Return sentinel for non-existant record

Signed-off-by: Mark Hannum <[email protected]>

Author: markhannum

db/phys_rep.c

@@ -672,6 +672,8 @@ int is_valid_lsn(unsigned int file, unsigned int offset)
            offset == get_next_offset(thedb->bdb_env->dbenv, info);
 }
 
+extern __thread int physrep_out_of_order;
+
 static LOG_INFO handle_record(cdb2_hndl_tp *repl_db, LOG_INFO prev_info)
 {
     /* vars for 1 record */
@@ -692,6 +694,14 @@ static LOG_INFO handle_record(cdb2_hndl_tp *repl_db, LOG_INFO prev_info)
         physrep_logmsg(LOGMSG_ERROR, "%s:%d: Could not parse lsn %s\n",
                        __func__, __LINE__, lsn);
     }
+    if (file == -1 && offset == -1) {
+        if (gbl_physrep_debug) {
+            physrep_logmsg(LOGMSG_USER, "%s:%d requested invalid record, force reconnect\n",
+                           __func__, __LINE__);
+        }
+        physrep_out_of_order = 1;
+        return prev_info;
+    }
     if (gbl_physrep_debug) {
         physrep_logmsg(LOGMSG_USER, "%s:%d: Processing record (lsn %d:%d)\n",
                        __func__, __LINE__, file, offset);
@@ -1362,7 +1372,6 @@ static int do_wait_for_reverse_conn(cdb2_hndl_tp *repl_metadb) {
        apply physical logs.
 */
 int gbl_physrep_pollms = 0;
-extern __thread int physrep_out_of_order;
 static void *physrep_worker(void *args)
 {
     comdb2_name_thread(__func__);
@@ -1567,9 +1576,9 @@ static void *physrep_worker(void *args)
         prev_info = info;
 
         rc = snprintf(sql_cmd, sql_cmd_len,
-                      "select * from comdb2_transaction_logs('{%u:%u}'%s)",
+                      "select * from comdb2_transaction_logs('{%u:%u}', NULL%s)",
                       info.file, info.offset,
-                      (gbl_blocking_physrep ? ", NULL, 1" : ""));
+                      (gbl_blocking_physrep ? ",9" : ",8"));
         if (rc < 0 || rc >= sql_cmd_len)
             physrep_logmsg(LOGMSG_ERROR, "%s:%d Command buffer is not long enough!\n", __func__, __LINE__);
         if (gbl_physrep_debug)

sqlite/ext/comdb2/tranlog.c

@@ -69,6 +69,7 @@ struct tranlog_cursor {
   int startAppRecGen;
   int starttime;
   int timeout;
+  int invalidRecord;
   DB_LOGC *logc;             /* Log Cursor */
   DBT data;
 };
@@ -154,6 +155,11 @@ static int tranlogNext(sqlite3_vtab_cursor *cur)
   uint32_t durable_gen=0, rc, getflags;
   bdb_state_type *bdb_state = thedb->bdb_env;
 
+  if (pCur->invalidRecord) {
+      pCur->hitLast = 1;
+      return SQLITE_OK;
+  }
+
   if (pCur->notDurable || pCur->hitLast)
       return SQLITE_OK;
 
@@ -252,9 +258,13 @@ static int tranlogNext(sqlite3_vtab_cursor *cur)
       }
 
       if (getflags != DB_NEXT && getflags != DB_PREV) {
-          logmsg(LOGMSG_ERROR, "%s line %d did not expect logc->get to fail with flag: %d."
-                               "got rc %d\n",
-                  __func__, __LINE__, getflags, rc);
+          if (pCur->flags & TRANLOG_FLAGS_SENTINEL) {
+              /* Return sentinel record with LSN {-1:-1} */
+              pCur->curLsn.file = pCur->curLsn.offset = -1;
+              pCur->invalidRecord = 1;
+              pCur->iRowid++;
+              return SQLITE_OK;
+          }
           return SQLITE_INTERNAL;
       }
 
@@ -665,7 +675,9 @@ static int tranlogColumn(
         }
         break;
     case TRANLOG_COLUMN_TXNID:
-        LOGCOPY_32(&txnid, &((char *) pCur->data.data)[4]); 
+        if (pCur->data.data) {
+            LOGCOPY_32(&txnid, &((char *) pCur->data.data)[4]); 
+        }
         sqlite3_result_int64(ctx, txnid);
         break;
     case TRANLOG_COLUMN_UTXNID:
@@ -717,7 +729,11 @@ static int tranlogColumn(
         }
         break;
     case TRANLOG_COLUMN_LOG:
-        sqlite3_result_blob(ctx, pCur->data.data, pCur->data.size, NULL);
+        if (pCur->data.data) {
+            sqlite3_result_blob(ctx, pCur->data.data, pCur->data.size, NULL);
+        } else {
+            sqlite3_result_null(ctx);
+        }
         break;
     case TRANLOG_COLUMN_CHILDUTXNID:
         if (pCur->data.data)

sqlite/ext/comdb2/tranlog.h

@@ -24,6 +24,7 @@ enum {
     TRANLOG_FLAGS_BLOCK             = 0x1,
     TRANLOG_FLAGS_DURABLE           = 0x2,
     TRANLOG_FLAGS_DESCENDING        = 0x4,
+    TRANLOG_FLAGS_SENTINEL          = 0x8,
 };
 
 u_int64_t get_timestamp_from_matchable_record(char *data);

tests/phys_rep_tiered.test/runit

@@ -1978,6 +1978,16 @@ function check_metadb_memstat
     fi
 }
 
+# Ask for a non-existent record with and without sentinel flag
+function non_existent_record
+{
+    x=$($CDB2SQL_EXE --tabs $CDB2_OPTIONS $DBNAME default "select lsn from comdb2_transaction_logs('{1:1}', NULL, 8)" 2>/dev/null)
+    [[ "$x" != '{-1:-1}' ]] && cleanFailExit "Expected {-1:-1} for non-existent record with sentinel flag, got $x"
+
+    x=$($CDB2SQL_EXE --tabs $CDB2_OPTIONS $DBNAME default "select lsn from comdb2_transaction_logs('{1:1}', NULL, 0)" 2>/dev/null)
+    [[ -n "$x" ]] && cleanFailExit "Expected no output, got $x, for non-existent record without sentinel flag"
+}
+
 # Halt a clustered physical replicant .. swing the master on that cluster
 # a number of times, then restart the physical replicant .. looking at the 
 # trace, verify that nothing ignores rectype 2, 11, or 10 ..
@@ -2197,6 +2207,11 @@ function run_tests
     check_metadb_memstat
     testcase_finish $testcase
 
+    testcase="non_existent_record"
+    testcase_preamble $testcase
+    non_existent_record
+    testcase_finish $testcase
+
     testcase="verify_non_ignored_reptype"
     testcase_preamble $testcase
     verify_non_ignored_reptype