Skip to content

feat(realtime): support compaction for deduplicate tables #449

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
lxy-9602 wants to merge 1 commit into
base: main
Choose a base branch
from
Open

feat(realtime): support compaction for deduplicate tables #449

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 4 additions & 0 deletions src/paimon/CMakeLists.txt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -374,6 +374,9 @@ set(PAIMON_CORE_SRCS
core/operation/commit/conflict_detection.cpp
core/operation/commit/commit_scanner.cpp
core/operation/commit/commit_changes_provider.cpp
core/operation/commit/realtime_deduplicate_commit_changes_provider.cpp
core/operation/commit/realtime_deduplicate_compaction_changes_provider.cpp
core/operation/commit/realtime_deduplicate_dv_materializer.cpp
core/operation/commit/compacted_changelog_path_resolver.cpp
core/operation/commit/overwrite_changes_provider.cpp
core/operation/commit/realtime_commit_properties.cpp
Expand Down Expand Up @@ -955,6 +958,7 @@ if(PAIMON_BUILD_TESTS)
core/operation/commit/row_id_column_conflict_checker_test.cpp
core/operation/commit/row_tracking_commit_utils_test.cpp
core/operation/commit/realtime_commit_properties_test.cpp
core/operation/commit/realtime_deduplicate_dv_materializer_test.cpp
core/operation/commit/sequence_snapshot_properties_test.cpp
core/operation/key_value_file_store_write_test.cpp
core/operation/internal_read_context_test.cpp
Expand Down
31 changes: 18 additions & 13 deletions src/paimon/core/append/bucketed_append_compact_manager.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,7 +30,8 @@ BucketedAppendCompactManager::BucketedAppendCompactManager(
const std::shared_ptr<BucketedDvMaintainer>& dv_maintainer, int32_t min_file_num,
int64_t target_file_size, int64_t compaction_file_size, bool force_rewrite_all_files,
CompactRewriter rewriter, const std::shared_ptr<CompactionMetrics::Reporter>& reporter,
const std::shared_ptr<CancellationController>& cancellation_controller)
const std::shared_ptr<CancellationController>& cancellation_controller,
bool materialize_dv_on_commit)
: executor_(executor),
dv_maintainer_(dv_maintainer),
min_file_num_(min_file_num),
Expand All @@ -43,7 +44,8 @@ BucketedAppendCompactManager::BucketedAppendCompactManager(
[](const std::shared_ptr<DataFileMeta>& lhs, const std::shared_ptr<DataFileMeta>& rhs) {
return lhs->min_sequence_number > rhs->min_sequence_number;
}),
cancellation_controller_(cancellation_controller) {
cancellation_controller_(cancellation_controller),
materialize_dv_on_commit_(materialize_dv_on_commit) {
assert(cancellation_controller_ != nullptr);
for (const auto& file : restored) {
to_compact_.push(file);
Expand Down Expand Up @@ -82,9 +84,9 @@ Status BucketedAppendCompactManager::TriggerFullCompaction() {
to_compact_.pop();
}
cancellation_controller_->Reset();
auto compact_task = std::make_shared<FullCompactTask>(reporter_, dv_maintainer_, compacting,
compaction_file_size_,
force_rewrite_all_files_, rewriter_);
auto compact_task = std::make_shared<FullCompactTask>(
reporter_, dv_maintainer_, compacting, compaction_file_size_, force_rewrite_all_files_,
rewriter_, materialize_dv_on_commit_);
task_future_ = Via(executor_.get(), [compact_task]() -> Result<std::shared_ptr<CompactResult>> {
return compact_task->Execute();
});
Expand All @@ -100,8 +102,8 @@ void BucketedAppendCompactManager::TriggerCompactionWithBestEffort() {
if (picked) {
cancellation_controller_->Reset();
compacting_ = picked.value();
auto compact_task = std::make_shared<AutoCompactTask>(reporter_, dv_maintainer_,
compacting_.value(), rewriter_);
auto compact_task = std::make_shared<AutoCompactTask>(
reporter_, dv_maintainer_, compacting_.value(), rewriter_, materialize_dv_on_commit_);
task_future_ =
Via(executor_.get(), [compact_task]() -> Result<std::shared_ptr<CompactResult>> {
return compact_task->Execute();
Expand Down Expand Up @@ -175,7 +177,8 @@ BucketedAppendCompactManager::GetCompactionResult(bool blocking) {

Result<std::shared_ptr<CompactResult>> BucketedAppendCompactManager::Compact(
const std::shared_ptr<BucketedDvMaintainer>& dv_maintainer,
const std::vector<std::shared_ptr<DataFileMeta>>& to_compact, CompactRewriter rewriter) {
const std::vector<std::shared_ptr<DataFileMeta>>& to_compact, CompactRewriter rewriter,
bool materialize_dv_on_commit) {
PAIMON_ASSIGN_OR_RAISE(std::vector<std::shared_ptr<DataFileMeta>> rewrite,
rewriter(to_compact));

Expand All @@ -184,9 +187,11 @@ Result<std::shared_ptr<CompactResult>> BucketedAppendCompactManager::Compact(
for (const auto& file : to_compact) {
dv_maintainer->RemoveDeletionVectorOf(file->file_name);
}
PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<CompactDeletionFile> deletion_file,
CompactDeletionFile::GenerateFiles(dv_maintainer));
result->SetDeletionFile(deletion_file);
if (!materialize_dv_on_commit) {
PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<CompactDeletionFile> deletion_file,
CompactDeletionFile::GenerateFiles(dv_maintainer));
result->SetDeletionFile(deletion_file);
}
}
return result;
}
Expand All @@ -209,7 +214,7 @@ Result<std::shared_ptr<CompactResult>> BucketedAppendCompactManager::FullCompact
return Compact(
dv_maintainer_,
std::vector<std::shared_ptr<DataFileMeta>>(to_compact_.begin(), to_compact_.end()),
rewriter_);
rewriter_, materialize_dv_on_commit_);
} else {
// compute small files
int32_t big = 0;
Expand All @@ -226,7 +231,7 @@ Result<std::shared_ptr<CompactResult>> BucketedAppendCompactManager::FullCompact
return Compact(
/*dv_maintainer=*/nullptr,
std::vector<std::shared_ptr<DataFileMeta>>(to_compact_.begin(), to_compact_.end()),
rewriter_);
rewriter_, materialize_dv_on_commit_);
} else {
return std::make_shared<CompactResult>();
}
Expand Down
22 changes: 15 additions & 7 deletions src/paimon/core/append/bucketed_append_compact_manager.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -81,7 +81,8 @@ class BucketedAppendCompactManager : public CompactFutureManager {
const std::shared_ptr<BucketedDvMaintainer>& dv_maintainer, int32_t min_file_num,
int64_t target_file_size, int64_t compaction_file_size, bool force_rewrite_all_files,
CompactRewriter rewriter, const std::shared_ptr<CompactionMetrics::Reporter>& reporter,
const std::shared_ptr<CancellationController>& cancellation_controller);
const std::shared_ptr<CancellationController>& cancellation_controller,
bool materialize_dv_on_commit);
~BucketedAppendCompactManager() override = default;

void RequestCancelCompaction() override;
Expand Down Expand Up @@ -125,7 +126,8 @@ class BucketedAppendCompactManager : public CompactFutureManager {

static Result<std::shared_ptr<CompactResult>> Compact(
const std::shared_ptr<BucketedDvMaintainer>& dv_maintainer,
const std::vector<std::shared_ptr<DataFileMeta>>& to_compact, CompactRewriter rewriter);
const std::vector<std::shared_ptr<DataFileMeta>>& to_compact, CompactRewriter rewriter,
bool materialize_dv_on_commit);

/// A `CompactTask` impl for full compaction of append-only table.
class FullCompactTask : public CompactTask {
Expand All @@ -134,13 +136,14 @@ class BucketedAppendCompactManager : public CompactFutureManager {
const std::shared_ptr<BucketedDvMaintainer>& dv_maintainer,
const std::vector<std::shared_ptr<DataFileMeta>>& inputs,
int64_t compaction_file_size, bool force_rewrite_all_files,
CompactRewriter rewriter)
CompactRewriter rewriter, bool materialize_dv_on_commit)
: CompactTask(reporter),
dv_maintainer_(dv_maintainer),
to_compact_(inputs.begin(), inputs.end()),
compaction_file_size_(compaction_file_size),
force_rewrite_all_files_(force_rewrite_all_files),
rewriter_(rewriter) {}
rewriter_(rewriter),
materialize_dv_on_commit_(materialize_dv_on_commit) {}

protected:
Result<std::shared_ptr<CompactResult>> DoCompact() override;
Expand All @@ -160,6 +163,7 @@ class BucketedAppendCompactManager : public CompactFutureManager {
int64_t compaction_file_size_;
bool force_rewrite_all_files_;
CompactRewriter rewriter_;
bool materialize_dv_on_commit_;
};

/// A `CompactTask` impl for append-only table auto-compaction.
Expand All @@ -172,21 +176,23 @@ class BucketedAppendCompactManager : public CompactFutureManager {
AutoCompactTask(const std::shared_ptr<CompactionMetrics::Reporter>& reporter,
const std::shared_ptr<BucketedDvMaintainer>& dv_maintainer,
const std::vector<std::shared_ptr<DataFileMeta>>& to_compact,
CompactRewriter rewriter)
CompactRewriter rewriter, bool materialize_dv_on_commit)
: CompactTask(reporter),
dv_maintainer_(dv_maintainer),
to_compact_(to_compact),
rewriter_(rewriter) {}
rewriter_(rewriter),
materialize_dv_on_commit_(materialize_dv_on_commit) {}

protected:
Result<std::shared_ptr<CompactResult>> DoCompact() override {
return Compact(dv_maintainer_, to_compact_, rewriter_);
return Compact(dv_maintainer_, to_compact_, rewriter_, materialize_dv_on_commit_);
}

private:
std::shared_ptr<BucketedDvMaintainer> dv_maintainer_;
std::vector<std::shared_ptr<DataFileMeta>> to_compact_;
CompactRewriter rewriter_;
bool materialize_dv_on_commit_;
};

DataFileMetaPriorityQueue GetToCompact() const {
Expand All @@ -208,6 +214,8 @@ class BucketedAppendCompactManager : public CompactFutureManager {
std::optional<std::vector<std::shared_ptr<DataFileMeta>>> compacting_;
DataFileMetaPriorityQueue to_compact_;
std::shared_ptr<CancellationController> cancellation_controller_;
// Realtime deduplicate commits rebuild the bucket DV against their publication snapshot.
bool materialize_dv_on_commit_;
};

} // namespace paimon
25 changes: 16 additions & 9 deletions src/paimon/core/append/bucketed_append_compact_manager_test.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -82,7 +82,8 @@ class BucketedAppendCompactManagerTest : public testing::Test {
executor_, to_compact_before_pick,
/*dv_maintainer=*/nullptr, min_file_num, target_file_size, threshold,
/*force_rewrite_all_files=*/false, /*rewriter=*/nullptr, /*reporter=*/nullptr,
/*cancellation_controller=*/std::make_shared<CancellationController>());
/*cancellation_controller=*/std::make_shared<CancellationController>(),
/*materialize_dv_on_commit=*/false);
auto actual = manager.PickCompactBefore();
if (expected_present) {
ASSERT_TRUE(actual.has_value());
Expand Down Expand Up @@ -332,7 +333,7 @@ TEST_F(BucketedAppendCompactManagerTest, TestCancelCompactionPropagatesToRewrite
/*target_file_size=*/1024,
/*compaction_file_size=*/700,
/*force_rewrite_all_files=*/false, rewriter,
/*reporter=*/nullptr, cancellation_controller);
/*reporter=*/nullptr, cancellation_controller, /*materialize_dv_on_commit=*/false);

ASSERT_OK(manager.TriggerCompaction(/*full_compaction=*/true));
manager.CancelAndWaitCompaction();
Expand All @@ -353,7 +354,7 @@ TEST_F(BucketedAppendCompactManagerTest, TestTriggerCompactionResetsCancelFlag)
/*target_file_size=*/1024,
/*compaction_file_size=*/700,
/*force_rewrite_all_files=*/false, rewriter,
/*reporter=*/nullptr, cancellation_controller);
/*reporter=*/nullptr, cancellation_controller, /*materialize_dv_on_commit=*/false);

ASSERT_OK(manager.TriggerCompaction(/*full_compaction=*/true));
ASSERT_FALSE(cancellation_controller->IsCancelled());
Expand Down Expand Up @@ -386,7 +387,8 @@ TEST_F(BucketedAppendCompactManagerTest, TestHasDeletionFileLargeFileWithDvRetai
/*compaction_file_size=*/500,
/*force_rewrite_all_files=*/false, rewriter,
/*reporter=*/nullptr,
/*cancellation_controller=*/std::make_shared<CancellationController>());
/*cancellation_controller=*/std::make_shared<CancellationController>(),
/*materialize_dv_on_commit=*/false);

ASSERT_OK(manager.TriggerCompaction(/*full_compaction=*/true));
ASSERT_OK_AND_ASSIGN(auto result, manager.GetCompactionResult(/*blocking=*/true));
Expand Down Expand Up @@ -426,7 +428,8 @@ TEST_F(BucketedAppendCompactManagerTest, TestHasDeletionFileLargeFileWithoutDvSk
/*compaction_file_size=*/500,
/*force_rewrite_all_files=*/false, rewriter,
/*reporter=*/nullptr,
/*cancellation_controller=*/std::make_shared<CancellationController>());
/*cancellation_controller=*/std::make_shared<CancellationController>(),
/*materialize_dv_on_commit=*/false);

ASSERT_OK(manager.TriggerCompaction(/*full_compaction=*/true));
ASSERT_OK_AND_ASSIGN(auto result, manager.GetCompactionResult(/*blocking=*/true));
Expand Down Expand Up @@ -458,7 +461,8 @@ TEST_F(BucketedAppendCompactManagerTest, TestDoCompactWithNullDvMaintainerWithLe
/*compaction_file_size=*/500,
/*force_rewrite_all_files=*/false, rewriter,
/*reporter=*/nullptr,
/*cancellation_controller=*/std::make_shared<CancellationController>());
/*cancellation_controller=*/std::make_shared<CancellationController>(),
/*materialize_dv_on_commit=*/false);

ASSERT_OK(manager.TriggerCompaction(/*full_compaction=*/true));
ASSERT_OK_AND_ASSIGN(auto result, manager.GetCompactionResult(/*blocking=*/true));
Expand Down Expand Up @@ -489,7 +493,8 @@ TEST_F(BucketedAppendCompactManagerTest, TestDoNoCompact) {
/*compaction_file_size=*/500,
/*force_rewrite_all_files=*/false, rewriter,
/*reporter=*/nullptr,
/*cancellation_controller=*/std::make_shared<CancellationController>());
/*cancellation_controller=*/std::make_shared<CancellationController>(),
/*materialize_dv_on_commit=*/false);

ASSERT_OK(manager.TriggerCompaction(/*full_compaction=*/true));
ASSERT_OK_AND_ASSIGN(auto result, manager.GetCompactionResult(/*blocking=*/true));
Expand All @@ -514,7 +519,8 @@ TEST_F(BucketedAppendCompactManagerTest, TestAllFilesWithoutCompacting) {
/*compaction_file_size=*/500,
/*force_rewrite_all_files=*/false, rewriter,
/*reporter=*/nullptr,
/*cancellation_controller=*/std::make_shared<CancellationController>());
/*cancellation_controller=*/std::make_shared<CancellationController>(),
/*materialize_dv_on_commit=*/false);

auto all_files = manager.AllFiles();
// to_compact_ is a min-heap sorted by min_sequence_number, so file1 comes first.
Expand Down Expand Up @@ -552,7 +558,8 @@ TEST_F(BucketedAppendCompactManagerTest, TestAllFilesWithCompacting) {
/*compaction_file_size=*/500,
/*force_rewrite_all_files=*/false, rewriter,
/*reporter=*/nullptr,
/*cancellation_controller=*/std::make_shared<CancellationController>());
/*cancellation_controller=*/std::make_shared<CancellationController>(),
/*materialize_dv_on_commit=*/false);

// Trigger compaction — all 4 small files will be picked
ASSERT_OK(manager.TriggerCompaction(/*full_compaction=*/true));
Expand Down
5 changes: 5 additions & 0 deletions src/paimon/core/operation/abstract_file_store_write.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -98,6 +98,11 @@ AbstractFileStoreWrite::AbstractFileStoreWrite(
}

Status AbstractFileStoreWrite::Write(std::unique_ptr<RecordBatch>&& batch) {
if (!options_.GetRealtimeDeduplicateKeyFields().empty() && !IsRealtimeWrite()) {
return Status::Invalid(
"real-time deduplicate write requires a real-time context; "
"this writer only supports compaction");
}
if (PAIMON_UNLIKELY(batch == nullptr)) {
return Status::Invalid("batch is null pointer");
}
Expand Down
3 changes: 2 additions & 1 deletion src/paimon/core/operation/append_only_file_store_write.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -302,7 +302,8 @@ Result<std::shared_ptr<BatchWriter>> AppendOnlyFileStoreWrite::CreateWriter(
options_.GetTargetFileSize(/*has_primary_key=*/false),
options_.GetCompactionFileSize(/*has_primary_key=*/false),
options_.CompactionForceRewriteAllFiles(), rewriter,
compaction_metrics_->CreateReporter(partition, bucket), cancellation_controller);
compaction_metrics_->CreateReporter(partition, bucket), cancellation_controller,
/*materialize_dv_on_commit=*/!options_.GetRealtimeDeduplicateKeyFields().empty());
}

auto writer = std::make_shared<AppendOnlyWriter>(
Expand Down
43 changes: 43 additions & 0 deletions src/paimon/core/operation/commit/manifest_entry_changes_test.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -35,6 +35,7 @@
#include "paimon/core/manifest/file_kind.h"
#include "paimon/core/manifest/index_manifest_entry.h"
#include "paimon/core/manifest/manifest_entry.h"
#include "paimon/core/operation/commit/realtime_deduplicate_compaction_changes_provider.h"
#include "paimon/core/stats/simple_stats.h"
#include "paimon/core/table/sink/commit_message_impl.h"
#include "paimon/data/timestamp.h"
Expand Down Expand Up @@ -128,6 +129,48 @@ TEST_F(ManifestEntryChangesTest, TestCollectAndSummary) {
ASSERT_NE(std::string::npos, summary.find("2 compact index files"));
}

TEST_F(ManifestEntryChangesTest, TestRealtimeCompactionNormalizesMultipleRewrites) {
const BinaryRow partition = CreateIntRow(10);
const auto original = CreateDataFileMeta("original");
const auto intermediate = CreateDataFileMeta("intermediate");
intermediate->extra_files = {"intermediate.index", "intermediate.offset"};
const auto final = CreateDataFileMeta("final");
// CollectChanges groups DELETE before ADD, not in rewrite execution order.
const std::vector<ManifestEntry> entries = {
{FileKind::Delete(), partition, 0, 2, original},
{FileKind::Delete(), partition, 0, 2, intermediate->CopyWithoutStats()},
{FileKind::Add(), partition, 0, 2, intermediate},
{FileKind::Add(), partition, 0, 2, final},
// The same filename in another bucket must not be cancelled.
{FileKind::Delete(), partition, 1, 2, intermediate},
};
ASSERT_OK_AND_ASSIGN(RealtimeCompactionPlan plan, RealtimeCompactionPlan::Create(entries));
ASSERT_EQ(3, plan.delta_files.size());
ASSERT_EQ("original", plan.delta_files[0].FileName());
ASSERT_EQ("final", plan.delta_files[1].FileName());
ASSERT_EQ(1, plan.delta_files[2].Bucket());
ASSERT_EQ(1, plan.intermediate_files.size());
ASSERT_EQ(intermediate->extra_files, plan.intermediate_files[0].File()->extra_files);
}

TEST_F(ManifestEntryChangesTest, TestRealtimeCompactionRejectsInvalidRewrites) {
const BinaryRow partition = CreateIntRow(10);
const auto file = CreateDataFileMeta("file");
const ManifestEntry input(FileKind::Delete(), partition, 0, 1, file);
const ManifestEntry output(FileKind::Add(), partition, 0, 1, file);
ASSERT_NOK_WITH_MSG(RealtimeCompactionPlan::Create({input, input}), "duplicate");
ASSERT_NOK_WITH_MSG(RealtimeCompactionPlan::Create({output, output}), "duplicate");
const auto inconsistent = CreateDataFileMeta("file");
inconsistent->row_count++;
ASSERT_NOK_WITH_MSG(RealtimeCompactionPlan::Create(
{input, ManifestEntry(FileKind::Add(), partition, 0, 1, inconsistent)}),
"inconsistent intermediate");
// All rows filtered by D0: a valid compaction can have no output files.
ASSERT_OK_AND_ASSIGN(RealtimeCompactionPlan plan, RealtimeCompactionPlan::Create({input}));
ASSERT_EQ(1, plan.delta_files.size());
ASSERT_TRUE(plan.intermediate_files.empty());
}

TEST_F(ManifestEntryChangesTest, TestDropStatsOnlyForDeleteEntries) {
SimpleStats value_stats =
BinaryRowGenerator::GenerateStats({1}, {8}, {0}, GetDefaultPool().get());
Expand Down
Loading
Loading