You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kudu.apache.org by al...@apache.org on 2023/04/03 16:26:27 UTC
[kudu] 01/02: KUDU-1945 Update auto incrementing counter during bootstrap
This is an automated email from the ASF dual-hosted git repository.
alexey pushed a commit to branch branch-1.17.x
in repository https://gitbox.apache.org/repos/asf/kudu.git
commit 7d107c24d7c9a61e3e8db3b1d00717ee787318d4
Author: Abhishek Chennaka <ac...@cloudera.com>
AuthorDate: Fri Jan 27 00:03:13 2023 +0530
KUDU-1945 Update auto incrementing counter during bootstrap
The auto incrementing counter would be reset to zero when the tablet
is being initialized. It is essential to have the correct value of
the counter. There are two cases:
1. WAL segments contain insert operations
In this scenario the WAL segments are replayed and since each insert
operation entry has auto incrementing counter which will be used for
the insert operations present in that entry, as long as we have the
latest insert operation entry in the WAL segments, the auto
incrementing counter is populated correctly during bootstrap.
2. WAL segments do not contain insert operations
In this case, we need to fetch the highest auto incrementing counter
value present in the data which is already flushed and update the
in-memory auto incrementing counter appropriately. This patch
accomplishes this task.
There are tests for the bootstrap scenarios where
1. We have no WAL segments with an INSERT OP containing the auto
incrementing column value. We rely on the auto incrementing counter
present in the data directories in this case.
2. We have no WAL segments with auto incrementing column value
and all the rows of the table are deleted. We reset the auto
incrementing counter in this case.
3. We have non committed replicate ops containing INSERT OPs with
the auto incrementing column values.
Manually tested the time taken to populate the auto incrementing
counter:
Columns - A Non Unique Primary Key column of type INT32
- 8 INT64 columns
- 5 STRING columns
Rows - 500k rows with data populated
Time taken in populating the counter during bootstrap:
Min - 235ms
Max - 466ms
Median - 312ms
The total time spent boostrapping the tablet was between 18-25
seconds.
Change-Id: I61b305efd7d5a065a2976327567163956c0c2184
Reviewed-on: http://gerrit.cloudera.org:8080/19445
Reviewed-by: Alexey Serbin <al...@apache.org>
Tested-by: Kudu Jenkins
(cherry picked from commit 8273792156d26c46f788558c896a0729ac2fedd1)
Reviewed-on: http://gerrit.cloudera.org:8080/19678
Reviewed-by: Yingchun Lai <la...@apache.org>
Tested-by: Yingchun Lai <la...@apache.org>
---
.../integration-tests/auto_incrementing-itest.cc | 319 ++++++++++++++++++---
src/kudu/tablet/tablet.cc | 47 ++-
src/kudu/tablet/tablet.h | 4 +
src/kudu/tserver/tablet_server-test-base.h | 8 +-
4 files changed, 336 insertions(+), 42 deletions(-)
diff --git a/src/kudu/integration-tests/auto_incrementing-itest.cc b/src/kudu/integration-tests/auto_incrementing-itest.cc
index ab2f9a9a3..28879890f 100644
--- a/src/kudu/integration-tests/auto_incrementing-itest.cc
+++ b/src/kudu/integration-tests/auto_incrementing-itest.cc
@@ -19,8 +19,11 @@
// of PK subsets, etc).
#include <memory>
+#include <ostream>
#include <string>
+#include <thread>
#include <type_traits>
+#include <utility>
#include <vector>
#include <glog/logging.h>
@@ -33,20 +36,22 @@
#include "kudu/common/partial_row.h"
#include "kudu/common/schema.h"
#include "kudu/common/wire_protocol.h"
-#include "kudu/gutil/ref_counted.h"
+#include "kudu/consensus/metadata.pb.h"
#include "kudu/gutil/strings/substitute.h"
-#include "kudu/mini-cluster/internal_mini_cluster.h"
+#include "kudu/mini-cluster/external_mini_cluster.h"
#include "kudu/rpc/rpc_controller.h"
-#include "kudu/tablet/tablet_replica.h"
-#include "kudu/tserver/mini_tablet_server.h"
+#include "kudu/tablet/tablet.pb.h"
#include "kudu/tserver/tablet_server-test-base.h"
-#include "kudu/tserver/tablet_server.h"
-#include "kudu/tserver/ts_tablet_manager.h"
#include "kudu/tserver/tserver.pb.h"
+#include "kudu/util/env_util.h"
+#include "kudu/util/monotime.h"
+#include "kudu/util/path_util.h"
#include "kudu/util/status.h"
#include "kudu/util/test_macros.h"
+#include "kudu/util/test_util.h"
using kudu::client::KuduClient;
+using kudu::client::KuduDelete;
using kudu::client::KuduColumnSchema;
using kudu::client::KuduInsert;
using kudu::client::KuduSchema;
@@ -55,8 +60,8 @@ using kudu::client::KuduSession;
using kudu::client::KuduTable;
using kudu::client::KuduTableCreator;
using kudu::client::sp::shared_ptr;
+using kudu::env_util::ListFilesInDir;
using kudu::rpc::RpcController;
-using kudu::tablet::TabletReplica;
using kudu::tserver::NewScanRequestPB;
using kudu::tserver::ScanResponsePB;
using kudu::tserver::ScanRequestPB;
@@ -73,26 +78,20 @@ static const char* const kTableName = "test-table";
static const int kNumTabletServers = 3;
static const int kNumRows = 200;
-class AutoIncrementingItest : public tserver::TabletServerTestBase {
+class AutoIncrementingItest : public KuduTest {
public:
-
void SetUp() override {
- TabletServerTestBase::SetUp();
- cluster::InternalMiniClusterOptions opts;
- opts.num_tablet_servers = kNumTabletServers;
- cluster_.reset(new cluster::InternalMiniCluster(env_, opts));
- ASSERT_OK(cluster_->Start());
- ASSERT_OK(cluster_->CreateClient(nullptr, &client_));
+ KuduTest::SetUp();
}
- Status CreateTableWithData() {
+ // Create a table with a single range partition.
+ Status CreateTableWithPartition() {
KuduSchemaBuilder b;
b.AddColumn("c0")->Type(KuduColumnSchema::INT32)->NotNull()->NonUniquePrimaryKey();
RETURN_NOT_OK(b.Build(&kudu_schema_));
- // Create a table with a range partition
int lower_bound = 0;
- int upper_bound = 200;
+ int upper_bound = 400;
unique_ptr<KuduPartialRow> lower(kudu_schema_.NewRow());
unique_ptr<KuduPartialRow> upper(kudu_schema_.NewRow());
@@ -108,19 +107,35 @@ class AutoIncrementingItest : public tserver::TabletServerTestBase {
.Create());
}
- Status InsertData() {
- RETURN_NOT_OK(client_->OpenTable(kTableName, &table_));
+ // Insert data into the above created table.
+ Status InsertData(int start_c0_value, int end_c0_value, int sleep_millis = 0) {
shared_ptr<KuduSession> session(client_->NewSession());
- for (int i = 0; i < kNumRows; i++) {
+ RETURN_NOT_OK(client_->OpenTable(kTableName, &table_));
+ for (int i = start_c0_value; i < end_c0_value; i++) {
unique_ptr<KuduInsert> insert(table_->NewInsert());
KuduPartialRow* row = insert->mutable_row();
RETURN_NOT_OK(row->SetInt32("c0", i));
RETURN_NOT_OK(session->Apply(insert.release()));
+ SleepFor(MonoDelta::FromMilliseconds(sleep_millis));
}
return Status::OK();
}
- // Returns a scan response from the tablet on the given tablet server.
+ // Delete row based on the row values passed.
+ Status DeleteRow(int c0_val, int auto_incrementing_id) {
+ shared_ptr<KuduSession> session = client_->NewSession();
+ session->SetTimeoutMillis(15 * 1000);
+ RETURN_NOT_OK(client_->OpenTable(kTableName, &table_));
+
+ unique_ptr<KuduDelete> del(table_->NewDelete());
+ auto* row = del->mutable_row();
+ RETURN_NOT_OK(row->SetInt32(0, c0_val));
+ RETURN_NOT_OK(row->SetInt64(1, auto_incrementing_id));
+ RETURN_NOT_OK(session->Apply(del.release()));
+ return Status::OK();
+ }
+
+ // Return a scan response from the tablet on the given tablet server.
Status ScanTablet(int ts, const string& tablet_id, vector<string>* results) {
ScanResponsePB resp;
RpcController rpc;
@@ -128,7 +143,8 @@ class AutoIncrementingItest : public tserver::TabletServerTestBase {
NewScanRequestPB* scan = req.mutable_new_scan_request();
scan->set_tablet_id(tablet_id);
- scan->set_read_mode(READ_LATEST);
+ scan->set_read_mode(READ_AT_SNAPSHOT);
+ scan->set_order_mode(ORDERED);
Schema schema = Schema({ ColumnSchema("c0", INT32),
ColumnSchema(Schema::GetAutoIncrementingColumnName(),
@@ -136,34 +152,265 @@ class AutoIncrementingItest : public tserver::TabletServerTestBase {
},2);
RETURN_NOT_OK(SchemaToColumnPBs(schema, scan->mutable_projected_columns()));
RETURN_NOT_OK(cluster_->tserver_proxy(ts)->Scan(req, &resp, &rpc));
- StringifyRowsFromResponse(schema, rpc, &resp, results);
+ tserver::TabletServerTestBase::StringifyRowsFromResponse(schema, rpc, &resp, results);
+ return Status::OK();
+ }
+
+ Status TestSetup(string* tablet_uuid) {
+ cluster::ExternalMiniClusterOptions opts;
+ // We ensure quick rolling and GC'ing of the wal segments and quick flushes to data.
+ // This is to make sure that when we are bootstrapping, we are not looking at the
+ // prior segments where the auto incrementing counter is present (INSERT ops).
+ opts.extra_tserver_flags = {
+ "--log_segment_size_bytes_for_tests=100",
+ "--log_max_segments_to_retain=1",
+ "--maintenance_manager_polling_interval_ms=10",
+ "--maintenance_manager_num_threads=4",
+ "--flush_threshold_secs=1",
+ "--flush_threshold_mb=0",
+ "--log_compression_codec=no_compression",
+ };
+ opts.num_tablet_servers = kNumTabletServers;
+ cluster_.reset(new cluster::ExternalMiniCluster(std::move(opts)));
+ RETURN_NOT_OK(cluster_->Start());
+ RETURN_NOT_OK(cluster_->CreateClient(nullptr, &client_));
+
+ // Create a table and insert data.
+ RETURN_NOT_OK(CreateTableWithPartition());
+ RETURN_NOT_OK(InsertData(0, kNumRows));
+
+ // Get the tablet UUID.
+ rpc::RpcController rpc;
+ tserver::ListTabletsRequestPB req;
+ tserver::ListTabletsResponsePB resp;
+ RETURN_NOT_OK(cluster_->tserver_proxy(0)->ListTablets(req, &resp, &rpc));
+ CHECK(resp.status_and_schema_size() == 1);
+ *tablet_uuid = resp.status_and_schema(0).tablet_status().tablet_id();
return Status::OK();
}
- protected:
- unique_ptr<cluster::InternalMiniCluster> cluster_;
+ protected:
+ unique_ptr<cluster::ExternalMiniCluster> cluster_;
KuduSchema kudu_schema_;
shared_ptr<KuduClient> client_;
shared_ptr<KuduTable> table_;
};
-TEST_F(AutoIncrementingItest, TestAutoIncrementingItest) {
- // Create a table and insert data
- ASSERT_OK(CreateTableWithData());
- ASSERT_OK(InsertData());
- // Scan all the tablet replicas
+TEST_F(AutoIncrementingItest, BasicInserts) {
+ cluster::ExternalMiniClusterOptions opts;
+ opts.num_tablet_servers = kNumTabletServers;
+ cluster_.reset(new cluster::ExternalMiniCluster(std::move(opts)));
+ ASSERT_OK(cluster_->Start());
+ ASSERT_OK(cluster_->CreateClient(nullptr, &client_));
+
+ // Create a table and insert data.
+ ASSERT_OK(CreateTableWithPartition());
+ ASSERT_OK(InsertData(0, kNumRows));
+
+ // Scan all the tablet replicas and validate the results.
for (int j = 0; j < kNumTabletServers; j++) {
- vector<scoped_refptr<TabletReplica>> replicas;
- auto* server = cluster_->mini_tablet_server(j);
- server->server()->tablet_manager()->GetTabletReplicas(&replicas);
- DCHECK_EQ(1, replicas.size());
+ auto server = cluster_->tserver_proxy(j);
+ rpc::RpcController rpc;
+ tserver::ListTabletsRequestPB req;
+ tserver::ListTabletsResponsePB resp;
+ server->ListTablets(req, &resp, &rpc);
+ ASSERT_EQ(1, resp.status_and_schema_size());
vector<string> results;
- ASSERT_OK(ScanTablet(j, replicas[0]->tablet_id(), &results));
+ ASSERT_OK(ScanTablet(j, resp.status_and_schema(0).tablet_status().tablet_id(), &results));
for (int i = 0; i < kNumRows; i++) {
ASSERT_EQ(Substitute("(int32 c0=$0, int64 $1=$2)", i,
Schema::GetAutoIncrementingColumnName(), i + 1), results[i]);
}
}
}
+
+TEST_F(AutoIncrementingItest, BootstrapWithNoWals) {
+ string tablet_uuid;
+ TestSetup(&tablet_uuid);
+
+ // Delete the first 100 rows to make sure at least latest 2 wals
+ // are not insert operations.
+ for (int i = 0; i < 100; i++) {
+ ASSERT_OK(DeleteRow(i, i + 1));
+ }
+ vector<string> wal_dir_files;
+
+ // Ensure the wals are GC'd in all the tablet servers.
+ int i = 0;
+ while (i < kNumTabletServers) {
+ ASSERT_OK(ListFilesInDir(env_,
+ JoinPathSegments(cluster_->tablet_server(0)->wal_dir(),
+ strings::Substitute("wals/$0", tablet_uuid)),
+ &wal_dir_files));
+ ASSERT_FALSE(wal_dir_files.empty());
+ if (wal_dir_files.size() > 3) {
+ // We are still yet to GC the wal segments, check back the server after 50ms.
+ SleepFor(MonoDelta::FromMilliseconds(50));
+ continue;
+ }
+ i++;
+ }
+
+ // Restart the cluster.
+ cluster_->Shutdown();
+ ASSERT_OK(cluster_->Restart());
+
+ // Insert new data and validate the auto incrementing column values.
+ ASSERT_OK(InsertData(kNumRows, kNumRows * 2));
+ for (int j = 0; j < kNumTabletServers; j++) {
+ vector<string> results;
+ ASSERT_OK(ScanTablet(j, tablet_uuid, &results));
+ LOG(INFO) << "Results size: " << results.size();
+ for (int i = 0; i < results.size(); i++) {
+ ASSERT_EQ(Substitute("(int32 c0=$0, int64 $1=$2)", i + 100,
+ Schema::GetAutoIncrementingColumnName(), i + 100 + 1), results[i]);
+
+ }
+ }
+}
+
+
+TEST_F(AutoIncrementingItest, BootstrapNoWalsNoData) {
+ string tablet_uuid;
+ TestSetup(&tablet_uuid);
+
+ // Delete all the rows.
+ for (int i = 0; i < kNumRows; i++) {
+ ASSERT_OK(DeleteRow(i, i + 1));
+ }
+
+ // Ensure the wals are GC'd in all the tablet servers and there are no
+ // rows present.
+ vector<string> wal_dir_files;
+ int i = 0;
+ while (i < kNumTabletServers) {
+ ASSERT_OK(ListFilesInDir(env_,
+ JoinPathSegments(cluster_->tablet_server(0)->wal_dir(),
+ strings::Substitute("wals/$0", tablet_uuid)),
+ &wal_dir_files));
+ ASSERT_FALSE(wal_dir_files.empty());
+ vector<string> results;
+ ASSERT_OK(ScanTablet(i, tablet_uuid, &results));
+ if (wal_dir_files.size() > 3 || !results.empty()) {
+ // We are still yet to GC the wal segments or flush the deletes,
+ // check back the server after 50ms.
+ SleepFor(MonoDelta::FromMilliseconds(50));
+ continue;
+ }
+ i++;
+ }
+
+ // Restart the cluster.
+ cluster_->Shutdown();
+ ASSERT_OK(cluster_->Restart());
+
+ // There could still be a WAL entry with a DELETE OP which is not persisted to the data.
+ // directory. This might cause the auto incrementing counter to be not set to 0 when fetching
+ // from the data directories. We restart servers to force elections which in turn
+ // writes more WAL entries which causes these flushes.
+ for (int i = 0; i < kNumTabletServers; i++) {
+ cluster_->tablet_server(i)->Shutdown();
+ ASSERT_OK(cluster_->tablet_server(i)->Restart());
+ }
+
+ // Insert new data and verify auto_incrementing_id starts from 1.
+ ASSERT_OK(InsertData(kNumRows, kNumRows * 2));
+ for (int j = 0; j < kNumTabletServers; j++) {
+ vector<string> results;
+ ASSERT_OK(ScanTablet(j, tablet_uuid, &results));
+ ASSERT_EQ(200, results.size());
+ for (int i = 0; i < results.size(); i++) {
+ ASSERT_EQ(Substitute("(int32 c0=$0, int64 $1=$2)", i + kNumRows,
+ Schema::GetAutoIncrementingColumnName(), i + 1), results[i]);
+ }
+ }
+}
+
+TEST_F(AutoIncrementingItest, BootstrapWalsDiverge) {
+ cluster::ExternalMiniClusterOptions opts;
+ opts.num_tablet_servers = kNumTabletServers;
+ cluster_.reset(new cluster::ExternalMiniCluster(std::move(opts)));
+ ASSERT_OK(cluster_->Start());
+ ASSERT_OK(cluster_->CreateClient(nullptr, &client_));
+
+ // Create a table.
+ ASSERT_OK(CreateTableWithPartition());
+
+ // Get the tablet server hosting the leader replica.
+ int leader_ts_index = -1;
+ string tablet_uuid;
+ for (int i = 0; i < kNumTabletServers; i++) {
+ rpc::RpcController rpc;
+ tserver::ListTabletsRequestPB req;
+ tserver::ListTabletsResponsePB resp;
+ cluster_->tserver_proxy(i)->ListTablets(req, &resp, &rpc);
+ ASSERT_EQ(1, resp.status_and_schema_size());
+ if (tablet_uuid.empty()) {
+ tablet_uuid = resp.status_and_schema(0).tablet_status().tablet_id();
+ }
+ if (resp.status_and_schema(0).role() == consensus::RaftPeerPB::LEADER) {
+ leader_ts_index = i;
+ break;
+ }
+ }
+ ASSERT_NE(-1, leader_ts_index);
+
+ // Shutdown the tablet servers in a separate thread but sleep for 100ms
+ // to ensure there is data being written to the server.
+ std::thread shutdown_thread([&]() {
+ // Ensure at least a couple of rows are attempted to be
+ // written before shutting down.
+ SleepFor(MonoDelta::FromMilliseconds(100));
+ // Shutdown the tablet servers hosting followers while the data is still being written.
+ for (int i = 0; i < kNumTabletServers; i++) {
+ if (i != leader_ts_index) {
+ cluster_->tablet_server(i)->Shutdown();
+ }
+ }
+ });
+ // Sleep between each write for less than Raft heartbeat interval.
+ // This is to ensure we won't write all the data before the servers are
+ // shutdown and we would have at least a few WAL non-committed replicate ops.
+ Status s = InsertData(0, kNumRows, 5);
+ shutdown_thread.join();
+ ASSERT_TRUE(!s.ok());
+
+ // Write 200 rows at the rate of 1 row every 5ms which are sent to the leader replica. After
+ // 100ms of starting to insert data, we shutdown the followers and at this point the write
+ // request is expected to 900ms more. Since the leader would mark the followers as
+ // unavailable after 3 lost hearbeats (1500ms), there will for sure be a situation where the
+ // leader has sent a write op and hasn't gotten the response from majority-1 number of
+ // followers. In this case the write op is not marked committed in the leader replica. All
+ // the writes including this are considered failed.
+ cluster_->tablet_server(leader_ts_index)->Shutdown();
+ for (int i = 0; i < kNumTabletServers; i++) {
+ if (i != leader_ts_index) {
+ ASSERT_OK(cluster_->tablet_server(i)->Restart());
+ }
+ }
+
+ // Insert more data.
+ ASSERT_OK(InsertData(kNumRows, kNumRows * 2));
+
+ // Start back the tablet server and ensure all the replicas contain the same data.
+ // We scan the replica on leader_ts_index last to give it reasonable time to start
+ // serving scans.
+ ASSERT_OK(cluster_->tablet_server(leader_ts_index)->Restart());
+ int j = (leader_ts_index + 1) % kNumTabletServers;
+ vector<vector<string>> results;
+ for (int i = 0; i < kNumTabletServers; i++) {
+ vector<string> result;
+ ASSERT_OK(ScanTablet(j, tablet_uuid, &result));
+ results.emplace_back(result);
+ j = ++j % kNumTabletServers;
+ }
+ ASSERT_EQ(kNumTabletServers, results.size());
+ for (int i = 0; i < kNumTabletServers - 1; i++) {
+ ASSERT_EQ(results[i].size(), results[i + 1].size());
+ for (int k = 0; k < results[i].size(); k++) {
+ ASSERT_EQ(results[i][k], results[i + 1][k]);
+ }
+ }
+}
} // namespace itest
} // namespace kudu
diff --git a/src/kudu/tablet/tablet.cc b/src/kudu/tablet/tablet.cc
index ef6b77cee..5041a19cf 100644
--- a/src/kudu/tablet/tablet.cc
+++ b/src/kudu/tablet/tablet.cc
@@ -47,6 +47,8 @@
#include "kudu/common/row_changelist.h"
#include "kudu/common/row_operations.h"
#include "kudu/common/row_operations.pb.h"
+#include "kudu/common/rowblock.h"
+#include "kudu/common/rowblock_memory.h"
#include "kudu/common/rowid.h"
#include "kudu/common/scan_spec.h"
#include "kudu/common/schema.h"
@@ -94,6 +96,7 @@
#include "kudu/util/process_memory.h"
#include "kudu/util/slice.h"
#include "kudu/util/status_callback.h"
+#include "kudu/util/stopwatch.h"
#include "kudu/util/throttler.h"
#include "kudu/util/trace.h"
#include "kudu/util/url-coding.h"
@@ -303,7 +306,6 @@ GROUP_FLAG_VALIDATOR(rowset_compaction, &ValidateRowsetCompactionGuard);
namespace kudu {
-class RowBlock;
struct IteratorStats;
namespace tablet {
@@ -430,10 +432,18 @@ Status Tablet::Open(const unordered_set<int64_t>& in_flight_txn_ids,
<< s.ToString();
return s;
}
-
rowsets_opened.emplace_back(std::move(rowset));
}
+ // Update the auto incrementing counter of the tablet from the data directories
+ if (schema()->has_auto_incrementing()) {
+ Status s = UpdateAutoIncrementingCounter(rowsets_opened);
+ if (!s.ok()) {
+ LOG_WITH_PREFIX(ERROR) << "Failed to update auto incrementing counter: " << s.ToString();
+ return s;
+ }
+ }
+
{
auto new_rowset_tree(make_shared<RowSetTree>());
RETURN_NOT_OK(new_rowset_tree->Reset(rowsets_opened));
@@ -482,6 +492,39 @@ Status Tablet::Open(const unordered_set<int64_t>& in_flight_txn_ids,
return Status::OK();
}
+Status Tablet::UpdateAutoIncrementingCounter(const RowSetVector& rowsets_opened) {
+ LOG_TIMING(INFO, "fetching auto increment counter") {
+ for (const shared_ptr<RowSet>& rowset: rowsets_opened) {
+ RowIteratorOptions opts;
+ opts.projection = schema().get();
+ opts.include_deleted_rows = false;
+ // TODO(achennaka): Materialize only the auto incrementing column for better performance
+ unique_ptr<RowwiseIterator> iter;
+ RETURN_NOT_OK(rowset->NewRowIterator(opts, &iter));
+ RETURN_NOT_OK(iter->Init(nullptr));
+ // The default size of 32K should be a good start as it would be increased if needed.
+ RowBlockMemory mem;
+ RowBlock block(&iter->schema(), 512, &mem);
+ while (iter->HasNext()) {
+ mem.Reset();
+ RETURN_NOT_OK(iter->NextBlock(&block));
+ const size_t nrows = block.nrows();
+ for (size_t i = 0; i < nrows; ++i) {
+ if (!block.selection_vector()->IsRowSelected(i)) {
+ continue;
+ }
+ int64 counter = *reinterpret_cast<const int64 *>(block.row(i).cell_ptr(
+ schema()->auto_incrementing_col_idx()));
+ if (counter > auto_incrementing_counter_) {
+ auto_incrementing_counter_ = counter;
+ }
+ }
+ }
+ }
+ }
+ return Status::OK();
+}
+
void Tablet::Stop() {
{
std::lock_guard<simple_spinlock> l(state_lock_);
diff --git a/src/kudu/tablet/tablet.h b/src/kudu/tablet/tablet.h
index 06f6a2a17..2bdf6dc34 100644
--- a/src/kudu/tablet/tablet.h
+++ b/src/kudu/tablet/tablet.h
@@ -130,6 +130,10 @@ class Tablet {
Status Open(const std::unordered_set<int64_t>& in_flight_txn_ids = std::unordered_set<int64_t>{},
const std::unordered_set<int64_t>& txn_ids_with_mrs = std::unordered_set<int64_t>{});
+ // Update the auto incrementing counter of the tablet
+ // if the tablet has the auto incrementing column in the schema.
+ Status UpdateAutoIncrementingCounter(const RowSetVector& rowsets_opened);
+
// Mark that the tablet has finished bootstrapping.
// This transitions from kBootstrapping to kOpen state.
// Returns an error if tablet has been stopped.
diff --git a/src/kudu/tserver/tablet_server-test-base.h b/src/kudu/tserver/tablet_server-test-base.h
index 8304202ae..43478e873 100644
--- a/src/kudu/tserver/tablet_server-test-base.h
+++ b/src/kudu/tserver/tablet_server-test-base.h
@@ -101,10 +101,10 @@ class TabletServerTestBase : public KuduTest {
TabletServerServiceProxy* proxy = nullptr,
uint32_t call_seq_id = 1);
- void StringifyRowsFromResponse(const Schema& projection,
- const rpc::RpcController& rpc,
- ScanResponsePB* resp,
- std::vector<std::string>* results);
+ static void StringifyRowsFromResponse(const Schema& projection,
+ const rpc::RpcController& rpc,
+ ScanResponsePB* resp,
+ std::vector<std::string>* results);
void ShutdownTablet();