[BACKPORT 2024.2][yugabyte#26060] docdb: Fast backward scan returns stale results for select statement with key columns only

Summary:
There are several scenarios where rows are requested by key columns only (including constraints and
aggregates like count). Doc reader have an optimization for this kind of reads: since actual values
does not matter (because key column values can be extracted from the row's doc key) then there is
no need to decode row values and it is enough to check if the row is tombstoned or not.

Unfortunately, this optimization was not updated to take fast backward scan into account:
an iterator stops reading when the firstly met record is not marked with a tombstone value,
which is the case for fast backward scan, which reads a row in the reverse order, from the
oldest to the newest change (per sub doc key), and the oldest record is generally corresponds
to the initially inserted values.

The change updates the algorithm if checking the row existence: the scanning continues until
all row changes have been taken into account to get the actual state of the row.

**Examples of row existence check before the fix**

Example #1
```
Record #1: KEY, T1 => PACKED ROW
Record #2: KEY, COL1, T2 => COL1 NEW VALUE 1

Forward scan / old backward scan:
#1 (alive, T1) => stop
result: row exists.

Fast backward scan:
#2 (alive, T2) => stop
result: row exists.
```

Example #2
```
Record #1: KEY, T3 => DEL
Record #2: KEY, T1 => PACKED ROW
Record #3: KEY, COL1, T2 => COL1 NEW VALUE

Forward scan / old backward scan:
#1 (deleted, T3), #2 (skipped: T1 < T3), #3 (skipped: T2 < T3) => stop
result: row does not exist.

Fast backward scan:
#3 (alive, T2) => stop
result: row exists.
```

Example #3
```
Record #1: KEY, T4 => PACKED ROW
Record #2: KEY, T3 => DEL
Record #3: KEY, T1 => PACKED ROW
Record #4: KEY, COL1, T2 => COL1 NEW VALUE

Forward scan / old backward scan:
#1 (alive, T4) => stop
result: row exists.

Fast backward scan:
#4 (alive, T2) => stop
result: row exists.
```

**Examples of row existence check with the fix**

Example #1
```
Record #1: KEY, T1 => PACKED ROW
Record #2: KEY, COL1, T2 => COL1 NEW VALUE 1

Fast backward scan:
#2 (alive, T2) => #1 (skipped, T1 < T2) => stop
result: row exists.
```

Example #2
```
Record #1: KEY, T3 => DEL
Record #2: KEY, T1 => PACKED ROW
Record #3: KEY, COL1, T2 => COL1 NEW VALUE

Fast backward scan:
#3 (alive, T2) => #2 (skipped: T1 < T2) => #1 (deleted: T3 > T2) => stop
result: row does not exist.
```

Example #3
```
Record #1: KEY, T4 => PACKED ROW
Record #2: KEY, T3 => DEL
Record #3: KEY, T1 => PACKED ROW
Record #4: KEY, COL1, T2 => COL1 NEW VALUE

Fast backward scan:
#4 (alive, T2) => #3 (skipped: T1 < T2) => #2 (deleted: T3 > T2) => #1 (alive: T4 > T3) => stop
result: row exists.
```

Original commit: d8bd1fd / D42212

Jira: DB-15387

Test Plan:
./yb_build.sh --cxx-test pgwrapper_pg_single_tserver-test --gtest_filter PgSingleTServerTest/PgFastBackwardScanTest.Simple/Fast_WithNulls_kV1
./yb_build.sh --cxx-test pgwrapper_pg_single_tserver-test --gtest_filter PgSingleTServerTest/PgFastBackwardScanTest.Simple/Fast_WithNulls_kV2
./yb_build.sh --cxx-test pgwrapper_pg_single_tserver-test --gtest_filter PgSingleTServerTest/PgFastBackwardScanTest.Simple/Fast_WithNulls_kNone
./yb_build.sh --cxx-test pgwrapper_pg_single_tserver-test --gtest_filter PgSingleTServerTest/PgFastBackwardScanTest.Simple/Fast_WithoutNulls_kV1
./yb_build.sh --cxx-test pgwrapper_pg_single_tserver-test --gtest_filter PgSingleTServerTest/PgFastBackwardScanTest.Simple/Fast_WithoutNulls_kV2
./yb_build.sh --cxx-test pgwrapper_pg_single_tserver-test --gtest_filter PgSingleTServerTest/PgFastBackwardScanTest.Simple/Fast_WithoutNulls_kNone
./yb_build.sh --cxx-test pgwrapper_pg_single_tserver-test --gtest_filter PgSingleTServerTest/PgFastBackwardScanTest.Simple/Slow_WithNulls_kV1
./yb_build.sh --cxx-test pgwrapper_pg_single_tserver-test --gtest_filter PgSingleTServerTest/PgFastBackwardScanTest.Simple/Slow_WithNulls_kV2
./yb_build.sh --cxx-test pgwrapper_pg_single_tserver-test --gtest_filter PgSingleTServerTest/PgFastBackwardScanTest.Simple/Slow_WithNulls_kNone
./yb_build.sh --cxx-test pgwrapper_pg_single_tserver-test --gtest_filter PgSingleTServerTest/PgFastBackwardScanTest.Simple/Slow_WithoutNulls_kV1
./yb_build.sh --cxx-test pgwrapper_pg_single_tserver-test --gtest_filter PgSingleTServerTest/PgFastBackwardScanTest.Simple/Slow_WithoutNulls_kV2
./yb_build.sh --cxx-test pgwrapper_pg_single_tserver-test --gtest_filter PgSingleTServerTest/PgFastBackwardScanTest.Simple/Slow_WithoutNulls_kNone

**manual testing:**
```
SET yb_use_hash_splitting_by_default = FALSE;
CREATE TABLE t1(c0 int UNIQUE DEFAULT 1);
INSERT INTO t1(c0) VALUES (2);
DELETE FROM t1;
SELECT * FROM t1;
SELECT * FROM t1 GROUP BY t1.c0 ORDER BY t1.c0 DESC;
```

Reviewers: sergei, rthallam

Reviewed By: rthallam

Subscribers: yql, ybase

Tags: #jenkins-ready

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

Author: arybochkin

src/yb/docdb/doc_reader.cc

@@ -838,7 +838,10 @@ class GetHelperBase : public PackedRowContext {
     auto& fetched_key = VERIFY_RESULT_REF(Prepare(prefetched_key, root_write_time));
 
     if constexpr (kCheckExistOnly) {
-      if (found_) {
+      // In case of fast backward scan enabled, row reading happens in the reverse order,
+      // from the overy oldest to very newest record, which mean it is required to scan all
+      // the records to understand if the row exists.
+      if (found_ && !kFastBackward) {
         return FoundResult(/* iter_valid= */ true);
       }
       auto iter_valid = VERIFY_RESULT(Scan(&fetched_key, root_write_time));
@@ -901,6 +904,7 @@ class GetHelperBase : public PackedRowContext {
 
     // Skip too old root records.
     if (key_result.write_time < root_write_time->encoded()) {
+      DVLOG_WITH_PREFIX_AND_FUNC(4) << "record is too old, skipping";
       Move<kFastBackward>(data_.iter);
       return true; // Continue scanning.
     }
@@ -911,11 +915,13 @@ class GetHelperBase : public PackedRowContext {
       auto has_row_update = data_.packed_row->HasUpdatesAfter(key_result.write_time);
       if (has_row_update.value_or(true)) {
         // Delete record is too old, just skip it.
+        DVLOG_WITH_PREFIX_AND_FUNC(4) << "tombstoned record is too old, skipping";
         Move<kFastBackward>(data_.iter);
         return true; // Continue scanning.
       }
 
       // This is a more recent delete, need to be taken it into account.
+      DVLOG_WITH_PREFIX_AND_FUNC(4) << "read tombstoned record, row is considered deleted";
       found_ = false;
     }
 
@@ -970,7 +976,8 @@ class GetHelperBase : public PackedRowContext {
     DVLOG_WITH_PREFIX_AND_FUNC(4)
         << "key: " << dockv::SubDocKey::DebugSliceToString(key_result.key)
         << ", write time: " << key_result.write_time.ToString()
-        << ", value: " << key_result.value.ToDebugHexString();
+        << ", value: " << key_result.value.ToDebugHexString()
+        << ", root write time: " << root_write_time->ToString();
     DCHECK(key_result.key.starts_with(root_doc_key_));
 
     // With the fast backward scan, the full packed row may be not the first record met during
@@ -1046,9 +1053,6 @@ class GetHelperBase : public PackedRowContext {
           << "current position: " << dockv::SubDocKey::DebugSliceToString(current_entry->key)
           << ", value: " << current_entry->value.ToDebugHexString();
       RETURN_NOT_OK(ProcessEntry(subkeys, current_entry->value, current_entry->write_time));
-      if (kCheckExistOnly && found_) {
-        return false;
-      }
 
       data_.packed_row->TrackColumnUpdate(current_column_->id, current_entry->write_time);
 
@@ -1071,7 +1075,7 @@ class GetHelperBase : public PackedRowContext {
       // Check if the iterator moved out the current column after the Prev().
       current_entry = &VERIFY_RESULT_REF(data_.iter->Fetch());
       if (!IsEntrySubKeysSame(*current_entry, subkeys)) {
-        // No more entries for the given root doc key or move out the given column.
+        // No more entries for the given root doc key or the iterator moved out the given column.
         break;
       }
 

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

@@ -871,10 +871,26 @@ class PgFastBackwardScanTest
     // A helper functor to execute a select statement for the test.
     auto fetch_and_validate = [this, &conn, &table_name, with_nulls](
         const std::string& expected_without_nulls, const std::string& expected_with_nulls) {
+      const auto stmt =
+          Format("SELECT c_1, c_5, r FROM $0 WHERE h = 1 ORDER BY r DESC", table_name);
+
+      // Make sure the statement is backward scan.
+      const auto explain = VERIFY_RESULT(conn.FetchAllAsString("EXPLAIN ANALYZE " + stmt));
+      SCHECK(Slice(explain).starts_with("Index Scan Backward"), RuntimeError, "");
+
       return FetchAndValidate(conn,
           Format("SELECT c_1, c_5, r FROM $0 WHERE h = 1 ORDER BY r DESC", table_name),
           with_nulls ? expected_with_nulls : expected_without_nulls);
     };
+    auto fetch_and_validate_pk_only = [this, &conn, &table_name](const std::string& expected) {
+      const auto stmt = Format("SELECT r FROM $0 WHERE h = 1 ORDER BY r DESC", table_name);
+
+      // Make sure the statement is backward scan.
+      const auto explain = VERIFY_RESULT(conn.FetchAllAsString("EXPLAIN ANALYZE " + stmt));
+      SCHECK(Slice(explain).starts_with("Index Scan Backward"), RuntimeError, "");
+
+      return FetchAndValidate(conn, stmt, expected);
+    };
 
     // Create table.
     {
@@ -906,6 +922,7 @@ class PgFastBackwardScanTest
     ASSERT_OK(fetch_and_validate(
         "1, 5, 3; 1, 5, 2; 1, 5, 1", "NULL, 5, 3; NULL, 5, 2; NULL, 5, 1"
     ));
+    ASSERT_OK(fetch_and_validate_pk_only("3; 2; 1"));
 
     if (intents_usage == IntentsUsage::kMixed) {
       ASSERT_OK(conn.StartTransaction(IsolationLevel::SERIALIZABLE_ISOLATION));
@@ -920,6 +937,7 @@ class PgFastBackwardScanTest
         "256, 260, 5; 256, 260, 4; 1, 5, 2; 4096, 5, 1",
         "256, 260, 5; 256, 260, 4; NULL, 5, 2; 4096, 5, 1"
     ));
+    ASSERT_OK(fetch_and_validate_pk_only("5; 4; 2; 1"));
 
     // Re-insert data for the deleted rows.
     ASSERT_OK(insert_values(build_values(/* r_keys */ keys(3), 16383)));
@@ -928,25 +946,29 @@ class PgFastBackwardScanTest
         "256, 260, 5; 256, 260, 4; 16384, 16388, 3; 1, 5, 2; 4096, 5, 1",
         "256, 260, 5; 256, 260, 4; 16384, 16388, 3; NULL, 5, 2; 4096, 5, 1"
     ));
+    ASSERT_OK(fetch_and_validate_pk_only("5; 4; 3; 2; 1"));
 
     // Delete the same records again.
     ASSERT_OK(conn.ExecuteFormat("DELETE FROM $0 WHERE r = 3", table_name));
     ASSERT_OK(cluster_->FlushTablets(tablet::FlushMode::kSync));
     ASSERT_OK(fetch_and_validate(
         "256, 260, 5; 256, 260, 4; 1, 5, 2; 4096, 5, 1",
         "256, 260, 5; 256, 260, 4; NULL, 5, 2; 4096, 5, 1"));
+    ASSERT_OK(fetch_and_validate_pk_only("5; 4; 2; 1"));
 
     // Delete records from the intents DB in such way the very first row would be in regular DB.
     ASSERT_OK(conn.ExecuteFormat("DELETE FROM $0 WHERE r IN (4, 5, 6)", table_name));
     ASSERT_OK(cluster_->FlushTablets(tablet::FlushMode::kSync));
     ASSERT_OK(fetch_and_validate("1, 5, 2; 4096, 5, 1", "NULL, 5, 2; 4096, 5, 1"));
+    ASSERT_OK(fetch_and_validate_pk_only("2; 1"));
 
     // Insert some data which would be positioned before all the existing rows.
     ASSERT_OK(insert_values(build_values(/* r_keys */ keys(0), 65535)));
     ASSERT_OK(cluster_->FlushTablets(tablet::FlushMode::kSync));
     ASSERT_OK(fetch_and_validate(
         "1, 5, 2; 4096, 5, 1; 65536, 65540, 0", "NULL, 5, 2; 4096, 5, 1; 65536, 65540, 0"
     ));
+    ASSERT_OK(fetch_and_validate_pk_only("2; 1; 0"));
 
     if (intents_usage != IntentsUsage::kRegularOnly) {
       ASSERT_OK(conn.CommitTransaction());
@@ -956,6 +978,7 @@ class PgFastBackwardScanTest
     ASSERT_OK(conn.ExecuteFormat("DELETE FROM $0 WHERE h = 1 AND r >= 0", table_name));
     ASSERT_OK(cluster_->FlushTablets(tablet::FlushMode::kSync));
     ASSERT_OK(fetch_and_validate("", ""));
+    ASSERT_OK(fetch_and_validate_pk_only(""));
 
     LOG_WITH_FUNC(INFO) << "Done";
   }