Fix for bug#17899 Partitions: crash, NDB, Select .. ORDER BY

mysql-test/r/ndb_partition_list.result

+drop table if exists t1;
+CREATE TABLE t1 ( f_int1 INTEGER NOT NULL, f_int2 INTEGER NOT NULL, 
+f_char1 CHAR(10),
+f_char2 CHAR(10), f_charbig VARCHAR(1000),
+PRIMARY KEY (f_int1,f_int2))
+PARTITION BY LIST(MOD(f_int1 + f_int2,4)) 
+(PARTITION part_3 VALUES IN (-3),
+PARTITION part_2 VALUES IN (-2),
+PARTITION part_1 VALUES IN (-1),
+PARTITION part0 VALUES IN (0),
+PARTITION part1 VALUES IN (1),
+PARTITION part2 VALUES IN (2),
+PARTITION part3 VALUES IN (3,4,5));
+INSERT INTO t1 SET f_int1 = -2, f_int2 = 20, f_char1 = '20', f_char2 = '20', f_charbig = '===20===';
+INSERT INTO t1 SET f_int1 = 1, f_int2 = 1, f_char1 = '1', f_char2 = '1', f_charbig = '===1===';
+INSERT INTO t1 SET f_int1 = 2, f_int2 = 1, f_char1 = '1', f_char2 = '1', f_charbig = '===1===';
+INSERT INTO t1 SET f_int1 = 3, f_int2 = 1, f_char1 = '1', f_char2 = '1', f_charbig = '===1===';
+INSERT INTO t1 SET f_int1 = 4, f_int2 = 1, f_char1 = '1', f_char2 = '1', f_charbig = '===1===';
+INSERT INTO t1 SET f_int1 = 5, f_int2 = 1, f_char1 = '1', f_char2 = '1', f_charbig = '===1===';
+INSERT INTO t1 SET f_int1 = 20, f_int2 = 1, f_char1 = '1', f_char2 = '1', f_charbig = '===1===';
+SELECT * FROM t1 ORDER BY f_int1;
+f_int1	f_int2	f_char1	f_char2	f_charbig
+-2	20	20	20	===20===
+1	1	1	1	===1===
+2	1	1	1	===1===
+3	1	1	1	===1===
+4	1	1	1	===1===
+5	1	1	1	===1===
+20	1	1	1	===1===
+DROP TABLE t1;
+CREATE TABLE t1 ( f_int1 INTEGER, f_int2 INTEGER, f_char1 CHAR(10), 
+f_char2 CHAR(10), f_charbig VARCHAR(1000))
+PARTITION BY LIST(f_int1) 
+(PARTITION part_1 VALUES IN (-1),
+PARTITION part0 VALUES IN (0,1),
+PARTITION part1 VALUES IN (2));
+INSERT INTO t1 SET f_int1 = -1, f_int2 = 20, f_char1 = '20', f_char2 = '20', f_charbig = '===20===';
+INSERT INTO t1 SET f_int1 = 0, f_int2 = 20, f_char1 = '20', f_char2 = '20', f_charbig = '===20===';
+INSERT INTO t1 SET f_int1 = 1, f_int2 = 1, f_char1 = '1', f_char2 = '1', f_charbig = '===1===';
+INSERT INTO t1 SET f_int1 = 2, f_int2 = 1, f_char1 = '1', f_char2 = '1', f_charbig = '===1===';
+INSERT INTO t1 SET f_int1 = 20, f_int2 = 1, f_char1 = '1', f_char2 = '1', f_charbig = '===1===';
+ERROR HY000: Table has no partition for value 20
+SELECT * FROM t1 ORDER BY f_int1;
+f_int1	f_int2	f_char1	f_char2	f_charbig
+-1	20	20	20	===20===
+0	20	20	20	===20===
+1	1	1	1	===1===
+2	1	1	1	===1===
+DROP TABLE t1;

mysql-test/t/ndb_partition_list.test

+--source include/have_ndb.inc
+#
+# Simple test for the partition storage engine
+# Focuses on range partitioning tests
+# 
+#-- source include/have_partition.inc
+
+--disable_warnings
+drop table if exists t1;
+--enable_warnings
+
+#
+# Partition by list, basic
+#
+
+CREATE TABLE t1 ( f_int1 INTEGER NOT NULL, f_int2 INTEGER NOT NULL, 
+	          f_char1 CHAR(10),
+                  f_char2 CHAR(10), f_charbig VARCHAR(1000),
+PRIMARY KEY (f_int1,f_int2))
+PARTITION BY LIST(MOD(f_int1 + f_int2,4)) 
+(PARTITION part_3 VALUES IN (-3),
+ PARTITION part_2 VALUES IN (-2),
+ PARTITION part_1 VALUES IN (-1),
+ PARTITION part0 VALUES IN (0),
+ PARTITION part1 VALUES IN (1),
+ PARTITION part2 VALUES IN (2),
+ PARTITION part3 VALUES IN (3,4,5));
+
+INSERT INTO t1 SET f_int1 = -2, f_int2 = 20, f_char1 = '20', f_char2 = '20', f_charbig = '===20===';
+INSERT INTO t1 SET f_int1 = 1, f_int2 = 1, f_char1 = '1', f_char2 = '1', f_charbig = '===1===';
+INSERT INTO t1 SET f_int1 = 2, f_int2 = 1, f_char1 = '1', f_char2 = '1', f_charbig = '===1===';
+INSERT INTO t1 SET f_int1 = 3, f_int2 = 1, f_char1 = '1', f_char2 = '1', f_charbig = '===1===';
+INSERT INTO t1 SET f_int1 = 4, f_int2 = 1, f_char1 = '1', f_char2 = '1', f_charbig = '===1===';
+INSERT INTO t1 SET f_int1 = 5, f_int2 = 1, f_char1 = '1', f_char2 = '1', f_charbig = '===1===';
+INSERT INTO t1 SET f_int1 = 20, f_int2 = 1, f_char1 = '1', f_char2 = '1', f_charbig = '===1===';
+
+SELECT * FROM t1 ORDER BY f_int1;
+
+DROP TABLE t1;
+
+#
+# Partition by list, no pk
+#
+
+CREATE TABLE t1 ( f_int1 INTEGER, f_int2 INTEGER, f_char1 CHAR(10), 
+                  f_char2 CHAR(10), f_charbig VARCHAR(1000))
+PARTITION BY LIST(f_int1) 
+(PARTITION part_1 VALUES IN (-1),
+ PARTITION part0 VALUES IN (0,1),
+ PARTITION part1 VALUES IN (2));
+
+INSERT INTO t1 SET f_int1 = -1, f_int2 = 20, f_char1 = '20', f_char2 = '20', f_charbig = '===20===';
+INSERT INTO t1 SET f_int1 = 0, f_int2 = 20, f_char1 = '20', f_char2 = '20', f_charbig = '===20===';
+INSERT INTO t1 SET f_int1 = 1, f_int2 = 1, f_char1 = '1', f_char2 = '1', f_charbig = '===1===';
+INSERT INTO t1 SET f_int1 = 2, f_int2 = 1, f_char1 = '1', f_char2 = '1', f_charbig = '===1===';
+--error 1504
+INSERT INTO t1 SET f_int1 = 20, f_int2 = 1, f_char1 = '1', f_char2 = '1', f_charbig = '===1===';
+
+SELECT * FROM t1 ORDER BY f_int1;
+
+DROP TABLE t1;
+

sql/ha_ndbcluster.cc

@@ -954,6 +954,19 @@ int ha_ndbcluster::get_ndb_value(NdbOperation *ndb_op, Field *field,
   DBUG_RETURN(m_value[fieldnr].rec == NULL);
 }
 
+/*
+  Instruct NDB to fetch the partition id (fragment id)
+*/
+int ha_ndbcluster::get_ndb_partition_id(NdbOperation *ndb_op)
+{
+  DBUG_ENTER("get_ndb_partition_id");
+  uint partition_id_fieldnr= table_share->fields + 1;
+
+  m_value[partition_id_fieldnr].rec= 
+    ndb_op->getValue(NdbDictionary::Column::FRAGMENT, 
+                     (char *)&m_part_id);
+  DBUG_RETURN(m_value[partition_id_fieldnr].rec == NULL);
+}
 
 /*
   Check if any set or get of blob value in current query.
@@ -1646,8 +1659,6 @@ int ha_ndbcluster::pk_read(const byte *key, uint key_len, byte *buf,
       op->readTuple(lm) != 0)
     ERR_RETURN(trans->getNdbError());
   
-  if (m_use_partition_function)
-    op->setPartitionId(part_id);
   if (table_share->primary_key == MAX_KEY) 
   {
     // This table has no primary key, use "hidden" primary key
@@ -1668,7 +1679,18 @@ int ha_ndbcluster::pk_read(const byte *key, uint key_len, byte *buf,
   
   if ((res= define_read_attrs(buf, op)))
     DBUG_RETURN(res);
-  
+
+  if (m_use_partition_function)
+  {
+    op->setPartitionId(part_id);
+    // If table has user defined partitioning
+    // and no indexes, we need to read the partition id
+    // to support ORDER BY queries
+    if (table_share->primary_key == MAX_KEY &&
+        get_ndb_partition_id(op))
+      ERR_RETURN(trans->getNdbError());
+  }
+
   if (execute_no_commit_ie(this,trans) != 0) 
   {
     table->status= STATUS_NOT_FOUND;
@@ -2187,12 +2209,23 @@ int ha_ndbcluster::ordered_index_scan(const key_range *start_key,
       DBUG_RETURN(res);
   }
 
-  if (!restart && generate_scan_filter(m_cond_stack, op))
-    DBUG_RETURN(ndb_err(trans));
-  
-  if (!restart && (res= define_read_attrs(buf, op)))
+  if (!restart)
   {
-    DBUG_RETURN(res);
+    if (generate_scan_filter(m_cond_stack, op))
+      DBUG_RETURN(ndb_err(trans));
+
+    if (res= define_read_attrs(buf, op))
+    {
+      DBUG_RETURN(res);
+    }
+    
+    // If table has user defined partitioning
+    // and no primary key, we need to read the partition id
+    // to support ORDER BY queries
+    if (m_use_partition_function &&
+        (table_share->primary_key == MAX_KEY) && 
+        (get_ndb_partition_id(op)))
+      ERR_RETURN(trans->getNdbError());
   }
 
   if (execute_no_commit(this,trans) != 0)
@@ -2249,6 +2282,12 @@ int ha_ndbcluster::full_table_scan(byte *buf)
       */
       m_active_cursor->setPartitionId(part_spec.start_part);
     }
+    // If table has user defined partitioning
+    // and no primary key, we need to read the partition id
+    // to support ORDER BY queries
+    if ((table_share->primary_key == MAX_KEY) && 
+        (get_ndb_partition_id(op)))
+      ERR_RETURN(trans->getNdbError());
   }
 
   if (generate_scan_filter(m_cond_stack, op))
@@ -3217,17 +3256,32 @@ int ha_ndbcluster::rnd_pos(byte *buf, byte *pos)
   // Perform a pk_read using primary key "index"
   {
     part_id_range part_spec;
+    uint key_length= ref_length;
     if (m_use_partition_function)
     {
-      key_range key_spec;
-      KEY *key_info= table->key_info + active_index;
-      key_spec.key= pos;
-      key_spec.length= ref_length;
-      key_spec.flag= HA_READ_KEY_EXACT;
-      get_full_part_id_from_key(table, buf, key_info, &key_spec, &part_spec);
-      DBUG_ASSERT(part_spec.start_part == part_spec.end_part);
+      if (table_share->primary_key == MAX_KEY)
+      {
+        /*
+          The partition id has been fetched from ndb
+          and has been stored directly after the hidden key
+        */
+        key_length= ref_length - sizeof(m_part_id);
+        part_spec.start_part= part_spec.end_part= *(pos + key_length);
+      }
+      else
+      {
+        key_range key_spec;
+        KEY *key_info= table->key_info + active_index;
+        key_spec.key= pos;
+        key_spec.length= key_length;
+        key_spec.flag= HA_READ_KEY_EXACT;
+        get_full_part_id_from_key(table, buf, key_info, 
+                                  &key_spec, &part_spec);
+        DBUG_ASSERT(part_spec.start_part == part_spec.end_part);
+      }
+      DBUG_PRINT("info", ("partition id %u", part_spec.start_part));
     }
-    DBUG_RETURN(pk_read(pos, ref_length, buf, part_spec.start_part));
+    DBUG_RETURN(pk_read(pos, key_length, buf, part_spec.start_part));
   }
 }
 
@@ -3244,6 +3298,7 @@ void ha_ndbcluster::position(const byte *record)
   KEY_PART_INFO *key_part;
   KEY_PART_INFO *end;
   byte *buff;
+  uint key_length= ref_length;
   DBUG_ENTER("position");
 
   if (table_share->primary_key != MAX_KEY) 
@@ -3290,18 +3345,25 @@ void ha_ndbcluster::position(const byte *record)
   {
     // No primary key, get hidden key
     DBUG_PRINT("info", ("Getting hidden key"));
+    // If table has user defined partition save the partition id as well
+    if(m_use_partition_function)
+    {
+      DBUG_PRINT("info", ("Saving partition id %u in %u", m_part_id));
+      key_length= ref_length - sizeof(m_part_id);
+      memcpy(ref+key_length, (void *)&m_part_id, sizeof(m_part_id));
+    }
 #ifndef DBUG_OFF
     int hidden_no= table->s->fields;
     const NDBTAB *tab= (const NDBTAB *) m_table;  
     const NDBCOL *hidden_col= tab->getColumn(hidden_no);
     DBUG_ASSERT(hidden_col->getPrimaryKey() && 
                 hidden_col->getAutoIncrement() &&
-                ref_length == NDB_HIDDEN_PRIMARY_KEY_LENGTH);
+                key_length == NDB_HIDDEN_PRIMARY_KEY_LENGTH);
 #endif
-    memcpy(ref, m_ref, ref_length);
+    memcpy(ref, m_ref, key_length);
   }
   
-  DBUG_DUMP("ref", (char*)ref, ref_length);
+  DBUG_DUMP("ref", (char*)ref, key_length);
   DBUG_VOID_RETURN;
 }
 
@@ -5173,8 +5235,17 @@ int ha_ndbcluster::open(const char *name, int mode, uint test_if_locked)
   {
     key= table->key_info+table_share->primary_key;
     ref_length= key->key_length;
-    DBUG_PRINT("info", (" ref_length: %d", ref_length));
   }
+  else // (table_share->primary_key == MAX_KEY) 
+  {
+    if (m_use_partition_function)
+    {
+      ref_length+= sizeof(m_part_id);
+    }
+  }
+
+  DBUG_PRINT("info", ("ref_length: %d", ref_length));
+
   // Init table lock structure 
   if (!(m_share=get_share(name, table)))
     DBUG_RETURN(1);

sql/ha_ndbcluster.h

@@ -761,6 +761,7 @@ private:
   int set_ndb_value(NdbOperation*, Field *field, uint fieldnr,
 		    int row_offset= 0, bool *set_blob_value= 0);
   int get_ndb_value(NdbOperation*, Field *field, uint fieldnr, byte*);
+  int ha_ndbcluster::get_ndb_partition_id(NdbOperation *);
   friend int g_get_ndb_blobs_value(NdbBlob *ndb_blob, void *arg);
   int get_ndb_blobs_value(NdbBlob *last_ndb_blob);
   int set_primary_key(NdbOperation *op, const byte *key);
@@ -824,6 +825,7 @@ private:
   NdbValue m_value[NDB_MAX_ATTRIBUTES_IN_TABLE];
   byte m_ref[NDB_HIDDEN_PRIMARY_KEY_LENGTH];
   partition_info *m_part_info;
+  uint32 m_part_id;
   byte *m_rec0;
   Field **m_part_field_array;
   bool m_use_partition_function;