diff --git a/src/paimon/CMakeLists.txt b/src/paimon/CMakeLists.txt index a8a4afe4c..02918fd5b 100644 --- a/src/paimon/CMakeLists.txt +++ b/src/paimon/CMakeLists.txt @@ -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 @@ -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 diff --git a/src/paimon/core/append/bucketed_append_compact_manager.cpp b/src/paimon/core/append/bucketed_append_compact_manager.cpp index f70421a5f..e1fd343cc 100644 --- a/src/paimon/core/append/bucketed_append_compact_manager.cpp +++ b/src/paimon/core/append/bucketed_append_compact_manager.cpp @@ -30,7 +30,8 @@ BucketedAppendCompactManager::BucketedAppendCompactManager( const std::shared_ptr& 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& reporter, - const std::shared_ptr& cancellation_controller) + const std::shared_ptr& cancellation_controller, + bool materialize_dv_on_commit) : executor_(executor), dv_maintainer_(dv_maintainer), min_file_num_(min_file_num), @@ -43,7 +44,8 @@ BucketedAppendCompactManager::BucketedAppendCompactManager( [](const std::shared_ptr& lhs, const std::shared_ptr& 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); @@ -82,9 +84,9 @@ Status BucketedAppendCompactManager::TriggerFullCompaction() { to_compact_.pop(); } cancellation_controller_->Reset(); - auto compact_task = std::make_shared(reporter_, dv_maintainer_, compacting, - compaction_file_size_, - force_rewrite_all_files_, rewriter_); + auto compact_task = std::make_shared( + reporter_, dv_maintainer_, compacting, compaction_file_size_, force_rewrite_all_files_, + rewriter_, materialize_dv_on_commit_); task_future_ = Via(executor_.get(), [compact_task]() -> Result> { return compact_task->Execute(); }); @@ -100,8 +102,8 @@ void BucketedAppendCompactManager::TriggerCompactionWithBestEffort() { if (picked) { cancellation_controller_->Reset(); compacting_ = picked.value(); - auto compact_task = std::make_shared(reporter_, dv_maintainer_, - compacting_.value(), rewriter_); + auto compact_task = std::make_shared( + reporter_, dv_maintainer_, compacting_.value(), rewriter_, materialize_dv_on_commit_); task_future_ = Via(executor_.get(), [compact_task]() -> Result> { return compact_task->Execute(); @@ -175,7 +177,8 @@ BucketedAppendCompactManager::GetCompactionResult(bool blocking) { Result> BucketedAppendCompactManager::Compact( const std::shared_ptr& dv_maintainer, - const std::vector>& to_compact, CompactRewriter rewriter) { + const std::vector>& to_compact, CompactRewriter rewriter, + bool materialize_dv_on_commit) { PAIMON_ASSIGN_OR_RAISE(std::vector> rewrite, rewriter(to_compact)); @@ -184,9 +187,11 @@ Result> BucketedAppendCompactManager::Compact( for (const auto& file : to_compact) { dv_maintainer->RemoveDeletionVectorOf(file->file_name); } - PAIMON_ASSIGN_OR_RAISE(std::shared_ptr deletion_file, - CompactDeletionFile::GenerateFiles(dv_maintainer)); - result->SetDeletionFile(deletion_file); + if (!materialize_dv_on_commit) { + PAIMON_ASSIGN_OR_RAISE(std::shared_ptr deletion_file, + CompactDeletionFile::GenerateFiles(dv_maintainer)); + result->SetDeletionFile(deletion_file); + } } return result; } @@ -209,7 +214,7 @@ Result> BucketedAppendCompactManager::FullCompact return Compact( dv_maintainer_, std::vector>(to_compact_.begin(), to_compact_.end()), - rewriter_); + rewriter_, materialize_dv_on_commit_); } else { // compute small files int32_t big = 0; @@ -226,7 +231,7 @@ Result> BucketedAppendCompactManager::FullCompact return Compact( /*dv_maintainer=*/nullptr, std::vector>(to_compact_.begin(), to_compact_.end()), - rewriter_); + rewriter_, materialize_dv_on_commit_); } else { return std::make_shared(); } diff --git a/src/paimon/core/append/bucketed_append_compact_manager.h b/src/paimon/core/append/bucketed_append_compact_manager.h index d4838f810..9da839dd2 100644 --- a/src/paimon/core/append/bucketed_append_compact_manager.h +++ b/src/paimon/core/append/bucketed_append_compact_manager.h @@ -81,7 +81,8 @@ class BucketedAppendCompactManager : public CompactFutureManager { const std::shared_ptr& 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& reporter, - const std::shared_ptr& cancellation_controller); + const std::shared_ptr& cancellation_controller, + bool materialize_dv_on_commit); ~BucketedAppendCompactManager() override = default; void RequestCancelCompaction() override; @@ -125,7 +126,8 @@ class BucketedAppendCompactManager : public CompactFutureManager { static Result> Compact( const std::shared_ptr& dv_maintainer, - const std::vector>& to_compact, CompactRewriter rewriter); + const std::vector>& to_compact, CompactRewriter rewriter, + bool materialize_dv_on_commit); /// A `CompactTask` impl for full compaction of append-only table. class FullCompactTask : public CompactTask { @@ -134,13 +136,14 @@ class BucketedAppendCompactManager : public CompactFutureManager { const std::shared_ptr& dv_maintainer, const std::vector>& 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> DoCompact() override; @@ -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. @@ -172,21 +176,23 @@ class BucketedAppendCompactManager : public CompactFutureManager { AutoCompactTask(const std::shared_ptr& reporter, const std::shared_ptr& dv_maintainer, const std::vector>& 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> DoCompact() override { - return Compact(dv_maintainer_, to_compact_, rewriter_); + return Compact(dv_maintainer_, to_compact_, rewriter_, materialize_dv_on_commit_); } private: std::shared_ptr dv_maintainer_; std::vector> to_compact_; CompactRewriter rewriter_; + bool materialize_dv_on_commit_; }; DataFileMetaPriorityQueue GetToCompact() const { @@ -208,6 +214,8 @@ class BucketedAppendCompactManager : public CompactFutureManager { std::optional>> compacting_; DataFileMetaPriorityQueue to_compact_; std::shared_ptr cancellation_controller_; + // Realtime deduplicate commits rebuild the bucket DV against their publication snapshot. + bool materialize_dv_on_commit_; }; } // namespace paimon diff --git a/src/paimon/core/append/bucketed_append_compact_manager_test.cpp b/src/paimon/core/append/bucketed_append_compact_manager_test.cpp index 712dd3667..91fa25650 100644 --- a/src/paimon/core/append/bucketed_append_compact_manager_test.cpp +++ b/src/paimon/core/append/bucketed_append_compact_manager_test.cpp @@ -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()); + /*cancellation_controller=*/std::make_shared(), + /*materialize_dv_on_commit=*/false); auto actual = manager.PickCompactBefore(); if (expected_present) { ASSERT_TRUE(actual.has_value()); @@ -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(); @@ -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()); @@ -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()); + /*cancellation_controller=*/std::make_shared(), + /*materialize_dv_on_commit=*/false); ASSERT_OK(manager.TriggerCompaction(/*full_compaction=*/true)); ASSERT_OK_AND_ASSIGN(auto result, manager.GetCompactionResult(/*blocking=*/true)); @@ -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()); + /*cancellation_controller=*/std::make_shared(), + /*materialize_dv_on_commit=*/false); ASSERT_OK(manager.TriggerCompaction(/*full_compaction=*/true)); ASSERT_OK_AND_ASSIGN(auto result, manager.GetCompactionResult(/*blocking=*/true)); @@ -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()); + /*cancellation_controller=*/std::make_shared(), + /*materialize_dv_on_commit=*/false); ASSERT_OK(manager.TriggerCompaction(/*full_compaction=*/true)); ASSERT_OK_AND_ASSIGN(auto result, manager.GetCompactionResult(/*blocking=*/true)); @@ -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()); + /*cancellation_controller=*/std::make_shared(), + /*materialize_dv_on_commit=*/false); ASSERT_OK(manager.TriggerCompaction(/*full_compaction=*/true)); ASSERT_OK_AND_ASSIGN(auto result, manager.GetCompactionResult(/*blocking=*/true)); @@ -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()); + /*cancellation_controller=*/std::make_shared(), + /*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. @@ -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()); + /*cancellation_controller=*/std::make_shared(), + /*materialize_dv_on_commit=*/false); // Trigger compaction — all 4 small files will be picked ASSERT_OK(manager.TriggerCompaction(/*full_compaction=*/true)); diff --git a/src/paimon/core/operation/abstract_file_store_write.cpp b/src/paimon/core/operation/abstract_file_store_write.cpp index 8f134f31f..a8eecace4 100644 --- a/src/paimon/core/operation/abstract_file_store_write.cpp +++ b/src/paimon/core/operation/abstract_file_store_write.cpp @@ -98,6 +98,11 @@ AbstractFileStoreWrite::AbstractFileStoreWrite( } Status AbstractFileStoreWrite::Write(std::unique_ptr&& 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"); } diff --git a/src/paimon/core/operation/append_only_file_store_write.cpp b/src/paimon/core/operation/append_only_file_store_write.cpp index 15b89ba90..0d6b699ef 100644 --- a/src/paimon/core/operation/append_only_file_store_write.cpp +++ b/src/paimon/core/operation/append_only_file_store_write.cpp @@ -302,7 +302,8 @@ Result> 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( diff --git a/src/paimon/core/operation/commit/manifest_entry_changes_test.cpp b/src/paimon/core/operation/commit/manifest_entry_changes_test.cpp index 5eb1608c4..af8e88fd5 100644 --- a/src/paimon/core/operation/commit/manifest_entry_changes_test.cpp +++ b/src/paimon/core/operation/commit/manifest_entry_changes_test.cpp @@ -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" @@ -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 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()); diff --git a/src/paimon/core/operation/commit/realtime_deduplicate_commit_changes_provider.cpp b/src/paimon/core/operation/commit/realtime_deduplicate_commit_changes_provider.cpp new file mode 100644 index 000000000..1d0506b70 --- /dev/null +++ b/src/paimon/core/operation/commit/realtime_deduplicate_commit_changes_provider.cpp @@ -0,0 +1,119 @@ +/* + * 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 "paimon/core/operation/commit/realtime_deduplicate_commit_changes_provider.h" + +#include +#include +#include + +#include "paimon/common/utils/scope_guard.h" +#include "paimon/core/utils/file_store_path_factory.h" + +namespace paimon { + +RealtimeDeduplicateCommitChangesProvider::RealtimeDeduplicateCommitChangesProvider( + ManifestEntryChanges changes, + std::map deleted_offsets, + std::shared_ptr table_schema, std::shared_ptr path_factory, + std::shared_ptr scanner, const CoreOptions& options, + std::shared_ptr pool) + : changes_(std::move(changes)), + deleted_offsets_(std::move(deleted_offsets)), + path_factory_(std::move(path_factory)), + materializer_(std::move(table_schema), path_factory_, std::move(scanner), options, + std::move(pool)) {} + +Result> RealtimeDeduplicateCommitChangesProvider::Provide( + const std::optional& latest_snapshot) const { + // Example: F1 offsets [10, 11, 12], baseline DV {0}; deleting offset 11 yields DV {0, 1}. + // If a retry sees offset 11 moved to F2 row 0, update F2's DV instead of F1's. + // Resolve against latest files and this commit's new files, preserving other baseline DVs. + if (changes_.HasCompactChanges()) { + return Status::Invalid( + "realtime deduplicate progress must not contain physical compaction changes"); + } + auto result = std::make_shared( + changes_.append_table_files, changes_.append_changelog, changes_.append_index_files); + ScopeGuard cleanup([&]() { materializer_.CleanUpUnpublishedFiles(*result); }); + + // 1. Group nonempty logical deletions by partition and bucket. + using BucketOffsets = std::map; + std::unordered_map buckets; + std::vector partitions; + for (const auto& [partition_bucket, offsets] : deleted_offsets_) { + if (offsets.IsEmpty()) { + continue; + } + PAIMON_ASSIGN_OR_RAISE(BinaryRow partition, + path_factory_->ToBinaryRow(partition_bucket.partition)); + buckets[partition].emplace(partition_bucket.bucket, &offsets); + partitions.push_back(partition); + } + if (partitions.empty()) { + cleanup.Release(); + return result; + } + // 2. Group latest-snapshot files with this commit's deltas, including all supplied prepares. + PAIMON_ASSIGN_OR_RAISE(RealtimeDeduplicateDvMaterializer::SnapshotBuckets snapshot_buckets, + materializer_.LoadSnapshot(latest_snapshot, partitions)); + for (const ManifestEntry& entry : result->delta_files) { + const auto partition_iter = buckets.find(entry.Partition()); + if (partition_iter != buckets.end() && partition_iter->second.count(entry.Bucket()) != 0) { + snapshot_buckets[entry.Partition()][entry.Bucket()].files.push_back(entry); + } + } + for (const auto& [partition, bucket_offsets] : buckets) { + for (const auto& [bucket, offsets] : bucket_offsets) { + const auto& snapshot = snapshot_buckets[partition][bucket]; + // 3. Apply ADD/DELETE entries to obtain this bucket's final active files. + std::map> active_files; + for (const ManifestEntry& entry : snapshot.files) { + if (entry.Kind() == FileKind::Add()) { + active_files[entry.FileName()] = entry.File(); + } else { + active_files.erase(entry.FileName()); + } + } + RealtimeDeduplicateDvMaterializer::BucketUpdate update; + for (const auto& [name, file] : active_files) { + update.final_file_names.insert(name); + update.lookup_files.push_back(file); + } + update.deleted_offsets = *offsets; + // 4. Map logical deleted offsets to physical row positions in the active files. + PAIMON_ASSIGN_OR_RAISE(RealtimeDeduplicateDvMaterializer::FilePositions positions, + materializer_.LookupFilePositions(partition, bucket, update)); + // Preserve the write-side no-hit fast path: do not read, prune or rewrite baseline + // DVs when none of the logical deletions maps to a current physical row. + if (positions.empty()) { + continue; + } + // 5. Load latest baseline DVs, merge deletions and emit index changes. + PAIMON_ASSIGN_OR_RAISE(RealtimeDeduplicateDvMaterializer::DeletionVectors baseline, + materializer_.ReadDeletionVectors(partition, bucket, snapshot)); + PAIMON_RETURN_NOT_OK(materializer_.MaterializeBucket( + partition, bucket, snapshot, std::move(baseline), update, positions, result.get())); + } + } + cleanup.Release(); + return result; +} + +} // namespace paimon diff --git a/src/paimon/core/operation/commit/realtime_deduplicate_commit_changes_provider.h b/src/paimon/core/operation/commit/realtime_deduplicate_commit_changes_provider.h new file mode 100644 index 000000000..0123c3a93 --- /dev/null +++ b/src/paimon/core/operation/commit/realtime_deduplicate_commit_changes_provider.h @@ -0,0 +1,53 @@ +/* + * 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. + */ + +#pragma once + +#include +#include +#include +#include + +#include "paimon/core/operation/commit/manifest_entry_changes.h" +#include "paimon/core/operation/commit/realtime_deduplicate_dv_materializer.h" +#include "paimon/realtime/realtime_context.h" + +namespace paimon { + +/// Materializes logical realtime deletions against each commit attempt's latest files. +class RealtimeDeduplicateCommitChangesProvider final : public CommitChangesProvider { + public: + RealtimeDeduplicateCommitChangesProvider( + ManifestEntryChanges changes, + std::map deleted_offsets, + std::shared_ptr table_schema, + std::shared_ptr path_factory, std::shared_ptr scanner, + const CoreOptions& options, std::shared_ptr pool); + + Result> Provide( + const std::optional& latest_snapshot) const override; + + private: + ManifestEntryChanges changes_; + std::map deleted_offsets_; + std::shared_ptr path_factory_; + RealtimeDeduplicateDvMaterializer materializer_; +}; + +} // namespace paimon diff --git a/src/paimon/core/operation/commit/realtime_deduplicate_compaction_changes_provider.cpp b/src/paimon/core/operation/commit/realtime_deduplicate_compaction_changes_provider.cpp new file mode 100644 index 000000000..61ec2ba22 --- /dev/null +++ b/src/paimon/core/operation/commit/realtime_deduplicate_compaction_changes_provider.cpp @@ -0,0 +1,312 @@ +/* + * 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 "paimon/core/operation/commit/realtime_deduplicate_compaction_changes_provider.h" + +#include +#include +#include +#include +#include + +#include "arrow/api.h" +#include "arrow/c/bridge.h" +#include "paimon/common/table/special_fields.h" +#include "paimon/common/types/data_field.h" +#include "paimon/common/utils/arrow/mem_utils.h" +#include "paimon/common/utils/checked_cast.h" +#include "paimon/common/utils/scope_guard.h" +#include "paimon/core/casting/casting_utils.h" +#include "paimon/core/deletionvectors/deletion_vectors_index_file.h" +#include "paimon/core/operation/internal_read_context.h" +#include "paimon/core/operation/raw_file_split_read.h" +#include "paimon/core/schema/table_schema.h" +#include "paimon/core/utils/file_store_path_factory.h" +#include "paimon/read_context.h" +#include "paimon/utils/range.h" + +namespace paimon { +namespace { + +Status CollectDeletedOffsets(const ManifestEntry& source, const std::shared_ptr& dv, + const std::shared_ptr& offset_schema, + const std::shared_ptr& pool, RawFileSplitRead* read, + RoaringBitmap64* deleted_offsets) { + std::vector ranges; + bool invalid_position = false; + PAIMON_RETURN_NOT_OK(dv->ForEachDeletedPosition([&](int64_t position) { + if (position < 0 || position >= source.File()->row_count || + position >= std::numeric_limits::max()) { + invalid_position = true; + } else if (!ranges.empty() && position == ranges.back().to + 1) { + ranges.back().to = position; + } else { + ranges.emplace_back(position, position); + } + })); + if (invalid_position) { + return Status::Invalid("compaction input DV contains an invalid row position"); + } + if (ranges.empty()) { + return Status::OK(); + } + ranges = Range::SortAndMergeOverlap(ranges, /*adjacent=*/true); + int64_t expected_rows = 0; + for (const Range& range : ranges) { + expected_rows += range.Count(); + } + // Read deleted physical rows to recover their stable offsets. Applying the input DV here + // would hide precisely the rows whose deletions need to be transferred to the output. + DeletionVector::Factory no_deletions = + [](const std::string&) -> Result> { + return std::shared_ptr(); + }; + PAIMON_ASSIGN_OR_RAISE( + std::unique_ptr reader, + read->CreateReader(source.Partition(), source.Bucket(), {source.File()}, no_deletions, + ranges, offset_schema, /*predicate=*/nullptr)); + ScopeGuard reader_guard([&]() { reader->Close(); }); + const std::shared_ptr arrow_pool = GetArrowPool(pool); + int64_t emitted_rows = 0; + while (true) { + PAIMON_ASSIGN_OR_RAISE(BatchReader::ReadBatchWithBitmap batch_with_bitmap, + reader->NextBatchWithBitmap()); + if (BatchReader::IsEofBatch(batch_with_bitmap)) { + break; + } + auto& [batch, selected_rows] = batch_with_bitmap; + PAIMON_ASSIGN_OR_RAISE_FROM_ARROW( + std::shared_ptr array, + arrow::ImportArray(batch.first.get(), batch.second.get())); + if (array->type_id() != arrow::Type::STRUCT) { + return Status::Invalid("compaction offset reader returned a non-struct array"); + } + std::shared_ptr raw_offsets = + checked_pointer_cast(array)->GetFieldByName( + SpecialFields::RealtimeOffset().Name()); + if (raw_offsets && raw_offsets->type_id() == arrow::Type::DICTIONARY) { + PAIMON_ASSIGN_OR_RAISE( + raw_offsets, + CastingUtils::Cast(raw_offsets, arrow::int64(), arrow::compute::CastOptions::Safe(), + arrow_pool.get())); + } + if (!raw_offsets || raw_offsets->type_id() != arrow::Type::INT64) { + return Status::Invalid("compaction input has no valid realtime offsets"); + } + const auto offsets = checked_pointer_cast(raw_offsets); + // The raw batch may contain unselected rows. Only selected offsets participate in + // migration and row-count validation; do not materialize a filtered Arrow array. + emitted_rows += selected_rows.Cardinality(); + for (auto selected = selected_rows.Begin(); selected != selected_rows.End(); ++selected) { + const int64_t row = *selected; + if (row >= offsets->length()) { + return Status::Invalid("compaction selected row is out of bounds"); + } + if (offsets->IsNull(row)) { + return Status::Invalid("compaction input contains a null realtime offset"); + } + if (offsets->Value(row) < 0) { + return Status::Invalid("compaction input contains a negative realtime offset"); + } + deleted_offsets->Add(offsets->Value(row)); + } + } + if (emitted_rows != expected_rows) { + return Status::Invalid("compaction offset reader did not return all deleted rows"); + } + return Status::OK(); +} + +} // namespace + +Result RealtimeCompactionPlan::Create( + const std::vector& entries) { + std::unordered_map inputs; + std::unordered_map outputs; + for (const ManifestEntry& entry : entries) { + if (!entry.File() || entry.File()->row_count < 0) { + return Status::Invalid("invalid realtime compaction file"); + } + if (!(entry.Kind() == FileKind::Add()) && !(entry.Kind() == FileKind::Delete())) { + return Status::Invalid("invalid realtime compaction entry kind"); + } + auto& files = entry.Kind() == FileKind::Add() ? outputs : inputs; + if (!files.emplace(entry.CreateIdentifier(), &entry).second) { + return Status::Invalid("duplicate realtime compaction input or output: ", + entry.FileName()); + } + } + // For F1/F2 -> F3 -> F4, delta_files contains DELETE F1/F2 and ADD F4. + // intermediate_files contains ADD F3, whose matching DELETE cancels it from the final changes. + RealtimeCompactionPlan plan; + for (const ManifestEntry& entry : entries) { + const auto id = entry.CreateIdentifier(); + if (entry.Kind() == FileKind::Delete()) { + if (outputs.count(id) == 0) { + plan.delta_files.push_back(entry); + } else { + const auto& output = outputs.at(id)->File(); + if (entry.File()->row_count != output->row_count || + entry.File()->schema_id != output->schema_id || + entry.File()->file_size != output->file_size) { + return Status::Invalid("inconsistent intermediate compaction file: ", + entry.FileName()); + } + } + } else if (inputs.count(id) == 0) { + plan.delta_files.push_back(entry); + } else { + plan.intermediate_files.push_back(entry); + } + } + return plan; +} + +RealtimeDeduplicateCompactionChangesProvider::RealtimeDeduplicateCompactionChangesProvider( + RealtimeCompactionPlan plan, std::vector changelog_files, + std::vector index_entries, std::shared_ptr table_schema, + std::shared_ptr path_factory, std::shared_ptr scanner, + const CoreOptions& options, std::shared_ptr executor, + std::shared_ptr pool) + : plan_(std::move(plan)), + changelog_files_(std::move(changelog_files)), + index_entries_(std::move(index_entries)), + table_schema_(std::move(table_schema)), + path_factory_(std::move(path_factory)), + options_(options), + executor_(std::move(executor)), + pool_(std::move(pool)), + materializer_(table_schema_, path_factory_, std::move(scanner), options_, pool_) {} + +Result> RealtimeDeduplicateCompactionChangesProvider::Provide( + const std::optional& latest_snapshot) const { + // Example: F1 offsets [10, 11, 12], DV {0} -> F2 offsets [11, 12]. + // A later deletion of offset 11 makes F1's latest DV {0, 1}; publish F2's DV {0}, + // retire F1's DV, and preserve DVs of other active files in the bucket. + // TODO(xinyu.lxy): Consider generating candidate DVs during compaction and carrying their + // bucket DV baseline in the commit message. Reuse candidates when the inputs remain active + // and the latest bucket baseline matches; otherwise rebase against the latest DVs here. + // Check the baseline on the first attempt too, not only after a snapshot CAS conflict. + if (!latest_snapshot) { + return Status::Invalid("realtime compaction requires a committed input snapshot"); + } + auto result = + std::make_shared(plan_.delta_files, changelog_files_, index_entries_); + for (const IndexManifestEntry& entry : index_entries_) { + if (entry.index_file->IndexType() == DeletionVectorsIndexFile::DELETION_VECTORS_INDEX) { + return Status::Invalid("realtime compaction must generate deletion vectors at commit"); + } + } + ScopeGuard cleanup([&]() { materializer_.CleanUpUnpublishedFiles(*result); }); + // 1. Group the normalized compaction changes by partition and bucket. + using BucketFiles = std::map>; + std::unordered_map buckets; + std::vector affected_partitions; + for (const ManifestEntry& entry : plan_.delta_files) { + buckets[entry.Partition()][entry.Bucket()].push_back(entry); + affected_partitions.push_back(entry.Partition()); + } + for (const ManifestEntry& entry : plan_.intermediate_files) { + affected_partitions.push_back(entry.Partition()); + } + // 2. Load the latest snapshot and verify intermediate outputs are not already committed. + PAIMON_ASSIGN_OR_RAISE(RealtimeDeduplicateDvMaterializer::SnapshotBuckets snapshot_buckets, + materializer_.LoadSnapshot(latest_snapshot, affected_partitions)); + for (const ManifestEntry& entry : plan_.intermediate_files) { + const auto id = entry.CreateIdentifier(); + for (const ManifestEntry& active : + snapshot_buckets[entry.Partition()][entry.Bucket()].files) { + if (active.CreateIdentifier() == id) { + return Status::Invalid("intermediate realtime compaction file is already active: ", + entry.FileName()); + } + } + } + ReadContextBuilder builder(path_factory_->RootPath()); + PAIMON_ASSIGN_OR_RAISE(std::shared_ptr read_context, + builder.SetOptions(options_.ToMap()) + .WithFileSystem(options_.GetFileSystem()) + .WithMemoryPool(pool_) + .Finish()); + PAIMON_ASSIGN_OR_RAISE( + std::shared_ptr context, + InternalReadContext::Create(read_context, table_schema_, options_.ToMap())); + RawFileSplitRead read(path_factory_, context, pool_, executor_); + const auto schema = DataField::ConvertDataFieldsToArrowSchema(table_schema_->Fields()); + const auto offset_schema = + arrow::schema({schema->GetFieldByName(SpecialFields::RealtimeOffset().Name())}); + for (const auto& [partition, bucket_files] : buckets) { + for (const auto& [bucket, files] : bucket_files) { + std::vector inputs; + const auto& snapshot = snapshot_buckets[partition][bucket]; + std::unordered_map active_files; + RealtimeDeduplicateDvMaterializer::BucketUpdate update; + for (const ManifestEntry& entry : snapshot.files) { + active_files.emplace(entry.CreateIdentifier(), entry); + update.final_file_names.insert(entry.FileName()); + } + // 3. Validate inputs and build the final file set; only final outputs need lookup. + for (const ManifestEntry& entry : files) { + if (entry.Kind() == FileKind::Delete()) { + const auto iter = active_files.find(entry.CreateIdentifier()); + if (iter == active_files.end()) { + return Status::Invalid("realtime compaction input is no longer active: ", + entry.FileName()); + } + // DELETE entries may omit stats; use the snapshot's complete metadata. + inputs.push_back(iter->second); + update.final_file_names.erase(entry.FileName()); + } else { + if (active_files.count(entry.CreateIdentifier()) != 0) { + return Status::Invalid("realtime compaction output is already active: ", + entry.FileName()); + } + update.lookup_files.push_back(entry.File()); + update.final_file_names.insert(entry.FileName()); + } + } + // 4. Load latest bucket DVs and collect deleted offsets from the original inputs. + PAIMON_ASSIGN_OR_RAISE(RealtimeDeduplicateDvMaterializer::DeletionVectors baseline, + materializer_.ReadDeletionVectors(partition, bucket, snapshot)); + if (!update.lookup_files.empty()) { + // Collect from the complete baseline before the common materializer retires + // input DVs. All inputs contribute to one bucket-level offset lookup. + for (const ManifestEntry& input : inputs) { + const auto iter = baseline.find(input.FileName()); + if (iter != baseline.end()) { + PAIMON_RETURN_NOT_OK(CollectDeletedOffsets(input, iter->second, + offset_schema, pool_, &read, + &update.deleted_offsets)); + } + } + } + // 5. Map the aggregated deleted offsets to physical positions in the final outputs. + PAIMON_ASSIGN_OR_RAISE(RealtimeDeduplicateDvMaterializer::FilePositions positions, + materializer_.LookupFilePositions(partition, bucket, update)); + // 6. Retire input DVs and merge output deletions, preserving other active files' DVs. + // Run even with no output hits, e.g. only offset 10 was deleted before the rewrite. + PAIMON_RETURN_NOT_OK(materializer_.MaterializeBucket( + partition, bucket, snapshot, std::move(baseline), update, positions, result.get())); + } + } + cleanup.Release(); + return result; +} + +} // namespace paimon diff --git a/src/paimon/core/operation/commit/realtime_deduplicate_compaction_changes_provider.h b/src/paimon/core/operation/commit/realtime_deduplicate_compaction_changes_provider.h new file mode 100644 index 000000000..ee7aa4193 --- /dev/null +++ b/src/paimon/core/operation/commit/realtime_deduplicate_compaction_changes_provider.h @@ -0,0 +1,73 @@ +/* + * 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. + */ + +#pragma once + +#include +#include +#include +#include + +#include "paimon/core/core_options.h" +#include "paimon/core/operation/commit/commit_changes_provider.h" +#include "paimon/core/operation/commit/realtime_deduplicate_dv_materializer.h" + +namespace paimon { + +class CommitScanner; +class Executor; +class FileStorePathFactory; +class MemoryPool; +class TableSchema; + +/// Net changes of any number of rewrites included in a single compaction commit. +struct RealtimeCompactionPlan { + static Result Create(const std::vector& entries); + + std::vector delta_files; + // Outputs consumed by later rewrites in the same commit. + std::vector intermediate_files; +}; + +/// Rebases realtime deduplicate compaction DVs without rewriting the data files again. +class RealtimeDeduplicateCompactionChangesProvider final : public CommitChangesProvider { + public: + RealtimeDeduplicateCompactionChangesProvider( + RealtimeCompactionPlan plan, std::vector changelog_files, + std::vector index_entries, std::shared_ptr table_schema, + std::shared_ptr path_factory, std::shared_ptr scanner, + const CoreOptions& options, std::shared_ptr executor, + std::shared_ptr pool); + + Result> Provide( + const std::optional& latest_snapshot) const override; + + private: + RealtimeCompactionPlan plan_; + std::vector changelog_files_; + std::vector index_entries_; + std::shared_ptr table_schema_; + std::shared_ptr path_factory_; + CoreOptions options_; + std::shared_ptr executor_; + std::shared_ptr pool_; + RealtimeDeduplicateDvMaterializer materializer_; +}; + +} // namespace paimon diff --git a/src/paimon/core/operation/commit/realtime_deduplicate_dv_materializer.cpp b/src/paimon/core/operation/commit/realtime_deduplicate_dv_materializer.cpp new file mode 100644 index 000000000..c96711568 --- /dev/null +++ b/src/paimon/core/operation/commit/realtime_deduplicate_dv_materializer.cpp @@ -0,0 +1,167 @@ +/* + * 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 "paimon/core/operation/commit/realtime_deduplicate_dv_materializer.h" + +#include + +#include "paimon/common/types/data_field.h" +#include "paimon/core/deletionvectors/bucketed_dv_maintainer.h" +#include "paimon/core/deletionvectors/deletion_vectors_index_file.h" +#include "paimon/core/operation/commit/commit_scanner.h" +#include "paimon/core/realtime/realtime_offset_file_index_lookup.h" +#include "paimon/core/schema/table_schema.h" +#include "paimon/core/utils/file_store_path_factory.h" +#include "paimon/fs/file_system.h" + +namespace paimon { + +RealtimeDeduplicateDvMaterializer::RealtimeDeduplicateDvMaterializer( + std::shared_ptr table_schema, std::shared_ptr path_factory, + std::shared_ptr scanner, const CoreOptions& options, + std::shared_ptr pool) + : table_schema_(std::move(table_schema)), + path_factory_(std::move(path_factory)), + scanner_(std::move(scanner)), + options_(options), + pool_(std::move(pool)) {} + +Result +RealtimeDeduplicateDvMaterializer::LoadSnapshot(const std::optional& snapshot, + const std::vector& partitions) const { + SnapshotBuckets buckets; + if (!snapshot || partitions.empty()) { + return buckets; + } + std::vector> filters; + const std::unordered_set unique_partitions(partitions.begin(), partitions.end()); + for (const BinaryRow& partition : unique_partitions) { + using PartitionValues = std::vector>; + PAIMON_ASSIGN_OR_RAISE(PartitionValues values, + path_factory_->GeneratePartitionVector(partition)); + filters.emplace_back(values.begin(), values.end()); + } + PAIMON_ASSIGN_OR_RAISE(std::vector files, + scanner_->ReadAllEntriesFromPartitions(*snapshot, filters)); + PAIMON_ASSIGN_OR_RAISE(std::vector indexes, + scanner_->ReadAllIndexEntriesFromPartitions(*snapshot, filters)); + for (const ManifestEntry& entry : files) { + buckets[entry.Partition()][entry.Bucket()].files.push_back(entry); + } + for (const IndexManifestEntry& entry : indexes) { + if (entry.index_file->IndexType() == DeletionVectorsIndexFile::DELETION_VECTORS_INDEX) { + buckets[entry.partition][entry.bucket].dv_index_files.push_back(entry.index_file); + } + } + return buckets; +} + +Result> RealtimeDeduplicateDvMaterializer::CreateDvFile( + const BinaryRow& partition, int32_t bucket) const { + PAIMON_ASSIGN_OR_RAISE(std::shared_ptr paths, + path_factory_->CreateIndexFileFactory(partition, bucket)); + return std::make_shared(options_.GetFileSystem(), paths, + /*bitmap64=*/false, pool_); +} + +Result +RealtimeDeduplicateDvMaterializer::ReadDeletionVectors(const BinaryRow& partition, int32_t bucket, + const BucketSnapshot& snapshot) const { + if (snapshot.dv_index_files.empty()) { + return DeletionVectors(); + } + PAIMON_ASSIGN_OR_RAISE(std::shared_ptr file, + CreateDvFile(partition, bucket)); + return file->ReadAllDeletionVectors(snapshot.dv_index_files); +} + +Result +RealtimeDeduplicateDvMaterializer::LookupFilePositions(const BinaryRow& partition, int32_t bucket, + const BucketUpdate& update) const { + FilePositions positions; + if (!update.deleted_offsets.IsEmpty() && !update.lookup_files.empty()) { + const auto key_fields = options_.GetRealtimeDeduplicateKeyFields(); + if (key_fields.empty()) { + return Status::Invalid("realtime DV materialization requires deduplicate key fields"); + } + const auto schema = DataField::ConvertDataFieldsToArrowSchema(table_schema_->Fields()); + PAIMON_ASSIGN_OR_RAISE(std::shared_ptr paths, + path_factory_->CreateDataFilePathFactory(partition, bucket)); + PAIMON_ASSIGN_OR_RAISE( + std::shared_ptr lookup, + RealtimeOffsetFileIndexLookup::Create( + schema, table_schema_->Id(), schema->GetFieldByName(key_fields.front()), + update.lookup_files, paths, options_.GetFileSystem(), pool_, options_)); + PAIMON_ASSIGN_OR_RAISE(positions, lookup->LookupFilePositions(update.deleted_offsets)); + } + + return positions; +} + +Status RealtimeDeduplicateDvMaterializer::MaterializeBucket( + const BinaryRow& partition, int32_t bucket, const BucketSnapshot& snapshot, + DeletionVectors baseline, const BucketUpdate& update, const FilePositions& positions, + CommitChanges* changes) const { + PAIMON_ASSIGN_OR_RAISE(std::shared_ptr file, + CreateDvFile(partition, bucket)); + // baseline is freshly read for this Provide call. Never mutate a cached or writer-owned DV. + BucketedDvMaintainer maintainer(file, baseline); + for (auto iter = baseline.begin(); iter != baseline.end();) { + if (update.final_file_names.count(iter->first) == 0) { + maintainer.RemoveDeletionVectorOf(iter->first); + iter = baseline.erase(iter); + } else { + ++iter; + } + } + bool has_dvs = !baseline.empty(); + for (const auto& [name, bitmap] : positions) { + if (update.final_file_names.count(name) == 0) { + return Status::Invalid("realtime DV lookup returned a file absent from the final set"); + } + for (auto iter = bitmap.Begin(); iter != bitmap.End(); ++iter) { + PAIMON_RETURN_NOT_OK(maintainer.NotifyNewDeletion(name, *iter)); + has_dvs = true; + } + } + if (!has_dvs) { + // Replacing a fixed bucket's DV with an empty map must remove the old index reference. + for (const auto& index : snapshot.dv_index_files) { + changes->index_entries.emplace_back(FileKind::Delete(), partition, bucket, index); + } + return Status::OK(); + } + PAIMON_ASSIGN_OR_RAISE(std::optional> index, + maintainer.WriteDeletionVectorsIndex()); + if (index) { + changes->temporary_files.push_back(file->Path(*index)); + changes->index_entries.emplace_back(FileKind::Add(), partition, bucket, *index); + } + return Status::OK(); +} + +void RealtimeDeduplicateDvMaterializer::CleanUpUnpublishedFiles( + const CommitChanges& changes) const { + for (const std::string& path : changes.temporary_files) { + [[maybe_unused]] Status status = + options_.GetFileSystem()->Delete(path, /*recursive=*/false); + } +} + +} // namespace paimon diff --git a/src/paimon/core/operation/commit/realtime_deduplicate_dv_materializer.h b/src/paimon/core/operation/commit/realtime_deduplicate_dv_materializer.h new file mode 100644 index 000000000..72cb7575e --- /dev/null +++ b/src/paimon/core/operation/commit/realtime_deduplicate_dv_materializer.h @@ -0,0 +1,99 @@ +/* + * 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. + */ + +#pragma once + +#include +#include +#include +#include +#include +#include +#include +#include + +#include "paimon/core/core_options.h" +#include "paimon/core/operation/commit/commit_changes_provider.h" +#include "paimon/utils/roaring_bitmap64.h" + +namespace paimon { + +class CommitScanner; +class DeletionVector; +class DeletionVectorsIndexFile; +class FileStorePathFactory; +class MemoryPool; +class TableSchema; + +/// Shared, attempt-local DV processing for realtime deduplicate writes and compaction. +class RealtimeDeduplicateDvMaterializer { + public: + using DeletionVectors = std::map>; + using FilePositions = std::map; + + struct BucketSnapshot { + std::vector files; + std::vector> dv_index_files; + }; + + using SnapshotBuckets = std::unordered_map>; + + // Compaction F1/F2 -> F3 with F5 retained: lookup_files={F3}, final_file_names={F3, F5}, + // and deleted_offsets contains offsets recovered from rows deleted by F1/F2's latest DVs. + struct BucketUpdate { + std::vector> lookup_files; + // Unlike lookup_files, this includes files not participating in compaction. + std::unordered_set final_file_names; + RoaringBitmap64 deleted_offsets; + }; + + RealtimeDeduplicateDvMaterializer(std::shared_ptr table_schema, + std::shared_ptr path_factory, + std::shared_ptr scanner, + const CoreOptions& options, std::shared_ptr pool); + + Result LoadSnapshot(const std::optional& snapshot, + const std::vector& partitions) const; + + Result ReadDeletionVectors(const BinaryRow& partition, int32_t bucket, + const BucketSnapshot& snapshot) const; + + Result LookupFilePositions(const BinaryRow& partition, int32_t bucket, + const BucketUpdate& update) const; + + Status MaterializeBucket(const BinaryRow& partition, int32_t bucket, + const BucketSnapshot& snapshot, DeletionVectors baseline, + const BucketUpdate& update, const FilePositions& positions, + CommitChanges* changes) const; + + /// Only for files whose provider has not yet returned successfully. + void CleanUpUnpublishedFiles(const CommitChanges& changes) const; + + private: + Result> CreateDvFile(const BinaryRow& partition, + int32_t bucket) const; + + std::shared_ptr table_schema_; + std::shared_ptr path_factory_; + std::shared_ptr scanner_; + CoreOptions options_; + std::shared_ptr pool_; +}; + +} // namespace paimon diff --git a/src/paimon/core/operation/commit/realtime_deduplicate_dv_materializer_test.cpp b/src/paimon/core/operation/commit/realtime_deduplicate_dv_materializer_test.cpp new file mode 100644 index 000000000..0faad1bf5 --- /dev/null +++ b/src/paimon/core/operation/commit/realtime_deduplicate_dv_materializer_test.cpp @@ -0,0 +1,248 @@ +/* + * 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 "paimon/core/operation/commit/realtime_deduplicate_dv_materializer.h" + +#include +#include +#include +#include +#include +#include +#include + +#include "arrow/type.h" +#include "gtest/gtest.h" +#include "paimon/core/deletionvectors/bitmap_deletion_vector.h" +#include "paimon/core/deletionvectors/deletion_vectors_index_file.h" +#include "paimon/core/operation/commit/realtime_deduplicate_commit_changes_provider.h" +#include "paimon/core/schema/table_schema.h" +#include "paimon/core/utils/file_store_path_factory.h" +#include "paimon/defs.h" +#include "paimon/fs/file_system.h" +#include "paimon/testing/utils/testharness.h" + +namespace paimon::test { + +class RealtimeDeduplicateDvMaterializerTest : public testing::Test { + protected: + using Materializer = RealtimeDeduplicateDvMaterializer; + + void SetUp() override { + dir_ = UniqueTestDirectory::Create("local"); + ASSERT_NE(nullptr, dir_); + const std::map options = {{Options::FILE_SYSTEM, "local"}, + {Options::BUCKET, "1"}}; + ASSERT_OK_AND_ASSIGN(options_, CoreOptions::FromMap(options)); + const auto schema = arrow::schema({arrow::field("id", arrow::int64())}); + ASSERT_OK_AND_ASSIGN(table_schema_, TableSchema::Create(0, schema, {}, {}, options)); + ASSERT_OK_AND_ASSIGN( + paths_, FileStorePathFactory::Create( + dir_->Str(), schema, {}, "__DEFAULT_PARTITION__", "parquet", "data-", + /*legacy_partition_name_enabled=*/true, {}, std::nullopt, + /*index_file_in_data_file_dir=*/false, GetDefaultPool())); + materializer_ = std::make_unique(table_schema_, paths_, /*scanner=*/nullptr, + options_, GetDefaultPool()); + } + + Result> CreateDvFile(int32_t bucket) const { + PAIMON_ASSIGN_OR_RAISE(std::shared_ptr paths, + paths_->CreateIndexFileFactory(BinaryRow(0), bucket)); + return std::make_shared(options_.GetFileSystem(), paths, + /*bitmap64=*/false, GetDefaultPool()); + } + + std::shared_ptr MakeDv(const std::vector& positions) const { + RoaringBitmap32 bitmap; + for (uint32_t position : positions) { + bitmap.Add(position); + } + return std::make_shared(bitmap); + } + + std::unique_ptr dir_; + CoreOptions options_; + std::shared_ptr table_schema_; + std::shared_ptr paths_; + std::unique_ptr materializer_; +}; + +TEST_F(RealtimeDeduplicateDvMaterializerTest, TestEmptySnapshotAndNoopUpdate) { + ASSERT_OK_AND_ASSIGN(Materializer::SnapshotBuckets buckets, + materializer_->LoadSnapshot(std::nullopt, {BinaryRow(0)})); + ASSERT_TRUE(buckets.empty()); + Materializer::BucketSnapshot snapshot; + Materializer::BucketUpdate update; + CommitChanges changes; + ASSERT_OK( + materializer_->MaterializeBucket(BinaryRow(0), 0, snapshot, {}, update, {}, &changes)); + ASSERT_TRUE(changes.index_entries.empty()); + ASSERT_TRUE(changes.temporary_files.empty()); +} + +TEST_F(RealtimeDeduplicateDvMaterializerTest, TestUnchangedBaselineDoesNotRewrite) { + ASSERT_OK_AND_ASSIGN(std::shared_ptr file, CreateDvFile(0)); + ASSERT_OK_AND_ASSIGN(std::shared_ptr index, + file->WriteSingleFile({{"kept", MakeDv({1, 3})}})); + Materializer::BucketSnapshot snapshot; + snapshot.dv_index_files = {index}; + Materializer::BucketUpdate update; + update.final_file_names = {"kept"}; + ASSERT_OK_AND_ASSIGN(Materializer::DeletionVectors baseline, + materializer_->ReadDeletionVectors(BinaryRow(0), 0, snapshot)); + CommitChanges changes; + ASSERT_OK(materializer_->MaterializeBucket(BinaryRow(0), 0, snapshot, std::move(baseline), + update, {}, &changes)); + ASSERT_TRUE(changes.index_entries.empty()); + ASSERT_TRUE(changes.temporary_files.empty()); + ASSERT_OK_AND_ASSIGN(bool exists, file->Exists(index)); + ASSERT_TRUE(exists); +} + +TEST_F(RealtimeDeduplicateDvMaterializerTest, TestPruneRetiredFilesAndCleanUpAttempt) { + ASSERT_OK_AND_ASSIGN(std::shared_ptr file, CreateDvFile(0)); + ASSERT_OK_AND_ASSIGN( + std::shared_ptr index, + file->WriteSingleFile({{"retired", MakeDv({0})}, {"kept", MakeDv({1, 3})}})); + Materializer::BucketSnapshot snapshot; + snapshot.dv_index_files = {index}; + Materializer::BucketUpdate update; + update.final_file_names = {"kept"}; + ASSERT_OK_AND_ASSIGN(Materializer::DeletionVectors baseline, + materializer_->ReadDeletionVectors(BinaryRow(0), 0, snapshot)); + CommitChanges changes; + ASSERT_OK(materializer_->MaterializeBucket(BinaryRow(0), 0, snapshot, std::move(baseline), + update, {}, &changes)); + ASSERT_EQ(1, changes.index_entries.size()); + ASSERT_EQ(FileKind::Add(), changes.index_entries[0].kind); + ASSERT_EQ(1, changes.temporary_files.size()); + ASSERT_OK_AND_ASSIGN(Materializer::DeletionVectors written, + file->ReadAllDeletionVectors(changes.index_entries[0].index_file)); + ASSERT_EQ(1, written.size()); + ASSERT_EQ(1, written.count("kept")); + ASSERT_OK_AND_ASSIGN(int64_t count, written.at("kept")->GetCardinality()); + ASSERT_EQ(2, count); + + materializer_->CleanUpUnpublishedFiles(changes); + ASSERT_OK_AND_ASSIGN(bool attempt_exists, + options_.GetFileSystem()->Exists(changes.temporary_files[0])); + ASSERT_FALSE(attempt_exists); + ASSERT_OK_AND_ASSIGN(bool baseline_exists, file->Exists(index)); + ASSERT_TRUE(baseline_exists); + // A retry reloads the original snapshot; cleanup and pruning never modify its DV file. + ASSERT_OK_AND_ASSIGN(Materializer::DeletionVectors reloaded, + materializer_->ReadDeletionVectors(BinaryRow(0), 0, snapshot)); + ASSERT_EQ(2, reloaded.size()); + ASSERT_EQ(1, reloaded.count("retired")); +} + +TEST_F(RealtimeDeduplicateDvMaterializerTest, TestEmptyFinalDvRemovesOldIndexReference) { + ASSERT_OK_AND_ASSIGN(std::shared_ptr file, CreateDvFile(0)); + ASSERT_OK_AND_ASSIGN(std::shared_ptr index, + file->WriteSingleFile({{"retired", MakeDv({0})}})); + Materializer::BucketSnapshot snapshot; + snapshot.dv_index_files = {index}; + Materializer::BucketUpdate update; + ASSERT_OK_AND_ASSIGN(Materializer::DeletionVectors baseline, + materializer_->ReadDeletionVectors(BinaryRow(0), 0, snapshot)); + CommitChanges changes; + ASSERT_OK(materializer_->MaterializeBucket(BinaryRow(0), 0, snapshot, std::move(baseline), + update, {}, &changes)); + ASSERT_EQ(1, changes.index_entries.size()); + ASSERT_EQ(FileKind::Delete(), changes.index_entries[0].kind); + ASSERT_EQ(index, changes.index_entries[0].index_file); + ASSERT_TRUE(changes.temporary_files.empty()); + // Publication removes the manifest reference, not the file needed by older snapshots. + ASSERT_OK_AND_ASSIGN(bool exists, file->Exists(index)); + ASSERT_TRUE(exists); +} + +TEST_F(RealtimeDeduplicateDvMaterializerTest, TestSameFileNameInDifferentBucketsIsIsolated) { + ASSERT_OK_AND_ASSIGN(std::shared_ptr file0, CreateDvFile(0)); + ASSERT_OK_AND_ASSIGN(std::shared_ptr file1, CreateDvFile(1)); + ASSERT_OK_AND_ASSIGN(std::shared_ptr index0, + file0->WriteSingleFile({{"same-name", MakeDv({0})}})); + ASSERT_OK_AND_ASSIGN(std::shared_ptr index1, + file1->WriteSingleFile({{"same-name", MakeDv({2, 4})}})); + for (int32_t bucket = 0; bucket < 2; ++bucket) { + Materializer::BucketSnapshot snapshot; + snapshot.dv_index_files = {bucket == 0 ? index0 : index1}; + Materializer::BucketUpdate update; + if (bucket == 1) { + update.final_file_names = {"same-name"}; + } + ASSERT_OK_AND_ASSIGN(Materializer::DeletionVectors baseline, + materializer_->ReadDeletionVectors(BinaryRow(0), bucket, snapshot)); + CommitChanges changes; + ASSERT_OK(materializer_->MaterializeBucket(BinaryRow(0), bucket, snapshot, + std::move(baseline), update, {}, &changes)); + ASSERT_EQ(bucket == 0 ? 1u : 0u, changes.index_entries.size()); + if (bucket == 0) { + ASSERT_EQ(0, changes.index_entries[0].bucket); + ASSERT_EQ(index0, changes.index_entries[0].index_file); + } + } + ASSERT_OK_AND_ASSIGN(Materializer::DeletionVectors untouched, + file1->ReadAllDeletionVectors(index1)); + ASSERT_OK_AND_ASSIGN(int64_t count, untouched.at("same-name")->GetCardinality()); + ASSERT_EQ(2, count); +} + +TEST_F(RealtimeDeduplicateDvMaterializerTest, TestProgressOnlyProviderNeedsNoSnapshotScan) { + auto index = std::make_shared("HASH", "unrelated-index", 10, 1, std::nullopt, + std::nullopt); + ManifestEntryChanges input(1, /*drop_delete_file_stats=*/false); + input.append_index_files.emplace_back(FileKind::Add(), BinaryRow(0), 0, index); + const std::map offsets = { + {RealtimePartitionBucket({}, 0), RoaringBitmap64()}}; + RealtimeDeduplicateCommitChangesProvider provider(std::move(input), offsets, table_schema_, + paths_, /*scanner=*/nullptr, options_, + GetDefaultPool()); + ASSERT_OK_AND_ASSIGN(std::shared_ptr changes, provider.Provide(std::nullopt)); + ASSERT_TRUE(changes->delta_files.empty()); + ASSERT_EQ(1, changes->index_entries.size()); + ASSERT_EQ(index, changes->index_entries[0].index_file); + ASSERT_TRUE(changes->temporary_files.empty()); +} + +TEST_F(RealtimeDeduplicateDvMaterializerTest, TestDeletionWithoutMatchingFilesNeedsNoDv) { + RoaringBitmap64 offsets; + offsets.Add(42); + RealtimeDeduplicateCommitChangesProvider provider( + ManifestEntryChanges(1, /*drop_delete_file_stats=*/false), + {{RealtimePartitionBucket({}, 0), offsets}}, table_schema_, paths_, + /*scanner=*/nullptr, options_, GetDefaultPool()); + ASSERT_OK_AND_ASSIGN(std::shared_ptr changes, provider.Provide(std::nullopt)); + ASSERT_TRUE(changes->index_entries.empty()); + ASSERT_TRUE(changes->temporary_files.empty()); +} + +TEST_F(RealtimeDeduplicateDvMaterializerTest, TestWriteProviderRejectsCompactionChanges) { + auto index = + std::make_shared("HASH", "compact-index", 10, 1, std::nullopt, std::nullopt); + ManifestEntryChanges input(1, /*drop_delete_file_stats=*/false); + input.compact_index_files.emplace_back(FileKind::Add(), BinaryRow(0), 0, index); + RealtimeDeduplicateCommitChangesProvider provider(std::move(input), {}, table_schema_, paths_, + /*scanner=*/nullptr, options_, + GetDefaultPool()); + ASSERT_NOK_WITH_MSG(provider.Provide(std::nullopt), + "must not contain physical compaction changes"); +} + +} // namespace paimon::test diff --git a/src/paimon/core/operation/file_store_commit_impl.cpp b/src/paimon/core/operation/file_store_commit_impl.cpp index 4d6422c04..69a52b0d2 100644 --- a/src/paimon/core/operation/file_store_commit_impl.cpp +++ b/src/paimon/core/operation/file_store_commit_impl.cpp @@ -46,8 +46,6 @@ #include "paimon/common/utils/path_util.h" #include "paimon/common/utils/scope_guard.h" #include "paimon/core/catalog/snapshot_commit.h" -#include "paimon/core/deletionvectors/bucketed_dv_maintainer.h" -#include "paimon/core/deletionvectors/deletion_vectors_index_file.h" #include "paimon/core/index/index_file_meta.h" #include "paimon/core/io/compact_increment.h" #include "paimon/core/io/data_file_meta.h" @@ -67,6 +65,8 @@ #include "paimon/core/operation/commit/compacted_changelog_path_resolver.h" #include "paimon/core/operation/commit/conflict_detection.h" #include "paimon/core/operation/commit/realtime_commit_properties.h" +#include "paimon/core/operation/commit/realtime_deduplicate_commit_changes_provider.h" +#include "paimon/core/operation/commit/realtime_deduplicate_compaction_changes_provider.h" #include "paimon/core/operation/commit/row_tracking_commit_utils.h" #include "paimon/core/operation/commit/sequence_snapshot_properties.h" #include "paimon/core/operation/expire_snapshots.h" @@ -74,7 +74,6 @@ #include "paimon/core/operation/metrics/commit_metrics.h" #include "paimon/core/operation/metrics/commit_stats.h" #include "paimon/core/partition/partition_statistics.h" -#include "paimon/core/realtime/realtime_offset_file_index_lookup.h" #include "paimon/core/schema/schema_manager.h" #include "paimon/core/schema/table_schema.h" #include "paimon/core/table/bucket_mode.h" @@ -98,23 +97,6 @@ namespace { constexpr const char* kCommitStrictModeLastSafeSnapshot = "commit.strict-mode.last-safe-snapshot"; constexpr const char* kSequenceSnapshotOrdering = "sequence.snapshot-ordering"; -class RealtimeCommitChangesProvider final : public CommitChangesProvider { - public: - using Materializer = - std::function>(const std::optional&)>; - - explicit RealtimeCommitChangesProvider(Materializer materializer) - : materializer_(std::move(materializer)) {} - - Result> Provide( - const std::optional& latest_snapshot) const override { - return materializer_(latest_snapshot); - } - - private: - Materializer materializer_; -}; - } // namespace Status FileStoreCommitImpl::ValidateCommitOptions(const CoreOptions& options) { @@ -934,9 +916,20 @@ Status FileStoreCommitImpl::Commit( } if (changes.HasCompactChanges()) { + std::shared_ptr provider; + if (!options_.GetRealtimeDeduplicateKeyFields().empty()) { + PAIMON_ASSIGN_OR_RAISE(RealtimeCompactionPlan plan, + RealtimeCompactionPlan::Create(changes.compact_table_files)); + provider = std::make_shared( + std::move(plan), changes.compact_changelog, changes.compact_index_files, + table_schema_, path_factory_, commit_scanner_, options_, executor_, memory_pool_); + } else { + provider = CommitChangesProvider::Provider(changes.compact_table_files, + changes.compact_changelog, + changes.compact_index_files); + } PAIMON_ASSIGN_OR_RAISE(TryCommitResult result, - TryCommit(changes.compact_table_files, changes.compact_changelog, - changes.compact_index_files, committable->Identifier(), + TryCommit(provider, committable->Identifier(), committable->Watermark(), committable->Properties(), /*realtime_ranges=*/{}, Snapshot::CommitKind::Compact(), /*reset_all_realtime_progress=*/false, @@ -1031,14 +1024,9 @@ Result FileStoreCommitImpl::CommitWithProgress( /*retry_on_conflict=*/true, realtime_ranges)); } else { PAIMON_ASSIGN_OR_RAISE(ManifestEntryChanges changes, CollectChanges(commit_messages)); - if (changes.HasCompactChanges()) { - return Status::Invalid( - "realtime deduplicate progress must not contain physical compaction changes"); - } - auto provider = std::make_shared( - [this, changes, deleted_offsets](const std::optional& snapshot) { - return MaterializeRealtimeDeletions(changes, deleted_offsets, snapshot); - }); + auto provider = std::make_shared( + changes, deleted_offsets, table_schema_, path_factory_, commit_scanner_, options_, + memory_pool_); const bool has_deletions = std::any_of(deleted_offsets.begin(), deleted_offsets.end(), [](const auto& entry) { return !entry.second.IsEmpty(); }); @@ -1066,120 +1054,6 @@ Result FileStoreCommitImpl::CommitWithProgress( return last_committed_snapshot_id_; } -Result> FileStoreCommitImpl::MaterializeRealtimeDeletions( - const ManifestEntryChanges& changes, - const std::map& deleted_offsets, - const std::optional& latest_snapshot) const { - auto result = std::make_shared( - changes.append_table_files, changes.append_changelog, changes.append_index_files); - // Nothing generated here can be referenced by a snapshot until Provide succeeds. - ScopeGuard cleanup([&]() { - for (const std::string& path : result->temporary_files) { - [[maybe_unused]] Status status = fs_->Delete(path, /*recursive=*/false); - } - }); - std::vector> partitions; - for (const auto& [partition_bucket, offsets] : deleted_offsets) { - if (!offsets.IsEmpty()) { - partitions.push_back(partition_bucket.partition); - } - } - if (partitions.empty()) { - cleanup.Release(); - return result; - } - std::vector committed_files; - std::vector committed_indexes; - if (latest_snapshot) { - PAIMON_ASSIGN_OR_RAISE(committed_files, commit_scanner_->ReadAllEntriesFromPartitions( - latest_snapshot.value(), partitions)); - PAIMON_ASSIGN_OR_RAISE(committed_indexes, - commit_scanner_->ReadAllIndexEntriesFromPartitions( - latest_snapshot.value(), partitions)); - } - for (const auto& [partition_bucket, offsets] : deleted_offsets) { - if (offsets.IsEmpty()) { - continue; - } - PAIMON_ASSIGN_OR_RAISE(BinaryRow partition, - path_factory_->ToBinaryRow(partition_bucket.partition)); - const int32_t bucket = partition_bucket.bucket; - std::map> active_files; - for (const ManifestEntry& entry : committed_files) { - if (entry.Partition() == partition && entry.Bucket() == bucket) { - active_files[entry.File()->file_name] = entry.File(); - } - } - // All earlier prepared-but-uncommitted progress is supplied with this commit. Include - // its new files as well, so handoff and multiple prepares need no writer-local directory. - for (const ManifestEntry& entry : result->delta_files) { - if (!(entry.Partition() == partition) || entry.Bucket() != bucket) { - continue; - } - if (entry.Kind() == FileKind::Add()) { - active_files[entry.File()->file_name] = entry.File(); - } else { - active_files.erase(entry.File()->file_name); - } - } - std::vector> data_files; - for (const auto& [_, file] : active_files) { - data_files.push_back(file); - } - PAIMON_ASSIGN_OR_RAISE(std::shared_ptr data_path_factory, - path_factory_->CreateDataFilePathFactory(partition, bucket)); - PAIMON_ASSIGN_OR_RAISE( - std::shared_ptr lookup, - RealtimeOffsetFileIndexLookup::Create( - schema_, table_schema_->Id(), - schema_->GetFieldByName(options_.GetRealtimeDeduplicateKeyFields().front()), - data_files, data_path_factory, fs_, memory_pool_, options_)); - using FilePositions = std::map; - PAIMON_ASSIGN_OR_RAISE(FilePositions positions, lookup->LookupFilePositions(offsets)); - if (positions.empty()) { - continue; - } - PAIMON_ASSIGN_OR_RAISE(std::shared_ptr index_path_factory, - path_factory_->CreateIndexFileFactory(partition, bucket)); - auto dv_file = std::make_shared(fs_, index_path_factory, - /*bitmap64=*/false, memory_pool_); - std::vector> baseline_files; - for (const IndexManifestEntry& entry : committed_indexes) { - if (entry.partition == partition && entry.bucket == bucket && - entry.index_file->IndexType() == DeletionVectorsIndexFile::DELETION_VECTORS_INDEX) { - baseline_files.push_back(entry.index_file); - } - } - using DeletionVectors = std::map>; - PAIMON_ASSIGN_OR_RAISE(DeletionVectors baseline, - dv_file->ReadAllDeletionVectors(baseline_files)); - // Compaction can retire old files and move their rows. Never carry old physical row - // positions across snapshots: resolve logical offsets in the current active files above. - for (auto iter = baseline.begin(); iter != baseline.end();) { - if (active_files.count(iter->first) == 0) { - iter = baseline.erase(iter); - } else { - ++iter; - } - } - BucketedDvMaintainer maintainer(dv_file, baseline); - for (const auto& [file_name, bitmap] : positions) { - for (RoaringBitmap32::Iterator position = bitmap.Begin(); position != bitmap.End(); - ++position) { - PAIMON_RETURN_NOT_OK(maintainer.NotifyNewDeletion(file_name, *position)); - } - } - PAIMON_ASSIGN_OR_RAISE(std::optional> index, - maintainer.WriteDeletionVectorsIndex()); - if (index) { - result->temporary_files.push_back(dv_file->Path(index.value())); - result->index_entries.emplace_back(FileKind::Add(), partition, bucket, index.value()); - } - } - cleanup.Release(); - return result; -} - Result FileStoreCommitImpl::TryCommit( const std::vector& delta_files, const std::vector& changelog_files, diff --git a/src/paimon/core/operation/file_store_commit_impl.h b/src/paimon/core/operation/file_store_commit_impl.h index 65d6c78e4..94f00bef0 100644 --- a/src/paimon/core/operation/file_store_commit_impl.h +++ b/src/paimon/core/operation/file_store_commit_impl.h @@ -70,8 +70,6 @@ class SchemaManager; class TableSchema; class BinaryRowPartitionComputer; class CommitChangesProvider; -struct CommitChanges; -class RoaringBitmap64; class CommitMessage; class Executor; class FileSystem; @@ -206,11 +204,6 @@ class FileStoreCommitImpl : public FileStoreCommit { void ReportCommit(const ManifestEntryChanges& changes, int64_t commit_duration, int32_t generated_snapshot, int32_t attempt); - Result> MaterializeRealtimeDeletions( - const ManifestEntryChanges& changes, - const std::map& deleted_offsets, - const std::optional& latest_snapshot) const; - Result TryCommit( const std::vector& delta_files, const std::vector& changelog_files, diff --git a/src/paimon/core/table/table.cpp b/src/paimon/core/table/table.cpp index bc7085521..b1c55c517 100644 --- a/src/paimon/core/table/table.cpp +++ b/src/paimon/core/table/table.cpp @@ -378,7 +378,11 @@ Result> FileStoreWrite::Create(std::unique_ptrGetWriteSchema().empty() || ignore_previous_files || options.WriteOnly()) { + return Status::Invalid( + "deduplicate compaction requires the full schema, restored files and " + "write-only=false"); + } } std::shared_ptr write_schema = arrow_schema; const auto& write_field_names = ctx->GetWriteSchema(); diff --git a/test/inte/realtime_write_inte_test.cpp b/test/inte/realtime_write_inte_test.cpp index 5de6c9fa7..6f018bc59 100644 --- a/test/inte/realtime_write_inte_test.cpp +++ b/test/inte/realtime_write_inte_test.cpp @@ -29,6 +29,7 @@ #include #include #include +#include #include #include #include @@ -52,7 +53,9 @@ #include "paimon/core/core_options.h" #include "paimon/core/deletionvectors/bitmap_deletion_vector.h" #include "paimon/core/deletionvectors/deletion_vectors_index_file.h" +#include "paimon/core/operation/commit/manifest_entry_changes.h" #include "paimon/core/operation/commit/realtime_commit_properties.h" +#include "paimon/core/operation/commit/realtime_deduplicate_compaction_changes_provider.h" #include "paimon/core/operation/metrics/commit_metrics.h" #include "paimon/core/realtime/realtime_context_impl.h" #include "paimon/core/realtime/realtime_schema_layout.h" @@ -739,6 +742,20 @@ class RealtimeWriteInteTest : public ::testing::Test { /*watermark=*/std::nullopt); } + Result WriteCommitAndRefresh(const std::vector& rows, bool partitioned, + int32_t bucket, + const std::vector& row_kinds, + int64_t commit_identifier, FileStoreWrite* writer) const { + PAIMON_ASSIGN_OR_RAISE(std::unique_ptr batch, + MakeBatch(rows, partitioned, bucket, row_kinds)); + PAIMON_RETURN_NOT_OK(writer->Write(std::move(batch))); + PAIMON_ASSIGN_OR_RAISE(std::vector progress, + writer->PrepareCommitWithProgress(commit_identifier)); + PAIMON_ASSIGN_OR_RAISE(int64_t snapshot, Commit(progress, commit_identifier)); + PAIMON_RETURN_NOT_OK(writer->RefreshCommittedSnapshot(snapshot)); + return snapshot; + } + Result DropPartition(const std::map& partition, int64_t commit_identifier) const { CommitContextBuilder builder(table_path_, kDropPartitionCommitUser); @@ -794,6 +811,39 @@ class RealtimeWriteInteTest : public ::testing::Test { return latest_snapshot->Id(); } + Result>> CompactAndPrepareCommit( + const std::vector& partition_buckets, int64_t commit_identifier, + bool full_compaction, std::unique_ptr* compactor) const { + WriteContextBuilder builder(table_path_, "regression-compactor"); + PAIMON_ASSIGN_OR_RAISE(std::unique_ptr context, + builder.SetOptions(options_).WithStreamingMode(true).Finish()); + PAIMON_ASSIGN_OR_RAISE(std::unique_ptr writer, + FileStoreWrite::Create(std::move(context))); + for (const RealtimePartitionBucket& scope : partition_buckets) { + PAIMON_RETURN_NOT_OK(writer->Compact(scope.partition, scope.bucket, full_compaction)); + } + PAIMON_ASSIGN_OR_RAISE(std::vector> messages, + writer->PrepareCommit(/*wait_compaction=*/true, commit_identifier)); + *compactor = std::move(writer); + return messages; + } + + Result CommitCompaction(const std::vector>& messages, + int64_t commit_identifier) const { + CommitContextBuilder builder(table_path_, "regression-compactor"); + PAIMON_ASSIGN_OR_RAISE(std::unique_ptr context, + builder.SetOptions(options_).Finish()); + PAIMON_ASSIGN_OR_RAISE(std::unique_ptr commit, + FileStoreCommit::Create(std::move(context))); + PAIMON_RETURN_NOT_OK(commit->Commit(messages, commit_identifier)); + SnapshotManager snapshot_manager(dir_->GetFileSystem(), table_path_); + PAIMON_ASSIGN_OR_RAISE(std::optional snapshot, snapshot_manager.LatestSnapshot()); + if (!snapshot) { + return Status::Invalid("compaction did not produce a snapshot"); + } + return *snapshot; + } + Result CompactAndCommit(const std::map& partition, int32_t bucket, int64_t commit_identifier, bool full_compaction = true) const { @@ -829,6 +879,16 @@ class RealtimeWriteInteTest : public ::testing::Test { return compact_snapshot.value(); } + Result CompactCommitAndRefresh(const std::map& partition, + int32_t bucket, int64_t commit_identifier, + bool full_compaction, FileStoreWrite* writer) const { + PAIMON_ASSIGN_OR_RAISE( + Snapshot snapshot, + CompactAndCommit(partition, bucket, commit_identifier, full_compaction)); + PAIMON_RETURN_NOT_OK(writer->RefreshCommittedSnapshot(snapshot.Id())); + return snapshot; + } + Result ExpireSnapshots() const { CommitContextBuilder builder(table_path_, commit_user_); PAIMON_ASSIGN_OR_RAISE(std::unique_ptr context, @@ -1134,6 +1194,39 @@ class RealtimeWriteInteTest : public ::testing::Test { return Status::OK(); } + struct FileState { + std::shared_ptr file; + std::string bucket_path; + std::optional deletion_file; + std::set deleted_positions; + }; + + Result> InspectFiles() const { + PAIMON_ASSIGN_OR_RAISE(std::shared_ptr plan, + CreatePlan(/*realtime_context=*/nullptr, /*predicate=*/nullptr)); + std::vector states; + for (const auto& split : plan->Splits()) { + auto data = std::dynamic_pointer_cast(split); + if (!data || (!data->DeletionFiles().empty() && + data->DeletionFiles().size() != data->DataFiles().size())) { + return Status::Invalid("expected data files with matching deletion files"); + } + for (size_t i = 0; i < data->DataFiles().size(); ++i) { + FileState state{data->DataFiles()[i], data->BucketPath(), std::nullopt, {}}; + if (!data->DeletionFiles().empty() && data->DeletionFiles()[i]) { + state.deletion_file = data->DeletionFiles()[i]; + PAIMON_ASSIGN_OR_RAISE(PAIMON_UNIQUE_PTR dv, + DeletionVector::Read(dir_->GetFileSystem().get(), + *state.deletion_file, pool_.get())); + PAIMON_RETURN_NOT_OK(dv->ForEachDeletedPosition( + [&](int64_t position) { state.deleted_positions.insert(position); })); + } + states.push_back(std::move(state)); + } + } + return states; + } + Result ReadCommittedOffsets() const { PAIMON_ASSIGN_OR_RAISE(CoreOptions options, CoreOptions::FromMap(options_)); SnapshotManager snapshot_manager(options.GetFileSystem(), table_path_); @@ -1536,8 +1629,7 @@ TEST_F(RealtimeWriteInteTest, TestRealtimeDeduplicateCommitRetryMergesLatestDv) ASSERT_OK(writer->Close()); } -// TODO(xinyu.lxy): Enable when realtime deduplicate compaction is supported. -TEST_F(RealtimeWriteInteTest, DISABLED_TestRealtimeDeduplicateCommitRetryRemapsCompactedOffsets) { +TEST_F(RealtimeWriteInteTest, TestRealtimeDeduplicateCommitRetryRemapsCompactedOffsets) { ConfigureRealtimeDeduplicateTable(); ASSERT_OK_AND_ASSIGN(std::shared_ptr realtime_context, RealtimeContext::Create()); @@ -1564,8 +1656,9 @@ TEST_F(RealtimeWriteInteTest, DISABLED_TestRealtimeDeduplicateCommitRetryRemapsC bool interleaved = false; auto fs = std::make_shared( PathUtil::JoinPath(table_path_, "snapshot/snapshot-3"), [&]() -> Status { - PAIMON_ASSIGN_OR_RAISE(Snapshot compacted, - CompactAndCommit({}, /*bucket=*/0, /*commit_identifier=*/1)); + PAIMON_ASSIGN_OR_RAISE( + Snapshot compacted, + CompactAndCommit(/*partition=*/{}, /*bucket=*/0, /*commit_identifier=*/1)); if (compacted.Id() != 3 || compacted.GetCommitKind() != Snapshot::CommitKind::Compact()) { return Status::Invalid("expected competing compaction snapshot"); @@ -1592,6 +1685,972 @@ TEST_F(RealtimeWriteInteTest, DISABLED_TestRealtimeDeduplicateCommitRetryRemapsC ASSERT_OK(writer->Close()); } +class RealtimeDeduplicateCompactionInteTest + : public RealtimeWriteInteTest, + public ::testing::WithParamInterface> {}; + +TEST_P(RealtimeDeduplicateCompactionInteTest, TestCommitTransfersLatestDvToCompactedFiles) { + const auto [force_retry, prepare_between_rounds] = GetParam(); + options_[Options::COMPACTION_FORCE_REWRITE_ALL_FILES] = "true"; + ConfigureRealtimeDeduplicateTable(); + ASSERT_OK_AND_ASSIGN(std::shared_ptr realtime_context, + RealtimeContext::Create()); + ASSERT_OK_AND_ASSIGN(std::unique_ptr writer, + CreateRealtimeWriter(realtime_context)); + // Establish D0 before compaction: offset 0 must not appear in either rewrite's output. + const std::vector> batches = { + {{1, "one", "p0"}, {2, "two", "p0"}, {3, "three", "p0"}}, + {{4, "four", "p0"}, {5, "five", "p0"}}, + {{1, "ignored", "p0"}}}; + for (int64_t identifier = 0; identifier < 3; ++identifier) { + const std::vector kinds = + identifier == 2 ? std::vector{RecordBatch::RowKind::DELETE} + : std::vector{}; + ASSERT_OK_AND_ASSIGN(std::unique_ptr batch, + MakeBatch(batches[identifier], /*partitioned=*/false, /*bucket=*/0, + /*row_kinds=*/kinds)); + ASSERT_OK(writer->Write(std::move(batch))); + ASSERT_OK_AND_ASSIGN(std::vector progress, + writer->PrepareCommitWithProgress(/*commit_identifier=*/identifier)); + ASSERT_OK_AND_ASSIGN(int64_t snapshot, Commit(progress, /*commit_identifier=*/identifier)); + ASSERT_OK(writer->RefreshCommittedSnapshot(snapshot)); + } + + WriteContextBuilder write_builder(table_path_, "compactor"); + ASSERT_OK_AND_ASSIGN(std::unique_ptr write_context, + write_builder.SetOptions(options_).WithStreamingMode(true).Finish()); + ASSERT_OK_AND_ASSIGN(std::unique_ptr compactor, + FileStoreWrite::Create(std::move(write_context))); + ASSERT_OK_AND_ASSIGN(std::unique_ptr forbidden_batch, + MakeBatch({{99, "not-written", "p0"}}, /*partitioned=*/false)); + ASSERT_NOK_WITH_MSG(compactor->Write(std::move(forbidden_batch)), "only supports compaction"); + ResetExternalOffset(/*partition=*/{}, /*bucket=*/0, /*next_offset=*/6); + // F1/F2 -> F3 -> F4, with no snapshot publication between the two rewrites. + std::vector> messages; + for (int32_t round = 0; round < 2; ++round) { + ASSERT_OK(compactor->Compact(/*partition=*/{}, /*bucket=*/0, /*full_compaction=*/true)); + if (prepare_between_rounds || round == 1) { + ASSERT_OK_AND_ASSIGN( + std::vector> prepared, + compactor->PrepareCommit(/*wait_compaction=*/true, /*commit_identifier=*/0)); + messages.insert(messages.end(), prepared.begin(), prepared.end()); + } + } + ManifestEntryChanges changes(1, /*drop_delete_file_stats=*/true); + for (const auto& message : messages) { + ASSERT_OK(changes.Collect(message)); + } + ASSERT_OK_AND_ASSIGN(RealtimeCompactionPlan compact_plan, + RealtimeCompactionPlan::Create(changes.compact_table_files)); + ASSERT_EQ(1, compact_plan.intermediate_files.size()); + std::shared_ptr final_output; + for (const ManifestEntry& entry : compact_plan.delta_files) { + if (entry.Kind() == FileKind::Add()) { + ASSERT_EQ(nullptr, final_output); + final_output = entry.File(); + } + } + ASSERT_NE(nullptr, final_output); + ASSERT_EQ(4, final_output->row_count); + ASSERT_FALSE(final_output->extra_files.empty()); + + // D1 is published against compact-before AFTER both rewrites have completed. + ASSERT_OK_AND_ASSIGN( + std::unique_ptr update_batch, + MakeBatch( + {{2, "updated", "p0"}, {3, "ignored", "p0"}}, /*partitioned=*/false, /*bucket=*/0, + /*row_kinds=*/{RecordBatch::RowKind::UPDATE_AFTER, RecordBatch::RowKind::DELETE})); + ASSERT_OK(writer->Write(std::move(update_batch))); + ASSERT_OK_AND_ASSIGN(std::vector update_progress, + writer->PrepareCommitWithProgress(/*commit_identifier=*/3)); + ASSERT_OK_AND_ASSIGN(int64_t update_snapshot, Commit(update_progress, /*commit_identifier=*/3)); + ASSERT_EQ(4, update_snapshot); + ASSERT_OK(writer->RefreshCommittedSnapshot(update_snapshot)); + + // D2 also deletes from a non-compacted file. Its DV must survive bucket replacement. + ASSERT_OK_AND_ASSIGN( + std::unique_ptr concurrent_batch, + MakeBatch( + {{2, "latest", "p0"}, {4, "ignored", "p0"}}, /*partitioned=*/false, /*bucket=*/0, + /*row_kinds=*/{RecordBatch::RowKind::UPDATE_AFTER, RecordBatch::RowKind::DELETE})); + ASSERT_OK(writer->Write(std::move(concurrent_batch))); + ASSERT_OK_AND_ASSIGN(std::vector concurrent_progress, + writer->PrepareCommitWithProgress(/*commit_identifier=*/4)); + bool interleaved = false; + auto publish_build = [&]() -> Status { + PAIMON_ASSIGN_OR_RAISE(int64_t snapshot, + Commit(concurrent_progress, /*commit_identifier=*/4)); + if (snapshot != 5) { + return Status::Invalid("expected interleaving build snapshot 5"); + } + interleaved = true; + return Status::OK(); + }; + std::shared_ptr fs = dir_->GetFileSystem(); + if (force_retry) { + fs = std::make_shared( + PathUtil::JoinPath(table_path_, "snapshot/snapshot-5"), publish_build); + } else { + ASSERT_OK(publish_build()); + } + CommitContextBuilder commit_builder(table_path_, "compactor"); + ASSERT_OK_AND_ASSIGN(std::unique_ptr commit_context, + commit_builder.SetOptions(options_).WithFileSystem(fs).Finish()); + ASSERT_OK_AND_ASSIGN(std::unique_ptr commit, + FileStoreCommit::Create(std::move(commit_context))); + ASSERT_OK(commit->Commit(messages, /*commit_identifier=*/0)); + ASSERT_TRUE(interleaved); + ASSERT_OK_AND_ASSIGN(uint64_t attempts, commit->GetCommitMetrics()->GetCounter( + CommitMetrics::LAST_COMMIT_ATTEMPTS)); + ASSERT_EQ(force_retry ? 2 : 1, attempts); + const std::vector expected = {{5, "five", "p0"}, {2, "latest", "p0"}}; + ASSERT_OK_AND_ASSIGN(std::vector disk_rows, ReadRows()); + ASSERT_EQ(expected, disk_rows); + ASSERT_OK_AND_ASSIGN(std::vector realtime_rows, ReadRows(realtime_context)); + ASSERT_EQ(expected, realtime_rows); + + // Prove migration, not an implicit re-rewrite: F4 is unchanged, with a physical DV {0,1,2}. + ASSERT_OK_AND_ASSIGN(std::shared_ptr plan, + CreatePlan(/*realtime_context=*/nullptr, /*predicate=*/nullptr)); + bool checked_output = false; + for (const auto& split : plan->Splits()) { + auto data = std::dynamic_pointer_cast(split); + ASSERT_NE(nullptr, data); + for (size_t i = 0; i < data->DataFiles().size(); ++i) { + if (data->DataFiles()[i]->file_name != final_output->file_name) { + continue; + } + checked_output = true; + ASSERT_EQ(4, data->DataFiles()[i]->row_count); + ASSERT_EQ(data->DataFiles().size(), data->DeletionFiles().size()); + ASSERT_TRUE(data->DeletionFiles()[i]); + ASSERT_OK_AND_ASSIGN( + PAIMON_UNIQUE_PTR dv, + DeletionVector::Read(fs.get(), *data->DeletionFiles()[i], pool_.get())); + ASSERT_OK_AND_ASSIGN(int64_t count, dv->GetCardinality()); + ASSERT_EQ(3, count); + for (int64_t position = 0; position < 4; ++position) { + ASSERT_OK_AND_ASSIGN(bool deleted, dv->IsDeleted(position)); + ASSERT_EQ(position < 3, deleted); + } + } + } + ASSERT_TRUE(checked_output); + ASSERT_OK(writer->RefreshCommittedSnapshot(6)); + ASSERT_OK_AND_ASSIGN(realtime_rows, ReadRows(realtime_context)); + ASSERT_EQ(expected, realtime_rows); + ASSERT_OK(writer->Close()); + ASSERT_OK(compactor->Close()); + + // Restart uses the newly compacted files' key->offset sidecars and keeps existing DVs. + ASSERT_OK_AND_ASSIGN(std::shared_ptr restarted_context, + RealtimeContext::Create()); + ASSERT_OK_AND_ASSIGN(writer, CreateRealtimeWriter(restarted_context)); + ASSERT_OK_AND_ASSIGN(realtime_rows, ReadRows(restarted_context)); + ASSERT_EQ(expected, realtime_rows); + ASSERT_OK_AND_ASSIGN(disk_rows, ReadRows()); + ASSERT_EQ(expected, disk_rows); + const std::vector expected_after_delete = {{2, "latest", "p0"}}; + ASSERT_OK_AND_ASSIGN(std::unique_ptr delete_batch, + MakeBatch({{5, "ignored", "p0"}}, /*partitioned=*/false, /*bucket=*/0, + /*row_kinds=*/{RecordBatch::RowKind::DELETE})); + ASSERT_OK(writer->Write(std::move(delete_batch))); + // The buffered deletion is visible in realtime, but is not durable yet. + ASSERT_OK_AND_ASSIGN(realtime_rows, ReadRows(restarted_context)); + ASSERT_EQ(expected_after_delete, realtime_rows); + ASSERT_OK_AND_ASSIGN(disk_rows, ReadRows()); + ASSERT_EQ(expected, disk_rows); + ASSERT_OK_AND_ASSIGN(std::vector delete_progress, + writer->PrepareCommitWithProgress(/*commit_identifier=*/5)); + ASSERT_OK_AND_ASSIGN(realtime_rows, ReadRows(restarted_context)); + ASSERT_EQ(expected_after_delete, realtime_rows); + ASSERT_OK_AND_ASSIGN(disk_rows, ReadRows()); + ASSERT_EQ(expected, disk_rows); + ASSERT_OK_AND_ASSIGN(int64_t snapshot, Commit(delete_progress, /*commit_identifier=*/5)); + ASSERT_EQ(7, snapshot); + // Publication must preserve realtime visibility even before the writer acknowledges it. + ASSERT_OK_AND_ASSIGN(realtime_rows, ReadRows(restarted_context)); + ASSERT_EQ(expected_after_delete, realtime_rows); + ASSERT_OK_AND_ASSIGN(disk_rows, ReadRows()); + ASSERT_EQ(expected_after_delete, disk_rows); + ASSERT_OK(writer->RefreshCommittedSnapshot(snapshot)); + ASSERT_OK_AND_ASSIGN(realtime_rows, ReadRows(restarted_context)); + ASSERT_EQ(expected_after_delete, realtime_rows); + ASSERT_OK_AND_ASSIGN(disk_rows, ReadRows()); + ASSERT_EQ(expected_after_delete, disk_rows); + ASSERT_OK(writer->Close()); +} + +INSTANTIATE_TEST_SUITE_P(Compaction, RealtimeDeduplicateCompactionInteTest, + ::testing::Combine(::testing::Bool(), ::testing::Bool())); + +TEST_F(RealtimeWriteInteTest, TestRealtimeDeduplicateCompactionRejectsReplacedInputs) { + ConfigureRealtimeDeduplicateTable(); + ASSERT_OK_AND_ASSIGN(std::unique_ptr writer, CreateRealtimeWriter()); + const std::vector expected = {{1, "one", "p0"}, {2, "two", "p0"}}; + ASSERT_OK_AND_ASSIGN(std::unique_ptr batch, + MakeBatch(expected, /*partitioned=*/false)); + ASSERT_OK(writer->Write(std::move(batch))); + ASSERT_OK_AND_ASSIGN(std::vector progress, + writer->PrepareCommitWithProgress(/*commit_identifier=*/0)); + ASSERT_OK_AND_ASSIGN(int64_t initial_snapshot, Commit(progress, /*commit_identifier=*/0)); + ASSERT_EQ(1, initial_snapshot); + ASSERT_OK(writer->Close()); + + WriteContextBuilder write_builder(table_path_, "stale-compactor"); + ASSERT_OK_AND_ASSIGN(std::unique_ptr write_context, + write_builder.SetOptions(options_).WithStreamingMode(true).Finish()); + ASSERT_OK_AND_ASSIGN(std::unique_ptr compactor, + FileStoreWrite::Create(std::move(write_context))); + ASSERT_OK(compactor->Compact(/*partition=*/{}, /*bucket=*/0, /*full_compaction=*/true)); + ASSERT_OK_AND_ASSIGN( + std::vector> messages, + compactor->PrepareCommit(/*wait_compaction=*/true, /*commit_identifier=*/0)); + ASSERT_EQ(1, messages.size()); + auto message = std::dynamic_pointer_cast(messages[0]); + ASSERT_NE(nullptr, message); + ASSERT_EQ(1, message->GetCompactIncrement().CompactAfter().size()); + + ASSERT_OK_AND_ASSIGN(Snapshot replacement, + CompactAndCommit(/*partition=*/{}, /*bucket=*/0, /*commit_identifier=*/1)); + ASSERT_EQ(2, replacement.Id()); + CommitContextBuilder commit_builder(table_path_, "stale-compactor"); + ASSERT_OK_AND_ASSIGN(std::unique_ptr commit_context, + commit_builder.SetOptions(options_).Finish()); + ASSERT_OK_AND_ASSIGN(std::unique_ptr commit, + FileStoreCommit::Create(std::move(commit_context))); + ASSERT_NOK_WITH_MSG(commit->Commit(messages, /*commit_identifier=*/0), + "input is no longer active"); + ASSERT_OK_AND_ASSIGN(std::vector rows, ReadRows()); + ASSERT_EQ(expected, rows); + // A known rejected commit can be aborted without damaging the winning compaction. + ASSERT_OK(commit->Abort(messages)); + for (const auto& file : message->GetCompactIncrement().CompactAfter()) { + const std::string bucket_path = PathUtil::JoinPath(table_path_, "bucket-0"); + ASSERT_OK_AND_ASSIGN(bool exists, dir_->GetFileSystem()->Exists( + PathUtil::JoinPath(bucket_path, file->file_name))); + ASSERT_FALSE(exists); + for (const auto& extra : file->extra_files) { + ASSERT_TRUE(extra); + ASSERT_OK_AND_ASSIGN( + exists, dir_->GetFileSystem()->Exists(PathUtil::JoinPath(bucket_path, *extra))); + ASSERT_FALSE(exists); + } + } + ASSERT_OK_AND_ASSIGN(rows, ReadRows()); + ASSERT_EQ(expected, rows); + ASSERT_OK(compactor->Close()); +} + +TEST_F(RealtimeWriteInteTest, TestRealtimeDeduplicateCompactionWithAllRowsDeleted) { + ConfigureRealtimeDeduplicateTable(); + ASSERT_OK_AND_ASSIGN(std::shared_ptr realtime_context, + RealtimeContext::Create()); + ASSERT_OK_AND_ASSIGN(std::unique_ptr writer, + CreateRealtimeWriter(realtime_context)); + const std::vector rows = {{1, "one", "p0"}, {2, "two", "p0"}}; + for (int64_t identifier = 0; identifier < 2; ++identifier) { + const std::vector expected_rows = identifier == 0 ? rows : std::vector{}; + const std::vector kinds = + identifier == 1 ? std::vector(2, RecordBatch::RowKind::DELETE) + : std::vector{}; + ASSERT_OK_AND_ASSIGN( + std::unique_ptr batch, + MakeBatch(rows, /*partitioned=*/false, /*bucket=*/0, /*row_kinds=*/kinds)); + ASSERT_OK(writer->Write(std::move(batch))); + // Check both the initial rows and their deletion before either is committed. + ASSERT_OK_AND_ASSIGN(std::vector realtime_rows, ReadRows(realtime_context)); + ASSERT_EQ(expected_rows, realtime_rows); + ASSERT_OK_AND_ASSIGN(std::vector progress, + writer->PrepareCommitWithProgress(/*commit_identifier=*/identifier)); + ASSERT_OK_AND_ASSIGN(int64_t snapshot, Commit(progress, /*commit_identifier=*/identifier)); + ASSERT_OK_AND_ASSIGN(std::vector disk_rows, ReadRows()); + ASSERT_EQ(expected_rows, disk_rows); + ASSERT_OK(writer->RefreshCommittedSnapshot(snapshot)); + ASSERT_OK_AND_ASSIGN(realtime_rows, ReadRows(realtime_context)); + ASSERT_EQ(expected_rows, realtime_rows); + } + ASSERT_OK_AND_ASSIGN(RealtimeOffsetMap offsets_before, ReadCommittedOffsets()); + ASSERT_OK_AND_ASSIGN(Snapshot compacted, + CompactAndCommit(/*partition=*/{}, /*bucket=*/0, /*commit_identifier=*/2)); + ASSERT_EQ(3, compacted.Id()); + ASSERT_EQ(Snapshot::CommitKind::Compact(), compacted.GetCommitKind()); + ASSERT_OK_AND_ASSIGN(std::shared_ptr plan, + CreatePlan(/*realtime_context=*/nullptr, /*predicate=*/nullptr)); + ASSERT_TRUE(plan->Splits().empty()); + // Empty split metadata alone is not enough: read through the disk and realtime paths. + ASSERT_OK_AND_ASSIGN(std::vector disk_rows, ReadRows(plan, nullptr)); + ASSERT_TRUE(disk_rows.empty()); + ASSERT_OK_AND_ASSIGN(std::vector realtime_rows, ReadRows(realtime_context)); + ASSERT_TRUE(realtime_rows.empty()); + ASSERT_OK_AND_ASSIGN(RealtimeOffsetMap offsets_after, ReadCommittedOffsets()); + ASSERT_EQ(offsets_before, offsets_after); + ASSERT_OK(writer->RefreshCommittedSnapshot(compacted.Id())); + ASSERT_OK_AND_ASSIGN(realtime_rows, ReadRows(realtime_context)); + ASSERT_TRUE(realtime_rows.empty()); + ASSERT_OK(writer->Close()); +} + +TEST_F(RealtimeWriteInteTest, TestDeletesAllRowsAfterCompactionRewrite) { + ConfigureRealtimeDeduplicateTable(); + ASSERT_OK_AND_ASSIGN(std::shared_ptr realtime_context, + RealtimeContext::Create()); + ASSERT_OK_AND_ASSIGN(std::unique_ptr writer, + CreateRealtimeWriter(realtime_context)); + const std::vector rows = {{1, "one", "p0"}, {2, "two", "p0"}, {3, "three", "p0"}}; + ASSERT_OK(WriteCommitAndRefresh(rows, /*partitioned=*/false, /*bucket=*/0, /*row_kinds=*/{}, + /*commit_identifier=*/0, writer.get())); + std::unique_ptr compactor; + ASSERT_OK_AND_ASSIGN( + std::vector> messages, + CompactAndPrepareCommit({RealtimePartitionBucket(/*partition=*/{}, /*bucket=*/0)}, + /*commit_identifier=*/0, /*full_compaction=*/true, &compactor)); + ASSERT_EQ(1, messages.size()); + auto message = std::dynamic_pointer_cast(messages[0]); + ASSERT_NE(nullptr, message); + ASSERT_EQ(1, message->GetCompactIncrement().CompactAfter().size()); + const auto& output = message->GetCompactIncrement().CompactAfter()[0]; + ASSERT_EQ(3, output->row_count); + + // D0 is empty; D1 deletes every row only after a real output has been written. + ASSERT_OK(WriteCommitAndRefresh( + rows, /*partitioned=*/false, /*bucket=*/0, + /*row_kinds=*/std::vector(3, RecordBatch::RowKind::DELETE), + /*commit_identifier=*/1, writer.get())); + { + ASSERT_OK_AND_ASSIGN(std::vector disk_rows, ReadRows()); + ASSERT_TRUE(disk_rows.empty()); + ASSERT_OK_AND_ASSIGN(std::vector realtime_rows, ReadRows(realtime_context)); + ASSERT_TRUE(realtime_rows.empty()); + } + ASSERT_OK_AND_ASSIGN(RealtimeOffsetMap offsets, ReadCommittedOffsets()); + ASSERT_EQ(1, offsets.size()); + ASSERT_EQ(6, offsets.at(RealtimePartitionBucket(/*partition=*/{}, /*bucket=*/0))); + ASSERT_OK_AND_ASSIGN(Snapshot compacted, CommitCompaction(messages, /*commit_identifier=*/0)); + ASSERT_EQ(3, compacted.Id()); + ASSERT_OK_AND_ASSIGN(std::vector files, InspectFiles()); + ASSERT_EQ(1, files.size()); + ASSERT_EQ(output->file_name, files[0].file->file_name); + ASSERT_EQ(3, files[0].file->row_count); + ASSERT_TRUE(files[0].deletion_file); + ASSERT_EQ((std::set{0, 1, 2}), files[0].deleted_positions); + { + ASSERT_OK_AND_ASSIGN(std::vector disk_rows, ReadRows()); + ASSERT_TRUE(disk_rows.empty()); + ASSERT_OK_AND_ASSIGN(std::vector realtime_rows, ReadRows(realtime_context)); + ASSERT_TRUE(realtime_rows.empty()); + } + ASSERT_OK(writer->RefreshCommittedSnapshot(compacted.Id())); + { + ASSERT_OK_AND_ASSIGN(std::vector disk_rows, ReadRows()); + ASSERT_TRUE(disk_rows.empty()); + ASSERT_OK_AND_ASSIGN(std::vector realtime_rows, ReadRows(realtime_context)); + ASSERT_TRUE(realtime_rows.empty()); + } + ASSERT_OK_AND_ASSIGN(RealtimeOffsetMap after, ReadCommittedOffsets()); + ASSERT_EQ(offsets, after); + ASSERT_OK(compactor->Close()); + + // A later rewrite retires the fully deleted output. + ASSERT_OK_AND_ASSIGN( + Snapshot reclaimed, + CompactCommitAndRefresh(/*partition=*/{}, /*bucket=*/0, /*commit_identifier=*/2, + /*full_compaction=*/true, writer.get())); + ASSERT_EQ(4, reclaimed.Id()); + ASSERT_OK_AND_ASSIGN(files, InspectFiles()); + ASSERT_TRUE(files.empty()); + ASSERT_OK_AND_ASSIGN(after, ReadCommittedOffsets()); + ASSERT_EQ(offsets, after); + { + ASSERT_OK_AND_ASSIGN(std::vector disk_rows, ReadRows()); + ASSERT_TRUE(disk_rows.empty()); + ASSERT_OK_AND_ASSIGN(std::vector realtime_rows, ReadRows(realtime_context)); + ASSERT_TRUE(realtime_rows.empty()); + } + ASSERT_OK(writer->Close()); +} + +TEST_F(RealtimeWriteInteTest, TestSparseDeletedOffsetsAcrossReaderBatches) { + options_[Options::READ_BATCH_SIZE] = "3"; + // Keep holes in the raw batch so NextBatchWithBitmap must honor its sparse selection. + options_["parquet.read.enable-page-index-filter"] = "false"; + ConfigureRealtimeDeduplicateTable(); + ASSERT_OK_AND_ASSIGN(std::shared_ptr realtime_context, + RealtimeContext::Create()); + ASSERT_OK_AND_ASSIGN(std::unique_ptr writer, + CreateRealtimeWriter(realtime_context)); + ASSERT_OK(WriteCommitAndRefresh(MakeRows(/*first_id=*/0, /*count=*/9, /*partition=*/"p0"), + /*partitioned=*/false, /*bucket=*/0, + /*row_kinds=*/{}, /*commit_identifier=*/0, writer.get())); + ASSERT_OK(WriteCommitAndRefresh(MakeRows(/*first_id=*/9, /*count=*/9, /*partition=*/"p0"), + /*partitioned=*/false, /*bucket=*/0, + /*row_kinds=*/{}, /*commit_identifier=*/1, writer.get())); + std::unique_ptr compactor; + ASSERT_OK_AND_ASSIGN( + std::vector> messages, + CompactAndPrepareCommit({RealtimePartitionBucket(/*partition=*/{}, /*bucket=*/0)}, + /*commit_identifier=*/0, /*full_compaction=*/true, &compactor)); + ASSERT_EQ(1, messages.size()); + auto message = std::dynamic_pointer_cast(messages[0]); + ASSERT_NE(nullptr, message); + ASSERT_EQ(2, message->GetCompactIncrement().CompactBefore().size()); + const std::vector deleted = {{0, "ignored", "p0"}, {2, "ignored", "p0"}, + {7, "ignored", "p0"}, {10, "ignored", "p0"}, + {14, "ignored", "p0"}, {17, "ignored", "p0"}}; + ASSERT_OK(WriteCommitAndRefresh( + deleted, /*partitioned=*/false, /*bucket=*/0, /*row_kinds=*/ + std::vector(deleted.size(), RecordBatch::RowKind::DELETE), + /*commit_identifier=*/2, writer.get())); + std::vector expected; + const std::set positions = {0, 2, 7, 10, 14, 17}; + for (const Row& row : MakeRows(/*first_id=*/0, /*count=*/18, /*partition=*/"p0")) { + if (positions.count(std::get<0>(row)) == 0) { + expected.push_back(row); + } + } + const std::multiset expected_rows(expected.begin(), expected.end()); + { + ASSERT_OK_AND_ASSIGN(std::vector disk_rows, ReadRows()); + ASSERT_EQ(expected_rows, std::multiset(disk_rows.begin(), disk_rows.end())); + ASSERT_OK_AND_ASSIGN(std::vector realtime_rows, ReadRows(realtime_context)); + ASSERT_EQ(expected_rows, std::multiset(realtime_rows.begin(), realtime_rows.end())); + } + ASSERT_OK_AND_ASSIGN(Snapshot compacted, CommitCompaction(messages, /*commit_identifier=*/0)); + ASSERT_OK_AND_ASSIGN(std::vector files, InspectFiles()); + ASSERT_EQ(1, files.size()); + ASSERT_EQ(18, files[0].file->row_count); + ASSERT_EQ(positions, files[0].deleted_positions); + { + ASSERT_OK_AND_ASSIGN(std::vector disk_rows, ReadRows()); + ASSERT_EQ(expected_rows, std::multiset(disk_rows.begin(), disk_rows.end())); + ASSERT_OK_AND_ASSIGN(std::vector realtime_rows, ReadRows(realtime_context)); + ASSERT_EQ(expected_rows, std::multiset(realtime_rows.begin(), realtime_rows.end())); + } + ASSERT_OK(writer->RefreshCommittedSnapshot(compacted.Id())); + { + ASSERT_OK_AND_ASSIGN(std::vector disk_rows, ReadRows()); + ASSERT_EQ(expected_rows, std::multiset(disk_rows.begin(), disk_rows.end())); + ASSERT_OK_AND_ASSIGN(std::vector realtime_rows, ReadRows(realtime_context)); + ASSERT_EQ(expected_rows, std::multiset(realtime_rows.begin(), realtime_rows.end())); + } + ASSERT_OK(writer->Close()); + ASSERT_OK(compactor->Close()); +} + +TEST_F(RealtimeWriteInteTest, TestCompactionRollsAndRemapsMultipleOutputs) { + ConfigureRealtimeDeduplicateTable(); + ASSERT_OK_AND_ASSIGN(std::shared_ptr realtime_context, + RealtimeContext::Create()); + ASSERT_OK_AND_ASSIGN(std::unique_ptr writer, + CreateRealtimeWriter(realtime_context)); + ASSERT_OK(WriteCommitAndRefresh(MakeRows(/*first_id=*/0, /*count=*/1200, /*partition=*/"p0"), + /*partitioned=*/false, /*bucket=*/0, + /*row_kinds=*/{}, /*commit_identifier=*/0, writer.get())); + ASSERT_OK(WriteCommitAndRefresh(MakeRows(/*first_id=*/1200, /*count=*/1200, /*partition=*/"p0"), + /*partitioned=*/false, /*bucket=*/0, + /*row_kinds=*/{}, /*commit_identifier=*/1, writer.get())); + // Parquet checks target size every 1000 records, independently of target-file-row-num. + options_[Options::TARGET_FILE_SIZE] = "1"; + options_[Options::READ_BATCH_SIZE] = "100"; + options_[Options::COMPACTION_FORCE_REWRITE_ALL_FILES] = "true"; + std::unique_ptr compactor; + ASSERT_OK_AND_ASSIGN( + std::vector> messages, + CompactAndPrepareCommit({RealtimePartitionBucket(/*partition=*/{}, /*bucket=*/0)}, + /*commit_identifier=*/0, /*full_compaction=*/true, &compactor)); + ASSERT_EQ(1, messages.size()); + auto message = std::dynamic_pointer_cast(messages[0]); + ASSERT_NE(nullptr, message); + const auto& outputs = message->GetCompactIncrement().CompactAfter(); + ASSERT_GT(outputs.size(), 1); + // Pick one row from every physical output, including across the original file boundary. + std::vector deleted; + std::map> expected_dvs; + int64_t first_id = 0; + for (const auto& output : outputs) { + ASSERT_GT(output->row_count, 1); + deleted.emplace_back(first_id, "ignored", "p0"); + expected_dvs[output->file_name] = {0}; + first_id += output->row_count; + bool has_offset_sidecar = false; + for (const auto& extra : output->extra_files) { + ASSERT_TRUE(extra); + has_offset_sidecar |= + extra->size() >= 7 && extra->compare(extra->size() - 7, 7, ".offset") == 0; + ASSERT_OK_AND_ASSIGN(bool exists, + dir_->GetFileSystem()->Exists(PathUtil::JoinPath( + PathUtil::JoinPath(table_path_, "bucket-0"), *extra))); + ASSERT_TRUE(exists); + } + ASSERT_TRUE(has_offset_sidecar); + } + ASSERT_EQ(2400, first_id); + ASSERT_OK(WriteCommitAndRefresh( + deleted, /*partitioned=*/false, /*bucket=*/0, /*row_kinds=*/ + std::vector(deleted.size(), RecordBatch::RowKind::DELETE), + /*commit_identifier=*/2, writer.get())); + std::vector expected = MakeRows(/*first_id=*/0, /*count=*/2400, /*partition=*/"p0"); + expected.erase(std::remove_if(expected.begin(), expected.end(), + [&](const Row& row) { + return std::any_of( + deleted.begin(), deleted.end(), [&](const Row& deletion) { + return std::get<0>(row) == std::get<0>(deletion); + }); + }), + expected.end()); + ASSERT_OK_AND_ASSIGN(Snapshot compacted, CommitCompaction(messages, /*commit_identifier=*/0)); + ASSERT_OK_AND_ASSIGN(std::vector files, InspectFiles()); + ASSERT_EQ(outputs.size(), files.size()); + for (const FileState& file : files) { + ASSERT_EQ(1, expected_dvs.count(file.file->file_name)); + ASSERT_EQ(expected_dvs.at(file.file->file_name), file.deleted_positions); + } + std::multiset expected_rows(expected.begin(), expected.end()); + { + ASSERT_OK_AND_ASSIGN(std::vector disk_rows, ReadRows()); + ASSERT_EQ(expected_rows, std::multiset(disk_rows.begin(), disk_rows.end())); + ASSERT_OK_AND_ASSIGN(std::vector realtime_rows, ReadRows(realtime_context)); + ASSERT_EQ(expected_rows, std::multiset(realtime_rows.begin(), realtime_rows.end())); + } + ASSERT_OK(writer->RefreshCommittedSnapshot(compacted.Id())); + { + ASSERT_OK_AND_ASSIGN(std::vector disk_rows, ReadRows()); + ASSERT_EQ(expected_rows, std::multiset(disk_rows.begin(), disk_rows.end())); + ASSERT_OK_AND_ASSIGN(std::vector realtime_rows, ReadRows(realtime_context)); + ASSERT_EQ(expected_rows, std::multiset(realtime_rows.begin(), realtime_rows.end())); + } + ASSERT_OK(writer->Close()); + ASSERT_OK(compactor->Close()); + + // Restore every rolled key->offset sidecar and delete a surviving row from every output. + ASSERT_OK_AND_ASSIGN(realtime_context, RealtimeContext::Create()); + ASSERT_OK_AND_ASSIGN(writer, CreateRealtimeWriter(realtime_context)); + { + ASSERT_OK_AND_ASSIGN(std::vector disk_rows, ReadRows()); + ASSERT_EQ(expected_rows, std::multiset(disk_rows.begin(), disk_rows.end())); + ASSERT_OK_AND_ASSIGN(std::vector realtime_rows, ReadRows(realtime_context)); + ASSERT_EQ(expected_rows, std::multiset(realtime_rows.begin(), realtime_rows.end())); + } + std::vector restart_deletes; + for (const Row& row : deleted) { + restart_deletes.emplace_back(std::get<0>(row) + 1, "ignored", "p0"); + } + ASSERT_OK(WriteCommitAndRefresh( + restart_deletes, /*partitioned=*/false, /*bucket=*/0, /*row_kinds=*/ + std::vector(restart_deletes.size(), RecordBatch::RowKind::DELETE), + /*commit_identifier=*/3, writer.get())); + expected.erase(std::remove_if(expected.begin(), expected.end(), + [&](const Row& row) { + return std::any_of( + restart_deletes.begin(), restart_deletes.end(), + [&](const Row& deletion) { + return std::get<0>(row) == std::get<0>(deletion); + }); + }), + expected.end()); + expected_rows = std::multiset(expected.begin(), expected.end()); + { + ASSERT_OK_AND_ASSIGN(std::vector disk_rows, ReadRows()); + ASSERT_EQ(expected_rows, std::multiset(disk_rows.begin(), disk_rows.end())); + ASSERT_OK_AND_ASSIGN(std::vector realtime_rows, ReadRows(realtime_context)); + ASSERT_EQ(expected_rows, std::multiset(realtime_rows.begin(), realtime_rows.end())); + } + ASSERT_OK_AND_ASSIGN(files, InspectFiles()); + ASSERT_EQ(outputs.size(), files.size()); + for (const FileState& file : files) { + ASSERT_EQ(1, expected_dvs.count(file.file->file_name)); + ASSERT_EQ((std::set{0, 1}), file.deleted_positions); + } + ASSERT_OK(writer->Close()); +} + +TEST_F(RealtimeWriteInteTest, TestCompactionIsolatesPartitionsAndBuckets) { + options_[Options::BUCKET] = "2"; + ConfigureRealtimeDeduplicateTable(/*partition_keys=*/{"pt"}); + ASSERT_OK_AND_ASSIGN(std::shared_ptr realtime_context, + RealtimeContext::Create()); + ASSERT_OK_AND_ASSIGN(std::unique_ptr writer, + CreateRealtimeWriter(realtime_context)); + const std::vector> scopes = { + {"p0", 0}, {"p0", 1}, {"p1", 0}, {"p1", 1}}; + for (const auto& [partition, bucket] : scopes) { + const std::vector rows = { + {0, fmt::format("{}-b{}-zero", partition, bucket), partition}, + {1, fmt::format("{}-b{}-one", partition, bucket), partition}, + {2, fmt::format("{}-b{}-two", partition, bucket), partition}}; + ASSERT_OK_AND_ASSIGN(std::unique_ptr batch, + MakeBatch(rows, /*partitioned=*/true, /*bucket=*/bucket)); + ASSERT_OK(writer->Write(std::move(batch))); + } + ASSERT_OK_AND_ASSIGN(std::vector initial, + writer->PrepareCommitWithProgress(/*commit_identifier=*/0)); + ASSERT_EQ(4, initial.size()); + ASSERT_OK_AND_ASSIGN(int64_t snapshot, Commit(initial, /*commit_identifier=*/0)); + ASSERT_OK(writer->RefreshCommittedSnapshot(snapshot)); + // The fourth scope is not compacted. Its existing DV must survive the other replacements. + ASSERT_OK(WriteCommitAndRefresh({{0, "ignored", "p1"}}, /*partitioned=*/true, /*bucket=*/1, + /*row_kinds=*/{RecordBatch::RowKind::DELETE}, + /*commit_identifier=*/1, writer.get())); + ASSERT_OK_AND_ASSIGN(std::vector baseline, InspectFiles()); + std::optional untouched_dv; + std::string untouched_name; + for (const FileState& file : baseline) { + if (file.deletion_file) { + ASSERT_FALSE(untouched_dv); + untouched_dv = file.deletion_file; + untouched_name = file.file->file_name; + } + } + ASSERT_TRUE(untouched_dv); + std::vector compact_scopes; + for (size_t i = 0; i < 3; ++i) { + compact_scopes.emplace_back(std::map{{"pt", scopes[i].first}}, + scopes[i].second); + } + std::unique_ptr compactor; + ASSERT_OK_AND_ASSIGN(std::vector> messages, + CompactAndPrepareCommit(compact_scopes, /*commit_identifier=*/0, + /*full_compaction=*/true, &compactor)); + ASSERT_EQ(3, messages.size()); + std::map> expected_dvs; + for (const auto& raw_message : messages) { + auto message = std::dynamic_pointer_cast(raw_message); + ASSERT_NE(nullptr, message); + ASSERT_EQ(1, message->GetCompactIncrement().CompactAfter().size()); + const std::string& name = message->GetCompactIncrement().CompactAfter()[0]->file_name; + // Resolve the scope through the bucket path of the corresponding compact-before file. + ASSERT_EQ(1, message->GetCompactIncrement().CompactBefore().size()); + const std::string& before = message->GetCompactIncrement().CompactBefore()[0]->file_name; + bool found = false; + for (const FileState& file : baseline) { + if (file.file->file_name != before) { + continue; + } + for (size_t i = 0; i < 3; ++i) { + const std::string suffix = + fmt::format("pt={}/bucket-{}", scopes[i].first, scopes[i].second); + if (file.bucket_path == PathUtil::JoinPath(table_path_, suffix)) { + expected_dvs[name] = {static_cast(i)}; + found = true; + } + } + } + ASSERT_TRUE(found); + } + // Every scope starts with offsets {0,1,2}; different deletions expose cross-scope leakage. + for (size_t i = 0; i < 3; ++i) { + ASSERT_OK_AND_ASSIGN(std::unique_ptr batch, + MakeBatch({{static_cast(i), "ignored", scopes[i].first}}, + /*partitioned=*/true, /*bucket=*/scopes[i].second, + /*row_kinds=*/{RecordBatch::RowKind::DELETE})); + ASSERT_OK(writer->Write(std::move(batch))); + } + ASSERT_OK_AND_ASSIGN(std::vector deleted, + writer->PrepareCommitWithProgress(/*commit_identifier=*/2)); + ASSERT_EQ(3, deleted.size()); + ASSERT_OK_AND_ASSIGN(snapshot, Commit(deleted, /*commit_identifier=*/2)); + ASSERT_OK(writer->RefreshCommittedSnapshot(snapshot)); + ASSERT_OK_AND_ASSIGN(RealtimeOffsetMap offsets, ReadCommittedOffsets()); + ASSERT_EQ(4, offsets.size()); + for (const auto& [partition, bucket] : scopes) { + ASSERT_EQ(4, offsets.at(RealtimePartitionBucket(/*partition=*/{{"pt", partition}}, + /*bucket=*/bucket))); + } + const std::multiset expected_rows = { + // p0/b0: id 0 was deleted. + {1, "p0-b0-one", "p0"}, + {2, "p0-b0-two", "p0"}, + // p0/b1: id 1 was deleted. + {0, "p0-b1-zero", "p0"}, + {2, "p0-b1-two", "p0"}, + // p1/b0: id 2 was deleted. + {0, "p1-b0-zero", "p1"}, + {1, "p1-b0-one", "p1"}, + // p1/b1: id 0 was already deleted; this bucket is not compacted. + {1, "p1-b1-one", "p1"}, + {2, "p1-b1-two", "p1"}, + }; + { + ASSERT_OK_AND_ASSIGN(std::vector disk_rows, ReadRows()); + ASSERT_EQ(expected_rows, std::multiset(disk_rows.begin(), disk_rows.end())); + ASSERT_OK_AND_ASSIGN(std::vector realtime_rows, ReadRows(realtime_context)); + ASSERT_EQ(expected_rows, std::multiset(realtime_rows.begin(), realtime_rows.end())); + } + ASSERT_OK_AND_ASSIGN(Snapshot compacted, CommitCompaction(messages, /*commit_identifier=*/0)); + ASSERT_EQ(4, compacted.Id()); + ASSERT_OK_AND_ASSIGN(std::vector files, InspectFiles()); + ASSERT_EQ(4, files.size()); + size_t checked_outputs = 0; + bool checked_untouched = false; + for (const FileState& file : files) { + if (file.file->file_name == untouched_name) { + checked_untouched = true; + ASSERT_TRUE(file.deletion_file); + ASSERT_EQ(untouched_dv->path, file.deletion_file->path); + ASSERT_EQ(untouched_dv->offset, file.deletion_file->offset); + ASSERT_EQ((std::set{0}), file.deleted_positions); + } else { + ASSERT_EQ(1, expected_dvs.count(file.file->file_name)); + ASSERT_EQ(expected_dvs.at(file.file->file_name), file.deleted_positions); + ++checked_outputs; + } + } + ASSERT_TRUE(checked_untouched); + ASSERT_EQ(3, checked_outputs); + ASSERT_OK_AND_ASSIGN(RealtimeOffsetMap after, ReadCommittedOffsets()); + ASSERT_EQ(offsets, after); + { + ASSERT_OK_AND_ASSIGN(std::vector disk_rows, ReadRows()); + ASSERT_EQ(expected_rows, std::multiset(disk_rows.begin(), disk_rows.end())); + ASSERT_OK_AND_ASSIGN(std::vector realtime_rows, ReadRows(realtime_context)); + ASSERT_EQ(expected_rows, std::multiset(realtime_rows.begin(), realtime_rows.end())); + } + ASSERT_OK(writer->RefreshCommittedSnapshot(compacted.Id())); + { + ASSERT_OK_AND_ASSIGN(std::vector disk_rows, ReadRows()); + ASSERT_EQ(expected_rows, std::multiset(disk_rows.begin(), disk_rows.end())); + ASSERT_OK_AND_ASSIGN(std::vector realtime_rows, ReadRows(realtime_context)); + ASSERT_EQ(expected_rows, std::multiset(realtime_rows.begin(), realtime_rows.end())); + } + ASSERT_OK(writer->Close()); + ASSERT_OK(compactor->Close()); +} + +TEST_F(RealtimeWriteInteTest, TestNoHitPreservesBaselineDv) { + ConfigureRealtimeDeduplicateTable(); + ASSERT_OK_AND_ASSIGN(std::shared_ptr realtime_context, + RealtimeContext::Create()); + ASSERT_OK_AND_ASSIGN(std::unique_ptr writer, + CreateRealtimeWriter(realtime_context)); + ASSERT_OK(WriteCommitAndRefresh({{0, "zero", "p0"}, {1, "one", "p0"}, {2, "two", "p0"}}, + /*partitioned=*/false, /*bucket=*/0, /*row_kinds=*/{}, + /*commit_identifier=*/0, writer.get())); + ASSERT_OK(WriteCommitAndRefresh({{0, "ignored", "p0"}}, /*partitioned=*/false, /*bucket=*/0, + /*row_kinds=*/{RecordBatch::RowKind::DELETE}, + /*commit_identifier=*/1, writer.get())); + ASSERT_OK(CompactCommitAndRefresh(/*partition=*/{}, /*bucket=*/0, /*commit_identifier=*/2, + /*full_compaction=*/true, writer.get())); + // Offset 0 is now physically absent, while a different offset creates a nonempty baseline. + ASSERT_OK(WriteCommitAndRefresh({{1, "new-one", "p0"}}, /*partitioned=*/false, /*bucket=*/0, + /*row_kinds=*/{RecordBatch::RowKind::UPDATE_AFTER}, + /*commit_identifier=*/3, writer.get())); + ASSERT_OK_AND_ASSIGN(std::vector baseline, InspectFiles()); + SnapshotManager snapshot_manager(dir_->GetFileSystem(), table_path_); + ASSERT_OK_AND_ASSIGN(std::optional before, snapshot_manager.LatestSnapshot()); + ASSERT_TRUE(before); + ASSERT_TRUE(before->IndexManifest()); + std::set indexes_before; + { + std::vector index_files; + ASSERT_OK( + dir_->GetFileSystem()->ListDir(PathUtil::JoinPath(table_path_, "index"), &index_files)); + for (const auto& file : index_files) { + if (!file.IsDir()) { + indexes_before.insert(file.GetPath()); + } + } + } + ASSERT_OK_AND_ASSIGN(std::unique_ptr batch, + MakeBatch({{3, "three", "p0"}}, /*partitioned=*/false)); + ASSERT_OK(writer->Write(std::move(batch))); + ASSERT_OK_AND_ASSIGN(std::vector progress, + writer->PrepareCommitWithProgress(/*commit_identifier=*/4)); + ASSERT_EQ(1, progress.size()); + // Replay a deletion whose offset has already been removed by compaction. + auto no_hit = std::make_shared(); + no_hit->Add(0); + progress[0].deleted_offsets = no_hit; + CommitContextBuilder builder(table_path_, commit_user_); + ASSERT_OK_AND_ASSIGN(std::unique_ptr context, + builder.SetOptions(options_).Finish()); + ASSERT_OK_AND_ASSIGN(std::unique_ptr commit, + FileStoreCommit::Create(std::move(context))); + ASSERT_OK_AND_ASSIGN( + int64_t snapshot, + commit->CommitWithProgress(progress, /*commit_identifier=*/4, /*watermark=*/std::nullopt)); + ASSERT_EQ(5, snapshot); + ASSERT_OK_AND_ASSIGN(std::optional after, snapshot_manager.LatestSnapshot()); + ASSERT_TRUE(after); + ASSERT_EQ(before->IndexManifest(), after->IndexManifest()); + std::set indexes_after; + { + std::vector index_files; + ASSERT_OK( + dir_->GetFileSystem()->ListDir(PathUtil::JoinPath(table_path_, "index"), &index_files)); + for (const auto& file : index_files) { + if (!file.IsDir()) { + indexes_after.insert(file.GetPath()); + } + } + } + ASSERT_EQ(indexes_before, indexes_after); + ASSERT_OK_AND_ASSIGN(std::vector files, InspectFiles()); + size_t checked = 0; + for (const FileState& old : baseline) { + if (!old.deletion_file) { + continue; + } + for (const FileState& file : files) { + if (old.file->file_name == file.file->file_name) { + ASSERT_TRUE(file.deletion_file); + ASSERT_EQ(old.deletion_file->path, file.deletion_file->path); + ASSERT_EQ(old.deletion_file->offset, file.deletion_file->offset); + ASSERT_EQ(old.deletion_file->length, file.deletion_file->length); + ASSERT_EQ(old.deleted_positions, file.deleted_positions); + ++checked; + } + } + } + ASSERT_EQ(1, checked); + const std::vector expected = {{2, "two", "p0"}, {1, "new-one", "p0"}, {3, "three", "p0"}}; + const std::multiset expected_rows(expected.begin(), expected.end()); + { + ASSERT_OK_AND_ASSIGN(std::vector disk_rows, ReadRows()); + ASSERT_EQ(expected_rows, std::multiset(disk_rows.begin(), disk_rows.end())); + ASSERT_OK_AND_ASSIGN(std::vector realtime_rows, ReadRows(realtime_context)); + ASSERT_EQ(expected_rows, std::multiset(realtime_rows.begin(), realtime_rows.end())); + } + ASSERT_OK(writer->RefreshCommittedSnapshot(snapshot)); + { + ASSERT_OK_AND_ASSIGN(std::vector disk_rows, ReadRows()); + ASSERT_EQ(expected_rows, std::multiset(disk_rows.begin(), disk_rows.end())); + ASSERT_OK_AND_ASSIGN(std::vector realtime_rows, ReadRows(realtime_context)); + ASSERT_EQ(expected_rows, std::multiset(realtime_rows.begin(), realtime_rows.end())); + } + ASSERT_OK(writer->Close()); +} + +TEST_F(RealtimeWriteInteTest, TestPublicationConflictCleansOnlyAttemptDv) { + ConfigureRealtimeDeduplicateTable(); + ASSERT_OK_AND_ASSIGN(std::shared_ptr realtime_context, + RealtimeContext::Create()); + ASSERT_OK_AND_ASSIGN(std::unique_ptr writer, + CreateRealtimeWriter(realtime_context)); + ASSERT_OK(WriteCommitAndRefresh(MakeRows(/*first_id=*/0, /*count=*/5, /*partition=*/"p0"), + /*partitioned=*/false, /*bucket=*/0, + /*row_kinds=*/{}, /*commit_identifier=*/0, writer.get())); + ASSERT_OK(WriteCommitAndRefresh({{0, "ignored", "p0"}}, /*partitioned=*/false, /*bucket=*/0, + /*row_kinds=*/{RecordBatch::RowKind::DELETE}, + /*commit_identifier=*/1, writer.get())); + std::unique_ptr compactor; + ASSERT_OK_AND_ASSIGN( + std::vector> messages, + CompactAndPrepareCommit({RealtimePartitionBucket(/*partition=*/{}, /*bucket=*/0)}, + /*commit_identifier=*/0, /*full_compaction=*/true, &compactor)); + ASSERT_EQ(1, messages.size()); + auto message = std::dynamic_pointer_cast(messages[0]); + ASSERT_NE(nullptr, message); + ASSERT_EQ(1, message->GetCompactIncrement().CompactAfter().size()); + const auto& output = message->GetCompactIncrement().CompactAfter()[0]; + ASSERT_EQ(4, output->row_count); + ASSERT_OK(WriteCommitAndRefresh({{1, "ignored", "p0"}}, /*partitioned=*/false, /*bucket=*/0, + /*row_kinds=*/{RecordBatch::RowKind::DELETE}, + /*commit_identifier=*/2, writer.get())); + ASSERT_OK_AND_ASSIGN(std::vector baseline, InspectFiles()); + ASSERT_EQ(1, baseline.size()); + ASSERT_TRUE(baseline[0].deletion_file); + std::set indexes_before; + { + std::vector index_files; + ASSERT_OK( + dir_->GetFileSystem()->ListDir(PathUtil::JoinPath(table_path_, "index"), &index_files)); + for (const auto& file : index_files) { + if (!file.IsDir()) { + indexes_before.insert(file.GetPath()); + } + } + } + ASSERT_OK_AND_ASSIGN(std::unique_ptr batch, + MakeBatch({{2, "ignored", "p0"}}, /*partitioned=*/false, /*bucket=*/0, + /*row_kinds=*/{RecordBatch::RowKind::DELETE})); + ASSERT_OK(writer->Write(std::move(batch))); + ASSERT_OK_AND_ASSIGN(std::vector concurrent, + writer->PrepareCommitWithProgress(/*commit_identifier=*/3)); + std::set failed_attempt_paths; + auto fs = std::make_shared( + PathUtil::JoinPath(table_path_, "snapshot/snapshot-4"), [&]() -> Status { + // Capture candidates before the competing build writes any new DV of its own. + std::vector index_files; + PAIMON_RETURN_NOT_OK(dir_->GetFileSystem()->ListDir( + PathUtil::JoinPath(table_path_, "index"), &index_files)); + std::set during; + for (const auto& file : index_files) { + if (!file.IsDir()) { + during.insert(file.GetPath()); + } + } + std::set_difference(during.begin(), during.end(), indexes_before.begin(), + indexes_before.end(), + std::inserter(failed_attempt_paths, failed_attempt_paths.end())); + PAIMON_ASSIGN_OR_RAISE(int64_t snapshot, Commit(concurrent, /*commit_identifier=*/3)); + if (snapshot != 4) { + return Status::Invalid("expected the competing snapshot to occupy snapshot 4"); + } + return Status::OK(); + }); + CommitContextBuilder builder(table_path_, "regression-compactor"); + ASSERT_OK_AND_ASSIGN(std::unique_ptr context, + builder.SetOptions(options_).WithFileSystem(fs).Finish()); + ASSERT_OK_AND_ASSIGN(std::unique_ptr commit, + FileStoreCommit::Create(std::move(context))); + ASSERT_OK(commit->Commit(messages, /*commit_identifier=*/0)); + ASSERT_OK_AND_ASSIGN(uint64_t attempts, commit->GetCommitMetrics()->GetCounter( + CommitMetrics::LAST_COMMIT_ATTEMPTS)); + ASSERT_EQ(2, attempts); + ASSERT_EQ(1, failed_attempt_paths.size()); + for (const std::string& path : failed_attempt_paths) { + ASSERT_OK_AND_ASSIGN(bool exists, dir_->GetFileSystem()->Exists(path)); + ASSERT_FALSE(exists) << path; + } + // Cleanup owns only unpublished DVs, not baseline DVs, source data, or compact outputs. + for (const std::string& path : indexes_before) { + ASSERT_OK_AND_ASSIGN(bool exists, dir_->GetFileSystem()->Exists(path)); + ASSERT_TRUE(exists) << path; + } + std::vector> retained = {baseline[0].file, output}; + for (const auto& file : retained) { + const std::string bucket_path = PathUtil::JoinPath(table_path_, "bucket-0"); + ASSERT_OK_AND_ASSIGN(bool exists, dir_->GetFileSystem()->Exists( + PathUtil::JoinPath(bucket_path, file->file_name))); + ASSERT_TRUE(exists); + for (const auto& extra : file->extra_files) { + ASSERT_TRUE(extra); + ASSERT_OK_AND_ASSIGN( + exists, dir_->GetFileSystem()->Exists(PathUtil::JoinPath(bucket_path, *extra))); + ASSERT_TRUE(exists); + } + } + ASSERT_OK_AND_ASSIGN(std::vector files, InspectFiles()); + ASSERT_EQ(1, files.size()); + ASSERT_EQ(output->file_name, files[0].file->file_name); + ASSERT_TRUE(files[0].deletion_file); + ASSERT_EQ(0, failed_attempt_paths.count(files[0].deletion_file->path)); + ASSERT_EQ((std::set{0, 1}), files[0].deleted_positions); + ASSERT_OK_AND_ASSIGN(bool exists, dir_->GetFileSystem()->Exists(files[0].deletion_file->path)); + ASSERT_TRUE(exists); + const std::vector expected = MakeRows(/*first_id=*/3, /*count=*/2, /*partition=*/"p0"); + const std::multiset expected_rows(expected.begin(), expected.end()); + { + ASSERT_OK_AND_ASSIGN(std::vector disk_rows, ReadRows()); + ASSERT_EQ(expected_rows, std::multiset(disk_rows.begin(), disk_rows.end())); + ASSERT_OK_AND_ASSIGN(std::vector realtime_rows, ReadRows(realtime_context)); + ASSERT_EQ(expected_rows, std::multiset(realtime_rows.begin(), realtime_rows.end())); + } + SnapshotManager snapshot_manager(dir_->GetFileSystem(), table_path_); + ASSERT_OK_AND_ASSIGN(std::optional published, snapshot_manager.LatestSnapshot()); + ASSERT_TRUE(published); + ASSERT_EQ(5, published->Id()); + ASSERT_OK(writer->RefreshCommittedSnapshot(published->Id())); + { + ASSERT_OK_AND_ASSIGN(std::vector disk_rows, ReadRows()); + ASSERT_EQ(expected_rows, std::multiset(disk_rows.begin(), disk_rows.end())); + ASSERT_OK_AND_ASSIGN(std::vector realtime_rows, ReadRows(realtime_context)); + ASSERT_EQ(expected_rows, std::multiset(realtime_rows.begin(), realtime_rows.end())); + } + ASSERT_OK(writer->Close()); + ASSERT_OK(compactor->Close()); +} + TEST_F(RealtimeWriteInteTest, TestRealtimeDeduplicateWriterHandoffRestoresCommittedFiles) { ConfigureRealtimeDeduplicateTable();