[#28960] DocDB: Fix DBIter::Next after SeekToLast

Summary:
DBIter has fast next mode, which works faster in YB scenarios while iteration goes only in forward direction.
But it does not work correctly in conjunction with backward iteration.
As result process could crash during unique index back in HasDuplicateUniqueIndexValueBackward.

The DB state that leads to crash is the following:
1) Suppose we are trying to check index entry with some key.
2) There are should be intents after this key (could be entries from different table), but no records in regular DB.
3) There is should be exacly only record in regular DB before this key (could be entry from different table).

This this case SeekToLast will be invoked by IntentAwareIterator::FindOldestRecord.
It will position regular DB DBIter to entry (3) and underlying memtable iterator to invalid (i.e. null) position.

Calling Next on DBIter will result in call to FastNext and it will call Next on underlying memtable iterator in null position.
Which least to crash.

Fixed by disabling fast next mode in SeekToLast, like it was already done in Prev.

Test Plan: PgIndexBackfillIgnoreApplyTest.Backward

Reviewers: rthallam, huapeng.yuan

Reviewed By: huapeng.yuan

Subscribers: ybase, yql

Tags: #jenkins-ready

Differential Revision: https://phorge.dev.yugabyte.com/D48206

Author: spolitov

src/yb/common/doc_hybrid_time.cc

@@ -284,6 +284,9 @@ void EncodedDocHybridTime::Reset() {
 }
 
 std::string EncodedDocHybridTime::ToString() const {
+  if (empty()) {
+    return "<EMPTY>";
+  }
   return DocHybridTime::FullyDecodeFrom(AsSlice()).ToString();
 }
 

src/yb/docdb/intent_aware_iterator.cc

@@ -1053,13 +1053,12 @@ Result<EncodedDocHybridTime> IntentAwareIterator::FindMatchingIntentRecordDocHyb
 
 Result<EncodedDocHybridTime> IntentAwareIterator::GetMatchingRegularRecordDocHybridTime(
     Slice key_without_ht) {
-  size_t other_encoded_ht_size =
-      VERIFY_RESULT(dockv::CheckHybridTimeSizeAndValueType(iter_->key()));
-  Slice iter_key_without_ht = iter_->key();
-  iter_key_without_ht.remove_suffix(1 + other_encoded_ht_size);
+  auto iter_key = iter_->key();
+  size_t other_encoded_ht_size = VERIFY_RESULT(dockv::CheckHybridTimeSizeAndValueType(iter_key));
+  auto iter_key_without_ht = iter_key.WithoutSuffix(1 + other_encoded_ht_size);
   if (key_without_ht == iter_key_without_ht) {
     EncodedDocHybridTime result;
-    RETURN_NOT_OK(DocHybridTime::EncodedFromEnd(iter_->key(), &result));
+    RETURN_NOT_OK(DocHybridTime::EncodedFromEnd(iter_key, &result));
     UpdateMaxSeenHt(result, key_without_ht);
     return result;
   }
@@ -1106,26 +1105,28 @@ Result<HybridTime> IntentAwareIterator::FindOldestRecord(
   RETURN_NOT_OK(status_);
 
   if (iter_->Valid()) {
+    VLOG_WITH_FUNC(4) << "Find prev: " << DebugDumpKeyToStr(iter_->key());
     SkipFutureRecords<Direction::kForward>(iter_->Prev());
   } else {
     HandleStatus(iter_->status());
     RETURN_NOT_OK(status_);
+    VLOG_WITH_FUNC(4) << "Seek to last";
     SkipFutureRecords<Direction::kForward>(iter_->SeekToLast());
   }
 
   if (regular_entry_) {
     auto regular_dht = VERIFY_RESULT(GetMatchingRegularRecordDocHybridTime(key_without_ht));
-    VLOG(4) << "Looking for Matching Regular Record found   =  " << regular_dht.ToString();
+    VLOG_WITH_FUNC(4) << "Looking for Matching Regular Record found = " << regular_dht.ToString();
     if (!regular_dht.empty()) {
       auto ht = VERIFY_RESULT(regular_dht.Decode()).hybrid_time();
       if (ht > min_hybrid_time) {
         result.MakeAtMost(ht);
       }
     }
   } else {
-    VLOG(4) << "regular_value_ is empty";
+    VLOG_WITH_FUNC(4) << "regular_value_ is empty";
   }
-  VLOG(4) << "Returning " << result;
+  VLOG_WITH_FUNC(4) << "Returning " << result;
   return result;
 }
 

src/yb/docdb/iter_util.cc

@@ -165,14 +165,14 @@ const rocksdb::KeyValueEntry& DoPerformRocksDBSeek(
       "    Seek() calls:     $8\n",
       file_name, line,
       dockv::BestEffortDocDBKeyToStr(seek_key),
-      FormatSliceAsStr(seek_key),
+      seek_key.ToDebugString(),
       iter->Valid()         ? dockv::BestEffortDocDBKeyToStr(iter->key())
       : iter->status().ok() ? "N/A"
                             : iter->status().ToString(),
-      iter->Valid()         ? FormatSliceAsStr(iter->key())
+      iter->Valid()         ? iter->key().ToDebugString()
       : iter->status().ok() ? "N/A"
                             : iter->status().ToString(),
-      iter->Valid()         ? FormatSliceAsStr(iter->value())
+      iter->Valid()         ? iter->value().ToDebugString()
       : iter->status().ok() ? "N/A"
                             : iter->status().ToString(),
       stats.num_nexts,

src/yb/docdb/pgsql_operation.cc

@@ -1213,7 +1213,7 @@ Result<bool> PgsqlWriteOperation::HasDuplicateUniqueIndexValue(const DocOperatio
 
 Result<bool> PgsqlWriteOperation::HasDuplicateUniqueIndexValueBackward(
     const DocOperationApplyData& data) {
-  VLOG(2) << "Looking for collision while going backward. Trying to insert " << doc_key_;
+  VLOG_WITH_FUNC(2) << "doc key: " << doc_key_;
 
   auto iter = CreateIntentAwareIterator(
       data.doc_write_batch->doc_db(),
@@ -1222,8 +1222,10 @@ Result<bool> PgsqlWriteOperation::HasDuplicateUniqueIndexValueBackward(
       txn_op_context_,
       data.read_operation_data.WithAlteredReadTime(ReadHybridTime::Max()));
 
+  VLOG_WITH_FUNC(4) << "whole row: " << doc_key_;
   HybridTime oldest_past_min_ht = VERIFY_RESULT(FindOldestOverwrittenTimestamp(
       iter.get(), SubDocKey(doc_key_), data.read_time().read));
+  VLOG_WITH_FUNC(4) << "liveness column: " << SubDocKey(doc_key_, KeyEntryValue::kLivenessColumn);
   const HybridTime oldest_past_min_ht_liveness =
       VERIFY_RESULT(FindOldestOverwrittenTimestamp(
           iter.get(),
@@ -1309,16 +1311,16 @@ Result<HybridTime> PgsqlWriteOperation::FindOldestOverwrittenTimestamp(
     const SubDocKey& sub_doc_key,
     HybridTime min_read_time) {
   HybridTime result;
-  VLOG(3) << "Doing iter->Seek " << doc_key_;
+  VLOG_WITH_FUNC(3) << doc_key_;
   iter->Seek(doc_key_);
   if (VERIFY_RESULT_REF(iter->Fetch())) {
     const auto bytes = sub_doc_key.EncodeWithoutHt();
     const Slice& sub_key_slice = bytes.AsSlice();
     result = VERIFY_RESULT(iter->FindOldestRecord(sub_key_slice, min_read_time));
-    VLOG(2) << "iter->FindOldestRecord returned " << result << " for "
-            << SubDocKey::DebugSliceToString(sub_key_slice);
+    VLOG_WITH_FUNC(2) << "iter->FindOldestRecord returned " << result << " for "
+                      << SubDocKey::DebugSliceToString(sub_key_slice);
   } else {
-    VLOG(3) << "iter->Seek " << doc_key_ << " turned out to be out of records";
+    VLOG_WITH_FUNC(3) << "iter->Seek " << doc_key_ << " turned out to be out of records";
   }
   return result;
 }

src/yb/master/catalog_manager.cc

@@ -110,6 +110,7 @@
 #include "yb/gutil/stl_util.h"
 #include "yb/gutil/strings/escaping.h"
 #include "yb/gutil/strings/join.h"
+#include "yb/gutil/strings/split.h"
 #include "yb/gutil/strings/substitute.h"
 #include "yb/gutil/sysinfo.h"
 #include "yb/gutil/walltime.h"
@@ -210,6 +211,7 @@
 #include "yb/util/status.h"
 #include "yb/util/status_format.h"
 #include "yb/util/status_log.h"
+#include "yb/util/stol_utils.h"
 #include "yb/util/stopwatch.h"
 #include "yb/util/string_case.h"
 #include "yb/util/string_util.h"
@@ -474,6 +476,10 @@ DEFINE_test_flag(bool, sequential_colocation_ids, false,
                  "rather than at random. This is especially useful for making pg_regress "
                  "tests output consistent and predictable.");
 
+DEFINE_test_flag(string, colocation_ids, "",
+                 "Comma separated list of colocation ids that should be used for the first created "
+                 "tables.");
+
 DEFINE_RUNTIME_bool(disable_truncate_table, false,
     "When enabled, truncate table will be disallowed");
 
@@ -3315,6 +3321,16 @@ Result<ColocationId> ConceiveColocationId(
     return req.colocation_id();
   }
 
+  if (PREDICT_FALSE(!FLAGS_TEST_colocation_ids.empty())) {
+    auto ids = VERIFY_RESULT(ParseCommaSeparatedListOfNumbers<ColocationId>(
+        FLAGS_TEST_colocation_ids));
+    for (auto colocation_id : ids) {
+      if (!contains_colocation_id(colocation_id)) {
+        return colocation_id;
+      }
+    }
+  }
+
   // Generate a random colocation ID unique within colocation group.
   ColocationId colocation_id = kColocationIdNotSet;
   if (FLAGS_TEST_sequential_colocation_ids) {

src/yb/rocksdb/db/db_iter.cc

@@ -123,7 +123,6 @@ class DBIter final : public Iterator {
         user_merge_operator_(ioptions.merge_operator),
         iter_(iter),
         sequence_(s),
-        direction_(kForward),
         current_entry_is_merged_(false),
         statistics_(statistics ? statistics : ioptions.statistics),
         version_number_(version_number),
@@ -365,6 +364,18 @@ class DBIter final : public Iterator {
   bool ParseKey(ParsedInternalKey* key);
   void MergeValuesNewToOld();
 
+  inline void SetForwardDirection() {
+    direction_ = kForward;
+  }
+
+  inline void SetReverseDirection() {
+    direction_ = kReverse;
+
+    // TODO(scanperf) allow fast next after reverse scan.
+    // Fallback to regular Next if reverse scan was used.
+    fast_next_ = false;
+  }
+
   inline void ClearSavedValue() {
     if (saved_value_.capacity() > 1048576) {
       std::string empty;
@@ -407,7 +418,8 @@ class DBIter final : public Iterator {
   Status status_;
   yb::ByteBuffer<kKeyBufferSize> key_buffer_;
   std::string saved_value_;
-  Direction direction_;
+  // Should be controlled via SetForwardDirection/SetReverseDirection
+  Direction direction_ = Direction::kForward;
   KeyValueEntry entry_;
 
 #ifdef ROCKSDB_CATCH_MISSING_STATUS_CHECK
@@ -467,7 +479,7 @@ const KeyValueEntry& DBIter::Next() {
 
   if (direction_ == kReverse) {
     FindNextUserKey();
-    direction_ = kForward;
+    SetForwardDirection();
     if (!iter_->Valid()) {
       iter_->SeekToFirst();
     }
@@ -498,6 +510,7 @@ const KeyValueEntry& DBIter::Next() {
 
 const KeyValueEntry& DBIter::FastNext() {
   DCHECK(entry_);
+  DCHECK_EQ(direction_, Direction::kForward);
 
   ++num_fast_next_calls_;
   const auto& entry = iter_->Next();
@@ -530,8 +543,8 @@ inline void DBIter::FindNextUserEntry(bool skipping) {
 // Actual implementation of DBIter::FindNextUserEntry()
 void DBIter::FindNextUserEntryInternal(bool skipping) {
   // Loop until we hit an acceptable entry to yield
-  assert(iter_->Valid());
-  assert(direction_ == kForward);
+  DCHECK(iter_->Valid());
+  DCHECK_EQ(direction_, Direction::kForward);
   current_entry_is_merged_ = false;
   uint64_t num_skipped = 0;
   do {
@@ -704,11 +717,7 @@ void DBIter::ReverseToBackward() {
 #endif
 
   FindPrevUserKey();
-  direction_ = kReverse;
-
-  // TODO(scanperf) allow fast next after reverse scan.
-  // Fallback to regular Next if reverse scan was used.
-  fast_next_ = false;
+  SetReverseDirection();
 }
 
 void DBIter::PrevInternal() {
@@ -968,7 +977,7 @@ const KeyValueEntry& DBIter::Seek(Slice target) {
   }
 
   if (iter_->Valid()) {
-    direction_ = kForward;
+    SetForwardDirection();
     ClearSavedValue();
     FindNextUserEntry(false /* not skipping */);
     RecordStats(NUMBER_DB_SEEK, NUMBER_DB_SEEK_FOUND);
@@ -988,7 +997,7 @@ const KeyValueEntry& DBIter::SeekToFirst() {
   if (prefix_extractor_ != nullptr) {
     max_skip_ = std::numeric_limits<uint64_t>::max();
   }
-  direction_ = kForward;
+  SetForwardDirection();
   ClearSavedValue();
 
   {
@@ -1015,7 +1024,7 @@ const KeyValueEntry& DBIter::SeekToLast() {
   if (prefix_extractor_ != nullptr) {
     max_skip_ = std::numeric_limits<uint64_t>::max();
   }
-  direction_ = kReverse;
+  SetReverseDirection();
   ClearSavedValue();
 
   {

src/yb/util/stol_utils.h

@@ -96,20 +96,18 @@ Result<T> CheckedParseNumber(Slice slice) {
       return *maybe_special_value;
     }
   }
-  if constexpr (std::is_same_v<T, int32_t> ||
-                std::is_same_v<T, uint32_t>) {
+  if constexpr (std::is_same_v<T, int32_t> || std::is_same_v<T, uint32_t>) {
     return CheckedStoInt<T>(slice);
   } else if constexpr (std::is_same_v<T, int64_t>) {
     return CheckedStoll(slice);
-  } else if constexpr (std::is_same_v<T, uint64_t> ||
-                       std::is_same_v<T, size_t>) {
+  } else if constexpr (std::is_same_v<T, uint64_t> || std::is_same_v<T, size_t>) { // NOLINT
     static_assert(sizeof(uint64_t) == sizeof(size_t),
                   "Assuming size_t is the same as uint64_t");
     return CheckedStoull(slice);
   } else if constexpr (std::is_same_v<T, long double>) {
     // For double, use CheckedStold directly
     return CheckedStold(slice);
-  } else if constexpr (std::is_floating_point_v<T>) {
+  } else if constexpr (std::is_floating_point_v<T>) { // NOLINT
     // For float and double, parse as double and cast
     auto long_double_value = VERIFY_RESULT(CheckedStold(slice));
     T result = static_cast<T>(long_double_value);
@@ -127,7 +125,7 @@ Result<T> CheckedParseNumber(Slice slice) {
 
 // Generalized function to parse comma-separated lists of numbers of a supported type into an
 // arbitrary container and enforce inclusive lower and upper bounds.
-template<ParseableNumber T, ContainerOf<T> Container>
+template<ParseableNumber T, ContainerOf<T> Container = std::vector<T>>
 Result<Container> ParseCommaSeparatedListOfNumbers(
     const std::string& input,
     std::optional<T> lower_bound = std::nullopt,

src/yb/yql/pgwrapper/pg_index_backfill-test.cc

@@ -2811,4 +2811,34 @@ TEST_P(PgIndexBackfillReadBeforeConcurrentUpdate, PartialIndex) {
   ASSERT_OK(CheckIndexConsistency(kPartialIndex));
 }
 
+class PgIndexBackfillIgnoreApplyTest : public PgIndexBackfillTest {
+ protected:
+  void UpdateMiniClusterOptions(ExternalMiniClusterOptions* options) override {
+    PgIndexBackfillTest::UpdateMiniClusterOptions(options);
+
+    options->extra_master_flags.push_back("--TEST_colocation_ids=1000,3000,2000");
+
+    options->extra_tserver_flags.push_back("--TEST_transaction_ignore_applying_probability=1.0");
+  }
+};
+
+TEST_P(PgIndexBackfillIgnoreApplyTest, Backward) {
+  const std::string kDbName = "colodb";
+  ASSERT_OK(conn_->ExecuteFormat("CREATE DATABASE $0 with COLOCATION = true", kDbName));
+  auto conn = ASSERT_RESULT(ConnectToDB(kDbName));
+  ASSERT_OK(conn.Execute(
+      "CREATE TABLE t2 (k INT, PRIMARY KEY (k ASC))"));
+  ASSERT_OK(conn.Execute(
+      "CREATE TABLE test (k INT, v INT, PRIMARY KEY (k ASC))"));
+  ASSERT_OK(conn.Execute("INSERT INTO t2 VALUES (48)"));
+
+  ASSERT_OK(conn.StartTransaction(IsolationLevel::SNAPSHOT_ISOLATION));
+  ASSERT_OK(conn.Execute("INSERT INTO test VALUES (11, 99)"));
+  ASSERT_OK(conn.CommitTransaction());
+
+  ASSERT_OK(conn.ExecuteFormat("CREATE UNIQUE INDEX idx ON test (v ASC)"));
+}
+
+INSTANTIATE_TEST_CASE_P(, PgIndexBackfillIgnoreApplyTest, ::testing::Bool());
+
 } // namespace yb::pgwrapper