ndb - bug#17929

  guess which scan type to use in handler


sql/ha_ndbcluster.cc:
  Add guessing of scan type when starting full-table-scan
storage/ndb/include/ndbapi/NdbDictionary.hpp:
  Add methods to check properties of a bitmap of columns
storage/ndb/include/ndbapi/NdbScanOperation.hpp:
  Add DiskScanFlag
storage/ndb/src/ndbapi/NdbDictionaryImpl.cpp:
  Add aggregate of #disk columns
storage/ndb/src/ndbapi/NdbDictionaryImpl.hpp:
  Add #disk columns
storage/ndb/src/ndbapi/NdbScanOperation.cpp:
  Add disk scan flag
storage/ndb/tools/delete_all.cpp:
  Add tupscan/diskscan to delete_all

sql/ha_ndbcluster.cc

@@ -2392,6 +2392,30 @@ int ha_ndbcluster::ordered_index_scan(const key_range *start_key,
   DBUG_RETURN(next_result(buf));
 }
 
+static
+int
+guess_scan_flags(NdbOperation::LockMode lm, 
+		 const NDBTAB* tab, const MY_BITMAP* readset)
+{
+  int flags= 0;
+  flags|= (lm == NdbOperation::LM_Read) ? NdbScanOperation::SF_KeyInfo : 0;
+  if (tab->checkColumns(0, 0) & 2)
+  {
+    int ret = tab->checkColumns(readset->bitmap, no_bytes_in_map(readset));
+    
+    if (ret & 2)
+    { // If disk columns...use disk scan
+      flags |= NdbScanOperation::SF_DiskScan;
+    }
+    else if ((ret & 4) == 0 && (lm == NdbOperation::LM_Exclusive))
+    {
+      // If no mem column is set and exclusive...guess disk scan
+      flags |= NdbScanOperation::SF_DiskScan;
+    }
+  }
+  return flags;
+}
+
 /*
   Start full table scan in NDB
  */
@@ -2409,11 +2433,9 @@ int ha_ndbcluster::full_table_scan(byte *buf)
 
   NdbOperation::LockMode lm=
     (NdbOperation::LockMode)get_ndb_lock_type(m_lock.type);
-  bool need_pk = (lm == NdbOperation::LM_Read);
+  int flags= guess_scan_flags(lm, m_table, table->read_set);
   if (!(op=trans->getNdbScanOperation(m_table)) ||
-      op->readTuples(lm, 
-		     (need_pk)?NdbScanOperation::SF_KeyInfo:0, 
-		     parallelism))
+      op->readTuples(lm, flags, parallelism))
     ERR_RETURN(trans->getNdbError());
   m_active_cursor= op;
 

storage/ndb/include/ndbapi/NdbDictionary.hpp

@@ -920,6 +920,16 @@ public:
 
     bool getTemporary();
     void setTemporary(bool); 
+
+    /**
+     * Check if any of column in bitmaps are disk columns
+     *   returns bitmap of different columns
+     *     bit 0 = atleast 1 pk column is set
+     *     bit 1 = atleast 1 disk column set
+     *     bit 2 = atleast 1 non disk column set
+     *   passing NULL pointer will equal to bitmap with all columns set
+     */
+    int checkColumns(const Uint32* bitmap, unsigned len_in_bytes) const;
 #endif
 
     // these 2 are not de-doxygenated

storage/ndb/include/ndbapi/NdbScanOperation.hpp

@@ -42,7 +42,8 @@ public:
    * readTuples.
    */
   enum ScanFlag {
-    SF_TupScan = (1 << 16),     // scan TUP
+    SF_TupScan = (1 << 16),     // scan TUP order
+    SF_DiskScan = (2 << 16),    // scan in DISK order
     SF_OrderBy = (1 << 24),     // index scan in order
     SF_Descending = (2 << 24),  // index scan in descending order
     SF_ReadRangeNo = (4 << 24), // enable @ref get_range_no

storage/ndb/src/ndbapi/NdbDictionaryImpl.cpp

@@ -773,6 +773,7 @@ NdbTableImpl::computeAggregates()
   m_keyLenInWords = 0;
   m_noOfDistributionKeys = 0;
   m_noOfBlobs = 0;
+  m_noOfDiskColumns = 0;
   Uint32 i, n;
   for (i = 0; i < m_columns.size(); i++) {
     NdbColumnImpl* col = m_columns[i];
@@ -785,6 +786,10 @@ NdbTableImpl::computeAggregates()
     
     if (col->getBlobType())
       m_noOfBlobs++;
+
+    if (col->getStorageType() == NdbDictionary::Column::StorageTypeDisk)
+      m_noOfDiskColumns++;
+    
     col->m_keyInfoPos = ~0;
   }
   if (m_noOfDistributionKeys == m_noOfKeys) {
@@ -1069,6 +1074,54 @@ NdbTableImpl::get_nodes(Uint32 hashValue, const Uint16 ** nodes) const
   return 0;
 }
 
+int
+NdbDictionary::Table::checkColumns(const Uint32* map, Uint32 len) const
+{
+  int ret = 0;
+  Uint32 colCnt = m_impl.m_columns.size();
+  if (map == 0)
+  {
+    ret |= 1;
+    ret |= (m_impl.m_noOfDiskColumns) ? 2 : 0;
+    ret |= (colCnt > m_impl.m_noOfDiskColumns) ? 4 : 0;
+    return ret;
+  }
+
+  NdbColumnImpl** cols = m_impl.m_columns.getBase();
+  const char * ptr = reinterpret_cast<const char*>(map);
+  const char * end = ptr + len;
+  Uint32 no = 0;
+  while (ptr < end)
+  {
+    Uint32 val = (Uint32)* ptr;
+    Uint32 idx = 1;
+    for (Uint32 i = 0; i<8; i++)
+    {
+      if (val & idx)
+      {
+	if (cols[no]->getPrimaryKey())
+	  ret |= 1;
+	else
+	{
+	  if (cols[no]->getStorageType() == NdbDictionary::Column::StorageTypeDisk)
+	    ret |= 2;
+	  else
+	    ret |= 4;
+	}
+      }
+      no ++;
+      idx *= 2; 
+      if (no == colCnt)
+	return ret;
+    }
+    
+    ptr++;
+  }
+  return ret;
+}
+
+
+  
 /**
  * NdbIndexImpl
  */

storage/ndb/src/ndbapi/NdbDictionaryImpl.hpp

@@ -225,7 +225,7 @@ public:
   // if all pk = dk then this is zero!
   Uint8 m_noOfDistributionKeys;
   Uint8 m_noOfBlobs;
-  
+  Uint8 m_noOfDiskColumns;
   Uint8 m_replicaCount;
 
   /**

storage/ndb/src/ndbapi/NdbScanOperation.cpp

@@ -174,6 +174,11 @@ NdbScanOperation::readTuples(NdbScanOperation::LockMode lm,
     }
   }
 #endif
+  if (scan_flags & SF_DiskScan)
+  {
+    tupScan = true;
+    m_no_disk_flag = false;
+  }
   
   bool rangeScan = false;
   if (m_accessTable->m_indexType == NdbDictionary::Index::OrderedIndex)

storage/ndb/tools/delete_all.cpp

@@ -29,6 +29,8 @@ NDB_STD_OPTS_VARS;
 
 static const char* _dbname = "TEST_DB";
 static my_bool _transactional = false;
+static my_bool _tupscan = 0;
+static my_bool _diskscan = 0;
 static struct my_option my_long_options[] =
 {
   NDB_STD_OPTS("ndb_desc"),
@@ -38,6 +40,12 @@ static struct my_option my_long_options[] =
   { "transactional", 't', "Single transaction (may run out of operations)",
     (gptr*) &_transactional, (gptr*) &_transactional, 0,
     GET_BOOL, NO_ARG, 0, 0, 0, 0, 0, 0 },
+  { "tupscan", 999, "Run tupscan",
+    (gptr*) &_tupscan, (gptr*) &_tupscan, 0,
+    GET_BOOL, NO_ARG, 0, 0, 0, 0, 0, 0 },
+  { "diskscan", 999, "Run diskcan",
+    (gptr*) &_diskscan, (gptr*) &_diskscan, 0,
+    GET_BOOL, NO_ARG, 0, 0, 0, 0, 0, 0 },
   { 0, 0, 0, 0, 0, 0, GET_NO_ARG, NO_ARG, 0, 0, 0, 0, 0, 0}
 };
 static void usage()
@@ -139,8 +147,11 @@ int clear_table(Ndb* pNdb, const NdbDictionary::Table* pTab,
       goto failed;
     }
     
+    int flags = 0;
+    flags |= _tupscan ? NdbScanOperation::SF_TupScan : 0;
+    flags |= _diskscan ? NdbScanOperation::SF_DiskScan : 0;
     if( pOp->readTuples(NdbOperation::LM_Exclusive, 
-			NdbScanOperation::SF_TupScan, par) ) {
+			flags, par) ) {
       goto failed;
     }