ndb - bug#19285 : document and check what blob ops are allowed

ndb/src/ndbapi/ndberror.c:
  distinguish blob method errors: 4265-wrong state 4275-wrong op type/lockmode
ndb/src/ndbapi/NdbBlobImpl.hpp:
  distinguish blob method errors: 4265-wrong state 4275-wrong op type/lockmode
  fix 4269 -> 4270
ndb/test/ndbapi/testBlobs.cpp:
  test lock upgrade, test 4275 errors
ndb/include/ndbapi/NdbScanOperation.hpp:
  fix comment
ndb/include/ndbapi/NdbBlob.hpp:
  upgrade LM_CommittedRead to LM_Read
  check if write allowed (new error 4275)
  dont invalidate blob state on error (just general principle)
ndb/src/ndbapi/NdbBlob.cpp:
  upgrade LM_CommittedRead to LM_Read
  check if write allowed (new error 4275)
  dont invalidate blob state on error (just general principle)

ndb/include/ndbapi/NdbBlob.hpp

@@ -74,19 +74,41 @@ class NdbColumnImpl;
  * NdbBlob methods return -1 on error and 0 on success, and use output
  * parameters when necessary.
  *
- * Operation types:
- * - insertTuple must use setValue if blob column is non-nullable
- * - readTuple with exclusive lock can also update existing value
- * - updateTuple can overwrite with setValue or update existing value
- * - writeTuple always overwrites and must use setValue if non-nullable
+ * Usage notes for different operation types:
+ *
+ * - insertTuple must use setValue if blob attribute is non-nullable
+ *
+ * - readTuple or scan readTuples with lock mode LM_CommittedRead is
+ *   automatically upgraded to lock mode LM_Read if any blob attributes
+ *   are accessed (to guarantee consistent view)
+ *
+ * - readTuple (with any lock mode) can only read blob value
+ *
+ * - updateTuple can either overwrite existing value with setValue or
+ *   update it in active phase
+ *
+ * - writeTuple always overwrites blob value and must use setValue if
+ *   blob attribute is non-nullable
+ *
  * - deleteTuple creates implicit non-accessible blob handles
- * - scan with exclusive lock can also update existing value
- * - scan "lock takeover" update op must do its own getBlobHandle
+ *
+ * - scan readTuples (any lock mode) can use its blob handles only
+ *   to read blob value
+ *
+ * - scan readTuples with lock mode LM_Exclusive can update row and blob
+ *   value using updateCurrentTuple, where the operation returned must
+ *   create its own blob handles explicitly
+ *
+ * - scan readTuples with lock mode LM_Exclusive can delete row (and
+ *   therefore blob values) using deleteCurrentTuple, which creates
+ *   implicit non-accessible blob handles
+ *
+ * - the operation returned by lockCurrentTuple cannot update blob value
  *
  * Bugs / limitations:
- * - lock mode upgrade should be handled automatically
- * - lock mode vs allowed operation is not checked
+ *
  * - too many pending blob ops can blow up i/o buffers
+ *
  * - table and its blob part tables are not created atomically
  */
 #ifndef DOXYGEN_SHOULD_SKIP_INTERNAL
@@ -194,6 +216,9 @@ public:
   /**
    * Return error object.  The error may be blob specific (below) or may
    * be copied from a failed implicit operation.
+   *
+   * The error code is copied back to the operation unless the operation
+   * already has a non-zero error code.
    */
   const NdbError& getNdbError() const;
   /**
@@ -290,6 +315,7 @@ private:
   bool isWriteOp();
   bool isDeleteOp();
   bool isScanOp();
+  bool isReadOnlyOp();
   bool isTakeOverOp();
   // computations
   Uint32 getPartNumber(Uint64 pos);
@@ -323,9 +349,9 @@ private:
   int preCommit();
   int atNextResult();
   // errors
-  void setErrorCode(int anErrorCode, bool invalidFlag = true);
-  void setErrorCode(NdbOperation* anOp, bool invalidFlag = true);
-  void setErrorCode(NdbTransaction* aCon, bool invalidFlag = true);
+  void setErrorCode(int anErrorCode, bool invalidFlag = false);
+  void setErrorCode(NdbOperation* anOp, bool invalidFlag = false);
+  void setErrorCode(NdbTransaction* aCon, bool invalidFlag = false);
 #ifdef VM_TRACE
   int getOperationType() const;
   friend class NdbOut& operator<<(NdbOut&, const NdbBlob&);

ndb/include/ndbapi/NdbScanOperation.hpp

@@ -41,7 +41,7 @@ public:
    * readTuples.
    */
   enum ScanFlag {
-    SF_TupScan = (1 << 16),     // scan TUP - only LM_CommittedRead
+    SF_TupScan = (1 << 16),     // scan TUP
     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

ndb/src/ndbapi/NdbBlob.cpp

@@ -265,6 +265,16 @@ NdbBlob::isScanOp()
     theNdbOp->theOperationType == NdbOperation::OpenRangeScanRequest;
 }
 
+inline bool
+NdbBlob::isReadOnlyOp()
+{
+  return ! (
+    theNdbOp->theOperationType == NdbOperation::InsertRequest ||
+    theNdbOp->theOperationType == NdbOperation::UpdateRequest ||
+    theNdbOp->theOperationType == NdbOperation::WriteRequest
+  );
+}
+
 inline bool
 NdbBlob::isTakeOverOp()
 {
@@ -438,12 +448,12 @@ NdbBlob::getValue(void* data, Uint32 bytes)
 {
   DBUG_ENTER("NdbBlob::getValue");
   DBUG_PRINT("info", ("data=%p bytes=%u", data, bytes));
-  if (theGetFlag || theState != Prepared) {
-    setErrorCode(NdbBlobImpl::ErrState);
+  if (! isReadOp() && ! isScanOp()) {
+    setErrorCode(NdbBlobImpl::ErrCompat);
     DBUG_RETURN(-1);
   }
-  if (! isReadOp() && ! isScanOp()) {
-    setErrorCode(NdbBlobImpl::ErrUsage);
+  if (theGetFlag || theState != Prepared) {
+    setErrorCode(NdbBlobImpl::ErrState);
     DBUG_RETURN(-1);
   }
   if (data == NULL && bytes != 0) {
@@ -461,12 +471,12 @@ NdbBlob::setValue(const void* data, Uint32 bytes)
 {
   DBUG_ENTER("NdbBlob::setValue");
   DBUG_PRINT("info", ("data=%p bytes=%u", data, bytes));
-  if (theSetFlag || theState != Prepared) {
-    setErrorCode(NdbBlobImpl::ErrState);
+  if (isReadOnlyOp()) {
+    setErrorCode(NdbBlobImpl::ErrCompat);
     DBUG_RETURN(-1);
   }
-  if (! isInsertOp() && ! isUpdateOp() && ! isWriteOp()) {
-    setErrorCode(NdbBlobImpl::ErrUsage);
+  if (theSetFlag || theState != Prepared) {
+    setErrorCode(NdbBlobImpl::ErrState);
     DBUG_RETURN(-1);
   }
   if (data == NULL && bytes != 0) {
@@ -533,6 +543,10 @@ int
 NdbBlob::setNull()
 {
   DBUG_ENTER("NdbBlob::setNull");
+  if (isReadOnlyOp()) {
+    setErrorCode(NdbBlobImpl::ErrCompat);
+    DBUG_RETURN(-1);
+  }
   if (theNullFlag == -1) {
     if (theState == Prepared) {
       DBUG_RETURN(setValue(0, 0));
@@ -571,6 +585,10 @@ NdbBlob::truncate(Uint64 length)
 {
   DBUG_ENTER("NdbBlob::truncate");
   DBUG_PRINT("info", ("length=%llu", length));
+  if (isReadOnlyOp()) {
+    setErrorCode(NdbBlobImpl::ErrCompat);
+    DBUG_RETURN(-1);
+  }
   if (theNullFlag == -1) {
     setErrorCode(NdbBlobImpl::ErrState);
     DBUG_RETURN(-1);
@@ -628,12 +646,14 @@ NdbBlob::setPos(Uint64 pos)
 int
 NdbBlob::readData(void* data, Uint32& bytes)
 {
+  DBUG_ENTER("NdbBlob::readData");
   if (theState != Active) {
     setErrorCode(NdbBlobImpl::ErrState);
-    return -1;
+    DBUG_RETURN(-1);
   }
   char* buf = static_cast<char*>(data);
-  return readDataPrivate(buf, bytes);
+  int ret = readDataPrivate(buf, bytes);
+  DBUG_RETURN(ret);
 }
 
 int
@@ -722,12 +742,18 @@ NdbBlob::readDataPrivate(char* buf, Uint32& bytes)
 int
 NdbBlob::writeData(const void* data, Uint32 bytes)
 {
+  DBUG_ENTER("NdbBlob::writeData");
+  if (isReadOnlyOp()) {
+    setErrorCode(NdbBlobImpl::ErrCompat);
+    DBUG_RETURN(-1);
+  }
   if (theState != Active) {
     setErrorCode(NdbBlobImpl::ErrState);
-    return -1;
+    DBUG_RETURN(-1);
   }
   const char* buf = static_cast<const char*>(data);
-  return writeDataPrivate(buf, bytes);
+  int ret = writeDataPrivate(buf, bytes);
+  DBUG_RETURN(0);
 }
 
 int
@@ -1130,6 +1156,9 @@ NdbBlob::atPrepare(NdbTransaction* aCon, NdbOperation* anOp, const NdbColumnImpl
       }
     }
     if (isReadOp()) {
+      // upgrade lock mode
+      if (theNdbOp->theLockMode == NdbOperation::LM_CommittedRead)
+        theNdbOp->theLockMode = NdbOperation::LM_Read;
       // add read of head+inline in this op
       if (getHeadInlineValue(theNdbOp) == -1)
         DBUG_RETURN(-1);
@@ -1148,6 +1177,9 @@ NdbBlob::atPrepare(NdbTransaction* aCon, NdbOperation* anOp, const NdbColumnImpl
     supportedOp = true;
   }
   if (isScanOp()) {
+    // upgrade lock mode
+    if (theNdbOp->theLockMode == NdbOperation::LM_CommittedRead)
+      theNdbOp->theLockMode = NdbOperation::LM_Read;
     // add read of head+inline in this op
     if (getHeadInlineValue(theNdbOp) == -1)
       DBUG_RETURN(-1);

ndb/src/ndbapi/NdbBlobImpl.hpp

@@ -24,7 +24,7 @@ public:
   STATIC_CONST( ErrTable = 4263 );
   // "Invalid usage of blob attribute" 
   STATIC_CONST( ErrUsage = 4264 );
-  // "Method is not valid in current blob state"
+  // "The blob method is not valid in current blob state"
   STATIC_CONST( ErrState = 4265 );
   // "Invalid blob seek position"
   STATIC_CONST( ErrSeek = 4266 );
@@ -33,7 +33,9 @@ public:
   // "Error in blob head update forced rollback of transaction"
   STATIC_CONST( ErrAbort = 4268 );
   // "Unknown blob error"
-  STATIC_CONST( ErrUnknown = 4269 );
+  STATIC_CONST( ErrUnknown = 4270 );
+  // "The blob method is incompatible with operation type or lock mode"
+  STATIC_CONST( ErrCompat = 4275 );
 };
 
 #endif

ndb/src/ndbapi/ndberror.c

@@ -513,14 +513,15 @@ ErrorBundle ErrorCodes[] = {
   { 4262, UD, "NdbScanFilter: Condition is out of bounds"},
   { 4263, IE, "Invalid blob attributes or invalid blob parts table" },
   { 4264, AE, "Invalid usage of blob attribute" },
-  { 4265, AE, "Method is not valid in current blob state" },
+  { 4265, AE, "The blob method is not valid in current blob state" },
   { 4266, AE, "Invalid blob seek position" },
   { 4267, IE, "Corrupted blob value" },
   { 4268, IE, "Error in blob head update forced rollback of transaction" },
   { 4269, IE, "No connection to ndb management server" },
   { 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()" }
+  { 4271, AE, "Invalid index object, not retrieved via getIndex()" },
+  { 4275, IE, "The blob method is incompatible with operation type or lock mode" }
 };
 
 static