« back to all changes in this revision
Viewing changes to drizzled/message/transaction_reader.cc
- browse files at revision 1166
- compare with another revision
- compare with revision 2422
- download diff
- download tarball
- view history from revision 1166
- Committer: Brian Aker
- Date: 2009-10-06 21:38:30 UTC
- mfrom: (1143.2.7 replication-group-commit)
- Revision ID: brian@gaz-20091006213830-avk8psae4iejhgap
Merge Jay
added
removed
drizzled/message/transaction_reader.cc
1
/* -*- mode: c++; c-basic-offset: 2; indent-tabs-mode: nil; -*-
2
* vim:expandtab:shiftwidth=2:tabstop=2:smarttab:
3
*
4
* Copyright (C) 2009 Sun Microsystems
5
*
6
* Authors:
7
*
8
* Jay Pipes <joinfu@sun.com>
9
*
10
* This program is free software; you can redistribute it and/or modify
11
* it under the terms of the GNU General Public License as published by
12
* the Free Software Foundation; version 2 of the License.
13
*
14
* This program is distributed in the hope that it will be useful,
15
* but WITHOUT ANY WARRANTY; without even the implied warranty of
16
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
17
* GNU General Public License for more details.
18
*
19
* You should have received a copy of the GNU General Public License
20
* along with this program; if not, write to the Free Software
21
* Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA
22
*/
23
1
24
#include <drizzled/global.h>
25
#include <drizzled/gettext.h>
2
26
#include <sys/types.h>
3
27
#include <sys/stat.h>
4
28
#include <fcntl.h>
5
29
#include <iostream>
6
#include <fstream>
7
30
#include <string>
31
#include <vector>
8
32
#include <unistd.h>
9
#include <drizzled/message/replication.pb.h>
33
#include <drizzled/message/transaction.pb.h>
34
#include <drizzled/message/statement_transform.h>
35
36
#include <google/protobuf/io/coded_stream.h>
37
#include <google/protobuf/io/zero_copy_stream_impl.h>
10
38
11
39
using namespace std;
12
40
using namespace drizzled;
13
14
/**
15
* @file Example application for reading change records and transactions
16
*/
17
18
static void printInsert(const message::Command &container,
19
const message::InsertRecord &record)
20
{
21
22
cout << "INSERT INTO `" << container.schema() << "`.`"
23
<< container.table() << "` (";
24
25
int32_t num_fields= record.insert_field_size();
26
27
int32_t x;
28
for (x= 0; x < num_fields; x++)
29
{
30
if (x != 0)
31
cout << ", ";
32
33
const message::Table::Field f= record.insert_field(x);
34
35
cout << "`" << f.name() << "`";
36
}
37
38
cout << ") VALUES ";
39
40
/*
41
* There may be an INSERT VALUES (),() type statement. We know the
42
* number of records is equal to the field_values array size divided
43
* by the number of fields.
44
*
45
* So, we do an inner and an outer loop. Outer loop is on the number
46
* of records and the inner loop on the number of fields. In this way,
47
* we know that record.field_values(outer_loop * num_fields) + inner_loop))
48
* always gives us our correct field value.
49
*/
50
int32_t num_records= (record.insert_value_size() / num_fields);
51
int32_t y;
52
for (x= 0; x < num_records; x++)
53
{
54
if (x != 0)
55
cout << ", ";
56
57
cout << "(";
58
for (y= 0; y < num_fields; y++)
59
{
60
if (y != 0)
61
cout << ", ";
62
63
cout << "\"" << record.insert_value((x * num_fields) + y) << "\"";
64
}
65
cout << ")";
66
}
67
68
cout << ";";
69
}
70
71
static void printDeleteWithPK(const message::Command &container,
72
const message::DeleteRecord &record)
73
{
74
cout << "DELETE FROM `" << container.schema() << "`.`" << container.table() << "`";
75
76
int32_t num_where_fields= record.where_field_size();
77
/*
78
* Make sure we catch anywhere we're not aligning the fields with
79
* the field_values arrays...
80
*/
81
assert(num_where_fields == record.where_value_size());
82
83
cout << " WHERE ";
84
int32_t x;
85
for (x= 0; x < num_where_fields; x++)
86
{
87
if (x != 0)
88
cout << " AND "; /* Always AND condition with a multi-column PK */
89
90
const message::Table::Field f= record.where_field(x);
91
92
/* Always equality conditions */
93
cout << "`" << f.name() << "` = \"" << record.where_value(x) << "\"";
94
}
95
}
96
97
static void printUpdateWithPK(const message::Command &container,
98
const message::UpdateRecord &record)
99
{
100
int32_t num_update_fields= record.update_field_size();
101
int32_t x;
102
103
cout << "UPDATE `" << container.schema() << "`.`" << container.table() << "` SET ";
104
105
for (x= 0;x < num_update_fields; x++)
106
{
107
message::Table::Field f= record.update_field(x);
108
109
if (x != 0)
110
cout << ", ";
111
112
cout << "`" << f.name() << "` = \"" << record.after_value(x) << "\"";
113
}
114
115
int32_t num_where_fields= record.where_field_size();
116
/*
117
* Make sure we catch anywhere we're not aligning the fields with
118
* the field_values arrays...
119
*/
120
assert(num_where_fields == record.where_value_size());
121
122
cout << " WHERE ";
123
for (x= 0;x < num_where_fields; x++)
124
{
125
if (x != 0)
126
cout << " AND "; /* Always AND condition with a multi-column PK */
127
128
const message::Table::Field f= record.where_field(x);
129
130
/* Always equality conditions */
131
cout << "`" << f.name() << "` = \"" << record.where_value(x) << "\"";
41
using namespace google;
42
43
static void printStatement(const message::Statement &statement)
44
{
45
cout << "/* Start Timestamp: " << statement.start_timestamp() << " ";
46
cout << " End Timestamp: " << statement.end_timestamp() << " */" << endl;
47
48
vector<string> sql_strings;
49
50
message::transformStatementToSql(statement, sql_strings, message::DRIZZLE);
51
52
vector<string>::iterator sql_string_iter= sql_strings.begin();
53
const std::string newline= "\n";
54
while (sql_string_iter != sql_strings.end())
55
{
56
string &sql= *sql_string_iter;
57
/*
58
* Replace \n with spaces so that SQL statements
59
* are always on a single line
60
*/
61
while (sql.find(newline) != std::string::npos)
62
sql.replace(sql.find(newline), 1, " ");
63
64
cout << sql << ';' << endl;
65
++sql_string_iter;
132
66
}
133
67
}
134
68
135
69
static void printTransaction(const message::Transaction &transaction)
136
70
{
137
int32_t e_size;
138
139
cout << "/* Start Time: " << transaction.start_timestamp() << " */ START TRANSACTION;"<< endl;
140
141
for (e_size= 0; e_size < transaction.command_size(); e_size++)
71
const message::TransactionContext trx= transaction.transaction_context();
72
73
cout << "/* SERVER ID: " << trx.server_id() << " TRX ID: " << trx.transaction_id() << " */ " << endl;
74
75
size_t num_statements= transaction.statement_size();
76
size_t x;
77
78
for (x= 0; x < num_statements; ++x)
142
79
{
143
const message::Command command= transaction.command(e_size);
144
145
message::TransactionContext trx= command.transaction_context();
146
147
cout << "/* SID: " << trx.server_id() << " XID: " << trx.transaction_id() << " */ ";
148
149
switch (command.type())
150
{
151
case message::Command::START_TRANSACTION:
152
cout << "START TRANSACTION;";
153
break;
154
case message::Command::COMMIT:
155
cout << "COMMIT;";
156
break;
157
case message::Command::ROLLBACK:
158
cout << "ROLLBACK;";
159
break;
160
case message::Command::INSERT:
161
{
162
printInsert(command, command.insert_record());
163
break;
164
}
165
case message::Command::DELETE:
166
{
167
printDeleteWithPK(command, command.delete_record());
168
break;
169
}
170
case message::Command::UPDATE:
171
{
172
printUpdateWithPK(command, command.update_record());
173
break;
174
}
175
default:
176
assert(0);
177
}
178
cout << endl;
80
const message::Statement &statement= transaction.statement(x);
81
printStatement(statement);
179
82
}
180
cout << "/* Commit Time: " << transaction.end_timestamp() << " */ COMMIT;" << endl;
181
83
}
182
84
183
85
int main(int argc, char* argv[])
187
89
188
90
if (argc != 2)
189
91
{
190
cerr << "Usage: " << argv[0] << " TRANSACTION_LOG" << endl;
92
fprintf(stderr, _("Usage: %s TRANSACTION_LOG\n"), argv[0]);
191
93
return -1;
192
94
}
193
95
194
96
message::Transaction transaction;
195
97
196
if ((file= open(argv[1], O_RDONLY)) == -1)
98
file= open(argv[1], O_RDONLY);
99
if (file == -1)
197
100
{
198
cerr << "Can not open file: " << argv[1] << endl;
101
fprintf(stderr, _("Cannot open file: %s\n"), argv[1]);
102
return -1;
199
103
}
200
104
105
protobuf::io::ZeroCopyInputStream *raw_input= new protobuf::io::FileInputStream(file);
106
protobuf::io::CodedInputStream *coded_input= new protobuf::io::CodedInputStream(raw_input);
107
201
108
char *buffer= NULL;
202
char *temp_buffer;
109
char *temp_buffer= NULL;
110
uint64_t length= 0;
111
uint64_t previous_length= 0;
112
bool result= true;
203
113
204
while (1)
114
/* Read in the length of the command */
115
while (result == true && coded_input->ReadLittleEndian64(&length) == true)
205
116
{
206
uint64_t length;
207
208
/* Read the size */
209
if (read(file, &length, sizeof(uint64_t)) != sizeof(uint64_t))
210
break;
211
212
117
if (length > SIZE_MAX)
213
118
{
214
cerr << "Attempted to read record bigger than SIZE_MAX" << endl;
119
fprintf(stderr, _("Attempted to read record bigger than SIZE_MAX\n"));
215
120
exit(1);
216
121
}
217
122
218
temp_buffer= (char *)realloc(buffer, (size_t)length);
123
if (buffer == NULL)
124
{
125
/*
126
* First time around...just malloc the length. This block gets rid
127
* of a GCC warning about uninitialized temp_buffer.
128
*/
129
temp_buffer= (char *) malloc((size_t) length);
130
}
131
/* No need to allocate if we have a buffer big enough... */
132
else if (length > previous_length)
133
{
134
temp_buffer= (char *) realloc(buffer, (size_t) length);
135
}
136
219
137
if (temp_buffer == NULL)
220
138
{
221
cerr << "Memory allocation failure trying to allocate " << length << " bytes." << endl;
222
exit(1);
223
}
224
memset(temp_buffer, 0, (size_t)length);
225
buffer= temp_buffer;
226
size_t read_bytes= 0;
227
228
/* Read the transaction */
229
if ((read_bytes= read(file, buffer, (size_t)length)) != (size_t)length)
230
{
231
cerr << "Could not read entire transaction. Read " << read_bytes << " bytes instead of " << length << " bytes." << endl;
232
exit(1);
233
}
234
transaction.ParseFromArray(buffer, (int) length);
139
fprintf(stderr, _("Memory allocation failure trying to allocate %" PRIu64 " bytes.\n"),
140
static_cast<uint64_t>(length));
141
break;
142
}
143
else
144
buffer= temp_buffer;
145
146
/* Read the Command */
147
result= coded_input->ReadRaw(buffer, length);
148
if (result == false)
149
{
150
fprintf(stderr, _("Could not read transaction message.\n"));
151
fprintf(stderr, _("GPB ERROR: %s.\n"), strerror(errno));
152
fprintf(stderr, _("Raw buffer read: %s.\n"), buffer);
153
break;
154
}
155
156
result= transaction.ParseFromArray(buffer, static_cast<size_t>(length));
157
if (result == false)
158
{
159
fprintf(stderr, _("Unable to parse command. Got error: %s.\n"), transaction.InitializationErrorString().c_str());
160
if (buffer != NULL)
161
fprintf(stderr, _("BUFFER: %s\n"), buffer);
162
break;
163
}
235
164
236
165
/* Print the transaction */
237
166
printTransaction(transaction);
167
168
previous_length= length;
238
169
}
239
return 0;
170
if (buffer)
171
free(buffer);
172
173
delete coded_input;
174
delete raw_input;
175
176
return (result == true ? 0 : 1);
240
177
}
Loggerhead is a web-based interface for Breezy
Version: 2.0.1