Merge sama.ndb.mysql.com:/export/space/pekka/ndb/version/my50-ndb

into  sama.ndb.mysql.com:/export/space/pekka/ndb/version/my51-bug29390


sql/ha_ndbcluster_cond.cc:
  Auto merged
mysql-test/suite/ndb/r/ndb_condition_pushdown.result:
  Auto merged
mysql-test/suite/ndb/t/ndb_condition_pushdown.test:
  Auto merged
storage/ndb/include/kernel/signaldata/ScanTab.hpp:
  Auto merged
storage/ndb/include/ndbapi/NdbScanFilter.hpp:
  Auto merged
storage/ndb/include/ndbapi/ndbapi_limits.h:
  Auto merged
storage/ndb/src/ndbapi/NdbScanFilter.cpp:
  Auto merged
storage/ndb/src/ndbapi/NdbScanOperation.cpp:
  Auto merged
storage/ndb/src/ndbapi/ndberror.c:
  ul, add code (4294) in post-merge
storage/ndb/include/ndbapi/Ndb.hpp:
  SCCS merged

mysql-test/suite/ndb/r/ndb_condition_pushdown.result

@@ -1888,5 +1888,27 @@ set engine_condition_pushdown = 1;
 SELECT fname, lname FROM t1 WHERE (fname like 'Y%') or (lname like 'F%');
 fname	lname
 Young	Foo
+drop table t1;
+create table t1 (a int, b int, c int, d int, primary key using hash(a))
+engine=ndbcluster;
+insert into t1 values (10,1,100,0+0x1111);
+insert into t1 values (20,2,200,0+0x2222);
+insert into t1 values (30,3,300,0+0x3333);
+insert into t1 values (40,4,400,0+0x4444);
+insert into t1 values (50,5,500,0+0x5555);
+set engine_condition_pushdown = on;
+select a,b,d from t1
+where b in (0,1,2,5)
+order by b;
+a	b	d
+10	1	4369
+20	2	8738
+50	5	21845
+a	b	d
+10	1	4369
+20	2	8738
+50	5	21845
+Warnings:
+Warning	4294	Scan filter is too large, discarded
 set engine_condition_pushdown = @old_ecpd;
 DROP TABLE t1,t2,t3,t4,t5;

sql/ha_ndbcluster_cond.cc

@@ -1338,9 +1338,23 @@ ha_ndbcluster_cond::generate_scan_filter(NdbScanOperation *op)
 
   if (m_cond_stack)
   {
-    NdbScanFilter filter(op);
+    NdbScanFilter filter(op, false); // don't abort on too large
     
-    DBUG_RETURN(generate_scan_filter_from_cond(filter));
+    int ret=generate_scan_filter_from_cond(filter);
+    if (ret != 0)
+    {
+      const NdbError& err=filter.getNdbError();
+      if (err.code == NdbScanFilter::FilterTooLarge)
+      {
+        // err.message has static storage
+        DBUG_PRINT("info", ("%s", err.message));
+        push_warning(current_thd, MYSQL_ERROR::WARN_LEVEL_WARN,
+                     err.code, err.message);
+        ret=0;
+      }
+    }
+    if (ret != 0)
+      DBUG_RETURN(ret);
   }
   else
   {  
@@ -1391,7 +1405,7 @@ int ha_ndbcluster_cond::generate_scan_filter_from_key(NdbScanOperation *op,
 {
   KEY_PART_INFO* key_part= key_info->key_part;
   KEY_PART_INFO* end= key_part+key_info->key_parts;
-  NdbScanFilter filter(op);
+  NdbScanFilter filter(op, true); // abort on too large
   int res;
   DBUG_ENTER("generate_scan_filter_from_key");
 

storage/ndb/include/kernel/signaldata/ScanTab.hpp

@@ -46,6 +46,7 @@ public:
    * Length of signal
    */
   STATIC_CONST( StaticLength = 11 );
+  STATIC_CONST( MaxTotalAttrInfo = 0xFFFF );
 
 private:
 

storage/ndb/include/ndbapi/Ndb.hpp

@@ -1056,6 +1056,7 @@ class Ndb
   friend class NdbBlob;
   friend class NdbImpl;
   friend class Ndb_internal;
+  friend class NdbScanFilterImpl;
 #endif
 
 public:

storage/ndb/include/ndbapi/NdbScanFilter.hpp

@@ -17,6 +17,7 @@
 #define NDB_SCAN_FILTER_HPP
 
 #include <ndb_types.h>
+#include <ndbapi_limits.h>
 
 /**
  * @class NdbScanFilter
@@ -31,8 +32,13 @@ public:
   /**
    * Constructor
    * @param op  The NdbOperation that the filter belongs to (is applied to).
+   * @param abort_on_too_large  abort transaction on filter too large
+   *                            default: true
+   * @param max_size  Maximum size of generated filter in words
    */
-  NdbScanFilter(class NdbOperation * op);
+  NdbScanFilter(class NdbOperation * op,
+                bool abort_on_too_large = true,
+                Uint32 max_size = NDB_MAX_SCANFILTER_SIZE_IN_WORDS);
   ~NdbScanFilter();
   
   /**
@@ -166,6 +172,25 @@ public:
   /** @} *********************************************************************/
 #endif
 
+  enum Error {
+    FilterTooLarge = 4294
+  };
+
+  /**
+   * Get filter level error.
+   *
+   * Most errors are set only on operation level, and they abort the
+   * transaction.  The error FilterTooLarge is set on filter level and
+   * by default it propagates to operation level and also aborts the
+   * transaction.
+   *
+   * If option abort_on_too_large is set to false, then FilterTooLarge
+   * does not propagate.  One can then either ignore this error (in
+   * which case no filtering is done) or try to define a new filter
+   * immediately.
+   */
+  const class NdbError & getNdbError() const;
+
 private:
 #ifndef DOXYGEN_SHOULD_SKIP_INTERNAL
   friend class NdbScanFilterImpl;

storage/ndb/include/ndbapi/ndbapi_limits.h

@@ -26,4 +26,6 @@
 #define NDB_MAX_TUPLE_SIZE (NDB_MAX_TUPLE_SIZE_IN_WORDS*4)
 #define NDB_MAX_ACTIVE_EVENTS 100
 
+#define NDB_MAX_SCANFILTER_SIZE_IN_WORDS 50000
+
 #endif

storage/ndb/src/ndbapi/NdbScanFilter.cpp

@@ -14,11 +14,14 @@
    Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA */
 
 #include <NdbScanFilter.hpp>
+#include <Ndb.hpp>
 #include <NdbOperation.hpp>
 #include "NdbDictionaryImpl.hpp"
 #include <Vector.hpp>
 #include <NdbOut.hpp>
 #include <Interpreter.hpp>
+#include <signaldata/AttrInfo.hpp>
+#include "NdbApiSignal.hpp"
 
 #ifdef VM_TRACE
 #include <NdbEnv.h>
@@ -52,14 +55,37 @@ public:
   
   int cond_col_const(Interpreter::BinaryCondition, Uint32 attrId, 
 		     const void * value, Uint32 len);
+
+  bool m_abort_on_too_large;
+
+  NdbOperation::OperationStatus m_initial_op_status;
+  Uint32 m_initial_AI_size;
+  Uint32 m_max_size;
+
+  Uint32 get_size() {
+    assert(m_operation->theTotalCurrAI_Len >= m_initial_AI_size);
+    return m_operation->theTotalCurrAI_Len - m_initial_AI_size;
+  }
+  bool check_size() {
+    if (get_size() <= m_max_size)
+      return true;
+    handle_filter_too_large();
+    return false;
+  }
+  void handle_filter_too_large();
+
+  NdbError m_error;
 };
 
 const Uint32 LabelExit = ~0;
 
 
-NdbScanFilter::NdbScanFilter(class NdbOperation * op)
+NdbScanFilter::NdbScanFilter(class NdbOperation * op,
+                             bool abort_on_too_large,
+                             Uint32 max_size)
   : m_impl(* new NdbScanFilterImpl())
 {
+  DBUG_ENTER("NdbScanFilter::NdbScanFilter");
   m_impl.m_current.m_group = (NdbScanFilter::Group)0;
   m_impl.m_current.m_popCount = 0;
   m_impl.m_current.m_ownLabel = 0;
@@ -69,6 +95,21 @@ NdbScanFilter::NdbScanFilter(class NdbOperation * op)
   m_impl.m_latestAttrib = ~0;
   m_impl.m_operation = op;
   m_impl.m_negative = 0;
+
+  DBUG_PRINT("info", ("op status: %d tot AI: %u in curr: %u",
+                      op->theStatus,
+                      op->theTotalCurrAI_Len, op->theAI_LenInCurrAI));
+
+  m_impl.m_abort_on_too_large = abort_on_too_large;
+
+  m_impl.m_initial_op_status = op->theStatus;
+  m_impl.m_initial_AI_size = op->theTotalCurrAI_Len;
+  if (max_size > NDB_MAX_SCANFILTER_SIZE_IN_WORDS)
+    max_size = NDB_MAX_SCANFILTER_SIZE_IN_WORDS;
+  m_impl.m_max_size = max_size;
+
+  m_impl.m_error.code = 0;
+  DBUG_VOID_RETURN;
 }
 
 NdbScanFilter::~NdbScanFilter(){
@@ -200,30 +241,38 @@ NdbScanFilter::end(){
   switch(tmp.m_group){
   case NdbScanFilter::AND:
     if(tmp.m_trueLabel == (Uint32)~0){
-      m_impl.m_operation->interpret_exit_ok();
+      if (m_impl.m_operation->interpret_exit_ok() == -1)
+        return -1;
     } else {
-      m_impl.m_operation->branch_label(tmp.m_trueLabel);
+      if (m_impl.m_operation->branch_label(tmp.m_trueLabel) == -1)
+        return -1;
     }
     break;
   case NdbScanFilter::NAND:
     if(tmp.m_trueLabel == (Uint32)~0){
-      m_impl.m_operation->interpret_exit_nok();
+      if (m_impl.m_operation->interpret_exit_nok() == -1)
+        return -1;
     } else {
-      m_impl.m_operation->branch_label(tmp.m_falseLabel);
+      if (m_impl.m_operation->branch_label(tmp.m_falseLabel) == -1)
+        return -1;
     }
     break;
   case NdbScanFilter::OR:
     if(tmp.m_falseLabel == (Uint32)~0){
-      m_impl.m_operation->interpret_exit_nok();
+      if (m_impl.m_operation->interpret_exit_nok() == -1)
+        return -1;
     } else {
-      m_impl.m_operation->branch_label(tmp.m_falseLabel);
+      if (m_impl.m_operation->branch_label(tmp.m_falseLabel) == -1)
+        return -1;
     }
     break;
   case NdbScanFilter::NOR:
     if(tmp.m_falseLabel == (Uint32)~0){
-      m_impl.m_operation->interpret_exit_ok();
+      if (m_impl.m_operation->interpret_exit_ok() == -1)
+        return -1;
     } else {
-      m_impl.m_operation->branch_label(tmp.m_trueLabel);
+      if (m_impl.m_operation->branch_label(tmp.m_trueLabel) == -1)
+        return -1;
     }
     break;
   default:
@@ -231,24 +280,29 @@ NdbScanFilter::end(){
     return -1;
   }
 
-  m_impl.m_operation->def_label(tmp.m_ownLabel);
+  if (m_impl.m_operation->def_label(tmp.m_ownLabel) == -1)
+    return -1;
 
   if(m_impl.m_stack.size() == 0){
     switch(tmp.m_group){
     case NdbScanFilter::AND:
     case NdbScanFilter::NOR:
-      m_impl.m_operation->interpret_exit_nok();
+      if (m_impl.m_operation->interpret_exit_nok() == -1)
+        return -1;
       break;
     case NdbScanFilter::OR:
     case NdbScanFilter::NAND:
-      m_impl.m_operation->interpret_exit_ok();
+      if (m_impl.m_operation->interpret_exit_ok() == -1)
+        return -1;
       break;
     default:
       m_impl.m_operation->setErrorCodeAbort(4260);
       return -1;
     }
   }
-  
+
+  if (!m_impl.check_size())
+    return -1;
   return 0;
 }
 
@@ -261,10 +315,16 @@ NdbScanFilter::istrue(){
   }
 
   if(m_impl.m_current.m_trueLabel == (Uint32)~0){
-    return m_impl.m_operation->interpret_exit_ok();
+    if (m_impl.m_operation->interpret_exit_ok() == -1)
+      return -1;
   } else {
-    return m_impl.m_operation->branch_label(m_impl.m_current.m_trueLabel);
+    if (m_impl.m_operation->branch_label(m_impl.m_current.m_trueLabel) == -1)
+      return -1;
   }
+
+  if (!m_impl.check_size())
+    return -1;
+  return 0;
 }
 
 int
@@ -276,10 +336,16 @@ NdbScanFilter::isfalse(){
   }
   
   if(m_impl.m_current.m_falseLabel == (Uint32)~0){
-    return m_impl.m_operation->interpret_exit_nok();
+    if (m_impl.m_operation->interpret_exit_nok() == -1)
+      return -1;
   } else {
-    return m_impl.m_operation->branch_label(m_impl.m_current.m_falseLabel);
+    if (m_impl.m_operation->branch_label(m_impl.m_current.m_falseLabel) == -1)
+      return -1;
   }
+
+  if (!m_impl.check_size())
+    return -1;
+  return 0;
 }
 
 
@@ -330,7 +396,11 @@ NdbScanFilterImpl::cond_col(Interpreter::UnaryCondition op, Uint32 AttrId){
   }
   
   Branch1 branch = table2[op].m_branches[m_current.m_group];
-  (m_operation->* branch)(AttrId, m_current.m_ownLabel);
+  if ((m_operation->* branch)(AttrId, m_current.m_ownLabel) == -1)
+    return -1;
+
+  if (!check_size())
+    return -1;
   return 0;
 }
 
@@ -463,8 +533,12 @@ NdbScanFilterImpl::cond_col_const(Interpreter::BinaryCondition op,
     return -1;
   }
   
-  int ret = (m_operation->* branch)(AttrId, value, len, false, m_current.m_ownLabel);
-  return ret;
+  if ((m_operation->* branch)(AttrId, value, len, false, m_current.m_ownLabel) == -1)
+    return -1;
+
+  if (!check_size())
+    return -1;
+  return 0;
 }
 
 int
@@ -490,7 +564,99 @@ NdbScanFilter::cmp(BinaryCondition cond, int ColId,
     return m_impl.cond_col_const(Interpreter::NOT_LIKE, ColId, val, len);
   }
   return -1;
-} 
+}
+
+void
+NdbScanFilterImpl::handle_filter_too_large()
+{
+  DBUG_ENTER("NdbScanFilterImpl::handle_filter_too_large");
+
+  NdbOperation* const op = m_operation;
+  m_error.code = NdbScanFilter::FilterTooLarge;
+  if (m_abort_on_too_large)
+    op->setErrorCodeAbort(m_error.code);
+
+  /*
+   * Possible interpreted parts at this point are:
+   *
+   * 1. initial read
+   * 2. interpreted program
+   *
+   * It is assumed that NdbScanFilter has created all of 2
+   * so that we don't have to save interpreter state.
+   */
+
+  const Uint32 size = get_size();
+  assert(size != 0);
+
+  // new ATTRINFO size
+  const Uint32 new_size = m_initial_AI_size;
+
+  // find last signal for new size
+  assert(op->theFirstATTRINFO != NULL);
+  NdbApiSignal* lastSignal = op->theFirstATTRINFO;
+  Uint32 n = 0;
+  while (n + AttrInfo::DataLength < new_size) {
+    lastSignal = lastSignal->next();
+    assert(lastSignal != NULL);
+    n += AttrInfo::DataLength;
+  }
+  assert(n < size);
+
+  // release remaining signals
+  NdbApiSignal* tSignal = lastSignal->next();
+  op->theNdb->releaseSignalsInList(&tSignal);
+  lastSignal->next(NULL);
+
+  // length of lastSignal
+  const Uint32 new_curr = AttrInfo::HeaderLength + new_size - n;
+  assert(new_curr <= 25);
+
+  DBUG_PRINT("info", ("op status: %d->%d tot AI: %u->%u in curr: %u->%u",
+                      op->theStatus, m_initial_op_status,
+                      op->theTotalCurrAI_Len, new_size,
+                      op->theAI_LenInCurrAI, new_curr));
+
+  // reset op state
+  op->theStatus = m_initial_op_status;
+
+  // reset interpreter state to initial
+  op->theFirstBranch = NULL;
+  op->theLastBranch = NULL;
+  op->theFirstCall = NULL;
+  op->theLastCall = NULL;
+  op->theFirstSubroutine = NULL;
+  op->theLastSubroutine = NULL;
+  op->theNoOfLabels = 0;
+  op->theNoOfSubroutines = 0;
+
+  // reset AI size
+  op->theTotalCurrAI_Len = new_size;
+  op->theAI_LenInCurrAI = new_curr;
+
+  // reset signal pointers
+  op->theCurrentATTRINFO = lastSignal;
+  op->theATTRINFOptr = &lastSignal->getDataPtrSend()[new_curr];
+
+  // interpreter sizes are set later somewhere
+
+  DBUG_VOID_RETURN;
+}
+
+static void
+update(const NdbError & _err){
+  NdbError & error = (NdbError &) _err;
+  ndberror_struct ndberror = (ndberror_struct)error;
+  ndberror_update(&ndberror);
+  error = NdbError(ndberror);
+}
+
+const NdbError &
+NdbScanFilter::getNdbError() const
+{
+  update(m_impl.m_error);
+  return m_impl.m_error;
+}
 
 
 #if 0

storage/ndb/src/ndbapi/NdbScanOperation.cpp

@@ -872,6 +872,10 @@ NdbScanOperation::doSendScan(int aProcessorId)
   // sending it. This could not be done in openScan because 
   // we created the ATTRINFO signals after the SCAN_TABREQ signal.
   ScanTabReq * const req = CAST_PTR(ScanTabReq, tSignal->getDataPtrSend());
+  if (unlikely(theTotalCurrAI_Len > ScanTabReq::MaxTotalAttrInfo)) {
+    setErrorCode(4257);
+    return -1;
+  }
   req->attrLenKeyLen = (tupKeyLen << 16) | theTotalCurrAI_Len;
   Uint32 tmp = req->requestInfo;
   ScanTabReq::setDistributionKeyFlag(tmp, theDistrKeyIndicator_);