Merge lp:~jaypipes/drizzle/replication-group-commit into lp:~drizzle-trunk/drizzle/development

[Jay Pipes]

[Jay Pipes (jaypipes)]

    This patch fixes a bug in the replication service and transaction
    proto file that was occurring when an UPDATE statement was SETting
    a field to a value by referencing the field itself.

    For instance, imagine the following scenario:

    CREATE TABLE t1 (id INT NOT NULL PRIMARY KEY, count INT NOT NULL);
    INSERT INTO t1 (id, counter) VALUES (1,1),(2,2),(3,3);
    UPDATE t1 SET counter = counter + 1 WHERE id IN (1,2);

    Previously, the UpdateHeader message contained a single set_value
    byte array containing the new value to set the field to. However,
    in scenarios such as the above, for each record being updated, the
    value that the field would be set to is different. Therefore, this
    patch moves the set_value byte array from the UpdateHeader message
    into the UpdateRecord message and names it after_value to match the
    existing before_value byte array.

    This patch adds a test case to the existing update.test case in the
    transaction log suite and modifies the statement_transform library to
    properly handle the above scenario.

[lp:~jaypipes/drizzle/replication-group-commit]

1177. By Jay Pipes <jpipes@serialcoder> on 2009-11-03

Updated filtered_replicator test due to new tests for UPDATE.

[Eric Lambert (elambert)]

looks good to me

review: Approve

=== modified file 'drizzled/message/statement_transform.cc'
--- drizzled/message/statement_transform.cc	2009-10-13 20:10:07 +0000
+++ drizzled/message/statement_transform.cc	2009-11-03 19:11:13 +0000
@@ -344,6 +344,23 @@
   destination->push_back(quoted_identifier);
   destination->append(" SET ");
 
+  return NONE;
+}
+
+enum message::TransformSqlError
+message::transformUpdateRecordToSql(const message::UpdateHeader &header,
+                                    const message::UpdateRecord &record,
+                                    std::string *destination,
+                                    enum message::TransformSqlVariant sql_variant)
+{
+  enum message::TransformSqlError error= transformUpdateHeaderToSql(header,
+                                                                    destination,
+                                                                    sql_variant);
+
+  char quoted_identifier= '`';
+  if (sql_variant == ANSI)
+    quoted_identifier= '"';
+
   /* Add field SET list to SQL string... */
   size_t num_set_fields= header.set_field_metadata_size();
   size_t x;
@@ -365,32 +382,13 @@
     if (should_quote_field_value)
       destination->push_back('\'');
 
-    destination->append(header.set_value(x));
+    destination->append(record.after_value(x));
 
     if (should_quote_field_value)
       destination->push_back('\'');
   }
 
-  return NONE;
-}
-
-enum message::TransformSqlError
-message::transformUpdateRecordToSql(const message::UpdateHeader &header,
-                                    const message::UpdateRecord &record,
-                                    std::string *destination,
-                                    enum message::TransformSqlVariant sql_variant)
-{
-  enum message::TransformSqlError error= transformUpdateHeaderToSql(header,
-                                                                    destination,
-                                                                    sql_variant);
-
-  char quoted_identifier= '`';
-  if (sql_variant == ANSI)
-    quoted_identifier= '"';
-
   size_t num_key_fields= header.key_field_metadata_size();
-  size_t x;
-  bool should_quote_field_value= false;
 
   destination->append(" WHERE ");
   for (x= 0; x < num_key_fields; ++x) 
@@ -423,64 +421,6 @@
 }
 
 enum message::TransformSqlError
-message::transformUpdateStatementToSql(const message::UpdateHeader &header,
-                                       const message::UpdateData &data,
-                                       std::string *destination,
-                                       enum message::TransformSqlVariant sql_variant)
-{
-  enum message::TransformSqlError error= transformUpdateHeaderToSql(header,
-                                                                    destination,
-                                                                    sql_variant);
-
-  char quoted_identifier= '`';
-  if (sql_variant == ANSI)
-    quoted_identifier= '"';
-
-  /* Add WHERE clause to SQL string... */
-  size_t num_key_fields= header.key_field_metadata_size();
-  size_t num_key_records= data.record_size();
-  size_t x, y;
-  bool should_quote_field_value= false;
-
-  destination->append(" WHERE ");
-  for (x= 0; x < num_key_records; ++x)
-  {
-    if (x != 0)
-      destination->append(" OR "); /* Always OR condition for multiple key records */
-
-    if (num_key_fields > 1)
-      destination->push_back('(');
-
-    for (y= 0; y < num_key_fields; ++y) 
-    {
-      const message::FieldMetadata &field_metadata= header.key_field_metadata(y);
-      
-      if (y != 0)
-        destination->append(" AND "); /* Always AND condition with a multi-column PK */
-
-      destination->push_back(quoted_identifier);
-      destination->append(field_metadata.name());
-      destination->push_back(quoted_identifier);
-
-      destination->push_back('=');
-
-      should_quote_field_value= message::shouldQuoteFieldValue(field_metadata.type());
-
-      if (should_quote_field_value)
-        destination->push_back('\'');
-
-      destination->append(data.record(x).key_value(y));
-
-      if (should_quote_field_value)
-        destination->push_back('\'');
-    }
-    if (num_key_fields > 1)
-      destination->push_back(')');
-  }
-  return error;
-}
-
-enum message::TransformSqlError
 message::transformDeleteHeaderToSql(const message::DeleteHeader &header,
                                     std::string *destination,
                                     enum message::TransformSqlVariant sql_variant)

=== modified file 'drizzled/message/statement_transform.h'
--- drizzled/message/statement_transform.h	2009-10-06 01:38:40 +0000
+++ drizzled/message/statement_transform.h	2009-11-03 19:11:13 +0000
@@ -154,32 +154,6 @@
 transformInsertHeaderToSql(const InsertHeader &header,
                            std::string *destination,
                            enum TransformSqlVariant sql_variant= DRIZZLE);
-/**
- * This function looks at a supplied UpdateHeader
- * and UpdateData message and constructs a correctly-formatted SQL
- * statement to the supplied destination string.
- *
- * @note
- *
- * This function constructs a <strong>single SQL statement</strong>
- * that contains all the update keys represented in all records in 
- * the UpdateData message.
- *
- * @param UpdateHeader message to transform
- * @param UpdateData message to transform
- * @param Destination string to append SQL to
- * @param Variation of SQL to generate
- *
- * @retval
- *  NONE if successful transformation
- * @retval
- *  Error code (see enum TransformSqlError definition) if failure
- */
-enum TransformSqlError
-transformUpdateStatementToSql(const UpdateHeader &header,
-                              const UpdateData &data,
-                              std::string *destination,
-                              enum TransformSqlVariant sql_variant= DRIZZLE);
 
 /**
  * Helper function to construct the header portion of an UPDATE

=== modified file 'drizzled/message/transaction.proto'
--- drizzled/message/transaction.proto	2009-10-06 01:38:40 +0000
+++ drizzled/message/transaction.proto	2009-11-03 19:11:13 +0000
@@ -364,7 +364,8 @@
 message UpdateRecord
 {
   repeated bytes key_value = 1; /* The value of keys of updated records (unique or primary key) */
-  repeated bytes before_value = 2; /* The value of the record before the update (optional) */
+  repeated bytes after_value = 2; /* The value of the field after the update */
+  repeated bytes before_value = 3; /* The value of the field before the update (optional) */
 }
 
 /*
@@ -372,6 +373,7 @@
  *
  * INSERT ... ON DUPLICATE KEY UPDATE
  * UPDATE
+ * REPLACE INTO when the UPDATE optimization occurs.
  *
  * The statement is composed of a header (UpdateHeader) containing
  * metadata about affected tables and fields, as well as one or more data
@@ -388,7 +390,6 @@
   required TableMetadata table_metadata = 1; /* Minimal metadata about the table affected */
   repeated FieldMetadata key_field_metadata = 2; /* Collection of metadata about key fields */
   repeated FieldMetadata set_field_metadata = 3; /* Collection of metadata about fields affected */
-  repeated bytes set_value = 4; /* The value of the field after the update */
 }
 
 message UpdateData

=== modified file 'drizzled/message/transaction_writer.cc'
--- drizzled/message/transaction_writer.cc	2009-10-19 19:19:38 +0000
+++ drizzled/message/transaction_writer.cc	2009-11-03 19:11:13 +0000
@@ -227,8 +227,6 @@
   sf_meta->set_name("a");
   sf_meta->set_type(message::Table::Field::VARCHAR);
 
-  header->add_set_value("5");
-
   /* Add new values... */
   message::UpdateData *data= statement->mutable_update_data();
   data->set_segment_id(1);
@@ -236,6 +234,7 @@
 
   message::UpdateRecord *record1= data->add_record();
 
+  record1->add_after_value("5");
   record1->add_key_value("1");
 
   statement->set_end_timestamp(getNanoTimestamp());
@@ -266,8 +265,6 @@
   sf_meta->set_name("a");
   sf_meta->set_type(message::Table::Field::VARCHAR);
 
-  header->add_set_value("5");
-
   /* Add new values... */
   message::UpdateData *data= statement->mutable_update_data();
   data->set_segment_id(1);
@@ -276,7 +273,9 @@
   message::UpdateRecord *record1= data->add_record();
   message::UpdateRecord *record2= data->add_record();
 
+  record1->add_after_value("5");
   record1->add_key_value("1");
+  record2->add_after_value("5");
   record2->add_key_value("2");
 
   statement->set_end_timestamp(getNanoTimestamp());

=== modified file 'drizzled/replication_services.cc'
--- drizzled/replication_services.cc	2009-10-19 19:19:38 +0000
+++ drizzled/replication_services.cc	2009-11-03 19:11:12 +0000
@@ -241,6 +241,8 @@
   {
     finalizeStatement(*statement, in_session);
   }
+  else
+    return; /* No data modification occurred inside the transaction */
   
   message::Transaction* transaction= getActiveTransaction(in_session);
 
@@ -463,7 +465,53 @@
       field_metadata= header->add_set_field_metadata();
       field_metadata->set_name(current_field->field_name);
       field_metadata->set_type(internalFieldTypeToFieldProtoType(current_field->type()));
-
+    }
+  }
+}
+void ReplicationServices::updateRecord(Session *in_session,
+                                       Table *in_table, 
+                                       const unsigned char *old_record, 
+                                       const unsigned char *new_record)
+{
+  if (! is_active)
+    return;
+
+  message::Statement &statement= getUpdateStatement(in_session, in_table, old_record, new_record);
+
+  message::UpdateData *data= statement.mutable_update_data();
+  data->set_segment_id(1);
+  data->set_end_segment(true);
+  message::UpdateRecord *record= data->add_record();
+
+  Field *current_field;
+  Field **table_fields= in_table->field;
+  String *string_value= new (in_session->mem_root) String(ReplicationServices::DEFAULT_RECORD_SIZE);
+  string_value->set_charset(system_charset_info);
+
+  while ((current_field= *table_fields++) != NULL) 
+  {
+    /*
+     * Here, we add the SET field values.  We used to do this in the setUpdateHeader() method, 
+     * but then realized that an UPDATE statement could potentially have different values for
+     * the SET field.  For instance, imagine this SQL scenario:
+     *
+     * CREATE TABLE t1 (id INT NOT NULL PRIMARY KEY, count INT NOT NULL);
+     * INSERT INTO t1 (id, counter) VALUES (1,1),(2,2),(3,3);
+     * UPDATE t1 SET counter = counter + 1 WHERE id IN (1,2);
+     *
+     * We will generate two UpdateRecord messages with different set_value byte arrays.
+     *
+     * The below really should be moved into the Field API and Record API.  But for now
+     * we do this crazy pointer fiddling to figure out if the current field
+     * has been updated in the supplied record raw byte pointers.
+     */
+    const unsigned char *old_ptr= (const unsigned char *) old_record + (ptrdiff_t) (current_field->ptr - in_table->record[0]); 
+    const unsigned char *new_ptr= (const unsigned char *) new_record + (ptrdiff_t) (current_field->ptr - in_table->record[0]); 
+
+    uint32_t field_length= current_field->pack_length(); /** @TODO This isn't always correct...check varchar diffs. */
+
+    if (memcmp(old_ptr, new_ptr, field_length) != 0)
+    {
       /* Store the original "read bit" for this field */
       bool is_read_set= current_field->isReadSet();
 
@@ -479,33 +527,10 @@
        */
       current_field->setReadSet(is_read_set);
 
-      header->add_set_value(string_value->c_ptr());
+      record->add_after_value(string_value->c_ptr());
       string_value->free();
     }
-  }
-}
-void ReplicationServices::updateRecord(Session *in_session,
-                                       Table *in_table, 
-                                       const unsigned char *old_record, 
-                                       const unsigned char *new_record)
-{
-  if (! is_active)
-    return;
-
-  message::Statement &statement= getUpdateStatement(in_session, in_table, old_record, new_record);
-
-  message::UpdateData *data= statement.mutable_update_data();
-  data->set_segment_id(1);
-  data->set_end_segment(true);
-  message::UpdateRecord *record= data->add_record();
-
-  Field *current_field;
-  Field **table_fields= in_table->field;
-  String *string_value= new (in_session->mem_root) String(ReplicationServices::DEFAULT_RECORD_SIZE);
-  string_value->set_charset(system_charset_info);
-
-  while ((current_field= *table_fields++) != NULL) 
-  {
+
     /* 
      * Add the WHERE clause values now...the fields which return true
      * for isReadSet() are in the WHERE clause.  For tables with no
@@ -520,6 +545,7 @@
        */
       string_value->free();
     }
+
   }
 }
 

=== modified file 'tests/suite/transaction_log/r/filtered_replicator.result'
--- tests/suite/transaction_log/r/filtered_replicator.result	2009-10-18 15:09:52 +0000
+++ tests/suite/transaction_log/r/filtered_replicator.result	2009-11-03 19:11:13 +0000
@@ -72,6 +72,15 @@
 INSERT INTO t1 (name,alias) VALUES ("jeff lebowski","dude");
 UPDATE t1 SET alias = "the dude" WHERE alias = "dude";
 DROP TABLE t1;
+CREATE TABLE t1 (
+id INT NOT NULL
+, counter INT NOT NULL
+, PRIMARY KEY (id)
+);
+INSERT INTO t1 (id, counter) VALUES (1,1),(2,2),(3,3);
+UPDATE t1 SET counter = counter + 1 WHERE id = 1;
+UPDATE t1 SET counter = counter + 1 WHERE id IN (2,3);
+DROP TABLE t1;
 DROP TABLE IF EXISTS `t1`;
 CREATE TABLE t1 ( id INT NOT NULL , padding VARCHAR(200) NOT NULL );
 INSERT INTO `test`.`t1` (`id`,`padding`) VALUES (1,'I love testing.');
@@ -118,4 +127,16 @@
 INSERT INTO `test`.`t1` (`id`,`name`,`alias`) VALUES (1,'jeff lebowski','dude');
 UPDATE `test`.`t1` SET `alias`='the dude' WHERE `id`=1;
 DROP TABLE `t1`;
+CREATE TABLE t1 ( id INT NOT NULL , counter INT NOT NULL , PRIMARY KEY (id) );
+START TRANSACTION;
+INSERT INTO `test`.`t1` (`id`,`counter`) VALUES (1,1);
+INSERT INTO `test`.`t1` (`id`,`counter`) VALUES (2,2);
+INSERT INTO `test`.`t1` (`id`,`counter`) VALUES (3,3);
+COMMIT;
+UPDATE `test`.`t1` SET `counter`=2 WHERE `id`=1;
+START TRANSACTION;
+UPDATE `test`.`t1` SET `counter`=3 WHERE `id`=2;
+UPDATE `test`.`t1` SET `counter`=4 WHERE `id`=3;
+COMMIT;
+DROP TABLE `t1`;
 SET GLOBAL transaction_log_truncate_debug= true;

=== added file 'tests/suite/transaction_log/r/no_modification.result'
--- tests/suite/transaction_log/r/no_modification.result	1970-01-01 00:00:00 +0000
+++ tests/suite/transaction_log/r/no_modification.result	2009-11-03 19:11:13 +0000
@@ -0,0 +1,20 @@
+DROP TABLE IF EXISTS t1;
+CREATE TABLE t1 (
+id INT NOT NULL
+, padding VARCHAR(200) NOT NULL
+);
+INSERT INTO t1 VALUES (1, "I love testing.");
+INSERT INTO t1 VALUES (2, "I hate testing.");
+START TRANSACTION;
+SELECT * FROM t1 WHERE id = 1;
+id	padding
+1	I love testing.
+SELECT * FROM t1 WHERE id = 2;
+id	padding
+2	I hate testing.
+COMMIT;
+DROP TABLE IF EXISTS `t1`;
+CREATE TABLE t1 ( id INT NOT NULL , padding VARCHAR(200) NOT NULL );
+INSERT INTO `test`.`t1` (`id`,`padding`) VALUES (1,'I love testing.');
+INSERT INTO `test`.`t1` (`id`,`padding`) VALUES (2,'I hate testing.');
+SET GLOBAL transaction_log_truncate_debug= true;

=== modified file 'tests/suite/transaction_log/r/update.result'
--- tests/suite/transaction_log/r/update.result	2009-10-18 15:09:52 +0000
+++ tests/suite/transaction_log/r/update.result	2009-11-03 19:11:13 +0000
@@ -17,6 +17,15 @@
 INSERT INTO t1 (name,alias) VALUES ("jeff lebowski","dude");
 UPDATE t1 SET alias = "the dude" WHERE alias = "dude";
 DROP TABLE t1;
+CREATE TABLE t1 (
+id INT NOT NULL
+, counter INT NOT NULL
+, PRIMARY KEY (id)
+);
+INSERT INTO t1 (id, counter) VALUES (1,1),(2,2),(3,3);
+UPDATE t1 SET counter = counter + 1 WHERE id = 1;
+UPDATE t1 SET counter = counter + 1 WHERE id IN (2,3);
+DROP TABLE t1;
 DROP TABLE IF EXISTS `t1`;
 CREATE TABLE t1 ( id INT NOT NULL , padding VARCHAR(200) NOT NULL , PRIMARY KEY (id) );
 INSERT INTO `test`.`t1` (`id`,`padding`) VALUES (1,'I love testing.');
@@ -31,4 +40,16 @@
 INSERT INTO `test`.`t1` (`id`,`name`,`alias`) VALUES (1,'jeff lebowski','dude');
 UPDATE `test`.`t1` SET `alias`='the dude' WHERE `id`=1;
 DROP TABLE `t1`;
+CREATE TABLE t1 ( id INT NOT NULL , counter INT NOT NULL , PRIMARY KEY (id) );
+START TRANSACTION;
+INSERT INTO `test`.`t1` (`id`,`counter`) VALUES (1,1);
+INSERT INTO `test`.`t1` (`id`,`counter`) VALUES (2,2);
+INSERT INTO `test`.`t1` (`id`,`counter`) VALUES (3,3);
+COMMIT;
+UPDATE `test`.`t1` SET `counter`=2 WHERE `id`=1;
+START TRANSACTION;
+UPDATE `test`.`t1` SET `counter`=3 WHERE `id`=2;
+UPDATE `test`.`t1` SET `counter`=4 WHERE `id`=3;
+COMMIT;
+DROP TABLE `t1`;
 SET GLOBAL transaction_log_truncate_debug= true;

=== added file 'tests/suite/transaction_log/t/no_modification-master.opt'
--- tests/suite/transaction_log/t/no_modification-master.opt	1970-01-01 00:00:00 +0000
+++ tests/suite/transaction_log/t/no_modification-master.opt	2009-11-03 19:11:13 +0000
@@ -0,0 +1,1 @@
+--default-replicator-enable --transaction-log-enable --scheduler=multi_thread

=== added file 'tests/suite/transaction_log/t/no_modification.inc'
--- tests/suite/transaction_log/t/no_modification.inc	1970-01-01 00:00:00 +0000
+++ tests/suite/transaction_log/t/no_modification.inc	2009-11-03 19:11:13 +0000
@@ -0,0 +1,26 @@
+# 
+# Simple test of the transaction log for when a START TRANSACTION ... COMMIT
+# is issued, but no data modifications actually occurred.
+# 
+# This situation occurs in, for instance, sysbench, which wraps even SELECTs
+# in a transaction.
+#
+
+--disable_warnings
+DROP TABLE IF EXISTS t1;
+--enable_warnings
+
+CREATE TABLE t1 (
+  id INT NOT NULL
+, padding VARCHAR(200) NOT NULL
+);
+
+INSERT INTO t1 VALUES (1, "I love testing.");
+INSERT INTO t1 VALUES (2, "I hate testing.");
+
+START TRANSACTION;
+
+SELECT * FROM t1 WHERE id = 1;
+SELECT * FROM t1 WHERE id = 2;
+
+COMMIT;

=== added file 'tests/suite/transaction_log/t/no_modification.test'
--- tests/suite/transaction_log/t/no_modification.test	1970-01-01 00:00:00 +0000
+++ tests/suite/transaction_log/t/no_modification.test	2009-11-03 19:11:13 +0000
@@ -0,0 +1,13 @@
+#
+# Tests when a transaction does not modify data.
+#
+
+# Populate log with some records...
+--source suite/transaction_log/t/no_modification.inc
+
+# Read in the transaction.log.
+
+--exec ../drizzled/message/transaction_reader var/master-data/transaction.log
+
+# Truncate the log file to reset for the next test
+--source suite/transaction_log/t/truncate_log.inc

=== modified file 'tests/suite/transaction_log/t/update.inc'
--- tests/suite/transaction_log/t/update.inc	2009-10-18 15:09:52 +0000
+++ tests/suite/transaction_log/t/update.inc	2009-11-03 19:11:13 +0000
@@ -43,3 +43,19 @@
 UPDATE t1 SET alias = "the dude" WHERE alias = "dude";
 
 DROP TABLE t1;
+
+# Tests UPDATE statement which changes an existing row
+# by referencing the changed field.
+
+CREATE TABLE t1 (
+  id INT NOT NULL
+, counter INT NOT NULL
+, PRIMARY KEY (id)
+);
+
+INSERT INTO t1 (id, counter) VALUES (1,1),(2,2),(3,3);
+
+UPDATE t1 SET counter = counter + 1 WHERE id = 1;
+UPDATE t1 SET counter = counter + 1 WHERE id IN (2,3);
+
+DROP TABLE t1;