You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kudu.apache.org by al...@apache.org on 2021/10/08 00:48:01 UTC
[kudu] branch master updated (7b0a94d -> a50091e)
This is an automated email from the ASF dual-hosted git repository.
alexey pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/kudu.git.
from 7b0a94d [common] add PartitionSchema::HasCustomHashSchemas() method
new 4fe61a2 [rpc] allow reuse of outbound request buffers when retrying
new 3884a63 [consensus] KUDU-1620: re-resolve consensus peers on network error
new a50091e KUDU-2671 introduce PartitionKey
The 3 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
src/kudu/client/batcher.cc | 2 +
src/kudu/client/client-test.cc | 54 +-
src/kudu/client/client.cc | 27 +-
src/kudu/client/client.h | 1 +
src/kudu/client/flex_partitioning_client-test.cc | 2 +-
src/kudu/client/meta_cache.cc | 98 ++--
src/kudu/client/meta_cache.h | 35 +-
src/kudu/client/partitioner-internal.cc | 16 +-
src/kudu/client/partitioner-internal.h | 5 +-
src/kudu/client/scan_configuration.cc | 12 +-
src/kudu/client/scan_configuration.h | 5 +-
src/kudu/client/scan_token-internal.cc | 27 +-
src/kudu/client/scanner-internal.cc | 12 +-
src/kudu/client/scanner-internal.h | 3 +-
src/kudu/common/common.proto | 2 +-
src/kudu/common/partition-test.cc | 558 +++++++++++----------
src/kudu/common/partition.cc | 255 +++++-----
src/kudu/common/partition.h | 118 ++++-
src/kudu/common/partition_pruner-test.cc | 148 +++---
src/kudu/common/partition_pruner.cc | 37 +-
src/kudu/common/partition_pruner.h | 12 +-
src/kudu/common/scan_spec.cc | 12 +-
src/kudu/common/scan_spec.h | 20 +-
src/kudu/consensus/consensus_peers.cc | 15 +-
src/kudu/integration-tests/dns_alias-itest.cc | 178 +++++++
.../integration-tests/table_locations-itest.cc | 43 +-
.../integration-tests/tablet_replacement-itest.cc | 8 +-
src/kudu/master/catalog_manager-test.cc | 70 ++-
src/kudu/master/catalog_manager.cc | 149 ++++--
src/kudu/master/catalog_manager.h | 16 +-
src/kudu/master/master.proto | 27 +-
src/kudu/master/master_path_handlers.cc | 4 +-
src/kudu/mini-cluster/external_mini_cluster.h | 6 +
src/kudu/mini-cluster/mini_cluster.cc | 5 +-
src/kudu/mini-cluster/mini_cluster.h | 4 +-
src/kudu/rpc/mt-rpc-test.cc | 6 +-
src/kudu/rpc/outbound_call.cc | 83 ++-
src/kudu/rpc/outbound_call.h | 95 +++-
src/kudu/rpc/proxy-test.cc | 48 ++
src/kudu/rpc/proxy.cc | 58 ++-
src/kudu/rpc/proxy.h | 8 +-
src/kudu/rpc/rpc-test-base.h | 29 +-
src/kudu/rpc/rpc_controller.cc | 14 +
src/kudu/rpc/rpc_controller.h | 17 +
src/kudu/tools/table_scanner.cc | 4 +-
src/kudu/tools/tool_action_table.cc | 4 +-
src/kudu/util/net/net_util.cc | 8 +-
47 files changed, 1560 insertions(+), 800 deletions(-)
[kudu] 02/03: [consensus] KUDU-1620: re-resolve consensus peers on
network error
Posted by al...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
alexey pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/kudu.git
commit 3884a6388b2696a826b8903144ae555faa595473
Author: Andrew Wong <aw...@cloudera.com>
AuthorDate: Sat Sep 18 00:50:39 2021 -0700
[consensus] KUDU-1620: re-resolve consensus peers on network error
This plumbs the work from KUDU-75 into the long-lived consensus proxy,
allowing Raft peers to re-resolve on error.
This has the knock-on effect that masters starting up also re-resolve
other masters' address when attempting to fetch UUIDs, since this
process also uses consensus proxies.
Change-Id: Ibd1b68c3c14d7d8f81168e16fe450d2ffcce840b
Reviewed-on: http://gerrit.cloudera.org:8080/17868
Tested-by: Kudu Jenkins
Reviewed-by: Alexey Serbin <as...@cloudera.com>
---
src/kudu/consensus/consensus_peers.cc | 15 +-
src/kudu/integration-tests/dns_alias-itest.cc | 178 +++++++++++++++++++++
.../integration-tests/tablet_replacement-itest.cc | 8 +-
src/kudu/mini-cluster/external_mini_cluster.h | 6 +
src/kudu/mini-cluster/mini_cluster.cc | 5 +-
src/kudu/mini-cluster/mini_cluster.h | 4 +-
src/kudu/rpc/proxy.cc | 4 +
src/kudu/util/net/net_util.cc | 8 +-
8 files changed, 209 insertions(+), 19 deletions(-)
diff --git a/src/kudu/consensus/consensus_peers.cc b/src/kudu/consensus/consensus_peers.cc
index 9e1b9fe..27dd2a1 100644
--- a/src/kudu/consensus/consensus_peers.cc
+++ b/src/kudu/consensus/consensus_peers.cc
@@ -48,9 +48,7 @@
#include "kudu/util/flag_tags.h"
#include "kudu/util/logging.h"
#include "kudu/util/monotime.h"
-#include "kudu/util/net/dns_resolver.h"
#include "kudu/util/net/net_util.h"
-#include "kudu/util/net/sockaddr.h"
#include "kudu/util/pb_util.h"
#include "kudu/util/status.h"
#include "kudu/util/threadpool.h"
@@ -321,7 +319,8 @@ void Peer::StartElection() {
// TODO(adar): lack of C++14 move capture makes for ugly code.
RunLeaderElectionResponsePB* resp = resp_uniq.release();
RpcController* controller = controller_uniq.release();
- proxy_->StartElectionAsync(req, resp, controller, [resp, controller, peer_uuid]() {
+ auto s_this = shared_from_this();
+ proxy_->StartElectionAsync(req, resp, controller, [resp, controller, peer_uuid, s_this]() {
unique_ptr<RunLeaderElectionResponsePB> r(resp);
unique_ptr<RpcController> c(controller);
string error_msg = Substitute("unable to start election on peer $0", peer_uuid);
@@ -588,14 +587,8 @@ Status CreateConsensusServiceProxyForHost(
const shared_ptr<Messenger>& messenger,
DnsResolver* dns_resolver,
unique_ptr<ConsensusServiceProxy>* new_proxy) {
- vector<Sockaddr> addrs;
- RETURN_NOT_OK(dns_resolver->ResolveAddresses(hostport, &addrs));
- if (addrs.size() > 1) {
- LOG(WARNING) << Substitute(
- "Peer address '$0' resolves to $1 different addresses. "
- "Using $2", hostport.ToString(), addrs.size(), addrs[0].ToString());
- }
- new_proxy->reset(new ConsensusServiceProxy(messenger, addrs[0], hostport.host()));
+ new_proxy->reset(new ConsensusServiceProxy(messenger, hostport, dns_resolver));
+ (*new_proxy)->Init();
return Status::OK();
}
diff --git a/src/kudu/integration-tests/dns_alias-itest.cc b/src/kudu/integration-tests/dns_alias-itest.cc
index 0e92cc3..a801005 100644
--- a/src/kudu/integration-tests/dns_alias-itest.cc
+++ b/src/kudu/integration-tests/dns_alias-itest.cc
@@ -16,6 +16,7 @@
// under the License.
#include <cstdint>
+#include <functional>
#include <memory>
#include <string>
#include <utility>
@@ -26,15 +27,19 @@
#include "kudu/client/client.h"
#include "kudu/gutil/stl_util.h"
+#include "kudu/gutil/strings/join.h"
#include "kudu/gutil/strings/split.h"
#include "kudu/gutil/strings/substitute.h"
#include "kudu/integration-tests/cluster_itest_util.h"
+#include "kudu/integration-tests/cluster_verifier.h"
#include "kudu/integration-tests/test_workload.h"
#include "kudu/mini-cluster/external_mini_cluster.h"
+#include "kudu/mini-cluster/mini_cluster.h"
#include "kudu/util/metrics.h"
#include "kudu/util/monotime.h"
#include "kudu/util/net/net_util.h"
#include "kudu/util/net/sockaddr.h"
+#include "kudu/util/net/socket.h"
#include "kudu/util/status.h"
#include "kudu/util/test_macros.h"
#include "kudu/util/test_util.h"
@@ -48,6 +53,7 @@ METRIC_DECLARE_entity(server);
using kudu::client::KuduClient;
using kudu::client::KuduTabletServer;
+using kudu::cluster::ExternalDaemon;
using kudu::cluster::ExternalMiniCluster;
using kudu::cluster::ExternalMiniClusterOptions;
using std::string;
@@ -89,6 +95,73 @@ class DnsAliasITest : public KuduTest {
void TearDown() override {
NO_FATALS(cluster_->AssertNoCrashes());
}
+
+ // Get the new DNS override string when restarting the last node of the given
+ // daemon type with the given reserved address.
+ string GetNewOverridesFlag(ExternalMiniCluster::DaemonType node_type,
+ const Sockaddr& new_addr) {
+ int master_end_idx = cluster_->num_masters();
+ int tserver_end_idx = cluster_->num_tablet_servers();
+ bool is_master = node_type == ExternalMiniCluster::DaemonType::MASTER;
+ if (is_master) {
+ --master_end_idx;
+ } else {
+ --tserver_end_idx;
+ }
+ vector<string> new_overrides;
+ new_overrides.reserve(cluster_->num_masters() + cluster_->num_tablet_servers());
+ for (int i = 0; i < master_end_idx; i++) {
+ new_overrides.emplace_back(Substitute("$0.$1=$2", kMasterHostPrefix, i,
+ cluster_->master(i)->bound_rpc_addr().ToString()));
+ }
+ for (int i = 0; i < tserver_end_idx; i++) {
+ new_overrides.emplace_back(
+ Substitute("$0.$1=$2", kTServerHostPrefix, i,
+ cluster_->tablet_server(i)->bound_rpc_addr().ToString()));
+ }
+ new_overrides.emplace_back(
+ Substitute("$0.$1=$2", is_master ? kMasterHostPrefix : kTServerHostPrefix,
+ is_master ? master_end_idx : tserver_end_idx,
+ new_addr.ToString()));
+ return JoinStrings(new_overrides, ",");
+ }
+
+ // Adds the appropriate flags for the given daemon to be restarted bound to
+ // the given address.
+ void SetUpDaemonForNewAddr(const Sockaddr& new_addr, const string& new_overrides_str,
+ ExternalDaemon* daemon) {
+ HostPort new_ip_hp(new_addr.host(), new_addr.port());
+ daemon->SetRpcBindAddress(new_ip_hp);
+ daemon->mutable_flags()->emplace_back("--rpc_reuseport=true");
+ // TODO(awong): more plumbing is needed to allow the server to startup with
+ // the webserver, so just disable it.
+ daemon->mutable_flags()->emplace_back("--webserver_enabled=false");
+ daemon->mutable_flags()->emplace_back(
+ Substitute("--dns_addr_resolution_override=$0", new_overrides_str));
+ }
+
+ // Sets the flags on all nodes in the cluster, except for the last node of
+ // the given 'node_type', which is expected to have been restarted with the
+ // appropriate flags.
+ void SetFlagsOnRemainingCluster(ExternalMiniCluster::DaemonType node_type,
+ const string& new_overrides_str) {
+ int master_end_idx = cluster_->num_masters();
+ int tserver_end_idx = cluster_->num_tablet_servers();
+ if (node_type == ExternalMiniCluster::DaemonType::MASTER) {
+ --master_end_idx;
+ } else {
+ --tserver_end_idx;
+ }
+ for (int i = 0; i < master_end_idx; i++) {
+ ASSERT_OK(cluster_->SetFlag(
+ cluster_->master(i), "dns_addr_resolution_override", new_overrides_str));
+ }
+ for (int i = 0; i < tserver_end_idx; i++) {
+ ASSERT_OK(cluster_->SetFlag(
+ cluster_->tablet_server(i), "dns_addr_resolution_override", new_overrides_str));
+ }
+ }
+
protected:
unique_ptr<ExternalMiniCluster> cluster_;
client::sp::shared_ptr<KuduClient> client_;
@@ -180,5 +253,110 @@ TEST_F(DnsAliasWithUnixSocketsITest, TestBasic) {
}
}
+// These tests depend on restarted servers being assigned a new IP address. On
+// MacOS, tservers are all assigned the same address, so don't run them there.
+#if defined(__linux__)
+
+// Regression test for KUDU-1620, wherein consensus proxies don't eventually
+// succeed when the address changes but the host/ports stays the same.
+TEST_F(DnsAliasITest, Kudu1620) {
+ TestWorkload w(cluster_.get());
+ w.set_num_replicas(3);
+ w.set_num_write_threads(1);
+ w.Setup();
+ w.Start();
+ while (w.rows_inserted() < 10) {
+ SleepFor(MonoDelta::FromMilliseconds(10));
+ }
+ w.StopAndJoin();
+
+ // Shut down a tablet server and start one up at a different IP.
+ auto* tserver = cluster_->tablet_server(cluster_->num_tablet_servers() - 1);
+ tserver->Shutdown();
+ unique_ptr<Socket> reserved_socket;
+ ASSERT_OK(cluster_->ReserveDaemonSocket(cluster::ExternalMiniCluster::DaemonType::TSERVER, 3,
+ kDefaultBindMode, &reserved_socket,
+ tserver->bound_rpc_hostport().port()));
+ Sockaddr new_addr;
+ ASSERT_OK(reserved_socket->GetSocketAddress(&new_addr));
+
+ // Once we start having the other servers communicate with the new tserver,
+ // ksck should return healthy.
+ auto new_overrides_str = GetNewOverridesFlag(ExternalMiniCluster::DaemonType::TSERVER, new_addr);
+ SetUpDaemonForNewAddr(new_addr, new_overrides_str, tserver);
+ ASSERT_OK(tserver->Restart());
+
+ // Running ksck should fail because the existing servers are still trying to
+ // communicate with the old port.
+ ClusterVerifier v(cluster_.get());
+ Status s = v.RunKsck();
+ ASSERT_TRUE(s.IsRuntimeError()) << s.ToString();
+
+ SetFlagsOnRemainingCluster(ExternalMiniCluster::DaemonType::TSERVER, new_overrides_str);
+
+ // Our test thread still thinks the old alias is still valid, so our ksck
+ // should fail.
+ s = v.RunKsck();
+ ASSERT_TRUE(s.IsRuntimeError()) << s.ToString();
+
+ // Once we set the DNS aliases in the test thread, ksck should succeed.
+ FLAGS_dns_addr_resolution_override = new_overrides_str;
+ ASSERT_EVENTUALLY([&] {
+ ASSERT_OK(v.RunKsck());
+ });
+}
+
+// Master-side regression test for KUDU-1620. Masters instantiate consensus
+// proxies to get the UUIDs of its peers. With KUDU-1620 resolved, the proxy
+// used should be able to re-resolve and retry upon failure, rather than
+// retrying at the same address.
+TEST_F(DnsAliasITest, TestMasterReresolveOnStartup) {
+ const int last_master_idx = cluster_->num_masters() - 1;
+ auto* master = cluster_->master(last_master_idx);
+
+ // Shut down and prepare the node that we're going to give a new address.
+ master->Shutdown();
+ unique_ptr<Socket> reserved_socket;
+ ASSERT_OK(cluster_->ReserveDaemonSocket(cluster::ExternalMiniCluster::DaemonType::MASTER, 3,
+ kDefaultBindMode, &reserved_socket,
+ master->bound_rpc_hostport().port()));
+ Sockaddr new_addr;
+ ASSERT_OK(reserved_socket->GetSocketAddress(&new_addr));
+ auto new_overrides_str = GetNewOverridesFlag(ExternalMiniCluster::DaemonType::MASTER, new_addr);
+ SetUpDaemonForNewAddr(new_addr, new_overrides_str, master);
+
+ // Shut down the other masters so we can test what happens when they come
+ // back up.
+ for (int i = 0; i < last_master_idx; i++) {
+ cluster_->master(i)->Shutdown();
+ }
+ for (int i = 0; i < last_master_idx; i++) {
+ ASSERT_OK(cluster_->master(i)->Restart());
+ }
+ // Since the rest of the cluster doesn't know about the address, ksck will
+ // fail.
+ ClusterVerifier v(cluster_.get());
+ Status s = v.RunKsck();
+ ASSERT_TRUE(s.IsRuntimeError()) << s.ToString();
+
+ // Even upon setting the DNS overrides on the rest of the nodes, since the
+ // master hasn't started, we should still see an error.
+ SetFlagsOnRemainingCluster(ExternalMiniCluster::DaemonType::MASTER, new_overrides_str);
+ s = v.RunKsck();
+ ASSERT_TRUE(s.IsRuntimeError()) << s.ToString();
+ FLAGS_dns_addr_resolution_override = new_overrides_str;
+ s = v.RunKsck();
+ ASSERT_TRUE(s.IsRuntimeError()) << s.ToString();
+
+ // Upon restarting the node, the other masters should be able to resolve and
+ // connect to it.
+ ASSERT_OK(master->Restart());
+ ASSERT_EVENTUALLY([&] {
+ ASSERT_OK(v.RunKsck());
+ });
+}
+
+#endif
+
} // namespace itest
} // namespace kudu
diff --git a/src/kudu/integration-tests/tablet_replacement-itest.cc b/src/kudu/integration-tests/tablet_replacement-itest.cc
index 65ca217..bc9b367 100644
--- a/src/kudu/integration-tests/tablet_replacement-itest.cc
+++ b/src/kudu/integration-tests/tablet_replacement-itest.cc
@@ -30,10 +30,10 @@
#include <glog/logging.h>
#include <gtest/gtest.h>
+#include "kudu/common/row_operations.pb.h"
#include "kudu/common/schema.h"
#include "kudu/common/wire_protocol-test-util.h"
#include "kudu/common/wire_protocol.h"
-#include "kudu/common/wire_protocol.pb.h"
#include "kudu/consensus/consensus.pb.h"
#include "kudu/consensus/metadata.pb.h"
#include "kudu/gutil/strings/substitute.h"
@@ -237,8 +237,10 @@ void TabletReplacementITest::TestDontEvictIfRemainingConfigIsUnstable(
consensus::ConsensusStatePB cstate;
ASSERT_OK(GetConsensusState(leader_ts, tablet_id, kTimeout, EXCLUDE_HEALTH_REPORT, &cstate));
SCOPED_TRACE(cstate.DebugString());
- ASSERT_FALSE(cstate.has_pending_config())
- << "Leader should not have issued any config change";
+ // It's possible the leader only registered one replica as failed when
+ // sending its report to the master, so the master may have requested a
+ // change config request to add non-voter. Regardless, there should be no
+ // new committed config since a majority is down.
ASSERT_EQ(cstate_initial.committed_config().opid_index(),
cstate.committed_config().opid_index())
<< "Leader should not have issued any config change";
diff --git a/src/kudu/mini-cluster/external_mini_cluster.h b/src/kudu/mini-cluster/external_mini_cluster.h
index 0a2e830..971727b 100644
--- a/src/kudu/mini-cluster/external_mini_cluster.h
+++ b/src/kudu/mini-cluster/external_mini_cluster.h
@@ -669,6 +669,12 @@ class ExternalDaemon : public RefCountedThreadSafe<ExternalDaemon> {
// Return the options used to create the daemon.
ExternalDaemonOptions opts() const { return opts_; }
+ void SetRpcBindAddress(HostPort rpc_hostport) {
+ DCHECK(!IsProcessAlive());
+ bound_rpc_ = std::move(rpc_hostport);
+ opts_.rpc_bind_address = bound_rpc_;
+ }
+
protected:
friend class RefCountedThreadSafe<ExternalDaemon>;
virtual ~ExternalDaemon();
diff --git a/src/kudu/mini-cluster/mini_cluster.cc b/src/kudu/mini-cluster/mini_cluster.cc
index 3a48cda..7da352b 100644
--- a/src/kudu/mini-cluster/mini_cluster.cc
+++ b/src/kudu/mini-cluster/mini_cluster.cc
@@ -60,10 +60,11 @@ string MiniCluster::GetBindIpForDaemonWithType(DaemonType type,
Status MiniCluster::ReserveDaemonSocket(DaemonType type,
int index,
BindMode bind_mode,
- unique_ptr<Socket>* socket) {
+ unique_ptr<Socket>* socket,
+ uint16_t port) {
string ip = GetBindIpForDaemonWithType(type, index, bind_mode);
Sockaddr sock_addr;
- RETURN_NOT_OK(sock_addr.ParseString(ip, 0));
+ RETURN_NOT_OK(sock_addr.ParseString(ip, port));
unique_ptr<Socket> sock(new Socket());
RETURN_NOT_OK(sock->Init(sock_addr.family(), 0));
diff --git a/src/kudu/mini-cluster/mini_cluster.h b/src/kudu/mini-cluster/mini_cluster.h
index 945a915..dd8e964 100644
--- a/src/kudu/mini-cluster/mini_cluster.h
+++ b/src/kudu/mini-cluster/mini_cluster.h
@@ -16,6 +16,7 @@
// under the License.
#pragma once
+#include <cstdint>
#include <memory>
#include <string>
#include <vector>
@@ -140,7 +141,8 @@ class MiniCluster {
static Status ReserveDaemonSocket(DaemonType type,
int index,
BindMode bind_mode,
- std::unique_ptr<Socket>* socket);
+ std::unique_ptr<Socket>* socket,
+ uint16_t port = 0);
protected:
// Return the IP address that the daemon with the given type and index will
diff --git a/src/kudu/rpc/proxy.cc b/src/kudu/rpc/proxy.cc
index fbda5de..9501a38 100644
--- a/src/kudu/rpc/proxy.cc
+++ b/src/kudu/rpc/proxy.cc
@@ -34,6 +34,7 @@
#include "kudu/rpc/rpc_controller.h"
#include "kudu/rpc/user_credentials.h"
#include "kudu/util/kernel_stack_watchdog.h"
+#include "kudu/util/logging.h"
#include "kudu/util/net/dns_resolver.h"
#include "kudu/util/net/net_util.h"
#include "kudu/util/net/sockaddr.h"
@@ -227,6 +228,9 @@ void Proxy::AsyncRequest(const string& method,
// TODO(awong): we should be more specific here -- consider having the RPC
// layer set a flag in the controller that warrants a retry.
if (PREDICT_FALSE(!controller->status().ok())) {
+ KLOG_EVERY_N_SECS(WARNING, 5)
+ << Substitute("Call had error, refreshing address and retrying: $0",
+ controller->status().ToString());
auto req_payload = controller->ReleaseRequestPayload();
controller->Reset();
RefreshDnsAndEnqueueRequest(method, std::move(req_payload), response, controller, callback);
diff --git a/src/kudu/util/net/net_util.cc b/src/kudu/util/net/net_util.cc
index f587041..54f577e 100644
--- a/src/kudu/util/net/net_util.cc
+++ b/src/kudu/util/net/net_util.cc
@@ -220,8 +220,12 @@ Status HostPort::ResolveAddresses(vector<Sockaddr>* addresses) const {
TRACE_EVENT1("net", "HostPort::ResolveAddresses",
"host", host_);
TRACE_COUNTER_SCOPE_LATENCY_US("dns_us");
- if (PREDICT_FALSE(!FLAGS_dns_addr_resolution_override.empty())) {
- vector<string> hosts_and_addrs = Split(FLAGS_dns_addr_resolution_override, ",");
+ // NOTE: we use this instead of the FLAGS_... variant because this flag may be
+ // changed at runtime in tests and thus needs to be thread-safe.
+ const auto dns_addr_resolution_override_flag =
+ google::GetCommandLineFlagInfoOrDie("dns_addr_resolution_override");
+ if (PREDICT_FALSE(!dns_addr_resolution_override_flag.current_value.empty())) {
+ vector<string> hosts_and_addrs = Split(dns_addr_resolution_override_flag.current_value, ",");
for (const auto& ha : hosts_and_addrs) {
vector<string> host_and_addr = Split(ha, "=");
if (host_and_addr.size() != 2) {
[kudu] 01/03: [rpc] allow reuse of outbound request buffers when
retrying
Posted by al...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
alexey pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/kudu.git
commit 4fe61a26ad5bbb0a41bb7be002a6b32d9c0ea916
Author: Andrew Wong <aw...@cloudera.com>
AuthorDate: Tue Sep 28 16:00:30 2021 -0700
[rpc] allow reuse of outbound request buffers when retrying
This patch fixes a heap-use-after-free bug in the address-re-resolving
proxy caused by the fact that requests may be stack allocated, are
serialized once when making the OutboundCall, and then thrown away once
calling the (typically) user-provided callback.
This patch splits out the state that may be used by retries (i.e. the
serialized request, header, and sidecars) into its own class, and passes
this around when retrying.
There are some methods that don't appear to be used in this codebase,
but I opted to keep existing behavior because they seem to be used in
other codebases that share library code. Specifically, I don't see usage
of RpcController::AddOutboundSidecar(), but this does seem to be used in
Impala[1].
1. https://github.com/apache/impala/blob/b28da054f3595bb92873433211438306fc22fbc7/be/src/rpc/sidecar-util.h#L48
Change-Id: I118f3559c4647bdd996617443bd371a041711295
Reviewed-on: http://gerrit.cloudera.org:8080/17892
Tested-by: Andrew Wong <aw...@cloudera.com>
Reviewed-by: Alexey Serbin <as...@cloudera.com>
---
src/kudu/rpc/mt-rpc-test.cc | 6 +--
src/kudu/rpc/outbound_call.cc | 83 ++++++++++++++++++++++++++----------
src/kudu/rpc/outbound_call.h | 95 ++++++++++++++++++++++++++++++++----------
src/kudu/rpc/proxy-test.cc | 48 +++++++++++++++++++++
src/kudu/rpc/proxy.cc | 54 ++++++++++++++++--------
src/kudu/rpc/proxy.h | 8 ++--
src/kudu/rpc/rpc-test-base.h | 29 ++++++-------
src/kudu/rpc/rpc_controller.cc | 14 +++++++
src/kudu/rpc/rpc_controller.h | 17 ++++++++
9 files changed, 272 insertions(+), 82 deletions(-)
diff --git a/src/kudu/rpc/mt-rpc-test.cc b/src/kudu/rpc/mt-rpc-test.cc
index 3496e26..773eb39 100644
--- a/src/kudu/rpc/mt-rpc-test.cc
+++ b/src/kudu/rpc/mt-rpc-test.cc
@@ -64,7 +64,7 @@ namespace rpc {
class MultiThreadedRpcTest : public RpcTestBase {
public:
// Make a single RPC call.
- void SingleCall(Sockaddr server_addr, const char* method_name,
+ void SingleCall(Sockaddr server_addr, const string& method_name,
Status* result, CountDownLatch* latch) {
LOG(INFO) << "Connecting to " << server_addr.ToString();
shared_ptr<Messenger> client_messenger;
@@ -76,7 +76,7 @@ class MultiThreadedRpcTest : public RpcTestBase {
}
// Make RPC calls until we see a failure.
- void HammerServer(Sockaddr server_addr, const char* method_name,
+ void HammerServer(Sockaddr server_addr, const string& method_name,
Status* last_result) {
shared_ptr<Messenger> client_messenger;
CHECK_OK(CreateMessenger("ClientHS", &client_messenger));
@@ -84,7 +84,7 @@ class MultiThreadedRpcTest : public RpcTestBase {
}
void HammerServerWithMessenger(
- Sockaddr server_addr, const char* method_name, Status* last_result,
+ Sockaddr server_addr, const string& method_name, Status* last_result,
const shared_ptr<Messenger>& messenger) {
LOG(INFO) << "Connecting to " << server_addr.ToString();
Proxy p(messenger, server_addr, server_addr.host(),
diff --git a/src/kudu/rpc/outbound_call.cc b/src/kudu/rpc/outbound_call.cc
index e2f1fd3..151195f 100644
--- a/src/kudu/rpc/outbound_call.cc
+++ b/src/kudu/rpc/outbound_call.cc
@@ -81,21 +81,24 @@ static const double kMicrosPerSecond = 1000000.0;
OutboundCall::OutboundCall(const ConnectionId& conn_id,
const RemoteMethod& remote_method,
+ unique_ptr<RequestPayload> payload,
+ CallbackBehavior cb_behavior,
google::protobuf::Message* response_storage,
RpcController* controller,
ResponseCallback callback)
- : state_(READY),
+ : cb_behavior_(cb_behavior),
+ state_(READY),
remote_method_(remote_method),
conn_id_(conn_id),
callback_(std::move(callback)),
controller_(DCHECK_NOTNULL(controller)),
response_(DCHECK_NOTNULL(response_storage)),
+ payload_(std::move(payload)),
cancellation_requested_(false) {
- DVLOG(4) << "OutboundCall " << this << " constructed with state_: " << StateName(state_)
- << " and RPC timeout: "
- << (controller->timeout().Initialized() ? controller->timeout().ToString() : "none");
- header_.set_call_id(kInvalidCallId);
- remote_method.ToPB(header_.mutable_remote_method());
+ DVLOG(4) << Substitute("OutboundCall $0 constructed with the state_: $1 and RPC timeout: $2",
+ this, StateName(state_),
+ (controller->timeout().Initialized() ? controller->timeout().ToString() : "none"));
+ payload_->header_.set_call_id(kInvalidCallId);
start_time_ = MonoTime::Now();
if (!controller_->required_server_features().empty()) {
@@ -103,41 +106,66 @@ OutboundCall::OutboundCall(const ConnectionId& conn_id,
}
if (controller_->request_id_) {
- header_.set_allocated_request_id(controller_->request_id_.release());
+ payload_->header_.set_allocated_request_id(controller_->request_id_.release());
}
}
+
+OutboundCall::OutboundCall(const ConnectionId& conn_id,
+ const RemoteMethod& remote_method,
+ google::protobuf::Message* response_storage,
+ RpcController* controller,
+ ResponseCallback callback)
+ : OutboundCall(conn_id, remote_method, std::make_unique<RequestPayload>(remote_method),
+ CallbackBehavior::kFreeSidecars, response_storage, controller,
+ std::move(callback)) {
+}
+
OutboundCall::~OutboundCall() {
DCHECK(IsFinished());
DVLOG(4) << "OutboundCall " << this << " destroyed with state_: " << StateName(state_);
}
void OutboundCall::SerializeTo(TransferPayload* slices) {
- DCHECK_LT(0, request_buf_.size())
+ DCHECK_LT(0, payload_->request_buf_.size())
<< "Must call SetRequestPayload() before SerializeTo()";
- const MonoDelta &timeout = controller_->timeout();
- if (timeout.Initialized()) {
- header_.set_timeout_millis(timeout.ToMilliseconds());
+ if (controller_->timeout().Initialized()) {
+ payload_->header_.set_timeout_millis(controller_->timeout().ToMilliseconds());
}
for (uint32_t feature : controller_->required_server_features()) {
- header_.add_required_feature_flags(feature);
+ payload_->header_.add_required_feature_flags(feature);
}
- DCHECK_LE(0, sidecar_byte_size_);
+ DCHECK_LE(0, payload_->sidecar_byte_size_);
serialization::SerializeHeader(
- header_, sidecar_byte_size_ + request_buf_.size(), &header_buf_);
+ payload_->header_, payload_->sidecar_byte_size_ + payload_->request_buf_.size(),
+ &payload_->header_buf_);
slices->clear();
- slices->push_back(header_buf_);
- slices->push_back(request_buf_);
- for (auto& sidecar : sidecars_) {
+ slices->push_back(payload_->header_buf_);
+ slices->push_back(payload_->request_buf_);
+ for (auto& sidecar : payload_->sidecars_) {
sidecar->AppendSlices(slices);
}
}
-void OutboundCall::SetRequestPayload(const Message& req,
+RequestPayload::RequestPayload(const RemoteMethod& remote_method) {
+ header_.set_call_id(kInvalidCallId);
+ remote_method.ToPB(header_.mutable_remote_method());
+}
+
+unique_ptr<RequestPayload> RequestPayload::CreateRequestPayload(
+ const RemoteMethod& remote_method,
+ const Message& req,
+ vector<unique_ptr<RpcSidecar>>&& sidecars) {
+ auto payload = std::make_unique<RequestPayload>(remote_method);
+ payload->PopulateRequestPayload(req, std::move(sidecars));
+ return payload;
+}
+
+void RequestPayload::PopulateRequestPayload(const Message& req,
vector<unique_ptr<RpcSidecar>>&& sidecars) {
DCHECK_EQ(-1, sidecar_byte_size_);
@@ -159,6 +187,11 @@ void OutboundCall::SetRequestPayload(const Message& req,
serialization::SerializeMessage(req, &request_buf_, sidecar_byte_size_, true);
}
+void OutboundCall::SetRequestPayload(const Message& req,
+ vector<unique_ptr<RpcSidecar>>&& sidecars) {
+ DCHECK_NOTNULL(payload_)->PopulateRequestPayload(req, std::move(sidecars));
+}
+
Status OutboundCall::status() const {
std::lock_guard<simple_spinlock> l(lock_);
return status_;
@@ -266,7 +299,9 @@ void OutboundCall::Cancel() {
void OutboundCall::CallCallback() {
// Clear references to outbound sidecars before invoking callback.
- sidecars_.clear();
+ if (cb_behavior_ == CallbackBehavior::kFreeSidecars) {
+ FreeSidecars();
+ }
int64_t start_cycles = CycleClock::Now();
{
@@ -317,6 +352,10 @@ void OutboundCall::SetResponse(unique_ptr<CallResponse> resp) {
}
}
+unique_ptr<RequestPayload> OutboundCall::ReleaseRequestPayload() {
+ return std::move(payload_);
+}
+
void OutboundCall::SetQueued() {
set_state(ON_OUTBOUND_QUEUE);
}
@@ -333,9 +372,9 @@ void OutboundCall::SetSent() {
// behavior is a lot more efficient if memory is freed from the same thread
// which allocated it -- this lets it keep to thread-local operations instead
// of taking a mutex to put memory back on the global freelist.
- delete [] header_buf_.release();
+ delete [] payload_->header_buf_.release();
- // request_buf_ is also done being used here, but since it was allocated by
+ // payload_ is also done being used here, but since it was allocated by
// the caller thread, we would rather let that thread free it whenever it
// deletes the RpcController.
@@ -452,7 +491,7 @@ string OutboundCall::ToString() const {
void OutboundCall::DumpPB(const DumpConnectionsRequestPB& req,
RpcCallInProgressPB* resp) {
std::lock_guard<simple_spinlock> l(lock_);
- resp->mutable_header()->CopyFrom(header_);
+ resp->mutable_header()->CopyFrom(payload_->header_);
resp->set_micros_elapsed((MonoTime::Now() - start_time_).ToMicroseconds());
switch (state_) {
diff --git a/src/kudu/rpc/outbound_call.h b/src/kudu/rpc/outbound_call.h
index 74db5ca..620a28b 100644
--- a/src/kudu/rpc/outbound_call.h
+++ b/src/kudu/rpc/outbound_call.h
@@ -57,6 +57,43 @@ class DumpConnectionsRequestPB;
class RpcCallInProgressPB;
class RpcController;
+// Encapsulates the request payload being sent by a call.
+class RequestPayload {
+ public:
+ // Creates a payload for the given remote method, serializing the given
+ // request, taking ownership of the sidecars, and populating the header as
+ // necessary.
+ static std::unique_ptr<RequestPayload> CreateRequestPayload(
+ const RemoteMethod& remote_method,
+ const google::protobuf::Message& req,
+ std::vector<std::unique_ptr<RpcSidecar>>&& sidecars);
+
+ // Creates an "empty" payload for the given remote method. Callers should
+ // also call PopulateRequestPayload() to form a usable payload.
+ explicit RequestPayload(const RemoteMethod& remote_method);
+
+ // Serializes the given 'req' and takes ownership of 'sidecars', populating
+ // the header as necessary.
+ void PopulateRequestPayload(const google::protobuf::Message& req,
+ std::vector<std::unique_ptr<RpcSidecar>>&& sidecars);
+ private:
+ friend class OutboundCall;
+
+ // The RPC header.
+ // Parts of this (eg the call ID) are only assigned once this request has
+ // been passed to the reactor thread and assigned a connection. Calls should
+ // re-assign the call ID if this payload is used in multiple calls (e.g.
+ // retries after re-resolving the address).
+ RequestHeader header_;
+ faststring header_buf_;
+ faststring request_buf_;
+ std::vector<std::unique_ptr<RpcSidecar>> sidecars_;
+
+ // Total size in bytes of all sidecars in 'sidecars_'. Set in SetRequestPayload().
+ // This cannot exceed TransferLimits::kMaxTotalSidecarBytes.
+ int32_t sidecar_byte_size_ = -1;
+};
+
// Tracks the status of a call on the client side.
//
// This is an internal-facing class -- clients interact with the
@@ -80,6 +117,20 @@ class OutboundCall {
REMOTE_CALL,
};
+ // Behavior when running the callback with regards to freeing resources. Some
+ // callers may expect the call itself free sidecars upon completion, while
+ // others may attempt to reuse the sidecars in another call attempt upon
+ // failure.
+ enum class CallbackBehavior {
+ kFreeSidecars,
+ kDontFreeSidecars,
+ };
+
+ OutboundCall(const ConnectionId& conn_id, const RemoteMethod& remote_method,
+ std::unique_ptr<RequestPayload> req_payload, CallbackBehavior cb_behavior,
+ google::protobuf::Message* response_storage,
+ RpcController* controller, ResponseCallback callback);
+
OutboundCall(const ConnectionId& conn_id, const RemoteMethod& remote_method,
google::protobuf::Message* response_storage,
RpcController* controller, ResponseCallback callback);
@@ -97,8 +148,8 @@ class OutboundCall {
// Assign the call ID for this call. This is called from the reactor
// thread once a connection has been assigned. Must only be called once.
void set_call_id(int32_t call_id) {
- DCHECK_EQ(header_.call_id(), kInvalidCallId) << "Already has a call ID";
- header_.set_call_id(call_id);
+ DCHECK_EQ(payload_->header_.call_id(), kInvalidCallId) << "Already has a call ID";
+ payload_->header_.set_call_id(call_id);
}
// Serialize the call for the wire. Requires that SetRequestPayload()
@@ -148,6 +199,12 @@ class OutboundCall {
return required_rpc_features_;
}
+ std::unique_ptr<RequestPayload> ReleaseRequestPayload();
+
+ void FreeSidecars() {
+ DCHECK_NOTNULL(payload_)->sidecars_.clear();
+ }
+
std::string ToString() const;
void DumpPB(const DumpConnectionsRequestPB& req, RpcCallInProgressPB* resp);
@@ -164,12 +221,12 @@ class OutboundCall {
// Return true if a call ID has been assigned to this call.
bool call_id_assigned() const {
- return header_.call_id() != kInvalidCallId;
+ return payload_->header_.call_id() != kInvalidCallId;
}
int32_t call_id() const {
DCHECK(call_id_assigned());
- return header_.call_id();
+ return payload_->header_.call_id();
}
// Returns true if cancellation has been requested. Must be called from
@@ -227,6 +284,16 @@ class OutboundCall {
// This will only be non-NULL if status().IsRemoteError().
const ErrorStatusPB* error_pb() const;
+ // Call the user-provided callback. Note that entries in 'sidecars_' are cleared
+ // prior to invoking the callback so the client can assume that the call doesn't
+ // hold references to outbound sidecars.
+ void CallCallback();
+
+ // The behavior defining whether to free sidecars upon calling the callback.
+ // Certain callbacks may perfer freeing the sidecars manually from within the
+ // callback.
+ const CallbackBehavior cb_behavior_;
+
// Lock for state_ status_, error_pb_ fields, since they
// may be mutated by the reactor thread while the client thread
// reads them.
@@ -235,16 +302,6 @@ class OutboundCall {
Status status_;
std::unique_ptr<ErrorStatusPB> error_pb_;
- // Call the user-provided callback. Note that entries in 'sidecars_' are cleared
- // prior to invoking the callback so the client can assume that the call doesn't
- // hold references to outbound sidecars.
- void CallCallback();
-
- // The RPC header.
- // Parts of this (eg the call ID) are only assigned once this call has been
- // passed to the reactor thread and assigned a connection.
- RequestHeader header_;
-
// The remote method being called.
RemoteMethod remote_method_;
@@ -259,20 +316,12 @@ class OutboundCall {
google::protobuf::Message* response_;
// Buffers for storing segments of the wire-format request.
- faststring header_buf_;
- faststring request_buf_;
+ std::unique_ptr<RequestPayload> payload_;
// Once a response has been received for this call, contains that response.
// Otherwise NULL.
std::unique_ptr<CallResponse> call_response_;
- // All sidecars to be sent with this call.
- std::vector<std::unique_ptr<RpcSidecar>> sidecars_;
-
- // Total size in bytes of all sidecars in 'sidecars_'. Set in SetRequestPayload().
- // This cannot exceed TransferLimits::kMaxTotalSidecarBytes.
- int32_t sidecar_byte_size_ = -1;
-
// True if cancellation was requested on this call.
bool cancellation_requested_;
diff --git a/src/kudu/rpc/proxy-test.cc b/src/kudu/rpc/proxy-test.cc
index 34b626c..c5ad52b 100644
--- a/src/kudu/rpc/proxy-test.cc
+++ b/src/kudu/rpc/proxy-test.cc
@@ -18,6 +18,7 @@
#include "kudu/rpc/proxy.h"
#include <cstdint>
+#include <functional>
#include <memory>
#include <string>
#include <thread>
@@ -29,6 +30,7 @@
#include "kudu/gutil/ref_counted.h"
#include "kudu/gutil/strings/substitute.h"
+#include "kudu/rpc/response_callback.h"
#include "kudu/rpc/rpc-test-base.h"
#include "kudu/rpc/rpc_controller.h"
#include "kudu/rpc/rtest.pb.h"
@@ -37,6 +39,7 @@
#include "kudu/util/net/dns_resolver.h"
#include "kudu/util/net/net_util.h"
#include "kudu/util/net/sockaddr.h"
+#include "kudu/util/notification.h"
#include "kudu/util/status.h"
#include "kudu/util/test_macros.h"
#include "kudu/util/test_util.h"
@@ -60,6 +63,13 @@ constexpr uint16_t kPort = 1111;
constexpr const char* kFakeHost = "fakehost";
const HostPort kFakeHostPort(kFakeHost, kPort);
+void SendRequestAsync(const ResponseCallback& cb,
+ Proxy* p, RpcController* controller, SleepResponsePB* resp) {
+ SleepRequestPB req;
+ req.set_sleep_micros(100 * 1000); // 100ms
+ p->AsyncRequest(GenericCalculatorService::kSleepMethodName, req, resp, controller, cb);
+}
+
Status SendRequest(Proxy* p) {
SleepRequestPB req;
req.set_sleep_micros(100 * 1000); // 100ms
@@ -74,6 +84,44 @@ Status SendRequest(Proxy* p) {
class RpcProxyTest : public RpcTestBase {
};
+TEST_F(RpcProxyTest, TestProxyRetriesWhenRequestLeavesScope) {
+ DnsResolver dns_resolver(1, 1024 * 1024);
+ shared_ptr<Messenger> client_messenger;
+ ASSERT_OK(CreateMessenger("client_messenger", &client_messenger));
+
+ // We're providing a fake hostport to encourage retries of DNS resolution,
+ // which will attempt to send the request payload after the request protobuf
+ // has already left scope.
+ Proxy p(client_messenger, kFakeHostPort, &dns_resolver,
+ CalculatorService::static_service_name());
+ p.Init();
+
+ SleepResponsePB resp;
+ {
+ Notification note;
+ RpcController controller;
+ controller.set_timeout(MonoDelta::FromMilliseconds(10000));
+ SendRequestAsync([¬e] () { note.Notify(); }, &p, &controller, &resp);
+ note.WaitForNotification();
+ Status s = controller.status();
+ ASSERT_TRUE(s.IsNetworkError()) << s.ToString();
+ }
+
+ // Now try a successful call.
+ string ip = GetBindIpForDaemon(/*index*/2, kDefaultBindMode);
+ Sockaddr addr;
+ ASSERT_OK(addr.ParseString(ip, kPort));
+ ASSERT_OK(StartTestServerWithGeneratedCode(&addr));
+ FLAGS_dns_addr_resolution_override = Substitute("$0=$1", kFakeHost, addr.ToString());
+
+ Notification note;
+ RpcController controller;
+ controller.set_timeout(MonoDelta::FromMilliseconds(10000));
+ SendRequestAsync([¬e] () { note.Notify(); }, &p, &controller, &resp);
+ note.WaitForNotification();
+ ASSERT_OK(controller.status());
+}
+
// Test that proxies initialized with a DnsResolver return errors when
// receiving a non-transient error.
TEST_F(RpcProxyTest, TestProxyReturnsOnNonTransientError) {
diff --git a/src/kudu/rpc/proxy.cc b/src/kudu/rpc/proxy.cc
index 8d810b8..fbda5de 100644
--- a/src/kudu/rpc/proxy.cc
+++ b/src/kudu/rpc/proxy.cc
@@ -33,6 +33,7 @@
#include "kudu/rpc/response_callback.h"
#include "kudu/rpc/rpc_controller.h"
#include "kudu/rpc/user_credentials.h"
+#include "kudu/util/kernel_stack_watchdog.h"
#include "kudu/util/net/dns_resolver.h"
#include "kudu/util/net/net_util.h"
#include "kudu/util/net/sockaddr.h"
@@ -132,16 +133,16 @@ Proxy::~Proxy() {
}
void Proxy::EnqueueRequest(const string& method,
- const google::protobuf::Message& req,
+ unique_ptr<RequestPayload> req_payload,
google::protobuf::Message* response,
RpcController* controller,
- const ResponseCallback& callback) const {
+ const ResponseCallback& callback,
+ OutboundCall::CallbackBehavior cb_behavior) const {
ConnectionId connection = conn_id();
DCHECK(connection.remote().is_initialized());
- RemoteMethod remote_method(service_name_, method);
controller->call_.reset(
- new OutboundCall(connection, remote_method, response, controller, callback));
- controller->SetRequestParam(req);
+ new OutboundCall(connection, {service_name_, method}, std::move(req_payload),
+ cb_behavior, response, controller, callback));
controller->SetMessenger(messenger_.get());
// If this fails to queue, the callback will get called immediately
@@ -150,14 +151,16 @@ void Proxy::EnqueueRequest(const string& method,
}
void Proxy::RefreshDnsAndEnqueueRequest(const std::string& method,
- const google::protobuf::Message& req,
+ unique_ptr<RequestPayload> req_payload,
google::protobuf::Message* response,
RpcController* controller,
const ResponseCallback& callback) {
DCHECK(!controller->call_);
vector<Sockaddr>* addrs = new vector<Sockaddr>();
DCHECK_NOTNULL(dns_resolver_)->RefreshAddressesAsync(hp_, addrs,
- [this, &req, &method, callback, response, controller, addrs] (const Status& s) {
+ [this, req_raw = req_payload.release(),
+ &method, callback, response, controller, addrs] (const Status& s) mutable {
+ unique_ptr<RequestPayload> req_payload(req_raw);
unique_ptr<vector<Sockaddr>> unique_addrs(addrs);
// If we fail to resolve the address, treat the call as failed.
if (!s.ok() || addrs->empty()) {
@@ -178,7 +181,10 @@ void Proxy::RefreshDnsAndEnqueueRequest(const std::string& method,
std::lock_guard<simple_spinlock> l(lock_);
conn_id_.set_remote(*addr);
}
- EnqueueRequest(method, req, response, controller, callback);
+ // NOTE: we don't expect the user-provided callback to free sidecars, so
+ // make sure the outbound call frees it for us.
+ EnqueueRequest(method, std::move(req_payload), response, controller, callback,
+ OutboundCall::CallbackBehavior::kFreeSidecars);
});
}
@@ -189,8 +195,16 @@ void Proxy::AsyncRequest(const string& method,
const ResponseCallback& callback) {
CHECK(!controller->call_) << "Controller should be reset";
base::subtle::NoBarrier_Store(&is_started_, true);
+ // TODO(awong): it would be great if we didn't have to heap allocate the
+ // payload.
+ auto req_payload = RequestPayload::CreateRequestPayload(
+ RemoteMethod{service_name_, method},
+ req, controller->ReleaseOutboundSidecars());
if (!dns_resolver_) {
- EnqueueRequest(method, req, response, controller, callback);
+ // NOTE: we don't expect the user-provided callback to free sidecars, so
+ // make sure the outbound call frees it for us.
+ EnqueueRequest(method, std::move(req_payload), response, controller, callback,
+ OutboundCall::CallbackBehavior::kFreeSidecars);
return;
}
@@ -202,26 +216,32 @@ void Proxy::AsyncRequest(const string& method,
remote_initialized = conn_id_.remote().is_initialized();
}
if (!remote_initialized) {
- RefreshDnsAndEnqueueRequest(method, req, response, controller, callback);
+ RefreshDnsAndEnqueueRequest(method, std::move(req_payload), response, controller, callback);
return;
}
- // Otherwise, just enqueue the request, but retry if there's a network error,
- // since it's possible the physical address of the host was changed. We only
- // retry once more before calling the callback.
- auto refresh_dns_and_cb = [this, &req, &method,
- callback, response, controller] () {
+ // Otherwise, just enqueue the request, but retry if there's an error, since
+ // it's possible the physical address of the host was changed. We only retry
+ // once more before calling the callback.
+ auto refresh_dns_and_cb = [this, &method, callback, response, controller] () {
// TODO(awong): we should be more specific here -- consider having the RPC
// layer set a flag in the controller that warrants a retry.
if (PREDICT_FALSE(!controller->status().ok())) {
+ auto req_payload = controller->ReleaseRequestPayload();
controller->Reset();
- RefreshDnsAndEnqueueRequest(method, req, response, controller, callback);
+ RefreshDnsAndEnqueueRequest(method, std::move(req_payload), response, controller, callback);
return;
}
// For any other status, OK or otherwise, just run the callback.
+ controller->FreeOutboundSidecars();
+ SCOPED_WATCH_STACK(100);
callback();
};
- EnqueueRequest(method, req, response, controller, refresh_dns_and_cb);
+ // Since we may end up using the request payload in the event of a retry,
+ // ensure the outbound call doesn't free the sidecars, and instead free
+ // manually from within our callback.
+ EnqueueRequest(method, std::move(req_payload), response, controller, refresh_dns_and_cb,
+ OutboundCall::CallbackBehavior::kDontFreeSidecars);
}
diff --git a/src/kudu/rpc/proxy.h b/src/kudu/rpc/proxy.h
index b0bc512..cb82205 100644
--- a/src/kudu/rpc/proxy.h
+++ b/src/kudu/rpc/proxy.h
@@ -24,6 +24,7 @@
#include "kudu/gutil/atomicops.h"
#include "kudu/gutil/macros.h"
#include "kudu/rpc/connection_id.h"
+#include "kudu/rpc/outbound_call.h"
#include "kudu/rpc/response_callback.h"
#include "kudu/util/locks.h"
#include "kudu/util/net/net_util.h"
@@ -148,7 +149,7 @@ class Proxy {
// Asynchronously refreshes the DNS, enqueueing the given request upon
// success, or failing the call and calling the callback upon failure.
void RefreshDnsAndEnqueueRequest(const std::string& method,
- const google::protobuf::Message& req,
+ std::unique_ptr<RequestPayload> req_payload,
google::protobuf::Message* response,
RpcController* controller,
const ResponseCallback& callback);
@@ -156,10 +157,11 @@ class Proxy {
// Queues the given request as an outbound call using the given messenger,
// controller, and response.
void EnqueueRequest(const std::string& method,
- const google::protobuf::Message& req,
+ std::unique_ptr<RequestPayload> req_payload,
google::protobuf::Message* response,
RpcController* controller,
- const ResponseCallback& callback) const;
+ const ResponseCallback& callback,
+ OutboundCall::CallbackBehavior cb_behavior) const;
// Returns a single Sockaddr from the 'addrs', logging a warning if there is
// more than one to choose from.
diff --git a/src/kudu/rpc/rpc-test-base.h b/src/kudu/rpc/rpc-test-base.h
index c32423e..0aa46ce 100644
--- a/src/kudu/rpc/rpc-test-base.h
+++ b/src/kudu/rpc/rpc-test-base.h
@@ -91,12 +91,12 @@ namespace rpc {
class GenericCalculatorService : public ServiceIf {
public:
static const std::string kFullServiceName;
- static const char *kAddMethodName;
- static const char *kSleepMethodName;
- static const char *kSleepWithSidecarMethodName;
- static const char *kPushStringsMethodName;
- static const char *kSendTwoStringsMethodName;
- static const char *kAddExactlyOnce;
+ static const std::string kAddMethodName;
+ static const std::string kSleepMethodName;
+ static const std::string kSleepWithSidecarMethodName;
+ static const std::string kPushStringsMethodName;
+ static const std::string kSendTwoStringsMethodName;
+ static const std::string kAddExactlyOnce;
static const char* kFirstString;
static const char* kSecondString;
@@ -400,13 +400,14 @@ class CalculatorService : public CalculatorServiceIf {
};
-const std::string GenericCalculatorService::kFullServiceName = "kudu.rpc.GenericCalculatorService";
-const char *GenericCalculatorService::kAddMethodName = "Add";
-const char *GenericCalculatorService::kSleepMethodName = "Sleep";
-const char *GenericCalculatorService::kSleepWithSidecarMethodName = "SleepWithSidecar";
-const char *GenericCalculatorService::kPushStringsMethodName = "PushStrings";
-const char *GenericCalculatorService::kSendTwoStringsMethodName = "SendTwoStrings";
-const char *GenericCalculatorService::kAddExactlyOnce = "AddExactlyOnce";
+const std::string GenericCalculatorService::kFullServiceName =
+ "kudu.rpc.GenericCalculatorService";
+const std::string GenericCalculatorService::kAddMethodName = "Add";
+const std::string GenericCalculatorService::kSleepMethodName = "Sleep";
+const std::string GenericCalculatorService::kSleepWithSidecarMethodName = "SleepWithSidecar";
+const std::string GenericCalculatorService::kPushStringsMethodName = "PushStrings";
+const std::string GenericCalculatorService::kSendTwoStringsMethodName = "SendTwoStrings";
+const std::string GenericCalculatorService::kAddExactlyOnce = "AddExactlyOnce";
const char *GenericCalculatorService::kFirstString =
"1111111111111111111111111111111111111111111111111111111111";
@@ -468,7 +469,7 @@ class RpcTestBase : public KuduTest {
return bld.Build(messenger);
}
- static Status DoTestSyncCall(Proxy* p, const char *method,
+ static Status DoTestSyncCall(Proxy* p, const std::string& method,
CredentialsPolicy policy = CredentialsPolicy::ANY_CREDENTIALS) {
AddRequestPB req;
req.set_x(rand());
diff --git a/src/kudu/rpc/rpc_controller.cc b/src/kudu/rpc/rpc_controller.cc
index d0e07dd..6730e7f 100644
--- a/src/kudu/rpc/rpc_controller.cc
+++ b/src/kudu/rpc/rpc_controller.cc
@@ -168,6 +168,20 @@ void RpcController::SetRequestParam(const google::protobuf::Message& req) {
call_->SetRequestPayload(req, std::move(outbound_sidecars_));
}
+void RpcController::FreeOutboundSidecars() {
+ outbound_sidecars_total_bytes_ = 0;
+ call_->FreeSidecars();
+}
+
+std::vector<unique_ptr<RpcSidecar>> RpcController::ReleaseOutboundSidecars() {
+ outbound_sidecars_total_bytes_ = 0;
+ return std::move(outbound_sidecars_);
+}
+
+unique_ptr<RequestPayload> RpcController::ReleaseRequestPayload() {
+ return DCHECK_NOTNULL(call_)->ReleaseRequestPayload();
+}
+
void RpcController::Cancel() {
DCHECK(call_);
DCHECK(messenger_);
diff --git a/src/kudu/rpc/rpc_controller.h b/src/kudu/rpc/rpc_controller.h
index a8c0af5..49bd527 100644
--- a/src/kudu/rpc/rpc_controller.h
+++ b/src/kudu/rpc/rpc_controller.h
@@ -42,6 +42,7 @@ class ErrorStatusPB;
class Messenger;
class OutboundCall;
class RequestIdPB;
+class RequestPayload;
class RpcSidecar;
// Authentication credentials policy for outbound RPCs. Some RPC methods
@@ -141,6 +142,21 @@ class RpcController {
// in some cases even when sent to multiple servers, enabling exactly once semantics.
void SetRequestIdPB(std::unique_ptr<RequestIdPB> request_id);
+ // Releases the outbound sidecars added to this controller. This is useful if
+ // callers want to create a request payload for a request.
+ std::vector<std::unique_ptr<RpcSidecar>> ReleaseOutboundSidecars();
+
+ // Frees the outbound sidecars added to this controller. OutboundCalls may
+ // call this before running the user-provided callback to ensure the state is
+ // freed before the callback is run. However, certain usages of OutboundCall
+ // warrants delaying the freeing until during the callback.
+ void FreeOutboundSidecars();
+
+ // Releases the request payload owned by this controller. This is useful if
+ // callers want to reuse a request payload in another attempt of a call, as
+ // it allows callers to transfer ownership to a new outbound call.
+ std::unique_ptr<RequestPayload> ReleaseRequestPayload();
+
// Returns whether a request id has been set on RPC header.
bool has_request_id() const;
@@ -273,6 +289,7 @@ class RpcController {
// Once the call is sent, it is tracked here.
std::shared_ptr<OutboundCall> call_;
+ // Owned by the controller until released and taken by a call.
std::vector<std::unique_ptr<RpcSidecar>> outbound_sidecars_;
// Total size of sidecars in outbound_sidecars_. This is limited to a maximum
[kudu] 03/03: KUDU-2671 introduce PartitionKey
Posted by al...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
alexey pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/kudu.git
commit a50091e2d4509feac2f29128107102ec52fcb7b0
Author: Alexey Serbin <al...@apache.org>
AuthorDate: Tue Sep 28 18:18:10 2021 -0700
KUDU-2671 introduce PartitionKey
This changelist introduces a dedicated PartitionKey data structure to
represent a table's partition key. The crux is in keeping the hash and
the range part of the key separately, so it's possible to properly work
with such an entity without the context of a particular range or tablet.
Before this patch, encoded partition keys were represented as strings that
had the hash and the range parts concatenated: it was always possible to
properly decode such a key having only table-wide information in the
context because every range had the same hash schema. However, once
custom hash schemas per range were introduced, now it's impossible to
properly decode such a compound key without knowing the particular hash
schema that was used to encode the key.
The ordering of PartitionKeys in this changelist follows the legacy
notation for comparison operator, first concatenating the hash and the
range parts and then comparing the result strings. However, it's going
to change in a follow-up changelist. That change induces corresponding
changes in PartitionPruner. Also, the updated comparison operator for
PartitionKey requires updating the code in the catalog manager and in
the client's meta-cache. I decided to split the changes into a few
changelists for easier tracking and reviewing.
As a by-product of this change, the following methods of the
PartitionSchema class started working as needed in case of a table with
per-range custom hash schemas:
* PartitionKeyDebugStringImpl()
* PartitionKeyDebugString()
I was also thinking about introducing an extra field to ScanTokenPB,
like the newly added GetTableLocationsRequestPB::partition_key_range
field, or rely on the ScanTokenPB::tablet_metadata field to extract
the hash and the range parts out of the strings representing lower and
upper scan boundaries, but I realized that deserves to be done in a
separate changelist.
Change-Id: I00255ec404beeb999117f5265de0d5d8deaf0d68
Reviewed-on: http://gerrit.cloudera.org:8080/17890
Tested-by: Kudu Jenkins
Reviewed-by: Andrew Wong <aw...@cloudera.com>
---
src/kudu/client/batcher.cc | 2 +
src/kudu/client/client-test.cc | 54 +-
src/kudu/client/client.cc | 27 +-
src/kudu/client/client.h | 1 +
src/kudu/client/flex_partitioning_client-test.cc | 2 +-
src/kudu/client/meta_cache.cc | 98 ++--
src/kudu/client/meta_cache.h | 35 +-
src/kudu/client/partitioner-internal.cc | 16 +-
src/kudu/client/partitioner-internal.h | 5 +-
src/kudu/client/scan_configuration.cc | 12 +-
src/kudu/client/scan_configuration.h | 5 +-
src/kudu/client/scan_token-internal.cc | 27 +-
src/kudu/client/scanner-internal.cc | 12 +-
src/kudu/client/scanner-internal.h | 3 +-
src/kudu/common/common.proto | 2 +-
src/kudu/common/partition-test.cc | 558 +++++++++++----------
src/kudu/common/partition.cc | 255 +++++-----
src/kudu/common/partition.h | 118 ++++-
src/kudu/common/partition_pruner-test.cc | 148 +++---
src/kudu/common/partition_pruner.cc | 37 +-
src/kudu/common/partition_pruner.h | 12 +-
src/kudu/common/scan_spec.cc | 12 +-
src/kudu/common/scan_spec.h | 20 +-
.../integration-tests/table_locations-itest.cc | 43 +-
src/kudu/master/catalog_manager-test.cc | 70 ++-
src/kudu/master/catalog_manager.cc | 149 ++++--
src/kudu/master/catalog_manager.h | 16 +-
src/kudu/master/master.proto | 27 +-
src/kudu/master/master_path_handlers.cc | 4 +-
src/kudu/tools/table_scanner.cc | 4 +-
src/kudu/tools/tool_action_table.cc | 4 +-
31 files changed, 1079 insertions(+), 699 deletions(-)
diff --git a/src/kudu/client/batcher.cc b/src/kudu/client/batcher.cc
index a50602d..2db5033 100644
--- a/src/kudu/client/batcher.cc
+++ b/src/kudu/client/batcher.cc
@@ -316,6 +316,8 @@ WriteRpc::WriteRpc(const scoped_refptr<Batcher>& batcher,
RowOperationsPBEncoder enc(requested);
for (InFlightOp* op : ops_) {
#ifndef NDEBUG
+ // Run the same verification that is about to be run by the tablet server
+ // the data is sent to.
const Partition& partition = op->tablet->partition();
const PartitionSchema& partition_schema = table()->partition_schema();
const KuduPartialRow& row = op->write_op->row();
diff --git a/src/kudu/client/client-test.cc b/src/kudu/client/client-test.cc
index 91d6063..9965f17 100644
--- a/src/kudu/client/client-test.cc
+++ b/src/kudu/client/client-test.cc
@@ -131,6 +131,10 @@
#include "kudu/util/test_util.h"
#include "kudu/util/thread_restrictions.h"
+namespace kudu {
+class PartitionKey;
+} // namespace kudu
+
DECLARE_bool(allow_unsafe_replication_factor);
DECLARE_bool(catalog_manager_support_live_row_count);
DECLARE_bool(catalog_manager_support_on_disk_size);
@@ -277,8 +281,8 @@ class ClientTest : public KuduTest {
}
// Looks up the remote tablet entry for a given partition key in the meta cache.
- scoped_refptr<internal::RemoteTablet> MetaCacheLookup(KuduTable* table,
- const string& partition_key) {
+ scoped_refptr<internal::RemoteTablet> MetaCacheLookup(
+ KuduTable* table, const PartitionKey& partition_key) {
scoped_refptr<internal::RemoteTablet> rt;
Synchronizer sync;
client_->data_->meta_cache_->LookupTabletByKey(
@@ -2198,22 +2202,22 @@ TEST_F(ClientTest, TestMetaCacheExpiry) {
// Clear the cache.
meta_cache->ClearCache();
internal::MetaCacheEntry entry;
- ASSERT_FALSE(meta_cache->LookupEntryByKeyFastPath(client_table_.get(), "", &entry));
+ ASSERT_FALSE(meta_cache->LookupEntryByKeyFastPath(client_table_.get(), {}, &entry));
// Prime the cache.
- CHECK_NOTNULL(MetaCacheLookup(client_table_.get(), "").get());
- ASSERT_TRUE(meta_cache->LookupEntryByKeyFastPath(client_table_.get(), "", &entry));
+ CHECK_NOTNULL(MetaCacheLookup(client_table_.get(), {}).get());
+ ASSERT_TRUE(meta_cache->LookupEntryByKeyFastPath(client_table_.get(), {}, &entry));
ASSERT_FALSE(entry.stale());
// Sleep in order to expire the cache.
SleepFor(MonoDelta::FromMilliseconds(FLAGS_table_locations_ttl_ms));
ASSERT_TRUE(entry.stale());
- ASSERT_FALSE(meta_cache->LookupEntryByKeyFastPath(client_table_.get(), "", &entry));
+ ASSERT_FALSE(meta_cache->LookupEntryByKeyFastPath(client_table_.get(), {}, &entry));
// Force a lookup and ensure the entry is refreshed.
- ASSERT_NE(nullptr, MetaCacheLookup(client_table_.get(), ""));
+ ASSERT_NE(nullptr, MetaCacheLookup(client_table_.get(), {}));
ASSERT_TRUE(entry.stale());
- ASSERT_TRUE(meta_cache->LookupEntryByKeyFastPath(client_table_.get(), "", &entry));
+ ASSERT_TRUE(meta_cache->LookupEntryByKeyFastPath(client_table_.get(), {}, &entry));
ASSERT_FALSE(entry.stale());
}
@@ -2316,11 +2320,11 @@ TEST_F(ClientTest, TestMetaCacheWithKeysAndIds) {
internal::MetaCacheEntry entry;
ASSERT_FALSE(meta_cache->LookupEntryByIdFastPath(first_tablet_id, &entry));
ASSERT_FALSE(entry.Initialized());
- ASSERT_FALSE(meta_cache->LookupEntryByKeyFastPath(client_table_.get(), "", &entry));
+ ASSERT_FALSE(meta_cache->LookupEntryByKeyFastPath(client_table_.get(), {}, &entry));
ASSERT_FALSE(entry.Initialized());
- ASSERT_NE(nullptr, MetaCacheLookup(client_table_.get(), ""));
- ASSERT_TRUE(meta_cache->LookupEntryByKeyFastPath(client_table_.get(), "", &entry));
+ ASSERT_NE(nullptr, MetaCacheLookup(client_table_.get(), {}));
+ ASSERT_TRUE(meta_cache->LookupEntryByKeyFastPath(client_table_.get(), {}, &entry));
ASSERT_FALSE(entry.stale());
}
// Just because we have a cache entry for key-based lookup doesn't mean we
@@ -2347,7 +2351,7 @@ TEST_F(ClientTest, TestMetaCacheWithKeysAndIds) {
// still be available.
{
internal::MetaCacheEntry entry;
- ASSERT_TRUE(meta_cache->LookupEntryByKeyFastPath(client_table_.get(), "", &entry));
+ ASSERT_TRUE(meta_cache->LookupEntryByKeyFastPath(client_table_.get(), {}, &entry));
ASSERT_FALSE(entry.stale());
}
// Let's do that again but with an id-based lookup first.
@@ -2356,7 +2360,7 @@ TEST_F(ClientTest, TestMetaCacheWithKeysAndIds) {
internal::MetaCacheEntry entry;
ASSERT_FALSE(meta_cache->LookupEntryByIdFastPath(first_tablet_id, &entry));
ASSERT_FALSE(entry.Initialized());
- ASSERT_FALSE(meta_cache->LookupEntryByKeyFastPath(client_table_.get(), "", &entry));
+ ASSERT_FALSE(meta_cache->LookupEntryByKeyFastPath(client_table_.get(), {}, &entry));
ASSERT_FALSE(entry.Initialized());
}
// Once we do the lookup by ID, we should be able to fast-path the id-based
@@ -2368,14 +2372,14 @@ TEST_F(ClientTest, TestMetaCacheWithKeysAndIds) {
ASSERT_NE(nullptr, rt);
ASSERT_TRUE(meta_cache->LookupEntryByIdFastPath(first_tablet_id, &entry));
ASSERT_FALSE(entry.stale());
- ASSERT_FALSE(meta_cache->LookupEntryByKeyFastPath(client_table_.get(), "", &entry));
+ ASSERT_FALSE(meta_cache->LookupEntryByKeyFastPath(client_table_.get(), {}, &entry));
}
// And once we do the key-based lookups, we should be able to see them both
// cached.
{
internal::MetaCacheEntry entry;
- ASSERT_NE(nullptr, MetaCacheLookup(client_table_.get(), ""));
- ASSERT_TRUE(meta_cache->LookupEntryByKeyFastPath(client_table_.get(), "", &entry));
+ ASSERT_NE(nullptr, MetaCacheLookup(client_table_.get(), {}));
+ ASSERT_TRUE(meta_cache->LookupEntryByKeyFastPath(client_table_.get(), {}, &entry));
ASSERT_FALSE(entry.stale());
}
{
@@ -2396,10 +2400,10 @@ TEST_F(ClientTest, TestMetaCacheExpiryWithKeysAndIds) {
FLAGS_client_tablet_locations_by_id_ttl_ms = 25;
internal::MetaCacheEntry entry;
ASSERT_FALSE(meta_cache->LookupEntryByIdFastPath(first_tablet_id, &entry));
- ASSERT_FALSE(meta_cache->LookupEntryByKeyFastPath(client_table_.get(), "", &entry));
+ ASSERT_FALSE(meta_cache->LookupEntryByKeyFastPath(client_table_.get(), {}, &entry));
// Perform both id-based and key-based lookups.
- ASSERT_NE(nullptr, MetaCacheLookup(client_table_.get(), ""));
+ ASSERT_NE(nullptr, MetaCacheLookup(client_table_.get(), {}));
scoped_refptr<internal::RemoteTablet> rt;
ASSERT_OK(MetaCacheLookupById(first_tablet_id, &rt));
ASSERT_NE(nullptr, rt);
@@ -2408,20 +2412,20 @@ TEST_F(ClientTest, TestMetaCacheExpiryWithKeysAndIds) {
// key-based entries.
SleepFor(MonoDelta::FromMilliseconds(FLAGS_client_tablet_locations_by_id_ttl_ms));
ASSERT_FALSE(meta_cache->LookupEntryByIdFastPath(first_tablet_id, &entry));
- ASSERT_TRUE(meta_cache->LookupEntryByKeyFastPath(client_table_.get(), "", &entry));
+ ASSERT_TRUE(meta_cache->LookupEntryByKeyFastPath(client_table_.get(), {}, &entry));
ASSERT_FALSE(entry.stale());
FLAGS_client_tablet_locations_by_id_ttl_ms = 10000;
FLAGS_table_locations_ttl_ms = 25;
meta_cache->ClearCache();
- ASSERT_NE(nullptr, MetaCacheLookup(client_table_.get(), ""));
+ ASSERT_NE(nullptr, MetaCacheLookup(client_table_.get(), {}));
ASSERT_OK(MetaCacheLookupById(first_tablet_id, &rt));
ASSERT_NE(nullptr, rt);
// Wait for our key-based entries to expire. This shouldn't affect our
// id-based entries.
SleepFor(MonoDelta::FromMilliseconds(FLAGS_table_locations_ttl_ms));
- ASSERT_FALSE(meta_cache->LookupEntryByKeyFastPath(client_table_.get(), "", &entry));
+ ASSERT_FALSE(meta_cache->LookupEntryByKeyFastPath(client_table_.get(), {}, &entry));
ASSERT_TRUE(meta_cache->LookupEntryByIdFastPath(first_tablet_id, &entry));
ASSERT_FALSE(entry.stale());
}
@@ -2478,7 +2482,7 @@ TEST_F(ClientTest, TestGetTabletServerBlacklist) {
// We have to loop since some replicas may have been created slowly.
scoped_refptr<internal::RemoteTablet> rt;
while (true) {
- rt = MetaCacheLookup(table.get(), "");
+ rt = MetaCacheLookup(table.get(), {});
ASSERT_TRUE(rt.get() != nullptr);
vector<internal::RemoteTabletServer*> tservers;
rt->GetRemoteTabletServers(&tservers);
@@ -2555,7 +2559,7 @@ TEST_F(ClientTest, TestGetTabletServerDeterministic) {
// We have to loop since some replicas may have been created slowly.
scoped_refptr<internal::RemoteTablet> rt;
while (true) {
- rt = MetaCacheLookup(table.get(), "");
+ rt = MetaCacheLookup(table.get(), {});
ASSERT_TRUE(rt.get() != nullptr);
vector<internal::RemoteTabletServer*> tservers;
rt->GetRemoteTabletServers(&tservers);
@@ -4918,7 +4922,7 @@ TEST_F(ClientTest, TestReplicatedMultiTabletTableFailover) {
NO_FATALS(InsertTestRows(table.get(), kNumRowsToWrite));
// Find the leader of the first tablet.
- scoped_refptr<internal::RemoteTablet> rt = MetaCacheLookup(table.get(), "");
+ scoped_refptr<internal::RemoteTablet> rt = MetaCacheLookup(table.get(), {});
internal::RemoteTabletServer *rts = rt->LeaderTServer();
// Kill the leader of the first tablet.
@@ -4968,7 +4972,7 @@ TEST_F(ClientTest, TestReplicatedTabletWritesWithLeaderElection) {
NO_FATALS(InsertTestRows(table.get(), kNumRowsToWrite));
// Find the leader replica
- scoped_refptr<internal::RemoteTablet> rt = MetaCacheLookup(table.get(), "");
+ scoped_refptr<internal::RemoteTablet> rt = MetaCacheLookup(table.get(), {});
internal::RemoteTabletServer *rts;
set<string> blacklist;
vector<internal::RemoteTabletServer*> candidates;
diff --git a/src/kudu/client/client.cc b/src/kudu/client/client.cc
index 7e3d141..19ed0e9 100644
--- a/src/kudu/client/client.cc
+++ b/src/kudu/client/client.cc
@@ -1314,7 +1314,7 @@ Status KuduTable::ListPartitions(vector<Partition>* partitions) {
while (pruner.HasMorePartitionKeyRanges()) {
scoped_refptr<client::internal::RemoteTablet> tablet;
Synchronizer sync;
- const string& partition_key = pruner.NextPartitionKey();
+ const auto& partition_key = pruner.NextPartitionKey();
client->data_->meta_cache_->LookupTabletByKey(
this,
partition_key,
@@ -1330,7 +1330,7 @@ Status KuduTable::ListPartitions(vector<Partition>* partitions) {
RETURN_NOT_OK(s);
partitions->emplace_back(tablet->partition());
- pruner.RemovePartitionKeyRange(tablet->partition().partition_key_end());
+ pruner.RemovePartitionKeyRange(tablet->partition().end());
}
return Status::OK();
}
@@ -1798,11 +1798,30 @@ Status KuduScanner::AddExclusiveUpperBoundRaw(const Slice& key) {
}
Status KuduScanner::AddLowerBoundPartitionKeyRaw(const Slice& partition_key) {
- return data_->mutable_configuration()->AddLowerBoundPartitionKeyRaw(partition_key);
+ // TODO(aserbin): use move semantics to pass PartitionKey arguments
+ if (const auto& table = GetKuduTable();
+ table->data_->partition_schema_.HasCustomHashSchemas()) {
+ return Status::InvalidArgument(Substitute(
+ "$0: cannot use AddLowerBoundPartitionKeyRaw() because "
+ "the table has custom per-range hash schemas", table->name()));
+ }
+ const auto& hash_schema = GetKuduTable()->partition_schema().hash_schema();
+ auto pkey = Partition::StringToPartitionKey(partition_key.ToString(),
+ hash_schema.size());
+ return data_->mutable_configuration()->AddLowerBoundPartitionKeyRaw(pkey);
}
Status KuduScanner::AddExclusiveUpperBoundPartitionKeyRaw(const Slice& partition_key) {
- return data_->mutable_configuration()->AddUpperBoundPartitionKeyRaw(partition_key);
+ if (const auto& table = GetKuduTable();
+ table->data_->partition_schema_.HasCustomHashSchemas()) {
+ return Status::InvalidArgument(Substitute(
+ "$0: cannot use AddExclusiveUpperBoundPartitionKeyRaw() because "
+ "the table has custom per-range hash schemas", table->name()));
+ }
+ const auto& hash_schema = GetKuduTable()->partition_schema().hash_schema();
+ auto pkey = Partition::StringToPartitionKey(partition_key.ToString(),
+ hash_schema.size());
+ return data_->mutable_configuration()->AddUpperBoundPartitionKeyRaw(pkey);
}
Status KuduScanner::SetCacheBlocks(bool cache_blocks) {
diff --git a/src/kudu/client/client.h b/src/kudu/client/client.h
index ea7638b..6d72f98 100644
--- a/src/kudu/client/client.h
+++ b/src/kudu/client/client.h
@@ -1761,6 +1761,7 @@ class KUDU_EXPORT KuduTable : public sp::enable_shared_from_this<KuduTable> {
friend class KuduClient;
friend class KuduPartitioner;
friend class KuduScanToken;
+ friend class KuduScanner;
KuduTable(const sp::shared_ptr<KuduClient>& client,
const std::string& name,
diff --git a/src/kudu/client/flex_partitioning_client-test.cc b/src/kudu/client/flex_partitioning_client-test.cc
index f7ce162..ed38612 100644
--- a/src/kudu/client/flex_partitioning_client-test.cc
+++ b/src/kudu/client/flex_partitioning_client-test.cc
@@ -486,7 +486,7 @@ TEST_F(FlexPartitioningCreateTableTest, NoUpperBoundRangeCustomHashSchema) {
// --------------------------------
// 0-111 x:{key} x:{key} x:{key}
// 111-222 x:{key} x:{key} -
- // >222 - - -
+ // >=222 - - -
constexpr const char* const kTableName = "NoUpperBoundRangeCustomHashSchema";
unique_ptr<KuduTableCreator> table_creator(client_->NewTableCreator());
diff --git a/src/kudu/client/meta_cache.cc b/src/kudu/client/meta_cache.cc
index 9f16beb..e8f2aa4 100644
--- a/src/kudu/client/meta_cache.cc
+++ b/src/kudu/client/meta_cache.cc
@@ -382,11 +382,19 @@ string RemoteTablet::ReplicasAsStringUnlocked() const {
return replicas_str;
}
-bool MetaCacheEntry::Contains(const string& partition_key) const {
+bool MetaCacheEntry::Contains(const PartitionKey& partition_key) const {
DCHECK(Initialized());
- return lower_bound_partition_key() <= partition_key &&
- (upper_bound_partition_key().empty() ||
- upper_bound_partition_key() > partition_key);
+ const auto& entry_hash_key = !lower_bound_partition_key().empty()
+ ? lower_bound_partition_key().hash_key()
+ : upper_bound_partition_key().hash_key();
+ // The hash part of the key has 'the exact' semantics: they should match
+ // for the partition key in question and the meta-cache entry.
+ if (partition_key.hash_key() != entry_hash_key) {
+ return false;
+ }
+ return lower_bound_partition_key().range_key() <= partition_key.range_key() &&
+ (upper_bound_partition_key().empty() ||
+ partition_key.range_key() < upper_bound_partition_key().range_key());
}
bool MetaCacheEntry::stale() const {
@@ -397,13 +405,16 @@ bool MetaCacheEntry::stale() const {
string MetaCacheEntry::DebugString(const KuduTable* table) const {
DCHECK(Initialized());
- const string& lower_bound = lower_bound_partition_key();
- const string& upper_bound = upper_bound_partition_key();
+ const auto& lower_bound = lower_bound_partition_key();
+ const auto& upper_bound = upper_bound_partition_key();
string lower_bound_string = MetaCache::DebugLowerBoundPartitionKey(table, lower_bound);
- string upper_bound_string = upper_bound.empty() ? "<end>" :
- table->partition_schema().PartitionKeyDebugString(upper_bound, *table->schema().schema_);
+ // The upper bound is exclusive, so it's necessary to get proper hash schema
+ // for the key.
+ string upper_bound_string = upper_bound.empty()
+ ? "<end>"
+ : table->partition_schema().PartitionKeyDebugString(upper_bound, *table->schema().schema_);
MonoDelta ttl = expiration_time_ - MonoTime::Now();
@@ -511,7 +522,7 @@ void MetaCacheServerPicker::PickLeader(const ServerPickedCallback& callback,
if (table_) {
meta_cache_->LookupTabletByKey(
table_,
- tablet_->partition().partition_key_start(),
+ tablet_->partition().begin(),
deadline,
MetaCache::LookupType::kPoint,
/*remote_tablet*/nullptr,
@@ -743,7 +754,7 @@ class LookupRpc
LookupRpc(scoped_refptr<MetaCache> meta_cache,
StatusCallback user_cb,
const KuduTable* table,
- string partition_key,
+ PartitionKey partition_key,
scoped_refptr<RemoteTablet>* remote_tablet,
const MonoTime& deadline,
MetaCache::LookupType lookup_type,
@@ -771,7 +782,7 @@ class LookupRpc
const GetTableLocationsResponsePB& resp() const { return resp_; }
const string& table_name() const { return table_->name(); }
const string& table_id() const { return table_->id(); }
- const string& partition_key() const { return partition_key_; }
+ const PartitionKey& partition_key() const { return partition_key_; }
bool is_exact_lookup() const {
return lookup_type_ == MetaCache::LookupType::kPoint;
}
@@ -815,7 +826,7 @@ class LookupRpc
const KuduTable* table_;
// Encoded partition key to lookup.
- string partition_key_;
+ PartitionKey partition_key_;
// When lookup finishes successfully, the selected tablet is written here
// prior to invoking the user-provided callback.
@@ -835,7 +846,7 @@ class LookupRpc
LookupRpc::LookupRpc(scoped_refptr<MetaCache> meta_cache,
StatusCallback user_cb, const KuduTable* table,
- string partition_key,
+ PartitionKey partition_key,
scoped_refptr<RemoteTablet>* remote_tablet,
const MonoTime& deadline,
MetaCache::LookupType lookup_type,
@@ -885,10 +896,22 @@ void LookupRpc::SendRpcSlowPath() {
return;
}
- // The end partition key is left unset intentionally so that we'll prefetch
- // some additional tablets.
+ // The end partition key is left unset intentionally to prefetch information
+ // on additional tablets.
req_.mutable_table()->set_table_id(table_->id());
- req_.set_partition_key_start(partition_key_);
+
+ // The range information is set using both the legacy and the contemporary
+ // notation, i.e. using 'partition_key_start' and 'partition_key_range' fields
+ // correspondingly. That's to allow a client of newer versions to work with
+ // servers of prior versions when sending request for a table without custom
+ // hash schemas per range, while also being able to query about table
+ // locations if a table has custom hash schemas per range when it served
+ // by a tablet server of newer versions.
+ auto* key_start = req_.mutable_key_start();
+ key_start->set_hash_key(partition_key_.hash_key());
+ key_start->set_range_key(partition_key_.range_key());
+ req_.set_partition_key_start(partition_key_.ToString());
+
req_.set_max_returned_locations(locations_to_fetch());
req_.set_intern_ts_infos_in_response(true);
if (replica_visibility_ == ReplicaController::Visibility::ALL) {
@@ -987,8 +1010,8 @@ Status MetaCache::ProcessGetTabletLocationsResponse(const string& tablet_id,
scoped_refptr<RemoteTablet> remote = FindPtrOrNull(tablets_by_id_, tablet_id);
if (remote) {
// Partition should not have changed.
- DCHECK_EQ(tablet_lower_bound, remote->partition().partition_key_start());
- DCHECK_EQ(tablet_upper_bound, remote->partition().partition_key_end());
+ DCHECK_EQ(tablet_lower_bound, remote->partition().begin().ToString());
+ DCHECK_EQ(tablet_upper_bound, remote->partition().end().ToString());
VLOG(3) << "Refreshing tablet " << tablet_id << ": " << SecureShortDebugString(tablet);
RETURN_NOT_OK_PREPEND(remote->Refresh(ts_cache_, tablet, ts_infos),
@@ -1017,7 +1040,7 @@ Status MetaCache::ProcessGetTabletLocationsResponse(const string& tablet_id,
}
Status MetaCache::ProcessGetTableLocationsResponse(const KuduTable* table,
- const string& partition_key,
+ const PartitionKey& partition_key,
bool is_exact_lookup,
const GetTableLocationsResponsePB& resp,
MetaCacheEntry* cache_entry,
@@ -1034,9 +1057,9 @@ Status MetaCache::ProcessGetTableLocationsResponse(const KuduTable* table,
if (tablet_locations.empty()) {
tablets_by_key.clear();
- MetaCacheEntry entry(expiration_time, "", "");
+ MetaCacheEntry entry(expiration_time, {}, {});
VLOG(3) << "Caching '" << table->name() << "' entry " << entry.DebugString(table);
- InsertOrDie(&tablets_by_key, "", entry);
+ InsertOrDie(&tablets_by_key, {}, entry);
} else {
// First, update the tserver cache, needed for the Refresh calls below.
// It's used for backward compatibility.
@@ -1067,24 +1090,30 @@ Status MetaCache::ProcessGetTableLocationsResponse(const KuduTable* table,
// non-covered range can only be inferred if the lookup partition key falls
// in A.
- const auto& first_lower_bound = tablet_locations.Get(0).partition().partition_key_start();
+ const auto& partition_pb = tablet_locations.Get(0).partition();
+ const PartitionKey first_lower_bound(
+ Partition::StringToPartitionKey(partition_pb.partition_key_start(),
+ partition_pb.hash_buckets_size()));
if (partition_key < first_lower_bound) {
// If the first tablet is past the requested partition key, then the
// partition key falls in an initial non-covered range, such as A.
// Clear any existing entries which overlap with the discovered non-covered range.
tablets_by_key.erase(tablets_by_key.begin(), tablets_by_key.lower_bound(first_lower_bound));
- MetaCacheEntry entry(expiration_time, "", first_lower_bound);
+ MetaCacheEntry entry(expiration_time, {}, first_lower_bound);
VLOG(3) << "Caching '" << table->name() << "' entry " << entry.DebugString(table);
- InsertOrDie(&tablets_by_key, "", entry);
+ InsertOrDie(&tablets_by_key, {}, entry);
}
// last_upper_bound tracks the upper bound of the previously processed
// entry, so that we can determine when we have found a non-covered range.
- string last_upper_bound = first_lower_bound;
+ auto last_upper_bound = first_lower_bound;
for (const TabletLocationsPB& tablet : tablet_locations) {
- const auto& tablet_lower_bound = tablet.partition().partition_key_start();
- const auto& tablet_upper_bound = tablet.partition().partition_key_end();
+ const auto& p = tablet.partition();
+ const PartitionKey tablet_lower_bound(Partition::StringToPartitionKey(
+ p.partition_key_start(), p.hash_buckets_size()));
+ const PartitionKey tablet_upper_bound(Partition::StringToPartitionKey(
+ p.partition_key_end(), p.hash_buckets_size()));
if (last_upper_bound < tablet_lower_bound) {
// There is a non-covered range between the previous tablet and this tablet.
@@ -1108,8 +1137,8 @@ Status MetaCache::ProcessGetTableLocationsResponse(const KuduTable* table,
scoped_refptr<RemoteTablet> remote = FindPtrOrNull(tablets_by_id_, tablet_id);
if (remote) {
// Partition should not have changed.
- DCHECK_EQ(tablet_lower_bound, remote->partition().partition_key_start());
- DCHECK_EQ(tablet_upper_bound, remote->partition().partition_key_end());
+ DCHECK_EQ(tablet_lower_bound, remote->partition().begin());
+ DCHECK_EQ(tablet_upper_bound, remote->partition().end());
VLOG(3) << Substitute("Refreshing tablet $0: $1",
tablet_id, SecureShortDebugString(tablet));
@@ -1161,7 +1190,7 @@ Status MetaCache::ProcessGetTableLocationsResponse(const KuduTable* table,
tablets_by_key.erase(tablets_by_key.lower_bound(last_upper_bound),
tablets_by_key.end());
- MetaCacheEntry entry(expiration_time, last_upper_bound, "");
+ MetaCacheEntry entry(expiration_time, last_upper_bound, {});
VLOG(3) << "Caching '" << table->name() << "' entry " << entry.DebugString(table);
InsertOrDie(&tablets_by_key, last_upper_bound, entry);
}
@@ -1178,7 +1207,7 @@ Status MetaCache::ProcessGetTableLocationsResponse(const KuduTable* table,
}
bool MetaCache::LookupEntryByKeyFastPath(const KuduTable* table,
- const string& partition_key,
+ const PartitionKey& partition_key,
MetaCacheEntry* entry) {
SCOPED_LOG_SLOW_EXECUTION(WARNING, 50, "looking up entry by key");
shared_lock<rw_spinlock> l(lock_.get_lock());
@@ -1208,7 +1237,7 @@ bool MetaCache::LookupEntryByKeyFastPath(const KuduTable* table,
}
Status MetaCache::DoFastPathLookup(const KuduTable* table,
- string* partition_key,
+ PartitionKey* partition_key,
MetaCache::LookupType lookup_type,
scoped_refptr<RemoteTablet>* remote_tablet) {
MetaCacheEntry entry;
@@ -1291,7 +1320,7 @@ void MetaCache::ClearCache() {
}
void MetaCache::LookupTabletByKey(const KuduTable* table,
- string partition_key,
+ PartitionKey partition_key,
const MonoTime& deadline,
MetaCache::LookupType lookup_type,
scoped_refptr<RemoteTablet>* remote_tablet,
@@ -1354,7 +1383,8 @@ void MetaCache::ReleaseMasterLookupPermit() {
master_lookup_sem_.Release();
}
-string MetaCache::DebugLowerBoundPartitionKey(const KuduTable* table, const string& partition_key) {
+string MetaCache::DebugLowerBoundPartitionKey(const KuduTable* table,
+ const PartitionKey& partition_key) {
return partition_key.empty() ? "<start>" :
table->partition_schema().PartitionKeyDebugString(partition_key, *table->schema().schema_);
}
diff --git a/src/kudu/client/meta_cache.h b/src/kudu/client/meta_cache.h
index 5dcc738..5eace9a 100644
--- a/src/kudu/client/meta_cache.h
+++ b/src/kudu/client/meta_cache.h
@@ -291,7 +291,6 @@ class RemoteTablet : public RefCountedThreadSafe<RemoteTablet> {
// instance, or a non-covered partition range.
class MetaCacheEntry {
public:
-
MetaCacheEntry() { }
// Construct a MetaCacheEntry representing a tablet.
@@ -303,8 +302,8 @@ class MetaCacheEntry {
// Construct a MetaCacheEntry representing a non-covered range with the
// provided range partition bounds.
MetaCacheEntry(MonoTime expiration_time,
- std::string lower_bound_partition_key,
- std::string upper_bound_partition_key)
+ PartitionKey lower_bound_partition_key,
+ PartitionKey upper_bound_partition_key)
: expiration_time_(expiration_time),
lower_bound_partition_key_(std::move(lower_bound_partition_key)),
upper_bound_partition_key_(std::move(upper_bound_partition_key)) {
@@ -325,23 +324,21 @@ class MetaCacheEntry {
}
// Returns the inclusive lower bound partition key for the entry.
- const std::string& lower_bound_partition_key() const {
+ const PartitionKey& lower_bound_partition_key() const {
DCHECK(Initialized());
if (is_non_covered_range()) {
return lower_bound_partition_key_;
- } else {
- return tablet_->partition().partition_key_start();
}
+ return tablet_->partition().begin();
}
// Returns the exclusive upper bound partition key for the entry.
- const std::string& upper_bound_partition_key() const {
+ const PartitionKey& upper_bound_partition_key() const {
DCHECK(Initialized());
if (is_non_covered_range()) {
return upper_bound_partition_key_;
- } else {
- return tablet_->partition().partition_key_end();
}
+ return tablet_->partition().end();
}
const MonoTime& expiration_time() const {
@@ -357,7 +354,7 @@ class MetaCacheEntry {
}
// Returns true if the partition key is contained in this meta cache entry.
- bool Contains(const std::string& partition_key) const;
+ bool Contains(const PartitionKey& partition_key) const;
// Returns true if this meta cache entry is stale.
bool stale() const;
@@ -382,10 +379,10 @@ class MetaCacheEntry {
scoped_refptr<RemoteTablet> tablet_;
// The lower bound partition key, if this is a non-covered range.
- std::string lower_bound_partition_key_;
+ PartitionKey lower_bound_partition_key_;
// The upper bound partition key, if this is a non-covered range.
- std::string upper_bound_partition_key_;
+ PartitionKey upper_bound_partition_key_;
};
// Manager of RemoteTablets and RemoteTabletServers. The client consults
@@ -419,7 +416,7 @@ class MetaCache : public RefCountedThreadSafe<MetaCache> {
// NOTE: the memory referenced by 'table' must remain valid until 'callback'
// is invoked.
void LookupTabletByKey(const KuduTable* table,
- std::string partition_key,
+ PartitionKey partition_key,
const MonoTime& deadline,
LookupType lookup_type,
scoped_refptr<RemoteTablet>* remote_tablet,
@@ -440,7 +437,7 @@ class MetaCache : public RefCountedThreadSafe<MetaCache> {
// Lookup the given tablet by key, only consulting local information.
// Returns true and sets *entry if successful.
bool LookupEntryByKeyFastPath(const KuduTable* table,
- const std::string& partition_key,
+ const PartitionKey& partition_key,
MetaCacheEntry* entry);
// Lookup the given tablet by tablet ID, only consulting local information.
// Returns true and sets *entry if successful.
@@ -450,7 +447,7 @@ class MetaCache : public RefCountedThreadSafe<MetaCache> {
// Process the response for the given key-based lookup parameters, indexing
// the location information as appropriate.
Status ProcessGetTableLocationsResponse(const KuduTable* table,
- const std::string& partition_key,
+ const PartitionKey& partition_key,
bool is_exact_lookup,
const master::GetTableLocationsResponsePB& resp,
MetaCacheEntry* cache_entry,
@@ -459,7 +456,7 @@ class MetaCache : public RefCountedThreadSafe<MetaCache> {
// Clears the non-covered range entries from a table's meta cache.
void ClearNonCoveredRangeEntries(const std::string& table_id);
- // Clears the meta cache.
+ // Clear meta-cache for the specified table identifier.
void ClearCache();
// Mark any replicas of any tablets hosted by 'ts' as failed. They will
@@ -475,7 +472,7 @@ class MetaCache : public RefCountedThreadSafe<MetaCache> {
// Return stringified representation of the given partition key, using "<start>" if empty.
static std::string DebugLowerBoundPartitionKey(const KuduTable* table,
- const std::string& partition_key);
+ const PartitionKey& partition_key);
private:
friend class LookupRpc;
@@ -504,7 +501,7 @@ class MetaCache : public RefCountedThreadSafe<MetaCache> {
// If 'lookup_type' is kLowerBound, then 'partition_key' will be updated to indicate the
// start of the range for the matched tablet.
Status DoFastPathLookup(const KuduTable* table,
- std::string* partition_key, // in-out parameter
+ PartitionKey* partition_key, // in-out parameter
LookupType lookup_type,
scoped_refptr<RemoteTablet>* remote_tablet);
@@ -553,7 +550,7 @@ class MetaCache : public RefCountedThreadSafe<MetaCache> {
// for key-based lookups.
//
// Protected by lock_.
- typedef std::map<std::string, MetaCacheEntry> TabletMap;
+ typedef std::map<PartitionKey, MetaCacheEntry> TabletMap;
std::unordered_map<std::string, TabletMap> tablets_by_table_and_key_;
// Cache entries for tablets, keyed by tablet ID, used for ID-based lookups.
diff --git a/src/kudu/client/partitioner-internal.cc b/src/kudu/client/partitioner-internal.cc
index 87788d9..34d73a5 100644
--- a/src/kudu/client/partitioner-internal.cc
+++ b/src/kudu/client/partitioner-internal.cc
@@ -46,8 +46,8 @@ Status KuduPartitionerBuilder::Data::Build(KuduPartitioner** partitioner) {
// Insert a sentinel for the beginning of the table, in case they
// query for any row which falls before the first partition.
- ret_data->partitions_by_start_key_[""] = -1;
- string next_part_key = "";
+ ret_data->partitions_by_start_key_[{}] = -1;
+ PartitionKey next_part_key;
int i = 0;
while (true) {
scoped_refptr<internal::RemoteTablet> tablet;
@@ -61,10 +61,12 @@ Status KuduPartitionerBuilder::Data::Build(KuduPartitioner** partitioner) {
break;
}
RETURN_NOT_OK(s);
- const auto& start_key = tablet->partition().partition_key_start();
- const auto& end_key = tablet->partition().partition_key_end();
+ const auto& start_key = tablet->partition().begin();
+ const auto& end_key = tablet->partition().end();
ret_data->partitions_by_start_key_[start_key] = i++;
- if (end_key.empty()) break;
+ if (end_key.empty()) {
+ break;
+ }
ret_data->partitions_by_start_key_[end_key] = -1;
next_part_key = end_key;
}
@@ -76,8 +78,8 @@ Status KuduPartitionerBuilder::Data::Build(KuduPartitioner** partitioner) {
Status KuduPartitioner::Data::PartitionRow(
const KuduPartialRow& row, int* partition) {
- tmp_buf_ = table_->data_->partition_schema_.EncodeKey(row);
- *partition = FindFloorOrDie(partitions_by_start_key_, tmp_buf_);
+ auto partition_key = table_->data_->partition_schema_.EncodeKey(row);
+ *partition = FindFloorOrDie(partitions_by_start_key_, partition_key);
return Status::OK();
}
diff --git a/src/kudu/client/partitioner-internal.h b/src/kudu/client/partitioner-internal.h
index 71f252e..d1aaaf7 100644
--- a/src/kudu/client/partitioner-internal.h
+++ b/src/kudu/client/partitioner-internal.h
@@ -18,11 +18,11 @@
#include <map>
#include <memory>
-#include <string>
#include <utility>
#include "kudu/client/client.h"
#include "kudu/client/shared_ptr.h" // IWYU pragma: keep
+#include "kudu/common/partition.h"
#include "kudu/util/monotime.h"
#include "kudu/util/status.h"
@@ -69,9 +69,8 @@ class KuduPartitioner::Data {
Status PartitionRow(const KuduPartialRow& row, int* partition);
sp::shared_ptr<KuduTable> table_;
- std::map<std::string, int> partitions_by_start_key_;
+ std::map<PartitionKey, int> partitions_by_start_key_;
int num_partitions_ = 0;
- std::string tmp_buf_;
};
diff --git a/src/kudu/client/scan_configuration.cc b/src/kudu/client/scan_configuration.cc
index 470b126..32bc784 100644
--- a/src/kudu/client/scan_configuration.cc
+++ b/src/kudu/client/scan_configuration.cc
@@ -33,6 +33,10 @@
#include "kudu/common/schema.h"
#include "kudu/gutil/strings/substitute.h"
+namespace kudu {
+class PartitionKey;
+} // namespace kudu
+
using std::unique_ptr;
using std::string;
using std::vector;
@@ -131,13 +135,13 @@ Status ScanConfiguration::AddUpperBoundRaw(const Slice& key) {
return Status::OK();
}
-Status ScanConfiguration::AddLowerBoundPartitionKeyRaw(const Slice& partition_key) {
- spec_.SetLowerBoundPartitionKey(partition_key);
+Status ScanConfiguration::AddLowerBoundPartitionKeyRaw(const PartitionKey& pkey) {
+ spec_.SetLowerBoundPartitionKey(pkey);
return Status::OK();
}
-Status ScanConfiguration::AddUpperBoundPartitionKeyRaw(const Slice& partition_key) {
- spec_.SetExclusiveUpperBoundPartitionKey(partition_key);
+Status ScanConfiguration::AddUpperBoundPartitionKeyRaw(const PartitionKey& pkey) {
+ spec_.SetExclusiveUpperBoundPartitionKey(pkey);
return Status::OK();
}
diff --git a/src/kudu/client/scan_configuration.h b/src/kudu/client/scan_configuration.h
index 39d34b6..1024ba7 100644
--- a/src/kudu/client/scan_configuration.h
+++ b/src/kudu/client/scan_configuration.h
@@ -40,6 +40,7 @@ namespace kudu {
class ColumnPredicate;
class KuduPartialRow;
+class PartitionKey;
namespace client {
@@ -68,9 +69,9 @@ class ScanConfiguration {
Status AddUpperBoundRaw(const Slice& key);
- Status AddLowerBoundPartitionKeyRaw(const Slice& partition_key);
+ Status AddLowerBoundPartitionKeyRaw(const PartitionKey& pkey);
- Status AddUpperBoundPartitionKeyRaw(const Slice& partition_key);
+ Status AddUpperBoundPartitionKeyRaw(const PartitionKey& pkey);
Status SetCacheBlocks(bool cache_blocks);
diff --git a/src/kudu/client/scan_token-internal.cc b/src/kudu/client/scan_token-internal.cc
index 117b70d..bb7540a 100644
--- a/src/kudu/client/scan_token-internal.cc
+++ b/src/kudu/client/scan_token-internal.cc
@@ -155,7 +155,7 @@ Status KuduScanToken::Data::PBIntoScanner(KuduClient* client,
Partition::FromPB(tablet_metadata.partition(), &partition);
MetaCacheEntry entry;
if (!client->data_->meta_cache_->LookupEntryByKeyFastPath(table.get(),
- partition.partition_key_start(), &entry)) {
+ partition.begin(), &entry)) {
// Generate a fake GetTableLocationsResponsePB to pass to the client
// meta cache in order to "inject" the tablet metadata into the client.
GetTableLocationsResponsePB mock_resp;
@@ -190,7 +190,7 @@ Status KuduScanToken::Data::PBIntoScanner(KuduClient* client,
}
client->data_->meta_cache_->ProcessGetTableLocationsResponse(
- table.get(), partition.partition_key_start(), true, mock_resp, &entry, 1);
+ table.get(), partition.begin(), true, mock_resp, &entry, 1);
}
}
@@ -439,28 +439,27 @@ Status KuduScanTokenBuilder::Data::Build(vector<KuduScanToken*>* tokens) {
while (pruner.HasMorePartitionKeyRanges()) {
scoped_refptr<internal::RemoteTablet> tablet;
Synchronizer sync;
- const string& partition_key = pruner.NextPartitionKey();
+ const auto& partition_key = pruner.NextPartitionKey();
client->data_->meta_cache_->LookupTabletByKey(table,
partition_key,
deadline,
MetaCache::LookupType::kLowerBound,
&tablet,
sync.AsStatusCallback());
- Status s = sync.Wait();
+ const auto s = sync.Wait();
if (s.IsNotFound()) {
// No more tablets in the table.
- pruner.RemovePartitionKeyRange("");
+ pruner.RemovePartitionKeyRange({});
continue;
- } else {
- RETURN_NOT_OK(s);
}
+ RETURN_NOT_OK(s);
// Check if the meta cache returned a tablet covering a partition key range past
// what we asked for. This can happen if the requested partition key falls
// in a non-covered range. In this case we can potentially prune the tablet.
- if (partition_key < tablet->partition().partition_key_start() &&
+ if (partition_key < tablet->partition().begin() &&
pruner.ShouldPrune(tablet->partition())) {
- pruner.RemovePartitionKeyRange(tablet->partition().partition_key_end());
+ pruner.RemovePartitionKeyRange(tablet->partition().end());
continue;
}
@@ -499,17 +498,15 @@ Status KuduScanTokenBuilder::Data::Build(vector<KuduScanToken*>* tokens) {
// Create the scan token itself.
ScanTokenPB message;
message.CopyFrom(pb);
- message.set_lower_bound_partition_key(
- tablet->partition().partition_key_start());
- message.set_upper_bound_partition_key(
- tablet->partition().partition_key_end());
+ message.set_lower_bound_partition_key(tablet->partition().begin().ToString());
+ message.set_upper_bound_partition_key(tablet->partition().end().ToString());
// Set the tablet metadata so that a call to the master is not needed to
// locate the tablet to scan when opening the scanner.
if (include_tablet_metadata_) {
internal::MetaCacheEntry entry;
if (client->data_->meta_cache_->LookupEntryByKeyFastPath(table,
- tablet->partition().partition_key_start(), &entry)) {
+ tablet->partition().begin(), &entry)) {
if (!entry.is_non_covered_range() && !entry.stale()) {
TabletMetadataPB tablet_pb;
tablet_pb.set_tablet_id(entry.tablet()->tablet_id());
@@ -561,7 +558,7 @@ Status KuduScanTokenBuilder::Data::Build(vector<KuduScanToken*>* tokens) {
std::move(message),
std::move(client_tablet));
tokens->push_back(client_scan_token.release());
- pruner.RemovePartitionKeyRange(tablet->partition().partition_key_end());
+ pruner.RemovePartitionKeyRange(tablet->partition().end());
}
return Status::OK();
}
diff --git a/src/kudu/client/scanner-internal.cc b/src/kudu/client/scanner-internal.cc
index 717fb32..1acb1cf 100644
--- a/src/kudu/client/scanner-internal.cc
+++ b/src/kudu/client/scanner-internal.cc
@@ -332,7 +332,7 @@ Status KuduScanner::Data::OpenNextTablet(const MonoTime& deadline,
Status KuduScanner::Data::ReopenCurrentTablet(const MonoTime& deadline,
std::set<std::string>* blacklist) {
- return OpenTablet(remote_->partition().partition_key_start(),
+ return OpenTablet(remote_->partition().begin(),
deadline,
blacklist);
}
@@ -400,7 +400,7 @@ ScanRpcStatus KuduScanner::Data::SendScanRpc(const MonoTime& overall_deadline,
return scan_status;
}
-Status KuduScanner::Data::OpenTablet(const string& partition_key,
+Status KuduScanner::Data::OpenTablet(const PartitionKey& partition_key,
const MonoTime& deadline,
set<string>* blacklist) {
SCOPED_LOG_SLOW_EXECUTION(WARNING, 500, "opening tablet");
@@ -510,7 +510,7 @@ Status KuduScanner::Data::OpenTablet(const string& partition_key,
Status s = sync.Wait();
if (s.IsNotFound()) {
// No more tablets in the table.
- partition_pruner_.RemovePartitionKeyRange("");
+ partition_pruner_.RemovePartitionKeyRange({});
return Status::OK();
}
RETURN_NOT_OK(s);
@@ -518,9 +518,9 @@ Status KuduScanner::Data::OpenTablet(const string& partition_key,
// Check if the meta cache returned a tablet covering a partition key range past
// what we asked for. This can happen if the requested partition key falls
// in a non-covered range. In this case we can potentially prune the tablet.
- if (partition_key < remote_->partition().partition_key_start() &&
+ if (partition_key < remote_->partition().begin() &&
partition_pruner_.ShouldPrune(remote_->partition())) {
- partition_pruner_.RemovePartitionKeyRange(remote_->partition().partition_key_end());
+ partition_pruner_.RemovePartitionKeyRange(remote_->partition().end());
return Status::OK();
}
@@ -568,7 +568,7 @@ Status KuduScanner::Data::OpenTablet(const string& partition_key,
RETURN_NOT_OK(HandleError(scan_status, deadline, blacklist, /* needs_reopen=*/ nullptr));
}
- partition_pruner_.RemovePartitionKeyRange(remote_->partition().partition_key_end());
+ partition_pruner_.RemovePartitionKeyRange(remote_->partition().end());
next_req_.clear_new_scan_request();
data_in_open_ = (last_response_.has_data() && last_response_.data().num_rows() > 0) ||
diff --git a/src/kudu/client/scanner-internal.h b/src/kudu/client/scanner-internal.h
index 98e89ea..a4c8901 100644
--- a/src/kudu/client/scanner-internal.h
+++ b/src/kudu/client/scanner-internal.h
@@ -46,6 +46,7 @@
namespace kudu {
class MonoTime;
+class PartitionKey;
class Schema;
namespace tserver {
@@ -162,7 +163,7 @@ class KuduScanner::Data {
Status ReopenCurrentTablet(const MonoTime& deadline, std::set<std::string>* blacklist);
// Open the tablet to scan.
- Status OpenTablet(const std::string& partition_key,
+ Status OpenTablet(const PartitionKey& partition_key,
const MonoTime& deadline,
std::set<std::string>* blacklist);
diff --git a/src/kudu/common/common.proto b/src/kudu/common/common.proto
index 0c08353..923b020 100644
--- a/src/kudu/common/common.proto
+++ b/src/kudu/common/common.proto
@@ -379,7 +379,7 @@ message PartitionSchemaPB {
// The serialized format of a Kudu table partition.
message PartitionPB {
// The hash buckets of the partition. The number of hash buckets must match
- // the number of hash bucket components in the partition's schema.
+ // the number of hash dimensions in the partition's schema.
repeated int32 hash_buckets = 1 [packed = true];
// The encoded start partition key (inclusive).
optional bytes partition_key_start = 2;
diff --git a/src/kudu/common/partition-test.cc b/src/kudu/common/partition-test.cc
index c8fa5a7..259ba28 100644
--- a/src/kudu/common/partition-test.cc
+++ b/src/kudu/common/partition-test.cc
@@ -37,6 +37,7 @@
#include "kudu/common/row_operations.pb.h"
#include "kudu/common/schema.h"
#include "kudu/gutil/strings/join.h"
+#include "kudu/gutil/strings/substitute.h"
#include "kudu/util/slice.h"
#include "kudu/util/status.h"
#include "kudu/util/test_macros.h"
@@ -48,6 +49,7 @@ using std::pair;
using std::string;
using std::vector;
using std::make_pair;
+using strings::Substitute;
namespace kudu {
@@ -116,12 +118,16 @@ void CheckCreateRangePartitions(const vector<pair<optional<string>, optional<str
for (int i = 0; i < partitions.size(); i++) {
const string& lower = expected_partition_ranges[i].first;
const string& upper = expected_partition_ranges[i].second;
-
- EXPECT_TRUE(partitions[i].hash_buckets().empty());
- EXPECT_EQ(lower, partitions[i].range_key_start());
- EXPECT_EQ(upper, partitions[i].range_key_end());
- EXPECT_EQ(lower, partitions[i].partition_key_start());
- EXPECT_EQ(upper, partitions[i].partition_key_end());
+ SCOPED_TRACE(Substitute("partition $0", i));
+
+ const auto& p = partitions[i];
+ ASSERT_TRUE(p.hash_buckets().empty());
+ ASSERT_TRUE(p.end().hash_key().empty());
+ ASSERT_TRUE(p.begin().hash_key().empty());
+ ASSERT_EQ(lower, p.begin().range_key());
+ ASSERT_EQ(upper, p.end().range_key());
+ ASSERT_EQ(lower, p.begin().ToString());
+ ASSERT_EQ(upper, p.end().ToString());
}
}
@@ -197,7 +203,7 @@ TEST_F(PartitionTest, TestCompoundRangeKeyEncoding) {
partition_schema.PartitionDebugString(partitions[3], schema));
}
-TEST_F(PartitionTest, TestPartitionKeyEncoding) {
+TEST_F(PartitionTest, PartitionKeyEncoding) {
// CREATE TABLE t (a INT32, b VARCHAR, c VARCHAR, PRIMARY KEY (a, b, c))
// PARTITION BY [HASH BUCKET (a, b), HASH BUCKET (c), RANGE (a, b, c)];
Schema schema({ ColumnSchema("a", INT32),
@@ -217,14 +223,21 @@ TEST_F(PartitionTest, TestPartitionKeyEncoding) {
{
KuduPartialRow row(&schema);
ASSERT_OK(row.SetInt32("a", 0));
- string key = partition_schema.EncodeKey(row);
-
- EXPECT_EQ(string("\0\0\0\0" // hash(0, "")
- "\0\0\0\x14" // hash("")
- "\x80\0\0\0" // a = 0
- "\0\0", // b = ""; c is elided
- 14), key);
- string expected = R"(HASH (a, b): 0, HASH (c): 20, RANGE (a, b, c): (0, "", ""))";
+ const auto key = partition_schema.EncodeKey(row);
+ EXPECT_EQ(string("\0\0\0\0" // hash(0, "")
+ "\0\0\0\x14", // hash("")
+ 8), key.hash_key());
+ EXPECT_EQ(string("\x80\0\0\0" // a = 0
+ "\0\0", // b = ""; c is elided
+ 6), key.range_key());
+ EXPECT_EQ(string("\0\0\0\0"
+ "\0\0\0\x14"
+ "\x80\0\0\0"
+ "\0\0",
+ 14), key.ToString());
+
+ const string expected =
+ R"(HASH (a, b): 0, HASH (c): 20, RANGE (a, b, c): (0, "", ""))";
EXPECT_EQ(expected, partition_schema.PartitionKeyDebugString(row));
EXPECT_EQ(expected, partition_schema.PartitionKeyDebugString(key, schema));
}
@@ -232,15 +245,22 @@ TEST_F(PartitionTest, TestPartitionKeyEncoding) {
{
KuduPartialRow row(&schema);
ASSERT_OK(row.SetInt32("a", 1));
- string key = partition_schema.EncodeKey(row);
+ const auto key = partition_schema.EncodeKey(row);
EXPECT_EQ(string("\0\0\0\x5" // hash(1, "")
- "\0\0\0\x14" // hash("")
- "\x80\0\0\x01" // a = 1
+ "\0\0\0\x14", // hash("")
+ 8), key.hash_key());
+ EXPECT_EQ(string("\x80\0\0\x01" // a = 1
"\0\0", // b = ""; c is elided
- 14), key);
-
- string expected = R"(HASH (a, b): 5, HASH (c): 20, RANGE (a, b, c): (1, "", ""))";
+ 6), key.range_key());
+ EXPECT_EQ(string("\0\0\0\x5"
+ "\0\0\0\x14"
+ "\x80\0\0\x01"
+ "\0\0",
+ 14), key.ToString());
+
+ const string expected =
+ R"(HASH (a, b): 5, HASH (c): 20, RANGE (a, b, c): (1, "", ""))";
EXPECT_EQ(expected, partition_schema.PartitionKeyDebugString(row));
EXPECT_EQ(expected, partition_schema.PartitionKeyDebugString(key, schema));
}
@@ -250,16 +270,24 @@ TEST_F(PartitionTest, TestPartitionKeyEncoding) {
ASSERT_OK(row.SetInt32("a", 0));
ASSERT_OK(row.SetStringCopy("b", "b"));
ASSERT_OK(row.SetStringCopy("c", "c"));
- string key = partition_schema.EncodeKey(row);
+ const auto key = partition_schema.EncodeKey(row);
EXPECT_EQ(string("\0\0\0\x1A" // hash(0, "b")
- "\0\0\0\x1D" // hash("c")
- "\x80\0\0\0" // a = 0
+ "\0\0\0\x1D",// hash("c")
+ 8), key.hash_key());
+ EXPECT_EQ(string("\x80\0\0\0" // a = 0
"b\0\0" // b = "b"
"c", // c = "c"
- 16), key);
-
- string expected = R"(HASH (a, b): 26, HASH (c): 29, RANGE (a, b, c): (0, "b", "c"))";
+ 8), key.range_key());
+ EXPECT_EQ(string("\0\0\0\x1A"
+ "\0\0\0\x1D"
+ "\x80\0\0\0"
+ "b\0\0"
+ "c",
+ 16), key.ToString());
+
+ const string expected =
+ R"(HASH (a, b): 26, HASH (c): 29, RANGE (a, b, c): (0, "b", "c"))";
EXPECT_EQ(expected, partition_schema.PartitionKeyDebugString(row));
EXPECT_EQ(expected, partition_schema.PartitionKeyDebugString(key, schema));
}
@@ -269,16 +297,24 @@ TEST_F(PartitionTest, TestPartitionKeyEncoding) {
ASSERT_OK(row.SetInt32("a", 1));
ASSERT_OK(row.SetStringCopy("b", "b"));
ASSERT_OK(row.SetStringCopy("c", "c"));
- string key = partition_schema.EncodeKey(row);
+ const auto key = partition_schema.EncodeKey(row);
EXPECT_EQ(string("\0\0\0\x0" // hash(1, "b")
- "\0\0\0\x1D" // hash("c")
- "\x80\0\0\x1" // a = 1
+ "\0\0\0\x1D", // hash("c")
+ 8), key.hash_key());
+ EXPECT_EQ(string("\x80\0\0\x1" // a = 1
"b\0\0" // b = "b"
"c", // c = "c"
- 16), key);
-
- string expected = R"(HASH (a, b): 0, HASH (c): 29, RANGE (a, b, c): (1, "b", "c"))";
+ 8), key.range_key());
+ EXPECT_EQ(string("\0\0\0\x0"
+ "\0\0\0\x1D"
+ "\x80\0\0\x1"
+ "b\0\0"
+ "c",
+ 16), key.ToString());
+
+ const string expected =
+ R"(HASH (a, b): 0, HASH (c): 29, RANGE (a, b, c): (1, "b", "c"))";
EXPECT_EQ(expected, partition_schema.PartitionKeyDebugString(row));
EXPECT_EQ(expected, partition_schema.PartitionKeyDebugString(key, schema));
}
@@ -290,24 +326,24 @@ TEST_F(PartitionTest, TestPartitionKeyEncoding) {
ASSERT_OK(row.SetInt32("a", 1));
ASSERT_OK(row.SetStringCopy("b", "b"));
ASSERT_OK(row.SetStringCopy("c", "c"));
- string key = partition_schema.EncodeKey(row);
+ const auto key = partition_schema.EncodeKey(row);
- string expected =
- R"(HASH (a, b): 0, HASH (c): 29, RANGE (a, b, c): (<redacted>, <redacted>, <redacted>))";
+ const string expected =
+ R"(HASH (a, b): 0, HASH (c): 29, RANGE (a, b, c): (<redacted>, <redacted>, <redacted>))";
EXPECT_EQ(expected, partition_schema.PartitionKeyDebugString(row));
EXPECT_EQ(expected, partition_schema.PartitionKeyDebugString(key, schema));
// Check that row values are redacted from error messages when
// --redact is set with 'log'.
- EXPECT_EQ("<range-decode-error>",
- partition_schema.PartitionKeyDebugString(string("\0\1\0\1", 4), schema));
+ EXPECT_EQ("<hash-decode-error>",
+ partition_schema.PartitionKeyDebugString(
+ PartitionKey(string("\0\1\0\1", 4), ""), schema));
EXPECT_EQ("HASH (a, b): 0, HASH (c): 0, RANGE (a, b, c): "
"<range-key-decode-error: Invalid argument: "
"Error decoding partition key range component 'a': key too short: <redacted>>",
- partition_schema.PartitionKeyDebugString(string("\0\0\0\0"
- "\0\0\0\0"
- "a", 9), schema));
+ partition_schema.PartitionKeyDebugString(
+ PartitionKey(string("\0\0\0\0\0\0\0\0", 8), "a"), schema));
}
}
@@ -331,7 +367,7 @@ TEST_F(PartitionTest, TestCreateRangePartitions) {
{ "2", string("2\0", 2) },
{ string("2\0", 2), "" },
};
- CheckCreateRangePartitions({}, splits, partitions);
+ NO_FATALS(CheckCreateRangePartitions({}, splits, partitions));
}
{
// Bounds
@@ -352,7 +388,7 @@ TEST_F(PartitionTest, TestCreateRangePartitions) {
{ "a", "m" },
{ "x", "z" },
};
- CheckCreateRangePartitions(bounds, {}, partitions);
+ NO_FATALS(CheckCreateRangePartitions(bounds, {}, partitions));
}
{
// Bounds:
@@ -393,7 +429,7 @@ TEST_F(PartitionTest, TestCreateRangePartitions) {
{ "h", string("h\0", 2) },
{ string("h\0", 2), "z" },
};
- CheckCreateRangePartitions(bounds, splits, partitions);
+ NO_FATALS(CheckCreateRangePartitions(bounds, splits, partitions));
}
{
// Bounds:
@@ -414,7 +450,7 @@ TEST_F(PartitionTest, TestCreateRangePartitions) {
{ "", "m" },
{ "m", "" },
};
- CheckCreateRangePartitions(bounds, splits, partitions);
+ NO_FATALS(CheckCreateRangePartitions(bounds, splits, partitions));
}
}
@@ -443,26 +479,26 @@ TEST_F(PartitionTest, TestCreateHashPartitions) {
ASSERT_EQ(3, partitions.size());
EXPECT_EQ(0, partitions[0].hash_buckets()[0]);
- EXPECT_EQ("", partitions[0].range_key_start());
- EXPECT_EQ("", partitions[0].range_key_end());
- EXPECT_EQ("", partitions[0].partition_key_start());
- EXPECT_EQ(string("\0\0\0\1", 4), partitions[0].partition_key_end());
+ EXPECT_TRUE(partitions[0].begin().range_key().empty());
+ EXPECT_TRUE(partitions[0].end().range_key().empty());
+ EXPECT_TRUE(partitions[0].begin().empty());
+ EXPECT_EQ(string("\0\0\0\1", 4), partitions[0].end().ToString());
EXPECT_EQ("HASH (a) PARTITION 0",
partition_schema.PartitionDebugString(partitions[0], schema));
EXPECT_EQ(1, partitions[1].hash_buckets()[0]);
- EXPECT_EQ("", partitions[1].range_key_start());
- EXPECT_EQ("", partitions[1].range_key_end());
- EXPECT_EQ(string("\0\0\0\1", 4), partitions[1].partition_key_start());
- EXPECT_EQ(string("\0\0\0\2", 4), partitions[1].partition_key_end());
+ EXPECT_TRUE(partitions[1].begin().range_key().empty());
+ EXPECT_TRUE(partitions[1].end().range_key().empty());
+ EXPECT_EQ(string("\0\0\0\1", 4), partitions[1].begin().ToString());
+ EXPECT_EQ(string("\0\0\0\2", 4), partitions[1].end().ToString());
EXPECT_EQ("HASH (a) PARTITION 1",
partition_schema.PartitionDebugString(partitions[1], schema));
EXPECT_EQ(2, partitions[2].hash_buckets()[0]);
- EXPECT_EQ("", partitions[2].range_key_start());
- EXPECT_EQ("", partitions[2].range_key_end());
- EXPECT_EQ(string("\0\0\0\2", 4), partitions[2].partition_key_start());
- EXPECT_EQ("", partitions[2].partition_key_end());
+ EXPECT_TRUE(partitions[2].begin().range_key().empty());
+ EXPECT_TRUE(partitions[2].end().range_key().empty());
+ EXPECT_EQ(string("\0\0\0\2", 4), partitions[2].begin().ToString());
+ EXPECT_EQ("", partitions[2].end().ToString());
EXPECT_EQ("HASH (a) PARTITION 2",
partition_schema.PartitionDebugString(partitions[2], schema));
}
@@ -520,12 +556,10 @@ TEST_F(PartitionTest, TestCreatePartitions) {
ASSERT_OK(split_a.SetStringCopy("a", "a1"));
ASSERT_OK(split_a.SetStringCopy("b", "b1"));
ASSERT_OK(split_a.SetStringCopy("c", "c1"));
- string partition_key_a = partition_schema.EncodeKey(split_a);
KuduPartialRow split_b(&schema);
ASSERT_OK(split_b.SetStringCopy("a", "a2"));
ASSERT_OK(split_b.SetStringCopy("b", "b2"));
- string partition_key_b = partition_schema.EncodeKey(split_b);
// Split keys need not be passed in sorted order.
vector<KuduPartialRow> split_rows = { split_b, split_a };
@@ -535,124 +569,127 @@ TEST_F(PartitionTest, TestCreatePartitions) {
EXPECT_EQ(0, partitions[0].hash_buckets()[0]);
EXPECT_EQ(0, partitions[0].hash_buckets()[1]);
- EXPECT_EQ("", partitions[0].range_key_start());
- EXPECT_EQ(string("a1\0\0b1\0\0c1", 10), partitions[0].range_key_end());
- EXPECT_EQ("", partitions[0].partition_key_start());
- EXPECT_EQ(string("\0\0\0\0" "\0\0\0\0" "a1\0\0b1\0\0c1", 18), partitions[0].partition_key_end());
+ EXPECT_EQ("", partitions[0].begin().range_key());
+ EXPECT_EQ(string("a1\0\0b1\0\0c1", 10), partitions[0].end().range_key());
+ EXPECT_EQ("", partitions[0].begin().ToString());
+ EXPECT_EQ(string("\0\0\0\0" "\0\0\0\0" "a1\0\0b1\0\0c1", 18),
+ partitions[0].end().ToString());
EXPECT_EQ("HASH (a) PARTITION 0, HASH (b) PARTITION 0, "
R"(RANGE (a, b, c) PARTITION VALUES < ("a1", "b1", "c1"))",
partition_schema.PartitionDebugString(partitions[0], schema));
+
EXPECT_EQ(0, partitions[1].hash_buckets()[0]);
EXPECT_EQ(0, partitions[1].hash_buckets()[1]);
- EXPECT_EQ(string("a1\0\0b1\0\0c1", 10), partitions[1].range_key_start());
- EXPECT_EQ(string("a2\0\0b2\0\0", 8), partitions[1].range_key_end());
+ EXPECT_EQ(string("a1\0\0b1\0\0c1", 10), partitions[1].begin().range_key());
+ EXPECT_EQ(string("a2\0\0b2\0\0", 8), partitions[1].end().range_key());
EXPECT_EQ(string("\0\0\0\0" "\0\0\0\0" "a1\0\0b1\0\0c1", 18),
- partitions[1].partition_key_start());
- EXPECT_EQ(string("\0\0\0\0" "\0\0\0\0" "a2\0\0b2\0\0", 16), partitions[1].partition_key_end());
+ partitions[1].begin().ToString());
+ EXPECT_EQ(string("\0\0\0\0" "\0\0\0\0" "a2\0\0b2\0\0", 16),
+ partitions[1].end().ToString());
EXPECT_EQ("HASH (a) PARTITION 0, HASH (b) PARTITION 0, "
R"(RANGE (a, b, c) PARTITION ("a1", "b1", "c1") <= VALUES < ("a2", "b2", ""))",
partition_schema.PartitionDebugString(partitions[1], schema));
EXPECT_EQ(0, partitions[2].hash_buckets()[0]);
EXPECT_EQ(0, partitions[2].hash_buckets()[1]);
- EXPECT_EQ(string("a2\0\0b2\0\0", 8), partitions[2].range_key_start());
- EXPECT_EQ("", partitions[2].range_key_end());
- EXPECT_EQ(string("\0\0\0\0" "\0\0\0\0" "a2\0\0b2\0\0", 16), partitions[2].partition_key_start());
- EXPECT_EQ(string("\0\0\0\0" "\0\0\0\1", 8), partitions[2].partition_key_end());
+ EXPECT_EQ(string("a2\0\0b2\0\0", 8), partitions[2].begin().range_key());
+ EXPECT_EQ("", partitions[2].end().range_key());
+ EXPECT_EQ(string("\0\0\0\0" "\0\0\0\0" "a2\0\0b2\0\0", 16), partitions[2].begin().ToString());
+ EXPECT_EQ(string("\0\0\0\0" "\0\0\0\1", 8), partitions[2].end().ToString());
EXPECT_EQ("HASH (a) PARTITION 0, HASH (b) PARTITION 0, "
R"(RANGE (a, b, c) PARTITION VALUES >= ("a2", "b2", ""))",
partition_schema.PartitionDebugString(partitions[2], schema));
EXPECT_EQ(0, partitions[3].hash_buckets()[0]);
EXPECT_EQ(1, partitions[3].hash_buckets()[1]);
- EXPECT_EQ("", partitions[3].range_key_start());
- EXPECT_EQ(string("a1\0\0b1\0\0c1", 10), partitions[3].range_key_end());
- EXPECT_EQ(string("\0\0\0\0" "\0\0\0\1", 8), partitions[3].partition_key_start());
- EXPECT_EQ(string("\0\0\0\0" "\0\0\0\1" "a1\0\0b1\0\0c1", 18), partitions[3].partition_key_end());
+ EXPECT_EQ("", partitions[3].begin().range_key());
+ EXPECT_EQ(string("a1\0\0b1\0\0c1", 10), partitions[3].end().range_key());
+ EXPECT_EQ(string("\0\0\0\0" "\0\0\0\1", 8), partitions[3].begin().ToString());
+ EXPECT_EQ(string("\0\0\0\0" "\0\0\0\1" "a1\0\0b1\0\0c1", 18), partitions[3].end().ToString());
EXPECT_EQ("HASH (a) PARTITION 0, HASH (b) PARTITION 1, "
R"(RANGE (a, b, c) PARTITION VALUES < ("a1", "b1", "c1"))",
partition_schema.PartitionDebugString(partitions[3], schema));
EXPECT_EQ(0, partitions[4].hash_buckets()[0]);
EXPECT_EQ(1, partitions[4].hash_buckets()[1]);
- EXPECT_EQ(string("a1\0\0b1\0\0c1", 10), partitions[4].range_key_start());
- EXPECT_EQ(string("a2\0\0b2\0\0", 8), partitions[4].range_key_end());
+ EXPECT_EQ(string("a1\0\0b1\0\0c1", 10), partitions[4].begin().range_key());
+ EXPECT_EQ(string("a2\0\0b2\0\0", 8), partitions[4].end().range_key());
EXPECT_EQ(string("\0\0\0\0" "\0\0\0\1" "a1\0\0b1\0\0c1", 18),
- partitions[4].partition_key_start());
- EXPECT_EQ(string("\0\0\0\0" "\0\0\0\1" "a2\0\0b2\0\0", 16), partitions[4].partition_key_end());
+ partitions[4].begin().ToString());
+ EXPECT_EQ(string("\0\0\0\0" "\0\0\0\1" "a2\0\0b2\0\0", 16), partitions[4].end().ToString());
EXPECT_EQ("HASH (a) PARTITION 0, HASH (b) PARTITION 1, "
R"(RANGE (a, b, c) PARTITION ("a1", "b1", "c1") <= VALUES < ("a2", "b2", ""))",
partition_schema.PartitionDebugString(partitions[4], schema));
EXPECT_EQ(0, partitions[5].hash_buckets()[0]);
EXPECT_EQ(1, partitions[5].hash_buckets()[1]);
- EXPECT_EQ(string("a2\0\0b2\0\0", 8), partitions[5].range_key_start());
- EXPECT_EQ("", partitions[5].range_key_end());
- EXPECT_EQ(string("\0\0\0\0" "\0\0\0\1" "a2\0\0b2\0\0", 16), partitions[5].partition_key_start());
- EXPECT_EQ(string("\0\0\0\1", 4), partitions[5].partition_key_end());
+ EXPECT_EQ(string("a2\0\0b2\0\0", 8), partitions[5].begin().range_key());
+ EXPECT_EQ("", partitions[5].end().range_key());
+ EXPECT_EQ(string("\0\0\0\0" "\0\0\0\1" "a2\0\0b2\0\0", 16), partitions[5].begin().ToString());
+ EXPECT_EQ(string("\0\0\0\1", 4), partitions[5].end().ToString());
EXPECT_EQ("HASH (a) PARTITION 0, HASH (b) PARTITION 1, "
R"(RANGE (a, b, c) PARTITION VALUES >= ("a2", "b2", ""))",
partition_schema.PartitionDebugString(partitions[5], schema));
EXPECT_EQ(1, partitions[6].hash_buckets()[0]);
EXPECT_EQ(0, partitions[6].hash_buckets()[1]);
- EXPECT_EQ("", partitions[6].range_key_start());
- EXPECT_EQ(string("a1\0\0b1\0\0c1", 10), partitions[6].range_key_end());
- EXPECT_EQ(string("\0\0\0\1", 4), partitions[6].partition_key_start());
- EXPECT_EQ(string("\0\0\0\1" "\0\0\0\0" "a1\0\0b1\0\0c1", 18), partitions[6].partition_key_end());
+ EXPECT_EQ("", partitions[6].begin().range_key());
+ EXPECT_EQ(string("a1\0\0b1\0\0c1", 10), partitions[6].end().range_key());
+ EXPECT_EQ(string("\0\0\0\1", 4), partitions[6].begin().ToString());
+ EXPECT_EQ(string("\0\0\0\1" "\0\0\0\0" "a1\0\0b1\0\0c1", 18), partitions[6].end().ToString());
EXPECT_EQ("HASH (a) PARTITION 1, HASH (b) PARTITION 0, "
R"(RANGE (a, b, c) PARTITION VALUES < ("a1", "b1", "c1"))",
partition_schema.PartitionDebugString(partitions[6], schema));
EXPECT_EQ(1, partitions[7].hash_buckets()[0]);
EXPECT_EQ(0, partitions[7].hash_buckets()[1]);
- EXPECT_EQ(string("a1\0\0b1\0\0c1", 10), partitions[7].range_key_start());
- EXPECT_EQ(string("a2\0\0b2\0\0", 8), partitions[7].range_key_end());
+ EXPECT_EQ(string("a1\0\0b1\0\0c1", 10), partitions[7].begin().range_key());
+ EXPECT_EQ(string("a2\0\0b2\0\0", 8), partitions[7].end().range_key());
EXPECT_EQ(string("\0\0\0\1" "\0\0\0\0" "a1\0\0b1\0\0c1", 18),
- partitions[7].partition_key_start());
- EXPECT_EQ(string("\0\0\0\1" "\0\0\0\0" "a2\0\0b2\0\0", 16), partitions[7].partition_key_end());
+ partitions[7].begin().ToString());
+ EXPECT_EQ(string("\0\0\0\1" "\0\0\0\0" "a2\0\0b2\0\0", 16), partitions[7].end().ToString());
EXPECT_EQ("HASH (a) PARTITION 1, HASH (b) PARTITION 0, "
R"(RANGE (a, b, c) PARTITION ("a1", "b1", "c1") <= VALUES < ("a2", "b2", ""))",
partition_schema.PartitionDebugString(partitions[7], schema));
EXPECT_EQ(1, partitions[8].hash_buckets()[0]);
EXPECT_EQ(0, partitions[8].hash_buckets()[1]);
- EXPECT_EQ(string("a2\0\0b2\0\0", 8), partitions[8].range_key_start());
- EXPECT_EQ("", partitions[8].range_key_end());
- EXPECT_EQ(string("\0\0\0\1" "\0\0\0\0" "a2\0\0b2\0\0", 16), partitions[8].partition_key_start());
- EXPECT_EQ(string("\0\0\0\1" "\0\0\0\1", 8), partitions[8].partition_key_end());
+ EXPECT_EQ(string("a2\0\0b2\0\0", 8), partitions[8].begin().range_key());
+ EXPECT_EQ("", partitions[8].end().range_key());
+ EXPECT_EQ(string("\0\0\0\1" "\0\0\0\0" "a2\0\0b2\0\0", 16), partitions[8].begin().ToString());
+ EXPECT_EQ(string("\0\0\0\1" "\0\0\0\1", 8), partitions[8].end().ToString());
EXPECT_EQ("HASH (a) PARTITION 1, HASH (b) PARTITION 0, "
R"(RANGE (a, b, c) PARTITION VALUES >= ("a2", "b2", ""))",
partition_schema.PartitionDebugString(partitions[8], schema));
EXPECT_EQ(1, partitions[9].hash_buckets()[0]);
EXPECT_EQ(1, partitions[9].hash_buckets()[1]);
- EXPECT_EQ("", partitions[9].range_key_start());
- EXPECT_EQ(string("a1\0\0b1\0\0c1", 10), partitions[9].range_key_end());
- EXPECT_EQ(string("\0\0\0\1" "\0\0\0\1", 8), partitions[9].partition_key_start());
- EXPECT_EQ(string("\0\0\0\1" "\0\0\0\1" "a1\0\0b1\0\0c1", 18), partitions[9].partition_key_end());
+ EXPECT_EQ("", partitions[9].begin().range_key());
+ EXPECT_EQ(string("a1\0\0b1\0\0c1", 10), partitions[9].end().range_key());
+ EXPECT_EQ(string("\0\0\0\1" "\0\0\0\1", 8), partitions[9].begin().ToString());
+ EXPECT_EQ(string("\0\0\0\1" "\0\0\0\1" "a1\0\0b1\0\0c1", 18), partitions[9].end().ToString());
EXPECT_EQ("HASH (a) PARTITION 1, HASH (b) PARTITION 1, "
R"(RANGE (a, b, c) PARTITION VALUES < ("a1", "b1", "c1"))",
partition_schema.PartitionDebugString(partitions[9], schema));
EXPECT_EQ(1, partitions[10].hash_buckets()[0]);
EXPECT_EQ(1, partitions[10].hash_buckets()[1]);
- EXPECT_EQ(string("a1\0\0b1\0\0c1", 10), partitions[10].range_key_start());
- EXPECT_EQ(string("a2\0\0b2\0\0", 8), partitions[10].range_key_end());
+ EXPECT_EQ(string("a1\0\0b1\0\0c1", 10), partitions[10].begin().range_key());
+ EXPECT_EQ(string("a2\0\0b2\0\0", 8), partitions[10].end().range_key());
EXPECT_EQ(string("\0\0\0\1" "\0\0\0\1" "a1\0\0b1\0\0c1", 18),
- partitions[10].partition_key_start());
- EXPECT_EQ(string("\0\0\0\1" "\0\0\0\1" "a2\0\0b2\0\0", 16), partitions[10].partition_key_end());
+ partitions[10].begin().ToString());
+ EXPECT_EQ(string("\0\0\0\1" "\0\0\0\1" "a2\0\0b2\0\0", 16), partitions[10].end().ToString());
EXPECT_EQ("HASH (a) PARTITION 1, HASH (b) PARTITION 1, "
R"(RANGE (a, b, c) PARTITION ("a1", "b1", "c1") <= VALUES < ("a2", "b2", ""))",
partition_schema.PartitionDebugString(partitions[10], schema));
EXPECT_EQ(1, partitions[11].hash_buckets()[0]);
EXPECT_EQ(1, partitions[11].hash_buckets()[1]);
- EXPECT_EQ(string("a2\0\0b2\0\0", 8), partitions[11].range_key_start());
- EXPECT_EQ("", partitions[11].range_key_end());
- EXPECT_EQ(string("\0\0\0\1" "\0\0\0\1" "a2\0\0b2\0\0", 16), partitions[11].partition_key_start());
- EXPECT_EQ("", partitions[11].partition_key_end());
+ EXPECT_EQ(string("a2\0\0b2\0\0", 8), partitions[11].begin().range_key());
+ EXPECT_EQ("", partitions[11].end().range_key());
+ EXPECT_EQ(string("\0\0\0\1" "\0\0\0\1" "a2\0\0b2\0\0", 16), partitions[11].begin().ToString());
+ EXPECT_EQ("", partitions[11].end().ToString());
EXPECT_EQ("HASH (a) PARTITION 1, HASH (b) PARTITION 1, "
R"(RANGE (a, b, c) PARTITION VALUES >= ("a2", "b2", ""))",
partition_schema.PartitionDebugString(partitions[11], schema));
@@ -951,163 +988,161 @@ TEST_F(PartitionTest, TestVaryingHashSchemasPerRange) {
ASSERT_EQ(1, partitions[0].hash_buckets().size());
EXPECT_EQ(0, partitions[0].hash_buckets()[0]);
- EXPECT_EQ(string("a1\0\0\0\0c1", 8), partitions[0].range_key_start());
- EXPECT_EQ(string("a2\0\0\0\0c2", 8), partitions[0].range_key_end());
+ EXPECT_EQ(string("a1\0\0\0\0c1", 8), partitions[0].begin().range_key());
+ EXPECT_EQ(string("a2\0\0\0\0c2", 8), partitions[0].end().range_key());
EXPECT_EQ(string("\0\0\0\0" "a1\0\0\0\0c1", 12),
- partitions[0].partition_key_start());
+ partitions[0].begin().ToString());
EXPECT_EQ(string("\0\0\0\0" "a2\0\0\0\0c2", 12),
- partitions[0].partition_key_end());
+ partitions[0].end().ToString());
ASSERT_EQ(1, partitions[1].hash_buckets().size());
EXPECT_EQ(1, partitions[1].hash_buckets()[0]);
- EXPECT_EQ(string("a1\0\0\0\0c1", 8), partitions[1].range_key_start());
- EXPECT_EQ(string("a2\0\0\0\0c2", 8), partitions[1].range_key_end());
+ EXPECT_EQ(string("a1\0\0\0\0c1", 8), partitions[1].begin().range_key());
+ EXPECT_EQ(string("a2\0\0\0\0c2", 8), partitions[1].end().range_key());
EXPECT_EQ(string("\0\0\0\1" "a1\0\0\0\0c1", 12),
- partitions[1].partition_key_start());
+ partitions[1].begin().ToString());
EXPECT_EQ(string("\0\0\0\1" "a2\0\0\0\0c2", 12),
- partitions[1].partition_key_end());
+ partitions[1].end().ToString());
ASSERT_EQ(1, partitions[2].hash_buckets().size());
EXPECT_EQ(2, partitions[2].hash_buckets()[0]);
- EXPECT_EQ(string("a1\0\0\0\0c1", 8), partitions[2].range_key_start());
- EXPECT_EQ(string("a2\0\0\0\0c2", 8), partitions[2].range_key_end());
+ EXPECT_EQ(string("a1\0\0\0\0c1", 8), partitions[2].begin().range_key());
+ EXPECT_EQ(string("a2\0\0\0\0c2", 8), partitions[2].end().range_key());
EXPECT_EQ(string("\0\0\0\2" "a1\0\0\0\0c1", 12),
- partitions[2].partition_key_start());
+ partitions[2].begin().ToString());
EXPECT_EQ(string("\0\0\0\2" "a2\0\0\0\0c2", 12),
- partitions[2].partition_key_end());
+ partitions[2].end().ToString());
ASSERT_EQ(1, partitions[3].hash_buckets().size());
EXPECT_EQ(3, partitions[3].hash_buckets()[0]);
- EXPECT_EQ(string("a1\0\0\0\0c1", 8), partitions[3].range_key_start());
- EXPECT_EQ(string("a2\0\0\0\0c2", 8), partitions[3].range_key_end());
+ EXPECT_EQ(string("a1\0\0\0\0c1", 8), partitions[3].begin().range_key());
+ EXPECT_EQ(string("a2\0\0\0\0c2", 8), partitions[3].end().range_key());
EXPECT_EQ(string("\0\0\0\3" "a1\0\0\0\0c1", 12),
- partitions[3].partition_key_start());
+ partitions[3].begin().ToString());
EXPECT_EQ(string("\0\0\0\3" "a2\0\0\0\0c2", 12),
- partitions[3].partition_key_end());
+ partitions[3].end().ToString());
ASSERT_EQ(2, partitions[4].hash_buckets().size());
EXPECT_EQ(0, partitions[4].hash_buckets()[0]);
EXPECT_EQ(0, partitions[4].hash_buckets()[1]);
- EXPECT_EQ(string("a3\0\0b3\0\0", 8),
- partitions[4].range_key_start());
- EXPECT_EQ(string("a4\0\0b4\0\0", 8),
- partitions[4].range_key_end());
+ EXPECT_EQ(string("a3\0\0b3\0\0", 8), partitions[4].begin().range_key());
+ EXPECT_EQ(string("a4\0\0b4\0\0", 8), partitions[4].end().range_key());
EXPECT_EQ(string("\0\0\0\0" "\0\0\0\0" "a3\0\0b3\0\0", 16),
- partitions[4].partition_key_start());
+ partitions[4].begin().ToString());
EXPECT_EQ(string("\0\0\0\0" "\0\0\0\0" "a4\0\0b4\0\0", 16),
- partitions[4].partition_key_end());
+ partitions[4].end().ToString());
ASSERT_EQ(2, partitions[5].hash_buckets().size());
EXPECT_EQ(0, partitions[5].hash_buckets()[0]);
EXPECT_EQ(1, partitions[5].hash_buckets()[1]);
- EXPECT_EQ(string("a3\0\0b3\0\0", 8), partitions[5].range_key_start());
- EXPECT_EQ(string("a4\0\0b4\0\0", 8), partitions[5].range_key_end());
+ EXPECT_EQ(string("a3\0\0b3\0\0", 8), partitions[5].begin().range_key());
+ EXPECT_EQ(string("a4\0\0b4\0\0", 8), partitions[5].end().range_key());
EXPECT_EQ(string("\0\0\0\0" "\0\0\0\1" "a3\0\0b3\0\0", 16),
- partitions[5].partition_key_start());
+ partitions[5].begin().ToString());
EXPECT_EQ(string("\0\0\0\0" "\0\0\0\1" "a4\0\0b4\0\0", 16),
- partitions[5].partition_key_end());
+ partitions[5].end().ToString());
ASSERT_EQ(2, partitions[6].hash_buckets().size());
EXPECT_EQ(1, partitions[6].hash_buckets()[0]);
EXPECT_EQ(0, partitions[6].hash_buckets()[1]);
- EXPECT_EQ(string("a3\0\0b3\0\0", 8), partitions[6].range_key_start());
- EXPECT_EQ(string("a4\0\0b4\0\0", 8), partitions[6].range_key_end());
+ EXPECT_EQ(string("a3\0\0b3\0\0", 8), partitions[6].begin().range_key());
+ EXPECT_EQ(string("a4\0\0b4\0\0", 8), partitions[6].end().range_key());
EXPECT_EQ(string("\0\0\0\1" "\0\0\0\0" "a3\0\0b3\0\0", 16),
- partitions[6].partition_key_start());
+ partitions[6].begin().ToString());
EXPECT_EQ(string("\0\0\0\1" "\0\0\0\0" "a4\0\0b4\0\0", 16),
- partitions[6].partition_key_end());
+ partitions[6].end().ToString());
ASSERT_EQ(2, partitions[7].hash_buckets().size());
EXPECT_EQ(1, partitions[7].hash_buckets()[0]);
EXPECT_EQ(1, partitions[7].hash_buckets()[1]);
- EXPECT_EQ(string("a3\0\0b3\0\0", 8), partitions[7].range_key_start());
- EXPECT_EQ(string("a4\0\0b4\0\0", 8), partitions[7].range_key_end());
+ EXPECT_EQ(string("a3\0\0b3\0\0", 8), partitions[7].begin().range_key());
+ EXPECT_EQ(string("a4\0\0b4\0\0", 8), partitions[7].end().range_key());
EXPECT_EQ(string("\0\0\0\1" "\0\0\0\1" "a3\0\0b3\0\0", 16),
- partitions[7].partition_key_start());
+ partitions[7].begin().ToString());
EXPECT_EQ(string("\0\0\0\1" "\0\0\0\1" "a4\0\0b4\0\0", 16),
- partitions[7].partition_key_end());
+ partitions[7].end().ToString());
ASSERT_EQ(2, partitions[8].hash_buckets().size());
EXPECT_EQ(2, partitions[8].hash_buckets()[0]);
EXPECT_EQ(0, partitions[8].hash_buckets()[1]);
- EXPECT_EQ(string("a3\0\0b3\0\0", 8), partitions[8].range_key_start());
- EXPECT_EQ(string("a4\0\0b4\0\0", 8), partitions[8].range_key_end());
+ EXPECT_EQ(string("a3\0\0b3\0\0", 8), partitions[8].begin().range_key());
+ EXPECT_EQ(string("a4\0\0b4\0\0", 8), partitions[8].end().range_key());
EXPECT_EQ(string("\0\0\0\2" "\0\0\0\0" "a3\0\0b3\0\0", 16),
- partitions[8].partition_key_start());
+ partitions[8].begin().ToString());
EXPECT_EQ(string("\0\0\0\2" "\0\0\0\0" "a4\0\0b4\0\0", 16),
- partitions[8].partition_key_end());
+ partitions[8].end().ToString());
ASSERT_EQ(2, partitions[9].hash_buckets().size());
EXPECT_EQ(2, partitions[9].hash_buckets()[0]);
EXPECT_EQ(1, partitions[9].hash_buckets()[1]);
- EXPECT_EQ(string("a3\0\0b3\0\0", 8), partitions[9].range_key_start());
- EXPECT_EQ(string("a4\0\0b4\0\0", 8), partitions[9].range_key_end());
+ EXPECT_EQ(string("a3\0\0b3\0\0", 8), partitions[9].begin().range_key());
+ EXPECT_EQ(string("a4\0\0b4\0\0", 8), partitions[9].end().range_key());
EXPECT_EQ(string("\0\0\0\2" "\0\0\0\1" "a3\0\0b3\0\0", 16),
- partitions[9].partition_key_start());
+ partitions[9].begin().ToString());
EXPECT_EQ(string("\0\0\0\2" "\0\0\0\1" "a4\0\0b4\0\0", 16),
- partitions[9].partition_key_end());
+ partitions[9].end().ToString());
ASSERT_EQ(2, partitions[10].hash_buckets().size());
EXPECT_EQ(0, partitions[10].hash_buckets()[0]);
EXPECT_EQ(0, partitions[10].hash_buckets()[1]);
EXPECT_EQ(string("a5\0\0b5\0\0", 8),
- partitions[10].range_key_start());
+ partitions[10].begin().range_key());
EXPECT_EQ(string("a6\0\0\0\0c6", 8),
- partitions[10].range_key_end());
+ partitions[10].end().range_key());
EXPECT_EQ(string("\0\0\0\0" "\0\0\0\0" "a5\0\0b5\0\0", 16),
- partitions[10].partition_key_start());
+ partitions[10].begin().ToString());
EXPECT_EQ(string("\0\0\0\0" "\0\0\0\0" "a6\0\0\0\0c6", 16),
- partitions[10].partition_key_end());
+ partitions[10].end().ToString());
ASSERT_EQ(2, partitions[11].hash_buckets().size());
EXPECT_EQ(0, partitions[11].hash_buckets()[0]);
EXPECT_EQ(1, partitions[11].hash_buckets()[1]);
- EXPECT_EQ(string("a5\0\0b5\0\0", 8),partitions[11].range_key_start());
- EXPECT_EQ(string("a6\0\0\0\0c6", 8),partitions[11].range_key_end());
+ EXPECT_EQ(string("a5\0\0b5\0\0", 8),partitions[11].begin().range_key());
+ EXPECT_EQ(string("a6\0\0\0\0c6", 8),partitions[11].end().range_key());
EXPECT_EQ(string("\0\0\0\0" "\0\0\0\1" "a5\0\0b5\0\0", 16),
- partitions[11].partition_key_start());
+ partitions[11].begin().ToString());
EXPECT_EQ(string("\0\0\0\0" "\0\0\0\1" "a6\0\0\0\0c6", 16),
- partitions[11].partition_key_end());
+ partitions[11].end().ToString());
ASSERT_EQ(2, partitions[12].hash_buckets().size());
EXPECT_EQ(0, partitions[12].hash_buckets()[0]);
EXPECT_EQ(2, partitions[12].hash_buckets()[1]);
- EXPECT_EQ(string("a5\0\0b5\0\0", 8), partitions[12].range_key_start());
- EXPECT_EQ(string("a6\0\0\0\0c6", 8), partitions[12].range_key_end());
+ EXPECT_EQ(string("a5\0\0b5\0\0", 8), partitions[12].begin().range_key());
+ EXPECT_EQ(string("a6\0\0\0\0c6", 8), partitions[12].end().range_key());
EXPECT_EQ(string("\0\0\0\0" "\0\0\0\2" "a5\0\0b5\0\0", 16),
- partitions[12].partition_key_start());
+ partitions[12].begin().ToString());
EXPECT_EQ(string("\0\0\0\0" "\0\0\0\2" "a6\0\0\0\0c6", 16),
- partitions[12].partition_key_end());
+ partitions[12].end().ToString());
ASSERT_EQ(2, partitions[13].hash_buckets().size());
EXPECT_EQ(1, partitions[13].hash_buckets()[0]);
EXPECT_EQ(0, partitions[13].hash_buckets()[1]);
- EXPECT_EQ(string("a5\0\0b5\0\0", 8), partitions[13].range_key_start());
- EXPECT_EQ(string("a6\0\0\0\0c6", 8), partitions[13].range_key_end());
+ EXPECT_EQ(string("a5\0\0b5\0\0", 8), partitions[13].begin().range_key());
+ EXPECT_EQ(string("a6\0\0\0\0c6", 8), partitions[13].end().range_key());
EXPECT_EQ(string("\0\0\0\1" "\0\0\0\0" "a5\0\0b5\0\0", 16),
- partitions[13].partition_key_start());
+ partitions[13].begin().ToString());
EXPECT_EQ(string("\0\0\0\1" "\0\0\0\0" "a6\0\0\0\0c6", 16),
- partitions[13].partition_key_end());
+ partitions[13].end().ToString());
ASSERT_EQ(2, partitions[14].hash_buckets().size());
EXPECT_EQ(1, partitions[14].hash_buckets()[0]);
EXPECT_EQ(1, partitions[14].hash_buckets()[1]);
- EXPECT_EQ(string("a5\0\0b5\0\0", 8), partitions[14].range_key_start());
- EXPECT_EQ(string("a6\0\0\0\0c6", 8), partitions[14].range_key_end());
+ EXPECT_EQ(string("a5\0\0b5\0\0", 8), partitions[14].begin().range_key());
+ EXPECT_EQ(string("a6\0\0\0\0c6", 8), partitions[14].end().range_key());
EXPECT_EQ(string("\0\0\0\1" "\0\0\0\1" "a5\0\0b5\0\0", 16),
- partitions[14].partition_key_start());
+ partitions[14].begin().ToString());
EXPECT_EQ(string("\0\0\0\1" "\0\0\0\1" "a6\0\0\0\0c6", 16),
- partitions[14].partition_key_end());
+ partitions[14].end().ToString());
ASSERT_EQ(2, partitions[15].hash_buckets().size());
EXPECT_EQ(1, partitions[15].hash_buckets()[0]);
EXPECT_EQ(2, partitions[15].hash_buckets()[1]);
- EXPECT_EQ(string("a5\0\0b5\0\0", 8), partitions[15].range_key_start());
- EXPECT_EQ(string("a6\0\0\0\0c6", 8), partitions[15].range_key_end());
+ EXPECT_EQ(string("a5\0\0b5\0\0", 8), partitions[15].begin().range_key());
+ EXPECT_EQ(string("a6\0\0\0\0c6", 8), partitions[15].end().range_key());
EXPECT_EQ(string("\0\0\0\1" "\0\0\0\2" "a5\0\0b5\0\0", 16),
- partitions[15].partition_key_start());
+ partitions[15].begin().ToString());
EXPECT_EQ(string("\0\0\0\1" "\0\0\0\2" "a6\0\0\0\0c6", 16),
- partitions[15].partition_key_end());
+ partitions[15].end().ToString());
};
vector<Partition> partitions;
@@ -1214,16 +1249,16 @@ TEST_F(PartitionTest, CustomHashSchemasPerRangeOnly) {
const auto& p = partitions[0];
ASSERT_EQ(1, p.hash_buckets().size());
EXPECT_EQ(0, p.hash_buckets()[0]);
- EXPECT_EQ(string("a1\0\0b1", 6), p.range_key_start());
- EXPECT_EQ(string("a2\0\0b2", 6), p.range_key_end());
+ EXPECT_EQ(string("a1\0\0b1", 6), p.begin().range_key());
+ EXPECT_EQ(string("a2\0\0b2", 6), p.end().range_key());
}
{
const auto& p = partitions[1];
ASSERT_EQ(1, p.hash_buckets().size());
EXPECT_EQ(1, p.hash_buckets()[0]);
- EXPECT_EQ(string("a1\0\0b1", 6), p.range_key_start());
- EXPECT_EQ(string("a2\0\0b2", 6), p.range_key_end());
+ EXPECT_EQ(string("a1\0\0b1", 6), p.begin().range_key());
+ EXPECT_EQ(string("a2\0\0b2", 6), p.end().range_key());
}
}
@@ -1290,85 +1325,85 @@ TEST_F(PartitionTest, TestVaryingHashSchemasPerUnboundedRanges) {
ASSERT_EQ(1, partitions[0].hash_buckets().size());
EXPECT_EQ(0, partitions[0].hash_buckets()[0]);
- EXPECT_EQ("", partitions[0].range_key_start());
- EXPECT_EQ(string("a1\0\0\0\0c1", 8), partitions[0].range_key_end());
- EXPECT_EQ("", partitions[0].partition_key_start());
- EXPECT_EQ(string("\0\0\0\0" "a1\0\0\0\0c1", 12), partitions[0].partition_key_end());
+ EXPECT_EQ("", partitions[0].begin().range_key());
+ EXPECT_EQ(string("a1\0\0\0\0c1", 8), partitions[0].end().range_key());
+ EXPECT_EQ("", partitions[0].begin().ToString());
+ EXPECT_EQ(string("\0\0\0\0" "a1\0\0\0\0c1", 12), partitions[0].end().ToString());
ASSERT_EQ(1, partitions[1].hash_buckets().size());
EXPECT_EQ(1, partitions[1].hash_buckets()[0]);
- EXPECT_EQ("", partitions[1].range_key_start());
- EXPECT_EQ(string("a1\0\0\0\0c1", 8), partitions[1].range_key_end());
- EXPECT_EQ(string("\0\0\0\1", 4),partitions[1].partition_key_start());
- EXPECT_EQ(string("\0\0\0\1" "a1\0\0\0\0c1", 12), partitions[1].partition_key_end());
+ EXPECT_EQ("", partitions[1].begin().range_key());
+ EXPECT_EQ(string("a1\0\0\0\0c1", 8), partitions[1].end().range_key());
+ EXPECT_EQ(string("\0\0\0\1", 4),partitions[1].begin().ToString());
+ EXPECT_EQ(string("\0\0\0\1" "a1\0\0\0\0c1", 12), partitions[1].end().ToString());
ASSERT_EQ(1, partitions[2].hash_buckets().size());
EXPECT_EQ(2, partitions[2].hash_buckets()[0]);
- EXPECT_EQ("", partitions[2].range_key_start());
- EXPECT_EQ(string("a1\0\0\0\0c1", 8), partitions[2].range_key_end());
- EXPECT_EQ(string("\0\0\0\2", 4), partitions[2].partition_key_start());
- EXPECT_EQ(string("\0\0\0\2" "a1\0\0\0\0c1", 12), partitions[2].partition_key_end());
+ EXPECT_EQ("", partitions[2].begin().range_key());
+ EXPECT_EQ(string("a1\0\0\0\0c1", 8), partitions[2].end().range_key());
+ EXPECT_EQ(string("\0\0\0\2", 4), partitions[2].begin().ToString());
+ EXPECT_EQ(string("\0\0\0\2" "a1\0\0\0\0c1", 12), partitions[2].end().ToString());
ASSERT_EQ(1, partitions[3].hash_buckets().size());
EXPECT_EQ(3, partitions[3].hash_buckets()[0]);
- EXPECT_EQ("", partitions[3].range_key_start());
- EXPECT_EQ(string("a1\0\0\0\0c1", 8), partitions[3].range_key_end());
- EXPECT_EQ(string("\0\0\0\3", 4), partitions[3].partition_key_start());
- EXPECT_EQ(string("\0\0\0\3" "a1\0\0\0\0c1", 12), partitions[3].partition_key_end());
+ EXPECT_EQ("", partitions[3].begin().range_key());
+ EXPECT_EQ(string("a1\0\0\0\0c1", 8), partitions[3].end().range_key());
+ EXPECT_EQ(string("\0\0\0\3", 4), partitions[3].begin().ToString());
+ EXPECT_EQ(string("\0\0\0\3" "a1\0\0\0\0c1", 12), partitions[3].end().ToString());
ASSERT_EQ(0, partitions[4].hash_buckets().size());
- EXPECT_EQ(string("a2\0\0b2\0\0", 8), partitions[4].range_key_start());
- EXPECT_EQ(string("a3\0\0b3\0\0", 8), partitions[4].range_key_end());
- EXPECT_EQ(string("a2\0\0b2\0\0", 8), partitions[4].partition_key_start());
- EXPECT_EQ(string("a3\0\0b3\0\0", 8), partitions[4].partition_key_end());
+ EXPECT_EQ(string("a2\0\0b2\0\0", 8), partitions[4].begin().range_key());
+ EXPECT_EQ(string("a3\0\0b3\0\0", 8), partitions[4].end().range_key());
+ EXPECT_EQ(string("a2\0\0b2\0\0", 8), partitions[4].begin().ToString());
+ EXPECT_EQ(string("a3\0\0b3\0\0", 8), partitions[4].end().ToString());
ASSERT_EQ(2, partitions[5].hash_buckets().size());
EXPECT_EQ(0, partitions[5].hash_buckets()[0]);
EXPECT_EQ(0, partitions[5].hash_buckets()[1]);
- EXPECT_EQ(string("a4\0\0b4\0\0", 8), partitions[5].range_key_start());
- EXPECT_EQ("", partitions[5].range_key_end());
- EXPECT_EQ(string("\0\0\0\0" "\0\0\0\0" "a4\0\0b4\0\0", 16), partitions[5].partition_key_start());
- EXPECT_EQ(string("\0\0\0\0" "\0\0\0\1", 8), partitions[5].partition_key_end());
+ EXPECT_EQ(string("a4\0\0b4\0\0", 8), partitions[5].begin().range_key());
+ EXPECT_EQ("", partitions[5].end().range_key());
+ EXPECT_EQ(string("\0\0\0\0" "\0\0\0\0" "a4\0\0b4\0\0", 16), partitions[5].begin().ToString());
+ EXPECT_EQ(string("\0\0\0\0" "\0\0\0\1", 8), partitions[5].end().ToString());
ASSERT_EQ(2, partitions[6].hash_buckets().size());
EXPECT_EQ(0, partitions[6].hash_buckets()[0]);
EXPECT_EQ(1, partitions[6].hash_buckets()[1]);
- EXPECT_EQ(string("a4\0\0b4\0\0", 8), partitions[6].range_key_start());
- EXPECT_EQ("", partitions[6].range_key_end());
- EXPECT_EQ(string("\0\0\0\0" "\0\0\0\1" "a4\0\0b4\0\0", 16),partitions[6].partition_key_start());
- EXPECT_EQ(string("\0\0\0\0" "\0\0\0\2", 8), partitions[6].partition_key_end());
+ EXPECT_EQ(string("a4\0\0b4\0\0", 8), partitions[6].begin().range_key());
+ EXPECT_EQ("", partitions[6].end().range_key());
+ EXPECT_EQ(string("\0\0\0\0" "\0\0\0\1" "a4\0\0b4\0\0", 16),partitions[6].begin().ToString());
+ EXPECT_EQ(string("\0\0\0\0" "\0\0\0\2", 8), partitions[6].end().ToString());
ASSERT_EQ(2, partitions[7].hash_buckets().size());
EXPECT_EQ(0, partitions[7].hash_buckets()[0]);
EXPECT_EQ(2, partitions[7].hash_buckets()[1]);
- EXPECT_EQ(string("a4\0\0b4\0\0", 8), partitions[7].range_key_start());
- EXPECT_EQ("", partitions[7].range_key_end());
- EXPECT_EQ(string("\0\0\0\0" "\0\0\0\2" "a4\0\0b4\0\0", 16), partitions[7].partition_key_start());
- EXPECT_EQ(string("\0\0\0\1", 4), partitions[7].partition_key_end());
+ EXPECT_EQ(string("a4\0\0b4\0\0", 8), partitions[7].begin().range_key());
+ EXPECT_EQ("", partitions[7].end().range_key());
+ EXPECT_EQ(string("\0\0\0\0" "\0\0\0\2" "a4\0\0b4\0\0", 16), partitions[7].begin().ToString());
+ EXPECT_EQ(string("\0\0\0\1", 4), partitions[7].end().ToString());
ASSERT_EQ(2, partitions[8].hash_buckets().size());
EXPECT_EQ(1, partitions[8].hash_buckets()[0]);
EXPECT_EQ(0, partitions[8].hash_buckets()[1]);
- EXPECT_EQ(string("a4\0\0b4\0\0", 8), partitions[8].range_key_start());
- EXPECT_EQ("", partitions[8].range_key_end());
- EXPECT_EQ(string("\0\0\0\1" "\0\0\0\0" "a4\0\0b4\0\0", 16), partitions[8].partition_key_start());
- EXPECT_EQ(string("\0\0\0\1" "\0\0\0\1", 8), partitions[8].partition_key_end());
+ EXPECT_EQ(string("a4\0\0b4\0\0", 8), partitions[8].begin().range_key());
+ EXPECT_EQ("", partitions[8].end().range_key());
+ EXPECT_EQ(string("\0\0\0\1" "\0\0\0\0" "a4\0\0b4\0\0", 16), partitions[8].begin().ToString());
+ EXPECT_EQ(string("\0\0\0\1" "\0\0\0\1", 8), partitions[8].end().ToString());
ASSERT_EQ(2, partitions[9].hash_buckets().size());
EXPECT_EQ(1, partitions[9].hash_buckets()[0]);
EXPECT_EQ(1, partitions[9].hash_buckets()[1]);
- EXPECT_EQ(string("a4\0\0b4\0\0", 8), partitions[9].range_key_start());
- EXPECT_EQ("", partitions[9].range_key_end());
- EXPECT_EQ(string("\0\0\0\1" "\0\0\0\1" "a4\0\0b4\0\0", 16),partitions[9].partition_key_start());
- EXPECT_EQ(string("\0\0\0\1" "\0\0\0\2", 8), partitions[9].partition_key_end());
+ EXPECT_EQ(string("a4\0\0b4\0\0", 8), partitions[9].begin().range_key());
+ EXPECT_EQ("", partitions[9].end().range_key());
+ EXPECT_EQ(string("\0\0\0\1" "\0\0\0\1" "a4\0\0b4\0\0", 16),partitions[9].begin().ToString());
+ EXPECT_EQ(string("\0\0\0\1" "\0\0\0\2", 8), partitions[9].end().ToString());
ASSERT_EQ(2, partitions[10].hash_buckets().size());
EXPECT_EQ(1, partitions[10].hash_buckets()[0]);
EXPECT_EQ(2, partitions[10].hash_buckets()[1]);
- EXPECT_EQ(string("a4\0\0b4\0\0", 8), partitions[10].range_key_start());
- EXPECT_EQ("", partitions[10].range_key_end());
- EXPECT_EQ(string("\0\0\0\1" "\0\0\0\2" "a4\0\0b4\0\0", 16), partitions[10].partition_key_start());
- EXPECT_EQ("", partitions[10].partition_key_end());
+ EXPECT_EQ(string("a4\0\0b4\0\0", 8), partitions[10].begin().range_key());
+ EXPECT_EQ("", partitions[10].end().range_key());
+ EXPECT_EQ(string("\0\0\0\1" "\0\0\0\2" "a4\0\0b4\0\0", 16), partitions[10].begin().ToString());
+ EXPECT_EQ("", partitions[10].end().ToString());
}
TEST_F(PartitionTest, NoHashSchemasForLastUnboundedRange) {
@@ -1433,97 +1468,96 @@ TEST_F(PartitionTest, NoHashSchemasForLastUnboundedRange) {
const auto& p = partitions[0];
ASSERT_EQ(1, p.hash_buckets().size());
EXPECT_EQ(0, p.hash_buckets()[0]);
- EXPECT_EQ("", p.range_key_start());
- EXPECT_EQ(string("a1\0\0", 4), p.range_key_end());
- EXPECT_EQ("", p.partition_key_start());
- EXPECT_EQ(string("\0\0\0\0" "a1\0\0c1", 8), p.partition_key_end());
+ EXPECT_EQ("", p.begin().range_key());
+ EXPECT_EQ(string("a1\0\0", 4), p.end().range_key());
+ EXPECT_EQ("", p.begin().ToString());
+ EXPECT_EQ(string("\0\0\0\0" "a1\0\0c1", 8), p.end().ToString());
}
{
const auto& p = partitions[1];
ASSERT_EQ(1, p.hash_buckets().size());
- ASSERT_EQ(1, p.hash_buckets().size());
EXPECT_EQ(1, p.hash_buckets()[0]);
- EXPECT_EQ("", p.range_key_start());
- EXPECT_EQ(string("a1\0\0", 4), p.range_key_end());
- EXPECT_EQ(string("\0\0\0\1", 4), p.partition_key_start());
- EXPECT_EQ(string("\0\0\0\1" "a1\0\0", 8), p.partition_key_end());
+ EXPECT_EQ("", p.begin().range_key());
+ EXPECT_EQ(string("a1\0\0", 4), p.end().range_key());
+ EXPECT_EQ(string("\0\0\0\1", 4), p.begin().ToString());
+ EXPECT_EQ(string("\0\0\0\1" "a1\0\0", 8), p.end().ToString());
}
{
const auto& p = partitions[2];
ASSERT_EQ(1, p.hash_buckets().size());
EXPECT_EQ(2, p.hash_buckets()[0]);
- EXPECT_EQ("", p.range_key_start());
- EXPECT_EQ(string("a1\0\0", 4), p.range_key_end());
- EXPECT_EQ(string("\0\0\0\2", 4), p.partition_key_start());
- EXPECT_EQ(string("\0\0\0\2" "a1\0\0", 8), p.partition_key_end());
+ EXPECT_EQ("", p.begin().range_key());
+ EXPECT_EQ(string("a1\0\0", 4), p.end().range_key());
+ EXPECT_EQ(string("\0\0\0\2", 4), p.begin().ToString());
+ EXPECT_EQ(string("\0\0\0\2" "a1\0\0", 8), p.end().ToString());
}
{
const auto& p = partitions[3];
ASSERT_EQ(2, p.hash_buckets().size());
EXPECT_EQ(0, p.hash_buckets()[0]);
EXPECT_EQ(0, p.hash_buckets()[1]);
- EXPECT_EQ(string("a2\0\0", 4), p.range_key_start());
- EXPECT_EQ(string("a3\0\0", 4), p.range_key_end());
- EXPECT_EQ(string("\0\0\0\0" "\0\0\0\0" "a2\0\0", 12), p.partition_key_start());
- EXPECT_EQ(string("\0\0\0\0" "\0\0\0\0" "a3\0\0", 12), p.partition_key_end());
+ EXPECT_EQ(string("a2\0\0", 4), p.begin().range_key());
+ EXPECT_EQ(string("a3\0\0", 4), p.end().range_key());
+ EXPECT_EQ(string("\0\0\0\0" "\0\0\0\0" "a2\0\0", 12), p.begin().ToString());
+ EXPECT_EQ(string("\0\0\0\0" "\0\0\0\0" "a3\0\0", 12), p.end().ToString());
}
{
const auto& p = partitions[4];
ASSERT_EQ(2, p.hash_buckets().size());
EXPECT_EQ(0, p.hash_buckets()[0]);
EXPECT_EQ(1, p.hash_buckets()[1]);
- EXPECT_EQ(string("a2\0\0", 4), p.range_key_start());
- EXPECT_EQ(string("a3\0\0", 4), p.range_key_end());
- EXPECT_EQ(string("\0\0\0\0" "\0\0\0\1" "a2\0\0", 12), p.partition_key_start());
- EXPECT_EQ(string("\0\0\0\0" "\0\0\0\1" "a3\0\0", 12), p.partition_key_end());
+ EXPECT_EQ(string("a2\0\0", 4), p.begin().range_key());
+ EXPECT_EQ(string("a3\0\0", 4), p.end().range_key());
+ EXPECT_EQ(string("\0\0\0\0" "\0\0\0\1" "a2\0\0", 12), p.begin().ToString());
+ EXPECT_EQ(string("\0\0\0\0" "\0\0\0\1" "a3\0\0", 12), p.end().ToString());
}
{
const auto& p = partitions[5];
ASSERT_EQ(2, p.hash_buckets().size());
EXPECT_EQ(1, p.hash_buckets()[0]);
EXPECT_EQ(0, p.hash_buckets()[1]);
- EXPECT_EQ(string("a2\0\0", 4), p.range_key_start());
- EXPECT_EQ(string("a3\0\0", 4), p.range_key_end());
- EXPECT_EQ(string("\0\0\0\1" "\0\0\0\0" "a2\0\0", 12), p.partition_key_start());
- EXPECT_EQ(string("\0\0\0\1" "\0\0\0\0" "a3\0\0", 12), p.partition_key_end());
+ EXPECT_EQ(string("a2\0\0", 4), p.begin().range_key());
+ EXPECT_EQ(string("a3\0\0", 4), p.end().range_key());
+ EXPECT_EQ(string("\0\0\0\1" "\0\0\0\0" "a2\0\0", 12), p.begin().ToString());
+ EXPECT_EQ(string("\0\0\0\1" "\0\0\0\0" "a3\0\0", 12), p.end().ToString());
}
{
const auto& p = partitions[6];
ASSERT_EQ(2, p.hash_buckets().size());
EXPECT_EQ(1, p.hash_buckets()[0]);
EXPECT_EQ(1, p.hash_buckets()[1]);
- EXPECT_EQ(string("a2\0\0", 4), p.range_key_start());
- EXPECT_EQ(string("a3\0\0", 4), p.range_key_end());
- EXPECT_EQ(string("\0\0\0\1" "\0\0\0\1" "a2\0\0", 12), p.partition_key_start());
- EXPECT_EQ(string("\0\0\0\1" "\0\0\0\1" "a3\0\0", 12), p.partition_key_end());
+ EXPECT_EQ(string("a2\0\0", 4), p.begin().range_key());
+ EXPECT_EQ(string("a3\0\0", 4), p.end().range_key());
+ EXPECT_EQ(string("\0\0\0\1" "\0\0\0\1" "a2\0\0", 12), p.begin().ToString());
+ EXPECT_EQ(string("\0\0\0\1" "\0\0\0\1" "a3\0\0", 12), p.end().ToString());
}
{
const auto& p = partitions[7];
ASSERT_EQ(2, p.hash_buckets().size());
EXPECT_EQ(2, p.hash_buckets()[0]);
EXPECT_EQ(0, p.hash_buckets()[1]);
- EXPECT_EQ(string("a2\0\0", 4), p.range_key_start());
- EXPECT_EQ(string("a3\0\0", 4), p.range_key_end());
- EXPECT_EQ(string("\0\0\0\2" "\0\0\0\0" "a2\0\0", 12), p.partition_key_start());
- EXPECT_EQ(string("\0\0\0\2" "\0\0\0\0" "a3\0\0", 12), p.partition_key_end());
+ EXPECT_EQ(string("a2\0\0", 4), p.begin().range_key());
+ EXPECT_EQ(string("a3\0\0", 4), p.end().range_key());
+ EXPECT_EQ(string("\0\0\0\2" "\0\0\0\0" "a2\0\0", 12), p.begin().ToString());
+ EXPECT_EQ(string("\0\0\0\2" "\0\0\0\0" "a3\0\0", 12), p.end().ToString());
}
{
const auto& p = partitions[8];
ASSERT_EQ(2, p.hash_buckets().size());
EXPECT_EQ(2, p.hash_buckets()[0]);
EXPECT_EQ(1, p.hash_buckets()[1]);
- EXPECT_EQ(string("a2\0\0", 4), p.range_key_start());
- EXPECT_EQ(string("a3\0\0", 4), p.range_key_end());
- EXPECT_EQ(string("\0\0\0\2" "\0\0\0\1" "a2\0\0", 12), p.partition_key_start());
- EXPECT_EQ(string("\0\0\0\2" "\0\0\0\1" "a3\0\0", 12), p.partition_key_end());
+ EXPECT_EQ(string("a2\0\0", 4), p.begin().range_key());
+ EXPECT_EQ(string("a3\0\0", 4), p.end().range_key());
+ EXPECT_EQ(string("\0\0\0\2" "\0\0\0\1" "a2\0\0", 12), p.begin().ToString());
+ EXPECT_EQ(string("\0\0\0\2" "\0\0\0\1" "a3\0\0", 12), p.end().ToString());
}
{
const auto& p = partitions[9];
ASSERT_EQ(0, p.hash_buckets().size());
- EXPECT_EQ(string("a4\0\0", 4), p.range_key_start());
- EXPECT_EQ("", p.range_key_end());
- EXPECT_EQ(string("a4\0\0", 4), p.partition_key_start());
- EXPECT_EQ("", p.partition_key_end());
+ EXPECT_EQ(string("a4\0\0", 4), p.begin().range_key());
+ EXPECT_EQ("", p.end().range_key());
+ EXPECT_EQ(string("a4\0\0", 4), p.begin().ToString());
+ EXPECT_EQ("", p.end().ToString());
}
}
diff --git a/src/kudu/common/partition.cc b/src/kudu/common/partition.cc
index 1925b87..cad1ab7 100644
--- a/src/kudu/common/partition.cc
+++ b/src/kudu/common/partition.cc
@@ -19,6 +19,8 @@
#include <algorithm>
#include <cstring>
+#include <iomanip>
+#include <iostream>
#include <iterator>
#include <memory>
#include <set>
@@ -65,32 +67,14 @@ class faststring;
// The encoded size of a hash bucket in a partition key.
static const size_t kEncodedBucketSize = sizeof(uint32_t);
-Slice Partition::range_key_start() const {
- return range_key(partition_key_start_);
-}
-
-Slice Partition::range_key_end() const {
- return range_key(partition_key_end_);
-}
-
-Slice Partition::range_key(const string& partition_key) const {
- const size_t hash_size = kEncodedBucketSize * hash_buckets_.size();
- if (partition_key.size() <= hash_size) {
- return Slice();
- }
- auto s = Slice(partition_key);
- s.remove_prefix(hash_size);
- return s;
-}
-
bool Partition::operator==(const Partition& rhs) const {
if (this == &rhs) {
return true;
}
- if (partition_key_start() != rhs.partition_key_start()) {
+ if (begin_ != rhs.begin_) {
return false;
}
- if (partition_key_end() != rhs.partition_key_end()) {
+ if (end_ != rhs.end_) {
return false;
}
if (hash_buckets_ != rhs.hash_buckets_) {
@@ -105,8 +89,8 @@ void Partition::ToPB(PartitionPB* pb) const {
for (int32_t bucket : hash_buckets_) {
pb->add_hash_buckets(bucket);
}
- pb->set_partition_key_start(partition_key_start());
- pb->set_partition_key_end(partition_key_end());
+ pb->set_partition_key_start(begin().ToString());
+ pb->set_partition_key_end(end().ToString());
}
void Partition::FromPB(const PartitionPB& pb, Partition* partition) {
@@ -116,8 +100,48 @@ void Partition::FromPB(const PartitionPB& pb, Partition* partition) {
partition->hash_buckets_.push_back(hash_bucket);
}
- partition->partition_key_start_ = pb.partition_key_start();
- partition->partition_key_end_ = pb.partition_key_end();
+ const size_t hash_buckets_num = partition->hash_buckets_.size();
+ partition->begin_ = StringToPartitionKey(
+ pb.partition_key_start(), hash_buckets_num);
+ partition->end_ = StringToPartitionKey(
+ pb.partition_key_end(), hash_buckets_num);
+ // Check the invariant: valid partition cannot span over multiple sets
+ // of hash buckets.
+ //DCHECK(partition->begin_.empty() || partition->end_.empty() ||
+ // partition->begin_.hash_key() == partition->end_.hash_key());
+}
+
+PartitionKey Partition::StringToPartitionKey(const std::string& key_str,
+ size_t hash_dimensions_num) {
+ if (hash_dimensions_num == 0) {
+ return PartitionKey("", key_str);
+ }
+ const size_t hash_part_size = kEncodedBucketSize * hash_dimensions_num;
+ if (key_str.size() <= hash_part_size) {
+ return PartitionKey(key_str, "");
+ }
+ return PartitionKey(key_str.substr(0, hash_part_size),
+ key_str.substr(hash_part_size));
+}
+
+string PartitionKey::DebugString() const {
+ std::ostringstream ss;
+ ss << "h:";
+ for (size_t i = 0; i < hash_key_.size(); ++i) {
+ ss << std::hex << std::setw(2) << std::setfill('0')
+ << static_cast<uint16_t>(hash_key_[i]);
+ }
+ ss << " r:";
+ for (size_t i = 0; i < range_key_.size(); ++i) {
+ ss << std::hex << std::setw(2) << std::setfill('0')
+ << static_cast<uint16_t>(range_key_[i]);
+ }
+ return ss.str();
+}
+
+std::ostream& operator<<(std::ostream& out, const PartitionKey& key) {
+ out << key.DebugString();
+ return out;
}
namespace {
@@ -319,30 +343,37 @@ Status PartitionSchema::ToPB(const Schema& schema, PartitionSchemaPB* pb) const
}
template<typename Row>
-void PartitionSchema::EncodeKeyImpl(const Row& row, string* buf) const {
+void PartitionSchema::EncodeKeyImpl(const Row& row,
+ string* range_buf,
+ string* hash_buf) const {
+ DCHECK(range_buf);
+ DCHECK(hash_buf);
string range_key;
EncodeColumns(row, range_schema_.column_ids, &range_key);
+
const auto& hash_schema = GetHashSchemaForRange(range_key);
const auto& hash_encoder = GetKeyEncoder<string>(GetTypeInfo(UINT32));
+ string hash_key;
for (const auto& hash_dimension : hash_schema) {
const auto bucket = HashValueForRow(row, hash_dimension);
- hash_encoder.Encode(&bucket, buf);
+ hash_encoder.Encode(&bucket, &hash_key);
}
- // The range portion of the key has been already encoded -- append it to the
- // result buffer.
- buf->append(range_key);
+ *range_buf = std::move(range_key);
+ *hash_buf = std::move(hash_key);
}
-string PartitionSchema::EncodeKey(const KuduPartialRow& row) const {
- string encoded_key;
- EncodeKeyImpl(row, &encoded_key);
- return encoded_key;
+PartitionKey PartitionSchema::EncodeKey(const KuduPartialRow& row) const {
+ string encoded_range;
+ string encoded_hash;
+ EncodeKeyImpl(row, &encoded_range, &encoded_hash);
+ return PartitionKey(std::move(encoded_hash), std::move(encoded_range));
}
-string PartitionSchema::EncodeKey(const ConstContiguousRow& row) const {
- string encoded_key;
- EncodeKeyImpl(row, &encoded_key);
- return encoded_key;
+PartitionKey PartitionSchema::EncodeKey(const ConstContiguousRow& row) const {
+ string encoded_range;
+ string encoded_hash;
+ EncodeKeyImpl(row, &encoded_range, &encoded_hash);
+ return PartitionKey(std::move(encoded_hash), std::move(encoded_range));
}
Status PartitionSchema::EncodeRangeKey(const KuduPartialRow& row,
@@ -574,10 +605,10 @@ Status PartitionSchema::CreatePartitions(
vector<Partition> new_partitions;
for (const Partition& base_partition : base_hash_partitions) {
for (const auto& bound : bounds_with_hash_schemas) {
- Partition partition = base_partition;
- partition.partition_key_start_.append(bound.lower);
- partition.partition_key_end_.append(bound.upper);
- new_partitions.emplace_back(std::move(partition));
+ Partition p = base_partition;
+ p.begin_.mutable_range_key()->append(bound.lower);
+ p.end_.mutable_range_key()->append(bound.upper);
+ new_partitions.emplace_back(std::move(p));
}
}
*partitions = std::move(new_partitions);
@@ -593,9 +624,11 @@ Status PartitionSchema::CreatePartitions(
vector<Partition> current_bound_hash_partitions = GenerateHashPartitions(
bound.hash_schema, hash_encoder);
// Add range information to the partition key.
- for (Partition& partition : current_bound_hash_partitions) {
- partition.partition_key_start_.append(bound.lower);
- partition.partition_key_end_.append(bound.upper);
+ for (Partition& p : current_bound_hash_partitions) {
+ DCHECK(p.begin_.range_key().empty()) << p.begin_.DebugString();
+ p.begin_.mutable_range_key()->assign(bound.lower);
+ DCHECK(p.end().range_key().empty());
+ p.end_.mutable_range_key()->assign(bound.upper);
partition_idx_to_hash_schema_idx.emplace_back(i);
}
result_partitions.insert(result_partitions.end(),
@@ -624,16 +657,20 @@ Status PartitionSchema::CreatePartitions(
// partition key beginning at the hash component. For a concrete example,
// see PartitionTest::TestCreatePartitions.
for (size_t partition_idx = 0; partition_idx < partitions->size(); ++partition_idx) {
- Partition& partition = (*partitions)[partition_idx];
- const int hash_buckets_num = static_cast<int>(partition.hash_buckets().size());
- // Find the first zero-valued bucket from the end and truncate the partition key
- // starting from that bucket onwards for zero-valued buckets.
- if (partition.range_key_start().empty()) {
+ auto& p = (*partitions)[partition_idx];
+ const auto hash_buckets_num = static_cast<int>(p.hash_buckets().size());
+ // Find the first nonzero-valued bucket from the end and truncate the
+ // partition key starting from that bucket onwards for zero-valued buckets.
+ //
+ // TODO(aserbin): is this really necessary -- zeros in hash key are the
+ // minimum possible values, and the range part is already
+ // empty?
+ if (p.begin().range_key().empty()) {
for (int i = hash_buckets_num - 1; i >= 0; --i) {
- if (partition.hash_buckets()[i] != 0) {
+ if (p.hash_buckets()[i] != 0) {
break;
}
- partition.partition_key_start_.erase(kEncodedBucketSize * i);
+ p.begin_.mutable_hash_key()->erase(kEncodedBucketSize * i);
}
}
// Starting from the last hash bucket, truncate the partition key until we hit the first
@@ -641,19 +678,19 @@ Status PartitionSchema::CreatePartitions(
// bucket value. For example, the following range end partition keys should be transformed,
// where the key is (hash_comp1, hash_comp2, range_comp):
//
- // [ (0, 0, "a2b2") -> (0, 1, "a2b2") ]
- // [ (0, 1, "a2b2") -> (1, _, "a2b2") ]
- // [ (1, 0, "a2b2") -> (1, 1, "a2b2") ]
- // [ (1, 1, "a2b2") -> (_, _, "a2b2") ]
- if (partition.range_key_end().empty()) {
+ // [ (0, 0, "") -> (0, 1, "") ]
+ // [ (0, 1, "") -> (1, _, "") ]
+ // [ (1, 0, "") -> (1, 1, "") ]
+ // [ (1, 1, "") -> (_, _, "") ]
+ if (p.end().range_key().empty()) {
const auto& hash_schema = range_hash_schemas.empty()
? hash_schema_
: bounds_with_hash_schemas[partition_idx_to_hash_schema_idx[partition_idx]].hash_schema;
for (int i = hash_buckets_num - 1; i >= 0; --i) {
- partition.partition_key_end_.erase(kEncodedBucketSize * i);
- const int32_t hash_bucket = partition.hash_buckets()[i] + 1;
+ p.end_.mutable_hash_key()->erase(kEncodedBucketSize * i);
+ const int32_t hash_bucket = p.hash_buckets()[i] + 1;
if (hash_bucket != hash_schema[i].num_buckets) {
- hash_encoder.Encode(&hash_bucket, &partition.partition_key_end_);
+ hash_encoder.Encode(&hash_bucket, p.end_.mutable_hash_key());
break;
}
}
@@ -697,16 +734,15 @@ bool PartitionSchema::RangePartitionContainsRowImpl(
const Partition& partition, const Row& row) const {
// If range partition is not used, column_ids would be empty and
// EncodedColumn() would return immediately.
- string range_partition_key;
- EncodeColumns(row, range_schema_.column_ids, &range_partition_key);
+ string key;
+ EncodeColumns(row, range_schema_.column_ids, &key);
// When all hash buckets match, then the row is contained in the partition
// if the row's key is greater or equal to the lower bound, and if there is
// either no upper bound or the row's key is less than the upper bound.
- const auto key = Slice(range_partition_key);
return
- (partition.range_key_start() <= key) &&
- (partition.range_key_end().empty() || key < partition.range_key_end());
+ (partition.begin().range_key() <= key) &&
+ (partition.end().range_key().empty() || key < partition.end().range_key());
}
bool PartitionSchema::PartitionContainsRow(const Partition& partition,
@@ -879,14 +915,16 @@ string PartitionSchema::PartitionDebugString(const Partition& partition,
// Partitions are considered metadata, so don't redact them.
ScopedDisableRedaction no_redaction;
- vector<string> components;
- if (partition.hash_buckets_.size() != hash_schema_.size()) {
- return "<hash-partition-error>";
+ const auto& hash_schema = GetHashSchemaForRange(partition.begin().range_key());
+ if (partition.hash_buckets_.size() != hash_schema.size()) {
+ return Substitute("<hash-partition-error: $0 vs $1 expected buckets>",
+ partition.hash_buckets_.size(), hash_schema.size());
}
- for (size_t i = 0; i < hash_schema_.size(); ++i) {
+ vector<string> components;
+ for (size_t i = 0; i < hash_schema.size(); ++i) {
string s = Substitute("HASH ($0) PARTITION $1",
- ColumnIdsToColumnNames(schema, hash_schema_[i].column_ids),
+ ColumnIdsToColumnNames(schema, hash_schema[i].column_ids),
partition.hash_buckets_[i]);
components.emplace_back(std::move(s));
}
@@ -894,8 +932,8 @@ string PartitionSchema::PartitionDebugString(const Partition& partition,
if (!range_schema_.column_ids.empty()) {
string s = Substitute("RANGE ($0) PARTITION $1",
ColumnIdsToColumnNames(schema, range_schema_.column_ids),
- RangePartitionDebugString(partition.range_key_start(),
- partition.range_key_end(),
+ RangePartitionDebugString(partition.begin().range_key(),
+ partition.end().range_key(),
schema));
components.emplace_back(std::move(s));
}
@@ -905,10 +943,12 @@ string PartitionSchema::PartitionDebugString(const Partition& partition,
template<typename Row>
string PartitionSchema::PartitionKeyDebugStringImpl(const Row& row) const {
- // TODO(aserbin): update this to adapt to custom hash schemas per range
+ string range_key;
+ EncodeColumns(row, range_schema_.column_ids, &range_key);
+ const auto& hash_schema = GetHashSchemaForRange(range_key);
vector<string> components;
- components.reserve(hash_schema_.size() + range_schema_.column_ids.size());
- for (const auto& hash_dimension : hash_schema_) {
+ components.reserve(hash_schema.size() + range_schema_.column_ids.size());
+ for (const auto& hash_dimension : hash_schema) {
components.emplace_back(Substitute(
"HASH ($0): $1",
ColumnIdsToColumnNames(*row.schema(), hash_dimension.column_ids),
@@ -937,55 +977,34 @@ string PartitionSchema::PartitionKeyDebugString(const KuduPartialRow& row) const
return PartitionKeyDebugStringImpl(row);
}
-string PartitionSchema::PartitionKeyDebugString(Slice key, const Schema& schema) const {
- // Since it's not known what's the right hash bucket schema for the key
- // (there might be custom hash bucket schemas per range), let's first find
- // what's the size of the range key and remove it as the trailing part
- // of the key. The rest is the hash partition key.
- //
- // TODO(aserbin): is there a better way to do this?
- const bool has_ranges = !range_schema_.column_ids.empty();
- string range_key_size_meter;
- if (has_ranges) {
- KuduPartialRow row(&schema);
- EncodeColumns(row, range_schema_.column_ids, &range_key_size_meter);
- }
-
- if (has_ranges) {
- if (key.size() < range_key_size_meter.size()) {
- return "<range-decode-error>";
- }
- } else {
- if (key.size() < kEncodedBucketSize * hash_schema_.size()) {
- return "<hash-decode-error>";
- }
- }
-
- Slice range_key(key);
- if (has_ranges) {
- DCHECK_GE(key.size(), range_key_size_meter.size());
- range_key.remove_prefix(key.size() - range_key_size_meter.size());
+string PartitionSchema::PartitionKeyDebugString(const PartitionKey& key,
+ const Schema& schema) const {
+ const auto& hash_schema = GetHashSchemaForRange(key.range_key());
+ Slice hash_key(key.hash_key());
+ if (hash_key.size() != kEncodedBucketSize * hash_schema.size()) {
+ return "<hash-decode-error>";
}
- const auto& hash_schemas = has_ranges
- ? GetHashSchemaForRange(range_key.ToString()) : hash_schema_;
vector<string> components;
- components.reserve(hash_schemas.size() + 1);
- for (const auto& hash_schema : hash_schemas) {
+ components.reserve(hash_schema.size() + 1);
+ for (const auto& hash_dimension : hash_schema) {
uint32_t big_endian;
- memcpy(&big_endian, key.data(), sizeof(uint32_t));
- key.remove_prefix(sizeof(uint32_t));
+ memcpy(&big_endian, hash_key.data(), kEncodedBucketSize);
+ hash_key.remove_prefix(kEncodedBucketSize);
components.emplace_back(
Substitute("HASH ($0): $1",
- ColumnIdsToColumnNames(schema, hash_schema.column_ids),
+ ColumnIdsToColumnNames(schema, hash_dimension.column_ids),
BigEndian::ToHost32(big_endian)));
}
+ DCHECK(hash_key.empty());
+ const bool has_ranges = !range_schema_.column_ids.empty();
if (has_ranges) {
- components.emplace_back(
- Substitute("RANGE ($0): $1",
- ColumnIdsToColumnNames(schema, range_schema_.column_ids),
- RangeKeyDebugString(key, schema)));
+ Slice range_key(key.range_key());
+ components.emplace_back(
+ Substitute("RANGE ($0): $1",
+ ColumnIdsToColumnNames(schema, range_schema_.column_ids),
+ RangeKeyDebugString(range_key, schema)));
}
return JoinStrings(components, ", ");
@@ -1155,8 +1174,8 @@ string PartitionSchema::PartitionTableEntry(const Schema& schema,
if (!range_schema_.column_ids.empty()) {
SubstituteAndAppend(&entry, "<td>$0</td>",
EscapeForHtmlToString(
- RangePartitionDebugString(partition.range_key_start(),
- partition.range_key_end(),
+ RangePartitionDebugString(partition.begin().range_key(),
+ partition.end().range_key(),
schema)));
}
return entry;
@@ -1264,8 +1283,10 @@ vector<Partition> PartitionSchema::GenerateHashPartitions(
for (uint32_t bucket = 0; bucket < hash_dimension.num_buckets; ++bucket) {
Partition partition = base_partition;
partition.hash_buckets_.push_back(bucket);
- hash_encoder.Encode(&bucket, &partition.partition_key_start_);
- hash_encoder.Encode(&bucket, &partition.partition_key_end_);
+
+ hash_encoder.Encode(&bucket, partition.begin_.mutable_hash_key());
+ hash_encoder.Encode(&bucket, partition.end_.mutable_hash_key());
+
new_partitions.emplace_back(std::move(partition));
}
}
diff --git a/src/kudu/common/partition.h b/src/kudu/common/partition.h
index 61e1a3e..f43b0d4 100644
--- a/src/kudu/common/partition.h
+++ b/src/kudu/common/partition.h
@@ -18,6 +18,7 @@
#include <cstddef>
#include <cstdint>
+#include <iosfwd>
#include <map>
#include <string>
#include <utility>
@@ -46,6 +47,82 @@ class PartitionSchemaPB;
class PartitionSchemaPB_HashBucketSchemaPB;
template <typename Buffer> class KeyEncoder;
+// This class is a representation of a table's row (or a row limit) according
+// to the table's partition schema. A table's partition can be defined by a pair
+// of such PartitionKey objects: the beginning and the end of a partition. Rows
+// that belong to a given partition all fall within the same partition key
+// range, i.e. they are between the beginning and the end of the partition
+// defined by two PartitionKey objects. Essentially, there is a way to serialize
+// a subset of primary keys columns in a table's row given the table's partition
+// schema, and PartitionKey consists of the hash and the range parts of that
+// serialized representation.
+class PartitionKey {
+ public:
+ // Build a PartitionKey that represents infinity. For the beginning of a
+ // partition represents '-inf'; for the end of a partition represents '+inf'.
+ PartitionKey() = default;
+
+ // Build PartitionKey object given hash and range parts.
+ PartitionKey(std::string hash_key, std::string range_key)
+ : hash_key_(std::move(hash_key)),
+ range_key_(std::move(range_key)) {
+ }
+
+ PartitionKey& operator=(const PartitionKey& other) {
+ hash_key_ = other.hash_key_;
+ range_key_ = other.range_key_;
+ return *this;
+ }
+
+ bool operator==(const PartitionKey& other) const {
+ return hash_key_ == other.hash_key_ && range_key_ == other.range_key_;
+ }
+
+ bool operator!=(const PartitionKey& other) const {
+ return !(*this == other);
+ }
+
+ bool operator<(const PartitionKey& other) const {
+ return ToString() < other.ToString();
+ }
+
+ bool operator<=(const PartitionKey& other) const {
+ return !(*this > other);
+ }
+
+ bool operator>(const PartitionKey& other) const {
+ return other < *this;
+ }
+
+ bool operator>=(const PartitionKey& other) const {
+ return !(*this < other);
+ }
+
+ bool empty() const {
+ return hash_key_.empty() && range_key_.empty();
+ }
+
+ const std::string& hash_key() const { return hash_key_; }
+ const std::string& range_key() const { return range_key_; }
+
+ std::string* mutable_hash_key() { return &hash_key_; }
+ std::string* mutable_range_key() { return &range_key_; }
+
+ // The serialized representation of this partition key. In legacy Kudu
+ // versions, this string representation was used as a partition key.
+ std::string ToString() const {
+ return hash_key_ + range_key_;
+ }
+
+ // Return a string with hexadecimal representation of the data comprising
+ // the key: useful for representing the key in human-readable format.
+ std::string DebugString() const;
+
+ private:
+ std::string hash_key_; // the hash part of the key, encoded
+ std::string range_key_; // the range part of the key, encoded
+};
+
// A Partition describes the set of rows that a Tablet is responsible for
// serving. Each tablet is assigned a single Partition.
//
@@ -62,16 +139,12 @@ class Partition {
return hash_buckets_;
}
- Slice range_key_start() const;
-
- Slice range_key_end() const;
-
- const std::string& partition_key_start() const {
- return partition_key_start_;
+ const PartitionKey& begin() const {
+ return begin_;
}
- const std::string& partition_key_end() const {
- return partition_key_end_;
+ const PartitionKey& end() const {
+ return end_;
}
// Returns true iff the given partition 'rhs' is equivalent to this one.
@@ -89,18 +162,23 @@ class Partition {
// to be created by the master process.
static void FromPB(const PartitionPB& pb, Partition* partition);
+ // Knowing the number of hash bucketing dimensions, it's possible to separate
+ // the encoded hash-related part from the range-related part, so it's possible
+ // to build corresponding PartitionKey object from a string where the hash-
+ // and the range-related parts are concatenated.
+ static PartitionKey StringToPartitionKey(const std::string& key_str,
+ size_t hash_dimensions_num);
private:
friend class PartitionSchema;
- // Helper function for accessing the range key portion of a partition key.
- Slice range_key(const std::string& partition_key) const;
-
std::vector<int32_t> hash_buckets_;
- std::string partition_key_start_;
- std::string partition_key_end_;
+ PartitionKey begin_;
+ PartitionKey end_;
};
+std::ostream& operator<<(std::ostream& out, const PartitionKey& key);
+
// A partition schema describes how the rows of a table are distributed among
// tablets.
//
@@ -225,10 +303,9 @@ class PartitionSchema {
// Requires a schema to encode the range bounds.
Status ToPB(const Schema& schema, PartitionSchemaPB* pb) const;
- // Appends the row's encoded partition key into the provided buffer.
- // On failure, the buffer may have data partially appended.
- std::string EncodeKey(const KuduPartialRow& row) const;
- std::string EncodeKey(const ConstContiguousRow& row) const;
+ // Returns partition key for the row.
+ PartitionKey EncodeKey(const KuduPartialRow& row) const;
+ PartitionKey EncodeKey(const ConstContiguousRow& row) const;
// Creates the set of table partitions for a partition schema and collection
// of split rows and split bounds.
@@ -284,7 +361,8 @@ class PartitionSchema {
std::string PartitionDebugString(const Partition& partition, const Schema& schema) const;
// Returns a text description of a partition key suitable for debug printing.
- std::string PartitionKeyDebugString(Slice key, const Schema& schema) const;
+ std::string PartitionKeyDebugString(const PartitionKey& key,
+ const Schema& schema) const;
std::string PartitionKeyDebugString(const KuduPartialRow& row) const;
std::string PartitionKeyDebugString(const ConstContiguousRow& row) const;
@@ -451,7 +529,9 @@ class PartitionSchema {
// Private templated helper for EncodeKey.
template<typename Row>
- void EncodeKeyImpl(const Row& row, std::string* buf) const;
+ void EncodeKeyImpl(const Row& row,
+ std::string* range_buf,
+ std::string* hash_buf) const;
// Returns true if all of the columns in the range partition key are unset in
// the row.
diff --git a/src/kudu/common/partition_pruner-test.cc b/src/kudu/common/partition_pruner-test.cc
index 2005f22..09bcff8 100644
--- a/src/kudu/common/partition_pruner-test.cc
+++ b/src/kudu/common/partition_pruner-test.cc
@@ -232,8 +232,8 @@ TEST_F(PartitionPrunerTest, TestPrimaryKeyRangePruning) {
spec.SetExclusiveUpperBoundKey(enc_upper_bound);
}
size_t pruner_ranges = remaining_tablets == 0 ? 0 : 1;
- NO_FATALS(CheckPrunedPartitions(schema, partition_schema, partitions, spec,
- remaining_tablets, pruner_ranges));
+ CheckPrunedPartitions(schema, partition_schema, partitions, spec,
+ remaining_tablets, pruner_ranges);
};
// No bounds
@@ -350,8 +350,8 @@ TEST_F(PartitionPrunerTest, TestPartialPrimaryKeyRangePruning) {
spec.SetExclusiveUpperBoundKey(enc_upper_bound);
}
size_t pruner_ranges = remaining_tablets == 0 ? 0 : 1;
- NO_FATALS(CheckPrunedPartitions(schema, partition_schema, partitions, spec,
- remaining_tablets, pruner_ranges));
+ CheckPrunedPartitions(schema, partition_schema, partitions, spec,
+ remaining_tablets, pruner_ranges);
};
// No bounds
@@ -459,8 +459,8 @@ TEST_F(PartitionPrunerTest, TestIntPartialPrimaryKeyRangePruning) {
spec.SetExclusiveUpperBoundKey(enc_upper_bound);
}
size_t pruner_ranges = remaining_tablets == 0 ? 0 : 1;
- NO_FATALS(CheckPrunedPartitions(schema, partition_schema, partitions, spec,
- remaining_tablets, pruner_ranges));
+ CheckPrunedPartitions(schema, partition_schema, partitions, spec,
+ remaining_tablets, pruner_ranges);
};
// No bounds
@@ -524,14 +524,12 @@ TEST_F(PartitionPrunerTest, TestRangePruning) {
// number of partitions are pruned.
const auto check = [&] (const vector<ColumnPredicate>& predicates, size_t remaining_tablets) {
ScanSpec spec;
-
for (const auto& pred : predicates) {
spec.AddPredicate(pred);
}
-
size_t pruner_ranges = remaining_tablets == 0 ? 0 : 1;
- NO_FATALS(CheckPrunedPartitions(schema, partition_schema, partitions, spec,
- remaining_tablets, pruner_ranges));
+ CheckPrunedPartitions(schema, partition_schema, partitions, spec,
+ remaining_tablets, pruner_ranges);
};
constexpr int8_t neg_ten = -10;
@@ -699,13 +697,11 @@ TEST_F(PartitionPrunerTest, TestHashPruning) {
size_t remaining_tablets,
size_t pruner_ranges) {
ScanSpec spec;
-
for (const auto& pred : predicates) {
spec.AddPredicate(pred);
}
-
- NO_FATALS(CheckPrunedPartitions(schema, partition_schema, partitions, spec,
- remaining_tablets, pruner_ranges));
+ CheckPrunedPartitions(schema, partition_schema, partitions, spec,
+ remaining_tablets, pruner_ranges);
};
constexpr int8_t zero = 0;
@@ -777,13 +773,11 @@ TEST_F(PartitionPrunerTest, TestInListHashPruning) {
size_t remaining_tablets,
size_t pruner_ranges) {
ScanSpec spec;
-
for (const auto& pred : predicates) {
spec.AddPredicate(pred);
}
-
- NO_FATALS(CheckPrunedPartitions(schema, partition_schema, partitions, spec,
- remaining_tablets, pruner_ranges));
+ CheckPrunedPartitions(schema, partition_schema, partitions, spec,
+ remaining_tablets, pruner_ranges);
};
// zero, one, eight are in different buckets when bucket number is 3 and seed is 0.
@@ -857,13 +851,11 @@ TEST_F(PartitionPrunerTest, TestMultiColumnInListHashPruning) {
size_t remaining_tablets,
size_t pruner_ranges) {
ScanSpec spec;
-
for (const auto& pred : predicates) {
spec.AddPredicate(pred);
}
-
- NO_FATALS(CheckPrunedPartitions(schema, partition_schema, partitions, spec,
- remaining_tablets, pruner_ranges));
+ CheckPrunedPartitions(schema, partition_schema, partitions, spec,
+ remaining_tablets, pruner_ranges);
};
// zero, one, eight are in different buckets when bucket number is 3 and seed is 0.
@@ -947,27 +939,25 @@ TEST_F(PartitionPrunerTest, TestPruning) {
ASSERT_OK(split.SetUnixTimeMicros("time", 10));
vector<Partition> partitions;
- ASSERT_OK(partition_schema.CreatePartitions(vector<KuduPartialRow>{ split }, {}, {},
- schema, &partitions));
+ ASSERT_OK(partition_schema.CreatePartitions(
+ vector<KuduPartialRow>{ split }, {}, {}, schema, &partitions));
ASSERT_EQ(4, partitions.size());
// Applies the specified predicates to a scan and checks that the expected
// number of partitions are pruned.
const auto check = [&] (const vector<ColumnPredicate>& predicates,
- const string& lower_bound_partition_key,
- const string& upper_bound_partition_key,
+ const PartitionKey& lower_bound_partition_key,
+ const PartitionKey& upper_bound_partition_key,
size_t remaining_tablets,
size_t pruner_ranges) {
ScanSpec spec;
-
spec.SetLowerBoundPartitionKey(lower_bound_partition_key);
spec.SetExclusiveUpperBoundPartitionKey(upper_bound_partition_key);
for (const auto& pred : predicates) {
spec.AddPredicate(pred);
}
-
- NO_FATALS(CheckPrunedPartitions(schema, partition_schema, partitions, spec,
- remaining_tablets, pruner_ranges));
+ CheckPrunedPartitions(schema, partition_schema, partitions, spec,
+ remaining_tablets, pruner_ranges);
};
Slice a = "a";
@@ -982,7 +972,7 @@ TEST_F(PartitionPrunerTest, TestPruning) {
NO_FATALS(check({ ColumnPredicate::Equality(schema.column(0), &a),
ColumnPredicate::Equality(schema.column(1), &a),
ColumnPredicate::Range(schema.column(2), &nine, nullptr) },
- "", "",
+ {}, {},
2, 1));
// host = "a"
@@ -992,7 +982,7 @@ TEST_F(PartitionPrunerTest, TestPruning) {
NO_FATALS(check({ ColumnPredicate::Equality(schema.column(0), &a),
ColumnPredicate::Equality(schema.column(1), &a),
ColumnPredicate::Range(schema.column(2), &ten, &twenty) },
- "", "",
+ {}, {},
1, 1));
// host = "a"
@@ -1001,7 +991,7 @@ TEST_F(PartitionPrunerTest, TestPruning) {
NO_FATALS(check({ ColumnPredicate::Equality(schema.column(0), &a),
ColumnPredicate::Equality(schema.column(1), &a),
ColumnPredicate::Range(schema.column(2), nullptr, &ten) },
- "", "",
+ {}, {},
1, 1));
// host = "a"
@@ -1010,7 +1000,7 @@ TEST_F(PartitionPrunerTest, TestPruning) {
NO_FATALS(check({ ColumnPredicate::Equality(schema.column(0), &a),
ColumnPredicate::Equality(schema.column(1), &a),
ColumnPredicate::Range(schema.column(2), &ten, nullptr) },
- "", "",
+ {}, {},
1, 1));
// host = "a"
@@ -1019,24 +1009,24 @@ TEST_F(PartitionPrunerTest, TestPruning) {
NO_FATALS(check({ ColumnPredicate::Equality(schema.column(0), &a),
ColumnPredicate::Equality(schema.column(1), &a),
ColumnPredicate::Equality(schema.column(2), &ten) },
- "", "",
+ {}, {},
1, 1));
// partition key < (hash=1)
- NO_FATALS(check({}, "", string("\0\0\0\1", 4), 2, 1));
+ NO_FATALS(check({}, {}, { string("\0\0\0\1", 4), "" }, 2, 1));
// partition key >= (hash=1)
- NO_FATALS(check({}, string("\0\0\0\1", 4), "", 2, 1));
+ NO_FATALS(check({}, { string("\0\0\0\1", 4), "" }, {}, 2, 1));
// timestamp = 10
// partition key < (hash=1)
NO_FATALS(check({ ColumnPredicate::Equality(schema.column(2), &ten) },
- "", string("\0\0\0\1", 4), 1, 1));
+ {}, { string("\0\0\0\1", 4), "" }, 1, 1));
// timestamp = 10
// partition key >= (hash=1)
NO_FATALS(check({ ColumnPredicate::Equality(schema.column(2), &ten) },
- string("\0\0\0\1", 4), "", 1, 1));
+ { string("\0\0\0\1", 4), "" }, {}, 1, 1));
}
TEST_F(PartitionPrunerTest, TestKudu2173) {
@@ -1063,13 +1053,12 @@ TEST_F(PartitionPrunerTest, TestKudu2173) {
// number of partitions are pruned.
const auto check = [&] (const vector<ColumnPredicate>& predicates, size_t remaining_tablets) {
ScanSpec spec;
-
for (const auto& pred : predicates) {
spec.AddPredicate(pred);
}
size_t pruner_ranges = remaining_tablets == 0 ? 0 : 1;
- NO_FATALS(CheckPrunedPartitions(schema, partition_schema, partitions, spec,
- remaining_tablets, pruner_ranges));
+ CheckPrunedPartitions(schema, partition_schema, partitions, spec,
+ remaining_tablets, pruner_ranges);
};
constexpr int8_t eleven = 11;
@@ -1146,8 +1135,8 @@ TEST_F(PartitionPrunerTest, TestHashSchemasPerRangePruning) {
// Applies the specified predicates to a scan and checks that the expected
// number of partitions are pruned.
const auto check = [&] (const vector<ColumnPredicate>& predicates,
- const string& lower_bound_partition_key,
- const string& upper_bound_partition_key,
+ const PartitionKey& lower_bound_partition_key,
+ const PartitionKey& upper_bound_partition_key,
size_t remaining_tablets,
size_t pruner_ranges) {
ScanSpec spec;
@@ -1172,97 +1161,118 @@ TEST_F(PartitionPrunerTest, TestHashSchemasPerRangePruning) {
const Slice m = "m";
// No Bounds
- NO_FATALS(check({}, "", "", 12, 12));
+ NO_FATALS(check({}, {}, {}, 12, 12));
// A = 1
NO_FATALS(check({ ColumnPredicate::Equality(schema.column(0), &one) },
- "", "", 7, 7));
+ {}, {}, 7, 7));
// B = 1
NO_FATALS(check({ ColumnPredicate::Equality(schema.column(1), &one) },
- "", "", 7, 7));
+ {}, {}, 7, 7));
// A = 0
// B = 1
// C >= "e"
NO_FATALS(check({ ColumnPredicate::Equality(schema.column(0), &zero),
ColumnPredicate::Equality(schema.column(1), &one),
ColumnPredicate::Range(schema.column(2), &e, nullptr) },
- "", "", 3, 3));
+ {}, {}, 3, 3));
// A = 0
// B = 1
// C = "e"
NO_FATALS(check({ ColumnPredicate::Equality(schema.column(0), &zero),
ColumnPredicate::Equality(schema.column(1), &one),
ColumnPredicate::Equality(schema.column(2), &e) },
- "", "", 1, 1));
+ {}, {}, 1, 1));
// B = 1
// C >= "b"
// C < "j"
NO_FATALS(check({ ColumnPredicate::Equality(schema.column(1), &one),
ColumnPredicate::Range(schema.column(2), &b, nullptr),
ColumnPredicate::Range(schema.column(2), nullptr, &i) },
- "", "", 6, 6));
+ {}, {}, 6, 6));
// B = 0
// C >= "e"
// C < "l"
NO_FATALS(check({ ColumnPredicate::Equality(schema.column(1), &zero),
ColumnPredicate::Range(schema.column(2), &e, nullptr),
ColumnPredicate::Range(schema.column(2), nullptr, &l) },
- "", "", 4, 4));
+ {}, {}, 4, 4));
// C >= "a"
// C < "b"
NO_FATALS(check({ ColumnPredicate::Range(schema.column(2), &a, &b) },
- "", "", 3, 3));
+ {}, {}, 3, 3));
// C >= "a"
// C < "e"
NO_FATALS(check({ ColumnPredicate::Range(schema.column(2), &a, &e) },
- "", "", 9, 9));
+ {}, {}, 9, 9));
// C >= "e"
// C < "i"
NO_FATALS(check({ ColumnPredicate::Range(schema.column(2), &e, &i) },
- "", "", 7, 7));
+ {}, {}, 7, 7));
// C >= "a"
// C < "l"
NO_FATALS(check({ ColumnPredicate::Range(schema.column(2), &a, &l) },
- "", "", 12, 12));
+ {}, {}, 12, 12));
// C >= "i"
// C < "l"
NO_FATALS(check({ ColumnPredicate::Range(schema.column(2), &i, &l) },
- "", "", 3, 3));
+ {}, {}, 3, 3));
// C >= "e"
NO_FATALS(check({ ColumnPredicate::Range(schema.column(2), &e, nullptr) },
- "", "", 9, 9));
+ {}, {}, 9, 9));
// C < "f"
NO_FATALS(check({ ColumnPredicate::Range(schema.column(2), nullptr, &f) },
- "", "", 9, 9));
+ {}, {}, 9, 9));
// C >= "f"
NO_FATALS(check({ ColumnPredicate::Range(schema.column(2), &f, nullptr) },
- "", "", 3, 3));
+ {}, {}, 3, 3));
// C < "a"
NO_FATALS(check({ ColumnPredicate::Range(schema.column(2), nullptr, &a) },
- "", "", 0, 0));
+ {}, {}, 0, 0));
// C >= "m"
NO_FATALS(check({ ColumnPredicate::Range(schema.column(2), &m, nullptr) },
- "", "", 0, 0));
+ {}, {}, 0, 0));
// Uses None predicate to short circuit scan
- NO_FATALS(check({ ColumnPredicate::None(schema.column(2))}, "", "", 0, 0));
+ NO_FATALS(check({ ColumnPredicate::None(schema.column(2))}, {}, {}, 0, 0));
// partition key >= (hash=1, hash=0)
- NO_FATALS(check({}, string("\0\0\0\1\0\0\0\0", 8), "", 7, 7));
+ NO_FATALS(check({}, {string("\0\0\0\1\0\0\0\0", 8), ""}, {}, 7, 7));
// partition key < (hash=1, hash=0)
- NO_FATALS(check({}, "", string("\0\0\0\1\0\0\0\0", 8), 5, 5));
+ NO_FATALS(check({}, {}, {string("\0\0\0\1\0\0\0\0", 8), ""}, 5, 5));
// C >= "e"
// C < "m"
// partition key >= (hash=1)
+ //
+ // The only non-pruned tablets are:
+ // * range d-f:
+ // hash keys:
+ // ** "\0\0\0\1\0\0\0\0"
+ // ** "\0\0\0\1\0\0\0\1"
+ // ** "\0\0\0\1\0\0\0\2"
+ // * range k-m:
+ // hash keys:
+ // ** "\0\0\0\1"
NO_FATALS(check({ColumnPredicate::Range(schema.column(2), &e, &m)},
- string("\0\0\0\1", 4), "", 5, 5));
+ { string("\0\0\0\1", 4), "" }, {}, 5, 5));
+
// C >= "e"
// C < "m"
// partition key < (hash=1)
- NO_FATALS(check({ColumnPredicate::Range(schema.column(2), &e, &m)}, "",
- string("\0\0\0\1", 4), 4, 4));
+ //
+ // The only non-pruned tablets are:
+ // * range d-f:
+ // hash keys:
+ // ** "\0\0\0\0\0\0\0\0"
+ // ** "\0\0\0\0\0\0\0\1"
+ // ** "\0\0\0\0\0\0\0\2"
+ // * range k-m:
+ // hash keys:
+ // ** "\0\0\0\0"
+ NO_FATALS(check({ColumnPredicate::Range(schema.column(2), &e, &m)},
+ {}, { string("\0\0\0\1", 4), "" }, 4, 4));
}
TEST_F(PartitionPrunerTest, TestHashSchemasPerRangeWithPartialPrimaryKeyRangePruning) {
@@ -1330,8 +1340,8 @@ TEST_F(PartitionPrunerTest, TestHashSchemasPerRangeWithPartialPrimaryKeyRangePru
enc_upper_bound = EncodedKey::FromContiguousRow(row, &arena);
spec.SetExclusiveUpperBoundKey(enc_upper_bound);
}
- NO_FATALS(CheckPrunedPartitions(schema, partition_schema, partitions, spec,
- remaining_tablets, pruner_ranges));
+ CheckPrunedPartitions(schema, partition_schema, partitions, spec,
+ remaining_tablets, pruner_ranges);
};
// No bounds
diff --git a/src/kudu/common/partition_pruner.cc b/src/kudu/common/partition_pruner.cc
index 3601abc..fea1d3d 100644
--- a/src/kudu/common/partition_pruner.cc
+++ b/src/kudu/common/partition_pruner.cc
@@ -284,18 +284,20 @@ vector<PartitionPruner::PartitionKeyRange> PartitionPruner::ConstructPartitionKe
bool is_last = hash_idx + 1 == constrained_index && range_bounds.upper.empty();
vector<PartitionKeyRange> new_partition_key_ranges;
- for (const auto& partition_key_range : partition_key_ranges) {
+ for (const auto& key_range : partition_key_ranges) {
const vector<bool>& buckets_bitset = hash_bucket_bitsets[hash_idx];
for (uint32_t bucket = 0; bucket < buckets_bitset.size(); ++bucket) {
if (!buckets_bitset[bucket]) {
continue;
}
uint32_t bucket_upper = is_last ? bucket + 1 : bucket;
- string lower = partition_key_range.start;
- string upper = partition_key_range.end;
+ auto lower = key_range.start.hash_key();
+ auto upper = key_range.end.hash_key();
hash_encoder.Encode(&bucket, &lower);
hash_encoder.Encode(&bucket_upper, &upper);
- new_partition_key_ranges.emplace_back(PartitionKeyRange{move(lower), move(upper)});
+ new_partition_key_ranges.emplace_back(PartitionKeyRange{
+ { lower, key_range.start.range_key() },
+ { upper, key_range.end.range_key() }});
}
}
partition_key_ranges.swap(new_partition_key_ranges);
@@ -303,8 +305,8 @@ vector<PartitionPruner::PartitionKeyRange> PartitionPruner::ConstructPartitionKe
// Append the (possibly empty) range bounds to the partition key ranges.
for (auto& range : partition_key_ranges) {
- range.start.append(range_bounds.lower);
- range.end.append(range_bounds.upper);
+ range.start.mutable_range_key()->append(range_bounds.lower);
+ range.end.mutable_range_key()->append(range_bounds.upper);
}
// Remove all partition key ranges past the scan spec's upper bound partition key.
@@ -500,12 +502,12 @@ bool PartitionPruner::HasMorePartitionKeyRanges() const {
return NumRangesRemaining() != 0;
}
-const string& PartitionPruner::NextPartitionKey() const {
+const PartitionKey& PartitionPruner::NextPartitionKey() const {
CHECK(HasMorePartitionKeyRanges());
return range_bounds_to_partition_key_ranges_.back().partition_key_ranges.back().start;
}
-void PartitionPruner::RemovePartitionKeyRange(const string& upper_bound) {
+void PartitionPruner::RemovePartitionKeyRange(const PartitionKey& upper_bound) {
if (upper_bound.empty()) {
range_bounds_to_partition_key_ranges_.clear();
return;
@@ -527,25 +529,26 @@ void PartitionPruner::RemovePartitionKeyRange(const string& upper_bound) {
}
bool PartitionPruner::ShouldPrune(const Partition& partition) const {
- for (const auto& [range_bounds, partition_key_range] : range_bounds_to_partition_key_ranges_) {
+ for (const auto& [range_bounds, partition_key_ranges] : range_bounds_to_partition_key_ranges_) {
// Check if the partition belongs to the same range as the partition key range.
- if (!range_bounds.lower.empty() && partition.range_key_start() != range_bounds.lower &&
- !range_bounds.upper.empty() && partition.range_key_end() != range_bounds.upper) {
+ if (!range_bounds.lower.empty() && partition.begin().range_key() != range_bounds.lower &&
+ !range_bounds.upper.empty() && partition.end().range_key() != range_bounds.upper) {
continue;
}
// range is an iterator that points to the first partition key range which
// overlaps or is greater than the partition.
- auto range = lower_bound(partition_key_range.rbegin(), partition_key_range.rend(),
+ auto range = lower_bound(partition_key_ranges.rbegin(), partition_key_ranges.rend(),
partition, [] (const PartitionKeyRange& scan_range,
const Partition& partition) {
// return true if scan_range < partition
- const string& scan_upper = scan_range.end;
+ const auto& scan_upper = scan_range.end;
return !scan_upper.empty()
- && scan_upper <= partition.partition_key_start();
+ && scan_upper <= partition.begin();
});
- if (!(range == partition_key_range.rend() ||
- (!partition.partition_key_end().empty() &&
- partition.partition_key_end() <= (*range).start))) {
+ if (range == partition_key_ranges.rend()) {
+ continue;
+ }
+ if (partition.end().empty() || (*range).start < partition.end()) {
return false;
}
}
diff --git a/src/kudu/common/partition_pruner.h b/src/kudu/common/partition_pruner.h
index 3077852..e3c3aa3 100644
--- a/src/kudu/common/partition_pruner.h
+++ b/src/kudu/common/partition_pruner.h
@@ -50,10 +50,10 @@ class PartitionPruner {
bool HasMorePartitionKeyRanges() const;
// Returns the inclusive lower bound partition key of the next tablet to scan.
- const std::string& NextPartitionKey() const;
+ const PartitionKey& NextPartitionKey() const;
// Removes all partition key ranges through the provided exclusive upper bound.
- void RemovePartitionKeyRange(const std::string& upper_bound);
+ void RemovePartitionKeyRange(const PartitionKey& upper_bound);
// Returns true if the provided partition should be pruned.
bool ShouldPrune(const Partition& partition) const;
@@ -78,8 +78,8 @@ class PartitionPruner {
};
struct PartitionKeyRange {
- std::string start;
- std::string end;
+ PartitionKey start;
+ PartitionKey end;
};
struct RangeBoundsAndPartitionKeyRanges {
@@ -102,8 +102,8 @@ class PartitionPruner {
const PartitionSchema::HashSchema& hash_schema,
const RangeBounds& range_bounds);
- // A vector of a pair of lower and upper range bounds mapped to a
- // reverse sorted set of partition key ranges. Each partition key range within the set
+ // A vector of a pair of lower and upper range bounds mapped to a reverse
+ // sorted set of partition key ranges. Each partition key range within the set
// has an inclusive lower bound and an exclusive upper bound.
std::vector<RangeBoundsAndPartitionKeyRanges> range_bounds_to_partition_key_ranges_;
diff --git a/src/kudu/common/scan_spec.cc b/src/kudu/common/scan_spec.cc
index 87b3d5b..f22b9c2 100644
--- a/src/kudu/common/scan_spec.cc
+++ b/src/kudu/common/scan_spec.cc
@@ -18,6 +18,7 @@
#include "kudu/common/scan_spec.h"
#include <algorithm>
+#include <cstddef>
#include <iterator>
#include <ostream>
#include <string>
@@ -41,6 +42,7 @@
#include "kudu/gutil/strings/substitute.h"
#include "kudu/util/array_view.h"
#include "kudu/util/memory/arena.h"
+#include "kudu/util/slice.h"
#include "kudu/util/status.h"
using std::any_of;
@@ -111,17 +113,17 @@ void ScanSpec::SetExclusiveUpperBoundKey(const EncodedKey* key) {
}
}
-void ScanSpec::SetLowerBoundPartitionKey(const Slice& partition_key) {
+void ScanSpec::SetLowerBoundPartitionKey(const PartitionKey& partition_key) {
if (lower_bound_partition_key_ < partition_key) {
- lower_bound_partition_key_ = partition_key.ToString();
+ lower_bound_partition_key_ = partition_key;
}
}
-void ScanSpec::SetExclusiveUpperBoundPartitionKey(const Slice& partition_key) {
+void ScanSpec::SetExclusiveUpperBoundPartitionKey(const PartitionKey& partition_key) {
if (exclusive_upper_bound_partition_key_.empty() ||
(!partition_key.empty() &&
partition_key < exclusive_upper_bound_partition_key_)) {
- exclusive_upper_bound_partition_key_ = partition_key.ToString();
+ exclusive_upper_bound_partition_key_ = partition_key;
}
}
@@ -135,7 +137,7 @@ string ScanSpec::ToString(const Schema& schema) const {
}
// List predicates in stable order.
- for (int idx = 0; idx < schema.num_columns(); idx++) {
+ for (size_t idx = 0; idx < schema.num_columns(); ++idx) {
const ColumnPredicate* predicate = FindOrNull(predicates_, schema.column(idx).name());
if (predicate != nullptr) {
preds.push_back(predicate->ToString());
diff --git a/src/kudu/common/scan_spec.h b/src/kudu/common/scan_spec.h
index a660502..d62e327 100644
--- a/src/kudu/common/scan_spec.h
+++ b/src/kudu/common/scan_spec.h
@@ -25,15 +25,13 @@
#include <glog/logging.h>
#include "kudu/common/column_predicate.h" // IWYU pragma: keep
-#include "kudu/util/slice.h"
+#include "kudu/common/partition.h"
namespace kudu {
class Arena;
class ColumnSchema;
class EncodedKey;
-class Partition;
-class PartitionSchema;
class Schema;
class ScanSpec {
@@ -107,17 +105,13 @@ class ScanSpec {
// Sets the lower bound (inclusive) partition key for the scan.
//
- // The scan spec makes a copy of 'slice'; the caller may free it afterward.
- //
// Only used in the client.
- void SetLowerBoundPartitionKey(const Slice& slice);
+ void SetLowerBoundPartitionKey(const PartitionKey& partition_key);
// Sets the upper bound (exclusive) partition key for the scan.
//
- // The scan spec makes a copy of 'slice'; the caller may free it afterward.
- //
// Only used in the client.
- void SetExclusiveUpperBoundPartitionKey(const Slice& slice);
+ void SetExclusiveUpperBoundPartitionKey(const PartitionKey& partition_key);
// Returns the scan predicates.
const std::unordered_map<std::string, ColumnPredicate>& predicates() const {
@@ -132,10 +126,10 @@ class ScanSpec {
return exclusive_upper_bound_key_;
}
- const std::string& lower_bound_partition_key() const {
+ const PartitionKey& lower_bound_partition_key() const {
return lower_bound_partition_key_;
}
- const std::string& exclusive_upper_bound_partition_key() const {
+ const PartitionKey& exclusive_upper_bound_partition_key() const {
return exclusive_upper_bound_partition_key_;
}
@@ -186,8 +180,8 @@ class ScanSpec {
std::unordered_map<std::string, ColumnPredicate> predicates_;
const EncodedKey* lower_bound_key_;
const EncodedKey* exclusive_upper_bound_key_;
- std::string lower_bound_partition_key_;
- std::string exclusive_upper_bound_partition_key_;
+ PartitionKey lower_bound_partition_key_;
+ PartitionKey exclusive_upper_bound_partition_key_;
bool cache_blocks_;
boost::optional<int64_t> limit_;
};
diff --git a/src/kudu/integration-tests/table_locations-itest.cc b/src/kudu/integration-tests/table_locations-itest.cc
index 5fe8ecd..983e2c8 100644
--- a/src/kudu/integration-tests/table_locations-itest.cc
+++ b/src/kudu/integration-tests/table_locations-itest.cc
@@ -39,6 +39,7 @@
#include "kudu/client/schema.h"
#include "kudu/common/common.pb.h"
#include "kudu/common/partial_row.h"
+#include "kudu/common/partition.h"
#include "kudu/common/row_operations.h"
#include "kudu/common/row_operations.pb.h"
#include "kudu/common/schema.h"
@@ -507,32 +508,32 @@ TEST_F(TableLocationsTest, TestRangeSpecificHashing) {
ASSERT_FALSE(resp.has_error());
ASSERT_EQ(19, resp.tablet_locations().size());
- vector<string> partition_key_starts = {
- string("\0\0\0\0" "\0\0\0\0" "a" "\0\0", 11),
- string("\0\0\0\0" "\0\0\0\1" "a" "\0\0", 11),
- string("\0\0\0\1" "\0\0\0\0" "a" "\0\0", 11),
- string("\0\0\0\1" "\0\0\0\1" "a" "\0\0", 11),
- string("\0\0\0\2" "\0\0\0\0" "a" "\0\0", 11),
- string("\0\0\0\2" "\0\0\0\1" "a" "\0\0", 11),
- string("\0\0\0\3" "\0\0\0\0" "a" "\0\0", 11),
- string("\0\0\0\3" "\0\0\0\1" "a" "\0\0", 11),
- string("\0\0\0\0" "c" "\0\0", 7),
- string("\0\0\0\1" "c" "\0\0", 7),
- string("\0\0\0\2" "c" "\0\0", 7),
- string("\0\0\0\3" "c" "\0\0", 7),
- string("\0\0\0\4" "c" "\0\0", 7),
- string("\0\0\0\5" "c" "\0\0", 7),
- string("\0\0\0\0" "e" "\0\0", 7),
- string("\0\0\0\1" "e" "\0\0", 7),
- string("\0\0\0\2" "e" "\0\0", 7),
- string("\0\0\0\3" "e" "\0\0", 7),
- string("\0\0\0\4" "e" "\0\0", 7)
+ vector<PartitionKey> partition_key_starts = {
+ { string("\0\0\0\0" "\0\0\0\0", 8), string("a" "\0\0", 3) },
+ { string("\0\0\0\0" "\0\0\0\1", 8), string("a" "\0\0", 3) },
+ { string("\0\0\0\1" "\0\0\0\0", 8), string("a" "\0\0", 3) },
+ { string("\0\0\0\1" "\0\0\0\1", 8), string("a" "\0\0", 3) },
+ { string("\0\0\0\2" "\0\0\0\0", 8), string("a" "\0\0", 3) },
+ { string("\0\0\0\2" "\0\0\0\1", 8), string("a" "\0\0", 3) },
+ { string("\0\0\0\3" "\0\0\0\0", 8), string("a" "\0\0", 3) },
+ { string("\0\0\0\3" "\0\0\0\1", 8), string("a" "\0\0", 3) },
+ { string("\0\0\0\0", 4), string("c" "\0\0", 3) },
+ { string("\0\0\0\1", 4), string("c" "\0\0", 3) },
+ { string("\0\0\0\2", 4), string("c" "\0\0", 3) },
+ { string("\0\0\0\3", 4), string("c" "\0\0", 3) },
+ { string("\0\0\0\4", 4), string("c" "\0\0", 3) },
+ { string("\0\0\0\5", 4), string("c" "\0\0", 3) },
+ { string("\0\0\0\0", 4), string("e" "\0\0", 3) },
+ { string("\0\0\0\1", 4), string("e" "\0\0", 3) },
+ { string("\0\0\0\2", 4), string("e" "\0\0", 3) },
+ { string("\0\0\0\3", 4), string("e" "\0\0", 3) },
+ { string("\0\0\0\4", 4), string("e" "\0\0", 3) },
};
// Sorting partition keys to match the tablets that are returned in sorted order.
sort(partition_key_starts.begin(), partition_key_starts.end());
ASSERT_EQ(partition_key_starts.size(), resp.tablet_locations_size());
for (int i = 0; i < resp.tablet_locations_size(); i++) {
- EXPECT_EQ(partition_key_starts[i],
+ EXPECT_EQ(partition_key_starts[i].ToString(),
resp.tablet_locations(i).partition().partition_key_start());
}
}
diff --git a/src/kudu/master/catalog_manager-test.cc b/src/kudu/master/catalog_manager-test.cc
index 7e00a7b..e4889ca 100644
--- a/src/kudu/master/catalog_manager-test.cc
+++ b/src/kudu/master/catalog_manager-test.cc
@@ -15,6 +15,8 @@
// specific language governing permissions and limitations
// under the License.
+#include "kudu/master/catalog_manager.h"
+
#include <ostream>
#include <string>
#include <vector>
@@ -23,13 +25,14 @@
#include <gtest/gtest.h>
#include "kudu/common/common.pb.h"
+#include "kudu/common/row_operations.pb.h"
#include "kudu/gutil/ref_counted.h"
#include "kudu/gutil/strings/substitute.h"
-#include "kudu/master/catalog_manager.h"
#include "kudu/master/master.pb.h"
#include "kudu/master/ts_descriptor.h"
#include "kudu/util/cow_object.h"
#include "kudu/util/monotime.h"
+#include "kudu/util/status.h"
#include "kudu/util/test_macros.h"
#include "kudu/util/test_util.h"
@@ -85,7 +88,7 @@ TEST(TableInfoTest, TestAssignmentRanges) {
req.mutable_table()->mutable_table_name()->assign(table_id);
req.mutable_partition_key_start()->assign(start_key);
vector<scoped_refptr<TabletInfo>> tablets_in_range;
- table->GetTabletsInRange(&req, &tablets_in_range);
+ ASSERT_OK(table->GetTabletsInRange(&req, &tablets_in_range));
// Only one tablet should own this key.
ASSERT_EQ(1, tablets_in_range.size());
@@ -95,6 +98,69 @@ TEST(TableInfoTest, TestAssignmentRanges) {
}
}
+TEST(TableInfoTest, GetTableLocationsLegacyCustomHashSchemas) {
+ const string table_id = CURRENT_TEST_NAME();
+ scoped_refptr<TableInfo> table(new TableInfo(table_id));
+
+ {
+ TableMetadataLock meta_lock(table.get(), LockMode::WRITE);
+ auto* ps = meta_lock.mutable_data()->pb.mutable_partition_schema();
+ // It's not really necessary to fill everything in the scope of this test.
+ auto* range = ps->add_custom_hash_schema_ranges();
+ range->mutable_range_bounds()->set_rows("a");
+ auto* hash_dimension = range->add_hash_schema();
+ hash_dimension->add_columns()->set_name("b");
+ hash_dimension->set_num_buckets(2);
+ meta_lock.Commit();
+ }
+
+ const string start_key = "a";
+ const string end_key = "";
+ string tablet_id = Substitute("tablet-$0-$1", start_key, end_key);
+
+ scoped_refptr<TabletInfo> tablet = new TabletInfo(table, tablet_id);
+ {
+ TabletMetadataLock meta_lock(tablet.get(), LockMode::WRITE);
+ PartitionPB* partition = meta_lock.mutable_data()->pb.mutable_partition();
+ partition->set_partition_key_start(start_key);
+ partition->set_partition_key_end(end_key);
+ meta_lock.mutable_data()->pb.set_state(SysTabletsEntryPB::RUNNING);
+ meta_lock.Commit();
+ }
+
+ TabletMetadataLock meta_lock(tablet.get(), LockMode::READ);
+ table->AddRemoveTablets({ tablet }, {});
+
+ // Query by specifying the start of the partition via the partition_key_start
+ // field: it should fail since the table has a range with custom hash schema.
+ {
+ GetTableLocationsRequestPB req;
+ req.set_max_returned_locations(1);
+ req.mutable_table()->mutable_table_name()->assign(table_id);
+ req.mutable_partition_key_start()->assign("a");
+ vector<scoped_refptr<TabletInfo>> tablets_in_range;
+ auto s = table->GetTabletsInRange(&req, &tablets_in_range);
+ ASSERT_TRUE(s.IsInvalidArgument()) << s.ToString();
+ ASSERT_STR_CONTAINS(s.ToString(),
+ "for a table with custom per-range hash schemas "
+ "the range must be specified using partition_key_range "
+ "field, not partition_key_{start,end} fields");
+ }
+
+ // Query by specifying the start of the partition via the partition_key_range
+ // field: it should succeed.
+ {
+ GetTableLocationsRequestPB req;
+ req.set_max_returned_locations(1);
+ req.mutable_table()->mutable_table_name()->assign(table_id);
+ req.mutable_key_start()->set_hash_key(string("\0\0\0\0", 4));
+ req.mutable_key_start()->set_range_key("a");
+ vector<scoped_refptr<TabletInfo>> tablets_in_range;
+ ASSERT_OK(table->GetTabletsInRange(&req, &tablets_in_range));
+ ASSERT_EQ(1, tablets_in_range.size());
+ }
+}
+
TEST(TestTSDescriptor, TestReplicaCreationsDecay) {
TSDescriptor ts("test");
ASSERT_EQ(0, ts.RecentReplicaCreations());
diff --git a/src/kudu/master/catalog_manager.cc b/src/kudu/master/catalog_manager.cc
index a170151..b9606fe 100644
--- a/src/kudu/master/catalog_manager.cc
+++ b/src/kudu/master/catalog_manager.cc
@@ -2597,8 +2597,8 @@ Status CatalogManager::ApplyAlterPartitioningSteps(
switch (step.type()) {
case AlterTableRequestPB::ADD_RANGE_PARTITION: {
for (const Partition& partition : partitions) {
- const string& lower_bound = partition.partition_key_start();
- const string& upper_bound = partition.partition_key_end();
+ const auto& lower_bound = partition.begin();
+ const auto& upper_bound = partition.end();
// Check that the new tablet does not overlap with any of the existing
// tablets. Since the elements of 'existing_tablets' are ordered by
@@ -2610,8 +2610,10 @@ Status CatalogManager::ApplyAlterPartitioningSteps(
TabletMetadataLock metadata(existing_iter->second.get(),
LockMode::READ);
const auto& p = metadata.data().pb.partition();
+ const auto p_begin = Partition::StringToPartitionKey(
+ p.partition_key_start(), p.hash_buckets_size());
// Check for the overlapping ranges.
- if (upper_bound.empty() || p.partition_key_start() < upper_bound) {
+ if (upper_bound.empty() || p_begin < upper_bound) {
return Status::InvalidArgument(
"new range partition conflicts with existing one",
partition_schema.RangePartitionDebugString(*ops[0].split_row,
@@ -2628,17 +2630,19 @@ Status CatalogManager::ApplyAlterPartitioningSteps(
TabletMetadataLock metadata(std::prev(existing_iter)->second.get(),
LockMode::READ);
const auto& p = metadata.data().pb.partition();
+ const auto p_begin = Partition::StringToPartitionKey(
+ p.partition_key_start(), p.hash_buckets_size());
+ const auto p_end = Partition::StringToPartitionKey(
+ p.partition_key_end(), p.hash_buckets_size());
// Check for the exact match of ranges.
- if (lower_bound == p.partition_key_start() &&
- upper_bound == p.partition_key_end()) {
+ if (lower_bound == p_begin && upper_bound == p_end) {
return Status::AlreadyPresent(
"range partition already exists",
partition_schema.RangePartitionDebugString(*ops[0].split_row,
*ops[1].split_row));
}
// Check for the overlapping ranges.
- if (p.partition_key_end().empty() ||
- p.partition_key_end() > lower_bound) {
+ if (p_end.empty() || p_end > lower_bound) {
return Status::InvalidArgument(
"new range partition conflicts with existing one",
partition_schema.RangePartitionDebugString(*ops[0].split_row,
@@ -2651,7 +2655,9 @@ Status CatalogManager::ApplyAlterPartitioningSteps(
if (new_iter != new_tablets.end()) {
// Check for the overlapping ranges.
const auto& p = new_iter->second->mutable_metadata()->dirty().pb.partition();
- if (upper_bound.empty() || p.partition_key_start() < upper_bound) {
+ const auto p_begin = Partition::StringToPartitionKey(
+ p.partition_key_start(), p.hash_buckets_size());
+ if (upper_bound.empty() || p_begin < upper_bound) {
return Status::InvalidArgument(
"new range partition conflicts with another newly added one",
partition_schema.RangePartitionDebugString(*ops[0].split_row,
@@ -2660,16 +2666,19 @@ Status CatalogManager::ApplyAlterPartitioningSteps(
}
if (new_iter != new_tablets.begin()) {
const auto& p = std::prev(new_iter)->second->mutable_metadata()->dirty().pb.partition();
+ const auto p_begin = Partition::StringToPartitionKey(
+ p.partition_key_start(), p.hash_buckets_size());
+ const auto p_end = Partition::StringToPartitionKey(
+ p.partition_key_end(), p.hash_buckets_size());
// Check for the exact match of ranges.
- if (lower_bound == p.partition_key_start() &&
- upper_bound == p.partition_key_end()) {
+ if (lower_bound == p_begin && upper_bound == p_end) {
return Status::AlreadyPresent(
"new range partition duplicates another newly added one",
partition_schema.RangePartitionDebugString(*ops[0].split_row,
*ops[1].split_row));
}
// Check for the overlapping ranges.
- if (p.partition_key_end().empty() || p.partition_key_end() > lower_bound) {
+ if (p_end.empty() || p_end > lower_bound) {
return Status::InvalidArgument(
"new range partition conflicts with another newly added one",
partition_schema.RangePartitionDebugString(*ops[0].split_row,
@@ -2691,8 +2700,8 @@ Status CatalogManager::ApplyAlterPartitioningSteps(
case AlterTableRequestPB::DROP_RANGE_PARTITION: {
for (const Partition& partition : partitions) {
- const string& lower_bound = partition.partition_key_start();
- const string& upper_bound = partition.partition_key_end();
+ const auto& lower_bound = partition.begin();
+ const auto& upper_bound = partition.end();
// Iter points to the tablet if it exists, or the next tablet, or the end.
auto existing_iter = existing_tablets.lower_bound(lower_bound);
@@ -2703,14 +2712,20 @@ Status CatalogManager::ApplyAlterPartitioningSteps(
if (existing_iter != existing_tablets.end()) {
TabletMetadataLock metadata(existing_iter->second.get(), LockMode::READ);
- const auto& partition = metadata.data().pb.partition();
- found_existing = partition.partition_key_start() == lower_bound &&
- partition.partition_key_end() == upper_bound;
+ const auto& p = metadata.data().pb.partition();
+ const auto p_begin = Partition::StringToPartitionKey(
+ p.partition_key_start(), p.hash_buckets_size());
+ const auto p_end = Partition::StringToPartitionKey(
+ p.partition_key_end(), p.hash_buckets_size());
+ found_existing = p_begin == lower_bound && p_end == upper_bound;
}
if (new_iter != new_tablets.end()) {
- const auto& partition = new_iter->second->mutable_metadata()->dirty().pb.partition();
- found_new = partition.partition_key_start() == lower_bound &&
- partition.partition_key_end() == upper_bound;
+ const auto& p = new_iter->second->mutable_metadata()->dirty().pb.partition();
+ const auto p_begin = Partition::StringToPartitionKey(
+ p.partition_key_start(), p.hash_buckets_size());
+ const auto p_end = Partition::StringToPartitionKey(
+ p.partition_key_end(), p.hash_buckets_size());
+ found_new = p_begin == lower_bound && p_end == upper_bound;
}
DCHECK(!found_existing || !found_new);
@@ -5679,6 +5694,13 @@ Status CatalogManager::GetTableLocations(const GetTableLocationsRequestPB* req,
&& req->partition_key_start() > req->partition_key_end())) {
return Status::InvalidArgument("start partition key is greater than the end partition key");
}
+ if (PREDICT_FALSE(req->has_key_start() && req->has_key_end() &&
+ req->key_start().has_range_key() &&
+ req->key_end().has_range_key() &&
+ req->key_start().range_key() > req->key_end().range_key())) {
+ return Status::InvalidArgument("start partition range key must not be "
+ "greater than the end partition range key");
+ }
if (PREDICT_FALSE(req->max_returned_locations() <= 0)) {
return Status::InvalidArgument("max_returned_locations must be greater than 0");
}
@@ -5699,7 +5721,7 @@ Status CatalogManager::GetTableLocations(const GetTableLocationsRequestPB* req,
RETURN_NOT_OK(CheckIfTableDeletedOrNotRunning(&l, resp));
vector<scoped_refptr<TabletInfo>> tablets_in_range;
- table->GetTabletsInRange(req, &tablets_in_range);
+ RETURN_NOT_OK(table->GetTabletsInRange(req, &tablets_in_range));
// Check for items in the cache.
if (table_locations_cache_) {
@@ -6239,10 +6261,12 @@ void TabletInfo::set_reported_schema_version(int64_t version) {
// We also need to hold the tablet metadata lock in order to read the partition
// key, but it's OK to make a local copy of it (and release the lock) because
// the key is immutable.
- string key_start;
+ PartitionKey key_start;
{
TabletMetadataLock l(this, LockMode::READ);
- key_start = l.data().pb.partition().partition_key_start();
+ const auto& p = l.data().pb.partition();
+ key_start = Partition::StringToPartitionKey(
+ p.partition_key_start(), p.hash_buckets_size());
}
std::lock_guard<rw_spinlock> table_l(table_->lock_);
std::lock_guard<simple_spinlock> tablet_l(lock_);
@@ -6319,17 +6343,20 @@ void TableInfo::AddRemoveTablets(const vector<scoped_refptr<TabletInfo>>& tablet
const vector<scoped_refptr<TabletInfo>>& tablets_to_drop) {
std::lock_guard<rw_spinlock> l(lock_);
for (const auto& tablet : tablets_to_drop) {
- const auto& lower_bound = tablet->metadata().state().pb.partition().partition_key_start();
+ const auto& p = tablet->metadata().state().pb.partition();
+ const auto& lower_bound = Partition::StringToPartitionKey(
+ p.partition_key_start(), p.hash_buckets_size());
CHECK(EraseKeyReturnValuePtr(&tablet_map_, lower_bound) != nullptr);
DecrementSchemaVersionCountUnlocked(tablet->reported_schema_version());
// Remove the table metrics for the deleted tablets.
RemoveMetrics(tablet->id(), tablet->GetStats());
}
for (const auto& tablet : tablets_to_add) {
+ const auto& p = tablet->metadata().state().pb.partition();
+ const auto& key_start = Partition::StringToPartitionKey(
+ p.partition_key_start(), p.hash_buckets_size());
TabletInfo* old = nullptr;
- if (UpdateReturnCopy(&tablet_map_,
- tablet->metadata().state().pb.partition().partition_key_start(),
- tablet.get(), &old)) {
+ if (UpdateReturnCopy(&tablet_map_, key_start, tablet.get(), &old)) {
VLOG(1) << Substitute("Replaced tablet $0 with $1",
old->id(), tablet->id());
DecrementSchemaVersionCountUnlocked(old->reported_schema_version());
@@ -6348,13 +6375,67 @@ void TableInfo::AddRemoveTablets(const vector<scoped_refptr<TabletInfo>>& tablet
#endif
}
-void TableInfo::GetTabletsInRange(const GetTableLocationsRequestPB* req,
- vector<scoped_refptr<TabletInfo>>* ret) const {
- shared_lock<rw_spinlock> l(lock_);
+Status TableInfo::GetTabletsInRange(
+ const GetTableLocationsRequestPB* req,
+ vector<scoped_refptr<TabletInfo>>* ret) const {
+
+ size_t hash_dimensions_num = 0;
+ bool has_custom_hash_schemas = false;
+ {
+ TableMetadataLock l(this, LockMode::READ);
+ const auto& ps = l.data().pb.partition_schema();
+ hash_dimensions_num = ps.hash_schema_size();
+ has_custom_hash_schemas = ps.custom_hash_schema_ranges_size() > 0;
+ }
+
+ // Find partition keys for the start and the end of the range in question.
+ // That's done with extra guardrails to ensure the table doesn't have custom
+ // hash schemas per range when using legacy fields
+ // GetTableLocationsRequestPB::{partition_key_start,partition_key_end}.
+ PartitionKey partition_key_start;
+ bool has_key_start = false;
+ if (req->has_key_start()) {
+ const auto& start = req->key_start();
+ if (start.has_hash_key() || start.has_range_key()) {
+ partition_key_start = PartitionKey(start.hash_key(), start.range_key());
+ has_key_start = true;
+ }
+ } else if (req->has_partition_key_start()) {
+ if (has_custom_hash_schemas) {
+ return Status::InvalidArgument(Substitute(
+ "$0: for a table with custom per-range hash schemas the range must "
+ "be specified using partition_key_range field, not "
+ "partition_key_{start,end} fields", ToString()));
+ }
+ partition_key_start = Partition::StringToPartitionKey(
+ req->partition_key_start(), hash_dimensions_num);
+ has_key_start = true;
+ }
+ PartitionKey partition_key_end;
+ bool has_key_end = false;
+ if (req->has_key_end()) {
+ const auto& end = req->key_end();
+ if (end.has_hash_key() || end.has_range_key()) {
+ partition_key_end = PartitionKey(end.hash_key(), end.range_key());
+ has_key_end = true;
+ }
+ } else if (req->has_partition_key_end()) {
+ if (has_custom_hash_schemas) {
+ return Status::InvalidArgument(Substitute(
+ "$0: for a table with custom per-range hash schemas the range must "
+ "be specified using partition_key_range field, not "
+ "partition_key_{start,end} fields", ToString()));
+ }
+ partition_key_end = Partition::StringToPartitionKey(
+ req->partition_key_end(), hash_dimensions_num);
+ has_key_end = true;
+ }
+
+ shared_lock<rw_spinlock> l(lock_);
RawTabletInfoMap::const_iterator it;
- if (req->has_partition_key_start()) {
- it = tablet_map_.upper_bound(req->partition_key_start());
+ if (has_key_start) {
+ it = tablet_map_.upper_bound(partition_key_start);
if (it != tablet_map_.begin()) {
--it;
}
@@ -6362,8 +6443,8 @@ void TableInfo::GetTabletsInRange(const GetTableLocationsRequestPB* req,
it = tablet_map_.begin();
}
- const RawTabletInfoMap::const_iterator it_end = req->has_partition_key_end()
- ? tablet_map_.upper_bound(req->partition_key_end())
+ const RawTabletInfoMap::const_iterator it_end = has_key_end
+ ? tablet_map_.upper_bound(partition_key_end)
: tablet_map_.end();
const size_t max_returned_locations = req->max_returned_locations();
@@ -6372,6 +6453,8 @@ void TableInfo::GetTabletsInRange(const GetTableLocationsRequestPB* req,
ret->emplace_back(make_scoped_refptr(it->second));
++count;
}
+
+ return Status::OK();
}
bool TableInfo::IsAlterInProgress(uint32_t version) const {
diff --git a/src/kudu/master/catalog_manager.h b/src/kudu/master/catalog_manager.h
index 8b76d5e..5e14bc2 100644
--- a/src/kudu/master/catalog_manager.h
+++ b/src/kudu/master/catalog_manager.h
@@ -36,6 +36,7 @@
#include <gtest/gtest_prod.h>
#include <sparsehash/dense_hash_map>
+#include "kudu/common/partition.h"
#include "kudu/consensus/metadata.pb.h"
#include "kudu/consensus/raft_consensus.h"
#include "kudu/gutil/macros.h"
@@ -71,7 +72,6 @@ class MetricRegistry;
class MonitoredTask;
class NodeInstancePB;
class PartitionPB;
-class PartitionSchema;
class Schema;
class TableExtraConfigPB;
class ThreadPool;
@@ -284,7 +284,7 @@ class TableInfo : public RefCountedThreadSafe<TableInfo> {
};
typedef PersistentTableInfo cow_state;
- typedef std::map<std::string, scoped_refptr<TabletInfo>> TabletInfoMap;
+ typedef std::map<PartitionKey, scoped_refptr<TabletInfo>> TabletInfoMap;
explicit TableInfo(std::string table_id);
@@ -301,9 +301,13 @@ class TableInfo : public RefCountedThreadSafe<TableInfo> {
void AddRemoveTablets(const std::vector<scoped_refptr<TabletInfo>>& tablets_to_add,
const std::vector<scoped_refptr<TabletInfo>>& tablets_to_drop);
- // This only returns tablets which are in RUNNING state.
- void GetTabletsInRange(const GetTableLocationsRequestPB* req,
- std::vector<scoped_refptr<TabletInfo>>* ret) const;
+ // This is only applicable to the tablets which are in RUNNING state. The
+ // method returns Status::InvalidArgument() status if the request happen to
+ // specify the range boundaries with legacy fields
+ // GetTableLocationsRequestPB::{partition_key_start,partition_key_end} when
+ // in fact the table has ranges with custom hash schemas.
+ Status GetTabletsInRange(const GetTableLocationsRequestPB* req,
+ std::vector<scoped_refptr<TabletInfo>>* ret) const;
// Fills the vector with all tablets in partition key sorted order.
void GetAllTablets(std::vector<scoped_refptr<TabletInfo>>* ret) const;
@@ -391,7 +395,7 @@ class TableInfo : public RefCountedThreadSafe<TableInfo> {
//
// Every TabletInfo has a strong backpointer to its TableInfo, so these
// pointers must be raw.
- typedef std::map<std::string, TabletInfo*> RawTabletInfoMap;
+ typedef std::map<PartitionKey, TabletInfo*> RawTabletInfoMap;
RawTabletInfoMap tablet_map_;
// Protects tablet_map_, pending_tasks_, and schema_version_counts_.
diff --git a/src/kudu/master/master.proto b/src/kudu/master/master.proto
index ca83159..a6e3f7a 100644
--- a/src/kudu/master/master.proto
+++ b/src/kudu/master/master.proto
@@ -616,10 +616,24 @@ message GetTableStatisticsResponsePB {
optional int64 row_count_limit = 5;
}
+// This data structure is used to specify a table's partition key.
+// That's used to find partitions (i.e. tablets) that cover the key range span
+// between two partition keys.
+message PartitionKeyPB {
+ // The range part of the encoded key.
+ optional bytes range_key = 1 [(kudu.REDACT) = true];
+
+ // The hash part of the encoded key.
+ optional bytes hash_key = 2;
+}
+
message GetTableLocationsRequestPB {
required TableIdentifierPB table = 1;
- // Partition-key range.
+ // These two fields are used to specify key range span if the table
+ // has a table-wide hash schema. Otherwise, neither of these two fields
+ // should not be present: use 'key_start' and 'key_end' fields below
+ // to specify key range information instead.
optional bytes partition_key_start = 3 [(kudu.REDACT) = true];
optional bytes partition_key_end = 4 [(kudu.REDACT) = true];
@@ -631,6 +645,17 @@ message GetTableLocationsRequestPB {
// Whether the response should use the 'interned_replicas' field.
optional bool intern_ts_infos_in_response = 7 [ default = false ];
+
+ // Partition key range, start and end: these fields make the
+ // 'partition_key_start' and 'partition_key_end' fields above obsolete.
+ // It's crucial to specify these (or at least one of these if the range is
+ // unbounded) when the table has custom hash schemas per range, otherwise
+ // the catalog manager responds back with an error. Even in case of
+ // table-wide hash schema the clients compiled with this proto specification
+ // are encouraged to use these fields instead of legacy 'partition_key_start'
+ // and 'partition_key_end'.
+ optional PartitionKeyPB key_start = 8;
+ optional PartitionKeyPB key_end = 9;
}
// The response to a GetTableLocations RPC. The master guarantees that:
diff --git a/src/kudu/master/master_path_handlers.cc b/src/kudu/master/master_path_handlers.cc
index 9de3632..fbcbf65 100644
--- a/src/kudu/master/master_path_handlers.cc
+++ b/src/kudu/master/master_path_handlers.cc
@@ -472,8 +472,8 @@ void MasterPathHandlers::HandleTablePage(const Webserver::WebRequest& req,
partition.hash_buckets().end(),
[] (const int32_t& bucket) { return bucket == 0; })) {
range_partitions.emplace_back(
- partition_schema.RangePartitionDebugString(partition.range_key_start(),
- partition.range_key_end(),
+ partition_schema.RangePartitionDebugString(partition.begin().range_key(),
+ partition.end().range_key(),
schema));
}
diff --git a/src/kudu/tools/table_scanner.cc b/src/kudu/tools/table_scanner.cc
index c9038c1..3c7923f 100644
--- a/src/kudu/tools/table_scanner.cc
+++ b/src/kudu/tools/table_scanner.cc
@@ -453,8 +453,8 @@ Status CreateDstTableIfNeeded(const client::sp::shared_ptr<KuduTable>& src_table
Arena arena(256);
std::unique_ptr<KuduPartialRow> lower(new KuduPartialRow(&schema_internal));
std::unique_ptr<KuduPartialRow> upper(new KuduPartialRow(&schema_internal));
- Slice range_key_start = partition.range_key_start();
- Slice range_key_end = partition.range_key_end();
+ Slice range_key_start(partition.begin().range_key());
+ Slice range_key_end(partition.end().range_key());
RETURN_NOT_OK(partition_schema.DecodeRangeKey(&range_key_start, lower.get(), &arena));
RETURN_NOT_OK(partition_schema.DecodeRangeKey(&range_key_end, upper.get(), &arena));
diff --git a/src/kudu/tools/tool_action_table.cc b/src/kudu/tools/tool_action_table.cc
index 15e2614..32bfd63 100644
--- a/src/kudu/tools/tool_action_table.cc
+++ b/src/kudu/tools/tool_action_table.cc
@@ -246,8 +246,8 @@ Status DescribeTable(const RunnerContext& context) {
continue;
}
auto range_partition_str =
- partition_schema.RangePartitionDebugString(partition.range_key_start(),
- partition.range_key_end(),
+ partition_schema.RangePartitionDebugString(partition.begin().range_key(),
+ partition.end().range_key(),
schema_internal);
partition_strs.emplace_back(std::move(range_partition_str));
}