[#28616] docdb: optimize index scans by adaptively deciding whether to use next or re-seek

Summary:
For index scan scenarios, optimisation to use up to `max_nexts_to_avoid_seek` nexts instead of seek could be useless when main table rows selected by index are not close enough one to each other.

This revision adds logic to avoid  useless next calls instead of seek using the following approach. If during RocksDB iterator lifetime optimisation was not useful for >=
`rocksdb_next_optimization_not_useful_limit` times and useful/num_nexts ratio is less than `rocksdb_next_optimization_skip_threshold_percentage` it stops using next instead of seek for this iterator.

That change allowed to reduce `select v3 from t3 where v1 = 123` query latency on serving data from block-cache from ~61ms to ~55ms (~10% improvement) for the following data set:
```
CREATE TABLE t3(k uuid default gen_random_uuid() primary key,
                v1 bigint, v2 bigint, v3 bigint, v4 varchar, v5 varchar)
       SPLIT INTO 3 TABLETS;
CREATE INDEX ON t3(v1) INCLUDE (v2);

WITH config AS ( SELECT 100000000::BIGINT AS total_records, 100000::BIGINT AS chunk_size )
   SELECT format( 'INSERT INTO t3(v1,v2,v3,v4,v5)
                                   SELECT i %% 20000, i, i, i || repeat(''abcd'',100), i || repeat(''foobar'',100)
                                   FROM generate_series(%s, %s) AS i;', (n * chunk_size) + 1, (n + 1) * chunk_size )
        FROM generate_series(0, (SELECT total_records / chunk_size - 1 FROM config)) AS n, config \gexec
```

This query uses the index on `v1` and will match/will lookup 5000 rows in the main table. Given that there are 3 tablets for the main table, we retrieve about 1666 rows from each tablet. And the metrics we track to decide whether to avoid the next calls and simply re-seek for each row are maintained in each tablet level request.
Jira: DB-18311

Test Plan: Jenkins

Reviewers: arybochkin

Reviewed By: arybochkin

Subscribers: kannan, mihnea, rthallam, ybase

Tags: #jenkins-ready

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

Author: ttyusupov

java/yb-pgsql/src/test/java/org/yb/pgsql/TestPgCostModelSeekNextEstimation.java

@@ -253,7 +253,9 @@ private void testSeekAndNextEstimationBitmapScanHelper(
                   .nodeType(NODE_YB_BITMAP_TABLE_SCAN)
                   .relationName(table_name)
                   .estimatedSeeks(expectedSeeksRange(expected_seeks))
-                  .estimatedNextsAndPrevs(expectedNextsRange(expected_nexts))
+// TODO(#28919): Fix cost model to take into account changes in DocDB seek/next behaviour made by
+// https://github.com/yugabyte/yugabyte-db/issues/28616 and uncomment this line:
+//                   .estimatedNextsAndPrevs(expectedNextsRange(expected_nexts))
                   .estimatedDocdbResultWidth(Checkers.equal(expected_docdb_result_width))
                   .readMetrics(makeMetricsBuilder()
                     .metric(METRIC_NUM_DB_SEEK, expectedSeeksRange(expected_seeks))
@@ -513,6 +515,7 @@ public void tearDown() throws Exception {
   protected Map<String, String> getTServerFlags() {
     Map<String, String> flagMap = super.getTServerFlags();
     flagMap.put("ysql_analyze_dump_metrics", "true");
+    flagMap.put("ysql_enable_packed_row", "true");
     flagMap.put("ysql_enable_packed_row_for_colocated_table", "true");
     // Disable auto analyze for CBO seek and next metrics test.
     flagMap.put("ysql_enable_auto_analyze", "false");
@@ -762,58 +765,58 @@ public void testSeekNextEstimationBitmapScan() throws Exception {
         T1_NAME, 4, 10, 5, makeBitmapIndexScanChecker(T1_INDEX_NAME, 4, 10));
       testSeekAndNextEstimationBitmapScanHelper(stmt, String.format("/*+BitmapScan(%s)*/ SELECT * "
         + "FROM %s WHERE k1 IN (4, 8, 12, 16)", T2_NAME, T2_NAME),
-        T2_NAME, 80, 240, 10, makeBitmapIndexScanChecker(T2_INDEX_NAME, 4, 86));
+        T2_NAME, 80, 100, 10, makeBitmapIndexScanChecker(T2_INDEX_NAME, 4, 86));
       testSeekAndNextEstimationBitmapScanHelper(stmt, String.format("/*+BitmapScan(%s)*/ SELECT * "
         + "FROM %s WHERE k2 IN (4, 8, 12, 16)", T2_NAME, T2_NAME),
-        T2_NAME, 80, 240, 10, makeBitmapIndexScanChecker(T2_INDEX_NAME, 101, 280));
+        T2_NAME, 80, 100, 10, makeBitmapIndexScanChecker(T2_INDEX_NAME, 101, 280));
       testSeekAndNextEstimationBitmapScanHelper(stmt, String.format("/*+BitmapScan(%s)*/ SELECT * "
         + "FROM %s WHERE k1 IN (4, 8, 12) AND k4 IN (4, 8, 12, 16)", T4_NAME, T4_NAME),
-        T4_NAME, 4830, 14490, 20, makeBitmapIndexScanChecker(T4_INDEX_NAME, 6007, 16808));
+        T4_NAME, 4800, 4900, 20, makeBitmapIndexScanChecker(T4_INDEX_NAME, 6007, 16808));
       testSeekAndNextEstimationBitmapScanHelper(stmt, String.format("/*+BitmapScan(%s)*/ SELECT * "
         + "FROM %s WHERE k2 IN (4, 8, 12, 16) AND k4 IN (4, 8, 12)", T4_NAME, T4_NAME),
-        T4_NAME, 4800, 14400, 20, makeBitmapIndexScanChecker(T4_INDEX_NAME, 6505, 17804));
+        T4_NAME, 4800, 4900, 20, makeBitmapIndexScanChecker(T4_INDEX_NAME, 6505, 17804));
       testSeekAndNextEstimationBitmapScanHelper(stmt, String.format("/*+BitmapScan(%s)*/ SELECT * "
         + "FROM %s WHERE k1 IN (4, 8, 12, 16) AND k2 IN (4, 8, 12, 16)", T4_NAME, T4_NAME),
-        T4_NAME, 6400, 19200, 20, makeBitmapIndexScanChecker(T4_INDEX_NAME, 22, 6436));
+        T4_NAME, 6400, 6540, 20, makeBitmapIndexScanChecker(T4_INDEX_NAME, 22, 6436));
       testSeekAndNextEstimationBitmapScanHelper(stmt, String.format("/*+BitmapScan(%s)*/ SELECT * "
         + "FROM %s WHERE k1 >= 4 and k1 < 14", T1_NAME, T1_NAME),
         T1_NAME, 10, 30, 5, makeBitmapIndexScanChecker(T1_INDEX_NAME, 1, 10));
       testSeekAndNextEstimationBitmapScanHelper(stmt, String.format("/*+BitmapScan(%s)*/ SELECT * "
         + "FROM %s WHERE k1 >= 4 and k1 < 14", T2_NAME, T2_NAME),
-        T2_NAME, 200, 600, 10, makeBitmapIndexScanChecker(T2_INDEX_NAME, 1, 200));
+        T2_NAME, 200, 220, 10, makeBitmapIndexScanChecker(T2_INDEX_NAME, 1, 200));
       testSeekAndNextEstimationBitmapScanHelper(stmt, String.format("/*+BitmapScan(%s)*/ SELECT * "
         + "FROM %s WHERE k1 >= 4 and k1 < 14", T3_NAME, T3_NAME),
-        T3_NAME, 4000, 12000, 15, makeBitmapIndexScanChecker(T3_INDEX_NAME, 4, 4000));
+        T3_NAME, 4000, 4080, 15, makeBitmapIndexScanChecker(T3_INDEX_NAME, 4, 4000));
       testSeekAndNextEstimationBitmapScanHelper(stmt, String.format("/*+BitmapScan(%s)*/ SELECT * "
         + "FROM %s WHERE k1 >= 4 and k1 < 14", T4_NAME, T4_NAME),
-        T4_NAME, 80000, 240000, 20, makeBitmapIndexScanChecker(T4_INDEX_NAME, 79, 80000));
+        T4_NAME, 80000, 81580, 20, makeBitmapIndexScanChecker(T4_INDEX_NAME, 79, 80000));
       testSeekAndNextEstimationBitmapScanHelper(stmt, String.format("/*+BitmapScan(%s)*/ SELECT * "
         + "FROM %s WHERE k2 >= 4 and k2 < 14", T2_NAME, T2_NAME),
-        T2_NAME, 200, 600, 10, makeBitmapIndexScanChecker(T2_INDEX_NAME, 41, 280));
+        T2_NAME, 200, 220, 10, makeBitmapIndexScanChecker(T2_INDEX_NAME, 41, 280));
       testSeekAndNextEstimationBitmapScanHelper(stmt, String.format("/*+BitmapScan(%s)*/ SELECT * "
         + "FROM %s WHERE k3 >= 4 and k3 < 14", T3_NAME, T3_NAME),
-        T3_NAME, 4000, 12000, 15, makeBitmapIndexScanChecker(T3_INDEX_NAME, 804, 5600));
+        T3_NAME, 4000, 4080, 15, makeBitmapIndexScanChecker(T3_INDEX_NAME, 804, 5600));
       testSeekAndNextEstimationBitmapScanHelper(stmt, String.format("/*+BitmapScan(%s)*/ SELECT * "
         + "FROM %s WHERE k4 >= 4 and k4 < 14", T4_NAME, T4_NAME),
-        T4_NAME, 80000, 240000, 20, makeBitmapIndexScanChecker(T4_INDEX_NAME, 16079, 112000));
+        T4_NAME, 80000, 81580, 20, makeBitmapIndexScanChecker(T4_INDEX_NAME, 16079, 112000));
       testSeekAndNextEstimationBitmapScanHelper(stmt, String.format("/*+BitmapScan(%s)*/ SELECT * "
         + "FROM %s WHERE k3 >= 4 and k3 < 14", T4_NAME, T4_NAME),
-        T4_NAME, 80000, 240000, 20, makeBitmapIndexScanChecker(T4_INDEX_NAME, 879, 81600));
+        T4_NAME, 80000, 81580, 20, makeBitmapIndexScanChecker(T4_INDEX_NAME, 879, 81600));
       testSeekAndNextEstimationBitmapScanHelper(stmt, String.format("/*+BitmapScan(%s)*/ SELECT * "
         + "FROM %s WHERE k2 >= 4 and k2 < 14", T4_NAME, T4_NAME),
-        T4_NAME, 80000, 240000, 20, makeBitmapIndexScanChecker(T4_INDEX_NAME, 120, 80000));
+        T4_NAME, 80000, 81580, 20, makeBitmapIndexScanChecker(T4_INDEX_NAME, 120, 80000));
       testSeekAndNextEstimationBitmapScanHelper(stmt, String.format("/*+BitmapScan(%s)*/ SELECT * "
         + "FROM %s WHERE k1 >= 4 and k1 < 14 and k3 >= 4 and k3 < 14", T4_NAME, T4_NAME),
-        T4_NAME, 40000, 120000, 20, makeBitmapIndexScanChecker(T4_INDEX_NAME, 440, 40800));
+        T4_NAME, 40000, 40800, 20, makeBitmapIndexScanChecker(T4_INDEX_NAME, 440, 40800));
       testSeekAndNextEstimationBitmapScanHelper(stmt, String.format("/*+BitmapScan(%s)*/ SELECT * "
         + "FROM %s WHERE k1 = 4 and k2 IN (4, 8, 12, 16)", T4_NAME, T4_NAME),
-        T4_NAME, 1600, 4800, 20, makeBitmapIndexScanChecker(T4_INDEX_NAME, 5, 1606));
+        T4_NAME, 1600, 1640, 20, makeBitmapIndexScanChecker(T4_INDEX_NAME, 5, 1606));
       testSeekAndNextEstimationBitmapScanHelper(stmt, String.format("/*+BitmapScan(%s)*/ SELECT * "
         + "FROM %s WHERE k1 IN (4, 8, 12, 16) and k2 = 4", T4_NAME, T4_NAME),
-        T4_NAME, 1600, 4800, 20, makeBitmapIndexScanChecker(T4_INDEX_NAME, 5, 1606));
+        T4_NAME, 1600, 1640, 20, makeBitmapIndexScanChecker(T4_INDEX_NAME, 5, 1606));
       testSeekAndNextEstimationBitmapScanHelper(stmt, String.format("/*+BitmapScan(%s)*/ SELECT * "
         + "FROM %s WHERE k3 IN (4, 8, 12, 16) and k4 = 4", T4_NAME, T4_NAME),
-        T4_NAME, 1600, 4800, 20, makeBitmapIndexScanChecker(T4_INDEX_NAME, 2002, 4000));
+        T4_NAME, 1600, 1640, 20, makeBitmapIndexScanChecker(T4_INDEX_NAME, 2002, 4000));
       testSeekAndNextEstimationBitmapScanHelper(stmt, String.format("/*+BitmapScan(%s)*/ SELECT * "
         + "FROM %s WHERE k1 >= 4 and k1 < 5 and k2 IN (4, 8, 12, 16)", T2_NAME, T2_NAME),
         T2_NAME, 4, 12, 10, makeBitmapIndexScanChecker(T2_INDEX_NAME, 5, 8));
@@ -825,31 +828,31 @@ public void testSeekNextEstimationBitmapScan() throws Exception {
         T2_NAME, 40, 120, 10, makeBitmapIndexScanChecker(T2_INDEX_NAME, 50, 98));
       testSeekAndNextEstimationBitmapScanHelper(stmt, String.format("/*+BitmapScan(%s)*/ SELECT * "
         + "FROM %s WHERE k1 >= 4 and k1 < 7 and k3 IN (4, 8, 12, 16)", T3_NAME, T3_NAME),
-        T3_NAME, 240, 720, 15, makeBitmapIndexScanChecker(T3_INDEX_NAME, 301, 600));
+        T3_NAME, 240, 260, 15, makeBitmapIndexScanChecker(T3_INDEX_NAME, 301, 600));
       testSeekAndNextEstimationBitmapScanHelper(stmt, String.format("/*+BitmapScan(%s)*/ SELECT * "
         + "FROM %s WHERE k2 >= 4 and k2 < 7 and k4 IN (4, 8, 12)", T4_NAME, T4_NAME),
-        T4_NAME, 3600, 10800, 20, makeBitmapIndexScanChecker(T4_INDEX_NAME, 4844, 9680));
+        T4_NAME, 3600, 3680, 20, makeBitmapIndexScanChecker(T4_INDEX_NAME, 4844, 9680));
       testSeekAndNextEstimationBitmapScanHelper(stmt, String.format("/*+BitmapScan(%s)*/ SELECT * "
         + "FROM %s WHERE k1 IN (1, 4, 7, 10)", T4_NAME, T4_NAME),
-        T4_NAME, 32000, 96000, 20, makeBitmapIndexScanChecker(T4_INDEX_NAME, 35, 32037));
+        T4_NAME, 32000, 32640, 20, makeBitmapIndexScanChecker(T4_INDEX_NAME, 35, 32037));
       testSeekAndNextEstimationBitmapScanHelper(stmt, String.format("/*+BitmapScan(%s)*/ SELECT * "
         + "FROM %s WHERE k1 >= 4 AND k1 < 14 AND k2 >= 4", T4_NAME, T4_NAME),
-        T4_NAME, 68000, 204000, 20, makeBitmapIndexScanChecker(T4_INDEX_NAME, 76, 68084));
+        T4_NAME, 68000, 69340, 20, makeBitmapIndexScanChecker(T4_INDEX_NAME, 76, 68084));
       testSeekAndNextEstimationBitmapScanHelper(stmt, String.format("/*+BitmapScan(%s)*/ SELECT * "
         + "FROM %s WHERE k1 >= 4 AND k1 < 14 AND k2 >= 4 AND k2 < 14", T4_NAME, T4_NAME),
-        T4_NAME, 40000, 120000, 20, makeBitmapIndexScanChecker(T4_INDEX_NAME, 59, 40077));
+        T4_NAME, 40000, 40802, 20, makeBitmapIndexScanChecker(T4_INDEX_NAME, 59, 40077));
       testSeekAndNextEstimationBitmapScanHelper(stmt, String.format("/*+BitmapScan(%s)*/ SELECT * "
         + "FROM %s WHERE k1 >= 4 AND k2 >= 4 AND k2 < 14", T4_NAME, T4_NAME),
-        T4_NAME, 68000, 204000, 20, makeBitmapIndexScanChecker(T4_INDEX_NAME, 100, 68132));
+        T4_NAME, 68000, 69340, 20, makeBitmapIndexScanChecker(T4_INDEX_NAME, 100, 68132));
       testSeekAndNextEstimationBitmapScanHelper(stmt, String.format("/*+BitmapScan(%s)*/ SELECT * "
         + "FROM %s WHERE k1 IN (1, 4, 7, 10) AND k2 IN (1, 4, 7, 10)", T4_NAME, T4_NAME),
-        T4_NAME, 6400, 19200, 20, makeBitmapIndexScanChecker(T4_INDEX_NAME, 22, 6436));
+        T4_NAME, 6400, 6450, 20, makeBitmapIndexScanChecker(T4_INDEX_NAME, 22, 6436));
       testSeekAndNextEstimationBitmapScanHelper(stmt, String.format("/*+BitmapScan(%s)*/ SELECT * "
         + "FROM %s WHERE k1 >= 4 AND k1 < 14 AND k3 >= 4 AND k3 < 14", T4_NAME, T4_NAME),
-        T4_NAME, 40000, 120000, 20, makeBitmapIndexScanChecker(T4_INDEX_NAME, 440, 40839));
+        T4_NAME, 40000, 40800, 20, makeBitmapIndexScanChecker(T4_INDEX_NAME, 440, 40839));
       testSeekAndNextEstimationBitmapScanHelper(stmt, String.format("/*+BitmapScan(%s)*/ SELECT * "
         + "FROM %s WHERE k1 >= 4 AND k3 >= 4 AND k3 < 14", T4_NAME, T4_NAME),
-        T4_NAME, 68000, 204000, 20, makeBitmapIndexScanChecker(T4_INDEX_NAME, 747, 69426));
+        T4_NAME, 68000, 69340, 20, makeBitmapIndexScanChecker(T4_INDEX_NAME, 747, 69426));
       stmt.execute("RESET work_mem");
     }
   }
@@ -958,12 +961,12 @@ public void testSeekNextEstimationBitmapScanWithAnd() throws Exception {
       final String query = "/*+ BitmapScan(t) */ SELECT * FROM %s AS t WHERE %s";
       testSeekAndNextEstimationBitmapScanHelper(stmt,
         String.format(query, T_NO_PKEY_NAME, "k1 <= 1 AND k2 <= 1"),
-        T_NO_PKEY_NAME, 100, 300, 10,
+        T_NO_PKEY_NAME, 100, 120, 10,
         makePlanBuilder().nodeType(NODE_BITMAP_INDEX_SCAN).build());
 
       testSeekAndNextEstimationBitmapScanHelper(stmt,
         String.format(query, T_NO_PKEY_NAME, "k1 <= 2 AND k2 <= 2"),
-        T_NO_PKEY_NAME, 200, 600, 10,
+        T_NO_PKEY_NAME, 200, 220, 10,
         makePlanBuilder().nodeType(NODE_BITMAP_INDEX_SCAN).build());
 
       testSeekAndNextEstimationBitmapScanHelper(stmt,
@@ -973,35 +976,35 @@ public void testSeekNextEstimationBitmapScanWithAnd() throws Exception {
 
       testSeekAndNextEstimationBitmapScanHelper(stmt,
         String.format(query, T_NO_PKEY_NAME, "k1 <= 10 AND k2 <= 10"),
-        T_NO_PKEY_NAME, 100, 300, 10,
+        T_NO_PKEY_NAME, 100, 120, 10,
         makePlanBuilder().nodeType(NODE_BITMAP_AND).build());
 
       testSeekAndNextEstimationBitmapScanHelper(stmt,
         String.format(query, T_NO_PKEY_NAME, "k1 <= 20 AND k2 <= 20"),
-        T_NO_PKEY_NAME, 400, 1200, 10,
+        T_NO_PKEY_NAME, 400, 420, 10,
         makePlanBuilder().nodeType(NODE_BITMAP_AND).build());
 
       testSeekAndNextEstimationBitmapScanHelper(stmt,
         String.format(query, T_NO_PKEY_NAME, "k1 <= 40 AND k2 <= 40"),
-        T_NO_PKEY_NAME, 1600, 4800, 10,
+        T_NO_PKEY_NAME, 1600, 1640, 10,
         makePlanBuilder().nodeType(NODE_BITMAP_AND).build());
 
       testSeekAndNextEstimationBitmapScanHelper(stmt,
         String.format(query, T_NO_PKEY_NAME, "k1 <= 80 AND k2 <= 80"),
-        T_NO_PKEY_NAME, 8000, 24000, 10,
+        T_NO_PKEY_NAME, 8000, 8160, 10,
         makePlanBuilder().nodeType(NODE_BITMAP_INDEX_SCAN).build());
 
       // If the two sets of ybctids are not similar sizes, it doesn't make sense
       // to collect both sets. Instead, we collect the smaller and filter out
       // the larger.
       testSeekAndNextEstimationBitmapScanHelper(stmt,
         String.format(query, T_NO_PKEY_NAME, "k1 <= 20 AND k2 <= 40"),
-        T_NO_PKEY_NAME, 2000, 6000, 10,
+        T_NO_PKEY_NAME, 2000, 2040, 10,
         makePlanBuilder().nodeType(NODE_BITMAP_INDEX_SCAN).indexName(T_K1_INDEX_NAME).build());
 
       testSeekAndNextEstimationBitmapScanHelper(stmt,
         String.format(query, T_NO_PKEY_NAME, "k1 <= 20 AND k2 <= 10"),
-        T_NO_PKEY_NAME, 1000, 3000, 10,
+        T_NO_PKEY_NAME, 1000, 1020, 10,
         makePlanBuilder().nodeType(NODE_BITMAP_INDEX_SCAN).indexName(T_K2_INDEX_NAME).build());
     }
   }

src/yb/docdb/conflict_resolution.cc

@@ -955,21 +955,21 @@ class StrongConflictChecker {
     if (PREDICT_FALSE(!FLAGS_TEST_file_to_dump_keys_checked_for_conflict_in_regular_db.empty())) {
       RETURN_NOT_OK(TEST_DumpRegularDbKeyToCheck(intent_key));
     }
-    if (!value_iter_.Initialized()) {
+    if (!value_iter_->Initialized()) {
       auto hybrid_time_file_filter =
           FLAGS_docdb_ht_filter_conflict_with_committed ? CreateHybridTimeFileFilter(read_time_)
                                                         : nullptr;
-      value_iter_ = CreateRocksDBIterator(
+      value_iter_ = OptimizedRocksDbIterator<BoundedRocksDbIterator>(CreateRocksDBIterator(
           resolver_.doc_db().regular,
           resolver_.doc_db().key_bounds,
           BloomFilterOptions::Variable(),
           rocksdb::kDefaultQueryId,
           hybrid_time_file_filter,
           /* iterate_upper_bound = */ nullptr,
-          rocksdb::CacheRestartBlockKeys::kFalse);
+          rocksdb::CacheRestartBlockKeys::kFalse));
     }
-    value_iter_.UpdateFilterKey(intent_key, Slice());
-    const auto* entry = &value_iter_.Seek(intent_key);
+    value_iter_->UpdateFilterKey(intent_key, Slice());
+    const auto* entry = &value_iter_->Seek(intent_key);
 
     VLOG_WITH_PREFIX_AND_FUNC(4)
         << "Overwrite; Seek: " << intent_key.ToDebugString() << " ("
@@ -1033,10 +1033,10 @@ class StrongConflictChecker {
       buffer_.Reset(existing_key);
       // Already have ValueType::kHybridTime at the end
       buffer_.AppendHybridTime(DocHybridTime::kMin);
-      entry = &ROCKSDB_SEEK(&value_iter_, buffer_.AsSlice());
+      entry = &ROCKSDB_SEEK(value_iter_, buffer_.AsSlice());
     }
 
-    return value_iter_.status();
+    return value_iter_->status();
   }
 
  private:
@@ -1059,7 +1059,7 @@ class StrongConflictChecker {
   KeyBytes& buffer_;
 
   // RocksDb iterator with bloom filter can be reused in case keys has same hash component.
-  BoundedRocksDbIterator value_iter_;
+  OptimizedRocksDbIterator<BoundedRocksDbIterator> value_iter_;
 };
 
 class ConflictResolverContextBase : public ConflictResolverContext {

src/yb/docdb/doc_operation-test.cc

@@ -18,6 +18,7 @@
 #include "yb/common/ql_value.h"
 #include "yb/common/transaction-test-util.h"
 
+#include "yb/docdb/bounded_rocksdb_iterator.h"
 #include "yb/docdb/cql_operation.h"
 #include "yb/docdb/doc_read_context.h"
 #include "yb/docdb/doc_rowwise_iterator.h"
@@ -396,8 +397,9 @@ TEST_F(DocOperationTest, TestRedisSetKVWithTTL) {
   // Read key from rocksdb.
   const auto doc_key = DocKey::FromRedisKey(123, "abc").Encode();
   rocksdb::ReadOptions read_opts;
-  auto iter = std::unique_ptr<rocksdb::Iterator>(db->NewIterator(read_opts));
-  ROCKSDB_SEEK(iter.get(), doc_key.AsSlice());
+  auto iter = OptimizedRocksDbIterator<BoundedRocksDbIterator>(
+      BoundedRocksDbIterator(db, read_opts, &KeyBounds::kNoBounds));
+  ROCKSDB_SEEK(iter, doc_key.AsSlice());
   ASSERT_TRUE(ASSERT_RESULT(iter->CheckedValid()));
 
   // Verify correct ttl.

src/yb/docdb/doc_rowwise_iterator.cc

@@ -142,7 +142,8 @@ Status DocRowwiseIterator::InitForTableType(
   CheckInitOnce();
   table_type_ = table_type;
   ignore_ttl_ = (table_type_ == TableType::PGSQL_TABLE_TYPE);
-  RETURN_NOT_OK(InitIterator(BloomFilterOptions::Inactive()));
+  RETURN_NOT_OK(
+      InitIterator(BloomFilterOptions::Inactive(), AvoidUselessNextInsteadOfSeek::kFalse));
 
   if (sub_doc_key.empty() || add_table_prefix_to_key) {
     dockv::DocKeyEncoder(&row_key_).Schema(*schema_);
@@ -160,7 +161,8 @@ Status DocRowwiseIterator::InitForTableType(
 
 Status DocRowwiseIterator::Init(
     const qlexpr::YQLScanSpec& doc_spec, SkipSeek skip_seek,
-    AllowVariableBloomFilter allow_variable_bloom_filter) {
+    AllowVariableBloomFilter allow_variable_bloom_filter,
+    AvoidUselessNextInsteadOfSeek avoid_useless_next_instead_of_seek) {
   table_type_ = doc_spec.client_type() == YQL_CLIENT_CQL ? TableType::YQL_TABLE_TYPE
                                                          : TableType::PGSQL_TABLE_TYPE;
   ignore_ttl_ = table_type_ == TableType::PGSQL_TABLE_TYPE;
@@ -210,7 +212,8 @@ Status DocRowwiseIterator::Init(
       allow_variable_bloom_filter));
 
   RETURN_NOT_OK(InitIterator(
-      scan_choices_->BloomFilterOptions(), doc_spec.QueryId(), CreateFileFilter(doc_spec)));
+      scan_choices_->BloomFilterOptions(), avoid_useless_next_instead_of_seek, doc_spec.QueryId(),
+      CreateFileFilter(doc_spec)));
 
   if (!skip_seek) {
     if (is_forward_scan_) {
@@ -504,6 +507,7 @@ Result<DocHybridTime> DocRowwiseIterator::GetTableTombstoneTime(Slice root_doc_k
 
 Status DocRowwiseIterator::InitIterator(
     const BloomFilterOptions& bloom_filter,
+    AvoidUselessNextInsteadOfSeek avoid_useless_next_instead_of_seek,
     const rocksdb::QueryId query_id,
     std::shared_ptr<rocksdb::ReadFileFilter> file_filter) {
   if (table_type_ == TableType::PGSQL_TABLE_TYPE) {
@@ -528,7 +532,8 @@ Status DocRowwiseIterator::InitIterator(
       read_operation_data_,
       file_filter,
       nullptr /* iterate_upper_bound */,
-      FastBackwardScan{use_fast_backward_scan_});
+      FastBackwardScan{use_fast_backward_scan_},
+      avoid_useless_next_instead_of_seek);
   InitResult();
 
   const auto scan_choices_has_upperbound =

src/yb/docdb/doc_rowwise_iterator.h

@@ -76,7 +76,9 @@ class DocRowwiseIterator final : public YQLRowwiseIteratorIf {
   // Init QL read scan.
   Status Init(
       const qlexpr::YQLScanSpec& spec, SkipSeek skip_seek = SkipSeek::kFalse,
-      AllowVariableBloomFilter allow_variable_bloom_filter = AllowVariableBloomFilter::kFalse);
+      AllowVariableBloomFilter allow_variable_bloom_filter = AllowVariableBloomFilter::kFalse,
+      AvoidUselessNextInsteadOfSeek avoid_useless_next_instead_of_seek =
+          AvoidUselessNextInsteadOfSeek::kFalse);
 
   bool IsFetchedRowStatic() const override;
 
@@ -175,6 +177,7 @@ class DocRowwiseIterator final : public YQLRowwiseIteratorIf {
 
   Status InitIterator(
       const BloomFilterOptions& bloom_filter,
+      AvoidUselessNextInsteadOfSeek avoid_useless_next_instead_of_seek,
       const rocksdb::QueryId query_id = rocksdb::kDefaultQueryId,
       std::shared_ptr<rocksdb::ReadFileFilter> file_filter = nullptr);