ndb - wl#2972 rbr blobs: write blob data to binlog

mysql-test/t/disabled.def:
  rbr blobs: write data + dict cache workarounds
sql/ha_ndbcluster.cc:
  rbr blobs: write data + dict cache workarounds
sql/ha_ndbcluster.h:
  rbr blobs: write data + dict cache workarounds
sql/ha_ndbcluster_binlog.cc:
  rbr blobs: write data + dict cache workarounds
storage/ndb/include/ndbapi/NdbDictionary.hpp:
  rbr blobs: write data + dict cache workarounds
storage/ndb/src/ndbapi/NdbBlob.cpp:
  rbr blobs: write data + dict cache workarounds
storage/ndb/src/ndbapi/NdbDictionary.cpp:
  rbr blobs: write data + dict cache workarounds
storage/ndb/src/ndbapi/NdbDictionaryImpl.cpp:
  rbr blobs: write data + dict cache workarounds
storage/ndb/src/ndbapi/NdbDictionaryImpl.hpp:
  rbr blobs: write data + dict cache workarounds

mysql-test/t/disabled.def

@@ -28,3 +28,7 @@ ndb_autodiscover : Needs to be fixed w.r.t binlog
 ndb_autodiscover2 : Needs to be fixed w.r.t binlog
 system_mysql_db : Needs fixing
 system_mysql_db_fix : Needs fixing
+#ndb_alter_table_row : sometimes wrong error 1015!=1046
+ndb_gis : garbled msgs from corrupt THD* + partitioning problem
+
+# vim: set filetype=conf:

sql/ha_ndbcluster.cc

@@ -35,6 +35,11 @@
 
 #include "ha_ndbcluster_binlog.h"
 
+#ifdef ndb_dynamite
+#undef assert
+#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
+
 // options from from mysqld.cc
 extern my_bool opt_ndb_optimized_node_selection;
 extern const char *opt_ndbcluster_connectstring;
@@ -791,10 +796,20 @@ int g_get_ndb_blobs_value(NdbBlob *ndb_blob, void *arg)
   if (ndb_blob->blobsNextBlob() != NULL)
     DBUG_RETURN(0);
   ha_ndbcluster *ha= (ha_ndbcluster *)arg;
-  DBUG_RETURN(ha->get_ndb_blobs_value(ndb_blob));
+  int ret= get_ndb_blobs_value(ha->table, ha->m_value,
+                               ha->m_blobs_buffer, ha->m_blobs_buffer_size,
+                               0);
+  DBUG_RETURN(ret);
 }
 
-int ha_ndbcluster::get_ndb_blobs_value(NdbBlob *last_ndb_blob)
+/*
+  This routine is shared by injector.  There is no common blobs buffer
+  so the buffer and length are passed by reference.  Injector also
+  passes a record pointer diff.
+ */
+int get_ndb_blobs_value(TABLE* table, NdbValue* value_array,
+                        byte*& buffer, uint& buffer_size,
+                        my_ptrdiff_t ptrdiff)
 {
   DBUG_ENTER("get_ndb_blobs_value");
 
@@ -803,14 +818,17 @@ int ha_ndbcluster::get_ndb_blobs_value(NdbBlob *last_ndb_blob)
   for (int loop= 0; loop <= 1; loop++)
   {
     uint32 offset= 0;
-    for (uint i= 0; i < table_share->fields; i++)
+    for (uint i= 0; i < table->s->fields; i++)
     {
       Field *field= table->field[i];
-      NdbValue value= m_value[i];
+      NdbValue value= value_array[i];
       if (value.ptr != NULL && (field->flags & BLOB_FLAG))
       {
         Field_blob *field_blob= (Field_blob *)field;
         NdbBlob *ndb_blob= value.blob;
+        int isNull;
+        ndb_blob->getDefined(isNull);
+        if (isNull == 0) { // XXX -1 should be allowed only for events
           Uint64 blob_len= 0;
           if (ndb_blob->getLength(blob_len) != 0)
             DBUG_RETURN(-1);
@@ -820,27 +838,31 @@ int ha_ndbcluster::get_ndb_blobs_value(NdbBlob *last_ndb_blob)
             blob_size+= 8 - blob_size % 8;
           if (loop == 1)
           {
-          char *buf= m_blobs_buffer + offset;
+            char *buf= buffer + offset;
             uint32 len= 0xffffffff;  // Max uint32
-          DBUG_PRINT("value", ("read blob ptr=%lx  len=%u",
+            DBUG_PRINT("info", ("read blob ptr=%p len=%u",
                                 buf, (uint) blob_len));
             if (ndb_blob->readData(buf, len) != 0)
               DBUG_RETURN(-1);
             DBUG_ASSERT(len == blob_len);
+            // Ugly hack assumes only ptr needs to be changed
+            field_blob->ptr += ptrdiff;
             field_blob->set_ptr(len, buf);
+            field_blob->ptr -= ptrdiff;
           }
           offset+= blob_size;
         }
       }
-    if (loop == 0 && offset > m_blobs_buffer_size)
+    }
+    if (loop == 0 && offset > buffer_size)
     {
-      my_free(m_blobs_buffer, MYF(MY_ALLOW_ZERO_PTR));
-      m_blobs_buffer_size= 0;
-      DBUG_PRINT("value", ("allocate blobs buffer size %u", offset));
-      m_blobs_buffer= my_malloc(offset, MYF(MY_WME));
-      if (m_blobs_buffer == NULL)
+      my_free(buffer, MYF(MY_ALLOW_ZERO_PTR));
+      buffer_size= 0;
+      DBUG_PRINT("info", ("allocate blobs buffer size %u", offset));
+      buffer= my_malloc(offset, MYF(MY_WME));
+      if (buffer == NULL)
         DBUG_RETURN(-1);
-      m_blobs_buffer_size= offset;
+      buffer_size= offset;
     }
   }
   DBUG_RETURN(0);
@@ -2713,15 +2735,23 @@ void ndb_unpack_record(TABLE *table, NdbValue *value,
       else
       {
         NdbBlob *ndb_blob= (*value).blob;
-        bool isNull= TRUE;
-#ifndef DBUG_OFF
-        int ret= 
-#endif
-          ndb_blob->getNull(isNull);
-        DBUG_ASSERT(ret == 0);
-        if (isNull)
+        int isNull;
+        ndb_blob->getDefined(isNull);
+        if (isNull != 0)
+        {
+          uint col_no = ndb_blob->getColumn()->getColumnNo();
+          if (isNull == 1)
+          {
+            DBUG_PRINT("info",("[%u] NULL", col_no))
             field->set_null(row_offset);
           }
+          else
+          {
+            DBUG_PRINT("info",("[%u] UNDEFINED", col_no));
+            bitmap_clear_bit(defined, col_no);
+          }
+        }
+      }
     }
   }
   DBUG_VOID_RETURN;
@@ -4713,6 +4743,7 @@ int ha_ndbcluster::alter_table_name(const char *to)
   NDBDICT *dict= ndb->getDictionary();
   const NDBTAB *orig_tab= (const NDBTAB *) m_table;
   DBUG_ENTER("alter_table_name");
+  DBUG_PRINT("info", ("from: %s to: %s", orig_tab->getName(), to));
 
   NdbDictionary::Table new_tab= *orig_tab;
   new_tab.setName(to);

sql/ha_ndbcluster.h

@@ -25,6 +25,9 @@
 #pragma interface                       /* gcc class implementation */
 #endif
 
+/* Blob tables and events are internal to NDB and must never be accessed */
+#define IS_NDB_BLOB_PREFIX(A) is_prefix(A, "NDB$BLOB")
+
 #include <NdbApi.hpp>
 #include <ndbapi_limits.h>
 
@@ -78,6 +81,10 @@ typedef struct ndb_index_data {
 
 typedef union { const NdbRecAttr *rec; NdbBlob *blob; void *ptr; } NdbValue;
 
+int get_ndb_blobs_value(TABLE* table, NdbValue* value_array,
+                        byte*& buffer, uint& buffer_size,
+                        my_ptrdiff_t ptrdiff);
+
 typedef enum {
   NSS_INITIAL= 0,
   NSS_DROPPED,
@@ -114,6 +121,7 @@ typedef struct st_ndbcluster_share {
 #ifdef HAVE_NDB_BINLOG
 /* NDB_SHARE.flags */
 #define NSF_HIDDEN_PK 1 /* table has hidden primary key */
+#define NSF_BLOB_FLAG 2 /* table has blob attributes */
 #define NSF_NO_BINLOG 4 /* table should not be binlogged */
 #endif
 

storage/ndb/include/ndbapi/NdbDictionary.hpp

@@ -883,6 +883,7 @@ public:
 
   private:
 #ifndef DOXYGEN_SHOULD_SKIP_INTERNAL
+    friend class NdbDictionaryImpl;
     friend class NdbTableImpl;
 #endif
     class NdbTableImpl & m_impl;
@@ -1764,6 +1765,7 @@ public:
 #ifndef DOXYGEN_SHOULD_SKIP_INTERNAL
     const Table * getTable(const char * name, void **data) const;
     void set_local_table_data_size(unsigned sz);
+    void fix_blob_events(const Table* table, const char* ev_name);
 #endif
   };
 };

storage/ndb/src/ndbapi/NdbBlob.cpp

@@ -1327,10 +1327,10 @@ NdbBlob::prepareColumn()
   assert((NDB_BLOB_HEAD_SIZE << 2) == sizeof(Head));
   assert(theColumn->m_attrSize * theColumn->m_arraySize == sizeof(Head) + theInlineSize);
   if (thePartSize > 0) {
-    const NdbDictionary::Table* bt = NULL;
-    const NdbDictionary::Column* bc = NULL;
+    const NdbTableImpl* bt = NULL;
+    const NdbColumnImpl* bc = NULL;
     if (theStripeSize == 0 ||
-        (bt = theColumn->getBlobTable()) == NULL ||
+        (bt = theColumn->m_blobTable) == NULL ||
         (bc = bt->getColumn("DATA")) == NULL ||
         bc->getType() != partType ||
         bc->getLength() != (int)thePartSize) {

storage/ndb/src/ndbapi/NdbDictionary.cpp

@@ -1478,6 +1478,12 @@ NdbDictionary::Dictionary::getNdbError() const {
   return m_impl.getNdbError();
 }
 
+void
+NdbDictionary::Dictionary::fix_blob_events(const Table* table, const char* ev_name)
+{
+  m_impl.fix_blob_events(table, ev_name);
+}
+
 // printers
 
 NdbOut&

storage/ndb/src/ndbapi/NdbDictionaryImpl.cpp

@@ -3398,12 +3398,14 @@ NdbDictionaryImpl::getEvent(const char * eventName)
     if (ev->m_tableId      == info->m_table_impl->m_id &&
 	ev->m_tableVersion == info->m_table_impl->m_version)
       break;
+    DBUG_PRINT("error",("%s: retry=%d: "
+                        "table version mismatch, event: [%u,%u] table: [%u,%u]",
+                        ev->getTableName(), retry,
+                        ev->m_tableId, ev->m_tableVersion,
+                        info->m_table_impl->m_id, info->m_table_impl->m_version));
     if (retry)
     {
       m_error.code= 241;
-      DBUG_PRINT("error",("%s: table version mismatch, event: [%u,%u] table: [%u,%u]",
-			  ev->getTableName(), ev->m_tableId, ev->m_tableVersion,
-			  info->m_table_impl->m_id, info->m_table_impl->m_version));
       delete ev;
       DBUG_RETURN(NULL);
     }
@@ -3607,7 +3609,7 @@ NdbDictionaryImpl::dropEvent(const char * eventName)
     if (m_error.code != 723 && // no such table
         m_error.code != 241)   // invalid table
       DBUG_RETURN(-1);
-    DBUG_PRINT("info", ("no table, drop by name alone"));
+    DBUG_PRINT("info", ("no table err=%d, drop by name alone", m_error.code));
     evnt = new NdbEventImpl();
     evnt->setName(eventName);
   }
@@ -3644,7 +3646,17 @@ NdbDictionaryImpl::dropBlobEvents(const NdbEventImpl& evnt)
       (void)dropEvent(bename);
     }
   } else {
-    // could loop over MAX_ATTRIBUTES_IN_TABLE ...
+    // loop over MAX_ATTRIBUTES_IN_TABLE ...
+    Uint32 i;
+    for (i = 0; i < MAX_ATTRIBUTES_IN_TABLE; i++) {
+      char bename[MAX_TAB_NAME_SIZE];
+      // XXX should get name from NdbBlob
+      sprintf(bename, "NDB$BLOBEVENT_%s_%u", evnt.getName(), i);
+      NdbEventImpl* bevnt = new NdbEventImpl();
+      bevnt->setName(bename);
+      (void)m_receiver.dropEvent(*bevnt);
+      delete bevnt;
+    }
   }
   DBUG_RETURN(0);
 }
@@ -4631,6 +4643,30 @@ NdbDictInterface::parseFileInfo(NdbFileImpl &dst,
   return 0;
 }
 
+// XXX temp
+void
+NdbDictionaryImpl::fix_blob_events(const NdbDictionary::Table* table, const char* ev_name)
+{
+  const NdbTableImpl& t = table->m_impl;
+  const NdbEventImpl* ev = getEvent(ev_name);
+  assert(ev != NULL && ev->m_tableImpl == &t);
+  Uint32 i;
+  for (i = 0; i < t.m_columns.size(); i++) {
+    assert(t.m_columns[i] != NULL);
+    const NdbColumnImpl& c = *t.m_columns[i];
+    if (! c.getBlobType() || c.getPartSize() == 0)
+      continue;
+    char bename[200];
+    NdbBlob::getBlobEventName(bename, ev, &c);
+    // following fixes dict cache blob table
+    NdbEventImpl* bev = getEvent(bename);
+    if (c.m_blobTable != bev->m_tableImpl) {
+      // XXX const violation
+      ((NdbColumnImpl*)&c)->m_blobTable = bev->m_tableImpl;
+    }
+  }
+}
+
 template class Vector<int>;
 template class Vector<Uint16>;
 template class Vector<Uint32>;

storage/ndb/src/ndbapi/NdbDictionaryImpl.hpp

@@ -592,6 +592,9 @@ public:
 
   NdbDictInterface m_receiver;
   Ndb & m_ndb;
+
+  // XXX temp
+  void fix_blob_events(const NdbDictionary::Table* table, const char* ev_name);
 private:
   NdbIndexImpl * getIndexImpl(const char * name,
                               const BaseString& internalName);