| // Licensed to the Apache Software Foundation (ASF) under one |
| // or more contributor license agreements. See the NOTICE file |
| // distributed with this work for additional information |
| // regarding copyright ownership. The ASF licenses this file |
| // to you under the Apache License, Version 2.0 (the |
| // "License"); you may not use this file except in compliance |
| // with the License. You may obtain a copy of the License at |
| // |
| // http://www.apache.org/licenses/LICENSE-2.0 |
| // |
| // Unless required by applicable law or agreed to in writing, |
| // software distributed under the License is distributed on an |
| // "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY |
| // KIND, either express or implied. See the License for the |
| // specific language governing permissions and limitations |
| // under the License. |
| |
| #include <cstdint> |
| #include <cstdlib> |
| #include <iostream> |
| #include <memory> |
| #include <string> |
| #include <utility> |
| #include <vector> |
| |
| #include <gflags/gflags.h> |
| #include <glog/logging.h> |
| #include <gtest/gtest.h> |
| |
| #include "kudu/client/client-test-util.h" |
| #include "kudu/client/client.h" |
| #include "kudu/client/scan_batch.h" |
| #include "kudu/client/scan_predicate.h" |
| #include "kudu/client/schema.h" |
| #include "kudu/client/value.h" |
| #include "kudu/client/write_op.h" |
| #include "kudu/common/partial_row.h" |
| #include "kudu/common/partition.h" |
| #include "kudu/gutil/port.h" |
| #include "kudu/gutil/ref_counted.h" |
| #include "kudu/gutil/stl_util.h" |
| #include "kudu/gutil/strings/substitute.h" |
| #include "kudu/master/catalog_manager.h" |
| #include "kudu/master/master.h" |
| #include "kudu/master/mini_master.h" |
| #include "kudu/mini-cluster/internal_mini_cluster.h" |
| #include "kudu/tablet/tablet.h" |
| #include "kudu/tablet/tablet_replica.h" |
| #include "kudu/tserver/mini_tablet_server.h" |
| #include "kudu/tserver/tablet_server.h" |
| #include "kudu/tserver/ts_tablet_manager.h" |
| #include "kudu/util/curl_util.h" |
| #include "kudu/util/faststring.h" |
| #include "kudu/util/metrics.h" |
| #include "kudu/util/net/sockaddr.h" |
| #include "kudu/util/slice.h" |
| #include "kudu/util/status.h" |
| #include "kudu/util/test_macros.h" |
| #include "kudu/util/test_util.h" |
| |
| DECLARE_bool(enable_per_range_hash_schemas); |
| DECLARE_int32(flush_threshold_secs); |
| DECLARE_int32(heartbeat_interval_ms); |
| DECLARE_string(webserver_doc_root); |
| |
| METRIC_DECLARE_counter(scans_started); |
| |
| using kudu::client::sp::shared_ptr; |
| using kudu::client::KuduValue; |
| using kudu::cluster::InternalMiniCluster; |
| using kudu::cluster::InternalMiniClusterOptions; |
| using kudu::master::CatalogManager; |
| using kudu::master::TableInfo; |
| using kudu::master::TabletInfo; |
| using kudu::tablet::TabletReplica; |
| using std::string; |
| using std::unique_ptr; |
| using std::vector; |
| using strings::Substitute; |
| |
| static constexpr const char* const kKeyColumn = "key"; |
| static constexpr const char* const kIntValColumn = "int_val"; |
| static constexpr const char* const kStringValColumn = "string_val"; |
| |
| namespace kudu { |
| namespace client { |
| |
| class FlexPartitioningTest : public KuduTest { |
| public: |
| FlexPartitioningTest() { |
| KuduSchemaBuilder b; |
| b.AddColumn(kKeyColumn)-> |
| Type(KuduColumnSchema::INT32)-> |
| NotNull()-> |
| PrimaryKey(); |
| b.AddColumn(kIntValColumn)-> |
| Type(KuduColumnSchema::INT32)-> |
| NotNull(); |
| b.AddColumn(kStringValColumn)-> |
| Type(KuduColumnSchema::STRING)-> |
| Nullable(); |
| CHECK_OK(b.Build(&schema_)); |
| } |
| |
| virtual int num_tablet_servers() const { |
| return 1; |
| } |
| |
| void SetUp() override { |
| KuduTest::SetUp(); |
| |
| // Reduce the TS<->Master heartbeat interval to speed up testing. |
| FLAGS_heartbeat_interval_ms = 10; |
| |
| // Ensure the static pages are not available as tests are written based |
| // on this value of the flag |
| FLAGS_webserver_doc_root = ""; |
| |
| InternalMiniClusterOptions opt; |
| opt.num_tablet_servers = num_tablet_servers(); |
| // Start minicluster and wait for tablet servers to connect to master. |
| cluster_.reset(new InternalMiniCluster(env_, std::move(opt))); |
| ASSERT_OK(cluster_->Start()); |
| |
| // Connect to the cluster. |
| ASSERT_OK(KuduClientBuilder() |
| .add_master_server_addr( |
| cluster_->mini_master()->bound_rpc_addr().ToString()) |
| .Build(&client_)); |
| } |
| |
| protected: |
| typedef unique_ptr<KuduRangePartition> RangePartition; |
| typedef vector<RangePartition> RangePartitions; |
| |
| static Status ApplyInsert(KuduSession* session, |
| const shared_ptr<KuduTable>& table, |
| int32_t key_val, |
| int32_t int_val, |
| string string_val) { |
| unique_ptr<KuduInsert> insert(table->NewInsert()); |
| RETURN_NOT_OK(insert->mutable_row()->SetInt32(kKeyColumn, key_val)); |
| RETURN_NOT_OK(insert->mutable_row()->SetInt32(kIntValColumn, int_val)); |
| RETURN_NOT_OK(insert->mutable_row()->SetStringCopy( |
| kStringValColumn, std::move(string_val))); |
| return session->Apply(insert.release()); |
| } |
| |
| static Status FetchSessionErrors(KuduSession* session, |
| vector<KuduError*>* errors = nullptr) { |
| if (errors) { |
| bool overflowed; |
| session->GetPendingErrors(errors, &overflowed); |
| if (PREDICT_FALSE(overflowed)) { |
| return Status::RuntimeError("session error buffer overflowed"); |
| } |
| } |
| return Status::OK(); |
| } |
| |
| Status InsertTestRows( |
| const char* table_name, |
| int32_t key_beg, |
| int32_t key_end, |
| KuduSession::FlushMode flush_mode = KuduSession::AUTO_FLUSH_SYNC, |
| vector<KuduError*>* errors = nullptr) { |
| CHECK_LE(key_beg, key_end); |
| shared_ptr<KuduTable> table; |
| RETURN_NOT_OK(client_->OpenTable(table_name, &table)); |
| shared_ptr<KuduSession> session = client_->NewSession(); |
| RETURN_NOT_OK(session->SetFlushMode(flush_mode)); |
| session->SetTimeoutMillis(60000); |
| for (int32_t key_val = key_beg; key_val < key_end; ++key_val) { |
| if (const auto s = ApplyInsert(session.get(), |
| table, |
| key_val, |
| rand(), |
| std::to_string(rand())); |
| !s.ok()) { |
| RETURN_NOT_OK(FetchSessionErrors(session.get(), errors)); |
| return s; |
| } |
| } |
| |
| const auto s = session->Flush(); |
| if (!s.ok()) { |
| RETURN_NOT_OK(FetchSessionErrors(session.get(), errors)); |
| } |
| |
| return s; |
| } |
| |
| Status CreateTable(const char* table_name, RangePartitions partitions) { |
| unique_ptr<KuduTableCreator> table_creator(client_->NewTableCreator()); |
| table_creator->table_name(table_name) |
| .schema(&schema_) |
| .num_replicas(1) |
| .set_range_partition_columns({ kKeyColumn }); |
| |
| for (auto& p : partitions) { |
| table_creator->add_custom_range_partition(p.release()); |
| } |
| |
| return table_creator->Create(); |
| } |
| |
| static RangePartition CreateRangePartition(const KuduSchema& schema, |
| const string& key_column, |
| int32_t lower_bound, |
| int32_t upper_bound) { |
| unique_ptr<KuduPartialRow> lower(schema.NewRow()); |
| CHECK_OK(lower->SetInt32(key_column, lower_bound)); |
| unique_ptr<KuduPartialRow> upper(schema.NewRow()); |
| CHECK_OK(upper->SetInt32(key_column, upper_bound)); |
| return unique_ptr<KuduRangePartition>( |
| new KuduRangePartition(lower.release(), upper.release())); |
| } |
| |
| RangePartition CreateRangePartition(int32_t lower_bound = 0, |
| int32_t upper_bound = 100) { |
| return CreateRangePartition(schema_, kKeyColumn, lower_bound, upper_bound); |
| } |
| |
| RangePartition CreateRangePartitionNoLowerBound(int32_t upper_bound) { |
| unique_ptr<KuduPartialRow> upper(schema_.NewRow()); |
| CHECK_OK(upper->SetInt32(kKeyColumn, upper_bound)); |
| return unique_ptr<KuduRangePartition>( |
| new KuduRangePartition(schema_.NewRow(), upper.release())); |
| } |
| |
| RangePartition CreateRangePartitionNoUpperBound(int32_t lower_bound) { |
| unique_ptr<KuduPartialRow> lower(schema_.NewRow()); |
| CHECK_OK(lower->SetInt32(kKeyColumn, lower_bound)); |
| return unique_ptr<KuduRangePartition>( |
| new KuduRangePartition(lower.release(), schema_.NewRow())); |
| } |
| |
| void CheckTabletCount(const char* table_name, int expected_count) { |
| shared_ptr<KuduTable> table; |
| ASSERT_OK(client_->OpenTable(table_name, &table)); |
| |
| scoped_refptr<master::TableInfo> table_info; |
| { |
| auto* cm = cluster_->mini_master(0)->master()->catalog_manager(); |
| CatalogManager::ScopedLeaderSharedLock l(cm); |
| ASSERT_OK(cm->GetTableInfo(table->id(), &table_info)); |
| } |
| ASSERT_EQ(expected_count, table_info->num_tablets()); |
| } |
| |
| void CheckTableRowsNum(const char* table_name, |
| int64_t expected_count) { |
| shared_ptr<KuduTable> table; |
| ASSERT_OK(client_->OpenTable(table_name, &table)); |
| KuduScanner scanner(table.get()); |
| ASSERT_OK(scanner.SetSelection(KuduClient::LEADER_ONLY)); |
| ASSERT_OK(scanner.SetReadMode(KuduScanner::READ_YOUR_WRITES)); |
| ASSERT_OK(scanner.Open()); |
| |
| int64_t count = 0; |
| KuduScanBatch batch; |
| while (scanner.HasMoreRows()) { |
| ASSERT_OK(scanner.NextBatch(&batch)); |
| count += batch.NumRows(); |
| } |
| ASSERT_EQ(expected_count, count); |
| } |
| |
| void CheckScanWithColumnPredicate(const string& table_name, Slice col_name, |
| int expected_count, int tablets, int lower, int upper) { |
| shared_ptr<KuduTable> table; |
| ASSERT_OK(client_->OpenTable(table_name, &table)); |
| |
| KuduScanner scanner(table.get()); |
| ASSERT_OK(scanner.SetTimeoutMillis(60000)); |
| if (lower != INT8_MIN) { |
| ASSERT_OK(scanner.AddConjunctPredicate(table->NewComparisonPredicate( |
| col_name, KuduPredicate::GREATER_EQUAL, KuduValue::FromInt(lower)))); |
| } |
| if (upper != INT8_MAX) { |
| ASSERT_OK(scanner.AddConjunctPredicate(table->NewComparisonPredicate( |
| col_name, KuduPredicate::LESS, KuduValue::FromInt(upper)))); |
| } |
| ASSERT_OK(scanner.Open()); |
| |
| int64_t count = 0; |
| while (scanner.HasMoreRows()) { |
| KuduScanBatch batch; |
| ASSERT_OK(scanner.NextBatch(&batch)); |
| count += batch.NumRows(); |
| } |
| ASSERT_EQ(count, expected_count); |
| |
| NO_FATALS(CheckTabletsScanned(table_name, tablets)); |
| } |
| |
| void CheckTabletsScanned(const string& table_name, int tablets) { |
| shared_ptr<KuduTable> table; |
| ASSERT_OK(client_->OpenTable(table_name, &table)); |
| vector<scoped_refptr<master::TabletInfo>> all_tablets_info; |
| { |
| auto* cm = cluster_->mini_master(0)->master()->catalog_manager(); |
| CatalogManager::ScopedLeaderSharedLock l(cm); |
| scoped_refptr<master::TableInfo> table_info; |
| ASSERT_OK(cm->GetTableInfo(table->id(), &table_info)); |
| table_info->GetAllTablets(&all_tablets_info); |
| } |
| |
| int64_t tablets_scanned = 0; |
| for (const auto& tablet_info : all_tablets_info) { |
| for (auto i = 0; i < cluster_->num_tablet_servers(); ++i) { |
| scoped_refptr<tablet::TabletReplica> replica; |
| ASSERT_TRUE(cluster_->mini_tablet_server(i)->server()-> |
| tablet_manager()->LookupTablet(tablet_info->id(), &replica)); |
| ASSERT_TRUE(replica->tablet()->GetMetricEntity()); |
| auto scans_started = METRIC_scans_started.Instantiate(replica->tablet()->GetMetricEntity()); |
| tablets_scanned += scans_started->value(); |
| scans_started->Reset(); |
| } |
| } |
| ASSERT_EQ(tablets, tablets_scanned); |
| } |
| |
| void CheckTableRowsNum(const char* table_name, const char* col_name, |
| int lower, int upper, int expected_row_count) { |
| shared_ptr<KuduTable> table; |
| ASSERT_OK(client_->OpenTable(table_name, &table)); |
| KuduScanner scanner(table.get()); |
| ASSERT_OK(scanner.SetTimeoutMillis(60000)); |
| ASSERT_OK(scanner.AddConjunctPredicate(table->NewComparisonPredicate( |
| col_name, KuduPredicate::GREATER_EQUAL, KuduValue::FromInt(lower)))); |
| ASSERT_OK(scanner.AddConjunctPredicate(table->NewComparisonPredicate( |
| col_name, KuduPredicate::LESS, KuduValue::FromInt(upper)))); |
| |
| ASSERT_OK(scanner.Open()); |
| KuduScanBatch batch; |
| int live_row_count = 0; |
| while (scanner.HasMoreRows()) { |
| ASSERT_OK(scanner.NextBatch(&batch)); |
| live_row_count += batch.NumRows(); |
| } |
| ASSERT_EQ(expected_row_count, live_row_count); |
| } |
| |
| KuduSchema schema_; |
| unique_ptr<InternalMiniCluster> cluster_; |
| shared_ptr<KuduClient> client_; |
| }; |
| |
| // Test for scenarios covering range partitioning with custom bucket schemas |
| // specified when creating a table. |
| class FlexPartitioningCreateTableTest : public FlexPartitioningTest {}; |
| |
| TEST_F(FlexPartitioningCreateTableTest, TableWideHashSchema) { |
| // Create a table with the following partitions: |
| // |
| // hash bucket |
| // key 0 1 |
| // ------------------------- |
| // <111 x:{key} x:{key} |
| constexpr const char* const kTableName = "TableWideHashSchema"; |
| |
| unique_ptr<KuduTableCreator> table_creator(client_->NewTableCreator()); |
| table_creator->table_name(kTableName) |
| .schema(&schema_) |
| .num_replicas(1) |
| .add_hash_partitions({ kKeyColumn }, 2) |
| .set_range_partition_columns({ kKeyColumn }); |
| |
| // Add a range partition with the table-wide hash partitioning rules. |
| { |
| unique_ptr<KuduPartialRow> lower(schema_.NewRow()); |
| ASSERT_OK(lower->SetInt32(kKeyColumn, INT32_MIN)); |
| unique_ptr<KuduPartialRow> upper(schema_.NewRow()); |
| ASSERT_OK(upper->SetInt32(kKeyColumn, 111)); |
| table_creator->add_range_partition(lower.release(), upper.release()); |
| } |
| |
| ASSERT_OK(table_creator->Create()); |
| NO_FATALS(CheckTabletCount(kTableName, 2)); |
| ASSERT_OK(InsertTestRows(kTableName, -111, 111, KuduSession::MANUAL_FLUSH)); |
| NO_FATALS(CheckTableRowsNum(kTableName, 222)); |
| } |
| |
| TEST_F(FlexPartitioningCreateTableTest, EmptyTableWideHashSchema) { |
| constexpr const char* const kTableName = "EmptyTableWideHashSchema"; |
| |
| unique_ptr<KuduTableCreator> table_creator(client_->NewTableCreator()); |
| table_creator->table_name(kTableName) |
| .schema(&schema_) |
| .num_replicas(1) |
| .set_range_partition_columns({ kKeyColumn }); |
| |
| // Add a range partition with the table-wide hash partitioning rules: no hash |
| // bucketing at all. |
| { |
| unique_ptr<KuduPartialRow> lower(schema_.NewRow()); |
| ASSERT_OK(lower->SetInt32(kKeyColumn, INT32_MIN)); |
| unique_ptr<KuduPartialRow> upper(schema_.NewRow()); |
| ASSERT_OK(upper->SetInt32(kKeyColumn, 111)); |
| table_creator->add_range_partition(lower.release(), upper.release()); |
| } |
| |
| // Add a custom range: no hash bucketing as well. |
| { |
| auto p = CreateRangePartition(111, 222); |
| table_creator->add_custom_range_partition(p.release()); |
| } |
| |
| ASSERT_OK(table_creator->Create()); |
| // There should be 2 tablets total: one per each range created. |
| NO_FATALS(CheckTabletCount(kTableName, 2)); |
| ASSERT_OK(InsertTestRows(kTableName, -111, 222, KuduSession::MANUAL_FLUSH)); |
| NO_FATALS(CheckTableRowsNum(kTableName, 333)); |
| } |
| |
| // Create tables with range partitions using custom hash bucket schemas only. |
| TEST_F(FlexPartitioningCreateTableTest, CustomHashSchemaDiffersFromTableWide) { |
| // Using one-level hash bucketing { 3, "key" } as the custom hash schema |
| // for the newly created range partition. Note that the table-wide hash schema |
| // is empty per FlexPartitioningTest::FlexPartitioningTest(), so the attempt |
| // to create such a table fails with the Status::NotSupported() status. |
| constexpr const char* const kTableName = "3@key"; |
| RangePartitions partitions; |
| partitions.emplace_back(CreateRangePartition(0, 100)); |
| auto& p = partitions.back(); |
| ASSERT_OK(p->add_hash_partitions({ kKeyColumn }, 3, 0)); |
| const auto s = CreateTable(kTableName, std::move(partitions)); |
| ASSERT_TRUE(s.IsNotSupported()) << s.ToString(); |
| ASSERT_STR_CONTAINS(s.ToString(), |
| "varying number of hash dimensions per range is not yet supported"); |
| } |
| |
| // TODO(aserbin): re-enable this scenario once varying hash dimensions per range |
| // are supported |
| TEST_F(FlexPartitioningCreateTableTest, DISABLED_SingleCustomRangeEmptyHashSchema) { |
| // Create a table with the following partitions: |
| // |
| // hash bucket |
| // key 0 1 |
| // ------------------------- |
| // <111 x:{key} x:{key} |
| // 111-222 - - |
| constexpr const char* const kTableName = "SingleCustomRangeEmptyHashSchema"; |
| |
| unique_ptr<KuduTableCreator> table_creator(client_->NewTableCreator()); |
| table_creator->table_name(kTableName) |
| .schema(&schema_) |
| .num_replicas(1) |
| .add_hash_partitions({ kKeyColumn }, 2) |
| .set_range_partition_columns({ kKeyColumn }); |
| |
| // Add a range partition with the table-wide hash partitioning rules. |
| { |
| unique_ptr<KuduPartialRow> lower(schema_.NewRow()); |
| ASSERT_OK(lower->SetInt32(kKeyColumn, INT32_MIN)); |
| unique_ptr<KuduPartialRow> upper(schema_.NewRow()); |
| ASSERT_OK(upper->SetInt32(kKeyColumn, 111)); |
| table_creator->add_range_partition(lower.release(), upper.release()); |
| } |
| |
| // Add a range partition with no hash bucketing. Not calling |
| // KuduRangePartition::add_hash_partitions() on the newly created range means |
| // the range doesn't have any hash bucketing. |
| { |
| auto p = CreateRangePartition(111, 222); |
| table_creator->add_custom_range_partition(p.release()); |
| } |
| |
| ASSERT_OK(table_creator->Create()); |
| NO_FATALS(CheckTabletCount(kTableName, 3)); |
| |
| // Make sure it's possible to insert rows into the table for all the existing |
| // the partitions: first check the range of table-wide schema, then check |
| // the ranges with custom hash schemas. |
| ASSERT_OK(InsertTestRows(kTableName, -111, 0)); |
| NO_FATALS(CheckTableRowsNum(kTableName, 111)); |
| ASSERT_OK(InsertTestRows(kTableName, 111, 222)); |
| NO_FATALS(CheckTableRowsNum(kTableName, 222)); |
| } |
| |
| // Create a table with mixed set of range partitions, using both table-wide and |
| // custom hash bucket schemas. |
| TEST_F(FlexPartitioningCreateTableTest, DefaultAndCustomHashSchemas) { |
| // Create a table with the following partitions: |
| // |
| // hash bucket |
| // key 0 1 2 3 |
| // ----------------------------------------------------------- |
| // <111 x:{key} x:{key} - - |
| // 111-222 x:{key} x:{key} x:{key} - |
| // 222-333 x:{key} x:{key} x:{key} x:{key} |
| // 333-444 x:{key} x:{key} - - |
| constexpr const char* const kTableName = "DefaultAndCustomHashSchemas"; |
| |
| unique_ptr<KuduTableCreator> table_creator(client_->NewTableCreator()); |
| table_creator->table_name(kTableName) |
| .schema(&schema_) |
| .num_replicas(1) |
| .add_hash_partitions({ kKeyColumn }, 2) |
| .set_range_partition_columns({ kKeyColumn }); |
| |
| // Add a range partition with the table-wide hash partitioning rules. |
| { |
| unique_ptr<KuduPartialRow> lower(schema_.NewRow()); |
| ASSERT_OK(lower->SetInt32(kKeyColumn, INT32_MIN)); |
| unique_ptr<KuduPartialRow> upper(schema_.NewRow()); |
| ASSERT_OK(upper->SetInt32(kKeyColumn, 111)); |
| table_creator->add_range_partition(lower.release(), upper.release()); |
| } |
| |
| // Add a range partition with custom hash sub-partitioning rules: |
| // 3 buckets with hash based on the "key" column with hash seed 1. |
| { |
| auto p = CreateRangePartition(111, 222); |
| ASSERT_OK(p->add_hash_partitions({ kKeyColumn }, 3, 1)); |
| table_creator->add_custom_range_partition(p.release()); |
| } |
| |
| // Add a range partition with custom hash sub-partitioning rules: |
| // 4 buckets with hash based on the "key" column with hash seed 2. |
| { |
| auto p = CreateRangePartition(222, 333); |
| ASSERT_OK(p->add_hash_partitions({ kKeyColumn }, 4, 2)); |
| table_creator->add_custom_range_partition(p.release()); |
| } |
| |
| // Add a range partition with custom hash sub-partitioning rules: |
| // 2 buckets hashing on the "key" column with hash seed 3. |
| { |
| auto p = CreateRangePartition(333, 444); |
| ASSERT_OK(p->add_hash_partitions({ kKeyColumn }, 2, 3)); |
| table_creator->add_custom_range_partition(p.release()); |
| } |
| |
| ASSERT_OK(table_creator->Create()); |
| NO_FATALS(CheckTabletCount(kTableName, 11)); |
| |
| // Make sure it's possible to insert rows into the table for all the existing |
| // the partitions: first check the range of table-wide schema, then check |
| // the ranges with custom hash schemas. |
| ASSERT_OK(InsertTestRows(kTableName, -111, 0)); |
| NO_FATALS(CheckTableRowsNum(kTableName, 111)); |
| ASSERT_OK(InsertTestRows(kTableName, 111, 444)); |
| NO_FATALS(CheckTableRowsNum(kTableName, 444)); |
| |
| // Meanwhile, inserting into non-covered ranges should result in a proper |
| // error status return to the client attempting such an operation. |
| { |
| vector<KuduError*> errors; |
| ElementDeleter drop(&errors); |
| auto s = InsertTestRows( |
| kTableName, 444, 445, KuduSession::AUTO_FLUSH_SYNC, &errors); |
| ASSERT_TRUE(s.IsIOError()) << s.ToString(); |
| ASSERT_STR_CONTAINS(s.ToString(), "failed to flush data"); |
| ASSERT_EQ(1, errors.size()); |
| const auto& err = errors[0]->status(); |
| EXPECT_TRUE(err.IsNotFound()) << err.ToString(); |
| ASSERT_STR_CONTAINS(err.ToString(), |
| "No tablet covering the requested range partition"); |
| } |
| // Try same as in the scope above, but do so for multiple rows to make sure |
| // custom hash bucketing isn't inducing any unexpected side effects. |
| { |
| constexpr int kNumRows = 10; |
| vector<KuduError*> errors; |
| ElementDeleter drop(&errors); |
| auto s = InsertTestRows( |
| kTableName, 445, 445 + kNumRows, KuduSession::MANUAL_FLUSH, &errors); |
| ASSERT_TRUE(s.IsIOError()) << s.ToString(); |
| ASSERT_STR_CONTAINS(s.ToString(), "failed to flush data"); |
| ASSERT_EQ(kNumRows, errors.size()); |
| for (const auto& e : errors) { |
| const auto& err = e->status(); |
| EXPECT_TRUE(err.IsNotFound()) << err.ToString(); |
| ASSERT_STR_CONTAINS(err.ToString(), |
| "No tablet covering the requested range partition"); |
| } |
| } |
| } |
| |
| TEST_F(FlexPartitioningCreateTableTest, TabletServerWebUI) { |
| // Create a table with the following partitions: |
| // |
| // hash bucket |
| // key 0 1 2 3 |
| // ----------------------------------------------------------- |
| // <111 x:{key} x:{key} - - |
| // 111-222 x:{key} x:{key} x:{key} - |
| // 222-333 x:{key} x:{key} x:{key} x:{key} |
| constexpr const char* const kTableName = "TabletServerWebUI"; |
| |
| unique_ptr<KuduTableCreator> table_creator(client_->NewTableCreator()); |
| table_creator->table_name(kTableName) |
| .schema(&schema_) |
| .num_replicas(1) |
| .add_hash_partitions({ kKeyColumn }, 2) |
| .set_range_partition_columns({ kKeyColumn }); |
| |
| // Add a range partition with the table-wide hash partitioning rules. |
| { |
| unique_ptr<KuduPartialRow> lower(schema_.NewRow()); |
| ASSERT_OK(lower->SetInt32(kKeyColumn, INT32_MIN)); |
| unique_ptr<KuduPartialRow> upper(schema_.NewRow()); |
| ASSERT_OK(upper->SetInt32(kKeyColumn, 111)); |
| table_creator->add_range_partition(lower.release(), upper.release()); |
| } |
| |
| // Add a range partition with custom hash sub-partitioning rules: |
| // 3 buckets with hash based on the "key" column with hash seed 1. |
| { |
| auto p = CreateRangePartition(111, 222); |
| ASSERT_OK(p->add_hash_partitions({ kKeyColumn }, 3, 1)); |
| table_creator->add_custom_range_partition(p.release()); |
| } |
| |
| // Add a range partition with custom hash sub-partitioning rules: |
| // 4 buckets with hash based on the "key" column with hash seed 2. |
| { |
| auto p = CreateRangePartition(222, 333); |
| ASSERT_OK(p->add_hash_partitions({ kKeyColumn }, 4, 2)); |
| table_creator->add_custom_range_partition(p.release()); |
| } |
| |
| ASSERT_OK(table_creator->Create()); |
| NO_FATALS(CheckTabletCount(kTableName, 9)); |
| |
| // Obtain the web page contents |
| EasyCurl c; |
| faststring buf; |
| ASSERT_OK(c.FetchURL(Substitute("http://$0/tablets", |
| cluster_->mini_tablet_server(0)->bound_http_addr().ToString()), |
| &buf)); |
| string raw = buf.ToString(); |
| |
| // Get the list of tablets present in this table |
| std::vector<scoped_refptr<TableInfo>> tables; |
| { |
| CatalogManager::ScopedLeaderSharedLock l( |
| cluster_->mini_master(0)->master()->catalog_manager()); |
| cluster_->mini_master(0)->master()->catalog_manager()->GetAllTables(&tables); |
| } |
| ASSERT_EQ(1, tables.size()); |
| vector<scoped_refptr<TabletInfo>> tablets; |
| tables.front()->GetAllTablets(&tablets); |
| |
| ASSERT_EQ(9, tablets.size()); |
| // Validate the partition information rendered in the page |
| ASSERT_STR_CONTAINS(raw, Substitute("id=$0\"},\"partition\":\"HASH (key) PARTITION 0, " |
| "RANGE (key) PARTITION -2147483648 <= VALUES < 111", |
| tablets[0]->id())); |
| ASSERT_STR_CONTAINS(raw, Substitute("id=$0\"},\"partition\":\"HASH (key) PARTITION 0, " |
| "RANGE (key) PARTITION 111 <= VALUES < 222", |
| tablets[1]->id())); |
| ASSERT_STR_CONTAINS(raw, Substitute("id=$0\"},\"partition\":\"HASH (key) PARTITION 0, " |
| "RANGE (key) PARTITION 222 <= VALUES < 333", |
| tablets[2]->id())); |
| ASSERT_STR_CONTAINS(raw, Substitute("id=$0\"},\"partition\":\"HASH (key) PARTITION 1, " |
| "RANGE (key) PARTITION -2147483648 <= VALUES < 111", |
| tablets[3]->id())); |
| ASSERT_STR_CONTAINS(raw, Substitute("id=$0\"},\"partition\":\"HASH (key) PARTITION 1, " |
| "RANGE (key) PARTITION 111 <= VALUES < 222", |
| tablets[4]->id())); |
| ASSERT_STR_CONTAINS(raw, Substitute("id=$0\"},\"partition\":\"HASH (key) PARTITION 1, " |
| "RANGE (key) PARTITION 222 <= VALUES < 333", |
| tablets[5]->id())); |
| ASSERT_STR_CONTAINS(raw, Substitute("id=$0\"},\"partition\":\"HASH (key) PARTITION 2, " |
| "RANGE (key) PARTITION 111 <= VALUES < 222", |
| tablets[6]->id())); |
| ASSERT_STR_CONTAINS(raw, Substitute("id=$0\"},\"partition\":\"HASH (key) PARTITION 2, " |
| "RANGE (key) PARTITION 222 <= VALUES < 333", |
| tablets[7]->id())); |
| ASSERT_STR_CONTAINS(raw, Substitute("id=$0\"},\"partition\":\"HASH (key) PARTITION 3, " |
| "RANGE (key) PARTITION 222 <= VALUES < 333", |
| tablets[8]->id())); |
| } |
| |
| // Parameters for a single hash dimension. |
| struct HashDimensionParameters { |
| vector<string> columns; // names of the columns to use for hash bucketing |
| int32_t num_buckets; // number of hash buckets |
| }; |
| std::ostream& operator<<(std::ostream& os, |
| const HashDimensionParameters& params) { |
| os << "{columns:{"; |
| for (const auto& c : params.columns) { |
| os << c << ","; |
| } |
| os << "},"; |
| os << "num_buckets:" << params.num_buckets << ","; |
| return os; |
| } |
| |
| // Hash bucketing parameters for a range partition. |
| typedef vector<HashDimensionParameters> HashPartitionParameters; |
| std::ostream& operator<<(std::ostream& os, |
| const HashPartitionParameters& params) { |
| os << "{"; |
| for (const auto& e : params) { |
| os << e << ","; |
| } |
| os << "}"; |
| return os; |
| } |
| |
| // Range partition bounds. |
| struct RangeBounds { |
| int32_t lower; // inclusive lower bound |
| int32_t upper; // exclusive upper bound |
| }; |
| |
| // The whole set of parameters per range. |
| struct PerRangeParameters { |
| RangeBounds bounds; |
| HashPartitionParameters hash_params; |
| }; |
| std::ostream& operator<<(std::ostream& os, |
| const PerRangeParameters& params) { |
| os << "{bounds:[" |
| << params.bounds.lower << "," |
| << params.bounds.upper << "),{"; |
| for (const auto& hp : params.hash_params) { |
| os << hp << ","; |
| } |
| os << "}}"; |
| return os; |
| } |
| |
| typedef vector<PerRangeParameters> TablePartitionParameters; |
| std::ostream& operator<<(std::ostream& os, |
| const TablePartitionParameters& params) { |
| os << "{"; |
| for (const auto& per_range_params : params) { |
| os << per_range_params; |
| } |
| os << "}"; |
| return os; |
| } |
| |
| struct PerRangeTestParameters { |
| bool allowed; |
| TablePartitionParameters partition_params; |
| vector<HashDimensionParameters> table_wide_hash_schema_params; |
| }; |
| std::ostream& operator<<(std::ostream& os, |
| const PerRangeTestParameters& params) { |
| os << "allowed:" << std::boolalpha << params.allowed |
| << " partition_params:" << params.partition_params; |
| return os; |
| } |
| |
| class FlexPartitioningCreateTableParamTest : |
| public FlexPartitioningCreateTableTest, |
| public testing::WithParamInterface<PerRangeTestParameters> { |
| public: |
| FlexPartitioningCreateTableParamTest() { |
| KuduSchemaBuilder b; |
| b.AddColumn(kKeyColumn)->Type(KuduColumnSchema::INT32)->NotNull(); |
| b.AddColumn(kIntValColumn)->Type(KuduColumnSchema::INT32)->NotNull(); |
| b.AddColumn(kStringValColumn)->Type(KuduColumnSchema::STRING); |
| b.SetPrimaryKey({ kKeyColumn, kIntValColumn }); |
| CHECK_OK(b.Build(&schema_)); |
| } |
| |
| void Run() { |
| constexpr const char* const kTableName = "PerRangeHashSchemaVariations"; |
| unique_ptr<KuduTableCreator> table_creator(client_->NewTableCreator()); |
| table_creator->table_name(kTableName) |
| .schema(&schema_) |
| .num_replicas(1) |
| .set_range_partition_columns({ kKeyColumn }); |
| |
| const PerRangeTestParameters& params = GetParam(); |
| for (const auto& per_range_params : params.partition_params) { |
| auto p = CreateRangePartition(per_range_params.bounds.lower, |
| per_range_params.bounds.upper); |
| const auto& hash_params = per_range_params.hash_params; |
| for (const auto& hd : hash_params) { |
| ASSERT_OK(p->add_hash_partitions(hd.columns, hd.num_buckets, 0)); |
| } |
| table_creator->add_custom_range_partition(p.release()); |
| } |
| |
| for (const auto& table_wide_params : params.table_wide_hash_schema_params) { |
| table_creator->add_hash_partitions(table_wide_params.columns, |
| table_wide_params.num_buckets); |
| } |
| |
| if (params.allowed) { |
| ASSERT_OK(table_creator->Create()); |
| } else { |
| const auto s = table_creator->Create(); |
| ASSERT_TRUE(s.IsNotSupported()) << s.ToString(); |
| ASSERT_STR_CONTAINS(s.ToString(), |
| "varying number of hash dimensions per range is not yet supported"); |
| } |
| } |
| }; |
| |
| const vector<PerRangeTestParameters> kPerRangeTestParameters = { |
| // A single range with no hashing at all. |
| { |
| true, |
| { |
| { { INT32_MIN, 0 }, {} }, |
| }, |
| {} // empty table-wide hash schema |
| }, |
| // A single range with minimal hash schema. |
| { |
| true, |
| { |
| { { INT32_MIN, 0 }, { { { kKeyColumn }, 2, } } }, |
| }, |
| { |
| { { kKeyColumn }, 3 }, |
| } |
| }, |
| // A single range with hashing on both columns of the primary key. |
| { |
| true, |
| { |
| { { INT32_MIN, 0 }, { { { kKeyColumn, kIntValColumn }, 2, } } }, |
| }, |
| { |
| { { kKeyColumn }, 5 }, |
| } |
| }, |
| // Varying the number of hash buckets. |
| { |
| true, |
| { |
| { { 0, 100 }, { { { kKeyColumn }, 2, } } }, |
| { { 100, 200 }, { { { kKeyColumn }, 3, } } }, |
| { { 200, 300 }, { { { kKeyColumn }, 2, } } }, |
| { { 300, 400 }, { { { kKeyColumn }, 10, } } }, |
| }, |
| { |
| { { kIntValColumn }, 5 }, |
| } |
| }, |
| // Varying the set of columns for hash bucketing. |
| { |
| true, |
| { |
| { { 100, 200 }, { { { kKeyColumn }, 2, } } }, |
| { { 200, 300 }, { { { kKeyColumn, kIntValColumn }, 2, } } }, |
| { { 300, 400 }, { { { kIntValColumn }, 2, } } }, |
| { { 400, 500 }, { { { kIntValColumn, kKeyColumn }, 2, } } }, |
| }, |
| { |
| { { kIntValColumn }, 3 }, |
| } |
| }, |
| // Varying the set of columns for hash bucketing and number of buckets. |
| { |
| true, |
| { |
| { { 200, 300 }, { { { kKeyColumn }, 10, } } }, |
| { { 300, 400 }, { { { kKeyColumn, kIntValColumn }, 2, } } }, |
| { { 400, 500 }, { { { kKeyColumn, kIntValColumn }, 5, } } }, |
| { { 500, 600 }, { { { kIntValColumn }, 8, } } }, |
| { { 600, 700 }, { { { kKeyColumn }, 3, } } }, |
| { { 700, 800 }, { { { kIntValColumn, kKeyColumn }, 16, } } }, |
| { { 800, 900 }, { { { kIntValColumn }, 32, } } }, |
| }, |
| { |
| { { kIntValColumn }, 7 }, |
| } |
| }, |
| // Below are multiple scenarios with varying number of hash dimensions |
| // for hash bucketing. |
| { |
| false, |
| { |
| { { INT32_MIN, 0 }, {} }, |
| { { 0, 100 }, { { { kKeyColumn }, 2 } } }, |
| }, |
| {} // an empty table-wide schema |
| }, |
| { |
| false, |
| { |
| { { INT32_MIN, 0 }, { { { kKeyColumn }, 3 } } }, |
| { { 0, 100 }, {} }, |
| }, |
| { |
| { { kKeyColumn }, 8 }, |
| } |
| }, |
| { |
| false, |
| { |
| { { INT32_MIN, 0 }, {} }, |
| { { 0, 100 }, { { { kKeyColumn }, 2 } } }, |
| { { 100, 200 }, {} }, |
| }, |
| {} // an empty table-wide schema |
| }, |
| { |
| false, |
| { |
| { { INT32_MIN, 0 }, { { { kKeyColumn }, 3 } } }, |
| { { 0, 100 }, {} }, |
| { { 100, 200 }, { { { kIntValColumn }, 2 } } }, |
| }, |
| {} // an empty table-wide schema |
| }, |
| { |
| false, |
| { |
| { { 0, 100 }, { { { kKeyColumn }, 2 }, { { kIntValColumn }, 2 } } }, |
| { { 100, 200 }, { { { kKeyColumn, kIntValColumn }, 2 } } }, |
| }, |
| { |
| { { kKeyColumn }, 7 }, |
| { { kIntValColumn }, 3 }, |
| } |
| }, |
| { |
| false, |
| { |
| { { 100, 200 }, { { { kKeyColumn }, 2 }, { { kIntValColumn }, 3 } } }, |
| { { 200, 300 }, { { { kKeyColumn }, 3 } } }, |
| }, |
| { |
| { { kKeyColumn }, 7 }, |
| } |
| }, |
| { |
| false, |
| { |
| { { 300, 400 }, { { { kKeyColumn }, 2 }, { { kIntValColumn }, 3 } } }, |
| { { 400, 500 }, { { { kKeyColumn }, 3 } } }, |
| }, |
| { |
| { { kKeyColumn }, 7 }, |
| } |
| }, |
| { |
| false, |
| { |
| { { 500, 600 }, { { { kKeyColumn }, 10 }, { { kIntValColumn }, 5 } } }, |
| { { 600, 700 }, { { { kKeyColumn }, 2 } } }, |
| { { 700, INT32_MAX }, {} }, |
| }, |
| {} // an empty table-wide schema |
| }, |
| }; |
| INSTANTIATE_TEST_SUITE_P(Variations, FlexPartitioningCreateTableParamTest, |
| testing::ValuesIn(kPerRangeTestParameters)); |
| TEST_P(FlexPartitioningCreateTableParamTest, PerRangeHashSchemaVariations) { |
| NO_FATALS(Run()); |
| } |
| |
| TEST_F(FlexPartitioningCreateTableTest, CustomHashSchemasVaryingDimensionsNumber) { |
| constexpr const char* const kTableName = "CustomHashSchemasVaryingDimensions"; |
| |
| // Try creating a table with the following partitions: |
| // |
| // hash bucket |
| // key 0 1 |
| // ------------------------- |
| // <111 x:{key} x:{key} |
| // 111-222 - - |
| unique_ptr<KuduTableCreator> table_creator(client_->NewTableCreator()); |
| table_creator->table_name(kTableName) |
| .schema(&schema_) |
| .num_replicas(1) |
| .add_hash_partitions({ kKeyColumn }, 2) |
| .set_range_partition_columns({ kKeyColumn }); |
| |
| // Add a range partition with the table-wide hash partitioning rules. |
| { |
| unique_ptr<KuduPartialRow> lower(schema_.NewRow()); |
| ASSERT_OK(lower->SetInt32(kKeyColumn, INT32_MIN)); |
| unique_ptr<KuduPartialRow> upper(schema_.NewRow()); |
| ASSERT_OK(upper->SetInt32(kKeyColumn, 111)); |
| table_creator->add_range_partition(lower.release(), upper.release()); |
| } |
| |
| // Add a range partition with no hash bucketing. Not calling |
| // KuduRangePartition::add_hash_partitions() on the newly created range |
| // means the range doesn't have any hash bucketing. |
| { |
| auto p = CreateRangePartition(111, 222); |
| table_creator->add_custom_range_partition(p.release()); |
| } |
| |
| const auto s = table_creator->Create(); |
| ASSERT_TRUE(s.IsNotSupported()) << s.ToString(); |
| ASSERT_STR_CONTAINS(s.ToString(), |
| "varying number of hash dimensions per range is not yet supported"); |
| } |
| |
| // This test scenario creates a table with a range partition having no upper |
| // bound. The range partition has a custom empty hash schema (i.e. no hash |
| // bucketing for the range) in the presence of non-empty table-wide hash schema. |
| // |
| // TODO(aserbin): re-enable this scenario once varying hash dimensions per range |
| // are supported |
| TEST_F(FlexPartitioningCreateTableTest, DISABLED_NoUpperBoundRangeCustomHashSchema) { |
| // Create a table with the following partitions: |
| // |
| // hash bucket |
| // key 0 1 2 |
| // -------------------------------- |
| // 0-111 x:{key} x:{key} x:{key} |
| // 111-222 x:{key} x:{key} - |
| // >=222 - - - |
| constexpr const char* const kTableName = "NoUpperBoundRangeCustomHashSchema"; |
| |
| unique_ptr<KuduTableCreator> table_creator(client_->NewTableCreator()); |
| table_creator->table_name(kTableName) |
| .schema(&schema_) |
| .num_replicas(1) |
| .add_hash_partitions({ kKeyColumn }, 3) |
| .set_range_partition_columns({ kKeyColumn }); |
| |
| // Add a range partition with the table-wide hash partitioning rules. |
| { |
| unique_ptr<KuduPartialRow> lower(schema_.NewRow()); |
| ASSERT_OK(lower->SetInt32(kKeyColumn, 0)); |
| unique_ptr<KuduPartialRow> upper(schema_.NewRow()); |
| ASSERT_OK(upper->SetInt32(kKeyColumn, 111)); |
| table_creator->add_range_partition(lower.release(), upper.release()); |
| } |
| |
| // Add a range partition with custom hash sub-partitioning rules: |
| // 2 buckets with hash based on the "key" column with hash seed 1. |
| { |
| auto p = CreateRangePartition(111, 222); |
| ASSERT_OK(p->add_hash_partitions({ kKeyColumn }, 2, 0)); |
| table_creator->add_custom_range_partition(p.release()); |
| } |
| |
| // Add unbounded range partition with no hash bucketing. |
| { |
| auto p = CreateRangePartitionNoUpperBound(222); |
| table_creator->add_custom_range_partition(p.release()); |
| } |
| |
| ASSERT_OK(table_creator->Create()); |
| NO_FATALS(CheckTabletCount(kTableName, 6)); |
| |
| // Make sure it's possible to insert rows into the table for all the existing |
| // partitions. |
| ASSERT_OK(InsertTestRows(kTableName, 0, 555)); |
| NO_FATALS(CheckTableRowsNum(kTableName, 555)); |
| } |
| |
| // Create a table unbounded ranges with custom hash schemas. |
| TEST_F(FlexPartitioningCreateTableTest, UnboundedRangesWithCustomHashSchemas) { |
| constexpr const char* const kTableName = |
| "UnboundedRangesWithCustomHashSchemas"; |
| |
| unique_ptr<KuduTableCreator> table_creator(client_->NewTableCreator()); |
| table_creator->table_name(kTableName) |
| .schema(&schema_) |
| .num_replicas(1) |
| .add_hash_partitions({ kKeyColumn }, 2) |
| .set_range_partition_columns({ kKeyColumn }); |
| |
| // Add a range partition with custom hash sub-partitioning rules: |
| // 3 buckets with hash based on the "key" column with hash seed 1. |
| { |
| auto p = CreateRangePartitionNoLowerBound(0); |
| ASSERT_OK(p->add_hash_partitions({ kKeyColumn }, 3, 1)); |
| table_creator->add_custom_range_partition(p.release()); |
| } |
| |
| // Add a range partition with custom hash sub-partitioning rules: |
| // 5 buckets with hash based on the "key" column with hash seed 2. |
| { |
| auto p = CreateRangePartitionNoUpperBound(0); |
| ASSERT_OK(p->add_hash_partitions({ kKeyColumn }, 5, 2)); |
| table_creator->add_custom_range_partition(p.release()); |
| } |
| |
| ASSERT_OK(table_creator->Create()); |
| NO_FATALS(CheckTabletCount(kTableName, 8)); |
| |
| ASSERT_OK(InsertTestRows(kTableName, -250, -125)); |
| NO_FATALS(CheckTableRowsNum(kTableName, 125)); |
| ASSERT_OK(InsertTestRows(kTableName, 125, 250)); |
| NO_FATALS(CheckTableRowsNum(kTableName, 250)); |
| ASSERT_OK(InsertTestRows(kTableName, -125, 125)); |
| NO_FATALS(CheckTableRowsNum(kTableName, 500)); |
| } |
| |
| // Similar to the UnboundedRangesWithCustomHashSchemas above, but with |
| // additional range having table-wide table schema. |
| TEST_F(FlexPartitioningCreateTableTest, |
| TwoUnboundedRangesWithCustomHashSchemasAndOneInBetween) { |
| constexpr const char* const kTableName = |
| "TwoUnboundedRangesWithCustomHashSchemasAndOneInBetween"; |
| |
| unique_ptr<KuduTableCreator> table_creator(client_->NewTableCreator()); |
| table_creator->table_name(kTableName) |
| .schema(&schema_) |
| .num_replicas(1) |
| .add_hash_partitions({ kKeyColumn }, 3) |
| .set_range_partition_columns({ kKeyColumn }); |
| |
| // Add a range partition with the table-wide hash partitioning rules. |
| { |
| unique_ptr<KuduPartialRow> lower(schema_.NewRow()); |
| ASSERT_OK(lower->SetInt32(kKeyColumn, -111)); |
| unique_ptr<KuduPartialRow> upper(schema_.NewRow()); |
| ASSERT_OK(upper->SetInt32(kKeyColumn, 111)); |
| table_creator->add_range_partition(lower.release(), upper.release()); |
| } |
| |
| // Add a range partition with custom hash sub-partitioning rules: |
| // 2 buckets with hash based on the "key" column with hash seed 1. |
| { |
| auto p = CreateRangePartitionNoLowerBound(-111); |
| ASSERT_OK(p->add_hash_partitions({ kKeyColumn }, 2, 1)); |
| table_creator->add_custom_range_partition(p.release()); |
| } |
| { |
| auto p = CreateRangePartitionNoUpperBound(111); |
| ASSERT_OK(p->add_hash_partitions({ kKeyColumn }, 5, 2)); |
| table_creator->add_custom_range_partition(p.release()); |
| } |
| |
| ASSERT_OK(table_creator->Create()); |
| NO_FATALS(CheckTabletCount(kTableName, 10)); // 3 + 2 + 5 = 10 |
| |
| // Make sure it's possible to insert rows into the table for all the existing |
| // partitions. |
| ASSERT_OK(InsertTestRows(kTableName, -500, 500)); |
| NO_FATALS(CheckTableRowsNum(kTableName, 1000)); |
| } |
| |
| TEST_F(FlexPartitioningCreateTableTest, ScansWithRangePredicates) { |
| |
| constexpr const char* const kTableName = "ScansWithRangePredicates"; |
| |
| unique_ptr<KuduTableCreator> table_creator(client_->NewTableCreator()); |
| table_creator->table_name(kTableName) |
| .schema(&schema_) |
| .num_replicas(1) |
| .add_hash_partitions({ kKeyColumn }, 2) |
| .set_range_partition_columns({ kKeyColumn }); |
| |
| // Add a range partition with the table-wide hash partitioning rules. |
| { |
| unique_ptr<KuduPartialRow> lower(schema_.NewRow()); |
| ASSERT_OK(lower->SetInt32(kKeyColumn, 0)); |
| unique_ptr<KuduPartialRow> upper(schema_.NewRow()); |
| ASSERT_OK(upper->SetInt32(kKeyColumn, 111)); |
| table_creator->add_range_partition(lower.release(), upper.release()); |
| } |
| |
| // Add a range partition with custom hash sub-partitioning rules: |
| // 4 buckets with hash based on the "key" column with hash seed 1. |
| { |
| auto p = CreateRangePartition(111, 222); |
| ASSERT_OK(p->add_hash_partitions({ kKeyColumn }, 4, 1)); |
| table_creator->add_custom_range_partition(p.release()); |
| } |
| |
| // Add a range partition with custom hash sub-partitioning rules: |
| // 3 buckets with hash based on the "key" column with hash seed 1. |
| { |
| auto p = CreateRangePartition(222, 333); |
| ASSERT_OK(p->add_hash_partitions({ kKeyColumn }, 3, 1)); |
| table_creator->add_custom_range_partition(p.release()); |
| } |
| |
| ASSERT_OK(table_creator->Create()); |
| NO_FATALS(CheckTabletCount(kTableName, 9)); |
| |
| ASSERT_OK(InsertTestRows(kTableName, 0, 111)); |
| NO_FATALS(CheckTableRowsNum(kTableName, 111)); |
| NO_FATALS(CheckTabletsScanned(kTableName, 9)); |
| |
| ASSERT_OK(InsertTestRows(kTableName, 111, 222)); |
| NO_FATALS(CheckTableRowsNum(kTableName, 222)); |
| NO_FATALS(CheckTabletsScanned(kTableName, 9)); |
| |
| ASSERT_OK(InsertTestRows(kTableName, 222, 333)); |
| NO_FATALS(CheckTableRowsNum(kTableName, 333)); |
| NO_FATALS(CheckTabletsScanned(kTableName, 9)); |
| |
| // Filler values for lower or upper bound when it's not meant to be set. |
| constexpr int lower_filler = INT8_MIN; |
| constexpr int upper_filler = INT8_MAX; |
| |
| CheckScanWithColumnPredicate(kTableName, "key", 150, 6, 0, 150); |
| CheckScanWithColumnPredicate(kTableName, "key", 183, 7, 150, 333); |
| CheckScanWithColumnPredicate(kTableName, "key", 75, 2, 0, 75); |
| CheckScanWithColumnPredicate(kTableName, "key", 175, 9, 75, 250); |
| CheckScanWithColumnPredicate(kTableName, "key", 83, 3, 250, 333); |
| CheckScanWithColumnPredicate(kTableName, "key", 1, 1, 0, 1); |
| CheckScanWithColumnPredicate(kTableName, "key", 2, 6, 110, 112); |
| CheckScanWithColumnPredicate(kTableName, "key", 0, 0, -10, -5); |
| CheckScanWithColumnPredicate(kTableName, "key", 0, 0, 350, 400); |
| CheckScanWithColumnPredicate(kTableName, "key", 1, 1, 332, 333); |
| CheckScanWithColumnPredicate(kTableName, "key", 0, 0, 333, 334); |
| CheckScanWithColumnPredicate(kTableName, "key", 100, 2, lower_filler, 100); |
| CheckScanWithColumnPredicate(kTableName, "key", 250, 9, lower_filler, 250); |
| CheckScanWithColumnPredicate(kTableName, "key", 333, 9, lower_filler, 333); |
| CheckScanWithColumnPredicate(kTableName, "key", 333, 9, lower_filler, 334); |
| CheckScanWithColumnPredicate(kTableName, "key", 333, 9, -10, upper_filler); |
| CheckScanWithColumnPredicate(kTableName, "key", 233, 9, 100, upper_filler); |
| CheckScanWithColumnPredicate(kTableName, "key", 133, 7, 200, upper_filler); |
| CheckScanWithColumnPredicate(kTableName, "key", 83, 3, 250, upper_filler); |
| CheckScanWithColumnPredicate(kTableName, "key", 0, 0, 333, upper_filler); |
| |
| // Meanwhile, inserting into non-covered ranges should result in a proper |
| // error status return to the client attempting such an operation. |
| { |
| constexpr int kNumRows = 10; |
| vector<KuduError*> errors; |
| ElementDeleter drop(&errors); |
| auto s = InsertTestRows( |
| kTableName, 334, 334 + kNumRows, KuduSession::MANUAL_FLUSH, &errors); |
| ASSERT_TRUE(s.IsIOError()) << s.ToString(); |
| ASSERT_STR_CONTAINS(s.ToString(), "failed to flush data"); |
| ASSERT_EQ(kNumRows, errors.size()); |
| for (const auto& e : errors) { |
| const auto& err = e->status(); |
| EXPECT_TRUE(err.IsNotFound()) << err.ToString(); |
| ASSERT_STR_CONTAINS(err.ToString(), |
| "No tablet covering the requested range partition"); |
| } |
| } |
| } |
| |
| TEST_F(FlexPartitioningCreateTableTest, ScansWithRangePredicatesWithSameHashBuckets) { |
| |
| constexpr const char* const kTableName = "ScansWithRangePredicatesWithSameHashBuckets"; |
| |
| unique_ptr<KuduTableCreator> table_creator(client_->NewTableCreator()); |
| table_creator->table_name(kTableName) |
| .schema(&schema_) |
| .num_replicas(1) |
| .add_hash_partitions({ kKeyColumn }, 3) |
| .set_range_partition_columns({ kKeyColumn }); |
| |
| // Add a range partition with the table-wide hash partitioning rules. |
| { |
| unique_ptr<KuduPartialRow> lower(schema_.NewRow()); |
| ASSERT_OK(lower->SetInt32(kKeyColumn, 0)); |
| unique_ptr<KuduPartialRow> upper(schema_.NewRow()); |
| ASSERT_OK(upper->SetInt32(kKeyColumn, 111)); |
| table_creator->add_range_partition(lower.release(), upper.release()); |
| } |
| |
| // Add a range partition with the table-wide hash partitioning rules. |
| { |
| unique_ptr<KuduPartialRow> lower(schema_.NewRow()); |
| ASSERT_OK(lower->SetInt32(kKeyColumn, 111)); |
| unique_ptr<KuduPartialRow> upper(schema_.NewRow()); |
| ASSERT_OK(upper->SetInt32(kKeyColumn, 222)); |
| table_creator->add_range_partition(lower.release(), upper.release()); |
| } |
| |
| // Add a range partition with the table-wide hash partitioning rules. |
| { |
| unique_ptr<KuduPartialRow> lower(schema_.NewRow()); |
| ASSERT_OK(lower->SetInt32(kKeyColumn, 222)); |
| unique_ptr<KuduPartialRow> upper(schema_.NewRow()); |
| ASSERT_OK(upper->SetInt32(kKeyColumn, 333)); |
| table_creator->add_range_partition(lower.release(), upper.release()); |
| } |
| |
| |
| ASSERT_OK(table_creator->Create()); |
| NO_FATALS(CheckTabletCount(kTableName, 9)); |
| |
| ASSERT_OK(InsertTestRows(kTableName, 0, 111)); |
| NO_FATALS(CheckTableRowsNum(kTableName, 111)); |
| NO_FATALS(CheckTabletsScanned(kTableName, 9)); |
| |
| ASSERT_OK(InsertTestRows(kTableName, 111, 222)); |
| NO_FATALS(CheckTableRowsNum(kTableName, 222)); |
| NO_FATALS(CheckTabletsScanned(kTableName, 9)); |
| |
| ASSERT_OK(InsertTestRows(kTableName, 222, 333)); |
| NO_FATALS(CheckTableRowsNum(kTableName, 333)); |
| NO_FATALS(CheckTabletsScanned(kTableName, 9)); |
| |
| // Filler values for lower or upper bound when it's not meant to be set. |
| constexpr int lower_filler = INT8_MIN; |
| constexpr int upper_filler = INT8_MAX; |
| |
| // The test cases commented out below fail. Initially, it was thought to be a regression but these |
| // failures are not reproducible in a real cluster. |
| CheckScanWithColumnPredicate(kTableName, "key", 150, 6, 0, 150); |
| CheckScanWithColumnPredicate(kTableName, "key", 183, 6, 150, 333); |
| CheckScanWithColumnPredicate(kTableName, "key", 75, 3, 0, 75); |
| CheckScanWithColumnPredicate(kTableName, "key", 175, 9, 75, 250); |
| CheckScanWithColumnPredicate(kTableName, "key", 83, 3, 250, 333); |
| CheckScanWithColumnPredicate(kTableName, "key", 1, 1, 0, 1); |
| CheckScanWithColumnPredicate(kTableName, "key", 1, 1, 110, 111); |
| CheckScanWithColumnPredicate(kTableName, "key", 3, 6, 110, 113); |
| CheckScanWithColumnPredicate(kTableName, "key", 1, 1, 332, 333); |
| CheckScanWithColumnPredicate(kTableName, "key", 0, 0, 333, 334); |
| CheckScanWithColumnPredicate(kTableName, "key", 0, 0, -10, -5); |
| CheckScanWithColumnPredicate(kTableName, "key", 0, 0, 350, 400); |
| CheckScanWithColumnPredicate(kTableName, "key", 100, 3, lower_filler, 100); |
| CheckScanWithColumnPredicate(kTableName, "key", 250, 9, lower_filler, 250); |
| CheckScanWithColumnPredicate(kTableName, "key", 333, 9, lower_filler, 333); |
| CheckScanWithColumnPredicate(kTableName, "key", 333, 9, lower_filler, 334); |
| CheckScanWithColumnPredicate(kTableName, "key", 333, 9, -10, upper_filler); |
| CheckScanWithColumnPredicate(kTableName, "key", 233, 9, 100, upper_filler); |
| CheckScanWithColumnPredicate(kTableName, "key", 133, 6, 200, upper_filler); |
| CheckScanWithColumnPredicate(kTableName, "key", 83, 3, 250, upper_filler); |
| CheckScanWithColumnPredicate(kTableName, "key", 0, 0, 333, upper_filler); |
| |
| // Meanwhile, inserting into non-covered ranges should result in a proper |
| // error status return to the client attempting such an operation. |
| { |
| constexpr int kNumRows = 10; |
| vector<KuduError*> errors; |
| ElementDeleter drop(&errors); |
| auto s = InsertTestRows( |
| kTableName, 334, 334 + kNumRows, KuduSession::MANUAL_FLUSH, &errors); |
| ASSERT_TRUE(s.IsIOError()) << s.ToString(); |
| ASSERT_STR_CONTAINS(s.ToString(), "failed to flush data"); |
| ASSERT_EQ(kNumRows, errors.size()); |
| for (const auto& e : errors) { |
| const auto& err = e->status(); |
| EXPECT_TRUE(err.IsNotFound()) << err.ToString(); |
| ASSERT_STR_CONTAINS(err.ToString(), |
| "No tablet covering the requested range partition"); |
| } |
| } |
| } |
| |
| TEST_F(FlexPartitioningCreateTableTest, ScansWithNonCoveringRanges) { |
| |
| constexpr const char* const kTableName = "ScansWithNonCoveringRanges"; |
| |
| unique_ptr<KuduTableCreator> table_creator(client_->NewTableCreator()); |
| table_creator->table_name(kTableName) |
| .schema(&schema_) |
| .num_replicas(1) |
| .add_hash_partitions({ kKeyColumn }, 2) |
| .set_range_partition_columns({ kKeyColumn }); |
| |
| // Add a range partition with custom hash sub-partitioning rules: |
| // 4 buckets with hash based on the "key" column with hash seed 1. |
| { |
| auto p = CreateRangePartition(0, 111); |
| ASSERT_OK(p->add_hash_partitions({ kKeyColumn }, 4, 1)); |
| table_creator->add_custom_range_partition(p.release()); |
| } |
| |
| // Add a range partition with the table-wide hash partitioning rules. |
| { |
| unique_ptr<KuduPartialRow> lower(schema_.NewRow()); |
| ASSERT_OK(lower->SetInt32(kKeyColumn, 222)); |
| unique_ptr<KuduPartialRow> upper(schema_.NewRow()); |
| ASSERT_OK(upper->SetInt32(kKeyColumn, 333)); |
| table_creator->add_range_partition(lower.release(), upper.release()); |
| } |
| |
| // Add a range partition with custom hash sub-partitioning rules: |
| // 3 buckets with hash based on the "key" column with hash seed 1. |
| { |
| auto p = CreateRangePartition(444, 555); |
| ASSERT_OK(p->add_hash_partitions({ kKeyColumn }, 3, 1)); |
| table_creator->add_custom_range_partition(p.release()); |
| } |
| |
| ASSERT_OK(table_creator->Create()); |
| NO_FATALS(CheckTabletCount(kTableName, 9)); |
| |
| ASSERT_OK(InsertTestRows(kTableName, 0, 111)); |
| NO_FATALS(CheckTableRowsNum(kTableName, 111)); |
| NO_FATALS(CheckTabletsScanned(kTableName, 9)); |
| |
| ASSERT_OK(InsertTestRows(kTableName, 222, 333)); |
| NO_FATALS(CheckTableRowsNum(kTableName, 222)); |
| NO_FATALS(CheckTabletsScanned(kTableName, 9)); |
| |
| ASSERT_OK(InsertTestRows(kTableName, 444, 555)); |
| NO_FATALS(CheckTableRowsNum(kTableName, 333)); |
| NO_FATALS(CheckTabletsScanned(kTableName, 9)); |
| |
| // Filler values for lower or upper bound when it's not meant to be set. |
| constexpr int lower_filler = INT8_MIN; |
| constexpr int upper_filler = INT8_MAX; |
| |
| CheckScanWithColumnPredicate(kTableName, "key", 111, 4, -10, 111); |
| CheckScanWithColumnPredicate(kTableName, "key", 111, 4, -10, 150); |
| CheckScanWithColumnPredicate(kTableName, "key", 111, 2, 150, 333); |
| CheckScanWithColumnPredicate(kTableName, "key", 33, 6, 100, 244); |
| CheckScanWithColumnPredicate(kTableName, "key", 111, 2, 144, 344); |
| CheckScanWithColumnPredicate(kTableName, "key", 333, 9, -50, 600); |
| CheckScanWithColumnPredicate(kTableName, "key", 222, 5, 150, 600); |
| CheckScanWithColumnPredicate(kTableName, "key", 1, 1, 0, 1); |
| CheckScanWithColumnPredicate(kTableName, "key", 0, 0, -10, -5); |
| CheckScanWithColumnPredicate(kTableName, "key", 0, 0, 600, 650); |
| CheckScanWithColumnPredicate(kTableName, "key", 1, 1, 554, 555); |
| CheckScanWithColumnPredicate(kTableName, "key", 0, 0, 555, 556); |
| CheckScanWithColumnPredicate(kTableName, "key", 111, 4, lower_filler, 150); |
| CheckScanWithColumnPredicate(kTableName, "key", 222, 6, lower_filler, 350); |
| CheckScanWithColumnPredicate(kTableName, "key", 333, 9, lower_filler, 560); |
| CheckScanWithColumnPredicate(kTableName, "key", 333, 9, -10, upper_filler); |
| CheckScanWithColumnPredicate(kTableName, "key", 222, 5, 150, upper_filler); |
| CheckScanWithColumnPredicate(kTableName, "key", 111, 3, 350, upper_filler); |
| } |
| |
| // Negative tests scenarios to cover non-OK status codes for various operations |
| // related to custom hash bucket schema per range. |
| TEST_F(FlexPartitioningCreateTableTest, Negatives) { |
| // Try adding hash partitions on an empty set of columns. |
| { |
| auto p = CreateRangePartition(); |
| const auto s = p->add_hash_partitions({}, 2, 0); |
| ASSERT_TRUE(s.IsInvalidArgument()) << s.ToString(); |
| ASSERT_STR_CONTAINS( |
| s.ToString(), "set of columns for hash partitioning must not be empty"); |
| } |
| |
| // Try adding hash partitions with just one bucket. |
| { |
| auto p = CreateRangePartition(); |
| const auto s = p->add_hash_partitions({ kKeyColumn }, 1, 0); |
| ASSERT_TRUE(s.IsInvalidArgument()) << s.ToString(); |
| ASSERT_STR_CONTAINS( |
| s.ToString(), |
| "at least two buckets are required to establish hash partitioning"); |
| } |
| |
| // Try adding hash partition on a non-existent column: appropriate error |
| // surfaces during table creation. |
| { |
| RangePartitions partitions; |
| partitions.emplace_back(CreateRangePartition()); |
| auto& p = partitions.back(); |
| ASSERT_OK(p->add_hash_partitions({ "nonexistent" }, 2, 0)); |
| const auto s = CreateTable("nicetry", std::move(partitions)); |
| ASSERT_TRUE(s.IsInvalidArgument()) << s.ToString(); |
| ASSERT_STR_CONTAINS(s.ToString(), "unknown column: name: \"nonexistent\""); |
| } |
| |
| // Try adding creating a table where both range splits and custom hash bucket |
| // schema per partition are both specified -- that should not be possible. |
| { |
| RangePartition p(CreateRangePartition(0, 100)); |
| ASSERT_OK(p->add_hash_partitions({ kKeyColumn }, 2, 0)); |
| |
| unique_ptr<KuduPartialRow> split(schema_.NewRow()); |
| ASSERT_OK(split->SetInt32(kKeyColumn, 50)); |
| |
| unique_ptr<KuduTableCreator> creator(client_->NewTableCreator()); |
| creator->table_name("nicetry") |
| .schema(&schema_) |
| .num_replicas(1) |
| .set_range_partition_columns({ kKeyColumn }) |
| .add_range_partition_split(split.release()) |
| .add_custom_range_partition(p.release()); |
| |
| const auto s = creator->Create(); |
| ASSERT_TRUE(s.IsInvalidArgument()) << s.ToString(); |
| ASSERT_STR_CONTAINS( |
| s.ToString(), |
| "split rows and custom hash bucket schemas for ranges are incompatible: " |
| "choose one or the other"); |
| } |
| |
| // Same as the sub-scenario above, but using deprecated client API to specify |
| // so-called split rows. |
| { |
| RangePartition p(CreateRangePartition(0, 100)); |
| ASSERT_OK(p->add_hash_partitions({ kKeyColumn }, 2, 0)); |
| |
| unique_ptr<KuduPartialRow> split_row(schema_.NewRow()); |
| ASSERT_OK(split_row->SetInt32(kKeyColumn, 50)); |
| |
| unique_ptr<KuduTableCreator> creator(client_->NewTableCreator()); |
| creator->table_name("nicetry") |
| .schema(&schema_) |
| .num_replicas(1) |
| .set_range_partition_columns({ kKeyColumn }) |
| .add_custom_range_partition(p.release()); |
| |
| #pragma GCC diagnostic push |
| #pragma GCC diagnostic ignored "-Wdeprecated-declarations" |
| creator->split_rows({ split_row.release() }); |
| #pragma GCC diagnostic pop |
| |
| const auto s = creator->Create(); |
| ASSERT_TRUE(s.IsInvalidArgument()) << s.ToString(); |
| ASSERT_STR_CONTAINS( |
| s.ToString(), |
| "split rows and custom hash bucket schemas for ranges are incompatible: " |
| "choose one or the other"); |
| } |
| |
| { |
| constexpr const char* const kTableName = "3@key_x_3@key"; |
| RangePartitions partitions; |
| partitions.emplace_back(CreateRangePartition()); |
| auto& p = partitions.back(); |
| ASSERT_OK(p->add_hash_partitions({ kKeyColumn }, 3, 0)); |
| ASSERT_OK(p->add_hash_partitions({ kKeyColumn }, 3, 1)); |
| const auto s = CreateTable(kTableName, std::move(partitions)); |
| ASSERT_TRUE(s.IsInvalidArgument()) << s.ToString(); |
| ASSERT_STR_CONTAINS(s.ToString(), |
| "hash bucket schema components must not contain " |
| "columns in common"); |
| } |
| |
| { |
| constexpr const char* const kTableName = "3@key_x_3@key:int_val"; |
| RangePartitions partitions; |
| partitions.emplace_back(CreateRangePartition()); |
| auto& p = partitions.back(); |
| ASSERT_OK(p->add_hash_partitions({ kKeyColumn }, 2, 0)); |
| ASSERT_OK(p->add_hash_partitions({ kKeyColumn, kIntValColumn }, 3, 1)); |
| const auto s = CreateTable(kTableName, std::move(partitions)); |
| ASSERT_TRUE(s.IsInvalidArgument()) << s.ToString(); |
| ASSERT_STR_CONTAINS(s.ToString(), |
| "hash bucket schema components must not contain " |
| "columns in common"); |
| } |
| |
| { |
| constexpr const char* const kTableName = "3@int_val"; |
| RangePartitions partitions; |
| partitions.emplace_back(CreateRangePartition()); |
| auto& p = partitions.back(); |
| ASSERT_OK(p->add_hash_partitions({ kIntValColumn }, 2, 0)); |
| const auto s = CreateTable(kTableName, std::move(partitions)); |
| ASSERT_TRUE(s.IsInvalidArgument()) << s.ToString(); |
| ASSERT_STR_CONTAINS(s.ToString(), |
| "must specify only primary key columns for hash " |
| "bucket partition components"); |
| } |
| |
| { |
| constexpr const char* const kTableName = "3@string_val"; |
| RangePartitions partitions; |
| partitions.emplace_back(CreateRangePartition()); |
| auto& p = partitions.back(); |
| ASSERT_OK(p->add_hash_partitions({ kStringValColumn }, 5, 0)); |
| const auto s = CreateTable(kTableName, std::move(partitions)); |
| ASSERT_TRUE(s.IsInvalidArgument()) << s.ToString(); |
| ASSERT_STR_CONTAINS(s.ToString(), |
| "must specify only primary key columns for hash " |
| "bucket partition components"); |
| } |
| |
| { |
| constexpr const char* const kTableName = "2@key_x_3@int_val"; |
| RangePartitions partitions; |
| partitions.emplace_back(CreateRangePartition()); |
| auto& p = partitions.back(); |
| ASSERT_OK(p->add_hash_partitions({ kKeyColumn }, 2, 0)); |
| ASSERT_OK(p->add_hash_partitions({ kIntValColumn }, 3, 1)); |
| const auto s = CreateTable(kTableName, std::move(partitions)); |
| ASSERT_TRUE(s.IsInvalidArgument()) << s.ToString(); |
| ASSERT_STR_CONTAINS(s.ToString(), |
| "must specify only primary key columns for hash " |
| "bucket partition components"); |
| } |
| } |
| |
| // This test scenario verifies that ListPartitions() returns correct number |
| // of partitions for tables with unbounded ranges with custom hash schemas. |
| // Essentially, this scenario makes sure the logic used by the code to iterate |
| // over range partitions by employing PartitionPruner's NextPartitionKey() and |
| // RemovePartitionKeyRange() methods works as expected. |
| TEST_F(FlexPartitioningCreateTableTest, UnboundedRangesOneDimensionHash) { |
| constexpr const char* const kTableName = "UnboundedRangesOneDimensionHash"; |
| |
| unique_ptr<KuduTableCreator> table_creator(client_->NewTableCreator()); |
| table_creator->table_name(kTableName) |
| .schema(&schema_) |
| .num_replicas(1) |
| .add_hash_partitions({ kKeyColumn }, 2) |
| .set_range_partition_columns({ kKeyColumn }); |
| |
| // Add a range partition with the table-wide hash schema. |
| { |
| unique_ptr<KuduPartialRow> lower(schema_.NewRow()); |
| unique_ptr<KuduPartialRow> upper(schema_.NewRow()); |
| ASSERT_OK(upper->SetInt32(kKeyColumn, -10)); |
| table_creator->add_range_partition(lower.release(), upper.release()); |
| } |
| |
| // Add two range partitions with custom hash schemas. |
| { |
| auto p = CreateRangePartition(-10, 10); |
| ASSERT_OK(p->add_hash_partitions({ kKeyColumn }, 5, 1)); |
| table_creator->add_custom_range_partition(p.release()); |
| } |
| { |
| auto p = CreateRangePartitionNoUpperBound(10); |
| ASSERT_OK(p->add_hash_partitions({ kKeyColumn }, 3, 2)); |
| table_creator->add_custom_range_partition(p.release()); |
| } |
| |
| ASSERT_OK(table_creator->Create()); |
| NO_FATALS(CheckTabletCount(kTableName, 10)); // 2 + 5 + 3 = 10 |
| shared_ptr<KuduTable> table; |
| ASSERT_OK(client_->OpenTable(kTableName, &table)); |
| |
| vector<Partition> partitions; |
| ASSERT_OK(table->ListPartitions(&partitions)); |
| ASSERT_EQ(10, partitions.size()); |
| |
| // Make sure it's possible to insert rows into the table for all the existing |
| // partitions. |
| ASSERT_OK(InsertTestRows(kTableName, -50, 50)); |
| NO_FATALS(CheckTableRowsNum(kTableName, 100)); |
| } |
| |
| // Similar to FlexPartitioningCreateTableTest.UnboundedRangesOneDimensionHash |
| // abobe, but with two hash dimensions. |
| TEST_F(FlexPartitioningCreateTableTest, UnboundedRangesTwoDimensionHash) { |
| constexpr const char* const kTableName = "UnboundedRangesTwoDimensionHash"; |
| constexpr const char* const kC0 = "c0"; |
| constexpr const char* const kC1 = "c1"; |
| constexpr const char* const kC2 = "c2"; |
| |
| KuduSchemaBuilder b; |
| b.AddColumn(kC0)->Type(KuduColumnSchema::INT32)->NotNull(); |
| b.AddColumn(kC1)->Type(KuduColumnSchema::INT32)->NotNull(); |
| b.AddColumn(kC2)->Type(KuduColumnSchema::STRING)->Nullable(); |
| b.SetPrimaryKey({ kC0, kC1 }); |
| KuduSchema schema; |
| ASSERT_OK(b.Build(&schema)); |
| |
| auto rows_inserter = [&](int32_t key_beg, int32_t key_end) { |
| vector<KuduError*> errors; |
| CHECK_LE(key_beg, key_end); |
| shared_ptr<KuduTable> table; |
| RETURN_NOT_OK(client_->OpenTable(kTableName, &table)); |
| shared_ptr<KuduSession> session = client_->NewSession(); |
| RETURN_NOT_OK(session->SetFlushMode(KuduSession::MANUAL_FLUSH)); |
| session->SetTimeoutMillis(60000); |
| for (int32_t key_val = key_beg; key_val < key_end; ++key_val) { |
| unique_ptr<KuduInsert> insert(table->NewInsert()); |
| RETURN_NOT_OK(insert->mutable_row()->SetInt32(kC0, key_val)); |
| RETURN_NOT_OK(insert->mutable_row()->SetInt32(kC1, key_val)); |
| RETURN_NOT_OK(insert->mutable_row()->SetStringCopy(kC2, std::to_string(rand()))); |
| RETURN_NOT_OK(session->Apply(insert.release())); |
| } |
| return session->Flush(); |
| }; |
| |
| // Two range partitions: [-inf, 0) [0, +inf) |
| { |
| unique_ptr<KuduTableCreator> table_creator(client_->NewTableCreator()); |
| table_creator->table_name(kTableName) |
| .schema(&schema) |
| .num_replicas(1) |
| .add_hash_partitions({ kC0 }, 3) |
| .add_hash_partitions({ kC1 }, 3) |
| .set_range_partition_columns({ kC0 }); |
| |
| // Add a range partition with the table-wide hash schema. |
| { |
| unique_ptr<KuduPartialRow> lower(schema.NewRow()); |
| unique_ptr<KuduPartialRow> upper(schema.NewRow()); |
| ASSERT_OK(upper->SetInt32(kC0, 0)); |
| table_creator->add_range_partition(lower.release(), upper.release()); |
| } |
| |
| // Add a range partition with custom hash schema. |
| { |
| unique_ptr<KuduPartialRow> lower(schema.NewRow()); |
| ASSERT_OK(lower->SetInt32(kC0, 0)); |
| unique_ptr<KuduRangePartition> p( |
| new KuduRangePartition(lower.release(), schema.NewRow())); |
| ASSERT_OK(p->add_hash_partitions({ kC0 }, 2, 0)); |
| ASSERT_OK(p->add_hash_partitions({ kC1 }, 2, 1)); |
| table_creator->add_custom_range_partition(p.release()); |
| } |
| |
| ASSERT_OK(table_creator->Create()); |
| NO_FATALS(CheckTabletCount(kTableName, 13)); |
| shared_ptr<KuduTable> table; |
| ASSERT_OK(client_->OpenTable(kTableName, &table)); |
| |
| vector<Partition> partitions; |
| ASSERT_OK(table->ListPartitions(&partitions)); |
| ASSERT_EQ(13, partitions.size()); |
| |
| ASSERT_OK(rows_inserter(-100, 100)); |
| NO_FATALS(CheckTableRowsNum(kTableName, 200)); |
| ASSERT_OK(client_->DeleteTable(kTableName)); |
| } |
| |
| // Three range partitions: [-inf, -10) [-10, 10) [10, +inf) |
| { |
| unique_ptr<KuduTableCreator> table_creator(client_->NewTableCreator()); |
| table_creator->table_name(kTableName) |
| .schema(&schema) |
| .num_replicas(1) |
| .add_hash_partitions({ kC0 }, 2) |
| .add_hash_partitions({ kC1 }, 2) |
| .set_range_partition_columns({ kC0 }); |
| |
| // Add a range partition with the table-wide hash schema. |
| { |
| unique_ptr<KuduPartialRow> lower(schema.NewRow()); |
| unique_ptr<KuduPartialRow> upper(schema.NewRow()); |
| ASSERT_OK(upper->SetInt32(kC0, -10)); |
| table_creator->add_range_partition(lower.release(), upper.release()); |
| } |
| |
| // Add two range partitions with custom hash schemas. |
| { |
| auto p = CreateRangePartition(schema, kC0, -10, 10); |
| ASSERT_OK(p->add_hash_partitions({ kC0 }, 4, 0)); |
| ASSERT_OK(p->add_hash_partitions({ kC1 }, 5, 1)); |
| table_creator->add_custom_range_partition(p.release()); |
| } |
| { |
| unique_ptr<KuduPartialRow> lower(schema.NewRow()); |
| ASSERT_OK(lower->SetInt32(kC0, 10)); |
| unique_ptr<KuduRangePartition> p( |
| new KuduRangePartition(lower.release(), schema.NewRow())); |
| ASSERT_OK(p->add_hash_partitions({ kC0 }, 3, 2)); |
| ASSERT_OK(p->add_hash_partitions({ kC1 }, 4, 3)); |
| table_creator->add_custom_range_partition(p.release()); |
| } |
| |
| ASSERT_OK(table_creator->Create()); |
| NO_FATALS(CheckTabletCount(kTableName, 36)); // 4 + 20 + 12 = 36 |
| shared_ptr<KuduTable> table; |
| ASSERT_OK(client_->OpenTable(kTableName, &table)); |
| |
| vector<Partition> partitions; |
| ASSERT_OK(table->ListPartitions(&partitions)); |
| ASSERT_EQ(36, partitions.size()); |
| |
| ASSERT_OK(rows_inserter(-250, 250)); |
| NO_FATALS(CheckTableRowsNum(kTableName, 500)); |
| ASSERT_OK(client_->DeleteTable(kTableName)); |
| } |
| } |
| |
| // When working with a cluster that doesn't support range-specific hash schemas |
| // for tables, the client should receive proper error while trying to create |
| // a table with custom hash schema for at least one of its ranges. |
| TEST_F(FlexPartitioningCreateTableTest, UnsupportedRangeSpecificHashSchema) { |
| // Turn off the support for range-specific hash schemas, emulating the |
| // situation when a Kudu cluster is running an older version released prior |
| // to the introduction of the feature. |
| FLAGS_enable_per_range_hash_schemas = false; |
| |
| constexpr const char* const kTableName = |
| "UnsupportedRangeSpecificHashSchemaCreateTable"; |
| |
| unique_ptr<KuduTableCreator> table_creator(client_->NewTableCreator()); |
| table_creator->table_name(kTableName) |
| .schema(&schema_) |
| .num_replicas(1) |
| .add_hash_partitions({ kKeyColumn }, 3) |
| .set_range_partition_columns({ kKeyColumn }); |
| |
| // Add a range partition with the table-wide hash partitioning rules. |
| unique_ptr<KuduPartialRow> lower(schema_.NewRow()); |
| ASSERT_OK(lower->SetInt32(kKeyColumn, 0)); |
| unique_ptr<KuduPartialRow> upper(schema_.NewRow()); |
| ASSERT_OK(upper->SetInt32(kKeyColumn, 111)); |
| table_creator->add_range_partition(lower.release(), upper.release()); |
| |
| // Add a range partition with custom hash schema. |
| auto p = CreateRangePartition(111, 222); |
| ASSERT_OK(p->add_hash_partitions({ kKeyColumn }, 5, 0)); |
| table_creator->add_custom_range_partition(p.release()); |
| |
| const auto s = table_creator->Create(); |
| ASSERT_TRUE(s.IsNotSupported()) << s.ToString(); |
| ASSERT_STR_CONTAINS(s.ToString(), "cluster does not support CreateTable with " |
| "feature(s) RANGE_SPECIFIC_HASH_SCHEMA"); |
| } |
| |
| // Test for scenarios covering range partitioning with custom hash schemas |
| // specified when adding a new custom hash schema partition to a table. |
| class FlexPartitioningAlterTableTest : public FlexPartitioningTest {}; |
| |
| TEST_F(FlexPartitioningAlterTableTest, AddAndDropCustomRangePartition) { |
| // Create a table with the following partitions: |
| // |
| // hash bucket |
| // key 0 1 |
| // ------------------------- |
| // <111 x:{key} x:{key} |
| // |
| // Later add the following partition: |
| // hash bucket |
| // key 0 1 2 |
| // -------------------------------- |
| // 111-222 x:{key} x:{key} x:{key} |
| constexpr const char *const kTableName = "AddAndDropCustomRangePartition"; |
| |
| // Create a table with a single Range partition. |
| unique_ptr<KuduTableCreator> table_creator(client_->NewTableCreator()); |
| unique_ptr<KuduPartialRow> lower(schema_.NewRow()); |
| ASSERT_OK(lower->SetInt32(kKeyColumn, INT32_MIN)); |
| unique_ptr<KuduPartialRow> upper(schema_.NewRow()); |
| ASSERT_OK(upper->SetInt32(kKeyColumn, 111)); |
| table_creator->table_name(kTableName) |
| .schema(&schema_) |
| .num_replicas(1) |
| .add_hash_partitions({ kKeyColumn }, 2) |
| .set_range_partition_columns({ kKeyColumn }) |
| .add_range_partition(lower.release(), upper.release()); |
| ASSERT_OK(table_creator->Create()); |
| |
| NO_FATALS(CheckTabletCount(kTableName, 2)); |
| |
| // Add a range partition with custom hash sub-partitioning rules: |
| // 3 buckets with hash based on the "key" column with hash seed 1. |
| { |
| unique_ptr<KuduTableAlterer> table_alterer(client_->NewTableAlterer(kTableName)); |
| auto p = CreateRangePartition(111, 222); |
| ASSERT_OK(p->add_hash_partitions({ kKeyColumn }, 3, 1)); |
| table_alterer->AddRangePartition(p.release()); |
| ASSERT_OK(table_alterer->Alter()); |
| } |
| |
| NO_FATALS(CheckTabletCount(kTableName, 5)); |
| |
| // Try adding a range partition with custom hash sub-partitioning rules: |
| // 3 buckets with hash based on the "key" column with hash seed 1. |
| // 2 buckets with hash based on the "key" column with hash seed 1. |
| { |
| unique_ptr<KuduTableAlterer> table_alterer(client_->NewTableAlterer(kTableName)); |
| auto p = CreateRangePartition(222, 333); |
| ASSERT_OK(p->add_hash_partitions({ kKeyColumn }, 3, 1)); |
| ASSERT_OK(p->add_hash_partitions({ kKeyColumn }, 2, 1)); |
| table_alterer->AddRangePartition(p.release()); |
| const auto s = table_alterer->Alter(); |
| ASSERT_TRUE(s.IsNotSupported()) << s.ToString(); |
| ASSERT_STR_CONTAINS(s.ToString(), |
| "varying number of hash dimensions per range is not yet supported"); |
| } |
| |
| // Drop the successfully created partition. |
| { |
| unique_ptr<KuduTableAlterer> table_alterer(client_->NewTableAlterer(kTableName)); |
| unique_ptr<KuduPartialRow> lower_drop(schema_.NewRow()); |
| ASSERT_OK(lower_drop->SetInt32(kKeyColumn, 111)); |
| unique_ptr<KuduPartialRow> upper_drop(schema_.NewRow()); |
| ASSERT_OK(upper_drop->SetInt32(kKeyColumn, 222)); |
| table_alterer->DropRangePartition(lower_drop.release(), upper_drop.release()); |
| ASSERT_OK(table_alterer->Alter()); |
| } |
| |
| NO_FATALS(CheckTabletCount(kTableName, 2)); |
| } |
| |
| TEST_F(FlexPartitioningAlterTableTest, ReadAndWriteToCustomRangePartition) { |
| // Create a table with the following partitions: |
| // |
| // hash bucket |
| // key 0 1 2 3 |
| // ----------------------------------------------------------- |
| // <111 x:{key} x:{key} - - |
| |
| // Later add the following partitions: |
| // hash bucket |
| // key 0 1 2 3 |
| // ----------------------------------------------------------- |
| // 111-222 x:{key} x:{key} x:{key} - |
| // 222-333 x:{key} x:{key} x:{key} x:{key} |
| // 333-444 x:{key} x:{key} - - |
| |
| constexpr const char* const kTableName = "ReadAndWriteToCustomRangePartition"; |
| unique_ptr<KuduTableCreator> table_creator(client_->NewTableCreator()); |
| unique_ptr<KuduPartialRow> lower(schema_.NewRow()); |
| ASSERT_OK(lower->SetInt32(kKeyColumn, INT32_MIN)); |
| unique_ptr<KuduPartialRow> upper(schema_.NewRow()); |
| ASSERT_OK(upper->SetInt32(kKeyColumn, 111)); |
| table_creator->table_name(kTableName) |
| .schema(&schema_) |
| .num_replicas(1) |
| .add_hash_partitions({ kKeyColumn }, 2) |
| .set_range_partition_columns({ kKeyColumn }) |
| .add_range_partition(lower.release(), upper.release()); |
| ASSERT_OK(table_creator->Create()); |
| |
| unique_ptr<KuduTableAlterer> table_alterer(client_->NewTableAlterer(kTableName)); |
| // Add a range partition with custom hash sub-partitioning rules: |
| // 3 buckets with hash based on the "key" column with hash seed 1. |
| { |
| auto p = CreateRangePartition(111, 222); |
| ASSERT_OK(p->add_hash_partitions({ kKeyColumn }, 3, 1)); |
| table_alterer->AddRangePartition(p.release()); |
| } |
| |
| // Add a range partition with custom hash sub-partitioning rules: |
| // 4 buckets with hash based on the "key" column with hash seed 2. |
| { |
| auto p = CreateRangePartition(222, 333); |
| ASSERT_OK(p->add_hash_partitions({ kKeyColumn }, 4, 2)); |
| table_alterer->AddRangePartition(p.release()); |
| } |
| |
| // Add a range partition with custom hash sub-partitioning rules: |
| // 2 buckets hashing on the "key" column with hash seed 3. |
| { |
| auto p = CreateRangePartition(333, 444); |
| ASSERT_OK(p->add_hash_partitions({ kKeyColumn }, 2, 3)); |
| table_alterer->AddRangePartition(p.release()); |
| } |
| |
| ASSERT_OK(table_alterer->Alter()); |
| NO_FATALS(CheckTabletCount(kTableName, 11)); |
| |
| // Make sure it's possible to insert rows into the table for all the existing |
| // the partitions: first check the range of table-wide schema, then check |
| // the ranges with custom hash schemas. |
| ASSERT_OK(InsertTestRows(kTableName, 0, 111)); |
| NO_FATALS(CheckTableRowsNum(kTableName, 111)); |
| ASSERT_OK(InsertTestRows(kTableName, 111, 444)); |
| NO_FATALS(CheckTableRowsNum(kTableName, 444)); |
| |
| NO_FATALS(CheckTableRowsNum(kTableName, kKeyColumn, 111, 222, 111)); |
| NO_FATALS(CheckTableRowsNum(kTableName, kKeyColumn, 0, 444, 444)); |
| // Drop a partition and re-scan |
| { |
| unique_ptr<KuduTableAlterer> table_alterer_drop(client_->NewTableAlterer(kTableName)); |
| unique_ptr<KuduPartialRow> lower_drop(schema_.NewRow()); |
| ASSERT_OK(lower_drop->SetInt32(kKeyColumn, 111)); |
| unique_ptr<KuduPartialRow> upper_drop(schema_.NewRow()); |
| ASSERT_OK(upper_drop->SetInt32(kKeyColumn, 222)); |
| table_alterer_drop->DropRangePartition(lower_drop.release(), upper_drop.release()); |
| ASSERT_OK(table_alterer_drop->Alter()); |
| } |
| NO_FATALS(CheckTableRowsNum(kTableName, kKeyColumn, 0, 444, 333)); |
| |
| // Meanwhile, inserting into non-covered ranges should result in a proper |
| // error status return to the client attempting such an operation. |
| { |
| constexpr int kNumRows = 10; |
| vector<KuduError*> errors; |
| ElementDeleter drop(&errors); |
| auto s = InsertTestRows( |
| kTableName, 445, 445 + kNumRows, KuduSession::MANUAL_FLUSH, &errors); |
| ASSERT_TRUE(s.IsIOError()) << s.ToString(); |
| ASSERT_STR_CONTAINS(s.ToString(), "failed to flush data"); |
| ASSERT_EQ(kNumRows, errors.size()); |
| for (const auto& e : errors) { |
| const auto& err = e->status(); |
| EXPECT_TRUE(err.IsNotFound()) << err.ToString(); |
| ASSERT_STR_CONTAINS(err.ToString(), |
| "No tablet covering the requested range partition"); |
| } |
| } |
| } |
| |
| TEST_F(FlexPartitioningAlterTableTest, ReadAndWriteToUnboundedCustomRangePartition) { |
| constexpr const char* const kTableName = |
| "ReadAndWriteToUnboundedCustomRangePartition"; |
| unique_ptr<KuduTableCreator> table_creator(client_->NewTableCreator()); |
| unique_ptr<KuduPartialRow> lower(schema_.NewRow()); |
| ASSERT_OK(lower->SetInt32(kKeyColumn, -100)); |
| unique_ptr<KuduPartialRow> upper(schema_.NewRow()); |
| ASSERT_OK(upper->SetInt32(kKeyColumn, 100)); |
| table_creator->table_name(kTableName) |
| .schema(&schema_) |
| .num_replicas(1) |
| .add_hash_partitions({ kKeyColumn }, 2) |
| .set_range_partition_columns({ kKeyColumn }) |
| .add_range_partition(lower.release(), upper.release()); |
| ASSERT_OK(table_creator->Create()); |
| |
| unique_ptr<KuduTableAlterer> table_alterer(client_->NewTableAlterer(kTableName)); |
| { |
| auto p = CreateRangePartitionNoLowerBound(-100); |
| ASSERT_OK(p->add_hash_partitions({ kKeyColumn }, 3, 1)); |
| table_alterer->AddRangePartition(p.release()); |
| } |
| { |
| auto p = CreateRangePartitionNoUpperBound(100); |
| ASSERT_OK(p->add_hash_partitions({ kKeyColumn }, 4, 2)); |
| table_alterer->AddRangePartition(p.release()); |
| } |
| ASSERT_OK(table_alterer->Alter()); |
| NO_FATALS(CheckTabletCount(kTableName, 9)); // 2 + 3 + 4 = 9 |
| |
| // Make sure it's possible to insert rows into the table for all the existing |
| // the partitions: first check the range of table-wide schema, then check |
| // the ranges with custom hash schemas. |
| ASSERT_OK(InsertTestRows(kTableName, -100, 100)); |
| NO_FATALS(CheckTableRowsNum(kTableName, 200)); |
| ASSERT_OK(InsertTestRows(kTableName, -200, -100)); |
| NO_FATALS(CheckTableRowsNum(kTableName, 300)); |
| ASSERT_OK(InsertTestRows(kTableName, 100, 200)); |
| NO_FATALS(CheckTableRowsNum(kTableName, 400)); |
| ASSERT_OK(InsertTestRows(kTableName, INT32_MIN, INT32_MIN + 100)); |
| NO_FATALS(CheckTableRowsNum(kTableName, 500)); |
| ASSERT_OK(InsertTestRows(kTableName, INT32_MAX - 100, INT32_MAX)); |
| NO_FATALS(CheckTableRowsNum(kTableName, 600)); |
| |
| NO_FATALS(CheckTableRowsNum(kTableName, kKeyColumn, -300, 300, 400)); |
| |
| // Drop a partition in the middle and re-scan with various ranges. |
| { |
| unique_ptr<KuduTableAlterer> table_alterer_drop(client_->NewTableAlterer(kTableName)); |
| unique_ptr<KuduPartialRow> lower_drop(schema_.NewRow()); |
| ASSERT_OK(lower_drop->SetInt32(kKeyColumn, -100)); |
| unique_ptr<KuduPartialRow> upper_drop(schema_.NewRow()); |
| ASSERT_OK(upper_drop->SetInt32(kKeyColumn, 100)); |
| table_alterer_drop->DropRangePartition(lower_drop.release(), upper_drop.release()); |
| ASSERT_OK(table_alterer_drop->Alter()); |
| } |
| NO_FATALS(CheckTableRowsNum(kTableName, 400)); |
| NO_FATALS(CheckTableRowsNum(kTableName, kKeyColumn, 0, 300, 100)); |
| NO_FATALS(CheckTableRowsNum(kTableName, kKeyColumn, -300, 0, 100)); |
| NO_FATALS(CheckTableRowsNum(kTableName, kKeyColumn, -200, 200, 200)); |
| NO_FATALS(CheckTableRowsNum(kTableName, kKeyColumn, -500, 500, 200)); |
| } |
| |
| // When working with a cluster that doesn't support range-specific hash schemas |
| // for tables, the client should receive proper error while trying to add |
| // a range with custom hash schema. |
| TEST_F(FlexPartitioningAlterTableTest, UnsupportedRangeSpecificHashSchema) { |
| // Turn off the support for range-specific hash schemas, emulating the |
| // situation when a Kudu cluster is running an older version released prior |
| // to the introduction of the feature. |
| FLAGS_enable_per_range_hash_schemas = false; |
| |
| constexpr const char* const kTableName = |
| "UnsupportedRangeSpecificHashSchemaAlterTable"; |
| constexpr const char* const kErrMsg = "cluster does not support AlterTable " |
| "with feature(s) RANGE_SPECIFIC_HASH_SCHEMA"; |
| |
| unique_ptr<KuduTableCreator> table_creator(client_->NewTableCreator()); |
| table_creator->table_name(kTableName) |
| .schema(&schema_) |
| .num_replicas(1) |
| .add_hash_partitions({ kKeyColumn }, 2) |
| .set_range_partition_columns({ kKeyColumn }); |
| |
| // Add a range partition with the table-wide hash partitioning rules. |
| unique_ptr<KuduPartialRow> lower(schema_.NewRow()); |
| ASSERT_OK(lower->SetInt32(kKeyColumn, 0)); |
| unique_ptr<KuduPartialRow> upper(schema_.NewRow()); |
| ASSERT_OK(upper->SetInt32(kKeyColumn, 111)); |
| table_creator->add_range_partition(lower.release(), upper.release()); |
| |
| ASSERT_OK(table_creator->Create()); |
| |
| // Try to add a single range with custom hash schema. |
| { |
| unique_ptr<KuduTableAlterer> alterer(client_->NewTableAlterer(kTableName)); |
| auto p = CreateRangePartition(111, 222); |
| ASSERT_OK(p->add_hash_partitions({ kKeyColumn }, 5, 0)); |
| alterer->AddRangePartition(p.release()); |
| |
| const auto s = alterer->Alter(); |
| ASSERT_TRUE(s.IsNotSupported()) << s.ToString(); |
| ASSERT_STR_CONTAINS(s.ToString(), kErrMsg); |
| } |
| |
| // Try to add a mix of ranges: with the table-wide and custom hash schemas. |
| { |
| unique_ptr<KuduTableAlterer> alterer(client_->NewTableAlterer(kTableName)); |
| |
| unique_ptr<KuduPartialRow> lower(schema_.NewRow()); |
| ASSERT_OK(lower->SetInt32(kKeyColumn, 111)); |
| unique_ptr<KuduPartialRow> upper(schema_.NewRow()); |
| ASSERT_OK(upper->SetInt32(kKeyColumn, 222)); |
| alterer->AddRangePartition(lower.release(), upper.release()); |
| |
| auto p = CreateRangePartition(222, 333); |
| ASSERT_OK(p->add_hash_partitions({ kKeyColumn }, 8, 0)); |
| alterer->AddRangePartition(p.release()); |
| |
| const auto s = alterer->Alter(); |
| ASSERT_TRUE(s.IsNotSupported()) << s.ToString(); |
| ASSERT_STR_CONTAINS(s.ToString(), kErrMsg); |
| } |
| |
| // Temporary enable support for the RANGE_SPECIFIC_HASH_SCHEMA feature to add |
| // a new range with custom hash schema to be dropped later. |
| FLAGS_enable_per_range_hash_schemas = true; |
| { |
| unique_ptr<KuduTableAlterer> alterer(client_->NewTableAlterer(kTableName)); |
| |
| auto p0 = CreateRangePartition(111, 222); |
| ASSERT_OK(p0->add_hash_partitions({ kKeyColumn }, 5, 0)); |
| alterer->AddRangePartition(p0.release()); |
| |
| auto p1 = CreateRangePartition(222, 333); |
| ASSERT_OK(p1->add_hash_partitions({ kKeyColumn }, 8, 0)); |
| alterer->AddRangePartition(p1.release()); |
| |
| ASSERT_OK(alterer->Alter()); |
| } |
| // Disable the support for the RANGE_SPECIFIC_HASH_SCHEMA. |
| FLAGS_enable_per_range_hash_schemas = false; |
| |
| // Dropping ranges with the table-wide or custom hash schemas should be fine |
| // even if the cluster doesn't support the RANGE_SPECIFIC_HASH_SCHEMA schema |
| // It's rather a hypothetical situation unless they toggle the flag after |
| // adding ranges with custom hash schemas or running older Kudu binaries with |
| // new data. |
| { |
| unique_ptr<KuduTableAlterer> alterer(client_->NewTableAlterer(kTableName)); |
| unique_ptr<KuduPartialRow> lower(schema_.NewRow()); |
| ASSERT_OK(lower->SetInt32(kKeyColumn, 111)); |
| unique_ptr<KuduPartialRow> upper(schema_.NewRow()); |
| ASSERT_OK(upper->SetInt32(kKeyColumn, 222)); |
| alterer->DropRangePartition(lower.release(), upper.release()); |
| ASSERT_OK(alterer->Alter()); |
| } |
| |
| // Dropping tables having ranges with custom hash schema should be fine |
| // even if the server side has no RANGE_SPECIFIC_HASH_SCHEMA feature. |
| ASSERT_OK(client_->DeleteTable(kTableName)); |
| } |
| |
| // Make sure adding and dropping ranges with the table-wide hash schema works |
| // as expected. |
| TEST_F(FlexPartitioningAlterTableTest, AddDropTableWideHashSchemaPartitions) { |
| constexpr const char* const kTableName = |
| "AddDropTableWideHashSchemaPartitions"; |
| |
| unique_ptr<KuduTableCreator> table_creator(client_->NewTableCreator()); |
| table_creator->table_name(kTableName) |
| .schema(&schema_) |
| .num_replicas(1) |
| .add_hash_partitions({ kKeyColumn }, 2) |
| .set_range_partition_columns({ kKeyColumn }); |
| |
| // Add a range partition with the table-wide hash partitioning rules. |
| { |
| unique_ptr<KuduPartialRow> lower(schema_.NewRow()); |
| ASSERT_OK(lower->SetInt32(kKeyColumn, 0)); |
| unique_ptr<KuduPartialRow> upper(schema_.NewRow()); |
| ASSERT_OK(upper->SetInt32(kKeyColumn, 111)); |
| table_creator->add_range_partition(lower.release(), upper.release()); |
| } |
| |
| ASSERT_OK(table_creator->Create()); |
| |
| ASSERT_OK(InsertTestRows(kTableName, 0, 111)); |
| NO_FATALS(CheckTableRowsNum(kTableName, 111)); |
| |
| // To have mix of ranges, add a single range with custom hash schema. |
| { |
| unique_ptr<KuduTableAlterer> alterer(client_->NewTableAlterer(kTableName)); |
| auto p = CreateRangePartition(-111, 0); |
| ASSERT_OK(p->add_hash_partitions({ kKeyColumn }, 5, 1)); |
| alterer->AddRangePartition(p.release()); |
| ASSERT_OK(alterer->Alter()); |
| } |
| |
| ASSERT_OK(InsertTestRows(kTableName, -111, 0)); |
| NO_FATALS(CheckTableRowsNum(kTableName, 222)); |
| |
| // Add one more range partition with the table-wide hash schema. |
| { |
| unique_ptr<KuduTableAlterer> alterer(client_->NewTableAlterer(kTableName)); |
| |
| unique_ptr<KuduPartialRow> lower(schema_.NewRow()); |
| ASSERT_OK(lower->SetInt32(kKeyColumn, 111)); |
| unique_ptr<KuduPartialRow> upper(schema_.NewRow()); |
| ASSERT_OK(upper->SetInt32(kKeyColumn, 222)); |
| alterer->AddRangePartition(lower.release(), upper.release()); |
| |
| ASSERT_OK(alterer->Alter()); |
| } |
| |
| ASSERT_OK(InsertTestRows(kTableName, 111, 222)); |
| NO_FATALS(CheckTableRowsNum(kTableName, 333)); |
| |
| // Drop the ranges with the table-wide hash partitions. |
| { |
| unique_ptr<KuduTableAlterer> alterer(client_->NewTableAlterer(kTableName)); |
| |
| { |
| unique_ptr<KuduPartialRow> lower(schema_.NewRow()); |
| ASSERT_OK(lower->SetInt32(kKeyColumn, 111)); |
| unique_ptr<KuduPartialRow> upper(schema_.NewRow()); |
| ASSERT_OK(upper->SetInt32(kKeyColumn, 222)); |
| alterer->DropRangePartition(lower.release(), upper.release()); |
| } |
| { |
| unique_ptr<KuduPartialRow> lower(schema_.NewRow()); |
| ASSERT_OK(lower->SetInt32(kKeyColumn, 0)); |
| unique_ptr<KuduPartialRow> upper(schema_.NewRow()); |
| ASSERT_OK(upper->SetInt32(kKeyColumn, 111)); |
| alterer->DropRangePartition(lower.release(), upper.release()); |
| } |
| |
| ASSERT_OK(alterer->Alter()); |
| } |
| |
| NO_FATALS(CheckTableRowsNum(kTableName, 111)); |
| |
| ASSERT_OK(client_->DeleteTable(kTableName)); |
| } |
| |
| // Try adding range partition with custom hash schema where hash columns are |
| // duplicated across different dimensions. That should not be possible and the |
| // client should receive a proper error in response. |
| TEST_F(FlexPartitioningAlterTableTest, AddRangeWithDuplicateHashColumns) { |
| constexpr const char* const kCol0 = "c0"; |
| constexpr const char* const kCol1 = "c1"; |
| constexpr const char* const kCol2 = "c2"; |
| constexpr const char* const kErrMsg = |
| "hash bucket schema components must not contain columns in common"; |
| constexpr const char* const kTableName = "AddRangeWithDuplicateHashColumns"; |
| |
| |
| KuduSchema schema; |
| { |
| KuduSchemaBuilder b; |
| b.AddColumn(kCol0)->Type(KuduColumnSchema::INT32)->NotNull(); |
| b.AddColumn(kCol1)->Type(KuduColumnSchema::INT32)->NotNull(); |
| b.AddColumn(kCol2)->Type(KuduColumnSchema::STRING)->Nullable(); |
| b.SetPrimaryKey({kCol0, kCol1}); |
| ASSERT_OK(b.Build(&schema)); |
| } |
| |
| unique_ptr<KuduTableCreator> table_creator(client_->NewTableCreator()); |
| table_creator->table_name(kTableName) |
| .schema(&schema) |
| .num_replicas(1) |
| .add_hash_partitions({ kCol0 }, 2) |
| .add_hash_partitions({ kCol1 }, 3) |
| .set_range_partition_columns({ kCol0 }); |
| |
| // Add a range partition with the table-wide hash schema. |
| { |
| unique_ptr<KuduPartialRow> lower(schema.NewRow()); |
| ASSERT_OK(lower->SetInt32(kCol0, 0)); |
| unique_ptr<KuduPartialRow> upper(schema.NewRow()); |
| ASSERT_OK(upper->SetInt32(kCol0, 111)); |
| table_creator->add_range_partition(lower.release(), upper.release()); |
| } |
| ASSERT_OK(table_creator->Create()); |
| |
| // Try to add hash partitions with duplicate hash columns across dimensions. |
| |
| { |
| unique_ptr<KuduTableAlterer> alterer(client_->NewTableAlterer(kTableName)); |
| auto p = CreateRangePartition(schema, kCol0, -111, 0); |
| ASSERT_OK(p->add_hash_partitions({ kCol0 }, 3, 1)); |
| ASSERT_OK(p->add_hash_partitions({ kCol0 }, 5, 2)); |
| alterer->AddRangePartition(p.release()); |
| const auto s = alterer->Alter(); |
| ASSERT_TRUE(s.IsInvalidArgument()) << s.ToString(); |
| ASSERT_STR_CONTAINS(s.ToString(), kErrMsg); |
| } |
| |
| { |
| unique_ptr<KuduTableAlterer> alterer(client_->NewTableAlterer(kTableName)); |
| auto p = CreateRangePartition(schema, kCol0, -111, 0); |
| ASSERT_OK(p->add_hash_partitions({ kCol0, kCol1 }, 3, 3)); |
| ASSERT_OK(p->add_hash_partitions({ kCol0 }, 2, 4)); |
| alterer->AddRangePartition(p.release()); |
| const auto s = alterer->Alter(); |
| ASSERT_TRUE(s.IsInvalidArgument()) << s.ToString(); |
| ASSERT_STR_CONTAINS(s.ToString(), kErrMsg); |
| } |
| |
| { |
| unique_ptr<KuduTableAlterer> alterer(client_->NewTableAlterer(kTableName)); |
| auto p = CreateRangePartition(schema, kCol0, -111, 0); |
| ASSERT_OK(p->add_hash_partitions({ kCol1 }, 5, 6)); |
| ASSERT_OK(p->add_hash_partitions({ kCol0, kCol1 }, 7, 5)); |
| alterer->AddRangePartition(p.release()); |
| const auto s = alterer->Alter(); |
| ASSERT_TRUE(s.IsInvalidArgument()) << s.ToString(); |
| ASSERT_STR_CONTAINS(s.ToString(), kErrMsg); |
| } |
| |
| { |
| unique_ptr<KuduTableAlterer> alterer(client_->NewTableAlterer(kTableName)); |
| auto p = CreateRangePartition(schema, kCol0, -111, 0); |
| ASSERT_OK(p->add_hash_partitions({ kCol0, kCol1 }, 3, 7)); |
| ASSERT_OK(p->add_hash_partitions({ kCol1, kCol0 }, 7, 8)); |
| alterer->AddRangePartition(p.release()); |
| const auto s = alterer->Alter(); |
| ASSERT_TRUE(s.IsInvalidArgument()) << s.ToString(); |
| ASSERT_STR_CONTAINS(s.ToString(), kErrMsg); |
| } |
| } |
| |
| class FlexPartitioningScanTest : public FlexPartitioningTest { |
| public: |
| int num_tablet_servers() const override { |
| return 3; |
| } |
| |
| void SetUp() override { |
| // This is necessary to make accumulated updates flush to the disk, |
| // so the DRS-level logic for optimizing the scan can kick in as expected. |
| FLAGS_flush_threshold_secs = 1; |
| FlexPartitioningTest::SetUp(); |
| } |
| }; |
| |
| // This scenario is to reproduce the issue described in KUDU-3384. |
| TEST_F(FlexPartitioningScanTest, MaxKeyValue) { |
| static constexpr const char* const kTableName = "max_key_value"; |
| |
| unique_ptr<KuduTableCreator> table_creator(client_->NewTableCreator()); |
| table_creator->table_name(kTableName) |
| .schema(&schema_) |
| .num_replicas(3) |
| .set_range_partition_columns({ kKeyColumn }); |
| { |
| unique_ptr<KuduPartialRow> lower(schema_.NewRow()); |
| unique_ptr<KuduPartialRow> upper(schema_.NewRow()); |
| ASSERT_OK(upper->SetInt32(kKeyColumn, 0)); |
| table_creator->add_range_partition(lower.release(), upper.release()); |
| } |
| { |
| unique_ptr<KuduPartialRow> lower(schema_.NewRow()); |
| ASSERT_OK(lower->SetInt32(kKeyColumn, 0)); |
| unique_ptr<KuduPartialRow> upper(schema_.NewRow()); |
| table_creator->add_range_partition(lower.release(), upper.release()); |
| } |
| ASSERT_OK(table_creator->Create()); |
| |
| shared_ptr<KuduSession> session = client_->NewSession(); |
| ASSERT_OK(session->SetFlushMode(KuduSession::AUTO_FLUSH_SYNC)); |
| |
| shared_ptr<KuduTable> table; |
| ASSERT_OK(client_->OpenTable(kTableName, &table)); |
| |
| // Not using InsertTestRows() since need to insert a row with key value |
| // of INT32_MAX. |
| for (int32_t i = INT32_MIN; i < INT32_MIN + 10; ++i) { |
| std::unique_ptr<client::KuduInsert> insert(table->NewInsert()); |
| KuduPartialRow* row = insert->mutable_row(); |
| ASSERT_OK(row->SetInt32(0, i)); |
| ASSERT_OK(row->SetInt32(1, i)); |
| ASSERT_OK(session->Apply(insert.release())); |
| } |
| |
| for (int32_t i = INT32_MAX; i > INT32_MAX - 10; --i) { |
| std::unique_ptr<client::KuduInsert> insert(table->NewInsert()); |
| KuduPartialRow* row = insert->mutable_row(); |
| ASSERT_OK(row->SetInt32(0, i)); |
| ASSERT_OK(row->SetInt32(1, i)); |
| ASSERT_OK(session->Apply(insert.release())); |
| } |
| |
| ASSERT_OK(session->Flush()); |
| |
| for (auto i = 0; i < 25; ++i) { |
| SCOPED_TRACE(Substitute("iteration $0", i)); |
| KuduScanTokenBuilder builder(table.get()); |
| ASSERT_OK(builder.SetTimeoutMillis(60000)); |
| |
| vector<KuduScanToken*> tokens; |
| ElementDeleter DeleteTable(&tokens); |
| ASSERT_OK(builder.Build(&tokens)); |
| |
| vector<string> rows; |
| for (auto token : tokens) { |
| KuduScanner* scanner_ptr; |
| ASSERT_OK(token->IntoKuduScanner(&scanner_ptr)); |
| ASSERT_OK(scanner_ptr->SetReadMode(KuduScanner::ReadMode::READ_AT_SNAPSHOT)); |
| unique_ptr<KuduScanner> scanner(scanner_ptr); |
| ASSERT_OK(ScanToStrings(scanner.get(), &rows)); |
| } |
| } |
| } |
| |
| // Try adding range partition with custom hash schema where the number of |
| // hash buckets is invalid. |
| TEST_F(FlexPartitioningAlterTableTest, AddRangeWithWrongHashBucketsNumber) { |
| constexpr const char* const kCol0 = "c0"; |
| constexpr const char* const kCol1 = "c1"; |
| constexpr const char* const kErrMsg = |
| "at least two buckets are required to establish hash partitioning"; |
| constexpr const char* const kTableName = "AddRangeWithWrongHashBucketsNumber"; |
| |
| KuduSchema schema; |
| { |
| KuduSchemaBuilder b; |
| b.AddColumn(kCol0)->Type(KuduColumnSchema::INT32)->NotNull()->PrimaryKey(); |
| b.AddColumn(kCol1)->Type(KuduColumnSchema::STRING)->Nullable(); |
| ASSERT_OK(b.Build(&schema)); |
| } |
| |
| unique_ptr<KuduTableCreator> table_creator(client_->NewTableCreator()); |
| table_creator->table_name(kTableName) |
| .schema(&schema) |
| .num_replicas(1) |
| .add_hash_partitions({ kCol0 }, 2) |
| .set_range_partition_columns({ kCol0 }); |
| |
| // Add a range partition with the table-wide hash schema. |
| { |
| unique_ptr<KuduPartialRow> lower(schema.NewRow()); |
| ASSERT_OK(lower->SetInt32(kCol0, -100)); |
| unique_ptr<KuduPartialRow> upper(schema.NewRow()); |
| ASSERT_OK(upper->SetInt32(kCol0, 0)); |
| table_creator->add_range_partition(lower.release(), upper.release()); |
| } |
| ASSERT_OK(table_creator->Create()); |
| |
| // Try to add hash partitions with wrong number of buckets in the |
| // range-specific hash schema. In Kudu C++ client, such mistakes are caught |
| // at the client side. |
| for (auto hash_bucket_num = -1; hash_bucket_num < 2; ++hash_bucket_num) { |
| SCOPED_TRACE(Substitute("hash schema with $0 buckets", hash_bucket_num)); |
| unique_ptr<KuduTableAlterer> alterer(client_->NewTableAlterer(kTableName)); |
| auto p = CreateRangePartition(schema, kCol0, 0, 100); |
| const auto s = p->add_hash_partitions({ kCol0 }, hash_bucket_num, 0); |
| ASSERT_TRUE(s.IsInvalidArgument()) << s.ToString(); |
| ASSERT_STR_CONTAINS(s.ToString(), kErrMsg); |
| } |
| } |
| |
| } // namespace client |
| } // namespace kudu |
