YugabyteDB (2.13.0.0-b42, bfc6a6643e7399ac8a0e81d06a3ee6d6571b33ab)

// Copyright (c) YugaByte, Inc.

//

// Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except

// in compliance with the License.  You may obtain a copy of the License at

//

// http://www.apache.org/licenses/LICENSE-2.0

//

// Unless required by applicable law or agreed to in writing, software distributed under the License

// is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express

// or implied.  See the License for the specific language governing permissions and limitations

// under the License.

//

#include <memory>

#include <string>

#include "yb/common/doc_hybrid_time.h"

#include "yb/common/hybrid_time.h"

#include "yb/common/ql_type.h"

#include "yb/common/ql_value.h"

#include "yb/docdb/consensus_frontier.h"

#include "yb/docdb/doc_key.h"

#include "yb/docdb/doc_reader.h"

#include "yb/docdb/doc_reader_redis.h"

#include "yb/docdb/docdb-internal.h"

#include "yb/docdb/docdb.h"

#include "yb/docdb/docdb.pb.h"

#include "yb/docdb/docdb_compaction_filter.h"

#include "yb/docdb/docdb_rocksdb_util.h"

#include "yb/docdb/docdb_test_base.h"

#include "yb/docdb/docdb_test_util.h"

#include "yb/docdb/in_mem_docdb.h"

#include "yb/docdb/primitive_value.h"

#include "yb/gutil/casts.h"

#include "yb/gutil/stringprintf.h"

#include "yb/gutil/walltime.h"

#include "yb/rocksdb/cache.h"

#include "yb/rocksdb/db.h"

#include "yb/server/hybrid_clock.h"

#include "yb/tablet/tablet_options.h"

#include "yb/util/debug-util.h"

#include "yb/util/minmax.h"

#include "yb/util/net/net_util.h"

#include "yb/util/random_util.h"

#include "yb/util/size_literals.h"

#include "yb/util/status_log.h"

#include "yb/util/string_trim.h"

#include "yb/util/strongly_typed_bool.h"

#include "yb/util/test_macros.h"

#include "yb/util/test_util.h"

#include "yb/util/tsan_util.h"

#include "yb/util/yb_partition.h"

using std::cout;

using std::endl;

using std::make_pair;

using std::map;

using std::string;

using std::unique_ptr;

using std::shared_ptr;

using std::make_shared;

using yb::util::TrimStr;

using yb::util::ApplyEagerLineContinuation;

using rocksdb::WriteOptions;

using namespace std::chrono_literals;

DECLARE_bool(use_docdb_aware_bloom_filter);

DECLARE_int32(max_nexts_to_avoid_seek);

DECLARE_bool(TEST_docdb_sort_weak_intents);

#define ASSERT_DOC_DB_DEBUG_DUMP_STR_EQ(str) ASSERT_NO_FATALS(AssertDocDbDebugDumpStrEq(str))

namespace yb {

namespace docdb {

CHECKED_STATUS GetPrimitiveValue(const rocksdb::UserBoundaryValues &values,

    size_t index,

    PrimitiveValue *out);

CHECKED_STATUS GetDocHybridTime(const rocksdb::UserBoundaryValues &values, DocHybridTime *out);

YB_STRONGLY_TYPED_BOOL(InitMarkerExpired);

YB_STRONGLY_TYPED_BOOL(UseIntermediateFlushes);

class DocDBTest : public DocDBTestBase {

 protected:

  DocDBTest() {

    SeedRandom();

  }

  ~DocDBTest() override {

  }

  virtual void GetSubDoc(

      const KeyBytes& subdoc_key, SubDocument* result, bool* found_result,

      const TransactionOperationContext& txn_op_context = TransactionOperationContext(),

      const ReadHybridTime& read_time = ReadHybridTime::Max()) = 0;

  // This is the baseline state of the database that we set up and come back to as we test various

  // operations.

  static constexpr const char *const kPredefinedDBStateDebugDumpStr = R"#(

SubDocKey(DocKey([], ["my_key_where_value_is_a_string"]), [HT{ physical: 1000 }]) -> "value1"

SubDocKey(DocKey([], ["mydockey", 123456]), [HT{ physical: 2000 }]) -> {}

SubDocKey(DocKey([], ["mydockey", 123456]), ["subkey_a"; HT{ physical: 2000 w: 1 }]) -> "value_a"

SubDocKey(DocKey([], ["mydockey", 123456]), ["subkey_b"; HT{ physical: 7000 }]) -> {}

SubDocKey(DocKey([], ["mydockey", 123456]), ["subkey_b"; HT{ physical: 6000 }]) -> DEL

SubDocKey(DocKey([], ["mydockey", 123456]), ["subkey_b"; HT{ physical: 3000 }]) -> {}

SubDocKey(DocKey([], ["mydockey", 123456]), ["subkey_b", "subkey_c"; HT{ physical: 7000 w: 1 }]) \

    -> "value_bc_prime"

SubDocKey(DocKey([], ["mydockey", 123456]), ["subkey_b", "subkey_c"; HT{ physical: 5000 }]) -> DEL

SubDocKey(DocKey([], ["mydockey", 123456]), ["subkey_b", "subkey_c"; HT{ physical: 3000 w: 1 }]) \

    -> "value_bc"

SubDocKey(DocKey([], ["mydockey", 123456]), ["subkey_b", "subkey_d"; HT{ physical: 3500 }]) -> \

    "value_bd"

      )#";

  static const DocKey kDocKey1;

  static const DocKey kDocKey2;

  static const KeyBytes kEncodedDocKey1;

  static const KeyBytes kEncodedDocKey2;

  void TestInsertion(

      DocPath doc_path,

      const PrimitiveValue &value,

      HybridTime hybrid_time,

      string expected_write_batch_str);

  void TestDeletion(

      DocPath doc_path,

      HybridTime hybrid_time,

      string expected_write_batch_str);

  void SetupRocksDBState(KeyBytes encoded_doc_key) {

    SubDocument root;

    SubDocument a, b, c, d, e, f, b2;

    // The test plan below:

    // Set root = {a: {1: 1, 2: 2}, b: {c: {1: 3}, d: {1: 5, 2: 6}}, u: 7}

    // Then set root.a.2 = 11

    // Then replace root.b = {e: {1: 8, 2: 9}, y: 10}

    // Then extend root.a by {1: 3, 3: 4}

    // Then Delete root.b.e.2

    // The end result should be {a: {1: 3, 2: 11, 3: 4, x: {}}, b: {e: {1: 8}, y: 10}, u: 7}

#define SET_CHILD(parent, child) parent.SetChild(PrimitiveValue(#child), std::move(child))

#define SET_VALUE(parent, key, value) parent.SetChild(PrimitiveValue(key), \

                                                      SubDocument(PrimitiveValue(value)))

    // Constructing top level document: "root"

    SET_VALUE(root, "u", "7");

    SET_VALUE(a, "1", "1");

    SET_VALUE(a, "2", "2");

    SET_VALUE(c, "1", "3");

    SET_VALUE(d, "1", "5");

    SET_VALUE(d, "2", "6");

    SET_CHILD(b, c);

    SET_CHILD(b, d);

    SET_CHILD(root, a);

    SET_CHILD(root, b);

    EXPECT_STR_EQ_VERBOSE_TRIMMED(R"#(

        {

          "a": {

            "1": "1",

            "2": "2"

          },

          "b": {

            "c": {

              "1": "3"

            },

            "d": {

              "1": "5",

              "2": "6"

            }

          },

          "u": "7"

        }

        )#", root.ToString());

    // Constructing new version of b = b2 to be inserted later.

    SET_VALUE(b2, "y", "10");

    SET_VALUE(e, "1", "8");

    SET_VALUE(e, "2", "9");

    SET_CHILD(b2, e);

    EXPECT_STR_EQ_VERBOSE_TRIMMED(R"#(

{

  "e": {

    "1": "8",

    "2": "9"

  },

  "y": "10"

}

      )#", b2.ToString());

    // Constructing a doc with which we will extend a later

    SET_VALUE(f, "1", "3");

    SET_VALUE(f, "3", "4");

    EXPECT_STR_EQ_VERBOSE_TRIMMED(R"#(

{

  "1": "3",

  "3": "4"

}

      )#", f.ToString());

#undef SET_CHILD

#undef SET_VALUE

    ASSERT_OK(InsertSubDocument(

        DocPath(encoded_doc_key), root, 1000_usec_ht));

    // The Insert above could have been an Extend with no difference in external behavior.

    // Internally however, an insert writes an extra key (with value tombstone).

    ASSERT_OK(SetPrimitive(

        DocPath(encoded_doc_key, PrimitiveValue("a"), PrimitiveValue("2")),

        Value(PrimitiveValue(11)), 2000_usec_ht));

    ASSERT_OK(InsertSubDocument(DocPath(encoded_doc_key, PrimitiveValue("b")), b2,

                                3000_usec_ht));

    ASSERT_OK(ExtendSubDocument(DocPath(encoded_doc_key, PrimitiveValue("a")), f,

                                4000_usec_ht));

    ASSERT_OK(SetPrimitive(

        DocPath(encoded_doc_key, PrimitiveValue("b"), PrimitiveValue("e"), PrimitiveValue("2")),

        Value(PrimitiveValue::kTombstone), 5000_usec_ht));

  }

  void VerifySubDocument(SubDocKey subdoc_key, HybridTime ht, string subdoc_string) {

    SubDocument doc_from_rocksdb;

    bool subdoc_found_in_rocksdb = false;

    SCOPED_TRACE("\n" + GetStackTrace(StackTraceLineFormat::CLION_CLICKABLE) + "\n" +

        DocDBDebugDumpToStr());

    // TODO(dtxn) - check both transaction and non-transaction path?

    // https://yugabyte.atlassian.net/browse/ENG-2177

    auto encoded_subdoc_key = subdoc_key.EncodeWithoutHt();

    GetSubDoc(

        encoded_subdoc_key, &doc_from_rocksdb, &subdoc_found_in_rocksdb,

        kNonTransactionalOperationContext, ReadHybridTime::SingleTime(ht));

    if (subdoc_string.empty()) {

      EXPECT_FALSE(subdoc_found_in_rocksdb);

      return;

    }

    EXPECT_TRUE(subdoc_found_in_rocksdb);

    EXPECT_STR_EQ_VERBOSE_TRIMMED(subdoc_string, doc_from_rocksdb.ToString());

  }

  // Tries to read some documents from the DB that is assumed to be in a state described by

  // kPredefinedDBStateDebugDumpStr, and verifies the result of those reads. Only the latest logical

  // state of documents matters for this check, so it is OK to call this after compacting previous

  // history.

  void CheckExpectedLatestDBState();

  // Checks bloom filter useful counter increment to be in range [1;expected_max_increment] and

  // table iterators number increment to be expected_num_iterators_increment.

  // Updates total_useful, total_iterators

  void CheckBloom(const int expected_max_increment, uint64_t *total_useful,

                  const int expected_num_iterators_increment, uint64_t *total_iterators) {

    if (FLAGS_use_docdb_aware_bloom_filter) {

      const auto total_useful_updated =

          regular_db_options().statistics->getTickerCount(rocksdb::BLOOM_FILTER_USEFUL);

      const auto total_iterators_updated =

          regular_db_options().statistics->getTickerCount(rocksdb::NO_TABLE_CACHE_ITERATORS);

      if (expected_max_increment > 0) {

        ASSERT_GT(total_useful_updated, *total_useful);

        ASSERT_LE(total_useful_updated, *total_useful + expected_max_increment);

        *total_useful = total_useful_updated;

      } else {

        ASSERT_EQ(*total_useful, total_useful_updated);

      }

      ASSERT_EQ(*total_iterators + expected_num_iterators_increment, total_iterators_updated);

      *total_iterators = total_iterators_updated;

    }

  }

  InetAddress GetInetAddress(const std::string &strval) {

    return InetAddress(CHECK_RESULT(ParseIpAddress(strval)));

  }

  void InsertInet(const std::string strval) {

    const DocKey doc_key(PrimitiveValues("mydockey"));

    KeyBytes encoded_doc_key(doc_key.Encode());

    ASSERT_OK(SetPrimitive(DocPath(encoded_doc_key, PrimitiveValue(GetInetAddress(strval))),

                           PrimitiveValue(),

                           1000_usec_ht));

  }

  // Inserts a bunch of subkeys starting with the provided doc key. It also, fills out the

  // expected_docdb_str with the expected state of DocDB after the operation.

  void AddSubKeys(const KeyBytes& encoded_doc_key, int num_subkeys, int base,

                  string* expected_docdb_str) {

    *expected_docdb_str = "";

    for (int i = 0; i < num_subkeys; i++) {

      string subkey = "subkey" + std::to_string(base + i);

      string value = "value" + std::to_string(i);

      MicrosTime hybrid_time = (i + 1) * 1000;

      ASSERT_OK(SetPrimitive(DocPath(encoded_doc_key, PrimitiveValue(subkey)),

                             Value(PrimitiveValue(value)), HybridTime::FromMicros(hybrid_time)));

      *expected_docdb_str += strings::Substitute(

          R"#(SubDocKey(DocKey([], ["key"]), ["$0"; HT{ physical: $1 }]) -> "$2")#",

          subkey, hybrid_time, value);

      *expected_docdb_str += "\n";

    }

  }

  static constexpr int kNumSubKeysForCollectionsWithTTL = 3;

  void SetUpCollectionWithTTL(DocKey collection_key, UseIntermediateFlushes intermediate_flushes) {

    SubDocument subdoc;

    for (int i = 0; i < kNumSubKeysForCollectionsWithTTL; i++) {

      string key = "k" + std::to_string(i);

      string value = "v" + std::to_string(i);

      subdoc.SetChildPrimitive(PrimitiveValue(key), PrimitiveValue(value));

    }

    ASSERT_OK(InsertSubDocument(DocPath(collection_key.Encode()), subdoc, 1000_usec_ht, 10s));

    ASSERT_DOC_DB_DEBUG_DUMP_STR_EQ(Format(R"#(

        SubDocKey($0, [HT{ physical: 1000 }]) -> {}; ttl: 10.000s

        SubDocKey($0, ["k0"; HT{ physical: 1000 w: 1 }]) -> "v0"; ttl: 10.000s

        SubDocKey($0, ["k1"; HT{ physical: 1000 w: 2 }]) -> "v1"; ttl: 10.000s

        SubDocKey($0, ["k2"; HT{ physical: 1000 w: 3 }]) -> "v2"; ttl: 10.000s

        )#", collection_key.ToString()));

    if (intermediate_flushes) {

      ASSERT_OK(FlushRocksDbAndWait());

    }

    // Set separate TTLs for each element.

    for (int i = 0; i < kNumSubKeysForCollectionsWithTTL; i++) {

      SubDocument subdoc;

      string key = "k" + std::to_string(i);

      string value = "vv" + std::to_string(i);

      subdoc.SetChildPrimitive(PrimitiveValue(key), PrimitiveValue(value));

      ASSERT_OK(ExtendSubDocument(

          DocPath(collection_key.Encode()), subdoc, 1100_usec_ht,

          MonoDelta::FromSeconds(20 + i)));

      if (intermediate_flushes) {

        ASSERT_OK(FlushRocksDbAndWait());

      }

    }

    // Add new keys as well.

    for (int i = kNumSubKeysForCollectionsWithTTL; i < kNumSubKeysForCollectionsWithTTL * 2; i++) {

      SubDocument subdoc;

      string key = "k" + std::to_string(i);

      string value = "vv" + std::to_string(i);

      subdoc.SetChildPrimitive(PrimitiveValue(key), PrimitiveValue(value));

      ASSERT_OK(ExtendSubDocument(

          DocPath(collection_key.Encode()), subdoc, 1100_usec_ht,

          MonoDelta::FromSeconds(20 + i)));

      if (intermediate_flushes) {

        ASSERT_OK(FlushRocksDbAndWait());

      }

    }

  }

  string ExpectedDebugDumpForCollectionWithTTL(DocKey collection_key,

                                               InitMarkerExpired init_marker_expired) {

    // The "file ..." comments below are for the case of intermediate_flushes = true above.

    const string result_template = init_marker_expired ?

        // After the init marker expires, we should not see a tombstone for it. We do not replace

        // timed-out collection init markers with tombstones on minor compactions, because that

        // could hide keys that

        R"#(

            SubDocKey($0, ["k0"; HT{ physical: 1100 }]) -> "vv0"; ttl: 20.000s

            SubDocKey($0, ["k1"; HT{ physical: 1100 }]) -> "vv1"; ttl: 21.000s

            SubDocKey($0, ["k2"; HT{ physical: 1100 }]) -> "vv2"; ttl: 22.000s

            SubDocKey($0, ["k3"; HT{ physical: 1100 }]) -> "vv3"; ttl: 23.000s

            SubDocKey($0, ["k4"; HT{ physical: 1100 }]) -> "vv4"; ttl: 24.000s

            SubDocKey($0, ["k5"; HT{ physical: 1100 }]) -> "vv5"; ttl: 25.000s

        )#" : R"#(

            SubDocKey($0, [HT{ physical: 1000 }]) -> {}; ttl: 10.000s               // file 1

            SubDocKey($0, ["k0"; HT{ physical: 1100 }]) -> "vv0"; ttl: 20.000s      // file 2

            SubDocKey($0, ["k0"; HT{ physical: 1000 w: 1 }]) -> "v0"; ttl: 10.000s  // file 1

            SubDocKey($0, ["k1"; HT{ physical: 1100 }]) -> "vv1"; ttl: 21.000s      // file 3

            SubDocKey($0, ["k1"; HT{ physical: 1000 w: 2 }]) -> "v1"; ttl: 10.000s  // file 1

            SubDocKey($0, ["k2"; HT{ physical: 1100 }]) -> "vv2"; ttl: 22.000s      // file 4

            SubDocKey($0, ["k2"; HT{ physical: 1000 w: 3 }]) -> "v2"; ttl: 10.000s  // file 1

            SubDocKey($0, ["k3"; HT{ physical: 1100 }]) -> "vv3"; ttl: 23.000s      // file 5

            SubDocKey($0, ["k4"; HT{ physical: 1100 }]) -> "vv4"; ttl: 24.000s      // file 6

            SubDocKey($0, ["k5"; HT{ physical: 1100 }]) -> "vv5"; ttl: 25.000s      // file 7

        )#";

    return Format(result_template, collection_key.ToString());

  }

  void InitializeCollection(const std::string& key_string,

                            std::string* val_string,

                            vector<HybridTime>::iterator* time_iter,

                            std::set<std::pair<string, string>>* docdb_dump) {

    SubDocument subdoc;

    DocKey collection_key(PrimitiveValues(key_string));

    for (int i = 0; i < kNumSubKeysForCollectionsWithTTL; i++) {

      string key = "sk" + std::to_string(i);

      subdoc.SetChildPrimitive(PrimitiveValue(key), PrimitiveValue(*val_string));

      (*val_string)[1]++;

    }

    ASSERT_OK(InsertSubDocument(DocPath(collection_key.Encode()), subdoc, **time_iter));

    ++*time_iter;

    SubDocument new_subdoc;

    // Add new keys as well.

    for (int i = kNumSubKeysForCollectionsWithTTL / 2;

         i < 3 * kNumSubKeysForCollectionsWithTTL / 2; i++) {

      string key = "sk" + std::to_string(i);

      new_subdoc.SetChildPrimitive(PrimitiveValue(key), PrimitiveValue(*val_string));

      (*val_string)[1]++;

    }

    ASSERT_OK(ExtendSubDocument(

      DocPath(collection_key.Encode()), new_subdoc, **time_iter));

    ++*time_iter;

  }

};

void GetSubDocQl(

      const DocDB& doc_db, const KeyBytes& subdoc_key, SubDocument* result, bool* found_result,

      const TransactionOperationContext& txn_op_context, const ReadHybridTime& read_time,

      const vector<PrimitiveValue>* projection = nullptr) {

  auto doc_from_rocksdb_opt = ASSERT_RESULT(TEST_GetSubDocument(

    subdoc_key, doc_db, rocksdb::kDefaultQueryId, txn_op_context,

    CoarseTimePoint::max() /* deadline */, read_time, projection));

  if (doc_from_rocksdb_opt) {

    *found_result = true;

    *result = *doc_from_rocksdb_opt;

  } else {

    *found_result = false;

    *result = SubDocument();

  }

}

void GetSubDocRedis(

      const DocDB& doc_db, const KeyBytes& subdoc_key, SubDocument* result, bool* found_result,

      const TransactionOperationContext& txn_op_context, const ReadHybridTime& read_time) {

  GetRedisSubDocumentData data = { subdoc_key, result, found_result };

  ASSERT_OK(GetRedisSubDocument(

      doc_db, data, rocksdb::kDefaultQueryId,

      txn_op_context, CoarseTimePoint::max() /* deadline */,

      read_time));

}

// The list of types we want to test.

YB_DEFINE_ENUM(TestDocDb, (kQlReader)(kRedisReader));

class DocDBTestWrapper : public DocDBTest, public testing::WithParamInterface<TestDocDb>  {

 public:

  void GetSubDoc(

      const KeyBytes& subdoc_key, SubDocument* result, bool* found_result,

      const TransactionOperationContext& txn_op_context = TransactionOperationContext(),

      const ReadHybridTime& read_time = ReadHybridTime::Max()) override {

    switch (GetParam()) {

      case TestDocDb::kQlReader: {

        GetSubDocQl(doc_db(), subdoc_key, result, found_result, txn_op_context, read_time);

        break;

      }

      case TestDocDb::kRedisReader: {

        GetSubDocRedis(

            doc_db(), subdoc_key, result, found_result, txn_op_context, read_time);

        break;

      }

    }

  }

};

INSTANTIATE_TEST_CASE_P(DocDBTests,

                        DocDBTestWrapper,

                        testing::Values(TestDocDb::kQlReader, TestDocDb::kRedisReader));

class DocDBTestQl : public DocDBTest {

 public:

  void GetSubDoc(

      const KeyBytes& subdoc_key, SubDocument* result, bool* found_result,

      const TransactionOperationContext& txn_op_context = TransactionOperationContext(),

      const ReadHybridTime& read_time = ReadHybridTime::Max()) override {

    GetSubDocQl(doc_db(), subdoc_key, result, found_result, txn_op_context, read_time);

  }

};

class DocDBTestRedis : public DocDBTest {

 public:

  void GetSubDoc(

      const KeyBytes& subdoc_key, SubDocument* result, bool* found_result,

      const TransactionOperationContext& txn_op_context = TransactionOperationContext(),

      const ReadHybridTime& read_time = ReadHybridTime::Max()) override {

    GetSubDocRedis(

        doc_db(), subdoc_key, result, found_result, txn_op_context, read_time);

  }

};

// Static constant initialization should be completely independent (cannot initialize one using the

// other).

const DocKey DocDBTest::kDocKey1(PrimitiveValues("row1", 11111));

const DocKey DocDBTest::kDocKey2(PrimitiveValues("row2", 22222));

const KeyBytes DocDBTest::kEncodedDocKey1(DocKey(PrimitiveValues("row1", 11111)).Encode());

const KeyBytes DocDBTest::kEncodedDocKey2(DocKey(PrimitiveValues("row2", 22222)).Encode());

void DocDBTest::TestInsertion(

    const DocPath doc_path,

    const PrimitiveValue &value,

    HybridTime hybrid_time,

    string expected_write_batch_str) {

  auto dwb = MakeDocWriteBatch();

  // Set write id to zero on the write path.

  ASSERT_OK(dwb.SetPrimitive(doc_path, value));

  ASSERT_OK(WriteToRocksDB(dwb, hybrid_time));

  string dwb_str;

  ASSERT_OK(FormatDocWriteBatch(dwb, &dwb_str));

  EXPECT_STR_EQ_VERBOSE_TRIMMED(ApplyEagerLineContinuation(expected_write_batch_str),

      dwb_str);

}

void DocDBTest::TestDeletion(

    DocPath doc_path,

    HybridTime hybrid_time,

    string expected_write_batch_str) {

  auto dwb = MakeDocWriteBatch();

  // Set write id to zero on the write path.

  ASSERT_OK(dwb.DeleteSubDoc(doc_path));

  ASSERT_OK(WriteToRocksDB(dwb, hybrid_time));

  string dwb_str;

  ASSERT_OK(FormatDocWriteBatch(dwb, &dwb_str));

  EXPECT_STR_EQ_VERBOSE_TRIMMED(ApplyEagerLineContinuation(expected_write_batch_str),

      dwb_str);

}

void DocDBTest::CheckExpectedLatestDBState() {

  const SubDocKey subdoc_key(DocKey(PrimitiveValues("mydockey", 123456)));

  SubDocument subdoc;

  bool doc_found = false;

  // TODO(dtxn) - check both transaction and non-transaction path?

  auto encoded_subdoc_key = subdoc_key.EncodeWithoutHt();

  GetSubDoc(

      encoded_subdoc_key, &subdoc, &doc_found,

      kNonTransactionalOperationContext);

  ASSERT_TRUE(doc_found);

  ASSERT_STR_EQ_VERBOSE_TRIMMED(

      R"#(

{

  "subkey_a": "value_a",

  "subkey_b": {

    "subkey_c": "value_bc_prime"

  }

}

      )#",

      subdoc.ToString()

  );

}

// ------------------------------------------------------------------------------------------------

TEST_P(DocDBTestWrapper, DocPathTest) {

  DocKey doc_key(PrimitiveValues("mydockey", 10, "mydockey", 20));

  DocPath doc_path(doc_key.Encode(), "first_subkey", 123);

  ASSERT_EQ(2, doc_path.num_subkeys());

  ASSERT_EQ("\"first_subkey\"", doc_path.subkey(0).ToString());

  ASSERT_EQ("123", doc_path.subkey(1).ToString());

}

TEST_P(DocDBTestWrapper, KeyAsEmptyObjectIsNotMasked) {

  const DocKey doc_key(PrimitiveValues(DocKeyHash(1234)));

  KeyBytes encoded_doc_key(doc_key.Encode());

  ASSERT_OK(SetPrimitive(

      DocPath(encoded_doc_key), PrimitiveValue::kObject, 252_usec_ht));

  ASSERT_OK(SetPrimitive(

      DocPath(encoded_doc_key, PrimitiveValue()), PrimitiveValue::kObject, 617_usec_ht));

  ASSERT_OK(DeleteSubDoc(

      DocPath(encoded_doc_key, PrimitiveValue(), PrimitiveValue(ValueType::kFalse)), 675_usec_ht));

  ASSERT_OK(SetPrimitive(

      DocPath(encoded_doc_key, PrimitiveValue(), PrimitiveValue(ValueType::kFalse)),

      PrimitiveValue(12345), 617_usec_ht));

  ASSERT_OK(SetPrimitive(

      DocPath(encoded_doc_key, "later"), PrimitiveValue(1), 336_usec_ht));

  ASSERT_DOC_DB_DEBUG_DUMP_STR_EQ(R"#(

      SubDocKey(DocKey([], [1234]), [HT{ physical: 252 }]) -> {}

      SubDocKey(DocKey([], [1234]), [null; HT{ physical: 617 }]) -> {}

      SubDocKey(DocKey([], [1234]), [null, false; HT{ physical: 675 }]) -> DEL

      SubDocKey(DocKey([], [1234]), [null, false; HT{ physical: 617 }]) -> 12345

      SubDocKey(DocKey([], [1234]), ["later"; HT{ physical: 336 }]) -> 1

      )#");

  VerifySubDocument(SubDocKey(doc_key), 4000_usec_ht,

                    R"#(

{

  null: {},

  "later": 1

}

      )#");

}

TEST_P(DocDBTestWrapper, NullChildObjectShouldMaskValues) {

  const DocKey doc_key(PrimitiveValues("mydockey", 123456));

  KeyBytes encoded_doc_key(doc_key.Encode());

  ASSERT_OK(SetPrimitive(

      DocPath(encoded_doc_key), PrimitiveValue::kObject, 1000_usec_ht));

  ASSERT_OK(SetPrimitive(

      DocPath(encoded_doc_key, "obj"), PrimitiveValue::kObject, 2000_usec_ht));

  ASSERT_OK(SetPrimitive(

      DocPath(encoded_doc_key, "obj", "key"), PrimitiveValue("value"), 2000_usec_ht));

  ASSERT_OK(SetPrimitive(

      DocPath(encoded_doc_key, "obj"), PrimitiveValue(ValueType::kNullHigh), 3000_usec_ht));

  ASSERT_DOC_DB_DEBUG_DUMP_STR_EQ(R"#(

      SubDocKey(DocKey([], ["mydockey", 123456]), [HT{ physical: 1000 }]) -> {}

      SubDocKey(DocKey([], ["mydockey", 123456]), ["obj"; HT{ physical: 3000 }]) -> null

      SubDocKey(DocKey([], ["mydockey", 123456]), ["obj"; HT{ physical: 2000 }]) -> {}

      SubDocKey(DocKey([], ["mydockey", 123456]), ["obj", "key"; HT{ physical: 2000 }]) -> "value"

      )#");

  VerifySubDocument(SubDocKey(doc_key), 4000_usec_ht,

                    R"#(

{

  "obj": null

}

      )#");

}

// This test confirms that we return the appropriate value for doc_found in the case that the last

// projection we look at is not present. Previously we had a bug where we would set doc_found to

// true if the last projection was present, and false otherwise, reguardless of other projections

// considered. This test ensures we have the correct behavior, returning true as long as any

// projection is present, even if the last one is absent.

TEST_F(DocDBTestQl, LastProjectionIsNull) {

  const DocKey doc_key(PrimitiveValues("mydockey", 123456));

  KeyBytes encoded_doc_key(doc_key.Encode());

  ASSERT_OK(SetPrimitive(

      DocPath(encoded_doc_key), PrimitiveValue::kObject, 1000_usec_ht));

  ASSERT_OK(SetPrimitive(

      DocPath(encoded_doc_key, "p1"), PrimitiveValue("value"), 2000_usec_ht));

  ASSERT_OK(SetPrimitive(

      DocPath(encoded_doc_key, "p2"), PrimitiveValue(ValueType::kTombstone), 2000_usec_ht));

  ASSERT_DOC_DB_DEBUG_DUMP_STR_EQ(R"#(

      SubDocKey(DocKey([], ["mydockey", 123456]), [HT{ physical: 1000 }]) -> {}

      SubDocKey(DocKey([], ["mydockey", 123456]), ["p1"; HT{ physical: 2000 }]) -> "value"

      SubDocKey(DocKey([], ["mydockey", 123456]), ["p2"; HT{ physical: 2000 }]) -> DEL

      )#");

  auto subdoc_key = SubDocKey(doc_key);

  auto encoded_subdoc_key = subdoc_key.EncodeWithoutHt();

  SubDocument doc_from_rocksdb;

  bool subdoc_found_in_rocksdb = false;

  const vector<PrimitiveValue> projection = {

    PrimitiveValue("p1"),

    PrimitiveValue("p2")

  };

  GetSubDocQl(

      doc_db(), encoded_subdoc_key, &doc_from_rocksdb, &subdoc_found_in_rocksdb,

      kNonTransactionalOperationContext, ReadHybridTime::SingleTime(4000_usec_ht),

      &projection);

  EXPECT_TRUE(subdoc_found_in_rocksdb);

  EXPECT_STR_EQ_VERBOSE_TRIMMED(R"#(

{

  "p1": "value",

  "p2": DEL

}

  )#", doc_from_rocksdb.ToString());

}

TEST_F(DocDBTestQl, ColocatedTableTombstoneTest) {

  constexpr PgTableOid pgtable_id(0x4001);

  DocKey doc_key_1(PrimitiveValues("mydockey", 123456));

  doc_key_1.set_pgtable_id(pgtable_id);

  DocKey doc_key_2(PrimitiveValues("mydockey", 789123));

  doc_key_2.set_pgtable_id(pgtable_id);

  ASSERT_OK(SetPrimitive(

      doc_key_1.Encode(), PrimitiveValue(1), 1000_usec_ht));

  ASSERT_OK(SetPrimitive(

      doc_key_2.Encode(), PrimitiveValue(2), 1000_usec_ht));

  DocKey doc_key_table;

  doc_key_table.set_pgtable_id(pgtable_id);

  ASSERT_OK(DeleteSubDoc(

      doc_key_table.Encode(), 2000_usec_ht));

  ASSERT_DOC_DB_DEBUG_DUMP_STR_EQ(R"#(

      SubDocKey(DocKey(PgTableId=16385, [], []), [HT{ physical: 2000 }]) -> DEL

      SubDocKey(DocKey(PgTableId=16385, [], ["mydockey", 123456]), [HT{ physical: 1000 }]) -> 1

      SubDocKey(DocKey(PgTableId=16385, [], ["mydockey", 789123]), [HT{ physical: 1000 }]) -> 2

      )#");

  VerifySubDocument(SubDocKey(doc_key_1), 4000_usec_ht, "");

  VerifySubDocument(SubDocKey(doc_key_1), 1500_usec_ht, "1");

}

TEST_P(DocDBTestWrapper, HistoryCompactionFirstRowHandlingRegression) {

  // A regression test for a bug in an initial version of compaction cleanup.

  const DocKey doc_key(PrimitiveValues("mydockey", 123456));

  KeyBytes encoded_doc_key(doc_key.Encode());

  ASSERT_OK(SetPrimitive(

      DocPath(encoded_doc_key), PrimitiveValue::kObject, 1000_usec_ht));

  ASSERT_OK(SetPrimitive(

      DocPath(encoded_doc_key, "subkey1"),

      PrimitiveValue("value1"),

      1000_usec_ht));

  ASSERT_OK(SetPrimitive(

      DocPath(encoded_doc_key, "subkey1"),

      PrimitiveValue("value2"),

      2000_usec_ht));

  ASSERT_OK(SetPrimitive(

      DocPath(encoded_doc_key, "subkey1"),

      PrimitiveValue("value3"),

      3000_usec_ht));

  ASSERT_OK(SetPrimitive(

      DocPath(encoded_doc_key), PrimitiveValue::kObject, 4000_usec_ht));

  ASSERT_DOC_DB_DEBUG_DUMP_STR_EQ(R"#(

      SubDocKey(DocKey([], ["mydockey", 123456]), [HT{ physical: 4000 }]) -> {}

      SubDocKey(DocKey([], ["mydockey", 123456]), [HT{ physical: 1000 }]) -> {}

      SubDocKey(DocKey([], ["mydockey", 123456]), ["subkey1"; HT{ physical: 3000 }]) -> "value3"

      SubDocKey(DocKey([], ["mydockey", 123456]), ["subkey1"; HT{ physical: 2000 }]) -> "value2"

      SubDocKey(DocKey([], ["mydockey", 123456]), ["subkey1"; HT{ physical: 1000 }]) -> "value1"

      )#");

  FullyCompactHistoryBefore(3500_usec_ht);

  ASSERT_DOC_DB_DEBUG_DUMP_STR_EQ(

      R"#(

SubDocKey(DocKey([], ["mydockey", 123456]), [HT{ physical: 4000 }]) -> {}

SubDocKey(DocKey([], ["mydockey", 123456]), [HT{ physical: 1000 }]) -> {}

SubDocKey(DocKey([], ["mydockey", 123456]), ["subkey1"; HT{ physical: 3000 }]) -> "value3"

      )#");

}

TEST_P(DocDBTestWrapper, SetPrimitiveQL) {

  const DocKey doc_key(PrimitiveValues("mydockey", 123456));

  SetupRocksDBState(doc_key.Encode());

  ASSERT_DOC_DB_DEBUG_DUMP_STR_EQ(

      R"#(

SubDocKey(DocKey([], ["mydockey", 123456]), [HT{ physical: 1000 }]) -> {}

SubDocKey(DocKey([], ["mydockey", 123456]), ["a", "1"; HT{ physical: 4000 }]) -> "3"

SubDocKey(DocKey([], ["mydockey", 123456]), ["a", "1"; HT{ physical: 1000 w: 1 }]) -> "1"

SubDocKey(DocKey([], ["mydockey", 123456]), ["a", "2"; HT{ physical: 2000 }]) -> 11

SubDocKey(DocKey([], ["mydockey", 123456]), ["a", "2"; HT{ physical: 1000 w: 2 }]) -> "2"

SubDocKey(DocKey([], ["mydockey", 123456]), ["a", "3"; HT{ physical: 4000 w: 1 }]) -> "4"

SubDocKey(DocKey([], ["mydockey", 123456]), ["b"; HT{ physical: 3000 }]) -> {}

SubDocKey(DocKey([], ["mydockey", 123456]), ["b", "c", "1"; HT{ physical: 1000 w: 3 }]) -> "3"

SubDocKey(DocKey([], ["mydockey", 123456]), ["b", "d", "1"; HT{ physical: 1000 w: 4 }]) -> "5"

SubDocKey(DocKey([], ["mydockey", 123456]), ["b", "d", "2"; HT{ physical: 1000 w: 5 }]) -> "6"

SubDocKey(DocKey([], ["mydockey", 123456]), ["b", "e", "1"; HT{ physical: 3000 w: 1 }]) -> "8"

SubDocKey(DocKey([], ["mydockey", 123456]), ["b", "e", "2"; HT{ physical: 5000 }]) -> DEL

SubDocKey(DocKey([], ["mydockey", 123456]), ["b", "e", "2"; HT{ physical: 3000 w: 2 }]) -> "9"

SubDocKey(DocKey([], ["mydockey", 123456]), ["b", "y"; HT{ physical: 3000 w: 3 }]) -> "10"

SubDocKey(DocKey([], ["mydockey", 123456]), ["u"; HT{ physical: 1000 w: 6 }]) -> "7"

     )#");

}

// This tests reads on data without init markers. Basic Test tests with init markers.

TEST_P(DocDBTestWrapper, GetSubDocumentTest) {

  const DocKey doc_key(PrimitiveValues("mydockey", 123456));

  SetupRocksDBState(doc_key.Encode());

  // We will test the state of the entire document after every operation, using timestamps

  // 500, 1500, 2500, 3500, 4500, 5500.

  VerifySubDocument(SubDocKey(doc_key), 500_usec_ht, "");

  VerifySubDocument(SubDocKey(doc_key), 1500_usec_ht,

                    R"#(

{

  "a": {

    "1": "1",

    "2": "2"

  },

  "b": {

    "c": {

      "1": "3"

    },

    "d": {

      "1": "5",

      "2": "6"

    }

  },

  "u": "7"

}

      )#");

  VerifySubDocument(SubDocKey(doc_key), 2500_usec_ht,

                    R"#(

{

  "a": {

    "1": "1",

    "2": 11

  },

  "b": {

    "c": {

      "1": "3"

    },

    "d": {

      "1": "5",

      "2": "6"

    }

  },

  "u": "7"

}

      )#");

  VerifySubDocument(SubDocKey(doc_key), 3500_usec_ht,

                    R"#(

{

  "a": {

    "1": "1",

    "2": 11

  },

  "b": {

    "e": {

      "1": "8",

      "2": "9"

    },

    "y": "10"

  },

  "u": "7"

}

      )#");

  VerifySubDocument(SubDocKey(doc_key), 4500_usec_ht,

                    R"#(

{

  "a": {

    "1": "3",

    "2": 11,

    "3": "4"

  },

  "b": {

    "e": {

      "1": "8",

      "2": "9"

    },

    "y": "10"

  },

  "u": "7"

}

      )#");

  VerifySubDocument(SubDocKey(doc_key), 5500_usec_ht,

                    R"#(

{

  "a": {

    "1": "3",

    "2": 11,

    "3": "4"

  },

  "b": {

    "e": {

      "1": "8"

    },

    "y": "10"

  },

  "u": "7"

}

      )#");

  // Test the evolution of SubDoc root.b at various timestamps.

  VerifySubDocument(SubDocKey(doc_key, PrimitiveValue("b")), 500_usec_ht, "");

  VerifySubDocument(SubDocKey(doc_key, PrimitiveValue("b")), 2500_usec_ht,

                    R"#(

{

  "c": {

    "1": "3"

  },

  "d": {

    "1": "5",

    "2": "6"

  }

}

      )#");

  VerifySubDocument(SubDocKey(doc_key, PrimitiveValue("b")), 3500_usec_ht,

                    R"#(

{

  "e": {

    "1": "8",

    "2": "9"

  },

  "y": "10"

}

      )#");

  VerifySubDocument(SubDocKey(doc_key, PrimitiveValue("b")), 5500_usec_ht,

                    R"#(

{

  "e": {

    "1": "8"

  },

  "y": "10"

}

      )#");

  VerifySubDocument(SubDocKey(

      doc_key, PrimitiveValue("b"), PrimitiveValue("d")), 10000_usec_ht, "");

  VerifySubDocument(SubDocKey(doc_key, PrimitiveValue("b"), PrimitiveValue("d")),

                    2500_usec_ht,

                    R"#(

  {

    "1": "5",

    "2": "6"

  }

        )#");

}

TEST_P(DocDBTestWrapper, ListInsertAndGetTest) {

  SubDocument parent;

  SubDocument list({PrimitiveValue(10), PrimitiveValue(2)});

  DocKey doc_key(PrimitiveValues("list_test", 231));

  KeyBytes encoded_doc_key = doc_key.Encode();

  parent.SetChild(PrimitiveValue("other"), SubDocument(PrimitiveValue("other_value")));

  parent.SetChild(PrimitiveValue("list2"), SubDocument(list));

  ASSERT_OK(InsertSubDocument(DocPath(encoded_doc_key), parent, HybridTime(100)));

  VerifySubDocument(SubDocKey(doc_key), HybridTime(250),

      R"#(

  {

    "list2": {

      ArrayIndex(1): 10,

      ArrayIndex(2): 2

    },

    "other": "other_value"

  }

        )#");

  ASSERT_OK(ExtendSubDocument(DocPath(encoded_doc_key, PrimitiveValue("list1")),

      SubDocument({PrimitiveValue(1), PrimitiveValue("3"), PrimitiveValue(2), PrimitiveValue(2)}),

      HybridTime(200)));