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 "yb/docdb/doc_rowwise_iterator.h"

#include <iterator>

#include <cstdint>

#include <ostream>

#include <string>

#include <vector>

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

#include "yb/common/doc_hybrid_time.h"

#include "yb/common/hybrid_time.h"

#include "yb/common/ql_expr.h"

#include "yb/common/ql_scanspec.h"

#include "yb/common/ql_value.h"

#include "yb/common/read_hybrid_time.h"

#include "yb/common/transaction.h"

#include "yb/docdb/docdb_fwd.h"

#include "yb/docdb/doc_key.h"

#include "yb/docdb/doc_path.h"

#include "yb/docdb/doc_ql_scanspec.h"

#include "yb/docdb/doc_reader.h"

#include "yb/docdb/doc_scanspec_util.h"

#include "yb/docdb/docdb_rocksdb_util.h"

#include "yb/docdb/docdb_types.h"

#include "yb/docdb/expiration.h"

#include "yb/docdb/intent_aware_iterator.h"

#include "yb/docdb/primitive_value.h"

#include "yb/docdb/subdocument.h"

#include "yb/docdb/value.h"

#include "yb/docdb/value_type.h"

#include "yb/gutil/strings/substitute.h"

#include "yb/util/flags.h"

#include "yb/rocksdb/db/compaction.h"

#include "yb/rocksutil/yb_rocksdb.h"

#include "yb/rocksdb/db.h"

#include "yb/util/flag_tags.h"

#include "yb/util/result.h"

#include "yb/util/status.h"

#include "yb/util/status_format.h"

#include "yb/util/status_log.h"

#include "yb/util/strongly_typed_bool.h"

DEFINE_bool(disable_hybrid_scan, false,

            "If true, hybrid scan will be disabled");

TAG_FLAG(disable_hybrid_scan, runtime);

using std::string;

namespace yb {

namespace docdb {

class ScanChoices {

 public:

  explicit ScanChoices(bool is_forward_scan) : is_forward_scan_(is_forward_scan) {}

  virtual ~ScanChoices() {}

  bool CurrentTargetMatchesKey(const Slice& curr) {

    VLOG(3) << __PRETTY_FUNCTION__ << " checking if acceptable ? "

            << (curr == current_scan_target_ ? "YEP" : "NOPE")

            << ": " << DocKey::DebugSliceToString(curr)

            << " vs " << DocKey::DebugSliceToString(current_scan_target_.AsSlice());

    return curr == current_scan_target_;

  }

  // Returns false if there are still target keys we need to scan, and true if we are done.

  virtual bool FinishedWithScanChoices() const { return finished_; }

  // Go to the next scan target if any.

  virtual CHECKED_STATUS DoneWithCurrentTarget() = 0;

  // Go (directly) to the new target (or the one after if new_target does not

  // exist in the desired list/range). If the new_target is larger than all scan target options it

  // means we are done.

  virtual CHECKED_STATUS SkipTargetsUpTo(const Slice& new_target) = 0;

  // If the given doc_key isn't already at the desired target, seek appropriately to go to the

  // current target.

  virtual CHECKED_STATUS SeekToCurrentTarget(IntentAwareIterator* db_iter) = 0;

 protected:

  const bool is_forward_scan_;

  KeyBytes current_scan_target_;

  bool finished_ = false;

};

class DiscreteScanChoices : public ScanChoices {

 public:

  DiscreteScanChoices(const DocQLScanSpec& doc_spec, const KeyBytes& lower_doc_key,

                      const KeyBytes& upper_doc_key)

      : ScanChoices(doc_spec.is_forward_scan()) {

    range_cols_scan_options_ = doc_spec.range_options();

    current_scan_target_idxs_.resize(range_cols_scan_options_->size());

    for (size_t i = 0; i < range_cols_scan_options_->size(); i++) {

      current_scan_target_idxs_[i] = range_cols_scan_options_->at(i).begin();

    }

    // Initialize target doc key.

    if (is_forward_scan_) {

      current_scan_target_ = lower_doc_key;

      if (CHECK_RESULT(ClearRangeComponents(&current_scan_target_))) {

        CHECK_OK(SkipTargetsUpTo(lower_doc_key));

      }

    } else {

      current_scan_target_ = upper_doc_key;

      if (CHECK_RESULT(ClearRangeComponents(&current_scan_target_))) {

        CHECK_OK(SkipTargetsUpTo(upper_doc_key));

      }

    }

  }

  DiscreteScanChoices(const DocPgsqlScanSpec& doc_spec, const KeyBytes& lower_doc_key,

                      const KeyBytes& upper_doc_key)

      : ScanChoices(doc_spec.is_forward_scan()) {

    range_cols_scan_options_ = doc_spec.range_options();

    current_scan_target_idxs_.resize(range_cols_scan_options_->size());

    for (size_t i = 0; i < range_cols_scan_options_->size(); i++) {

      current_scan_target_idxs_[i] = range_cols_scan_options_->at(i).begin();

    }

    // Initialize target doc key.

    if (is_forward_scan_) {

      current_scan_target_ = lower_doc_key;

      if (CHECK_RESULT(ClearRangeComponents(&current_scan_target_))) {

        CHECK_OK(SkipTargetsUpTo(lower_doc_key));

      }

    } else {

      current_scan_target_ = upper_doc_key;

      if (CHECK_RESULT(ClearRangeComponents(&current_scan_target_))) {

        CHECK_OK(SkipTargetsUpTo(upper_doc_key));

      }

    }

  }

  CHECKED_STATUS DoneWithCurrentTarget() override;

  CHECKED_STATUS SkipTargetsUpTo(const Slice& new_target) override;

  CHECKED_STATUS SeekToCurrentTarget(IntentAwareIterator* db_iter) override;

 protected:

  // Utility function for (multi)key scans. Updates the target scan key by incrementing the option

  // index for one column. Will handle overflow by setting current column index to 0 and

  // incrementing the previous column instead. If it overflows at first column it means we are done,

  // so it clears the scan target idxs array.

  CHECKED_STATUS IncrementScanTargetAtColumn(size_t start_col);

  // Utility function for (multi)key scans to initialize the range portion of the current scan

  // target, scan target with the first option.

  // Only needed for scans that include the static row, otherwise Init will take care of this.

  Result<bool> InitScanTargetRangeGroupIfNeeded();

 private:

  // For (multi)key scans (e.g. selects with 'IN' condition on the range columns) we hold the

  // options for each range column as we iteratively seek to each target key.

  // e.g. for a query "h = 1 and r1 in (2,3) and r2 in (4,5) and r3 = 6":

  //  range_cols_scan_options_   [[2, 3], [4, 5], [6]] -- value options for each column.

  //  current_scan_target_idxs_  goes from [0, 0, 0] up to [1, 1, 0] -- except when including the

  //                             static row when it starts from [0, 0, -1] instead.

  //  current_scan_target_       goes from [1][2,4,6] up to [1][3,5,6] -- is the doc key containing,

  //                             for each range column, the value (option) referenced by the

  //                             corresponding index (updated along with current_scan_target_idxs_).

  std::shared_ptr<std::vector<std::vector<PrimitiveValue>>> range_cols_scan_options_;

  mutable std::vector<std::vector<PrimitiveValue>::const_iterator> current_scan_target_idxs_;

};

Status DiscreteScanChoices::IncrementScanTargetAtColumn(size_t start_col) {

  DCHECK_LE(start_col, current_scan_target_idxs_.size());

  // Increment start col, move backwards in case of overflow.

  ssize_t col_idx = start_col;

  for (; col_idx >= 0; col_idx--) {

    const auto& choices = (*range_cols_scan_options_)[col_idx];

    auto& it = current_scan_target_idxs_[col_idx];

    if (++it != choices.end()) {

      break;

    }

    it = choices.begin();

  }

  if (col_idx < 0) {

    // If we got here we finished all the options and are done.

    finished_ = true;

    return Status::OK();

  }

  DocKeyDecoder decoder(current_scan_target_);

  RETURN_NOT_OK(decoder.DecodeToRangeGroup());

  for (int i = 0; i != col_idx; ++i) {

    RETURN_NOT_OK(decoder.DecodePrimitiveValue());

  }

  current_scan_target_.Truncate(

      decoder.left_input().cdata() - current_scan_target_.AsSlice().cdata());

  for (size_t i = col_idx; i <= start_col; ++i) {

    current_scan_target_idxs_[i]->AppendToKey(&current_scan_target_);

  }

  return Status::OK();

}

Result<bool> DiscreteScanChoices::InitScanTargetRangeGroupIfNeeded() {

  DocKeyDecoder decoder(current_scan_target_.AsSlice());

  RETURN_NOT_OK(decoder.DecodeToRangeGroup());

  // Initialize the range key values if needed (i.e. we scanned the static row until now).

  if (!VERIFY_RESULT(decoder.HasPrimitiveValue())) {

    current_scan_target_.mutable_data()->pop_back();

    for (size_t col_idx = 0; col_idx < range_cols_scan_options_->size(); col_idx++) {

      current_scan_target_idxs_[col_idx]->AppendToKey(&current_scan_target_);

    }

    current_scan_target_.AppendValueType(ValueType::kGroupEnd);

    return true;

  }

  return false;

}

Status DiscreteScanChoices::DoneWithCurrentTarget() {

  VLOG(2) << __PRETTY_FUNCTION__ << " moving on to next target";

  DCHECK(!FinishedWithScanChoices());

  // Initialize the first target/option if not done already, otherwise go to the next one.

  if (!VERIFY_RESULT(InitScanTargetRangeGroupIfNeeded())) {

    RETURN_NOT_OK(IncrementScanTargetAtColumn(range_cols_scan_options_->size() - 1));

    current_scan_target_.AppendValueType(ValueType::kGroupEnd);

  }

  return Status::OK();

}

Status DiscreteScanChoices::SkipTargetsUpTo(const Slice& new_target) {

  VLOG(2) << __PRETTY_FUNCTION__

            << " Updating current target to be >= "

            << DocKey::DebugSliceToString(new_target);

  DCHECK(!FinishedWithScanChoices());

  RETURN_NOT_OK(InitScanTargetRangeGroupIfNeeded());

  DocKeyDecoder decoder(new_target);

  RETURN_NOT_OK(decoder.DecodeToRangeGroup());

  current_scan_target_.Reset(Slice(new_target.data(), decoder.left_input().data()));

  size_t col_idx = 0;

  PrimitiveValue target_value;

  while (col_idx < range_cols_scan_options_->size()) {

    RETURN_NOT_OK(decoder.DecodePrimitiveValue(&target_value));

    const auto& choices = (*range_cols_scan_options_)[col_idx];

    auto& it = current_scan_target_idxs_[col_idx];

    // Fast-path in case the existing value for this column already matches the new target.

    if (target_value == *it) {

      col_idx++;

      target_value.AppendToKey(&current_scan_target_);

      continue;

    }

    // Search for the option that matches new target value (for the current column).

    if (is_forward_scan_) {

      it = std::lower_bound(choices.begin(), choices.end(), target_value);

    } else {

      it = std::lower_bound(choices.begin(), choices.end(), target_value, std::greater<>());

    }

    // If we overflowed, the new target value for this column is larger than all our options, so

    // we go back and increment the previous column instead.

    if (it == choices.end()) {

      RETURN_NOT_OK(IncrementScanTargetAtColumn(col_idx - 1));

      break;

    }

    // Else, update the current target value for this column.

    it->AppendToKey(&current_scan_target_);

    // If we did not find an exact match we are already beyond the new target so we can stop.

    if (target_value != *it) {

      col_idx++;

      break;

    }

    col_idx++;

  }

  // If there are any columns left (i.e. we stopped early), it means we did not find an exact

  // match and we reached beyond the new target key. So we need to include all options for the

  // leftover columns (i.e. set all following indexes to 0).

  for (size_t i = col_idx; i < current_scan_target_idxs_.size(); i++) {

    current_scan_target_idxs_[i] = (*range_cols_scan_options_)[i].begin();

    current_scan_target_idxs_[i]->AppendToKey(&current_scan_target_);

  }

  current_scan_target_.AppendValueType(ValueType::kGroupEnd);

  VLOG(2) << "After " << __PRETTY_FUNCTION__ << " current_scan_target_ is "

          << DocKey::DebugSliceToString(current_scan_target_);

  return Status::OK();

}

Status DiscreteScanChoices::SeekToCurrentTarget(IntentAwareIterator* db_iter) {

  VLOG(2) << __PRETTY_FUNCTION__ << " Advancing iterator towards target";

  // Seek to the current target doc key if needed.

  if (!FinishedWithScanChoices()) {

    if (is_forward_scan_) {

      VLOG(2) << __PRETTY_FUNCTION__ << " Seeking to " << current_scan_target_;

      db_iter->Seek(current_scan_target_);

    } else {

      auto tmp = current_scan_target_;

      tmp.AppendValueType(ValueType::kHighest);

      VLOG(2) << __PRETTY_FUNCTION__ << " Going to PrevDocKey " << tmp;

      db_iter->PrevDocKey(tmp);

    }

  }

  return Status::OK();

}

// This class combines the notions of option filters (col1 IN (1,2,3)) and

// singular range bound filters (col1 < 4 AND col1 >= 1) into a single notion of

// lists of ranges. So a filter for a column given in the

// Doc(QL/PGSQL)ScanSpec is converted into a range bound filter.

// In the end, each HybridScanChoices

// instance should have a sorted list of disjoint ranges to filter each column.

// Right now this supports a conjunction of range bound and discrete filters.

// Disjunctions are also supported but are UNTESTED.

// TODO: Test disjunctions when YSQL and YQL support pushing those down

class HybridScanChoices : public ScanChoices {

 public:

  // Constructs a list of ranges for each column from the given scanspec.

  // A filter of the form col1 IN (1,4,5) is converted to a filter

  // in the form col1 IN ([1, 1], [4, 4], [5, 5]).

  HybridScanChoices(const Schema& schema,

                    const KeyBytes &lower_doc_key,

                    const KeyBytes &upper_doc_key,

                    bool is_forward_scan,

                    const std::vector<ColumnId> &range_options_indexes,

                    const

                    std::shared_ptr<std::vector<std::vector<PrimitiveValue>>>&

                        range_options,

                    const std::vector<ColumnId> range_bounds_indexes,

                    const QLScanRange *range_bounds)

                    : ScanChoices(is_forward_scan),

                        lower_doc_key_(lower_doc_key),

                        upper_doc_key_(upper_doc_key) {

    auto range_cols_scan_options = range_options;

    size_t idx = 0;

    range_cols_scan_options_lower_.reserve(schema.num_range_key_columns());

    range_cols_scan_options_upper_.reserve(schema.num_range_key_columns());

    size_t num_hash_cols = schema.num_hash_key_columns();

    for (idx = schema.num_hash_key_columns();

            idx < schema.num_key_columns(); idx++) {

      const ColumnId col_idx = schema.column_id(idx);

      range_cols_scan_options_lower_.push_back({});

      range_cols_scan_options_upper_.push_back({});

      // If this is a range bound filter, we create a singular

      // list of the given range bound

      if ((std::find(range_bounds_indexes.begin(),

                        range_bounds_indexes.end(), col_idx)

                    != range_bounds_indexes.end())

            && (std::find(range_options_indexes.begin(),

                            range_options_indexes.end(), col_idx)

                        == range_options_indexes.end())) {

        const auto col_sort_type = schema.column(idx).sorting_type();

        const QLScanRange::QLRange range = range_bounds->RangeFor(col_idx);

        const auto lower = GetQLRangeBoundAsPVal(range, col_sort_type,

                                                    true /* lower_bound */);

        const auto upper = GetQLRangeBoundAsPVal(range, col_sort_type,

                                                    false /* upper_bound */);

        range_cols_scan_options_lower_[idx - num_hash_cols].push_back(lower);

        range_cols_scan_options_upper_[idx - num_hash_cols].push_back(upper);

      } else {

        // If this is an option filter, we turn each option into a

        // range bound to produce a list of singular range bounds

        if(std::find(range_options_indexes.begin(),

                        range_options_indexes.end(), col_idx)

                    != range_options_indexes.end()) {

          auto &options = (*range_cols_scan_options)[idx - num_hash_cols];

          if (options.empty()) {

            // If there is nothing specified in the IN list like in

            // SELECT * FROM ... WHERE c1 IN ();

            // then nothing should pass the filter.

            // To enforce this, we create a range bound (kHighest, kLowest)

            range_cols_scan_options_lower_[idx

              - num_hash_cols].push_back(PrimitiveValue(ValueType::kHighest));

            range_cols_scan_options_upper_[idx

              - num_hash_cols].push_back(PrimitiveValue(ValueType::kLowest));

          }

          for (auto val : options) {

            const auto lower = val;

            const auto upper = val;

            range_cols_scan_options_lower_[idx

              - num_hash_cols].push_back(lower);

            range_cols_scan_options_upper_[idx

              - num_hash_cols].push_back(upper);

          }

        } else {

            // If no filter is specified, we just impose an artificial range

            // filter [kLowest, kHighest]

            range_cols_scan_options_lower_[idx - num_hash_cols]

                                .push_back(PrimitiveValue(ValueType::kLowest));

            range_cols_scan_options_upper_[idx - num_hash_cols]

                                .push_back(PrimitiveValue(ValueType::kHighest));

        }

      }

    }

    current_scan_target_idxs_.resize(range_cols_scan_options_lower_.size());

    if (is_forward_scan_) {

      current_scan_target_ = lower_doc_key;

    } else {

      current_scan_target_ = upper_doc_key;

    }

  }

  HybridScanChoices(const Schema& schema,

                    const DocPgsqlScanSpec& doc_spec,

                    const KeyBytes &lower_doc_key,

                    const KeyBytes &upper_doc_key)

      : HybridScanChoices(schema, lower_doc_key, upper_doc_key,

      doc_spec.is_forward_scan(), doc_spec.range_options_indexes(),

      doc_spec.range_options(), doc_spec.range_bounds_indexes(),

      doc_spec.range_bounds()) {

  }

  HybridScanChoices(const Schema& schema,

                    const DocQLScanSpec& doc_spec,

                    const KeyBytes &lower_doc_key,

                    const KeyBytes &upper_doc_key)

      : HybridScanChoices(schema, lower_doc_key, upper_doc_key,

      doc_spec.is_forward_scan(), doc_spec.range_options_indexes(),

      doc_spec.range_options(), doc_spec.range_bounds_indexes(),

      doc_spec.range_bounds()) {

  }

  CHECKED_STATUS SkipTargetsUpTo(const Slice& new_target) override;

  CHECKED_STATUS DoneWithCurrentTarget() override;

  CHECKED_STATUS SeekToCurrentTarget(IntentAwareIterator* db_iter) override;

 protected:

  // Utility function for (multi)key scans. Updates the target scan key by

  // incrementing the option

  // index for one column. Will handle overflow by setting current column

  // index to 0 and incrementing the previous column instead. If it overflows

  // at first column it means we are done, so it clears the scan target idxs

  // array.

  CHECKED_STATUS IncrementScanTargetAtColumn(int start_col);

 private:

  KeyBytes prev_scan_target_;

  // The following encodes the list of ranges we are iterating over

  std::vector<std::vector<PrimitiveValue>> range_cols_scan_options_lower_;

  std::vector<std::vector<PrimitiveValue>> range_cols_scan_options_upper_;

  std::vector<ColumnId> range_options_indexes_;

  mutable std::vector<size_t> current_scan_target_idxs_;

  bool is_options_done_ = false;

  const KeyBytes lower_doc_key_;

  const KeyBytes upper_doc_key_;

};

// Sets current_scan_target_ to the first tuple in the filter space

// that is >= new_target.

Status HybridScanChoices::SkipTargetsUpTo(const Slice& new_target) {

  VLOG(2) << __PRETTY_FUNCTION__ << " Updating current target to be >= "

          << DocKey::DebugSliceToString(new_target);

  DCHECK(!FinishedWithScanChoices());

  is_options_done_ = false;

  /*

   Let's say we have a row key with (A B) as the hash part and C, D as the range part:

   ((A B) C D) E F

   Let's say our current constraints :

    l_c_k <= C <= u_c_k

     4            6

    l_d_j <= D <= u_d_j

      3           5

    a b  0 d  -> a  b l_c  d

    a b  5 d  -> a  b  5   d

                  [ Will subsequently seek out of document on reading the subdoc]

    a b  7 d  -> a b l_c_(k+1) 0

                [ If there is another range bound filter that's higher than the

                  current one, effectively, moving this column to the next

                  range in the filter list.]

              -> a b Inf

                [ This will seek to <b_next> and on the next invocation update:

                   a <b_next> ? ? -> a <b_next> l_c_0 0 ]

    a b  c 6  -> a b c l_d_(j+1)

                [ If there is another range bound filter that's higher than the

                  d, effectively, moving column D to the next

                  range in the filter list.]

              -> a b c Inf

                [ If c_next is between l_c_k and u_c_k. This will seek to <a b

                   <c_next>> and on the next invocation update:

                   a b <c_next> ? -> a b <c_next> l_d_0 ]

              -> a b l_c_(k+1) l_d_0

                 [ If c_next is above u_c_k. We do this because we know

                   exactly what the next tuple in our filter space should be.]

  */

  DocKeyDecoder decoder(new_target);

  RETURN_NOT_OK(decoder.DecodeToRangeGroup());

  current_scan_target_.Reset(Slice(new_target.data(),

                                decoder.left_input().data()));

  size_t col_idx = 0;

  PrimitiveValue target_value;

  for (col_idx = 0; col_idx < current_scan_target_idxs_.size(); col_idx++) {

    RETURN_NOT_OK(decoder.DecodePrimitiveValue(&target_value));

    const auto& lower_choices = (range_cols_scan_options_lower_)[col_idx];

    const auto& upper_choices = (range_cols_scan_options_upper_)[col_idx];

    auto current_ind = current_scan_target_idxs_[col_idx];

    DCHECK(current_ind < lower_choices.size());

    const auto& lower = lower_choices[current_ind];

    const auto& upper = upper_choices[current_ind];

    // If it's in range then good, continue after appending the target value

    // column.

    if (target_value >= lower && target_value <= upper) {

      target_value.AppendToKey(&current_scan_target_);

      continue;

    }

    // If target_value is not in the current range then we must find a range

    // that works for it.

    // If we are above all ranges then increment the index of the previous

    // column.

    // Else, target_value is below at least one range: find the lowest lower

    // bound above target_value and use that, this relies on the assumption

    // that all our filter ranges are disjoint.

    auto it = lower_choices.begin();

    size_t ind = 0;

    // Find an upper (lower) bound closest to target_value

    if (is_forward_scan_) {

      it = std::lower_bound(upper_choices.begin(),

                                upper_choices.end(), target_value);

      ind = it - upper_choices.begin();

    } else {

      it = std::lower_bound(lower_choices.begin(), lower_choices.end(),

              target_value, std::greater<>());

      ind = it - lower_choices.begin();

    }

    if (ind == lower_choices.size()) {

      // target value is higher than all range options and

      // we need to increment.

      RETURN_NOT_OK(IncrementScanTargetAtColumn(static_cast<int>(col_idx) - 1));

      col_idx = current_scan_target_idxs_.size();

      break;

    }

    current_scan_target_idxs_[col_idx] = ind;

    // If we are within a range then target_value itself should work.

    if (lower_choices[ind] <= target_value

        && upper_choices[ind] >= target_value) {

      target_value.AppendToKey(&current_scan_target_);

      continue;

    }

    // Otherwise we must set it to the next lower bound.

    // This only works as we are assuming all given ranges are

    // disjoint.

    DCHECK((is_forward_scan_ && lower_choices[ind] > target_value)

              || (!is_forward_scan_ && upper_choices[ind]

              < target_value));

    if (is_forward_scan_) {

      lower_choices[ind].AppendToKey(&current_scan_target_);

    } else {

      upper_choices[ind].AppendToKey(&current_scan_target_);

    }

    col_idx++;

    break;

  }

  // Reset the remaining range columns to lower bounds for forward scans

  // or upper bounds for backward scans.

  for (size_t i = col_idx; i < range_cols_scan_options_lower_.size(); i++) {

    current_scan_target_idxs_[i] = 0;

    if (is_forward_scan_) {

      range_cols_scan_options_lower_[i][0]

                    .AppendToKey(&current_scan_target_);

    } else {

      range_cols_scan_options_upper_[i][0]

                    .AppendToKey(&current_scan_target_);

    }

  }

  current_scan_target_.AppendValueType(ValueType::kGroupEnd);

  VLOG(2) << "After " << __PRETTY_FUNCTION__ << " current_scan_target_ is "

          << DocKey::DebugSliceToString(current_scan_target_);

  return Status::OK();

}

// Update the value at start column by setting it up for incrementing to the

// next allowed value in the filter space

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

// There are two important cases to consider here.

// Let's say the value of current_scan_target_ at start_col, c,

// is currently V and the current bounds for that column

// is l_c_k <= V <= u_c_k. In the usual case where V != u_c_k

// (or V != l_c_k for backwards scans) such that V_next is still in the given

// restriction, we set column c + 1 to kHighest (kLowest), such that the next

// invocation of GetNext() produces V_next at column similar to what is done

// in SkipTargetsUpTo. In this case, doing a SkipTargetsUpTo on the resulting

// current_scan_target_ should yield the next allowed value in the filter space

// In the case where V = u_c_k (V = l_c_k), or in other words V is at the

// EXTREMAL boundary of the current range, we know exactly what the next value

// of column C will be. So we move column c to the next

// range k+1 and set that column to the new value l_c_(k+1) (u_c_(k+1))

// while setting all columns, b > c to l_b_0 (u_b_0)

// In the case of overflow on a column c (we want to increment the

// restriction range of c to the next range bound for that column but there

// are no restriction ranges remaining), we set the

// current column to the 0th range and move on to increment c - 1

// Note that in almost all cases the resulting current_scan_target_ is strictly

// greater (lesser in the case of backwards scans) than the original

// current_scan_target_. This is necessary to allow the iterator seek out

// of the current scan target. The exception to this rule is below.

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

// This function leaves the scan target as is if the next tuple in the current

// scan direction is also the next tuple in the filter space and start_col

// is given as the last column

Status HybridScanChoices::IncrementScanTargetAtColumn(int start_col) {

  VLOG(2) << __PRETTY_FUNCTION__

          << " Incrementing at " << start_col;

  // Increment start col, move backwards in case of overflow.

  int col_idx = start_col;

  // lower and upper here are taken relative to the scan order

  auto &lower_extremal_vector = is_forward_scan_

                          ? range_cols_scan_options_lower_

                            : range_cols_scan_options_upper_;

  auto &upper_extremal_vector = is_forward_scan_

                                ? range_cols_scan_options_upper_

                                  : range_cols_scan_options_lower_;

  DocKeyDecoder t_decoder(current_scan_target_);

  RETURN_NOT_OK(t_decoder.DecodeToRangeGroup());

  // refer to the documentation of this function to see what extremal

  // means here

  std::vector<bool> is_extremal;

  PrimitiveValue target_value;

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

    RETURN_NOT_OK(t_decoder.DecodePrimitiveValue(&target_value));

    is_extremal.push_back(target_value ==

      upper_extremal_vector[i][current_scan_target_idxs_[i]]);

  }

  // this variable tells us whether we start by appending

  // kHighest/kLowest at col_idx after the following for loop

  bool start_with_infinity = true;

  for (; col_idx >= 0; col_idx--) {

    const auto& choices = lower_extremal_vector[col_idx];

    auto it = current_scan_target_idxs_[col_idx];

    if (!is_extremal[col_idx]) {

      col_idx++;

      start_with_infinity = true;

      break;

    }

    if (++it < choices.size()) {

      // and if this value is at the extremal bound

      if (is_extremal[col_idx]) {

        current_scan_target_idxs_[col_idx]++;

        start_with_infinity = false;

      }

      break;

    }

    current_scan_target_idxs_[col_idx] = 0;

  }

  DocKeyDecoder decoder(current_scan_target_);

  RETURN_NOT_OK(decoder.DecodeToRangeGroup());

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

    RETURN_NOT_OK(decoder.DecodePrimitiveValue());

  }

  if (col_idx < 0) {

    // If we got here we finished all the options and are done.

    col_idx++;

    start_with_infinity = true;

    is_options_done_ = true;

  }

  current_scan_target_.Truncate(

      decoder.left_input().cdata() - current_scan_target_.AsSlice().cdata());

  if (start_with_infinity &&

        (col_idx < static_cast<int64>(current_scan_target_idxs_.size()))) {

    if (is_forward_scan_) {

      PrimitiveValue(ValueType::kHighest).AppendToKey(&current_scan_target_);

    } else {

      PrimitiveValue(ValueType::kLowest).AppendToKey(&current_scan_target_);

    }

    col_idx++;

  }

  if (start_with_infinity) {

    // there's no point in appending anything after infinity

    return Status::OK();

  }

  for (int i = col_idx; i <= start_col; ++i) {

      lower_extremal_vector[i][current_scan_target_idxs_[i]]

                                      .AppendToKey(&current_scan_target_);

  }

  for (size_t i = start_col + 1; i < current_scan_target_idxs_.size(); ++i) {

    current_scan_target_idxs_[i] = 0;

    lower_extremal_vector[i][current_scan_target_idxs_[i]]

                                    .AppendToKey(&current_scan_target_);

  }

  return Status::OK();

}

// Method called when the scan target is done being used

Status HybridScanChoices::DoneWithCurrentTarget() {

  // prev_scan_target_ is necessary for backwards scans

  prev_scan_target_ = current_scan_target_;

  RETURN_NOT_OK(IncrementScanTargetAtColumn(

                                  static_cast<int>(current_scan_target_idxs_.size()) - 1));

  current_scan_target_.AppendValueType(ValueType::kGroupEnd);

  // if we we incremented the last index then

  // if this is a forward scan it doesn't matter what we do

  // if this is a backwards scan then dont clear current_scan_target and we

  // stay live

  VLOG(2) << "After " << __PRETTY_FUNCTION__ << " current_scan_target_ is "

          << DocKey::DebugSliceToString(current_scan_target_);

  VLOG(2) << __PRETTY_FUNCTION__ << " moving on to next target";

  DCHECK(!FinishedWithScanChoices());

  if (is_options_done_) {

      // It could be possible that we finished all our options but are not

      // done because we haven't hit the bound key yet. This would usually be

      // the case if we are moving onto the next hash key where we will

      // restart our range options.

      const KeyBytes &bound_key = is_forward_scan_ ?

                                    upper_doc_key_ : lower_doc_key_;

      finished_ = bound_key.empty() ? false

                    : is_forward_scan_

                        == (current_scan_target_.CompareTo(bound_key) >= 0);

      VLOG(4) << "finished_ = " << finished_;

  }

  VLOG(4) << "current_scan_target_ is "

          << DocKey::DebugSliceToString(current_scan_target_)

          << " and prev_scan_target_ is "

          << DocKey::DebugSliceToString(prev_scan_target_);

  // The below condition is either indicative of the special case

  // where IncrementScanTargetAtColumn didn't change the target due

  // to the case specified in the last section of the

  // documentation for IncrementScanTargetAtColumn or we have exhausted

  // all available range keys for the given hash key (indicated

  // by is_options_done_)

  // We clear the scan target in these cases to indicate that the

  // current_scan_target_ has been used and is invalid

  // In all other cases, IncrementScanTargetAtColumn has updated

  // current_scan_target_ to the new value that we want to seek to.

  // Hence, we shouldn't clear it in those cases

  if ((prev_scan_target_ == current_scan_target_) || is_options_done_) {

      current_scan_target_.Clear();

  }

  return Status::OK();

}

// Seeks the given iterator to the current target as specified by

// current_scan_target_ and prev_scan_target_ (relevant in backwards

// scans)

Status HybridScanChoices::SeekToCurrentTarget(IntentAwareIterator* db_iter) {

  VLOG(2) << __PRETTY_FUNCTION__ << " Advancing iterator towards target";

  if (!FinishedWithScanChoices()) {

    // if current_scan_target_ is valid we use it to determine

    // what to seek to

    if (!current_scan_target_.empty()) {

      VLOG(3) << __PRETTY_FUNCTION__

              << " current_scan_target_ is non-empty. "

              << DocKey::DebugSliceToString(current_scan_target_);

      if (is_forward_scan_) {

        VLOG(3) << __PRETTY_FUNCTION__

                << " Seeking to "

                << DocKey::DebugSliceToString(current_scan_target_);

        db_iter->Seek(current_scan_target_);

      } else {

        // seek to the highest key <= current_scan_target_

        // seeking to the highest key < current_scan_target_ + kHighest

        // is equivalent to seeking to the highest key <=

        // current_scan_target_

        auto tmp = current_scan_target_;

        PrimitiveValue(ValueType::kHighest).AppendToKey(&tmp);

        VLOG(3) << __PRETTY_FUNCTION__ << " Going to PrevDocKey " << tmp;

        db_iter->PrevDocKey(tmp);

      }

    } else {

      if (!is_forward_scan_ && !prev_scan_target_.empty()) {

        db_iter->PrevDocKey(prev_scan_target_);

      }

    }

  }

  return Status::OK();

}

class RangeBasedScanChoices : public ScanChoices {

 public:

  RangeBasedScanChoices(const Schema& schema, const DocQLScanSpec& doc_spec)

      : ScanChoices(doc_spec.is_forward_scan()) {

    DCHECK(doc_spec.range_bounds());

    lower_.reserve(schema.num_range_key_columns());

    upper_.reserve(schema.num_range_key_columns());

    size_t idx = 0;

    for (idx = schema.num_hash_key_columns(); idx < schema.num_key_columns(); idx++) {

      const ColumnId col_idx = schema.column_id(idx);

      const auto col_sort_type = schema.column(idx).sorting_type();

      const QLScanRange::QLRange range = doc_spec.range_bounds()->RangeFor(col_idx);

      const auto lower = GetQLRangeBoundAsPVal(range, col_sort_type, true /* lower_bound */);

      const auto upper = GetQLRangeBoundAsPVal(range, col_sort_type, false /* upper_bound */);

      lower_.emplace_back(lower);

      upper_.emplace_back(upper);

    }

  }

  RangeBasedScanChoices(const Schema& schema, const DocPgsqlScanSpec& doc_spec)

      : ScanChoices(doc_spec.is_forward_scan()) {

    DCHECK(doc_spec.range_bounds());

    lower_.reserve(schema.num_range_key_columns());

    upper_.reserve(schema.num_range_key_columns());

    for (auto idx = schema.num_hash_key_columns(); idx < schema.num_key_columns(); idx++) {

      const ColumnId col_idx = schema.column_id(idx);

      const auto col_sort_type = schema.column(idx).sorting_type();

      const QLScanRange::QLRange range = doc_spec.range_bounds()->RangeFor(col_idx);

      const auto lower = GetQLRangeBoundAsPVal(range, col_sort_type, true /* lower_bound */);

      const auto upper = GetQLRangeBoundAsPVal(range, col_sort_type, false /* upper_bound */);

      lower_.emplace_back(lower);

      upper_.emplace_back(upper);

    }

  }

  CHECKED_STATUS SkipTargetsUpTo(const Slice& new_target) override;

  CHECKED_STATUS DoneWithCurrentTarget() override;

  CHECKED_STATUS SeekToCurrentTarget(IntentAwareIterator* db_iter) override;

 private:

  std::vector<PrimitiveValue> lower_, upper_;

  KeyBytes prev_scan_target_;

};

Status RangeBasedScanChoices::SkipTargetsUpTo(const Slice& new_target) {

  VLOG(2) << __PRETTY_FUNCTION__ << " Updating current target to be >= "

          << DocKey::DebugSliceToString(new_target);

  DCHECK(!FinishedWithScanChoices());

  /*

   Let's say we have a row key with (A B) as the hash part and C, D as the range part:

   ((A B) C D) E F

   Let's say we have a range constraint :

    l_c < C < u_c

     4        6

    a b  0 d  -> a  b l_c  d

    a b  5 d  -> a  b  5   d

                  [ Will subsequently seek out of document on reading the subdoc]

    a b  7 d  -> a <b> MAX

                [ This will seek to <b_next> and on the next invocation update:

                   a <b_next> ? ? -> a <b_next> l_c d ]

  */

  DocKeyDecoder decoder(new_target);

  RETURN_NOT_OK(decoder.DecodeToRangeGroup());

  current_scan_target_.Reset(Slice(new_target.data(), decoder.left_input().data()));

  size_t col_idx = 0;

  PrimitiveValue target_value;

  bool last_was_infinity = false;

  for (col_idx = 0; VERIFY_RESULT(decoder.HasPrimitiveValue()); col_idx++) {

    RETURN_NOT_OK(decoder.DecodePrimitiveValue(&target_value));

    VLOG(3) << "col_idx " << col_idx << " is " << target_value << " in ["

            << yb::ToString(lower_[col_idx]) << " , " << yb::ToString(upper_[col_idx]) << " ] ?";

    const auto& lower = lower_[col_idx];

    if (target_value < lower) {

      const auto tgt = (is_forward_scan_ ? lower : PrimitiveValue(ValueType::kLowest));

      tgt.AppendToKey(&current_scan_target_);

      last_was_infinity = tgt.IsInfinity();

      VLOG(3) << " Updating idx " << col_idx << " from " << target_value << " to " << tgt;

      break;

    }

    const auto& upper = upper_[col_idx];

    if (target_value > upper) {

      const auto tgt = (!is_forward_scan_ ? upper : PrimitiveValue(ValueType::kHighest));

      VLOG(3) << " Updating idx " << col_idx << " from " << target_value << " to " << tgt;

      tgt.AppendToKey(&current_scan_target_);

      last_was_infinity = tgt.IsInfinity();

      break;

    }

    target_value.AppendToKey(&current_scan_target_);

    last_was_infinity = target_value.IsInfinity();

  }

  // Reset the remaining range columns to kHighest/lower for forward scans

  // or kLowest/upper for backward scans.

  while (++col_idx < lower_.size()) {

    if (last_was_infinity) {

      // No point having more components after +/- Inf.

      break;

    }

    if (is_forward_scan_) {

      VLOG(3) << " Updating col_idx " << col_idx << " to " << lower_[col_idx];

      lower_[col_idx].AppendToKey(&current_scan_target_);

      last_was_infinity = lower_[col_idx].IsInfinity();

    } else {

      VLOG(3) << " Updating col_idx " << col_idx << " to " << upper_[col_idx];

      upper_[col_idx].AppendToKey(&current_scan_target_);

      last_was_infinity = upper_[col_idx].IsInfinity();

    }

  }

  current_scan_target_.AppendValueType(ValueType::kGroupEnd);

  VLOG(2) << "After " << __PRETTY_FUNCTION__ << " current_scan_target_ is "

          << DocKey::DebugSliceToString(current_scan_target_);

  return Status::OK();

}

Status RangeBasedScanChoices::DoneWithCurrentTarget() {

  prev_scan_target_ = current_scan_target_;

  current_scan_target_.Clear();

  return Status::OK();

}

Status RangeBasedScanChoices::SeekToCurrentTarget(IntentAwareIterator* db_iter) {

  VLOG(2) << __PRETTY_FUNCTION__ << " Advancing iterator towards target";

  if (!FinishedWithScanChoices()) {

    if (!current_scan_target_.empty()) {

      VLOG(3) << __PRETTY_FUNCTION__

              << " current_scan_target_ is non-empty. "

              << current_scan_target_;

      if (is_forward_scan_) {

        VLOG(3) << __PRETTY_FUNCTION__

                << " Seeking to "

                << DocKey::DebugSliceToString(current_scan_target_);

        db_iter->Seek(current_scan_target_);

      } else {