/Users/deen/code/yugabyte-db/src/yb/tserver/read_query.cc

Source (jump to first uncovered line)
// 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/tserver/read_query.h"

#include "yb/common/row_mark.h"
#include "yb/common/transaction.h"

#include "yb/gutil/bind.h"

#include "yb/tablet/operations/write_operation.h"
#include "yb/tablet/read_result.h"
#include "yb/tablet/tablet.h"
#include "yb/tablet/tablet_metadata.h"
#include "yb/tablet/tablet_metrics.h"
#include "yb/tablet/transaction_participant.h"
#include "yb/tablet/write_query.h"

#include "yb/tserver/service_util.h"
#include "yb/tserver/tablet_server_interface.h"
#include "yb/tserver/ts_tablet_manager.h"
#include "yb/tserver/tserver.pb.h"

#include "yb/util/countdown_latch.h"
#include "yb/util/debug/trace_event.h"
#include "yb/util/flag_tags.h"
#include "yb/util/metrics.h"
#include "yb/util/scope_exit.h"
#include "yb/util/trace.h"

using namespace std::literals;

DEFINE_test_flag(int32, transactional_read_delay_ms, 0,
                 "Amount of time to delay between transaction status check and reading start.");

DEFINE_test_flag(int32, simulate_time_out_failures_msecs, 0, "If greater than 0, we will randomly "
                 "mark read requests as timed out and sleep for the specificed amount of time by "
                 "this flag to simulate time out failures. The requester will mark the timed out "
                 "replica as failed, and its periodic refresh mechanism for the lookup cache will "
                 "mark them as available.");

DEFINE_test_flag(bool, assert_reads_served_by_follower, false, "If set, we verify that the "
                 "consistency level is CONSISTENT_PREFIX, and that this server is not the leader "
                 "for the tablet");

DEFINE_bool(parallelize_read_ops, true,
            "Controls whether multiple (Redis) read ops that are present in a operation "
            "should be executed in parallel.");
TAG_FLAG(parallelize_read_ops, advanced);
TAG_FLAG(parallelize_read_ops, runtime);

namespace yb {
namespace tserver {

namespace {

void HandleRedisReadRequestAsync(
    tablet::AbstractTablet* tablet,
    CoarseTimePoint deadline,
    const ReadHybridTime& read_time,
    const RedisReadRequestPB& redis_read_request,
    RedisResponsePB* response,
    const std::function<void(const Status& s)>& status_cb
) {
  status_cb(tablet->HandleRedisReadRequest(deadline, read_time, redis_read_request, response));
}

class ReadQuery : public std::enable_shared_from_this<ReadQuery>, public rpc::ThreadPoolTask {
 public:
  ReadQuery(
      TabletServerIf* server, ReadTabletProvider* read_tablet_provider, const ReadRequestPB* req,
      ReadResponsePB* resp, rpc::RpcContext context)
      : server_(*server), read_tablet_provider_(*read_tablet_provider), req_(req), resp_(resp),
        context_(std::move(context)) {}

  void Perform() {
    RespondIfFailed(DoPerform());
  }

  void RespondIfFailed(const Status& status) {
    if (!status.ok()) {
      RespondFailure(status);
    }
  }

  void RespondFailure(const Status& status) {
    SetupErrorAndRespond(resp_->mutable_error(), status, &context_);
  }

  virtual ~ReadQuery() = default;

 private:
  CHECKED_STATUS DoPerform();

  // Picks read based for specified read context.
  CHECKED_STATUS DoPickReadTime(server::Clock* clock);

  bool transactional() const;

  tablet::Tablet* tablet() const;

  ReadHybridTime FormRestartReadHybridTime(const HybridTime& restart_time) const;

  CHECKED_STATUS PickReadTime(server::Clock* clock);

  bool IsForBackfill() const;

  // Read implementation. If restart is required returns restart time, in case of success
  // returns invalid ReadHybridTime. Otherwise returns error status.
  Result<ReadHybridTime> DoRead();
  Result<ReadHybridTime> DoReadImpl();

  CHECKED_STATUS Complete();

  void UpdateConsistentPrefixMetrics();

  // Used when we write intents during read, i.e. for serializable isolation.
  // We cannot proceed with read from completion callback, to avoid holding
  // replica state lock for too long.
  // So ThreadPool is used to proceed with read.
  void Run() override {
    auto status = PickReadTime(server_.Clock());
    if (status.ok()) {
      status = Complete();
    }
    RespondIfFailed(status);
  }

  void Done(const Status& status) override {
    RespondIfFailed(status);
    retained_self_ = nullptr;
  }

  TabletServerIf& server_;
  ReadTabletProvider& read_tablet_provider_;
  const ReadRequestPB* req_;
  ReadResponsePB* resp_;
  rpc::RpcContext context_;

  std::shared_ptr<tablet::AbstractTablet> abstract_tablet_;

  ReadHybridTime read_time_;
  HybridTime safe_ht_to_read_;
  ReadHybridTime used_read_time_;
  tablet::RequireLease require_lease_ = tablet::RequireLease::kFalse;
  HostPortPB host_port_pb_;
  bool allow_retry_ = false;
  RequestScope request_scope_;
  std::shared_ptr<ReadQuery> retained_self_;
};

bool ReadQuery::transactional() const {
  return abstract_tablet_->IsTransactionalRequest(req_->pgsql_batch_size() > 0);
}

tablet::Tablet* ReadQuery::tablet() const {
  return down_cast<tablet::Tablet*>(abstract_tablet_.get());
}

ReadHybridTime ReadQuery::FormRestartReadHybridTime(const HybridTime& restart_time) const {
  DCHECK_GT(restart_time, read_time_.read);
  VLOG(1) << "Restart read required at: " << restart_time << ", original: " << read_time_;
  auto result = read_time_;
  result.read = std::min(std::max(restart_time, safe_ht_to_read_), read_time_.global_limit);
  result.local_limit = std::min(safe_ht_to_read_, read_time_.global_limit);
  return result;
}

CHECKED_STATUS ReadQuery::PickReadTime(server::Clock* clock) {
  auto result = DoPickReadTime(clock);
  if (!result.ok()) {
    TRACE(result.ToString());
  }
  return result;
}

bool ReadQuery::IsForBackfill() const {
  if (req_->pgsql_batch_size() > 0) {
    if (req_->pgsql_batch(0).is_for_backfill()) {
      // Currently, read requests for backfill should only come by themselves, not in batches.
      DCHECK_EQ(req_->pgsql_batch_size(), 1);
      return true;
    }
  }
  // YCQL doesn't send read RPCs for scanning the indexed table and instead directly reads using
  // iterator, so there's no equivalent logic for YCQL here.
  return false;
}

CHECKED_STATUS ReadQuery::DoPerform() {
  TRACE("Start Read");
  TRACE_EVENT1("tserver", "TabletServiceImpl::Read", "tablet_id", req_->tablet_id());
  VLOG(2) << "Received Read RPC: " << req_->DebugString();
  // Unfortunately, determining the isolation level is not as straightforward as it seems. All but
  // the first request to a given tablet by a particular transaction assume that the tablet already
  // has the transaction metadata, including the isolation level, and those requests expect us to
  // retrieve the isolation level from that metadata. Failure to do so was the cause of a
  // serialization anomaly tested by TestOneOrTwoAdmins
  // (https://github.com/yugabyte/yugabyte-db/issues/1572).

  bool serializable_isolation = false;
  TabletPeerTablet peer_tablet;
  if (req_->has_transaction()) {
    IsolationLevel isolation_level;
    if (req_->transaction().has_isolation()) {
      // This must be the first request to this tablet by this particular transaction.
      isolation_level = req_->transaction().isolation();
    } else {
      peer_tablet = VERIFY_RESULT(LookupTabletPeer(
          server_.tablet_peer_lookup(), req_->tablet_id()));
      isolation_level = VERIFY_RESULT(peer_tablet.tablet->GetIsolationLevelFromPB(*req_));
    }
    serializable_isolation = isolation_level == IsolationLevel::SERIALIZABLE_ISOLATION;

    if (PREDICT_FALSE(FLAGS_TEST_transactional_read_delay_ms > 0)) {
      LOG(INFO) << "Delaying transactional read for "
                << FLAGS_TEST_transactional_read_delay_ms << " ms.";
      SleepFor(MonoDelta::FromMilliseconds(FLAGS_TEST_transactional_read_delay_ms));
    }

#if defined(DUMP_READ)
    if (req->pgsql_batch().size() > 0) {
      LOG(INFO) << CHECK_RESULT(FullyDecodeTransactionId(req->transaction().transaction_id()))
                << " READ: " << req->pgsql_batch(0).partition_column_values(0).value().int32_value()
                << ", " << isolation_level;
    }
#endif
  }

  // Get the most restrictive row mark present in the batch of PostgreSQL requests.
  // TODO: rather handle individual row marks once we start batching read requests (issue #2495)
  RowMarkType batch_row_mark = RowMarkType::ROW_MARK_ABSENT;
  if (!req_->pgsql_batch().empty()) {
    uint64_t last_breaking_catalog_version = 0; // unset.
    for (const auto& pg_req : req_->pgsql_batch()) {
      // For postgres requests check that the syscatalog version matches.
      if (pg_req.has_ysql_catalog_version()) {
        if (last_breaking_catalog_version == 0) {
          // Initialize last breaking version if not yet set.
          server_.get_ysql_catalog_version(
              nullptr /* current_version */, &last_breaking_catalog_version);
        }
        if (pg_req.ysql_catalog_version() < last_breaking_catalog_version) {
          return STATUS(
              QLError, "The catalog snapshot used for this transaction has been invalidated",
              TabletServerError(TabletServerErrorPB::MISMATCHED_SCHEMA));
        }
      }
      RowMarkType current_row_mark = GetRowMarkTypeFromPB(pg_req);
      if (IsValidRowMarkType(current_row_mark)) {
        if (!req_->has_transaction()) {
          return STATUS(
              NotSupported, "Read request with row mark types must be part of a transaction",
              TabletServerError(TabletServerErrorPB::OPERATION_NOT_SUPPORTED));
        }
        batch_row_mark = GetStrongestRowMarkType({current_row_mark, batch_row_mark});
      }
    }
  }
  const bool has_row_mark = IsValidRowMarkType(batch_row_mark);

  LeaderTabletPeer leader_peer;

  if (serializable_isolation || has_row_mark) {
    // At this point we expect that we don't have pure read serializable transactions, and
    // always write read intents to detect conflicts with other writes.
    leader_peer = VERIFY_RESULT(LookupLeaderTablet(
        server_.tablet_peer_lookup(), req_->tablet_id(), std::move(peer_tablet)));
    // Serializable read adds intents, i.e. writes data.
    // We should check for memory pressure in this case.
    RETURN_NOT_OK(CheckWriteThrottling(req_->rejection_score(), leader_peer.peer.get()));
    abstract_tablet_ = leader_peer.peer->shared_tablet();
  } else {
    abstract_tablet_ = VERIFY_RESULT(read_tablet_provider_.GetTabletForRead(
        req_->tablet_id(), std::move(peer_tablet.tablet_peer),
        req_->consistency_level(), AllowSplitTablet::kFalse));
    leader_peer.leader_term = OpId::kUnknownTerm;
  }

  if (PREDICT_FALSE(FLAGS_TEST_assert_reads_served_by_follower)) {
    if (req_->consistency_level() == YBConsistencyLevel::STRONG) {
      LOG(FATAL) << "--TEST_assert_reads_served_by_follower is true but consistency level is "
                    "invalid: YBConsistencyLevel::STRONG";
    }
    tablet::TabletPeerPtr tablet_peer;
    RETURN_NOT_OK(server_.tablet_peer_lookup()->GetTabletPeer(req_->tablet_id(), &tablet_peer));
    if (CheckPeerIsLeader(*tablet_peer).ok()) {
      LOG(FATAL) << "--TEST_assert_reads_served_by_follower is true but read is being served by "
                 << " peer " << tablet_peer->permanent_uuid()
                 << " which is the leader for tablet " << req_->tablet_id();
    }
  }

  if (!abstract_tablet_->system() && tablet()->metadata()->hidden()) {
    return STATUS(NotFound, "Tablet not found", req_->tablet_id());
  }

  if (FLAGS_TEST_simulate_time_out_failures_msecs > 0 && RandomUniformInt(0, 10) < 2) {
    LOG(INFO) << "Marking request as timed out for test: " << req_->ShortDebugString();
    SleepFor(MonoDelta::FromMilliseconds(FLAGS_TEST_simulate_time_out_failures_msecs));
    return STATUS(TimedOut, "timed out for test");
  }

  if (server_.Clock()) {
    server::UpdateClock(*req_, server_.Clock());
  }

  // safe_ht_to_read is used only for read restart, so if read_time is valid, then we would respond
  // with "restart required".
  read_time_ = ReadHybridTime::FromReadTimePB(*req_);

  allow_retry_ = !read_time_;
  require_lease_ = tablet::RequireLease(req_->consistency_level() == YBConsistencyLevel::STRONG);
  // TODO: should check all the tables referenced by the requests to decide if it is transactional.
  const bool transactional = this->transactional();
  // Should not pick read time for serializable isolation, since it is picked after read intents
  // are added. Also conflict resolution for serializable isolation should be done without read time
  // specified. So we use max hybrid time for conflict resolution in such case.
  // It was implemented as part of #655.
  if (!serializable_isolation) {
    RETURN_NOT_OK(PickReadTime(server_.Clock()));
  }

  if (transactional) {
    // Serial number is used to check whether this operation was initiated before
    // transaction status request. So we should initialize it as soon as possible.
    request_scope_ = RequestScope(tablet()->transaction_participant());
    read_time_.serial_no = request_scope_.request_id();
  }

  const auto& remote_address = context_.remote_address();
  host_port_pb_.set_host(remote_address.address().to_string());
  host_port_pb_.set_port(remote_address.port());

  if (serializable_isolation || has_row_mark) {
    auto deadline = context_.GetClientDeadline();
    auto query = std::make_unique<tablet::WriteQuery>(
        leader_peer.leader_term, deadline, leader_peer.peer.get(),
        leader_peer.peer->tablet(), nullptr /* response */,
        docdb::OperationKind::kRead);

    auto& write = *query->operation().AllocateRequest();
    auto& write_batch = *write.mutable_write_batch();
    *write_batch.mutable_transaction() = req_->transaction();
    if (has_row_mark) {
      write_batch.set_row_mark_type(batch_row_mark);
      query->set_read_time(read_time_);
    }
    write.set_unused_tablet_id(""); // For backward compatibility.
    write_batch.set_deprecated_may_have_metadata(true);
    write.set_batch_idx(req_->batch_idx());
    // TODO(dtxn) write request id

    RETURN_NOT_OK(leader_peer.peer->tablet()->CreateReadIntents(
        req_->transaction(), req_->subtransaction(), req_->ql_batch(), req_->pgsql_batch(),
        &write_batch));

    query->AdjustYsqlQueryTransactionality(req_->pgsql_batch_size());

    query->set_callback([peer = leader_peer.peer, self = shared_from_this()](const Status& status) {
      if (!status.ok()) {
        self->RespondFailure(status);
      } else {
        self->retained_self_ = self;
        peer->Enqueue(self.get());
      }
    });
    leader_peer.peer->WriteAsync(std::move(query));
    return Status::OK();
  }

  if (req_->consistency_level() == YBConsistencyLevel::CONSISTENT_PREFIX) {
    if (abstract_tablet_) {
      tablet()->metrics()->consistent_prefix_read_requests->Increment();
    }
  }

  return Complete();
}

CHECKED_STATUS ReadQuery::DoPickReadTime(server::Clock* clock) {
  if (!read_time_) {
    safe_ht_to_read_ = VERIFY_RESULT(abstract_tablet_->SafeTime(require_lease_));
    // If the read time is not specified, then it is a single-shard read.
    // So we should restart it in server in case of failure.
    read_time_.read = safe_ht_to_read_;
    if (transactional()) {
      read_time_.global_limit = clock->MaxGlobalNow();
      read_time_.local_limit = std::min(safe_ht_to_read_, read_time_.global_limit);

      VLOG(1) << "Read time: " << read_time_.ToString();
    } else {
      read_time_.local_limit = read_time_.read;
      read_time_.global_limit = read_time_.read;
    }
  } else {
    safe_ht_to_read_ = VERIFY_RESULT(abstract_tablet_->SafeTime(
        require_lease_, read_time_.read, context_.GetClientDeadline()));
  }
  return Status::OK();
}

CHECKED_STATUS ReadQuery::Complete() {
  for (;;) {
    resp_->Clear();
    context_.ResetRpcSidecars();
    VLOG(1) << "Read time: " << read_time_ << ", safe: " << safe_ht_to_read_;
    const auto result = VERIFY_RESULT(DoRead());
    if (allow_retry_ && read_time_ && read_time_ == result) {
      YB_LOG_EVERY_N_SECS(DFATAL, 5)
          << __func__ << ", restarting read with the same read time: " << result << THROTTLE_MSG;
      allow_retry_ = false;
    }
    read_time_ = result;
    // If read was successful, then restart time is invalid. Finishing.
    // (If a read restart was requested, then read_time would be set to the time at which we have
    // to restart.)
    if (!read_time_) {
      // allow_retry means that that the read time was not set in the request and therefore we can
      // retry read restarts on the tablet server.
      if (!allow_retry_) {
        auto local_limit = std::min(safe_ht_to_read_, used_read_time_.global_limit);
        resp_->set_local_limit_ht(local_limit.ToUint64());
      }
      break;
    }
    if (!allow_retry_) {
      // If the read time is specified, then we read as part of a transaction. So we should restart
      // whole transaction. In this case we report restart time and abort reading.
      resp_->Clear();
      auto restart_read_time = resp_->mutable_restart_read_time();
      restart_read_time->set_read_ht(read_time_.read.ToUint64());
      restart_read_time->set_deprecated_max_of_read_time_and_local_limit_ht(
          read_time_.local_limit.ToUint64());
      restart_read_time->set_local_limit_ht(read_time_.local_limit.ToUint64());
      // Global limit is ignored by caller, so we don't set it.
      tablet()->metrics()->restart_read_requests->Increment();
      break;
    }

    if (CoarseMonoClock::now() > context_.GetClientDeadline()) {
      TRACE("Read timed out");
      return STATUS(TimedOut, "Read timed out");
    }
  }
  if (req_->include_trace() && Trace::CurrentTrace() != nullptr) {
    resp_->set_trace_buffer(Trace::CurrentTrace()->DumpToString(true));
  }

  // In case read time was not specified (i.e. allow_retry is true)
  // we just picked a read time and we should communicate it back to the caller.
  if (allow_retry_) {
    used_read_time_.ToPB(resp_->mutable_used_read_time());
  }

  // Useful when debugging transactions
#if defined(DUMP_READ)
  if (read_context->req->has_transaction() && read_context->req->pgsql_batch().size() == 1 &&
      read_context->req->pgsql_batch()[0].partition_column_values().size() == 1 &&
      read_context->resp->pgsql_batch().size() == 1 &&
      read_context->resp->pgsql_batch()[0].rows_data_sidecar() == 0) {
    auto txn_id = CHECK_RESULT(FullyDecodeTransactionId(
        read_context->req->transaction().transaction_id()));
    auto value_slice = read_context->context->RpcSidecar(0).as_slice();
    auto num = BigEndian::Load64(value_slice.data());
    std::string result;
    if (num == 0) {
      result = "<NONE>";
    } else if (num == 1) {
      auto len = BigEndian::Load64(value_slice.data() + 14) - 1;
      result = Slice(value_slice.data() + 22, len).ToBuffer();
    } else {
      result = value_slice.ToDebugHexString();
    }
    auto key = read_context->req->pgsql_batch(0).partition_column_values(0).value().int32_value();
    LOG(INFO) << txn_id << " READ DONE: " << key << " = " << result;
  }
#endif

  MakeRpcOperationCompletionCallback<ReadResponsePB>(
      std::move(context_), resp_, server_.Clock())(Status::OK());
  TRACE("Done Read");

  return Status::OK();
}

Result<ReadHybridTime> ReadQuery::DoRead() {
  Result<ReadHybridTime> result{ReadHybridTime()};
  {
    LongOperationTracker long_operation_tracker("Read", 1s);
    result = DoReadImpl();
  }
  // Check transaction is still alive in case read was successful
  // and data has been written earlier into current tablet in context of current transaction.
  const auto* transaction =
      req_->has_transaction() ? &req_->transaction() : nullptr;
  if (result.ok() && transaction && transaction->isolation() == NON_TRANSACTIONAL) {
    const auto txn_id = VERIFY_RESULT(FullyDecodeTransactionId(transaction->transaction_id()));
    auto& txn_participant = *tablet()->transaction_participant();
    RETURN_NOT_OK(txn_participant.CheckAborted(txn_id));
  }
  return result;
}

Result<ReadHybridTime> ReadQuery::DoReadImpl() {
  ReadHybridTime read_time;
  tablet::ScopedReadOperation read_tx;
  if (IsForBackfill()) {
    read_time = read_time_;
  } else {
    read_tx = VERIFY_RESULT(
        tablet::ScopedReadOperation::Create(abstract_tablet_.get(), require_lease_, read_time_));
    read_time = read_tx.read_time();
  }
  used_read_time_ = read_time;
  if (!req_->redis_batch().empty()) {
    // Assert the primary table is a redis table.
    DCHECK_EQ(abstract_tablet_->table_type(), TableType::REDIS_TABLE_TYPE);
    auto count = req_->redis_batch_size();
    std::vector<Status> rets(count);
    CountDownLatch latch(count);
    for (int idx = 0; idx < count; idx++) {
      const RedisReadRequestPB& redis_read_req = req_->redis_batch(idx);
      Status &failed_status_ = rets[idx];
      auto cb = [&latch, &failed_status_] (const Status &status) -> void {
                  if (!status.ok())
                    failed_status_ = status;
                  latch.CountDown(1);
                };
      auto func = Bind(
          &HandleRedisReadRequestAsync,
          Unretained(abstract_tablet_.get()),
          context_.GetClientDeadline(),
          read_time,
          redis_read_req,
          Unretained(resp_->add_redis_batch()),
          cb);

      Status s;
      bool run_async = FLAGS_parallelize_read_ops && (idx != count - 1);
      if (run_async) {
        s = server_.tablet_manager()->read_pool()->SubmitClosure(func);
      }

      if (!s.ok() || !run_async) {
        func.Run();
      }
    }
    latch.Wait();
    std::vector<Status> failed;
    for (auto& status : rets) {
      if (!status.ok()) {
        failed.push_back(status);
      }
    }
    if (failed.size() == 0) {
      // TODO(dtxn) implement read restart for Redis.
      return ReadHybridTime();
    } else if (failed.size() == 1) {
      return failed[0];
    } else {
      return STATUS(Combined, VectorToString(failed));
    }
  }

  if (!req_->ql_batch().empty()) {
    // Assert the primary table is a YQL table.
    DCHECK_EQ(abstract_tablet_->table_type(), TableType::YQL_TABLE_TYPE);
    ReadRequestPB* mutable_req = const_cast<ReadRequestPB*>(req_);
    for (QLReadRequestPB& ql_read_req : *mutable_req->mutable_ql_batch()) {
      // Update the remote endpoint.
      ql_read_req.set_allocated_remote_endpoint(&host_port_pb_);
      ql_read_req.set_allocated_proxy_uuid(mutable_req->mutable_proxy_uuid());
      auto se = ScopeExit([&ql_read_req] {
        ql_read_req.release_remote_endpoint();
        ql_read_req.release_proxy_uuid();
      });

      tablet::QLReadRequestResult result;
      TRACE("Start HandleQLReadRequest");
      RETURN_NOT_OK(abstract_tablet_->HandleQLReadRequest(
          context_.GetClientDeadline(), read_time, ql_read_req, req_->transaction(), &result));
      TRACE("Done HandleQLReadRequest");
      if (result.restart_read_ht.is_valid()) {
        return FormRestartReadHybridTime(result.restart_read_ht);
      }
      result.response.set_rows_data_sidecar(
          narrow_cast<int32_t>(context_.AddRpcSidecar(result.rows_data)));
      resp_->add_ql_batch()->Swap(&result.response);
    }
    return ReadHybridTime();
  }

  if (!req_->pgsql_batch().empty()) {
    ReadRequestPB* mutable_req = const_cast<ReadRequestPB*>(req_);
    size_t total_num_rows_read = 0;
    for (PgsqlReadRequestPB& pgsql_read_req : *mutable_req->mutable_pgsql_batch()) {
      tablet::PgsqlReadRequestResult result;
      TRACE("Start HandlePgsqlReadRequest");
      size_t num_rows_read;
      RETURN_NOT_OK(abstract_tablet_->HandlePgsqlReadRequest(
          context_.GetClientDeadline(), read_time,
          !allow_retry_ /* is_explicit_request_read_time */, pgsql_read_req, req_->transaction(),
          req_->subtransaction(), &result, &num_rows_read));

      total_num_rows_read += num_rows_read;

      TRACE("Done HandlePgsqlReadRequest");
      if (result.restart_read_ht.is_valid()) {
        return FormRestartReadHybridTime(result.restart_read_ht);
      }
      result.response.set_rows_data_sidecar(
          narrow_cast<int32_t>(context_.AddRpcSidecar(result.rows_data)));
      resp_->add_pgsql_batch()->Swap(&result.response);
    }

    if (req_->consistency_level() == YBConsistencyLevel::CONSISTENT_PREFIX &&
        total_num_rows_read > 0) {
      tablet()->metrics()->pgsql_consistent_prefix_read_rows->IncrementBy(total_num_rows_read);
    }
    return ReadHybridTime();
  }

  if (abstract_tablet_->table_type() == TableType::TRANSACTION_STATUS_TABLE_TYPE) {
    return STATUS(NotSupported, "Transaction status table does not support read");
  }

  return ReadHybridTime();
}

} // namespace

void PerformRead(
    TabletServerIf* server, ReadTabletProvider* read_tablet_provider,
    const ReadRequestPB* req, ReadResponsePB* resp, rpc::RpcContext context) {
  auto read_query = std::make_shared<ReadQuery>(
      server, read_tablet_provider, req, resp, std::move(context));
  read_query->Perform();
}

}  // namespace tserver
}  // namespace yb

YugabyteDB (2.13.0.0-b42, bfc6a6643e7399ac8a0e81d06a3ee6d6571b33ab)

Coverage Report

Created: 2022-03-09 17:30