text: // 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 "yb/docdb/intent_aware_iterator.h"
  
  #include <future>
  
  #include "yb/common/doc_hybrid_time.h"
  #include "yb/common/hybrid_time.h"
  #include "yb/common/transaction.h"
  
  #include "yb/docdb/docdb_fwd.h"
  #include "yb/docdb/shared_lock_manager_fwd.h"
  #include "yb/docdb/conflict_resolution.h"
  #include "yb/docdb/doc_key.h"
  #include "yb/docdb/doc_kv_util.h"
  #include "yb/docdb/docdb-internal.h"
  #include "yb/docdb/docdb_rocksdb_util.h"
  #include "yb/docdb/intent.h"
  #include "yb/docdb/key_bounds.h"
  #include "yb/docdb/transaction_dump.h"
  #include "yb/docdb/value.h"
  #include "yb/docdb/value_type.h"
  
  #include "yb/util/bytes_formatter.h"
  #include "yb/util/debug-util.h"
  #include "yb/util/result.h"
  #include "yb/util/status_format.h"
  #include "yb/util/trace.h"
  
  using namespace std::literals;
  
  namespace yb {
  namespace docdb {
  
  namespace {
  
  void GetIntentPrefixForKeyWithoutHt(const Slice& key, KeyBytes* out) {
    out->Clear();
    // Since caller guarantees that key_bytes doesn't have hybrid time, we can simply use it
    // to get prefix for all related intents.
    out->AppendRawBytes(key);
  }
  
  KeyBytes GetIntentPrefixForKeyWithoutHt(const Slice& key) {
    KeyBytes result;
    GetIntentPrefixForKeyWithoutHt(key, &result);
    return result;
  }
  
  void AppendEncodedDocHt(const Slice& encoded_doc_ht, KeyBytes* key_bytes) {
    key_bytes->AppendValueType(ValueType::kHybridTime);
    key_bytes->AppendRawBytes(encoded_doc_ht);
  }
  
  const char kStrongWriteTail[] = {
      ValueTypeAsChar::kIntentTypeSet,
      static_cast<char>(IntentTypeSet({IntentType::kStrongWrite}).ToUIntPtr()) };
  
  const Slice kStrongWriteTailSlice = Slice(kStrongWriteTail, sizeof(kStrongWriteTail));
  
  char kEmptyKeyStrongWriteTail[] = {
      ValueTypeAsChar::kGroupEnd,
      ValueTypeAsChar::kIntentTypeSet,
      static_cast<char>(IntentTypeSet({IntentType::kStrongWrite}).ToUIntPtr()) };
  
  const Slice kEmptyKeyStrongWriteTailSlice =
      Slice(kEmptyKeyStrongWriteTail, sizeof(kEmptyKeyStrongWriteTail));
  
  Slice StrongWriteSuffix(const KeyBytes& key) {
    return key.empty() ? kEmptyKeyStrongWriteTailSlice : kStrongWriteTailSlice;
  }
  
  // We are not interested in weak and read intents here.
  // So could just skip them.
  void AppendStrongWrite(KeyBytes* out) {
    out->AppendRawBytes(StrongWriteSuffix(*out));
  }
  
  } // namespace
  
  namespace {
  
  struct DecodeStrongWriteIntentResult {
    Slice intent_prefix;
    Slice intent_value;
    DocHybridTime intent_time;
    DocHybridTime value_time;
    IntentTypeSet intent_types;
  
    // Whether this intent from the same transaction as specified in context.
    bool same_transaction = false;
  
    std::string ToString() const {
      return Format("{ intent_prefix: $0 intent_value: $1 intent_time: $2 value_time: $3 "
                    "same_transaction: $4 intent_types: $5 }",
                    intent_prefix.ToDebugHexString(), intent_value.ToDebugHexString(), intent_time,
                    value_time, same_transaction, intent_types);
    }
  
    // Returns the upper limit for the "value time" of an intent in order for the intent to be visible
    // in the read results. The "value time" is defined as follows:
    //   - For uncommitted transactions, the "value time" is the time when the intent was written.
    //     Note that same_transaction or in_txn_limit could only be set for uncommited transactions.
    //   - For committed transactions, the "value time" is the commit time.
    //
    // The logic here is as follows:
    //   - When a transaction is reading its own intents, the in_txn_limit allows a statement to
    //     avoid seeing its own partial results. This is necessary for statements such as INSERT ...
    //     SELECT to avoid reading rows that the same statement generated and going into an infinite
    //     loop.
    //   - If an intent's hybrid time is greater than the tablet's local limit, then this intent
    //     cannot lead to a read restart and we only need to see it if its commit time is less than or
    //     equal to read_time.
    //   - If an intent's hybrid time is <= than the tablet's local limit, then we cannot claim that
    //     the intent was written after the read transaction began based on the local limit, and we
    //     must compare the intent's commit time with global_limit and potentially perform a read
    //     restart, because the transaction that wrote the intent might have been committed before our
    //     read transaction begin.
    HybridTime MaxAllowedValueTime(const ReadHybridTime& read_time) const {
      if (same_transaction) {
        return read_time.in_txn_limit;
      }
      return intent_time.hybrid_time() > read_time.local_limit
          ? read_time.read : read_time.global_limit;
    }
  };
  
  std::ostream& operator<<(std::ostream& out, const DecodeStrongWriteIntentResult& result) {
    return out << result.ToString();
  }
  
  // Decodes intent based on intent_iterator and its transaction commit time if intent is a strong
  // write intent, intent is not for row locking, and transaction is already committed at specified
  // time or is current transaction.
  // Returns HybridTime::kMin as value_time otherwise.
  // For current transaction returns intent record hybrid time as value_time.
  // Consumes intent from value_slice leaving only value itself.
  Result<DecodeStrongWriteIntentResult> DecodeStrongWriteIntent(
      HybridTime global_limit,
      const TransactionOperationContext& txn_op_context,
      rocksdb::Iterator* intent_iter,
      TransactionStatusCache* transaction_status_cache) {
    DecodeStrongWriteIntentResult result;
    auto decoded_intent_key = VERIFY_RESULT(DecodeIntentKey(intent_iter->key()));
    result.intent_prefix = decoded_intent_key.intent_prefix;
    result.intent_types = decoded_intent_key.intent_types;
    if (result.intent_types.Test(IntentType::kStrongWrite)) {
      auto intent_value = intent_iter->value();
      auto decoded_intent_value = VERIFY_RESULT(DecodeIntentValue(intent_value));
  
      auto decoded_txn_id = decoded_intent_value.transaction_id;
      auto decoded_subtxn_id = decoded_intent_value.subtransaction_id;
  
      result.intent_value = decoded_intent_value.body;
      result.intent_time = decoded_intent_key.doc_ht;
      result.same_transaction = decoded_txn_id == txn_op_context.transaction_id;
  
      // By setting the value time to kMin, we ensure the caller ignores this intent. This is true
      // because the caller is skipping all intents written before or at the same time as
      // intent_dht_from_same_txn_ or resolved_intent_txn_dht_, which of course are greater than or
      // equal to DocHybridTime::kMin.
      if (result.intent_value.starts_with(ValueTypeAsChar::kRowLock)) {
        result.value_time = DocHybridTime::kMin;
      } else if (result.same_transaction) {
        if (txn_op_context.subtransaction.aborted.Test(decoded_subtxn_id)) {
          // If this intent is from the same transaction, we can check the aborted set from this
          // txn_op_context to see whether the intent is still live. If not, mask it from the caller.
          result.value_time = DocHybridTime::kMin;
        } else {
          result.value_time = decoded_intent_key.doc_ht;
        }
      } else if (result.intent_time.hybrid_time() > global_limit) {
        VTRACE(1, "Ignoring intent from a different txn written after read.global_limit");
        result.value_time = DocHybridTime::kMin;
      } else {
        auto commit_data = VERIFY_RESULT(transaction_status_cache->GetCommitData(decoded_txn_id));
        auto commit_ht = commit_data.commit_ht;
        auto aborted_subtxn_set = commit_data.aborted_subtxn_set;
        auto is_aborted_subtxn = aborted_subtxn_set.Test(decoded_subtxn_id);
        result.value_time = commit_ht == HybridTime::kMin || is_aborted_subtxn
            ? DocHybridTime::kMin
            : DocHybridTime(commit_ht, decoded_intent_value.write_id);
        VLOG(4) << "Transaction id: " << decoded_txn_id
                << ", subtransaction id: " << decoded_subtxn_id
                << ", value time: " << result.value_time
                << ", value: " << result.intent_value.ToDebugHexString()
                << ", aborted subtxn set: " << aborted_subtxn_set.ToString();
      }
    } else {
      result.value_time = DocHybridTime::kMin;
    }
    return result;
  }
  
  // Given that key is well-formed DocDB encoded key, checks if it is an intent key for the same key
  // as intent_prefix. If key is not well-formed DocDB encoded key, result could be true or false.
  bool IsIntentForTheSameKey(const Slice& key, const Slice& intent_prefix) {
    return key.starts_with(intent_prefix) &&
           key.size() > intent_prefix.size() &&
           IntentValueType(key[intent_prefix.size()]);
  }
  
  std::string DebugDumpKeyToStr(const Slice &key) {
    return key.ToDebugString() + " (" + SubDocKey::DebugSliceToString(key) + ")";
  }
  
  std::string DebugDumpKeyToStr(const KeyBytes &key) {
    return DebugDumpKeyToStr(key.AsSlice());
  }
  
  bool DebugHasHybridTime(const Slice& subdoc_key_encoded) {
    SubDocKey subdoc_key;
    CHECK(subdoc_key.FullyDecodeFromKeyWithOptionalHybridTime(subdoc_key_encoded).ok());
    return subdoc_key.has_hybrid_time();
  }
  
  std::string EncodeHybridTime(HybridTime value) {
    return DocHybridTime(value, kMaxWriteId).EncodedInDocDbFormat();
  }
  
  } // namespace
  
  IntentAwareIterator::IntentAwareIterator(
      const DocDB& doc_db,
      const rocksdb::ReadOptions& read_opts,
      CoarseTimePoint deadline,
      const ReadHybridTime& read_time,
      const TransactionOperationContext& txn_op_context)
      : read_time_(read_time),
        encoded_read_time_read_(EncodeHybridTime(read_time_.read)),
        encoded_read_time_local_limit_(EncodeHybridTime(read_time_.local_limit)),
        encoded_read_time_global_limit_(EncodeHybridTime(read_time_.global_limit)),
        encoded_read_time_regular_limit_(
            read_time_.local_limit > read_time_.read ? Slice(encoded_read_time_local_limit_)
                                                     : Slice(encoded_read_time_read_)),
        txn_op_context_(txn_op_context),
        transaction_status_cache_(txn_op_context_, read_time, deadline) {
    VTRACE(1, __func__);
    VLOG(4) << "IntentAwareIterator, read_time: " << read_time
            << ", txn_op_context: " << txn_op_context_;
  
    if (txn_op_context) {
      VTRACE(1, "Checking MinRunningTime");
      const auto min_running_ht = txn_op_context.txn_status_manager->MinRunningHybridTime();
      if (min_running_ht != HybridTime::kMax && min_running_ht < read_time.global_limit) {
        intent_iter_ = docdb::CreateRocksDBIterator(doc_db.intents,
                                                    doc_db.key_bounds,
                                                    docdb::BloomFilterMode::DONT_USE_BLOOM_FILTER,
                                                    boost::none,
                                                    rocksdb::kDefaultQueryId,
                                                    nullptr /* file_filter */,
                                                    &intent_upperbound_);
      } else {
        VLOG(4) << "No releavant transactions running";
      }
    }
    VTRACE(2, "Done Checking MinRunningTime");
    // WARNING: Is is important for regular DB iterator to be created after intents DB iterator,
    // otherwise consistency could break, for example in following scenario:
    // 1) Transaction is T1 committed with value v1 for k1, but not yet applied to regular DB.
    // 2) Client reads v1 for k1.
    // 3) Regular DB iterator is created on a regular DB snapshot containing no values for k1.
    // 4) Transaction T1 is applied, k1->v1 is written into regular DB, intent k1->v1 is deleted.
    // 5) Intents DB iterator is created on an intents DB snapshot containing no intents for k1.
    // 6) Client reads no values for k1.
    iter_ = BoundedRocksDbIterator(doc_db.regular, read_opts, doc_db.key_bounds);
    VTRACE(2, "Created iterator");
  }
  
  void IntentAwareIterator::Seek(const DocKey &doc_key) {
    Seek(doc_key.Encode());
  }
  
  void IntentAwareIterator::Seek(const Slice& key) {
    VLOG(4) << "Seek(" << SubDocKey::DebugSliceToString(key) << ")";
    DOCDB_DEBUG_SCOPE_LOG(
        key.ToDebugString(),
        std::bind(&IntentAwareIterator::DebugDump, this));
    if (!status_.ok()) {
      return;
    }
  
    ROCKSDB_SEEK(&iter_, key);
    skip_future_records_needed_ = true;
  
    if (intent_iter_.Initialized()) {
      seek_intent_iter_needed_ = SeekIntentIterNeeded::kSeek;
      GetIntentPrefixForKeyWithoutHt(key, &seek_key_buffer_);
      AppendStrongWrite(&seek_key_buffer_);
    }
  }
  
  void IntentAwareIterator::SeekForward(const Slice& key) {
    KeyBytes key_bytes;
    // Reserve space for key plus kMaxBytesPerEncodedHybridTime + 1 bytes for SeekForward() below to
    // avoid extra realloc while appending the read time.
    key_bytes.Reserve(key.size() + kMaxBytesPerEncodedHybridTime + 1);
    key_bytes.AppendRawBytes(key);
    SeekForward(&key_bytes);
  }
  
  void IntentAwareIterator::SeekForward(KeyBytes* key_bytes) {
    VLOG(4) << "SeekForward(" << SubDocKey::DebugSliceToString(*key_bytes) << ")";
    DOCDB_DEBUG_SCOPE_LOG(
        SubDocKey::DebugSliceToString(*key_bytes),
        std::bind(&IntentAwareIterator::DebugDump, this));
    if (!status_.ok()) {
      return;
    }
  
    const size_t key_size = key_bytes->size();
    AppendEncodedDocHt(encoded_read_time_global_limit_, key_bytes);
    SeekForwardRegular(*key_bytes);
    key_bytes->Truncate(key_size);
    if (intent_iter_.Initialized() && status_.ok()) {
      UpdatePlannedIntentSeekForward(
          *key_bytes, StrongWriteSuffix(*key_bytes), /* use_suffix_for_prefix= */ false);
    }
  }
  
  void IntentAwareIterator::UpdatePlannedIntentSeekForward(const Slice& key,
                                                           const Slice& suffix,
                                                           bool use_suffix_for_prefix) {
    if (seek_intent_iter_needed_ != SeekIntentIterNeeded::kNoNeed &&
        seek_key_buffer_.AsSlice().GreaterOrEqual(key, suffix)) {
      return;
    }
    seek_key_buffer_.Clear();
    seek_key_buffer_.AppendRawBytes(key);
    seek_key_buffer_.AppendRawBytes(suffix);
    if (seek_intent_iter_needed_ == SeekIntentIterNeeded::kNoNeed) {
      seek_intent_iter_needed_ = SeekIntentIterNeeded::kSeekForward;
    }
    seek_key_prefix_ = seek_key_buffer_.AsSlice();
    if (!use_suffix_for_prefix) {
      seek_key_prefix_.remove_suffix(suffix.size());
    }
  }
  
  // TODO: If TTL rows are ever supported on subkeys, this may need to change appropriately.
  // Otherwise, this function might seek past the TTL merge record, but not the original
  // record for the actual subkey.
  void IntentAwareIterator::SeekPastSubKey(const Slice& key) {
    VLOG(4) << "SeekPastSubKey(" << SubDocKey::DebugSliceToString(key) << ")";
    if (!status_.ok()) {
      return;
    }
  
    docdb::SeekPastSubKey(key, &iter_);
    skip_future_records_needed_ = true;
    if (intent_iter_.Initialized() && status_.ok()) {
      // Skip all intents for subdoc_key.
      char kSuffix = ValueTypeAsChar::kGreaterThanIntentType;
      UpdatePlannedIntentSeekForward(key, Slice(&kSuffix, 1));
    }
  }
  
  void IntentAwareIterator::SeekOutOfSubDoc(KeyBytes* key_bytes) {
    VLOG(4) << "SeekOutOfSubDoc(" << SubDocKey::DebugSliceToString(*key_bytes) << ")";
    if (!status_.ok()) {
      return;
    }
  
    docdb::SeekOutOfSubKey(key_bytes, &iter_);
    skip_future_records_needed_ = true;
    if (intent_iter_.Initialized() && status_.ok()) {
      // See comment for SubDocKey::AdvanceOutOfSubDoc.
      const char kSuffix = ValueTypeAsChar::kMaxByte;
      UpdatePlannedIntentSeekForward(*key_bytes, Slice(&kSuffix, 1));
    }
  }
  
  void IntentAwareIterator::SeekOutOfSubDoc(const Slice& key) {
    KeyBytes key_bytes;
    // Reserve space for key + 1 byte for docdb::SeekOutOfSubKey() above to avoid extra realloc while
    // appending kMaxByte.
    key_bytes.Reserve(key.size() + 1);
    key_bytes.AppendRawBytes(key);
    SeekOutOfSubDoc(&key_bytes);
  }
  
  bool IntentAwareIterator::HasCurrentEntry() {
    return iter_valid_ || resolved_intent_state_ == ResolvedIntentState::kValid;
  }
  
  void IntentAwareIterator::SeekToLastDocKey() {
    iter_.SeekToLast();
    SkipFutureRecords(Direction::kBackward);
    if (intent_iter_.Initialized()) {
      ResetIntentUpperbound();
      intent_iter_.SeekToLast();
      SeekToSuitableIntent<Direction::kBackward>();
      seek_intent_iter_needed_ = SeekIntentIterNeeded::kNoNeed;
      skip_future_intents_needed_ = false;
    }
    if (HasCurrentEntry()) {
      SeekToLatestDocKeyInternal();
    }
  }
  
  template <class T>
  void Assign(const T& value, T* out) {
    if (out) {
      *out = value;
    }
  }
  
  // If we reach a different key, stop seeking.
  Status IntentAwareIterator::NextFullValue(
      DocHybridTime* latest_record_ht,
      Slice* result_value,
      Slice* final_key) {
    if (!latest_record_ht || !result_value)
      return STATUS(Corruption, "The arguments latest_record_ht and "
                                "result_value cannot be null pointers.");
    RETURN_NOT_OK(status_);
    Slice v;
    if (!valid() || !IsMergeRecord(v = value())) {
      auto key_data = VERIFY_RESULT(FetchKey());
      Assign(key_data.key, final_key);
      Assign(key_data.write_time, latest_record_ht);
      *result_value = v;
      return status_;
    }
  
    *latest_record_ht = DocHybridTime::kMin;
    const auto key_data = VERIFY_RESULT(FetchKey());
    auto key = key_data.key;
    const size_t key_size = key.size();
    bool found_record = false;
  
    while ((found_record = iter_.Valid()) &&  // as long as we're pointing to a record
           (key = iter_.key()).starts_with(key_data.key) &&  // with the same key we started with
           key[key_size] == ValueTypeAsChar::kHybridTime && // whose key ends with a HT
           IsMergeRecord(v = iter_.value())) { // and whose value is a merge record
      iter_.Next(); // advance the iterator
    }
  
    if (found_record) {
      *result_value = v;
      *latest_record_ht = VERIFY_RESULT(DocHybridTime::DecodeFromEnd(&key));
      Assign(key, final_key);
    }
  
    found_record = false;
    if (intent_iter_.Initialized()) {
      while ((found_record = IsIntentForTheSameKey(intent_iter_.key(), key_data.key)) &&
             IsMergeRecord(v = intent_iter_.value())) {
        intent_iter_.Next();
      }
      DocHybridTime doc_ht;
      if (found_record && !(key = intent_iter_.key()).empty() &&
          (doc_ht = VERIFY_RESULT(DocHybridTime::DecodeFromEnd(&key))) >= *latest_record_ht) {
        *latest_record_ht = doc_ht;
        *result_value = v;
        Assign(key, final_key);
      }
    }
  
    if (*latest_record_ht == DocHybridTime::kMin) {
      iter_valid_ = false;
    }
    return status_;
  }
  
  bool IntentAwareIterator::PreparePrev(const Slice& key) {
    VLOG(4) << __func__ << "(" << SubDocKey::DebugSliceToString(key) << ")";
  
    ROCKSDB_SEEK(&iter_, key);
  
    if (iter_.Valid()) {
      iter_.Prev();
    } else {
      iter_.SeekToLast();
    }
    SkipFutureRecords(Direction::kBackward);
  
    if (intent_iter_.Initialized()) {
      ResetIntentUpperbound();
      ROCKSDB_SEEK(&intent_iter_, GetIntentPrefixForKeyWithoutHt(key));
      if (intent_iter_.Valid()) {
        intent_iter_.Prev();
      } else {
        intent_iter_.SeekToLast();
      }
      SeekToSuitableIntent<Direction::kBackward>();
      seek_intent_iter_needed_ = SeekIntentIterNeeded::kNoNeed;
      skip_future_intents_needed_ = false;
    }
  
    return HasCurrentEntry();
  }
  
  void IntentAwareIterator::PrevSubDocKey(const KeyBytes& key_bytes) {
    if (PreparePrev(key_bytes)) {
      SeekToLatestSubDocKeyInternal();
    }
  }
  
  void IntentAwareIterator::PrevDocKey(const DocKey& doc_key) {
    PrevDocKey(doc_key.Encode().AsSlice());
  }
  
  void IntentAwareIterator::PrevDocKey(const Slice& encoded_doc_key) {
    if (PreparePrev(encoded_doc_key)) {
      SeekToLatestDocKeyInternal();
    }
  }
  
  Slice IntentAwareIterator::LatestSubDocKey() {
    DCHECK(HasCurrentEntry())
        << "Expected iter_valid(" << iter_valid_ << ") || resolved_intent_state_("
        << resolved_intent_state_ << ") == ResolvedIntentState::kValid";
    return IsEntryRegular(/* descending */ true) ? iter_.key()
                                                 : resolved_intent_key_prefix_.AsSlice();
  }
  
  void IntentAwareIterator::SeekToLatestSubDocKeyInternal() {
    auto subdockey_slice = LatestSubDocKey();
  
    // Strip the hybrid time and seek the slice.
    auto doc_ht = DocHybridTime::DecodeFromEnd(&subdockey_slice);
    if (!doc_ht.ok()) {
      status_ = doc_ht.status();
      return;
    }
    subdockey_slice.remove_suffix(1);
    Seek(subdockey_slice);
  }
  
  void IntentAwareIterator::SeekToLatestDocKeyInternal() {
    auto subdockey_slice = LatestSubDocKey();
  
    // Seek to the first key for row containing found subdockey.
    auto dockey_size = DocKey::EncodedSize(subdockey_slice, DocKeyPart::kWholeDocKey);
    if (!dockey_size.ok()) {
      status_ = dockey_size.status();
      return;
    }
    Seek(Slice(subdockey_slice.data(), *dockey_size));
  }
  
  void IntentAwareIterator::SeekIntentIterIfNeeded() {
    if (seek_intent_iter_needed_ == SeekIntentIterNeeded::kNoNeed || !status_.ok()) {
      return;
    }
    status_ = SetIntentUpperbound();
    if (!status_.ok()) {
      return;
    }
    switch (seek_intent_iter_needed_) {
      case SeekIntentIterNeeded::kNoNeed:
        break;
      case SeekIntentIterNeeded::kSeek:
        VLOG(4) << __func__ << ", seek: " << SubDocKey::DebugSliceToString(seek_key_buffer_);
        ROCKSDB_SEEK(&intent_iter_, seek_key_buffer_);
        SeekToSuitableIntent<Direction::kForward>();
        seek_intent_iter_needed_ = SeekIntentIterNeeded::kNoNeed;
        return;
      case SeekIntentIterNeeded::kSeekForward:
        SeekForwardToSuitableIntent();
        seek_intent_iter_needed_ = SeekIntentIterNeeded::kNoNeed;
        return;
    }
    FATAL_INVALID_ENUM_VALUE(SeekIntentIterNeeded, seek_intent_iter_needed_);
  }
  
  bool IntentAwareIterator::valid() {
    if (skip_future_records_needed_) {
      SkipFutureRecords(Direction::kForward);
    }
    SeekIntentIterIfNeeded();
    if (skip_future_intents_needed_) {
      SkipFutureIntents();
    }
    return !status_.ok() || HasCurrentEntry();
  }
  
  bool IntentAwareIterator::IsEntryRegular(bool descending) {
    if (PREDICT_FALSE(!iter_valid_)) {
      return false;
    }
    if (resolved_intent_state_ == ResolvedIntentState::kValid) {
      return (iter_.key().compare(resolved_intent_sub_doc_key_encoded_) < 0) != descending;
    }
    return true;
  }
  
  Result<FetchKeyResult> IntentAwareIterator::FetchKey() {
    RETURN_NOT_OK(status_);
    FetchKeyResult result;
    if (IsEntryRegular()) {
      result.key = iter_.key();
      result.write_time = VERIFY_RESULT(DocHybridTime::DecodeFromEnd(&result.key));
      DCHECK(result.key.ends_with(ValueTypeAsChar::kHybridTime)) << result.key.ToDebugString();
      result.key.remove_suffix(1);
      result.same_transaction = false;
      max_seen_ht_.MakeAtLeast(result.write_time.hybrid_time());
    } else {
      DCHECK_EQ(ResolvedIntentState::kValid, resolved_intent_state_);
      result.key = resolved_intent_key_prefix_.AsSlice();
      result.write_time = GetIntentDocHybridTime();
      result.same_transaction = ResolvedIntentFromSameTransaction();
      max_seen_ht_.MakeAtLeast(resolved_intent_txn_dht_.hybrid_time());
    }
    VLOG(4) << "Fetched key " << SubDocKey::DebugSliceToString(result.key)
            << ", regular: " << IsEntryRegular()
            << ", with time: " << result.write_time
            << ", while read bounds are: " << read_time_;
  
    YB_TRANSACTION_DUMP(
        Read, txn_op_context_ ? txn_op_context_.txn_status_manager->tablet_id() : TabletId(),
        txn_op_context_ ? txn_op_context_.transaction_id : TransactionId::Nil(),
        read_time_, result.write_time, result.same_transaction,
        result.key.size(), result.key, value().size(), value());
  
    return result;
  }
  
  Slice IntentAwareIterator::value() {
    if (IsEntryRegular()) {
      VLOG(4) << "IntentAwareIterator::value() returning iter_.value(): "
              << iter_.value().ToDebugHexString() << " or " << FormatSliceAsStr(iter_.value());
      return iter_.value();
    } else {
      DCHECK_EQ(ResolvedIntentState::kValid, resolved_intent_state_);
      VLOG(4) << "IntentAwareIterator::value() returning resolved_intent_value_: "
              << resolved_intent_value_.AsSlice().ToDebugHexString();
      return resolved_intent_value_;
    }
  }
  
  void IntentAwareIterator::SeekForwardRegular(const Slice& slice) {
    VLOG(4) << "SeekForwardRegular(" << SubDocKey::DebugSliceToString(slice) << ")";
    docdb::SeekForward(slice, &iter_);
    skip_future_records_needed_ = true;
  }
  
  bool IntentAwareIterator::SatisfyBounds(const Slice& slice) {
    return upperbound_.empty() || slice.compare(upperbound_) <= 0;
  }
  
  void IntentAwareIterator::ProcessIntent() {
    auto decode_result = DecodeStrongWriteIntent(
        read_time_.global_limit, txn_op_context_, &intent_iter_, &transaction_status_cache_);
    if (!decode_result.ok()) {
      status_ = decode_result.status();
      return;
    }
    VLOG(4) << "Intent decode: " << DebugIntentKeyToString(intent_iter_.key())
            << " => " << intent_iter_.value().ToDebugHexString() << ", result: " << *decode_result;
    DOCDB_DEBUG_LOG(
        "resolved_intent_txn_dht_: $0 value_time: $1 read_time: $2",
        resolved_intent_txn_dht_.ToString(),
        decode_result->value_time.ToString(),
        read_time_.ToString());
    auto resolved_intent_time = decode_result->same_transaction ? intent_dht_from_same_txn_
                                                                : resolved_intent_txn_dht_;
    // If we already resolved intent that is newer that this one, we should ignore current
    // intent because we are interested in the most recent intent only.
    if (decode_result->value_time <= resolved_intent_time) {
      return;
    }
  
    // Ignore intent past read limit.
    if (decode_result->value_time.hybrid_time() > decode_result->MaxAllowedValueTime(read_time_)) {
      return;
    }
  
    if (resolved_intent_state_ == ResolvedIntentState::kNoIntent) {
      resolved_intent_key_prefix_.Reset(decode_result->intent_prefix);
      auto prefix = CurrentPrefix();
      if (!decode_result->intent_prefix.starts_with(prefix)) {
        resolved_intent_state_ = ResolvedIntentState::kInvalidPrefix;
      } else if (!SatisfyBounds(decode_result->intent_prefix)) {
        resolved_intent_state_ = ResolvedIntentState::kNoIntent;
      } else {
        resolved_intent_state_ = ResolvedIntentState::kValid;
      }
    }
    if (decode_result->same_transaction) {
      intent_dht_from_same_txn_ = decode_result->value_time;
      // We set resolved_intent_txn_dht_ to maximum possible time (time higher than read_time_.read
      // will cause read restart or will be ignored if higher than read_time_.global_limit) in
      // order to ignore intents/values from other transactions. But we save origin intent time into
      // intent_dht_from_same_txn_, so we can compare time of intents for the same key from the same
      // transaction and select the latest one.
      resolved_intent_txn_dht_ = DocHybridTime(read_time_.read, kMaxWriteId);
    } else {
      resolved_intent_txn_dht_ = decode_result->value_time;
    }
    resolved_intent_value_.Reset(decode_result->intent_value);
  }
  
  void IntentAwareIterator::UpdateResolvedIntentSubDocKeyEncoded() {
    resolved_intent_sub_doc_key_encoded_.Reset(resolved_intent_key_prefix_.AsSlice());
    resolved_intent_sub_doc_key_encoded_.AppendValueType(ValueType::kHybridTime);
    resolved_intent_sub_doc_key_encoded_.AppendHybridTime(resolved_intent_txn_dht_);
    VLOG(4) << "Resolved intent SubDocKey: "
            << DebugDumpKeyToStr(resolved_intent_sub_doc_key_encoded_);
  }
  
  void IntentAwareIterator::SeekForwardToSuitableIntent() {
    VLOG(4) << __func__ << "(" << DebugDumpKeyToStr(seek_key_buffer_) << ")";
  
    DOCDB_DEBUG_SCOPE_LOG(seek_key_buffer_.ToString(),
                          std::bind(&IntentAwareIterator::DebugDump, this));
    if (resolved_intent_state_ != ResolvedIntentState::kNoIntent &&
        resolved_intent_key_prefix_.CompareTo(seek_key_prefix_) >= 0) {
      VLOG(4) << __func__ << ", has suitable " << AsString(resolved_intent_state_) << " intent: "
              << DebugDumpKeyToStr(resolved_intent_key_prefix_);
      return;
    }
  
    if (VLOG_IS_ON(4)) {
      if (resolved_intent_state_ != ResolvedIntentState::kNoIntent) {
        VLOG(4) << __func__ << ", has NOT suitable " << AsString(resolved_intent_state_)
                << " intent: " << DebugDumpKeyToStr(resolved_intent_key_prefix_);
      }
  
      if (intent_iter_.Valid()) {
        VLOG(4) << __func__ << ", current position: " << DebugDumpKeyToStr(intent_iter_.key());
      } else {
        VLOG(4) << __func__ << ", iterator invalid";
      }
    }
  
    docdb::SeekForward(seek_key_buffer_.AsSlice(), &intent_iter_);
    SeekToSuitableIntent<Direction::kForward>();
  }
  
  template<Direction direction>
  void IntentAwareIterator::SeekToSuitableIntent() {
    DOCDB_DEBUG_SCOPE_LOG(/* msg */ "", std::bind(&IntentAwareIterator::DebugDump, this));
    resolved_intent_state_ = ResolvedIntentState::kNoIntent;
    resolved_intent_txn_dht_ = DocHybridTime::kMin;
    intent_dht_from_same_txn_ = DocHybridTime::kMin;
    auto prefix = CurrentPrefix();
  
    // Find latest suitable intent for the first SubDocKey having suitable intents.
    while (intent_iter_.Valid()) {
      auto intent_key = intent_iter_.key();
      if (intent_key[0] == ValueTypeAsChar::kTransactionId) {
        // If the intent iterator ever enters the transaction metadata and reverse index region, skip
        // past it.
        switch (direction) {
          case Direction::kForward: {
            static const std::array<char, 1> kAfterTransactionId{ValueTypeAsChar::kTransactionId + 1};
            static const Slice kAfterTxnRegion(kAfterTransactionId);
            intent_iter_.Seek(kAfterTxnRegion);
            break;
          }
          case Direction::kBackward:
            intent_upperbound_keybytes_.Clear();
            intent_upperbound_keybytes_.AppendValueType(ValueType::kTransactionId);
            intent_upperbound_ = intent_upperbound_keybytes_.AsSlice();
            intent_iter_.SeekToLast();
            break;
        }
        continue;
      }
      VLOG(4) << "Intent found: " << DebugIntentKeyToString(intent_key)
              << ", resolved state: " << yb::ToString(resolved_intent_state_);
      if (resolved_intent_state_ != ResolvedIntentState::kNoIntent &&
          // Only scan intents for the first SubDocKey having suitable intents.
          !IsIntentForTheSameKey(intent_key, resolved_intent_key_prefix_)) {
        break;
      }
      if (!intent_key.starts_with(prefix) || !SatisfyBounds(intent_key)) {
        break;
      }
      ProcessIntent();
      if (!status_.ok()) {
        return;
      }
      switch (direction) {
        case Direction::kForward:
          intent_iter_.Next();
          break;
        case Direction::kBackward:
          intent_iter_.Prev();
          break;
      }
    }
    if (resolved_intent_state_ != ResolvedIntentState::kNoIntent) {
      UpdateResolvedIntentSubDocKeyEncoded();
    }
  }
  
  void IntentAwareIterator::DebugDump() {
    bool is_valid = valid();
    LOG(INFO) << ">> IntentAwareIterator dump";
    LOG(INFO) << "iter_.Valid(): " << iter_.Valid();
    if (iter_.Valid()) {
      LOG(INFO) << "iter_.key(): " << DebugDumpKeyToStr(iter_.key());
    }
    if (intent_iter_.Initialized()) {
      LOG(INFO) << "intent_iter_.Valid(): " << intent_iter_.Valid();
      if (intent_iter_.Valid()) {
        LOG(INFO) << "intent_iter_.key(): " << intent_iter_.key().ToDebugHexString();
      }
    }
    LOG(INFO) << "resolved_intent_state_: " << yb::ToString(resolved_intent_state_);
    if (resolved_intent_state_ != ResolvedIntentState::kNoIntent) {
      LOG(INFO) << "resolved_intent_sub_doc_key_encoded_: "
                << DebugDumpKeyToStr(resolved_intent_sub_doc_key_encoded_);
    }
    LOG(INFO) << "valid(): " << is_valid;
    if (valid()) {
      auto key_data = FetchKey();
      if (key_data.ok()) {
        LOG(INFO) << "key(): " << DebugDumpKeyToStr(key_data->key)
                  << ", doc_ht: " << key_data->write_time;
      } else {
        LOG(INFO) << "key(): fetch failed: " << key_data.status();
      }
    }
    LOG(INFO) << "<< IntentAwareIterator dump";
  }
  
  Result<DocHybridTime>
  IntentAwareIterator::FindMatchingIntentRecordDocHybridTime(const Slice& key_without_ht) {
    VLOG(4) << __func__ << "(" << SubDocKey::DebugSliceToString(key_without_ht) << ")";
    GetIntentPrefixForKeyWithoutHt(key_without_ht, &seek_key_buffer_);
    seek_key_prefix_ = seek_key_buffer_.AsSlice();
  
    SeekForwardToSuitableIntent();
    RETURN_NOT_OK(status_);
  
    if (resolved_intent_state_ != ResolvedIntentState::kValid) {
      return DocHybridTime::kInvalid;
    }
  
    if (resolved_intent_key_prefix_.CompareTo(seek_key_buffer_) == 0) {
      max_seen_ht_.MakeAtLeast(resolved_intent_txn_dht_.hybrid_time());
      return GetIntentDocHybridTime();
    }
    return DocHybridTime::kInvalid;
  }
  
  Result<DocHybridTime>
  IntentAwareIterator::GetMatchingRegularRecordDocHybridTime(
      const Slice& key_without_ht) {
    DocHybridTime doc_ht;
    size_t other_encoded_ht_size = 0;
    RETURN_NOT_OK(CheckHybridTimeSizeAndValueType(iter_.key(), &other_encoded_ht_size));
    Slice iter_key_without_ht = iter_.key();
    iter_key_without_ht.remove_suffix(1 + other_encoded_ht_size);
    if (key_without_ht == iter_key_without_ht) {
      RETURN_NOT_OK(DecodeHybridTimeFromEndOfKey(iter_.key(), &doc_ht));
      max_seen_ht_.MakeAtLeast(doc_ht.hybrid_time());
      return doc_ht;
    }
    return DocHybridTime::kInvalid;
  }
  
  Result<HybridTime> IntentAwareIterator::FindOldestRecord(
      const Slice& key_without_ht, HybridTime min_hybrid_time) {
    VLOG(4) << "FindOldestRecord("
            << SubDocKey::DebugSliceToString(key_without_ht) << " = "
            << key_without_ht.ToDebugHexString() << " , " << min_hybrid_time
            << ")";
  #define DOCDB_DEBUG
    DOCDB_DEBUG_SCOPE_LOG(SubDocKey::DebugSliceToString(key_without_ht) + ", " +
                              yb::ToString(min_hybrid_time),
                          std::bind(&IntentAwareIterator::DebugDump, this));
  #undef DOCDB_DEBUG
    DCHECK(!DebugHasHybridTime(key_without_ht));
  
    RETURN_NOT_OK(status_);
    if (!valid()) {
      VLOG(4) << "Returning kInvalid";
      return HybridTime::kInvalid;
    }
  
    HybridTime result;
    if (intent_iter_.Initialized()) {
      auto intent_dht = VERIFY_RESULT(FindMatchingIntentRecordDocHybridTime(key_without_ht));
      VLOG(4) << "Looking for Intent Record found ?  =  "
              << (intent_dht != DocHybridTime::kInvalid);
      if (intent_dht != DocHybridTime::kInvalid &&
          intent_dht.hybrid_time() > min_hybrid_time) {
        result = intent_dht.hybrid_time();
        VLOG(4) << " oldest_record_ht is now " << result;
      }
    } else {
      VLOG(4) << "intent_iter_ not Initialized";
    }
  
    seek_key_buffer_.Reserve(key_without_ht.size() +
                             kMaxBytesPerEncodedHybridTime);
    seek_key_buffer_.Reset(key_without_ht);
    seek_key_buffer_.AppendValueType(ValueType::kHybridTime);
    seek_key_buffer_.AppendHybridTime(
        DocHybridTime(min_hybrid_time, kMaxWriteId));
    SeekForwardRegular(seek_key_buffer_);
    RETURN_NOT_OK(status_);
    if (iter_.Valid()) {
      iter_.Prev();
    } else {
      iter_.SeekToLast();
    }
    SkipFutureRecords(Direction::kForward);
  
    if (iter_valid_) {
      DocHybridTime regular_dht =
          VERIFY_RESULT(GetMatchingRegularRecordDocHybridTime(key_without_ht));
      VLOG(4) << "Looking for Matching Regular Record found   =  " << regular_dht;
      if (regular_dht != DocHybridTime::kInvalid &&
          regular_dht.hybrid_time() > min_hybrid_time) {
        result.MakeAtMost(regular_dht.hybrid_time());
      }
    } else {
      VLOG(4) << "iter_valid_ is false";
    }
    VLOG(4) << "Returning " << result;
    return result;
  }
  
  Status IntentAwareIterator::FindLatestRecord(
      const Slice& key_without_ht,
      DocHybridTime* latest_record_ht,
      Slice* result_value) {
    if (!latest_record_ht) {
      return STATUS(Corruption, "latest_record_ht should not be a null pointer");
    }
    VLOG(4) << __func__ << "(" << SubDocKey::DebugSliceToString(key_without_ht) << ", "
            << *latest_record_ht << ")";
    DOCDB_DEBUG_SCOPE_LOG(
        SubDocKey::DebugSliceToString(key_without_ht) + ", " + yb::ToString(latest_record_ht) + ", "
        + yb::ToString(result_value),
        std::bind(&IntentAwareIterator::DebugDump, this));
    DCHECK(!DebugHasHybridTime(key_without_ht)) << SubDocKey::DebugSliceToString(key_without_ht);
  
    RETURN_NOT_OK(status_);
    if (!valid()) {
      return Status::OK();
    }
  
    bool found_later_intent_result = false;
    if (intent_iter_.Initialized()) {
      DocHybridTime dht = VERIFY_RESULT(FindMatchingIntentRecordDocHybridTime(key_without_ht));
      if (dht != DocHybridTime::kInvalid && dht > *latest_record_ht) {
        *latest_record_ht = dht;
        found_later_intent_result = true;
      }
    }
  
    seek_key_buffer_.Reserve(key_without_ht.size() + encoded_read_time_global_limit_.size() + 1);
    seek_key_buffer_.Reset(key_without_ht);
    AppendEncodedDocHt(encoded_read_time_global_limit_, &seek_key_buffer_);
  
    SeekForwardRegular(seek_key_buffer_);
    RETURN_NOT_OK(status_);
    // After SeekForwardRegular(), we need to call valid() to skip future records and see if the
    // current key still matches the pushed prefix if any. If it does not, we are done.
    if (!valid()) {
      return Status::OK();
    }
  
    bool found_later_regular_result = false;
    if (iter_valid_) {
      DocHybridTime dht = VERIFY_RESULT(GetMatchingRegularRecordDocHybridTime(key_without_ht));
      if (dht != DocHybridTime::kInvalid && dht > *latest_record_ht) {
        *latest_record_ht = dht;
        found_later_regular_result = true;
      }
    }
  
    if (result_value) {
      if (found_later_regular_result) {
        *result_value = iter_.value();
      } else if (found_later_intent_result) {
        *result_value = resolved_intent_value_;
      }
    }
    return Status::OK();
  }
  
  void IntentAwareIterator::PushPrefix(const Slice& prefix) {
    VLOG(4) << "PushPrefix: " << SubDocKey::DebugSliceToString(prefix);
    prefix_stack_.push_back(prefix);
    skip_future_records_needed_ = true;
    skip_future_intents_needed_ = true;
  }
  
  void IntentAwareIterator::PopPrefix() {
    prefix_stack_.pop_back();
    skip_future_records_needed_ = true;
    skip_future_intents_needed_ = true;
    VLOG(4) << "PopPrefix: "
            << (prefix_stack_.empty() ? std::string()
                : SubDocKey::DebugSliceToString(prefix_stack_.back()));
  }
  
  Slice IntentAwareIterator::CurrentPrefix() const {
    return prefix_stack_.empty() ? Slice() : prefix_stack_.back();
  }
  
  void IntentAwareIterator::SkipFutureRecords(const Direction direction) {
    skip_future_records_needed_ = false;
    if (!status_.ok()) {
      return;
    }
    auto prefix = CurrentPrefix();
    while (iter_.Valid()) {
      if (!iter_.key().starts_with(prefix)) {
        VLOG(4) << "Unmatched prefix: " << SubDocKey::DebugSliceToString(iter_.key())
                << ", prefix: " << SubDocKey::DebugSliceToString(prefix);
        iter_valid_ = false;
        return;
      }
      if (!SatisfyBounds(iter_.key())) {
        VLOG(4) << "Out of bounds: " << SubDocKey::DebugSliceToString(iter_.key())
                << ", upperbound: " << SubDocKey::DebugSliceToString(upperbound_);
        iter_valid_ = false;
        return;
      }
      Slice encoded_doc_ht = iter_.key();
      if (encoded_doc_ht.TryConsumeByte(ValueTypeAsChar::kTransactionApplyState)) {
        if (!NextRegular(direction)) {
          return;
        }
        continue;
      }
      size_t doc_ht_size = 0;
      auto decode_status = DocHybridTime::CheckAndGetEncodedSize(encoded_doc_ht, &doc_ht_size);
      if (!decode_status.ok()) {
        LOG(ERROR) << "Decode doc ht from key failed: " << decode_status
                   << ", key: " << iter_.key().ToDebugHexString();
        status_ = std::move(decode_status);
        return;
      }
      encoded_doc_ht.remove_prefix(encoded_doc_ht.size() - doc_ht_size);
      auto value = iter_.value();
      auto value_type = DecodeValueType(value);
      VLOG(4) << "Checking for skip, type " << value_type << ", encoded_doc_ht: "
              << DocHybridTime::DebugSliceToString(encoded_doc_ht)
              << " value: " << value.ToDebugHexString();
      if (value_type == ValueType::kHybridTime) {
        // Value came from a transaction, we could try to filter it by original intent time.
        Slice encoded_intent_doc_ht = value;
        encoded_intent_doc_ht.consume_byte();
        // The logic here replicates part of the logic in
        // DecodeStrongWriteIntentResult:: MaxAllowedValueTime for intents that have been committed
        // and applied to regular RocksDB only. Note that here we are comparing encoded hybrid times,
        // so comparisons are reversed vs. the un-encoded case. If a value is found "invalid", it
        // can't cause a read restart. If it is found "valid", it will cause a read restart if it is
        // greater than read_time.read. That last comparison is done outside this function.
        Slice max_allowed = encoded_intent_doc_ht.compare(encoded_read_time_local_limit_) > 0
            ? Slice(encoded_read_time_global_limit_)
            : Slice(encoded_read_time_read_);
        if (encoded_doc_ht.compare(max_allowed) > 0) {
          iter_valid_ = true;
          return;
        }
      } else if (encoded_doc_ht.compare(encoded_read_time_regular_limit_) > 0) {
        // If a value does not contain the hybrid time of the intent that wrote the original
        // transaction, then it either (a) originated from a single-shard transaction or (b) the
        // intent hybrid time has already been garbage-collected during a compaction because the
        // corresponding transaction's commit time (stored in the key) became lower than the history
        // cutoff. See the following commit for the details of this intent hybrid time GC.
        //
        // https://github.com/yugabyte/yugabyte-db/commit/26260e0143e521e219d93f4aba6310fcc030a628
        //
        // encoded_read_time_regular_limit_ is simply the encoded value of max(read_ht, local_limit).
        // The above condition
        //
        //   encoded_doc_ht.compare(encoded_read_time_regular_limit_) >= 0
        //
        // corresponds to the following in terms of decoded hybrid times (order is reversed):
        //
        //   commit_ht <= max(read_ht, local_limit)
        //
        // and the inverse of that can be written as
        //
        //   commit_ht > read_ht && commit_ht > local_limit
        //
        // The reason this is correct here is that in case (a) the event of writing a single-shard
        // record to the tablet would certainly be after our read transaction's start time in case
        // commit_ht > local_limit, so it can never cause a read restart. In case (b) we know that
        // commit_ht < history_cutoff and read_ht >= history_cutoff (by definition of history cutoff)
        // so commit_ht < read_ht, and in this case read restart is impossible regardless of the
        // value of local_limit.
        iter_valid_ = true;
        return;
      }
      VLOG(4) << "Skipping because of time: " << SubDocKey::DebugSliceToString(iter_.key())
              << ", read time: " << read_time_;
      if (!NextRegular(direction)) {
        return;
      }
    }
    iter_valid_ = false;
  }
  
  bool IntentAwareIterator::NextRegular(Direction direction) {
    switch (direction) {
      case Direction::kForward:
        iter_.Next(); // TODO(dtxn) use seek with the same key, but read limit as doc hybrid time.
        return true;
      case Direction::kBackward:
        iter_.Prev();
        return true;
    }
  
    status_ = STATUS_FORMAT(Corruption, "Unexpected direction: $0", direction);
    LOG(ERROR) << status_;
    iter_valid_ = false;
    return false;
  }
  
  void IntentAwareIterator::SkipFutureIntents() {
    skip_future_intents_needed_ = false;
    if (!intent_iter_.Initialized() || !status_.ok()) {
      return;
    }
    auto prefix = CurrentPrefix();
    if (resolved_intent_state_ != ResolvedIntentState::kNoIntent) {
      auto compare_result = resolved_intent_key_prefix_.AsSlice().compare_prefix(prefix);
      VLOG(4) << "Checking resolved intent subdockey: "
              << DebugDumpKeyToStr(resolved_intent_key_prefix_)
              << ", against new prefix: " << DebugDumpKeyToStr(prefix) << ": "
              << compare_result;
      if (compare_result == 0) {
        if (!SatisfyBounds(resolved_intent_key_prefix_.AsSlice())) {
          resolved_intent_state_ = ResolvedIntentState::kNoIntent;
        } else {
          resolved_intent_state_ = ResolvedIntentState::kValid;
        }
        return;
      } else if (compare_result > 0) {
        resolved_intent_state_ = ResolvedIntentState::kInvalidPrefix;
        return;
      }
    }
    SeekToSuitableIntent<Direction::kForward>();
  }
  
  Status IntentAwareIterator::SetIntentUpperbound() {
    if (iter_.Valid()) {
      intent_upperbound_keybytes_.Clear();
      // Strip ValueType::kHybridTime + DocHybridTime at the end of SubDocKey in iter_ and append
      // to upperbound with 0xff.
      Slice subdoc_key = iter_.key();
      size_t doc_ht_size = 0;
      RETURN_NOT_OK(DocHybridTime::CheckAndGetEncodedSize(subdoc_key, &doc_ht_size));
      subdoc_key.remove_suffix(1 + doc_ht_size);
      intent_upperbound_keybytes_.AppendRawBytes(subdoc_key);
      VLOG(4) << "SetIntentUpperbound = "
              << SubDocKey::DebugSliceToString(intent_upperbound_keybytes_.AsSlice());
      intent_upperbound_keybytes_.AppendValueType(ValueType::kMaxByte);
      intent_upperbound_ = intent_upperbound_keybytes_.AsSlice();
      intent_iter_.RevalidateAfterUpperBoundChange();
    } else {
      // In case the current position of the regular iterator is invalid, set the exclusive intent
      // upperbound high to be able to find all intents higher than the last regular record.
      ResetIntentUpperbound();
    }
    return Status::OK();
  }
  
  void IntentAwareIterator::ResetIntentUpperbound() {
    intent_upperbound_keybytes_.Clear();
    intent_upperbound_keybytes_.AppendValueType(ValueType::kHighest);
    intent_upperbound_ = intent_upperbound_keybytes_.AsSlice();
    intent_iter_.RevalidateAfterUpperBoundChange();
    VLOG(4) << "ResetIntentUpperbound = " << intent_upperbound_.ToDebugString();
  }
  
  }  // namespace docdb
  }  // namespace yb
  