ndb - bug#29390: if ScanFilter is too large, abort or optionally discard it

mysql-test/r/ndb_condition_pushdown.result:
  if ScanFilter is too large, abort or optionaly discard it
mysql-test/t/ndb_condition_pushdown.test:
  if ScanFilter is too large, abort or optionaly discard it
ndb/include/kernel/signaldata/ScanTab.hpp:
  if ScanFilter is too large, abort or optionaly discard it
ndb/include/ndbapi/Ndb.hpp:
  if ScanFilter is too large, abort or optionaly discard it
ndb/include/ndbapi/NdbScanFilter.hpp:
  if ScanFilter is too large, abort or optionaly discard it
ndb/include/ndbapi/ndbapi_limits.h:
  if ScanFilter is too large, abort or optionaly discard it
ndb/src/ndbapi/NdbScanFilter.cpp:
  if ScanFilter is too large, abort or optionaly discard it
ndb/src/ndbapi/NdbScanOperation.cpp:
  if ScanFilter is too large, abort or optionaly discard it
ndb/src/ndbapi/ndberror.c:
  if ScanFilter is too large, abort or optionaly discard it
sql/ha_ndbcluster_cond.cc:
  if ScanFilter is too large, abort or optionaly discard it

7 files changed, 223 insertions, 23 deletions

diff --git a/ndb/include/kernel/signaldata/ScanTab.hpp b/ndb/include/kernel/signaldata/ScanTab.hpp
index 15a022e2cba..38ec4cccf7b 100644
--- a/ndb/include/kernel/signaldata/ScanTab.hpp
+++ b/ndb/include/kernel/signaldata/ScanTab.hpp
@@ -46,6 +46,7 @@ public:
    * Length of signal
    */
   STATIC_CONST( StaticLength = 11 );
+  STATIC_CONST( MaxTotalAttrInfo = 0xFFFF );
 
 private:
 
diff --git a/ndb/include/ndbapi/Ndb.hpp b/ndb/include/ndbapi/Ndb.hpp
index f83db77739e..01bc899b4e1 100644
--- a/ndb/include/ndbapi/Ndb.hpp
+++ b/ndb/include/ndbapi/Ndb.hpp
@@ -1052,6 +1052,7 @@ class Ndb
   friend class NdbDictInterface;
   friend class NdbBlob;
   friend class NdbImpl;
+  friend class NdbScanFilterImpl;
 #endif
 
 public:
diff --git a/ndb/include/ndbapi/NdbScanFilter.hpp b/ndb/include/ndbapi/NdbScanFilter.hpp
index 1ef62558560..02fcb6215ba 100644
--- a/ndb/include/ndbapi/NdbScanFilter.hpp
+++ b/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;
diff --git a/ndb/include/ndbapi/ndbapi_limits.h b/ndb/include/ndbapi/ndbapi_limits.h
index 63399e4bd0a..e283913d059 100644
--- a/ndb/include/ndbapi/ndbapi_limits.h
+++ b/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
diff --git a/ndb/src/ndbapi/NdbScanFilter.cpp b/ndb/src/ndbapi/NdbScanFilter.cpp
index fb47772fdea..624122b5c55 100644
--- a/ndb/src/ndbapi/NdbScanFilter.cpp
+++ b/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
diff --git a/ndb/src/ndbapi/NdbScanOperation.cpp b/ndb/src/ndbapi/NdbScanOperation.cpp
index aec98a7f5d5..9176fb47297 100644
--- a/ndb/src/ndbapi/NdbScanOperation.cpp
+++ b/ndb/src/ndbapi/NdbScanOperation.cpp
@@ -849,6 +849,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_);
diff --git a/ndb/src/ndbapi/ndberror.c b/ndb/src/ndbapi/ndberror.c
index 24ccb1d07c2..56eb2c0f8bd 100644
--- a/ndb/src/ndbapi/ndberror.c
+++ b/ndb/src/ndbapi/ndberror.c
@@ -527,7 +527,8 @@ ErrorBundle ErrorCodes[] = {
   { 4270, IE, "Unknown blob error" },
   { 4335, AE, "Only one autoincrement column allowed per table. Having a table without primary key uses an autoincremented hidden key, i.e. a table without a primary key can not have an autoincremented column" },
   { 4271, AE, "Invalid index object, not retrieved via getIndex()" },
-  { 4275, AE, "The blob method is incompatible with operation type or lock mode" }
+  { 4275, AE, "The blob method is incompatible with operation type or lock mode" },
+  { 4294, AE, "Scan filter is too large, discarded" }
 };
 
 static