You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kudu.apache.org by to...@apache.org on 2016/09/09 19:13:46 UTC
[4/4] kudu git commit: KUDU-236 (part 1). Implement tablet history GC
KUDU-236 (part 1). Implement tablet history GC
This patch defines the concept of an "ancient history mark" (AHM) and a
background job that removes history from disk that represents changes
occurring prior to the ancient history mark.
There is also a mechanism to reject requests for scans if they attempt
to open a snapshot scan at a timestamp prior to the AHM.
Included tests:
* Test for Redo GC via major delta compaction
* Test for history GC via tablet flush
* Test for Undo GC via merge compaction
* Test for entire-row GC via merge compaction
* Test for ghost rows in multiple RS reinsert case
* Test for reupdating missed deltas
* Test for partial rowset history GC using alternating rows
* Test for major delta compaction against a subset of columns
* Test for reinsert after delete that is prior to AHM
* Test for opening a scanner at TS < AHM and rejecting that scanner at
the RPC level
Tests coming in a follow-up commit:
* Randomized test
Missing features coming in follow-up commits:
* Background task that schedules and performs undo GC
* GC history on delta flush
Change-Id: If9833a863f118eb82be80ea56204d0d9141611c2
Reviewed-on: http://gerrit.cloudera.org:8080/3076
Tested-by: Kudu Jenkins
Reviewed-by: Todd Lipcon <to...@apache.org>
Project: http://git-wip-us.apache.org/repos/asf/kudu/repo
Commit: http://git-wip-us.apache.org/repos/asf/kudu/commit/be719edc
Tree: http://git-wip-us.apache.org/repos/asf/kudu/tree/be719edc
Diff: http://git-wip-us.apache.org/repos/asf/kudu/diff/be719edc
Branch: refs/heads/master
Commit: be719edc3581802e094c3af6a88d67acba44ba71
Parents: d5cc572
Author: Mike Percy <mp...@apache.org>
Authored: Mon Jul 18 13:40:14 2016 -0700
Committer: Todd Lipcon <to...@apache.org>
Committed: Fri Sep 9 19:13:18 2016 +0000
----------------------------------------------------------------------
src/kudu/common/timestamp.h | 10 +
src/kudu/integration-tests/CMakeLists.txt | 1 +
.../tablet_history_gc-itest.cc | 52 ++
src/kudu/server/clock.h | 6 +
src/kudu/server/hybrid_clock.cc | 8 +-
src/kudu/server/hybrid_clock.h | 4 +-
src/kudu/tablet/CMakeLists.txt | 1 +
src/kudu/tablet/compaction-test.cc | 12 +-
src/kudu/tablet/compaction.cc | 161 ++++--
src/kudu/tablet/compaction.h | 66 ++-
src/kudu/tablet/delta_compaction.cc | 39 +-
src/kudu/tablet/delta_compaction.h | 6 +-
src/kudu/tablet/diskrowset.cc | 15 +-
src/kudu/tablet/diskrowset.h | 8 +-
src/kudu/tablet/mutation.h | 6 +-
src/kudu/tablet/rowset_metadata.h | 2 +-
src/kudu/tablet/tablet-harness.h | 16 +-
src/kudu/tablet/tablet-test-base.h | 79 ++-
src/kudu/tablet/tablet-test-util.h | 9 +-
src/kudu/tablet/tablet-test.cc | 11 +-
src/kudu/tablet/tablet.cc | 78 ++-
src/kudu/tablet/tablet.h | 11 +
src/kudu/tablet/tablet_history_gc-test.cc | 511 +++++++++++++++++++
src/kudu/tserver/tablet_service.cc | 103 ++--
24 files changed, 1047 insertions(+), 168 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kudu/blob/be719edc/src/kudu/common/timestamp.h
----------------------------------------------------------------------
diff --git a/src/kudu/common/timestamp.h b/src/kudu/common/timestamp.h
index 9c9731a..c1d46c9 100644
--- a/src/kudu/common/timestamp.h
+++ b/src/kudu/common/timestamp.h
@@ -50,8 +50,14 @@ class Timestamp {
// Encode the timestamp to the given buffer.
void EncodeTo(faststring *dst) const;
+ // Returns -1 if this < other.
+ // Returns 1 if this > other.
+ // Returns 0 if this == other.
int CompareTo(const Timestamp &other) const;
+ // Returns true if this comes before other.
+ bool ComesBefore(const Timestamp &other) const;
+
std::string ToString() const;
// Returns this Timestamp as an uint64_t
@@ -92,6 +98,10 @@ inline int Timestamp::CompareTo(const Timestamp &other) const {
return 0;
}
+inline bool Timestamp::ComesBefore(const Timestamp &other) const {
+ return v < other.v;
+}
+
} // namespace kudu
#endif /* KUDU_COMMON_TIMESTAMP_H_ */
http://git-wip-us.apache.org/repos/asf/kudu/blob/be719edc/src/kudu/integration-tests/CMakeLists.txt
----------------------------------------------------------------------
diff --git a/src/kudu/integration-tests/CMakeLists.txt b/src/kudu/integration-tests/CMakeLists.txt
index 89b5fb8..aa6a673 100644
--- a/src/kudu/integration-tests/CMakeLists.txt
+++ b/src/kudu/integration-tests/CMakeLists.txt
@@ -64,6 +64,7 @@ ADD_KUDU_TEST(raft_consensus-itest RUN_SERIAL true)
ADD_KUDU_TEST(registration-test RESOURCE_LOCK "master-web-port")
ADD_KUDU_TEST(table_locations-itest)
ADD_KUDU_TEST(tablet_copy-itest)
+ADD_KUDU_TEST(tablet_history_gc-itest)
ADD_KUDU_TEST(tablet_replacement-itest)
ADD_KUDU_TEST(ts_recovery-itest)
ADD_KUDU_TEST(ts_tablet_manager-itest)
http://git-wip-us.apache.org/repos/asf/kudu/blob/be719edc/src/kudu/integration-tests/tablet_history_gc-itest.cc
----------------------------------------------------------------------
diff --git a/src/kudu/integration-tests/tablet_history_gc-itest.cc b/src/kudu/integration-tests/tablet_history_gc-itest.cc
new file mode 100644
index 0000000..0ee404c
--- /dev/null
+++ b/src/kudu/integration-tests/tablet_history_gc-itest.cc
@@ -0,0 +1,52 @@
+// 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 "kudu/integration-tests/external_mini_cluster-itest-base.h"
+#include "kudu/integration-tests/test_workload.h"
+
+using kudu::client::KuduScanner;
+using kudu::client::KuduTable;
+using kudu::client::sp::shared_ptr;
+
+namespace kudu {
+
+class TabletHistoryGcITest : public ExternalMiniClusterITestBase {
+};
+
+// Check that attempts to scan prior to the ancient history mark fail.
+TEST_F(TabletHistoryGcITest, TestSnapshotScanBeforeAHM) {
+ NO_FATALS(StartCluster({ "--tablet_history_max_age_sec=0" }));
+
+ // Create a tablet so we can scan it.
+ TestWorkload workload(cluster_.get());
+ workload.Setup();
+
+ // When the tablet history max age is set to 0, it's not possible to do a
+ // snapshot scan without a timestamp because it's illegal to open a snapshot
+ // prior to the AHM. When a snapshot timestamp is not specified, we decide on
+ // the timestamp of the snapshot before checking that it's lower than the
+ // current AHM. This test verifies that scans prior to the AHM are rejected.
+ shared_ptr<KuduTable> table;
+ ASSERT_OK(client_->OpenTable(TestWorkload::kDefaultTableName, &table));
+ KuduScanner scanner(table.get());
+ ASSERT_OK(scanner.SetReadMode(KuduScanner::READ_AT_SNAPSHOT));
+ Status s = scanner.Open();
+ ASSERT_TRUE(s.IsInvalidArgument()) << s.ToString();
+ ASSERT_STR_CONTAINS(s.ToString(), "Snapshot timestamp is earlier than the ancient history mark");
+}
+
+} // namespace kudu
http://git-wip-us.apache.org/repos/asf/kudu/blob/be719edc/src/kudu/server/clock.h
----------------------------------------------------------------------
diff --git a/src/kudu/server/clock.h b/src/kudu/server/clock.h
index e55fa5b..777667b 100644
--- a/src/kudu/server/clock.h
+++ b/src/kudu/server/clock.h
@@ -65,6 +65,12 @@ class Clock : public RefCountedThreadSafe<Clock> {
// Indicates whether this clock supports the required external consistency mode.
virtual bool SupportsExternalConsistencyMode(ExternalConsistencyMode mode) = 0;
+ // Indicates whether the clock has a physical component to its timestamps
+ // (wallclock time).
+ virtual bool HasPhysicalComponent() {
+ return false;
+ }
+
// Update the clock with a transaction timestamp originating from
// another server. For instance replicas can call this so that,
// if elected leader, they are guaranteed to generate timestamps
http://git-wip-us.apache.org/repos/asf/kudu/blob/be719edc/src/kudu/server/hybrid_clock.cc
----------------------------------------------------------------------
diff --git a/src/kudu/server/hybrid_clock.cc b/src/kudu/server/hybrid_clock.cc
index 6d79d1f..cc30015 100644
--- a/src/kudu/server/hybrid_clock.cc
+++ b/src/kudu/server/hybrid_clock.cc
@@ -299,6 +299,10 @@ bool HybridClock::SupportsExternalConsistencyMode(ExternalConsistencyMode mode)
return true;
}
+bool HybridClock::HasPhysicalComponent() {
+ return true;
+}
+
Status HybridClock::WaitUntilAfter(const Timestamp& then_latest,
const MonoTime& deadline) {
TRACE_EVENT0("clock", "HybridClock::WaitUntilAfter");
@@ -343,8 +347,8 @@ Status HybridClock::WaitUntilAfter(const Timestamp& then_latest,
return Status::OK();
}
- Status HybridClock::WaitUntilAfterLocally(const Timestamp& then,
- const MonoTime& deadline) {
+Status HybridClock::WaitUntilAfterLocally(const Timestamp& then,
+ const MonoTime& deadline) {
while (true) {
Timestamp now;
uint64_t error;
http://git-wip-us.apache.org/repos/asf/kudu/blob/be719edc/src/kudu/server/hybrid_clock.h
----------------------------------------------------------------------
diff --git a/src/kudu/server/hybrid_clock.h b/src/kudu/server/hybrid_clock.h
index 1110f9b..d712c82 100644
--- a/src/kudu/server/hybrid_clock.h
+++ b/src/kudu/server/hybrid_clock.h
@@ -60,6 +60,8 @@ class HybridClock : public Clock {
// HybridClock supports all external consistency modes.
virtual bool SupportsExternalConsistencyMode(ExternalConsistencyMode mode) OVERRIDE;
+ virtual bool HasPhysicalComponent() OVERRIDE;
+
// Blocks the caller thread until the true time is after 'then'.
// In other words, waits until the HybridClock::Now() on _all_ nodes
// will return a value greater than 'then'.
@@ -136,7 +138,7 @@ class HybridClock : public Clock {
uint64_t logical_value);
// Creates a new timestamp whose physical time is GetPhysicalValue(original) +
- // 'micros_to_add' and which retains the same logical value.
+ // 'to_add' and which retains the same logical value.
static Timestamp AddPhysicalTimeToTimestamp(const Timestamp& original,
const MonoDelta& to_add);
http://git-wip-us.apache.org/repos/asf/kudu/blob/be719edc/src/kudu/tablet/CMakeLists.txt
----------------------------------------------------------------------
diff --git a/src/kudu/tablet/CMakeLists.txt b/src/kudu/tablet/CMakeLists.txt
index 9e8a00f..8fda82c 100644
--- a/src/kudu/tablet/CMakeLists.txt
+++ b/src/kudu/tablet/CMakeLists.txt
@@ -115,6 +115,7 @@ ADD_KUDU_TEST(delta_compaction-test)
ADD_KUDU_TEST(mt-rowset_delta_compaction-test)
ADD_KUDU_TEST(major_delta_compaction-test)
ADD_KUDU_TEST(transactions/transaction_tracker-test)
+ADD_KUDU_TEST(tablet_history_gc-test)
ADD_KUDU_TEST(tablet_peer-test)
ADD_KUDU_TEST(tablet_random_access-test)
ADD_KUDU_TEST(tablet_throttle-test)
http://git-wip-us.apache.org/repos/asf/kudu/blob/be719edc/src/kudu/tablet/compaction-test.cc
----------------------------------------------------------------------
diff --git a/src/kudu/tablet/compaction-test.cc b/src/kudu/tablet/compaction-test.cc
index 738c8ce..923f3ea 100644
--- a/src/kudu/tablet/compaction-test.cc
+++ b/src/kudu/tablet/compaction-test.cc
@@ -208,7 +208,7 @@ class TestCompaction : public KuduRowSetTest {
BloomFilterSizing::BySizeAndFPRate(32*1024, 0.01f),
roll_threshold);
ASSERT_OK(rsw.Open());
- ASSERT_OK(FlushCompactionInput(input, snap, &rsw));
+ ASSERT_OK(FlushCompactionInput(input, snap, HistoryGcOpts::Disabled(), &rsw));
ASSERT_OK(rsw.Finish());
vector<shared_ptr<RowSetMetadata> > metas;
@@ -378,7 +378,8 @@ class TestCompaction : public KuduRowSetTest {
BloomFilterSizing::BySizeAndFPRate(32 * 1024, 0.01f),
1024 * 1024); // 1 MB
ASSERT_OK(rdrsw.Open());
- ASSERT_OK(FlushCompactionInput(compact_input.get(), merge_snap, &rdrsw));
+ ASSERT_OK(FlushCompactionInput(compact_input.get(), merge_snap, HistoryGcOpts::Disabled(),
+ &rdrsw));
ASSERT_OK(rdrsw.Finish());
}
}
@@ -575,7 +576,8 @@ TEST_F(TestCompaction, TestOneToOne) {
string dummy_name = "";
- ASSERT_OK(ReupdateMissedDeltas(dummy_name, input.get(), snap, snap2, { rs }));
+ ASSERT_OK(ReupdateMissedDeltas(dummy_name, input.get(), HistoryGcOpts::Disabled(), snap, snap2,
+ { rs }));
// If we look at the contents of the DiskRowSet now, we should see the "re-updated" data.
vector<string> out;
@@ -628,10 +630,10 @@ TEST_F(TestCompaction, TestKUDU102) {
string dummy_name = "";
// This would fail without KUDU-102
- ASSERT_OK(ReupdateMissedDeltas(dummy_name, input.get(), snap, snap2, { rs, rs_b }));
+ ASSERT_OK(ReupdateMissedDeltas(dummy_name, input.get(), HistoryGcOpts::Disabled(), snap, snap2,
+ { rs, rs_b }));
}
-
// Test compacting when all of the inputs and the output have the same schema
TEST_F(TestCompaction, TestMerge) {
vector<Schema> schemas;
http://git-wip-us.apache.org/repos/asf/kudu/blob/be719edc/src/kudu/tablet/compaction.cc
----------------------------------------------------------------------
diff --git a/src/kudu/tablet/compaction.cc b/src/kudu/tablet/compaction.cc
index 37a8877..53bf505 100644
--- a/src/kudu/tablet/compaction.cc
+++ b/src/kudu/tablet/compaction.cc
@@ -30,6 +30,7 @@
#include "kudu/gutil/map-util.h"
#include "kudu/gutil/stl_util.h"
#include "kudu/gutil/strings/substitute.h"
+#include "kudu/server/hybrid_clock.h"
#include "kudu/tablet/cfile_set.h"
#include "kudu/tablet/delta_store.h"
#include "kudu/tablet/delta_tracker.h"
@@ -38,6 +39,7 @@
#include "kudu/tablet/transactions/write_transaction.h"
#include "kudu/util/debug/trace_event.h"
+using kudu::server::HybridClock;
using std::shared_ptr;
using std::unique_ptr;
using std::unordered_set;
@@ -593,9 +595,34 @@ void RowSetsInCompaction::DumpToLog() const {
}
}
+void RemoveAncientUndos(const HistoryGcOpts& history_gc_opts, CompactionInputRow* row) {
+ if (!history_gc_opts.gc_enabled()) {
+ return;
+ }
+
+ DVLOG(5) << "Ancient history mark: " << history_gc_opts.ancient_history_mark().ToString()
+ << ": " << HybridClock::StringifyTimestamp(history_gc_opts.ancient_history_mark());
+
+ Mutation *prev_undo = nullptr;
+ Mutation *undo_mut = row->undo_head;
+ while (undo_mut != nullptr) {
+ if (history_gc_opts.IsAncientHistory(undo_mut->timestamp())) {
+ // Drop all undos following this one in the list; Their timestamps will be lower.
+ if (prev_undo != nullptr) {
+ prev_undo->set_next(nullptr);
+ } else {
+ row->undo_head = nullptr;
+ }
+ break;
+ }
+ prev_undo = undo_mut;
+ undo_mut = undo_mut->next();
+ }
+}
Status ApplyMutationsAndGenerateUndos(const MvccSnapshot& snap,
const CompactionInputRow& src_row,
+ const HistoryGcOpts& history_gc_opts,
const Schema* base_schema,
Mutation** new_undo_head,
Mutation** new_redo_head,
@@ -603,12 +630,14 @@ Status ApplyMutationsAndGenerateUndos(const MvccSnapshot& snap,
RowBlockRow* dst_row,
bool* is_garbage_collected,
uint64_t* num_rows_history_truncated) {
- // TODO actually perform garbage collection (KUDU-236).
- // Right now we persist all mutations.
+ // In most cases, we don't want to remove all traces of this row after
+ // compaction.
*is_garbage_collected = false;
const Schema* dst_schema = dst_row->schema();
+ // At the time of writing, REINSERT is never encoded as an UNDO and the base
+ // data can never be in a deleted state (see KUDU-237).
bool is_deleted = false;
#define ERROR_LOG_CONTEXT \
@@ -628,6 +657,8 @@ Status ApplyMutationsAndGenerateUndos(const MvccSnapshot& snap,
Mutation* undo_head = const_cast<Mutation*>(src_row.undo_head);
Mutation* redo_head = nullptr;
+ // Convert the redos into undos. Any redos that are eligible for history GC
+ // are applied and then thrown away (they don't generate undos).
for (const Mutation *redo_mut = src_row.redo_head;
redo_mut != nullptr;
redo_mut = redo_mut->acquire_next()) {
@@ -667,6 +698,11 @@ Status ApplyMutationsAndGenerateUndos(const MvccSnapshot& snap,
continue;
}
+ if (history_gc_opts.IsAncientHistory(redo_mut->timestamp())) {
+ // Don't generate an undo for this.
+ continue;
+ }
+
// create the UNDO mutation in the provided arena.
current_undo = Mutation::CreateInArena(arena, redo_mut->timestamp(),
undo_encoder.as_changelist());
@@ -685,6 +721,8 @@ Status ApplyMutationsAndGenerateUndos(const MvccSnapshot& snap,
redo_decoder.TwiddleDeleteStatus(&is_deleted);
if (redo_decoder.is_reinsert()) {
+ *is_garbage_collected = false;
+
// Right now when a REINSERT mutation is found it is treated as a new insert and it
// clears the whole row history before it.
@@ -708,22 +746,30 @@ Status ApplyMutationsAndGenerateUndos(const MvccSnapshot& snap,
redo_head = nullptr;
if ((*num_rows_history_truncated)++ == 0) {
- LOG(WARNING) << "Found REINSERT REDO truncating row history for "
- << ERROR_LOG_CONTEXT << " Note: this warning will appear "
- "only for the first truncated row";
+ LOG(WARNING) << "Found REINSERT REDO truncating row history for " << ERROR_LOG_CONTEXT
+ << " Note: this warning will appear only for the first truncated row";
}
if (PREDICT_FALSE(VLOG_IS_ON(2))) {
VLOG(2) << "Found REINSERT REDO, cannot create UNDO for it, resetting row history "
- " under snap: " << snap.ToString() << ERROR_LOG_CONTEXT;
+ "under snapshot: " << snap.ToString() << ERROR_LOG_CONTEXT;
}
+
+ // Is a delete.
} else {
- // Delete mutations are left as redos
+ // Delete mutations are left as redos. Encode the DELETE as a redo.
undo_encoder.SetToDelete();
- // Encode the DELETE as a redo
redo_head = Mutation::CreateInArena(arena,
redo_mut->timestamp(),
undo_encoder.as_changelist());
+
+ if (history_gc_opts.IsAncientHistory(redo_mut->timestamp())) {
+ // This row was deleted prior to the ancient history mark. Remove all
+ // traces of this row. This flag is set to false once again if there
+ // is a subsequent reinsert (see above).
+ *is_garbage_collected = true;
+ }
+
}
} else {
LOG(FATAL) << "Unknown mutation type!" << ERROR_LOG_CONTEXT;
@@ -740,6 +786,7 @@ Status ApplyMutationsAndGenerateUndos(const MvccSnapshot& snap,
Status FlushCompactionInput(CompactionInput* input,
const MvccSnapshot& snap,
+ const HistoryGcOpts& history_gc_opts,
RollingDiskRowSetWriter* out) {
RETURN_NOT_OK(input->Init());
vector<CompactionInputRow> rows;
@@ -754,28 +801,32 @@ Status FlushCompactionInput(CompactionInput* input,
RETURN_NOT_OK(input->PrepareBlock(&rows));
int n = 0;
- for (const CompactionInputRow &input_row : rows) {
+ for (int i = 0; i < rows.size(); i++) {
+ CompactionInputRow* input_row = &rows[i];
RETURN_NOT_OK(out->RollIfNecessary());
- const Schema* schema = input_row.row.schema();
+ const Schema* schema = input_row->row.schema();
DCHECK_SCHEMA_EQ(*schema, out->schema());
DCHECK(schema->has_column_ids());
RowBlockRow dst_row = block.row(n);
- RETURN_NOT_OK(CopyRow(input_row.row, &dst_row, static_cast<Arena*>(nullptr)));
+ RETURN_NOT_OK(CopyRow(input_row->row, &dst_row, static_cast<Arena*>(nullptr)));
- DVLOG(2) << "Input Row: " << dst_row.schema()->DebugRow(dst_row) <<
- " RowId: " << input_row.row.row_index() <<
- " Undo Mutations: " << Mutation::StringifyMutationList(*schema, input_row.undo_head) <<
- " Redo Mutations: " << Mutation::StringifyMutationList(*schema, input_row.redo_head);
+ DVLOG(3) << "Input Row: " << dst_row.schema()->DebugRow(dst_row) <<
+ " RowIndex: " << input_row->row.row_index() <<
+ " Undo Mutations: " << Mutation::StringifyMutationList(*schema, input_row->undo_head) <<
+ " Redo Mutations: " << Mutation::StringifyMutationList(*schema, input_row->redo_head);
// Collect the new UNDO/REDO mutations.
Mutation* new_undos_head = nullptr;
Mutation* new_redos_head = nullptr;
bool is_garbage_collected;
+ uint64_t prev_num_rows_history_truncated = num_rows_history_truncated;
+ RemoveAncientUndos(history_gc_opts, input_row);
RETURN_NOT_OK(ApplyMutationsAndGenerateUndos(snap,
- input_row,
+ *input_row,
+ history_gc_opts,
schema,
&new_undos_head,
&new_redos_head,
@@ -784,33 +835,32 @@ Status FlushCompactionInput(CompactionInput* input,
&is_garbage_collected,
&num_rows_history_truncated));
+ DVLOG(4) << "Output Row: " << dst_row.schema()->DebugRow(dst_row) <<
+ "; RowIndex: " << dst_row.row_index() <<
+ "; Undo Mutations: " << Mutation::StringifyMutationList(*schema, new_undos_head) <<
+ "; Redo Mutations: " << Mutation::StringifyMutationList(*schema, new_redos_head) <<
+ "; Was garbage collected? " << is_garbage_collected <<
+ "; Was history truncated? " <<
+ (num_rows_history_truncated > prev_num_rows_history_truncated);
+
// Whether this row was garbage collected
if (is_garbage_collected) {
- DVLOG(2) << "Garbage Collected!";
// Don't flush the row.
continue;
}
- rowid_t index_in_current_drs_;
+ rowid_t index_in_current_drs;
- // We should always have UNDO deltas, until we implement delta GC. For now,
- // this is a convenient assertion to catch bugs like KUDU-632.
- CHECK(new_undos_head != nullptr) <<
- "Writing an output row with no UNDOs: "
- "Input Row: " << dst_row.schema()->DebugRow(dst_row) <<
- " RowId: " << input_row.row.row_index() <<
- " Undo Mutations: " << Mutation::StringifyMutationList(*schema, input_row.undo_head) <<
- " Redo Mutations: " << Mutation::StringifyMutationList(*schema, input_row.redo_head);
- out->AppendUndoDeltas(dst_row.row_index(), new_undos_head, &index_in_current_drs_);
+ if (new_undos_head != nullptr) {
+ out->AppendUndoDeltas(dst_row.row_index(), new_undos_head, &index_in_current_drs);
+ }
if (new_redos_head != nullptr) {
- out->AppendRedoDeltas(dst_row.row_index(), new_redos_head, &index_in_current_drs_);
+ out->AppendRedoDeltas(dst_row.row_index(), new_redos_head, &index_in_current_drs);
}
- DVLOG(2) << "Output Row: " << dst_row.schema()->DebugRow(dst_row) <<
- " RowId: " << index_in_current_drs_
- << " Undo Mutations: " << Mutation::StringifyMutationList(*schema, new_undos_head)
- << " Redo Mutations: " << Mutation::StringifyMutationList(*schema, new_redos_head);
+ DVLOG(5) << "Output Row: " << dst_row.schema()->DebugRow(dst_row)
+ << "; RowId: " << index_in_current_drs;
n++;
if (n == block.nrows()) {
@@ -829,20 +879,21 @@ Status FlushCompactionInput(CompactionInput* input,
if (num_rows_history_truncated > 0) {
LOG(WARNING) << "Total " << num_rows_history_truncated
- << " rows lost some history due to REINSERT after DELETE";
+ << " rows lost some history due to REINSERT after DELETE";
}
return Status::OK();
}
Status ReupdateMissedDeltas(const string &tablet_name,
CompactionInput *input,
+ const HistoryGcOpts& history_gc_opts,
const MvccSnapshot &snap_to_exclude,
const MvccSnapshot &snap_to_include,
const RowSetVector &output_rowsets) {
TRACE_EVENT0("tablet", "ReupdateMissedDeltas");
RETURN_NOT_OK(input->Init());
- VLOG(1) << "Re-updating missed deltas between snapshot " <<
+ VLOG(1) << "Reupdating missed deltas between snapshot " <<
snap_to_exclude.ToString() << " and " << snap_to_include.ToString();
// Collect the delta trackers that we'll push the updates into.
@@ -851,7 +902,7 @@ Status ReupdateMissedDeltas(const string &tablet_name,
delta_trackers.push_back(down_cast<DiskRowSet *>(rs.get())->delta_tracker());
}
- // The map of updated delta trackers, indexed by id.
+ // The set of updated delta trackers.
unordered_set<DeltaTracker*> updated_trackers;
// When we apply the updates to the new DMS, there is no need to anchor them
@@ -873,27 +924,39 @@ Status ReupdateMissedDeltas(const string &tablet_name,
RETURN_NOT_OK(key_projector.Init());
faststring buf;
- rowid_t row_idx = 0;
+ rowid_t output_row_offset = 0;
while (input->HasMoreBlocks()) {
RETURN_NOT_OK(input->PrepareBlock(&rows));
for (const CompactionInputRow &row : rows) {
- DVLOG(2) << "Revisiting row: " << schema->DebugRow(row.row) <<
+ DVLOG(4) << "Revisiting row: " << schema->DebugRow(row.row) <<
" Redo Mutations: " << Mutation::StringifyMutationList(*schema, row.redo_head) <<
" Undo Mutations: " << Mutation::StringifyMutationList(*schema, row.undo_head);
+ bool is_garbage_collected = false;
for (const Mutation *mut = row.redo_head;
mut != nullptr;
mut = mut->acquire_next()) {
+ is_garbage_collected = false;
RowChangeListDecoder decoder(mut->changelist());
RETURN_NOT_OK(decoder.Init());
if (snap_to_exclude.IsCommitted(mut->timestamp())) {
// This update was already taken into account in the first phase of the
- // compaction.
+ // compaction. We don't need to reapply it.
+
+ // But was this row GCed at flush time?
+ if (decoder.is_delete() &&
+ history_gc_opts.IsAncientHistory(mut->timestamp())) {
+ is_garbage_collected = true;
+ }
continue;
}
+ // We should never be able to get to this point if a row has been
+ // garbage collected.
+ DCHECK(!is_garbage_collected);
+
// We should never see a REINSERT in an input RowSet which was not
// caught in the original flush. REINSERT only occurs when an INSERT is
// done to a row when a ghost is already present for that row in
@@ -920,7 +983,7 @@ Status ReupdateMissedDeltas(const string &tablet_name,
// Therefore, it's already present in the output rowset, and we don't need
// to copy it in.
- DVLOG(2) << "Skipping already-duplicated delta for row " << row_idx
+ DVLOG(3) << "Skipping already-duplicated delta for row " << output_row_offset
<< " @" << mut->timestamp() << ": " << mut->changelist().ToString(*schema);
continue;
}
@@ -928,15 +991,15 @@ Status ReupdateMissedDeltas(const string &tablet_name,
// Otherwise, this is an update that arrived after the snapshot for the first
// pass, but before the DuplicatingRowSet was swapped in. We need to transfer
// this over to the output rowset.
- DVLOG(1) << "Flushing missed delta for row " << row_idx
- << " @" << mut->timestamp() << ": " << mut->changelist().ToString(*schema);
+ DVLOG(3) << "Flushing missed delta for row " << output_row_offset
+ << " @" << mut->timestamp() << ": " << mut->changelist().ToString(*schema);
DeltaTracker *cur_tracker = delta_trackers.front();
// The index on the input side isn't necessarily the index on the output side:
// we may have output several small DiskRowSets, so we need to find the index
// relative to the current one.
- int64_t idx_in_delta_tracker = row_idx - delta_tracker_base_row;
+ int64_t idx_in_delta_tracker = output_row_offset - delta_tracker_base_row;
while (idx_in_delta_tracker >= cur_tracker->num_rows()) {
// If the current index is higher than the total number of rows in the current
// DeltaTracker, that means we're now processing the next one in the list.
@@ -955,7 +1018,7 @@ Status ReupdateMissedDeltas(const string &tablet_name,
mut->changelist(),
max_op_id,
result.get());
- DCHECK(s.ok()) << "Failed update on compaction for row " << row_idx
+ DCHECK(s.ok()) << "Failed update on compaction for row " << output_row_offset
<< " @" << mut->timestamp() << ": " << mut->changelist().ToString(*schema);
if (s.ok()) {
// Update the set of delta trackers with the one we've just updated.
@@ -963,15 +1026,19 @@ Status ReupdateMissedDeltas(const string &tablet_name,
}
}
- // TODO when garbage collection kicks in we need to take care that
- // CGed rows do not increment this.
- row_idx++;
+ if (is_garbage_collected) {
+ DVLOG(4) << "Skipping GCed input row: " << schema->DebugRow(row.row)
+ << " while reupdating missed deltas";
+ } else {
+ // Important: We must ensure that GCed rows do not increment the output
+ // row offset.
+ output_row_offset++;
+ }
}
RETURN_NOT_OK(input->FinishBlock());
}
-
// Flush the trackers that got updated, this will make sure that all missed deltas
// get flushed before we update the tablet's metadata at the end of compaction/flush.
// Note that we don't flush the metadata here, as to we will update the metadata
http://git-wip-us.apache.org/repos/asf/kudu/blob/be719edc/src/kudu/tablet/compaction.h
----------------------------------------------------------------------
diff --git a/src/kudu/tablet/compaction.h b/src/kudu/tablet/compaction.h
index 9258ea0..b706879 100644
--- a/src/kudu/tablet/compaction.h
+++ b/src/kudu/tablet/compaction.h
@@ -32,6 +32,49 @@ namespace tablet {
struct CompactionInputRow;
class WriteTransactionState;
+// Options related to tablet history garbage collection.
+class HistoryGcOpts {
+ public:
+ static HistoryGcOpts Enabled(Timestamp ahm) {
+ return HistoryGcOpts(true, ahm);
+ }
+
+ static HistoryGcOpts Disabled() {
+ return HistoryGcOpts(false, Timestamp(0));
+ }
+
+ // Returns true if Timestamp 't' is considered "ancient history" and is
+ // eligible for garbage collection. If GC is disabled, will return false for
+ // any Timestamp with a value >= 0 (that is, all valid Timestamps).
+ bool IsAncientHistory(Timestamp t) const {
+ return t.ComesBefore(ancient_history_mark_);
+ }
+
+ // Returns true if history GC is enabled.
+ bool gc_enabled() const {
+ return gc_enabled_;
+ }
+
+ // Returns the ancient history mark.
+ Timestamp ancient_history_mark() const {
+ return ancient_history_mark_;
+ }
+
+ private:
+ HistoryGcOpts(bool gc_enabled, Timestamp ahm)
+ : gc_enabled_(gc_enabled),
+ ancient_history_mark_(ahm) {
+ }
+
+ // Whether historical records prior to the ancient history mark should be
+ // garbage-collected (deleted).
+ const bool gc_enabled_;
+
+ // A timestamp prior to which no history will be preserved.
+ // Ignored if 'enabled' != GC_ENABLED.
+ const Timestamp ancient_history_mark_;
+};
+
// Interface for an input feeding into a compaction or flush.
class CompactionInput {
public:
@@ -110,6 +153,8 @@ class RowSetsInCompaction {
};
// One row yielded by CompactionInput::PrepareBlock.
+// Looks like this (assuming n UNDO records and m REDO records):
+// UNDO_n <- ... <- UNDO_1 <- UNDO_head <- row -> REDO_head -> REDO_1 -> ... -> REDO_m
struct CompactionInputRow {
// The compaction input base row.
RowBlockRow row;
@@ -119,6 +164,11 @@ struct CompactionInputRow {
Mutation* undo_head;
};
+// Function shared by flushes and compactions. Removes UNDO Mutations
+// considered "ancient" from the given CompactionInputRow, modifying the undo
+// mutation list in-place.
+void RemoveAncientUndos(const HistoryGcOpts& history_gc_opts, CompactionInputRow* row);
+
// Function shared by flushes, compactions and major delta compactions. Applies all the REDO
// mutations from 'src_row' to the 'dst_row', and generates the related UNDO mutations. Some
// handling depends on the nature of the operation being performed:
@@ -128,9 +178,20 @@ struct CompactionInputRow {
// belonging to 'dst_row'. Those that don't belong to that schema are
// ignored.
//
-// Currently, 'is_garbage_collected' is always false (KUDU-236).
+// 'history_gc_opts': Options indicating whether row history should be
+// garbage-collected.
+//
+// 'is_garbage_collected': Set to true if the row was marked as deleted prior
+// to the ancient history mark, with no reinsertions after that. In such a
+// case, all traces of the row should be removed from disk by the caller.
+//
+// 'num_rows_history_truncated': Set to the number of rows that had a "reinsert
+// after delete" that was compacted, resulting in the loss of history prior to
+// the reinsert. This is related to a known issue that will eventually be
+// fixed. See KUDU-237.
Status ApplyMutationsAndGenerateUndos(const MvccSnapshot& snap,
const CompactionInputRow& src_row,
+ const HistoryGcOpts& history_gc_opts,
const Schema* base_schema,
Mutation** new_undo_head,
Mutation** new_redo_head,
@@ -139,7 +200,6 @@ Status ApplyMutationsAndGenerateUndos(const MvccSnapshot& snap,
bool* is_garbage_collected,
uint64_t* num_rows_history_truncated);
-
// Iterate through this compaction input, flushing all rows to the given RollingDiskRowSetWriter.
// The 'snap' argument should match the MvccSnapshot used to create the compaction input.
//
@@ -147,6 +207,7 @@ Status ApplyMutationsAndGenerateUndos(const MvccSnapshot& snap,
// no longer be useful.
Status FlushCompactionInput(CompactionInput *input,
const MvccSnapshot &snap,
+ const HistoryGcOpts& history_gc_opts,
RollingDiskRowSetWriter *out);
// Iterate through this compaction input, finding any mutations which came between
@@ -161,6 +222,7 @@ Status FlushCompactionInput(CompactionInput *input,
// yield no further rows.
Status ReupdateMissedDeltas(const string &tablet_name,
CompactionInput *input,
+ const HistoryGcOpts& history_gc_opts,
const MvccSnapshot &snap_to_exclude,
const MvccSnapshot &snap_to_include,
const RowSetVector &output_rowsets);
http://git-wip-us.apache.org/repos/asf/kudu/blob/be719edc/src/kudu/tablet/delta_compaction.cc
----------------------------------------------------------------------
diff --git a/src/kudu/tablet/delta_compaction.cc b/src/kudu/tablet/delta_compaction.cc
index 62bca8e..59a6da2 100644
--- a/src/kudu/tablet/delta_compaction.cc
+++ b/src/kudu/tablet/delta_compaction.cc
@@ -61,17 +61,19 @@ MajorDeltaCompaction::MajorDeltaCompaction(
FsManager* fs_manager, const Schema& base_schema, CFileSet* base_data,
unique_ptr<DeltaIterator> delta_iter,
vector<shared_ptr<DeltaStore> > included_stores,
- const vector<ColumnId>& col_ids)
+ vector<ColumnId> col_ids,
+ HistoryGcOpts history_gc_opts)
: fs_manager_(fs_manager),
base_schema_(base_schema),
- column_ids_(col_ids),
+ column_ids_(std::move(col_ids)),
+ history_gc_opts_(std::move(history_gc_opts)),
base_data_(base_data),
included_stores_(std::move(included_stores)),
delta_iter_(std::move(delta_iter)),
redo_delta_mutations_written_(0),
undo_delta_mutations_written_(0),
state_(kInitialized) {
- CHECK(!col_ids.empty());
+ CHECK(!column_ids_.empty());
}
MajorDeltaCompaction::~MajorDeltaCompaction() {
@@ -132,24 +134,32 @@ Status MajorDeltaCompaction::FlushRowSetAndDeltas() {
vector<CompactionInputRow> input_rows;
input_rows.resize(block.nrows());
for (int i = 0; i < block.nrows(); i++) {
- CompactionInputRow &input_row = input_rows.at(i);
- input_row.row.Reset(&block, i);
- input_row.redo_head = redo_mutation_block[i];
- Mutation::ReverseMutationList(&input_row.redo_head);
- input_row.undo_head = nullptr;
+ CompactionInputRow* input_row = &input_rows[i];
+ input_row->row.Reset(&block, i);
+ input_row->redo_head = redo_mutation_block[i];
+ Mutation::ReverseMutationList(&input_row->redo_head);
+ input_row->undo_head = nullptr;
RowBlockRow dst_row = block.row(i);
- RETURN_NOT_OK(CopyRow(input_row.row, &dst_row, static_cast<Arena*>(nullptr)));
+ RETURN_NOT_OK(CopyRow(input_row->row, &dst_row, static_cast<Arena*>(nullptr)));
Mutation* new_undos_head = nullptr;
// We're ignoring the result from new_redos_head because we'll find them
// later at step 5.
Mutation* new_redos_head = nullptr;
+ DVLOG(3) << "MDC: Input Row: " << dst_row.schema()->DebugRow(dst_row)
+ << " RowIndex: " << input_row->row.row_index()
+ << " Undo Mutations: (not shown)"
+ << " Redo Mutations: "
+ << Mutation::StringifyMutationList(base_schema_, input_row->redo_head);
+
bool is_garbage_collected;
+ RemoveAncientUndos(history_gc_opts_, input_row);
RETURN_NOT_OK(ApplyMutationsAndGenerateUndos(snap,
- input_row,
+ *input_row,
+ history_gc_opts_,
&base_schema_,
&new_undos_head,
&new_redos_head,
@@ -158,9 +168,12 @@ Status MajorDeltaCompaction::FlushRowSetAndDeltas() {
&is_garbage_collected,
&num_rows_history_truncated));
- VLOG(2) << "Output Row: " << dst_row.schema()->DebugRow(dst_row)
- << " Undo Mutations: " << Mutation::StringifyMutationList(partial_schema_, new_undos_head)
- << " Redo Mutations: " << Mutation::StringifyMutationList(partial_schema_, new_redos_head);
+ DVLOG(3) << "MDC: Output Row: " << dst_row.schema()->DebugRow(dst_row)
+ << " RowIndex: " << input_row->row.row_index()
+ << " Generated Undo Mutations: "
+ << Mutation::StringifyMutationList(partial_schema_, new_undos_head)
+ << " Updated Redo Mutations: "
+ << Mutation::StringifyMutationList(partial_schema_, new_redos_head);
// We only create a new undo delta file if we need to.
if (new_undos_head != nullptr && !new_undo_delta_writer_) {
http://git-wip-us.apache.org/repos/asf/kudu/blob/be719edc/src/kudu/tablet/delta_compaction.h
----------------------------------------------------------------------
diff --git a/src/kudu/tablet/delta_compaction.h b/src/kudu/tablet/delta_compaction.h
index d797072..85760bb 100644
--- a/src/kudu/tablet/delta_compaction.h
+++ b/src/kudu/tablet/delta_compaction.h
@@ -24,6 +24,7 @@
#include <vector>
#include "kudu/cfile/cfile_writer.h"
+#include "kudu/tablet/compaction.h"
#include "kudu/tablet/deltafile.h"
namespace kudu {
@@ -56,7 +57,8 @@ class MajorDeltaCompaction {
FsManager* fs_manager, const Schema& base_schema, CFileSet* base_data,
std::unique_ptr<DeltaIterator> delta_iter,
std::vector<std::shared_ptr<DeltaStore> > included_stores,
- const std::vector<ColumnId>& col_ids);
+ std::vector<ColumnId> col_ids,
+ HistoryGcOpts history_gc_opts);
~MajorDeltaCompaction();
// Executes the compaction.
@@ -103,6 +105,8 @@ class MajorDeltaCompaction {
// The column ids to compact.
const std::vector<ColumnId> column_ids_;
+ const HistoryGcOpts history_gc_opts_;
+
// Inputs:
//-----------------
http://git-wip-us.apache.org/repos/asf/kudu/blob/be719edc/src/kudu/tablet/diskrowset.cc
----------------------------------------------------------------------
diff --git a/src/kudu/tablet/diskrowset.cc b/src/kudu/tablet/diskrowset.cc
index 5feeab4..e7fabf4 100644
--- a/src/kudu/tablet/diskrowset.cc
+++ b/src/kudu/tablet/diskrowset.cc
@@ -507,7 +507,7 @@ Status DiskRowSet::MinorCompactDeltaStores() {
return delta_tracker_->Compact();
}
-Status DiskRowSet::MajorCompactDeltaStores() {
+Status DiskRowSet::MajorCompactDeltaStores(HistoryGcOpts history_gc_opts) {
vector<ColumnId> col_ids;
delta_tracker_->GetColumnIdsWithUpdates(&col_ids);
@@ -515,16 +515,17 @@ Status DiskRowSet::MajorCompactDeltaStores() {
return Status::OK();
}
- return MajorCompactDeltaStoresWithColumnIds(col_ids);
+ return MajorCompactDeltaStoresWithColumnIds(col_ids, std::move(history_gc_opts));
}
-Status DiskRowSet::MajorCompactDeltaStoresWithColumnIds(const vector<ColumnId>& col_ids) {
+Status DiskRowSet::MajorCompactDeltaStoresWithColumnIds(const vector<ColumnId>& col_ids,
+ HistoryGcOpts history_gc_opts) {
TRACE_EVENT0("tablet", "DiskRowSet::MajorCompactDeltaStores");
std::lock_guard<Mutex> l(*delta_tracker()->compact_flush_lock());
// TODO: do we need to lock schema or anything here?
gscoped_ptr<MajorDeltaCompaction> compaction;
- RETURN_NOT_OK(NewMajorDeltaCompaction(col_ids, &compaction));
+ RETURN_NOT_OK(NewMajorDeltaCompaction(col_ids, std::move(history_gc_opts), &compaction));
RETURN_NOT_OK(compaction->Compact());
@@ -547,6 +548,7 @@ Status DiskRowSet::MajorCompactDeltaStoresWithColumnIds(const vector<ColumnId>&
}
Status DiskRowSet::NewMajorDeltaCompaction(const vector<ColumnId>& col_ids,
+ HistoryGcOpts history_gc_opts,
gscoped_ptr<MajorDeltaCompaction>* out) const {
DCHECK(open_);
shared_lock<rw_spinlock> l(component_lock_.get_lock());
@@ -566,8 +568,9 @@ Status DiskRowSet::NewMajorDeltaCompaction(const vector<ColumnId>& col_ids,
*schema,
base_data_.get(),
std::move(delta_iter),
- included_stores,
- col_ids));
+ std::move(included_stores),
+ col_ids,
+ std::move(history_gc_opts)));
return Status::OK();
}
http://git-wip-us.apache.org/repos/asf/kudu/blob/be719edc/src/kudu/tablet/diskrowset.h
----------------------------------------------------------------------
diff --git a/src/kudu/tablet/diskrowset.h b/src/kudu/tablet/diskrowset.h
index d638336..d096793 100644
--- a/src/kudu/tablet/diskrowset.h
+++ b/src/kudu/tablet/diskrowset.h
@@ -61,6 +61,7 @@ class CFileSet;
class DeltaFileWriter;
class DeltaStats;
class DeltaTracker;
+class HistoryGcOpts;
class MultiColumnWriter;
class Mutation;
class OperationResultPB;
@@ -346,7 +347,7 @@ class DiskRowSet : public RowSet {
double DeltaStoresCompactionPerfImprovementScore(DeltaCompactionType type) const OVERRIDE;
// Major compacts all the delta files for all the columns.
- Status MajorCompactDeltaStores();
+ Status MajorCompactDeltaStores(HistoryGcOpts history_gc_opts);
std::mutex *compact_flush_lock() OVERRIDE {
return &compact_flush_lock_;
@@ -370,6 +371,7 @@ class DiskRowSet : public RowSet {
FRIEND_TEST(TestRowSet, TestRowSetUpdate);
FRIEND_TEST(TestRowSet, TestDMSFlush);
FRIEND_TEST(TestCompaction, TestOneToOne);
+ FRIEND_TEST(TabletHistoryGcTest, TestMajorDeltaCompactionOnSubsetOfColumns);
friend class CompactionInput;
friend class Tablet;
@@ -382,10 +384,12 @@ class DiskRowSet : public RowSet {
// Create a new major delta compaction object to compact the specified columns.
Status NewMajorDeltaCompaction(const std::vector<ColumnId>& col_ids,
+ HistoryGcOpts history_gc_opts,
gscoped_ptr<MajorDeltaCompaction>* out) const;
// Major compacts all the delta files for the specified columns.
- Status MajorCompactDeltaStoresWithColumnIds(const std::vector<ColumnId>& col_ids);
+ Status MajorCompactDeltaStoresWithColumnIds(const std::vector<ColumnId>& col_ids,
+ HistoryGcOpts history_gc_opts);
std::shared_ptr<RowSetMetadata> rowset_metadata_;
http://git-wip-us.apache.org/repos/asf/kudu/blob/be719edc/src/kudu/tablet/mutation.h
----------------------------------------------------------------------
diff --git a/src/kudu/tablet/mutation.h b/src/kudu/tablet/mutation.h
index 1eb73ca..bb3e464 100644
--- a/src/kudu/tablet/mutation.h
+++ b/src/kudu/tablet/mutation.h
@@ -51,6 +51,8 @@ class Mutation {
}
Timestamp timestamp() const { return timestamp_; }
+
+ Mutation *next() { return next_; }
const Mutation *next() const { return next_; }
// Same as 'next()' but loads with 'Acquire' ordering semantics.
@@ -91,8 +93,6 @@ class Mutation {
template<bool ATOMIC>
void DoAppendToList(Mutation **list);
- DISALLOW_COPY_AND_ASSIGN(Mutation);
-
// The transaction ID which made this mutation. If this transaction is not
// committed in the snapshot of the reader, this mutation should be ignored.
Timestamp timestamp_;
@@ -104,6 +104,8 @@ class Mutation {
// The actual encoded RowChangeList
char changelist_data_[0];
+
+ DISALLOW_COPY_AND_ASSIGN(Mutation);
};
template<class ArenaType>
http://git-wip-us.apache.org/repos/asf/kudu/blob/be719edc/src/kudu/tablet/rowset_metadata.h
----------------------------------------------------------------------
diff --git a/src/kudu/tablet/rowset_metadata.h b/src/kudu/tablet/rowset_metadata.h
index 14ed190..c195d8f 100644
--- a/src/kudu/tablet/rowset_metadata.h
+++ b/src/kudu/tablet/rowset_metadata.h
@@ -43,7 +43,7 @@ namespace tablet {
class RowSetMetadataUpdate;
class TabletMetadata;
-// Keeps tracks of the RowSet data blocks.
+// Keeps track of the RowSet data blocks.
//
// On each tablet MemRowSet flush, a new RowSetMetadata is created,
// and the DiskRowSetWriter will create and write the "immutable" blocks for
http://git-wip-us.apache.org/repos/asf/kudu/blob/be719edc/src/kudu/tablet/tablet-harness.h
----------------------------------------------------------------------
diff --git a/src/kudu/tablet/tablet-harness.h b/src/kudu/tablet/tablet-harness.h
index f8f376c..fd09c5a 100644
--- a/src/kudu/tablet/tablet-harness.h
+++ b/src/kudu/tablet/tablet-harness.h
@@ -25,6 +25,7 @@
#include "kudu/common/schema.h"
#include "kudu/consensus/log_anchor_registry.h"
#include "kudu/consensus/metadata.pb.h"
+#include "kudu/server/hybrid_clock.h"
#include "kudu/server/logical_clock.h"
#include "kudu/tablet/tablet.h"
#include "kudu/util/env.h"
@@ -60,16 +61,22 @@ static std::pair<PartitionSchema, Partition> CreateDefaultPartition(const Schema
class TabletHarness {
public:
struct Options {
+ enum ClockType {
+ HYBRID_CLOCK,
+ LOGICAL_CLOCK
+ };
explicit Options(string root_dir)
: env(Env::Default()),
tablet_id("test_tablet_id"),
root_dir(std::move(root_dir)),
- enable_metrics(true) {}
+ enable_metrics(true),
+ clock_type(LOGICAL_CLOCK) {}
Env* env;
string tablet_id;
string root_dir;
bool enable_metrics;
+ ClockType clock_type;
};
TabletHarness(const Schema& schema, Options options)
@@ -99,7 +106,12 @@ class TabletHarness {
metrics_registry_.reset(new MetricRegistry());
}
- clock_ = server::LogicalClock::CreateStartingAt(Timestamp::kInitialTimestamp);
+ if (options_.clock_type == Options::LOGICAL_CLOCK) {
+ clock_.reset(server::LogicalClock::CreateStartingAt(Timestamp::kInitialTimestamp));
+ } else {
+ clock_.reset(new server::HybridClock());
+ RETURN_NOT_OK(clock_->Init());
+ }
tablet_.reset(new Tablet(metadata,
clock_,
std::shared_ptr<MemTracker>(),
http://git-wip-us.apache.org/repos/asf/kudu/blob/be719edc/src/kudu/tablet/tablet-test-base.h
----------------------------------------------------------------------
diff --git a/src/kudu/tablet/tablet-test-base.h b/src/kudu/tablet/tablet-test-base.h
index 78d2916..a00c62b 100644
--- a/src/kudu/tablet/tablet-test-base.h
+++ b/src/kudu/tablet/tablet-test-base.h
@@ -18,6 +18,7 @@
#define KUDU_TABLET_TABLET_TEST_BASE_H
#include <algorithm>
+#include <boost/optional.hpp>
#include <glog/logging.h>
#include <gtest/gtest.h>
#include <limits>
@@ -304,8 +305,11 @@ typedef ::testing::Types<
template<class TESTSETUP>
class TabletTestBase : public KuduTabletTest {
public:
- TabletTestBase() :
- KuduTabletTest(TESTSETUP::CreateSchema()),
+ typedef std::function<bool(int32_t, int32_t)> TestRowVerifier;
+
+ TabletTestBase(TabletHarness::Options::ClockType clock_type =
+ TabletHarness::Options::ClockType::LOGICAL_CLOCK) :
+ KuduTabletTest(TESTSETUP::CreateSchema(), clock_type),
setup_(),
max_rows_(setup_.GetMaxRows()),
arena_(1024, 4*1024*1024)
@@ -398,59 +402,90 @@ class TabletTestBase : public KuduTabletTest {
}
template <class RowType>
- void VerifyRow(const RowType& row, int64_t key_idx, int32_t val) {
+ void VerifyRow(const RowType& row, int32_t key_idx, int32_t val) {
ASSERT_EQ(setup_.FormatDebugRow(key_idx, val, false), schema_.DebugRow(row));
}
- void VerifyTestRows(int64_t first_row, uint64_t expected_count) {
+ void VerifyTestRows(int32_t first_row, uint64_t expected_count) {
+ VerifyTestRowsWithVerifier(first_row, expected_count, boost::none);
+ }
+
+ void VerifyTestRowsWithVerifier(int32_t first_row, uint64_t expected_count,
+ boost::optional<TestRowVerifier> verifier) {
gscoped_ptr<RowwiseIterator> iter;
ASSERT_OK(tablet()->NewRowIterator(client_schema_, &iter));
+ VerifyTestRowsWithRowIteratorAndVerifier(first_row, expected_count, std::move(iter), verifier);
+ }
+
+ void VerifyTestRowsWithTimestampAndVerifier(int32_t first_row, uint64_t expected_count,
+ Timestamp timestamp,
+ boost::optional<TestRowVerifier> verifier) {
+ gscoped_ptr<RowwiseIterator> iter;
+ ASSERT_OK(tablet()->NewRowIterator(client_schema_, MvccSnapshot(timestamp), Tablet::UNORDERED,
+ &iter));
+ VerifyTestRowsWithRowIteratorAndVerifier(first_row, expected_count, std::move(iter), verifier);
+ }
+
+ void VerifyTestRowsWithRowIteratorAndVerifier(int32_t first_row, uint64_t expected_row_count,
+ gscoped_ptr<RowwiseIterator> iter,
+ boost::optional<TestRowVerifier> verifier) {
ASSERT_OK(iter->Init(NULL));
- int batch_size = std::max(
- (size_t)1, std::min((size_t)(expected_count / 10),
- 4*1024*1024 / schema_.byte_size()));
+ int batch_size = std::max<size_t>(1, std::min<size_t>(expected_row_count / 10,
+ 4L * 1024 * 1024 / schema_.byte_size()));
Arena arena(32*1024, 256*1024);
RowBlock block(schema_, batch_size, &arena);
- if (expected_count > INT_MAX) {
- LOG(INFO) << "Not checking rows for duplicates -- duplicates expected since "
- << "there were more than " << INT_MAX << " rows inserted.";
- return;
+ bool check_for_dups = true;
+ if (expected_row_count > INT_MAX) {
+ check_for_dups = false;
+ LOG(WARNING) << "Not checking rows for duplicates -- duplicates expected since "
+ << "there were more than " << INT_MAX << " rows inserted.";
}
// Keep a bitmap of which rows have been seen from the requested
// range.
std::vector<bool> seen_rows;
- seen_rows.resize(expected_count);
+ seen_rows.resize(expected_row_count);
+ uint64_t actual_row_count = 0;
while (iter->HasNext()) {
ASSERT_OK_FAST(iter->NextBlock(&block));
RowBlockRow rb_row = block.row(0);
if (VLOG_IS_ON(2)) {
VLOG(2) << "Fetched batch of " << block.nrows() << "\n"
- << "First row: " << schema_.DebugRow(rb_row);
+ << "First row: " << schema_.DebugRow(rb_row);
}
for (int i = 0; i < block.nrows(); i++) {
rb_row.Reset(&block, i);
int32_t key_idx = *schema_.ExtractColumnFromRow<INT32>(rb_row, 1);
- if (key_idx >= first_row && key_idx < first_row + expected_count) {
- size_t rel_idx = key_idx - first_row;
- if (seen_rows[rel_idx]) {
- FAIL() << "Saw row " << key_idx << " twice!\n"
- << "Row: " << schema_.DebugRow(rb_row);
+ if (key_idx >= first_row && block.selection_vector()->IsRowSelected(i)) {
+ actual_row_count++;
+ if (key_idx < first_row + expected_row_count) {
+ size_t rel_idx = key_idx - first_row;
+ if (check_for_dups && seen_rows[rel_idx]) {
+ FAIL() << "Saw row " << key_idx << " twice!\n"
+ << "Row: " << schema_.DebugRow(rb_row);
+ }
+ seen_rows[rel_idx] = true;
+ if (verifier) {
+ int32_t val = *schema_.ExtractColumnFromRow<INT32>(rb_row, 2);
+ ASSERT_TRUE((*verifier)(key_idx, val))
+ << "Key index: " << key_idx << ", value: " << val;
+ }
}
- seen_rows[rel_idx] = true;
}
}
}
// Verify that all the rows were seen.
- for (int i = 0; i < expected_count; i++) {
- ASSERT_EQ(true, seen_rows[i]) << "Never saw row: " << (i + first_row);
+ for (int i = 0; i < expected_row_count; i++) {
+ ASSERT_EQ(true, seen_rows[i]) << "Never saw row " << (i + first_row);
}
- LOG(INFO) << "Successfully verified " << expected_count << "rows";
+ ASSERT_EQ(expected_row_count, actual_row_count)
+ << "Expected row count didn't match actual row count";
+ LOG(INFO) << "Successfully verified " << expected_row_count << "rows";
}
// Iterate through the full table, stringifying the resulting rows
http://git-wip-us.apache.org/repos/asf/kudu/blob/be719edc/src/kudu/tablet/tablet-test-util.h
----------------------------------------------------------------------
diff --git a/src/kudu/tablet/tablet-test-util.h b/src/kudu/tablet/tablet-test-util.h
index 7cc8516..977987f 100644
--- a/src/kudu/tablet/tablet-test-util.h
+++ b/src/kudu/tablet/tablet-test-util.h
@@ -45,9 +45,12 @@ using std::vector;
class KuduTabletTest : public KuduTest {
public:
- explicit KuduTabletTest(const Schema& schema)
+ explicit KuduTabletTest(const Schema& schema,
+ TabletHarness::Options::ClockType clock_type =
+ TabletHarness::Options::ClockType::LOGICAL_CLOCK)
: schema_(schema.CopyWithColumnIds()),
- client_schema_(schema) {
+ client_schema_(schema),
+ clock_type_(clock_type) {
// Keep unit tests fast, but only if no one has set the flag explicitly.
if (google::GetCommandLineFlagInfoOrDie("enable_data_block_fsync").is_default) {
FLAGS_enable_data_block_fsync = false;
@@ -64,6 +67,7 @@ class KuduTabletTest : public KuduTest {
string dir = root_dir.empty() ? GetTestPath("fs_root") : root_dir;
TabletHarness::Options opts(dir);
opts.enable_metrics = true;
+ opts.clock_type = clock_type_;
bool first_time = harness_ == NULL;
harness_.reset(new TabletHarness(schema_, opts));
CHECK_OK(harness_->Create(first_time));
@@ -115,6 +119,7 @@ class KuduTabletTest : public KuduTest {
protected:
const Schema schema_;
const Schema client_schema_;
+ const TabletHarness::Options::ClockType clock_type_;
gscoped_ptr<TabletHarness> harness_;
};
http://git-wip-us.apache.org/repos/asf/kudu/blob/be719edc/src/kudu/tablet/tablet-test.cc
----------------------------------------------------------------------
diff --git a/src/kudu/tablet/tablet-test.cc b/src/kudu/tablet/tablet-test.cc
index a8f97bc..f704f0d 100644
--- a/src/kudu/tablet/tablet-test.cc
+++ b/src/kudu/tablet/tablet-test.cc
@@ -553,17 +553,22 @@ TYPED_TEST(TestTablet, TestInsertsPersist) {
this->InsertTestRows(0, max_rows, 0);
ASSERT_EQ(max_rows, this->TabletCount());
+ // Get current timestamp.
+ Timestamp t = this->tablet()->clock()->Now();
+
// Flush it.
ASSERT_OK(this->tablet()->Flush());
ASSERT_EQ(max_rows, this->TabletCount());
- // Close and re-open tablet
+ // Close and re-open tablet.
+ // TODO: Should we be reopening the tablet in a different way to persist the
+ // clock / timestamps?
this->TabletReOpen();
// Ensure that rows exist
ASSERT_EQ(max_rows, this->TabletCount());
- this->VerifyTestRows(0, max_rows);
+ this->VerifyTestRowsWithTimestampAndVerifier(0, max_rows, t, boost::none);
// TODO: add some more data, re-flush
}
@@ -645,8 +650,6 @@ TYPED_TEST(TestTablet, TestMultipleUpdates) {
ASSERT_EQ(this->setup_.FormatDebugRow(1, 0, false), out_rows[1]);
}
-
-
TYPED_TEST(TestTablet, TestCompaction) {
uint64_t max_rows = this->ClampRowCount(FLAGS_testcompaction_num_rows);
http://git-wip-us.apache.org/repos/asf/kudu/blob/be719edc/src/kudu/tablet/tablet.cc
----------------------------------------------------------------------
diff --git a/src/kudu/tablet/tablet.cc b/src/kudu/tablet/tablet.cc
index faf1ae1..d223c1f 100644
--- a/src/kudu/tablet/tablet.cc
+++ b/src/kudu/tablet/tablet.cc
@@ -41,6 +41,7 @@
#include "kudu/gutil/stl_util.h"
#include "kudu/gutil/strings/numbers.h"
#include "kudu/gutil/strings/substitute.h"
+#include "kudu/server/hybrid_clock.h"
#include "kudu/tablet/compaction.h"
#include "kudu/tablet/compaction_policy.h"
#include "kudu/tablet/delta_compaction.h"
@@ -91,7 +92,6 @@ DEFINE_double(fault_crash_before_flush_tablet_meta_after_flush_mrs, 0.0,
"Fraction of the time, while flushing an MRS, to crash before flushing metadata");
TAG_FLAG(fault_crash_before_flush_tablet_meta_after_flush_mrs, unsafe);
-
DEFINE_int64(tablet_throttler_rpc_per_sec, 0,
"Maximum write RPC rate (op/s) allowed for a tablet, write RPC exceeding this "
"limit will be throttled. 0 means no limit.");
@@ -108,6 +108,12 @@ DEFINE_double(tablet_throttler_burst_factor, 1.0f,
"base rate.");
TAG_FLAG(tablet_throttler_burst_factor, experimental);
+DEFINE_int32(tablet_history_max_age_sec, -1,
+ "Number of seconds to retain tablet history. Reads initiated at a "
+ "snapshot that is older than this age will be rejected. "
+ "To disable history removal, set to -1.");
+TAG_FLAG(tablet_history_max_age_sec, advanced);
+
METRIC_DEFINE_entity(tablet);
METRIC_DEFINE_gauge_size(tablet, memrowset_size, "MemRowSet Memory Usage",
kudu::MetricUnit::kBytes,
@@ -116,21 +122,22 @@ METRIC_DEFINE_gauge_size(tablet, on_disk_size, "Tablet Size On Disk",
kudu::MetricUnit::kBytes,
"Size of this tablet on disk.");
+using base::subtle::Barrier_AtomicIncrement;
+using kudu::MaintenanceManager;
+using kudu::consensus::OpId;
+using kudu::consensus::MaximumOpId;
+using kudu::log::LogAnchorRegistry;
+using kudu::server::HybridClock;
using std::shared_ptr;
using std::string;
+using std::unique_ptr;
using std::unordered_set;
using std::vector;
+using strings::Substitute;
namespace kudu {
namespace tablet {
-using kudu::MaintenanceManager;
-using consensus::OpId;
-using consensus::MaximumOpId;
-using log::LogAnchorRegistry;
-using strings::Substitute;
-using base::subtle::Barrier_AtomicIncrement;
-
static CompactionPolicy *CreateCompactionPolicy() {
return new BudgetedCompactionPolicy(FLAGS_tablet_compaction_budget_mb);
}
@@ -683,10 +690,40 @@ Status Tablet::DoMajorDeltaCompaction(const vector<ColumnId>& col_ids,
shared_ptr<RowSet> input_rs) {
CHECK_EQ(state_, kOpen);
Status s = down_cast<DiskRowSet*>(input_rs.get())
- ->MajorCompactDeltaStoresWithColumnIds(col_ids);
+ ->MajorCompactDeltaStoresWithColumnIds(col_ids, GetHistoryGcOpts());
return s;
}
+bool Tablet::GetTabletAncientHistoryMark(Timestamp* ancient_history_mark) const {
+ // We currently only support history GC through a fully-instantiated tablet
+ // when using the HybridClock, since we can calculate the age of a mutation.
+ if (!clock_->HasPhysicalComponent() || FLAGS_tablet_history_max_age_sec < 0) {
+ return false;
+ }
+ Timestamp now = clock_->Now();
+ uint64_t now_micros = HybridClock::GetPhysicalValueMicros(now);
+ uint64_t max_age_micros = FLAGS_tablet_history_max_age_sec * 1000000ULL;
+ // Ensure that the AHM calculation doesn't underflow when
+ // '--tablet_history_max_age_sec' is set to a very high value.
+ if (max_age_micros <= now_micros) {
+ *ancient_history_mark =
+ HybridClock::TimestampFromMicrosecondsAndLogicalValue(
+ now_micros - max_age_micros,
+ HybridClock::GetLogicalValue(now));
+ } else {
+ *ancient_history_mark = Timestamp(0);
+ }
+ return true;
+}
+
+HistoryGcOpts Tablet::GetHistoryGcOpts() const {
+ Timestamp ancient_history_mark;
+ if (GetTabletAncientHistoryMark(&ancient_history_mark)) {
+ return HistoryGcOpts::Enabled(ancient_history_mark);
+ }
+ return HistoryGcOpts::Disabled();
+}
+
Status Tablet::Flush() {
TRACE_EVENT1("tablet", "Tablet::Flush", "id", tablet_id());
std::lock_guard<Semaphore> lock(rowsets_flush_sem_);
@@ -1255,7 +1292,9 @@ Status Tablet::DoCompactionOrFlush(const RowSetsInCompaction &input, int64_t mrs
RollingDiskRowSetWriter drsw(metadata_.get(), merge->schema(), bloom_sizing(),
compaction_policy_->target_rowset_size());
RETURN_NOT_OK_PREPEND(drsw.Open(), "Failed to open DiskRowSet for flush");
- RETURN_NOT_OK_PREPEND(FlushCompactionInput(merge.get(), flush_snap, &drsw),
+
+ HistoryGcOpts history_gc_opts = GetHistoryGcOpts();
+ RETURN_NOT_OK_PREPEND(FlushCompactionInput(merge.get(), flush_snap, history_gc_opts, &drsw),
"Flush to disk failed");
RETURN_NOT_OK_PREPEND(drsw.Finish(), "Failed to finish DRS writer");
@@ -1389,6 +1428,7 @@ Status Tablet::DoCompactionOrFlush(const RowSetsInCompaction &input, int64_t mrs
// rowsets is flushed.
RETURN_NOT_OK_PREPEND(ReupdateMissedDeltas(metadata_->tablet_id(),
merge.get(),
+ history_gc_opts,
flush_snap,
non_duplicated_txns_snap,
new_disk_rowsets),
@@ -1449,10 +1489,6 @@ Status Tablet::Compact(CompactFlags flags) {
// Step 1. Capture the rowsets to be merged
RETURN_NOT_OK_PREPEND(PickRowSetsToCompact(&input, flags),
"Failed to pick rowsets to compact");
- if (input.num_rowsets() < 2) {
- VLOG_WITH_PREFIX(1) << "Not enough rowsets to run compaction! Aborting...";
- return Status::OK();
- }
LOG_WITH_PREFIX(INFO) << "Compaction: stage 1 complete, picked "
<< input.num_rowsets() << " rowsets to compact";
if (compaction_hooks_) {
@@ -1725,6 +1761,7 @@ Status Tablet::FlushBiggestDMS() {
Status Tablet::CompactWorstDeltas(RowSet::DeltaCompactionType type) {
CHECK_EQ(state_, kOpen);
shared_ptr<RowSet> rs;
+
// We're required to grab the rowset's compact_flush_lock under the compact_select_lock_.
std::unique_lock<std::mutex> lock;
double perf_improv;
@@ -1734,12 +1771,11 @@ Status Tablet::CompactWorstDeltas(RowSet::DeltaCompactionType type) {
// under this lock.
std::lock_guard<std::mutex> compact_lock(compact_select_lock_);
perf_improv = GetPerfImprovementForBestDeltaCompactUnlocked(type, &rs);
- if (rs) {
- lock = std::unique_lock<std::mutex>(*rs->compact_flush_lock(), std::try_to_lock);
- CHECK(lock.owns_lock());
- } else {
+ if (!rs) {
return Status::OK();
}
+ lock = std::unique_lock<std::mutex>(*rs->compact_flush_lock(), std::try_to_lock);
+ CHECK(lock.owns_lock());
}
// We just released compact_select_lock_ so other compactions can select and run, but the
@@ -1749,14 +1785,14 @@ Status Tablet::CompactWorstDeltas(RowSet::DeltaCompactionType type) {
RETURN_NOT_OK_PREPEND(rs->MinorCompactDeltaStores(),
"Failed minor delta compaction on " + rs->ToString());
} else if (type == RowSet::MAJOR_DELTA_COMPACTION) {
- RETURN_NOT_OK_PREPEND(down_cast<DiskRowSet*>(rs.get())->MajorCompactDeltaStores(),
- "Failed major delta compaction on " + rs->ToString());
+ RETURN_NOT_OK_PREPEND(down_cast<DiskRowSet*>(rs.get())->MajorCompactDeltaStores(
+ GetHistoryGcOpts()), "Failed major delta compaction on " + rs->ToString());
}
return Status::OK();
}
double Tablet::GetPerfImprovementForBestDeltaCompact(RowSet::DeltaCompactionType type,
- shared_ptr<RowSet>* rs) const {
+ shared_ptr<RowSet>* rs) const {
std::lock_guard<std::mutex> compact_lock(compact_select_lock_);
return GetPerfImprovementForBestDeltaCompactUnlocked(type, rs);
}
http://git-wip-us.apache.org/repos/asf/kudu/blob/be719edc/src/kudu/tablet/tablet.h
----------------------------------------------------------------------
diff --git a/src/kudu/tablet/tablet.h b/src/kudu/tablet/tablet.h
index 7663cde..f0b8b31 100644
--- a/src/kudu/tablet/tablet.h
+++ b/src/kudu/tablet/tablet.h
@@ -64,6 +64,7 @@ namespace tablet {
class AlterSchemaTransactionState;
class CompactionPolicy;
+class HistoryGcOpts;
class MemRowSet;
class MvccSnapshot;
struct RowOp;
@@ -346,6 +347,14 @@ class Tablet {
Status DoMajorDeltaCompaction(const std::vector<ColumnId>& column_ids,
std::shared_ptr<RowSet> input_rowset);
+ // Calculates the ancient history mark and returns true iff tablet history GC
+ // is enabled, which requires the use of a HybridClock.
+ // Otherwise, returns false.
+ bool GetTabletAncientHistoryMark(Timestamp* ancient_history_mark) const WARN_UNUSED_RESULT;
+
+ // Calculates history GC options based on properties of the Clock implementation.
+ HistoryGcOpts GetHistoryGcOpts() const;
+
// Method used by tests to retrieve all rowsets of this table. This
// will be removed once code for selecting the appropriate RowSet is
// finished and delta files is finished is part of Tablet class.
@@ -374,6 +383,8 @@ class Tablet {
// Return true if this RPC is allowed.
bool ShouldThrottleAllow(int64_t bytes);
+ scoped_refptr<server::Clock> clock() const { return clock_; }
+
static const char* kDMSMemTrackerId;
private:
friend class Iterator;