bk commit into 5.1 tree (knielsen:1.2509)

[knielsen]

Below is the list of changes that have just been committed into a local
5.1 repository of knielsen. When knielsen does a push these changes will
be propagated to the main repository and, within 24 hours after the
push, to the public repository.
For information on how to access the public repository
see http://dev.mysql.com/doc/mysql/en/installing-source-tree.html

ChangeSet@stripped, 2007-05-25 15:07:03+02:00, knielsen@ymer.(none) +14 -0
  NdbRecord interpreted update + conflict resolution.

  sql/ha_ndbcluster.cc@stripped, 2007-05-25 15:06:57+02:00, knielsen@ymer.(none) +194 -30
    NdbRecord interpreted update + conflict resolution.

  sql/ha_ndbcluster.h@stripped, 2007-05-25 15:06:57+02:00, knielsen@ymer.(none) +8 -0
    NdbRecord interpreted update + conflict resolution.

  sql/ha_ndbcluster_binlog.cc@stripped, 2007-05-25 15:06:57+02:00, knielsen@ymer.(none) +7 -2
    NdbRecord interpreted update + conflict resolution.

  sql/mysqld.cc@stripped, 2007-05-25 15:06:57+02:00, knielsen@ymer.(none) +7 -0
    NdbRecord interpreted update + conflict resolution.

  storage/ndb/include/Makefile.am@stripped, 2007-05-25 15:06:57+02:00, knielsen@ymer.(none) +2
-1
    NdbRecord interpreted update + conflict resolution.

  storage/ndb/include/ndbapi/NdbInterpretedCode.hpp@stripped, 2007-05-25 15:06:58+02:00,
knielsen@ymer.(none) +131 -0
    New BitKeeper file ``storage/ndb/include/ndbapi/NdbInterpretedCode.hpp''

  storage/ndb/include/ndbapi/NdbInterpretedCode.hpp@stripped, 2007-05-25 15:06:58+02:00,
knielsen@ymer.(none) +0 -0

  storage/ndb/include/ndbapi/NdbOperation.hpp@stripped, 2007-05-25 15:06:57+02:00,
knielsen@ymer.(none) +3 -0
    NdbRecord interpreted update + conflict resolution.

  storage/ndb/include/ndbapi/NdbTransaction.hpp@stripped, 2007-05-25 15:06:57+02:00,
knielsen@ymer.(none) +9 -2
    NdbRecord interpreted update + conflict resolution.

  storage/ndb/src/kernel/blocks/dbtc/DbtcMain.cpp@stripped, 2007-05-25 15:06:57+02:00,
knielsen@ymer.(none) +0 -13
    NdbRecord interpreted update + conflict resolution.

  storage/ndb/src/ndbapi/Makefile.am@stripped, 2007-05-25 15:06:58+02:00, knielsen@ymer.(none)
+2 -1
    NdbRecord interpreted update + conflict resolution.

  storage/ndb/src/ndbapi/NdbInterpretedCode.cpp@stripped, 2007-05-25 15:06:58+02:00,
knielsen@ymer.(none) +158 -0
    New BitKeeper file ``storage/ndb/src/ndbapi/NdbInterpretedCode.cpp''

  storage/ndb/src/ndbapi/NdbInterpretedCode.cpp@stripped, 2007-05-25 15:06:58+02:00,
knielsen@ymer.(none) +0 -0

  storage/ndb/src/ndbapi/NdbOperation.cpp@stripped, 2007-05-25 15:06:58+02:00,
knielsen@ymer.(none) +1 -0
    NdbRecord interpreted update + conflict resolution.

  storage/ndb/src/ndbapi/NdbOperationExec.cpp@stripped, 2007-05-25 15:06:58+02:00,
knielsen@ymer.(none) +41 -1
    NdbRecord interpreted update + conflict resolution.

  storage/ndb/src/ndbapi/NdbTransaction.cpp@stripped, 2007-05-25 15:06:58+02:00,
knielsen@ymer.(none) +18 -3
    NdbRecord interpreted update + conflict resolution.

# This is a BitKeeper patch.  What follows are the unified diffs for the
# set of deltas contained in the patch.  The rest of the patch, the part
# that BitKeeper cares about, is below these diffs.
# User:	knielsen
# Host:	ymer.(none)
# Root:	/usr/local/mysql/mysql-5.1-telco-ndbrecord

--- 1.643/sql/mysqld.cc	2007-05-25 15:07:11 +02:00
+++ 1.644/sql/mysqld.cc	2007-05-25 15:07:11 +02:00
@@ -379,6 +379,7 @@ ulong ndb_extra_logging;
 #ifdef HAVE_NDB_BINLOG
 ulong ndb_report_thresh_binlog_epoch_slip;
 ulong ndb_report_thresh_binlog_mem_usage;
+my_bool opt_ndb_log_update_as_write;
 #endif
 
 extern const char *ndb_distribution_names[];
@@ -4927,6 +4928,7 @@ enum options_mysqld
   OPT_NDB_REPORT_THRESH_BINLOG_EPOCH_SLIP,
   OPT_NDB_REPORT_THRESH_BINLOG_MEM_USAGE,
   OPT_NDB_USE_COPYING_ALTER_TABLE,
+  OPT_NDB_LOG_UPDATE_AS_WRITE,
   OPT_SKIP_SAFEMALLOC,
   OPT_TEMP_POOL, OPT_TX_ISOLATION, OPT_COMPLETION_TYPE,
   OPT_SKIP_STACK_TRACE, OPT_SKIP_SYMLINKS,
@@ -5509,6 +5511,11 @@ master-ssl",
    (gptr*) &opt_ndb_shm,
    (gptr*) &opt_ndb_shm,
    0, GET_BOOL, OPT_ARG, OPT_NDB_SHM_DEFAULT, 0, 0, 0, 0, 0},
+  {"ndb-log-update-as-write", OPT_NDB_LOG_UPDATE_AS_WRITE,
+   "",
+   (gptr*) &opt_ndb_log_update_as_write,
+   (gptr*) &opt_ndb_log_update_as_write,
+   0, GET_BOOL, OPT_ARG, 1, 0, 0, 0, 0, 0},
   {"ndb-optimized-node-selection", OPT_NDB_OPTIMIZED_NODE_SELECTION,
    "Select nodes for transactions in a more optimal way.",
    (gptr*) &opt_ndb_optimized_node_selection,
--- New file ---
+++ storage/ndb/include/ndbapi/NdbInterpretedCode.hpp	07/05/25 15:06:58
/* Copyright (C) 2003 MySQL AB

   This program is free software; you can redistribute it and/or modify
   it under the terms of the GNU General Public License as published by
   the Free Software Foundation; version 2 of the License.

   This program is distributed in the hope that it will be useful,
   but WITHOUT ANY WARRANTY; without even the implied warranty of
   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
   GNU General Public License for more details.

   You should have received a copy of the GNU General Public License
   along with this program; if not, write to the Free Software
   Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA */

#ifndef NdbInterpretedCode_H
#define NdbInterpretedCode_H

#include <ndb_global.h>
#include <ndb_types.h>
#include "NdbDictionary.hpp"

class NdbTableImpl;
class NdbColumnImpl;


#ifndef DOXYGEN_SHOULD_SKIP_INTERNAL
/*
  @brief Stand-alone interpreted programs, for use with NdbRecord

  @details This class is used to prepare an NDB interpreted program for use
  in operations created using NdbRecord.

  The caller supplies the memory buffer to be used for the operations. If
  this is too small, an error will be returned when attempting to add
  instructions. The caller must deallocate the buffer if necessary after
  the InterpretedCode object is destroyed.

  Methods have minimal error checks, for efficiency.

  Note that this is currently considered an internal interface, and subject
  to change without notice.

  ToDo: We should add placeholders to this, so that one can use a single
  InterpretedCode object to create many operations from different embedded
  constants. Once we do this, we could add more error checks, as the cost
  would then not be paid for each operation creation.

*/
class NdbInterpretedCode
{
private:
  friend class NdbOperation;

  static const Uint32 MaxReg= 8;
  Uint32 *m_buffer;
  const Uint32 m_buffer_length;               // In words
  const Uint32 m_number_of_labels;
  /* Number of words used for instructions. */
  Uint32 m_instructions_length;
  /*
    Number of words left in buffer.
    This can be smaller than m_buffer_length-m_instructions_length, as
    the end of the buffer is used to store jump and label addresses for
    back-patching jumps.
  */
  Uint32 m_available_length;
  Uint32 m_number_of_jumps;
  enum Flags {
    /*
      We will set m_got_error if an error occurs, so that we can
      refuse to create an operation from InterpretedCode that the user
      forgot to do error checks on.
    */
    GotError= 0x1,
    /* Set if reading disk column. */
    UsesDisk= 0x2
  };
  Uint32 m_flags;

  int error()
  {
    m_flags|= GotError;
    return -1;
  }

  int add1(Uint32 x1)
  {
    if (unlikely(m_available_length < 1))
      return error();
    Uint32 current = m_instructions_length;
    m_buffer[current    ] = x1;
    m_instructions_length = current + 1;
    m_available_length--;
    return 0;
  }

  int add2(Uint32 x1, Uint32 x2)
  {
    if (unlikely(m_available_length < 2))
      return error();
    Uint32 current = m_instructions_length;
    m_buffer[current    ] = x1;
    m_buffer[current + 1] = x2;
    m_instructions_length = current + 2;
    m_available_length -= 2;
    return 0;
  }

  int add_branch(Uint32 type,
                 Uint32 RegLvalue, Uint32 RegRvalue, Uint32 Label);
  int read_attr_impl(const NdbColumnImpl *c, Uint32 RegDest);

public:
  NdbInterpretedCode(Uint32 *buffer, Uint32 buffer_word_size,
                     Uint32 num_labels);
  int load_const_u32(Uint32 RegDest, Uint32 Constant);
  int read_attr(const NdbDictionary::Table *table, Uint32 attrId,
                Uint32 RegDest);
  int read_attr(const NdbDictionary::Column *column, Uint32 RegDest);
  int branch_gt(Uint32 RegLvalue, Uint32 RegRvalue, Uint32 Label);
  int branch_eq(Uint32 RegLvalue, Uint32 RegRvalue, Uint32 Label);
  int branch_le(Uint32 RegLvalue, Uint32 RegRvalue, Uint32 Label);
  int def_label(int tLabelNo);
  int interpret_exit_nok(Uint32 ErrorCode);
  int interpret_exit_ok();
  int backpatch();
};
#endif

#endif

--- New file ---
+++ storage/ndb/src/ndbapi/NdbInterpretedCode.cpp	07/05/25 15:06:58
/* Copyright (C) 2003 MySQL AB

   This program is free software; you can redistribute it and/or modify
   it under the terms of the GNU General Public License as published by
   the Free Software Foundation; version 2 of the License.

   This program is distributed in the hope that it will be useful,
   but WITHOUT ANY WARRANTY; without even the implied warranty of
   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
   GNU General Public License for more details.

   You should have received a copy of the GNU General Public License
   along with this program; if not, write to the Free Software
   Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA */

#include "NdbInterpretedCode.hpp"
#include "Interpreter.hpp"
#include "NdbDictionaryImpl.hpp"


int NdbInterpretedCode::add_branch(Uint32 type, Uint32 RegLvalue,
                                   Uint32 RegRvalue, Uint32 Label)
{
  /* NOTE: subroutines not supported yet. */
  /* Store the jump location for backpatching. */
  if (m_available_length < 2 ||
      m_instructions_length > 0xffff ||
      Label > 0xffff)
    return error();
  m_number_of_jumps++;
  m_available_length--;
  Uint32 pos= m_buffer_length - (m_number_of_labels + m_number_of_jumps);
  m_buffer[pos]= m_instructions_length | (Label << 16);
  return add1(Interpreter::Branch(type, RegLvalue, RegRvalue));
}

NdbInterpretedCode::NdbInterpretedCode(Uint32 *buffer, Uint32 buffer_word_size,
                                       Uint32 num_labels) :
  m_buffer(buffer),
  m_buffer_length(buffer_word_size),
  m_number_of_labels(num_labels),
  m_instructions_length(0),
  m_available_length(m_buffer_length),
  m_number_of_jumps(0),
  m_flags(0)
{
  Uint32 i;
  /*
    At the end of the buffer we store the address of all labels as they are
    defined, so that we can back-patch forward jumps later.
    If the supplied buffer is too small, we cannot fail in the constructor,
    but all attempts to add instructions will fail due to full buffer.
  */
  for (i = 0; i < num_labels && i < buffer_word_size; i++)
  {
    buffer[buffer_word_size - (i + 1)] = ~(Uint32)0;
    m_available_length--;
  }
}

int NdbInterpretedCode::load_const_u32(Uint32 RegDest, Uint32 Constant)
{
  return add2(Interpreter::LoadConst32(RegDest % MaxReg), Constant);
}

int NdbInterpretedCode::read_attr_impl(const NdbColumnImpl *c, Uint32 RegDest)
{
  if (c->m_storageType == NDB_STORAGETYPE_DISK)
    m_flags|= UsesDisk;
  return add1(Interpreter::Read(c->m_attrId, RegDest % MaxReg));
}

int NdbInterpretedCode::read_attr(const NdbDictionary::Table *table,
                                  Uint32 attrId, Uint32 RegDest)
{
  const NdbTableImpl *t= & NdbTableImpl::getImpl(*table);
  const NdbColumnImpl *c= t->getColumn(attrId);
  if (unlikely(c == NULL))
    return error();
  return read_attr_impl(c, RegDest);
}

int NdbInterpretedCode::read_attr(const NdbDictionary::Column *column,
                                  Uint32 RegDest)
{
  return read_attr_impl(&NdbColumnImpl::getImpl(*column), RegDest);
}
int NdbInterpretedCode::branch_gt(Uint32 RegLvalue, Uint32 RegRvalue,
                                  Uint32 Label)
{
  return add_branch(Interpreter::BRANCH_GT_REG_REG,
                    RegLvalue, RegRvalue, Label);
}

int NdbInterpretedCode::branch_eq(Uint32 RegLvalue, Uint32 RegRvalue,
                                  Uint32 Label)
{
  return add_branch(Interpreter::BRANCH_EQ_REG_REG,
                    RegLvalue, RegRvalue, Label);
}

int NdbInterpretedCode::branch_le(Uint32 RegLvalue, Uint32 RegRvalue,
                                  Uint32 Label)
{
  return add_branch(Interpreter::BRANCH_LE_REG_REG,
                    RegLvalue, RegRvalue, Label);
}

int NdbInterpretedCode::def_label(int tLabelNo)
{
  if (tLabelNo < 0 ||
      (Uint32)tLabelNo >= m_number_of_labels)
    return error();
  Uint32 pos= m_buffer_length - (tLabelNo + 1);
  /* Check label not already defined. */
  if (m_buffer[pos] != ~(Uint32)0)
    return error();
  m_buffer[pos]= m_instructions_length;
  return 0;
}

int NdbInterpretedCode::interpret_exit_nok(Uint32 ErrorCode)
{
  return add1((ErrorCode << 16) + Interpreter::EXIT_REFUSE);
}

int NdbInterpretedCode::interpret_exit_ok()
{
  return add1(Interpreter::EXIT_OK);
}

int NdbInterpretedCode::backpatch()
{
  Uint32 i;
  Uint32 jumplist_end = m_buffer_length - m_number_of_labels;
  Uint32 jumplist_start = jumplist_end - m_number_of_jumps;

  for (i = jumplist_start; i < jumplist_end; i++)
  {
    Uint32 value = m_buffer[i];
    Uint32 instruction_address = value & 0xffff;
    Uint32 label_idx = value >> 16;
    assert(label_idx < m_number_of_labels);
    assert(instruction_address < m_instructions_length);
    Uint32 label_address = m_buffer[m_buffer_length - (label_idx + 1)];
    /* Check for jump to undefined label. */
    if (label_address == ~(Uint32)0)
      return error();
    /* Check if backwards or forwards jump. */
    if (label_address < instruction_address)
      m_buffer[instruction_address] |=
        (((instruction_address - label_address) << 16) | ((Uint32)1 << 31));
    else
      m_buffer[instruction_address] |=
        ((label_address - instruction_address) << 16);
  }
  return 0;
}


--- 1.65/storage/ndb/include/ndbapi/NdbTransaction.hpp	2007-05-25 15:07:11 +02:00
+++ 1.66/storage/ndb/include/ndbapi/NdbTransaction.hpp	2007-05-25 15:07:11 +02:00
@@ -30,6 +30,7 @@ class NdbIndexOperation;
 class NdbApiSignal;
 class Ndb;
 class NdbBlob;
+class NdbInterpretedCode;
 
 #ifndef DOXYGEN_SHOULD_SKIP_INTERNAL
 // to be documented later
@@ -590,7 +591,10 @@ public:
                             const unsigned char *mask= 0);
   NdbOperation *updateTuple(const NdbRecord *key_rec, const char *key_row,
                             const NdbRecord *attr_rec, const char *attr_row,
-                            const unsigned char *mask= 0);
+                            const unsigned char *mask= 0,
+                            const Uint32 *setPartitionId = 0,
+                            const void *getSetValue = 0,
+                            const NdbInterpretedCode *interpreted_code = 0);
   NdbOperation *writeTuple(const NdbRecord *key_rec, const char *key_row,
                            const NdbRecord *attr_rec, const char *attr_row,
                            const unsigned char *mask= 0);
@@ -814,7 +818,10 @@ private:						
                               const char *key_row,
                               const NdbRecord *attribute_record,
                               const char *attribute_row,
-                              const unsigned char *mask);
+                              const unsigned char *mask,
+                              const Uint32 *setPartitionId = 0,
+                              const void *getSetValue = 0,
+                              const NdbInterpretedCode *interpreted_code = 0);
 
   void		handleExecuteCompletion();
   

--- 1.52/storage/ndb/include/ndbapi/NdbOperation.hpp	2007-05-25 15:07:11 +02:00
+++ 1.53/storage/ndb/include/ndbapi/NdbOperation.hpp	2007-05-25 15:07:11 +02:00
@@ -32,6 +32,7 @@ class NdbColumnImpl;
 class NdbBlob;
 class TcKeyReq;
 class NdbRecord;
+class NdbInterpretedCode;
 
 /**
  * @class NdbOperation
@@ -1183,6 +1184,8 @@ protected:
     #include <Bitmask.hpp> in application code.
   */
   Uint32 m_read_mask[(NDB_MAX_ATTRIBUTES_IN_TABLE+31)>>5];
+  /* Interpreted program for NdbRecord operations. */
+  const NdbInterpretedCode *m_interpreted_code;
 
   Uint32 m_any_value;                           // Valid if m_use_any_value!=0
 

--- 1.154/storage/ndb/src/kernel/blocks/dbtc/DbtcMain.cpp	2007-05-25 15:07:11 +02:00
+++ 1.155/storage/ndb/src/kernel/blocks/dbtc/DbtcMain.cpp	2007-05-25 15:07:11 +02:00
@@ -5211,19 +5211,6 @@ void Dbtc::execLQHKEYREF(Signal* signal)
 	return;
       }
       
-      /* Only ref in certain situations */
-      {
-	const Uint32 opType = regTcPtr->operation;
-	if (   (opType == ZDELETE && errCode != ZNOT_FOUND)
-	    || (opType == ZINSERT && errCode != ZALREADYEXIST)
-	    || (opType == ZUPDATE && errCode != ZNOT_FOUND)
-	    || (opType == ZWRITE  && errCode != 839 && errCode != 840))
-	{
-	  TCKEY_abort(signal, 49);
-	  return;
-	}
-      }
-
       /* *************** */
       /*    TCKEYREF   < */
       /* *************** */

--- 1.87/storage/ndb/src/ndbapi/NdbTransaction.cpp	2007-05-25 15:07:11 +02:00
+++ 1.88/storage/ndb/src/ndbapi/NdbTransaction.cpp	2007-05-25 15:07:11 +02:00
@@ -2107,7 +2107,10 @@ NdbTransaction::setupRecordOp(NdbOperati
                               const char *key_row,
                               const NdbRecord *attribute_record,
                               const char *attribute_row,
-                              const unsigned char *mask)
+                              const unsigned char *mask,
+                              const Uint32 *setPartitionId,
+                              const void *getSetValue,
+                              const NdbInterpretedCode *interpreted_code)
 {
   NdbOperation *op;
   /*
@@ -2144,6 +2147,14 @@ NdbTransaction::setupRecordOp(NdbOperati
   op->m_attribute_row= attribute_row;
   attribute_record->copyMask(op->m_read_mask, mask);
 
+  assert(getSetValue == NULL);                  // ToDo
+  if (setPartitionId != NULL)
+  {
+    op->theDistributionKey= *setPartitionId;
+    op->theDistrKeyIndicator_= 1;
+  }
+  op->m_interpreted_code= interpreted_code;
+
   if (unlikely(attribute_record->flags & NdbRecord::RecHasBlob))
   {
     if (op->getBlobHandlesNdbRecord(this) == -1)
@@ -2222,7 +2233,10 @@ NdbTransaction::insertTuple(const NdbRec
 NdbOperation *
 NdbTransaction::updateTuple(const NdbRecord *key_rec, const char *key_row,
                             const NdbRecord *attr_rec, const char *attr_row,
-                            const unsigned char *mask)
+                            const unsigned char *mask,
+                            const Uint32 *setPartitionId,
+                            const void *getSetValue,
+                            const NdbInterpretedCode *interpreted_code)
 {
   /* Check that the NdbRecord specifies the full primary key. */
   if (!(key_rec->flags & NdbRecord::RecIsKeyRecord))
@@ -2233,7 +2247,8 @@ NdbTransaction::updateTuple(const NdbRec
 
   NdbOperation *op= setupRecordOp(NdbOperation::UpdateRequest,
                                   NdbOperation::LM_Exclusive, key_rec, key_row,
-                                  attr_rec, attr_row, mask);
+                                  attr_rec, attr_row, mask, setPartitionId,
+                                  getSetValue, interpreted_code);
   if(!op)
     return op;
 

--- 1.24/storage/ndb/src/ndbapi/NdbOperation.cpp	2007-05-25 15:07:11 +02:00
+++ 1.25/storage/ndb/src/ndbapi/NdbOperation.cpp	2007-05-25 15:07:11 +02:00
@@ -164,6 +164,7 @@ NdbOperation::init(const NdbTableImpl* t
   theBlobList = NULL;
   m_abortOption = -1;
   m_no_disk_flag = 1;
+  m_interpreted_code = NULL;
   m_use_any_value = 0;
 
   tSignal = theNdb->getSignal();

--- 1.39/storage/ndb/src/ndbapi/NdbOperationExec.cpp	2007-05-25 15:07:11 +02:00
+++ 1.40/storage/ndb/src/ndbapi/NdbOperationExec.cpp	2007-05-25 15:07:11 +02:00
@@ -21,6 +21,7 @@
 #include <Ndb.hpp>
 #include <NdbRecAttr.hpp>
 #include "NdbUtil.hpp"
+#include "NdbInterpretedCode.hpp"
 
 #include "Interpreter.hpp"
 #include <AttributeHeader.hpp>
@@ -617,6 +618,8 @@ NdbOperation::prepareSendNdbRecord(Uint3
   Uint32 remain;
   int res;
   Uint32 no_disk_flag;
+  Uint32 interpreted_code_end;
+  Uint32 *update_len_addr;
 
   assert(theStatus==UseNdbRecord);
   /* Not yet support for NdbRecord with interpreted operations. */
@@ -706,6 +709,34 @@ NdbOperation::prepareSendNdbRecord(Uint3
 
   no_disk_flag= m_no_disk_flag;
 
+  /* Handle any interpreted program. */
+  const NdbInterpretedCode *code= m_interpreted_code;
+  if (code)
+  {
+    if (code->m_flags & NdbInterpretedCode::UsesDisk)
+      no_disk_flag = 0;
+    Uint32 sizes[5];
+    sizes[0] = 0;               // Initial read.
+    sizes[1] = code->m_instructions_length;
+    sizes[2] = 0;               // Update size, filled later
+    update_len_addr= attrInfoPtr + 2;
+    sizes[3] = 0;               // Final read size, ToDo
+    sizes[4] = 0;               // Subroutine size, ToDo
+    res = insertATTRINFOData_NdbRecord(aTC_ConnectPtr, aTransId,
+                                       (const char *)sizes,
+                                       sizeof(sizes),
+                                       &attrInfoPtr, &remain);
+    if (res)
+      return res;
+    res = insertATTRINFOData_NdbRecord(aTC_ConnectPtr, aTransId,
+                                       (const char *)(code->m_buffer),
+                                       code->m_instructions_length*4,
+                                       &attrInfoPtr, &remain);
+    if (res)
+      return res;
+    interpreted_code_end= theTotalCurrAI_Len;
+  }
+
   OperationType tOpType= theOperationType;
   if ((tOpType == InsertRequest) || (tOpType == WriteRequest) ||
       (tOpType == UpdateRequest))
@@ -854,6 +885,15 @@ NdbOperation::prepareSendNdbRecord(Uint3
       return res;
   }
 
+  if (code)
+  {
+    /* ToDo: This only handles update currently. */
+    Uint32 update_word_length= theTotalCurrAI_Len - interpreted_code_end;
+    *update_len_addr = update_word_length;
+for (Uint32 i= 0; i < 5; i++) {fprintf(stderr, "%2d %p 0x%08x    %u  %p 0x%08x\n", i,
tcKeyReq->attrInfo, tcKeyReq->attrInfo[i], update_word_length, update_len_addr-2,
update_len_addr[(int)i - 2]);}
+assert(update_word_length > 0);
+  }
+
   Uint32 signalLength= hdrSize +
     (theTupKeyLen > TcKeyReq::MaxKeyInfo ?
          TcKeyReq::MaxKeyInfo : theTupKeyLen) +
@@ -906,7 +946,7 @@ NdbOperation::fillTcKeyReqHdr(TcKeyReq *
   TcKeyReq::setSimpleFlag(reqInfo, theSimpleIndicator);
   TcKeyReq::setCommitFlag(reqInfo, theCommitIndicator);
   TcKeyReq::setStartFlag(reqInfo, theStartIndicator);
-  TcKeyReq::setInterpretedFlag(reqInfo, theInterpretIndicator);
+  TcKeyReq::setInterpretedFlag(reqInfo, (m_interpreted_code != NULL));
   /* We will setNoDiskFlag() later when we have checked all columns. */
   TcKeyReq::setDirtyFlag(reqInfo, theDirtyIndicator);
   TcKeyReq::setOperationType(reqInfo, theOperationType);

--- 1.471/sql/ha_ndbcluster.cc	2007-05-25 15:07:11 +02:00
+++ 1.472/sql/ha_ndbcluster.cc	2007-05-25 15:07:11 +02:00
@@ -33,6 +33,7 @@
 #include "ha_ndbcluster_cond.h"
 #include <../util/Bitmask.hpp>
 #include <ndbapi/NdbIndexStat.hpp>
+#include <ndbapi/NdbInterpretedCode.hpp>
 
 #include "ha_ndbcluster_binlog.h"
 #include "ha_ndbcluster_tables.h"
@@ -290,51 +291,72 @@ void ha_ndbcluster::reset_state_at_execu
   thd_ndb->m_unsent_bytes= 0;
 }
 
-int execute_no_commit_ignore_no_key(ha_ndbcluster *h, NdbTransaction *trans)
+static int
+check_completed_operations(NdbTransaction *trans, const NdbOperation *first)
 {
-  int res= trans->execute(NdbTransaction::NoCommit,
-                          NdbOperation::AO_IgnoreError,
-                          h->m_force_send);
-  h->reset_state_at_execute();
-  if (res == -1)
-    return -1;
-
-  const NdbError &err= trans->getNdbError();
-  if (err.classification != NdbError::NoError &&
-      err.classification != NdbError::ConstraintViolation &&
-      err.classification != NdbError::NoDataFound)
-    return -1;
-
-  return 0;
+  DBUG_ENTER("check_completed_operations");
+  /*
+    Check that all errors are "accepted" errors
+  */
+  while (first)
+  {
+    const NdbError &err= first->getNdbError();
+    if (err.classification != NdbError::NoError &&
+        err.classification != NdbError::ConstraintViolation &&
+        err.classification != NdbError::NoDataFound &&
+        err.code != 9999)
+      DBUG_RETURN(err.code);
+    first= trans->getNextCompletedOperation(first);
+  }
+  DBUG_RETURN(0);
 }
 
 inline
 int execute_no_commit(ha_ndbcluster *h, NdbTransaction *trans,
 		      bool force_release)
 {
+  DBUG_ENTER("execute_no_commit");
   h->release_completed_operations(trans, force_release);
-  if (h->m_ignore_no_key)
-    return execute_no_commit_ignore_no_key(h,trans);
-  else
+  const NdbOperation *first= trans->getFirstDefinedOperation();
+  if (trans->execute(NdbTransaction::NoCommit,
+                      NdbOperation::AO_IgnoreError,
+                      h->m_force_send))
   {
-    int res= trans->execute(NdbTransaction::NoCommit,
-                            NdbOperation::AbortOnError,
-                            h->m_force_send);
     h->reset_state_at_execute();
-    return res;
+    DBUG_RETURN(-1);
   }
+  h->reset_state_at_execute();
+  if (!h->m_ignore_no_key || trans->getNdbError().code == 0)
+    DBUG_RETURN(trans->getNdbError().code);
+
+  DBUG_RETURN(check_completed_operations(trans, first));
+}
+
+inline
+int execute_commit(NdbTransaction *trans, int force_send, int ignore_error)
+{
+  DBUG_ENTER("execute_commit");
+  const NdbOperation *first= trans->getFirstDefinedOperation();
+  if (trans->execute(NdbTransaction::Commit,
+                     NdbOperation::AO_IgnoreError,
+                     force_send))
+  {
+    DBUG_RETURN(-1);
+  }
+  if (!ignore_error || trans->getNdbError().code == 0)
+    DBUG_RETURN(trans->getNdbError().code);
+  DBUG_RETURN(check_completed_operations(trans, first));
 }
 
 inline
 int execute_commit(ha_ndbcluster *h, NdbTransaction *trans)
 {
-  int res= trans->execute(NdbTransaction::Commit,
-                          NdbOperation::AbortOnError,
-                          h->m_force_send);
+  int res= execute_commit(trans, h->m_force_send, h->m_ignore_no_key);
   h->reset_state_at_execute();
   return res;
 }
 
+static
 inline
 int execute_commit(THD *thd, NdbTransaction *trans)
 {
@@ -343,21 +365,22 @@ int execute_commit(THD *thd, NdbTransact
     end time, and we will not take further action after this (nor can we
     easily, as we execute outside the context of the ha_ndbcluster object).
   */
-  return trans->execute(NdbTransaction::Commit,
-                        NdbOperation::AbortOnError,
-                        thd->variables.ndb_force_send);
+
+/* CHECK THIS */
+  return execute_commit(trans, thd->variables.ndb_force_send, FALSE);
 }
 
 inline
 int execute_no_commit_ie(ha_ndbcluster *h, NdbTransaction *trans,
 			 bool force_release)
 {
+  DBUG_ENTER("execute_no_commit_ie");
   h->release_completed_operations(trans, force_release);
   int res= trans->execute(NdbTransaction::NoCommit,
                           NdbOperation::AO_IgnoreError,
                           h->m_force_send);
   h->reset_state_at_execute();
-  return res;
+  DBUG_RETURN(res);
 }
 
 /*
@@ -794,6 +817,14 @@ ha_ndbcluster::get_row_buffer()
                     table->s->reclength + m_extra_reclength);
 }
 
+/* Return a row buffer, valid until next execute(). */
+char *
+ha_ndbcluster::get_buffer(uint size)
+{
+  Thd_ndb *thd_ndb= get_thd_ndb(table->in_use);
+  return alloc_root(&(thd_ndb->m_batch_mem_root), size);
+}
+
 /*
   When using extra hidden columns, the mysqld column bitmaps do not
   include bits for the extra columns, so we use this method to initialize
@@ -1215,6 +1246,32 @@ int ha_ndbcluster::get_metadata(const ch
   if (error)
     goto err;
 
+#ifdef HAVE_NDB_BINLOG
+  if (m_share->m_resolve_column == 0xFFFFFFFF)
+  {
+    m_share->m_resolve_column= 0;
+    for (int j= 0; j < tab->getNoOfColumns(); j++)
+    {
+      const NDBCOL *c= tab->getColumn(j);
+      if (strcmp("X", c->getName()) == 0)
+      {
+        switch (c->getType())
+        {
+        case  NDBCOL::Unsigned:
+          m_share->m_resolve_column= j+1;
+          DBUG_PRINT("info", ("resolve column %u", m_share->m_resolve_column));
+          break;
+        default:
+          DBUG_PRINT("info", ("resolve column %u has wrong type",
+                              m_share->m_resolve_column));
+          break;
+        }
+        break;
+      }
+    }
+  }
+#endif /* HAVE_NDB_BINLOG */
+
   DBUG_PRINT("info", ("fetched table %s", tab->getName()));
   m_table= tab;
 
@@ -3506,6 +3563,93 @@ int ha_ndbcluster::primary_key_cmp(const
   Update one record in NDB using primary key
 */
 
+#ifdef HAVE_NDB_BINLOG
+int
+ha_ndbcluster::update_row_timestamp_resolve(const byte *old_data, byte *new_data,
NdbInterpretedCode *code)
+{
+  const int do_before_image_check= 0;
+  uint32 resolve_column= m_share->m_resolve_column;
+
+  const uint error_wrong_before_image= 9998;
+  const uint error_lower_timestamp= 9999;
+  const uint label_0= 0, label_1= 1;
+  const Uint32 RegNewValue= 1, RegCurrentValue= 2;
+  int r;
+  Field *field= table->field[resolve_column-1];
+  DBUG_PRINT("info", ("interpreted update post equal"));
+  /*
+   * read new value from record
+   */
+  uint32 new_value;
+  {
+    const byte* field_ptr= field->ptr + (new_data - table->record[0]);
+    memcpy(&new_value, field_ptr, 4);
+  }
+  DBUG_PRINT("info", ("new_value: %u", new_value));
+  /*
+   * Load registers RegNewValue and RegCurrentValue
+   */
+  r= code->load_const_u32(RegNewValue, new_value);
+  DBUG_ASSERT(r == 0);
+  r= code->read_attr(m_table, resolve_column-1, RegCurrentValue);
+  DBUG_ASSERT(r == 0);
+  /*
+   * if RegNewValue > RegCurrentValue goto label_0
+   * else raise error for this row
+   */
+  r= code->branch_gt(RegCurrentValue, RegNewValue, label_0);
+  DBUG_ASSERT(r == 0);
+  r= code->interpret_exit_nok(error_lower_timestamp);
+  DBUG_ASSERT(r == 0);
+  r= code->def_label(label_0);
+  DBUG_ASSERT(r == 0);
+  if (do_before_image_check)
+  {
+    const Uint32 RegOldValue= RegNewValue; /* reuse register, new value not needed */
+    /*
+     * Now also check that old values are the same
+     */
+
+    /*
+     * read old value from record
+     */
+    uint32 old_value;
+    {
+      const byte* field_ptr= field->ptr + (old_data - table->record[0]);
+      memcpy(&old_value, field_ptr, 4);
+    }
+    DBUG_PRINT("info", ("old_value: %u", old_value));
+    /*
+     * Load register RegOldValue
+     */
+    r= code->load_const_u32(RegOldValue, old_value);
+    DBUG_ASSERT(r == 0);
+    // RegCurrentValue already stored
+    /*
+     * if RegOldValue == RegCurrentValue goto label label_1
+     * else raise error for this row
+     */
+    r= code->branch_eq(RegOldValue, RegCurrentValue, label_1);
+    DBUG_ASSERT(r == 0);
+    if (error_wrong_before_image)
+      r= code->interpret_exit_nok(error_wrong_before_image);
+    else
+      r= code->interpret_exit_nok(0);
+    DBUG_ASSERT(r == 0);
+    /*
+     * Exit ok and let transaction continue without error
+     */
+    r= code->def_label(label_1);
+    DBUG_ASSERT(r == 1);
+  }
+  r= code->interpret_exit_ok();
+  DBUG_ASSERT(r == 0);
+  r= code->backpatch();
+  DBUG_ASSERT(r == 0);
+  return r;
+}
+#endif
+
 int ha_ndbcluster::update_row(const byte *old_data, byte *new_data)
 {
   THD *thd= table->in_use;
@@ -3679,7 +3823,27 @@ int ha_ndbcluster::update_row(const byte
       key_row= (const char *)(&m_ref);
     }
 
-    if (!(op= trans->updateTuple(key_rec, key_row, m_ndb_record, row, mask)))
+    NdbInterpretedCode *code= NULL;
+#ifdef HAVE_NDB_BINLOG
+    if (m_share->m_resolve_column)
+    {
+      DBUG_PRINT("info", ("interpreted update pre equal on column %u",
+                          m_share->m_resolve_column));
+      /* Room for 10 instruction words, two labels, and two jumps. */
+      uint code_size= 14;
+      uint struct_size= sizeof(*code);
+      char *mem= get_buffer(struct_size + code_size*sizeof(Uint32));
+      Uint32* buffer= (Uint32 *)(mem + struct_size);
+      code= new(mem) NdbInterpretedCode(buffer, code_size, 2);
+      if (update_row_timestamp_resolve(old_data, new_data, code))
+      {
+        /* ToDo error handling */
+        abort();
+      }
+    }
+#endif
+    if (!(op= trans->updateTuple(key_rec, key_row, m_ndb_record, row, mask,
+                                 NULL, NULL, code)))
       ERR_RETURN(trans->getNdbError());  
   }
 

--- 1.182/sql/ha_ndbcluster.h	2007-05-25 15:07:11 +02:00
+++ 1.183/sql/ha_ndbcluster.h	2007-05-25 15:07:11 +02:00
@@ -41,6 +41,7 @@ class NdbIndexScanOperation; 
 class NdbBlob;
 class NdbIndexStat;
 class NdbEventOperation;
+class NdbInterpretedCode;
 class ha_ndbcluster_cond;
 
 // connectstring to cluster if given by mysqld
@@ -120,6 +121,7 @@ typedef struct st_ndbcluster_share {
 #ifdef HAVE_NDB_BINLOG
   uint32 connect_count;
   uint32 flags;
+  uint32 m_resolve_column;
   NdbEventOperation *op;
   NdbEventOperation *op_old; // for rename table
   char *old_names; // for rename table
@@ -248,6 +250,10 @@ class ha_ndbcluster: public handler
 
   int write_row(byte *buf);
   int update_row(const byte *old_data, byte *new_data);
+#ifdef HAVE_NDB_BINLOG
+  int update_row_timestamp_resolve(const byte *old_data, byte *new_data,
+                                   NdbInterpretedCode *);
+#endif
   int delete_row(const byte *buf);
   int index_init(uint index, bool sorted);
   int index_end();
@@ -506,6 +512,7 @@ private:
                                  uint op_batch_size, bool & batch_full);
   char *copy_row_to_buffer(Thd_ndb *thd_ndb, const byte *record);
   char *get_row_buffer();
+  char *get_buffer(uint size);
   void clear_extended_column_set(uchar *mask);
   uchar *copy_column_set(MY_BITMAP *bitmap);
 
@@ -547,6 +554,7 @@ private:
 
   void reset_state_at_execute();
   friend int execute_commit(ha_ndbcluster*, NdbTransaction*);
+  friend int execute_commit(NdbTransaction *, int, int);
   friend int execute_no_commit_ignore_no_key(ha_ndbcluster*, NdbTransaction*);
   friend int execute_no_commit(ha_ndbcluster*, NdbTransaction*, bool);
   friend int execute_no_commit_ie(ha_ndbcluster*, NdbTransaction*, bool);

--- 1.20/storage/ndb/src/ndbapi/Makefile.am	2007-05-25 15:07:11 +02:00
+++ 1.21/storage/ndb/src/ndbapi/Makefile.am	2007-05-25 15:07:11 +02:00
@@ -56,7 +56,8 @@ libndbapi_la_SOURCES = \
 	NdbBlob.cpp \
 	NdbIndexStat.cpp \
         SignalSender.cpp \
-        ObjectMap.cpp
+        ObjectMap.cpp \
+	NdbInterpretedCode.cpp
 
 INCLUDES_LOC = -I$(top_srcdir)/storage/ndb/src/mgmapi
 

--- 1.22/storage/ndb/include/Makefile.am	2007-05-25 15:07:11 +02:00
+++ 1.23/storage/ndb/include/Makefile.am	2007-05-25 15:07:11 +02:00
@@ -41,7 +41,8 @@ ndbapi/NdbScanFilter.hpp \
 ndbapi/NdbScanOperation.hpp \
 ndbapi/NdbIndexScanOperation.hpp \
 ndbapi/NdbIndexStat.hpp \
-ndbapi/ndberror.h
+ndbapi/ndberror.h \
+ndbapi/NdbInterpretedCode.hpp
 
 mgmapiinclude_HEADERS = \
 mgmapi/mgmapi.h \

--- 1.120/sql/ha_ndbcluster_binlog.cc	2007-05-25 15:07:11 +02:00
+++ 1.121/sql/ha_ndbcluster_binlog.cc	2007-05-25 15:07:11 +02:00
@@ -33,6 +33,8 @@
 #define assert(x) do { if(x) break; ::printf("%s %d: assert failed: %s\n", __FILE__,
__LINE__, #x); ::fflush(stdout); ::signal(SIGABRT,SIG_DFL); ::abort();
::kill(::getpid(),6); ::kill(::getpid(),9); } while (0)
 #endif
 
+extern my_bool opt_ndb_log_update_as_write;
+
 /*
   defines for cluster replication table names
 */
@@ -385,6 +387,7 @@ void ndbcluster_binlog_init_share(NDB_SH
   DBUG_ENTER("ndbcluster_binlog_init_share");
 
   share->connect_count= g_ndb_cluster_connection->get_connect_count();
+  share->m_resolve_column= 0xFFFFFFFF;
 
   share->op= 0;
   share->table= 0;
@@ -3499,7 +3502,8 @@ ndb_binlog_thread_handle_data_event(Ndb 
         since we do not have an after image
       */
       int n;
-      if (table->s->primary_key != MAX_KEY)
+      if (table->s->primary_key != MAX_KEY &&
+          opt_ndb_log_update_as_write)
         n= 0; /*
                 use the primary key only as it save time and space and
                 it is the only thing needed to log the delete
@@ -3546,7 +3550,8 @@ ndb_binlog_thread_handle_data_event(Ndb 
       ndb_unpack_record(table, share->ndb_value[0],
                         &b, table->record[0]);
       DBUG_EXECUTE("info", print_records(table, table->record[0]););
-      if (table->s->primary_key != MAX_KEY) 
+      if (table->s->primary_key != MAX_KEY &&
+          opt_ndb_log_update_as_write) 
       {
         /*
           since table has a primary key, we can do a write