diff --git a/include/paimon/defs.h b/include/paimon/defs.h index ff5b4661..46b20c31 100644 --- a/include/paimon/defs.h +++ b/include/paimon/defs.h @@ -173,6 +173,10 @@ struct PAIMON_EXPORT Options { /// compaction of manifest, default value is 16MB. static const char MANIFEST_FULL_COMPACTION_FILE_SIZE[]; + /// "manifest.delete-file-drop-stats" - Whether final DELETE manifest entries should omit + /// value statistics. Default is false only for compatibility with old readers. + static const char MANIFEST_DELETE_FILE_DROP_STATS[]; + /// "source.split.target-size" - Target size of a source split when scanning a bucket. Default /// value is 128MB. static const char SOURCE_SPLIT_TARGET_SIZE[]; diff --git a/src/paimon/common/defs.cpp b/src/paimon/common/defs.cpp index 2d9a8765..53f18c3f 100644 --- a/src/paimon/common/defs.cpp +++ b/src/paimon/common/defs.cpp @@ -52,6 +52,7 @@ const char Options::MANIFEST_COMPRESSION[] = "manifest.compression"; const char Options::MANIFEST_MERGE_MIN_COUNT[] = "manifest.merge-min-count"; const char Options::MANIFEST_FULL_COMPACTION_FILE_SIZE[] = "manifest.full-compaction-threshold-size"; +const char Options::MANIFEST_DELETE_FILE_DROP_STATS[] = "manifest.delete-file-drop-stats"; const char Options::SOURCE_SPLIT_TARGET_SIZE[] = "source.split.target-size"; const char Options::SOURCE_SPLIT_OPEN_FILE_COST[] = "source.split.open-file-cost"; const char Options::SCAN_SNAPSHOT_ID[] = "scan.snapshot-id"; diff --git a/src/paimon/core/append/append_compact_coordinator.cpp b/src/paimon/core/append/append_compact_coordinator.cpp index 37693a58..f840e9c7 100644 --- a/src/paimon/core/append/append_compact_coordinator.cpp +++ b/src/paimon/core/append/append_compact_coordinator.cpp @@ -244,6 +244,9 @@ Result>>> Sca CreateFileStoreScan(snapshot_manager, schema_manager, table_schema, arrow_schema, partition_schema, core_options, path_factory, scan_filter, executor, pool)); + if (core_options.ManifestDeleteFileDropStats()) { + scan->EnableDropStats(); + } PAIMON_ASSIGN_OR_RAISE(std::shared_ptr plan, scan->CreatePlan()); std::vector add_entries = plan->Files(FileKind::Add()); diff --git a/src/paimon/core/append/append_compact_coordinator_test.cpp b/src/paimon/core/append/append_compact_coordinator_test.cpp index f1b82a90..2d1ff12c 100644 --- a/src/paimon/core/append/append_compact_coordinator_test.cpp +++ b/src/paimon/core/append/append_compact_coordinator_test.cpp @@ -141,7 +141,7 @@ class AppendCompactCoordinatorTest : public ::testing::Test { } void CheckCommitMessage(const std::shared_ptr& msg, size_t expected_before_files, - int64_t expected_total_rows) { + int64_t expected_total_rows, bool expect_dropped_stats = false) { auto impl = dynamic_cast(msg.get()); ASSERT_TRUE(impl); ASSERT_EQ(impl->Bucket(), 0); @@ -154,9 +154,17 @@ class AppendCompactCoordinatorTest : public ::testing::Test { int64_t total_before_rows = 0; for (const auto& file : compact_before) { total_before_rows += file->row_count; + if (expect_dropped_stats) { + ASSERT_EQ(SimpleStats::EmptyStats(), file->value_stats); + ASSERT_TRUE(file->value_stats_cols.has_value()); + ASSERT_TRUE(file->value_stats_cols->empty()); + } } ASSERT_EQ(total_before_rows, expected_total_rows); ASSERT_EQ(compact_after[0]->row_count, expected_total_rows); + if (expect_dropped_stats) { + ASSERT_FALSE(compact_after[0]->value_stats == SimpleStats::EmptyStats()); + } } private: @@ -179,6 +187,7 @@ TEST_F(AppendCompactCoordinatorTest, TestRunCompactsAllPartitions) { {Options::BUCKET, "-1"}, {Options::FILE_SYSTEM, "local"}, {Options::COMPACTION_MIN_FILE_NUM, "2"}, + {Options::MANIFEST_DELETE_FILE_DROP_STATS, "true"}, }; arrow::FieldVector fields = { @@ -246,11 +255,13 @@ TEST_F(AppendCompactCoordinatorTest, TestRunCompactsAllPartitions) { // f1=10: 2 files compacted into 1, total 7 rows CheckCommitMessage(compact_messages[0], /*expected_before_files=*/2, - /*expected_total_rows=*/7); + /*expected_total_rows=*/7, + /*expect_dropped_stats=*/true); // f1=20: 2 files compacted into 1, total 3 rows CheckCommitMessage(compact_messages[1], /*expected_before_files=*/2, - /*expected_total_rows=*/3); + /*expected_total_rows=*/3, + /*expect_dropped_stats=*/true); // Commit compact results ASSERT_OK(Commit(table_path, compact_messages)); diff --git a/src/paimon/core/core_options.cpp b/src/paimon/core/core_options.cpp index 965c3b13..e3366cd0 100644 --- a/src/paimon/core/core_options.cpp +++ b/src/paimon/core/core_options.cpp @@ -448,6 +448,7 @@ struct CoreOptions::Impl { int64_t write_buffer_spill_max_disk_size = std::numeric_limits::max(); bool ignore_delete = false; + bool manifest_delete_file_drop_stats = false; bool write_buffer_spillable = true; bool write_only = false; bool bucket_append_ordered = false; @@ -657,6 +658,9 @@ struct CoreOptions::Impl { // Parse manifest.full-compaction-threshold-size - size threshold for full compaction PAIMON_RETURN_NOT_OK(parser.ParseMemorySize(Options::MANIFEST_FULL_COMPACTION_FILE_SIZE, &manifest_full_compaction_file_size)); + // Parse manifest.delete-file-drop-stats - drop stats from DELETE entries, default false + PAIMON_RETURN_NOT_OK(parser.Parse(Options::MANIFEST_DELETE_FILE_DROP_STATS, + &manifest_delete_file_drop_stats)); return Status::OK(); } @@ -1172,6 +1176,10 @@ int64_t CoreOptions::GetManifestFullCompactionThresholdSize() const { return impl_->manifest_full_compaction_file_size; } +bool CoreOptions::ManifestDeleteFileDropStats() const { + return impl_->manifest_delete_file_drop_stats; +} + const std::string& CoreOptions::GetManifestCompression() const { return impl_->manifest_compression; } diff --git a/src/paimon/core/core_options.h b/src/paimon/core/core_options.h index 3367f45d..ebf4eceb 100644 --- a/src/paimon/core/core_options.h +++ b/src/paimon/core/core_options.h @@ -96,6 +96,12 @@ class PAIMON_EXPORT CoreOptions { const std::string& GetManifestCompression() const; int32_t GetManifestMergeMinCount() const; int64_t GetManifestFullCompactionThresholdSize() const; + + /// Return whether final DELETE manifest entries should omit value statistics. + /// + /// @return True when DELETE entries should omit value statistics. + bool ManifestDeleteFileDropStats() const; + int64_t GetSourceSplitTargetSize() const; int64_t GetSourceSplitOpenFileCost() const; std::optional GetScanSnapshotId() const; diff --git a/src/paimon/core/core_options_test.cpp b/src/paimon/core/core_options_test.cpp index 0fea6744..ab6fc690 100644 --- a/src/paimon/core/core_options_test.cpp +++ b/src/paimon/core/core_options_test.cpp @@ -62,6 +62,7 @@ TEST(CoreOptionsTest, TestDefaultValue) { ASSERT_EQ(8 * 1024 * 1024L, core_options.GetManifestTargetFileSize()); ASSERT_EQ(16 * 1024 * 1024L, core_options.GetManifestFullCompactionThresholdSize()); ASSERT_EQ(30, core_options.GetManifestMergeMinCount()); + ASSERT_FALSE(core_options.ManifestDeleteFileDropStats()); ASSERT_EQ(0, core_options.GetScanManifestEntryCacheMaxSnapshots()); ASSERT_EQ(nullptr, core_options.GetCache()); ASSERT_EQ(128 * 1024 * 1024L, core_options.GetSourceSplitTargetSize()); @@ -194,6 +195,7 @@ TEST(CoreOptionsTest, TestFromMap) { {Options::MANIFEST_TARGET_FILE_SIZE, "16MB"}, {Options::MANIFEST_FULL_COMPACTION_FILE_SIZE, "32MB"}, {Options::MANIFEST_MERGE_MIN_COUNT, "2"}, + {Options::MANIFEST_DELETE_FILE_DROP_STATS, "true"}, {Options::SOURCE_SPLIT_TARGET_SIZE, "24MB"}, {Options::SOURCE_SPLIT_OPEN_FILE_COST, "32MB"}, {Options::READ_BATCH_SIZE, "2048"}, @@ -328,6 +330,7 @@ TEST(CoreOptionsTest, TestFromMap) { ASSERT_EQ(16 * 1024 * 1024L, core_options.GetManifestTargetFileSize()); ASSERT_EQ(32 * 1024 * 1024L, core_options.GetManifestFullCompactionThresholdSize()); ASSERT_EQ(2, core_options.GetManifestMergeMinCount()); + ASSERT_TRUE(core_options.ManifestDeleteFileDropStats()); ASSERT_EQ(nullptr, core_options.GetCache()); ASSERT_EQ(24 * 1024 * 1024L, core_options.GetSourceSplitTargetSize()); ASSERT_EQ(32 * 1024 * 1024L, core_options.GetSourceSplitOpenFileCost()); diff --git a/src/paimon/core/io/data_file_meta.cpp b/src/paimon/core/io/data_file_meta.cpp index 86b4d62b..7ce767f4 100644 --- a/src/paimon/core/io/data_file_meta.cpp +++ b/src/paimon/core/io/data_file_meta.cpp @@ -96,6 +96,14 @@ std::shared_ptr DataFileMeta::CopyWithExtraFiles( first_row_id, write_cols); } +std::shared_ptr DataFileMeta::CopyWithoutStats() const { + return std::make_shared( + file_name, file_size, row_count, min_key, max_key, key_stats, SimpleStats::EmptyStats(), + min_sequence_number, max_sequence_number, schema_id, level, extra_files, creation_time, + delete_row_count, embedded_index, file_source, std::vector(), external_path, + first_row_id, write_cols); +} + DataFileMeta::DataFileMeta( const std::string& _file_name, int64_t _file_size, int64_t _row_count, const BinaryRow& _min_key, const BinaryRow& _max_key, const SimpleStats& _key_stats, diff --git a/src/paimon/core/io/data_file_meta.h b/src/paimon/core/io/data_file_meta.h index 443e9c46..98aaa0ee 100644 --- a/src/paimon/core/io/data_file_meta.h +++ b/src/paimon/core/io/data_file_meta.h @@ -83,6 +83,11 @@ struct DataFileMeta { std::shared_ptr CopyWithExtraFiles( const std::vector>& new_extra_files) const; + /// Create a copy without value statistics. All other metadata is preserved. + /// + /// @return A new metadata object with empty value statistics and value-stat columns. + std::shared_ptr CopyWithoutStats() const; + std::optional AddRowCount() const { return delete_row_count == std::nullopt ? std::optional() : row_count - delete_row_count.value(); diff --git a/src/paimon/core/io/data_file_meta_test.cpp b/src/paimon/core/io/data_file_meta_test.cpp index 78b27b8b..f7283f15 100644 --- a/src/paimon/core/io/data_file_meta_test.cpp +++ b/src/paimon/core/io/data_file_meta_test.cpp @@ -20,9 +20,44 @@ #include "gtest/gtest.h" #include "paimon/status.h" +#include "paimon/testing/utils/binary_row_generator.h" #include "paimon/testing/utils/testharness.h" namespace paimon::test { +TEST(DataFileMetaTest, TestCopyWithoutStats) { + std::shared_ptr pool = GetDefaultPool(); + SimpleStats value_stats = BinaryRowGenerator::GenerateStats( + {1, std::string("a")}, {5, std::string("z")}, {0, 1}, pool.get()); + auto file_meta = std::make_shared( + "data-0.orc", /*file_size=*/645, /*row_count=*/5, BinaryRow::EmptyRow(), + BinaryRow::EmptyRow(), SimpleStats::EmptyStats(), value_stats, + /*min_sequence_number=*/0, /*max_sequence_number=*/4, /*schema_id=*/0, + /*level=*/0, /*extra_files=*/std::vector>(), + /*creation_time=*/Timestamp(1737111915429ll, 0), + /*delete_row_count=*/2, /*embedded_index=*/nullptr, FileSource::Append(), + /*value_stats_cols=*/std::vector({"f0", "f1"}), + /*external_path=*/"file:/tmp/bucket-0/data-0.orc", /*first_row_id=*/100, + /*write_cols=*/std::vector({"f0"})); + + std::shared_ptr result = file_meta->CopyWithoutStats(); + + ASSERT_NE(file_meta.get(), result.get()); + DataFileMeta expected = *file_meta; + expected.value_stats = SimpleStats::EmptyStats(); + expected.value_stats_cols = std::vector(); + ASSERT_EQ(expected, *result); + ASSERT_EQ(value_stats, file_meta->value_stats); + ASSERT_EQ(std::vector({"f0", "f1"}), file_meta->value_stats_cols.value()); + + // Upgrade cannot restore stats once they have been dropped. Writer restore must therefore + // retain full stats for metadata-only ADD entries; see the Paimon Java bug at + // https://github.com/apache/paimon/issues/7026. + ASSERT_OK_AND_ASSIGN(std::shared_ptr upgraded, result->Upgrade(/*new_level=*/1)); + ASSERT_EQ(SimpleStats::EmptyStats(), upgraded->value_stats); + ASSERT_TRUE(upgraded->value_stats_cols.has_value()); + ASSERT_TRUE(upgraded->value_stats_cols->empty()); +} + TEST(DataFileMetaTest, TestAddRowCount) { DataFileMeta file_meta("data-80110e15-97b5-4bcf-ac09-6ca2659a4950-0.orc", /*file_size=*/645, /*row_count=*/5, BinaryRow::EmptyRow(), BinaryRow::EmptyRow(), diff --git a/src/paimon/core/manifest/manifest_entry.cpp b/src/paimon/core/manifest/manifest_entry.cpp index c0fbac0c..849407ba 100644 --- a/src/paimon/core/manifest/manifest_entry.cpp +++ b/src/paimon/core/manifest/manifest_entry.cpp @@ -24,6 +24,10 @@ class DataType; } // namespace arrow namespace paimon { +ManifestEntry ManifestEntry::CopyWithoutStats() const { + return ManifestEntry(kind_, partition_, bucket_, total_buckets_, file_->CopyWithoutStats()); +} + const std::shared_ptr& ManifestEntry::DataType() { static std::shared_ptr data_type = arrow::struct_({arrow::field("_KIND", arrow::int8(), /*nullable=*/false), diff --git a/src/paimon/core/manifest/manifest_entry.h b/src/paimon/core/manifest/manifest_entry.h index e39fa51c..ca6e8ffa 100644 --- a/src/paimon/core/manifest/manifest_entry.h +++ b/src/paimon/core/manifest/manifest_entry.h @@ -130,6 +130,11 @@ class ManifestEntry : public FileEntry { return file_; } + /// Create a copy whose data file has empty value statistics. + /// + /// @return A new manifest entry preserving all metadata except value statistics. + ManifestEntry CopyWithoutStats() const; + bool operator==(const ManifestEntry& other) const { if (this == &other) { return true; diff --git a/src/paimon/core/operation/abstract_file_store_write.cpp b/src/paimon/core/operation/abstract_file_store_write.cpp index d7dafa18..653e6d92 100644 --- a/src/paimon/core/operation/abstract_file_store_write.cpp +++ b/src/paimon/core/operation/abstract_file_store_write.cpp @@ -302,6 +302,10 @@ Result> AbstractFileStoreWrite::ScanExistingFileMe if (dv_maintainer_factory_) { index_file_handler = dv_maintainer_factory_->GetIndexFileHandler(); } + // Paimon Java currently drops value stats during writer restore. This is a known bug: a + // restored file can become a compact-after ADD via metadata-only level upgrade and lose its + // stats (https://github.com/apache/paimon/issues/7026). C++ intentionally does not align with + // that behavior; stats are dropped later only when the final entry kind is DELETE. FileSystemWriteRestore restore(snapshot_manager_, std::move(scan), index_file_handler); PAIMON_ASSIGN_OR_RAISE( std::shared_ptr restore_files, diff --git a/src/paimon/core/operation/append_only_file_store_scan_test.cpp b/src/paimon/core/operation/append_only_file_store_scan_test.cpp index 1fe75338..f319498a 100644 --- a/src/paimon/core/operation/append_only_file_store_scan_test.cpp +++ b/src/paimon/core/operation/append_only_file_store_scan_test.cpp @@ -40,6 +40,7 @@ #include "paimon/fs/local/local_file_system.h" #include "paimon/memory/memory_pool.h" #include "paimon/metrics.h" +#include "paimon/predicate/literal.h" #include "paimon/predicate/predicate_builder.h" #include "paimon/scan_context.h" #include "paimon/status.h" @@ -184,7 +185,8 @@ namespace { std::shared_ptr BuildScan(const std::string& table_path, const std::shared_ptr& cache, - const std::optional& bucket = std::nullopt) { + const std::optional& bucket = std::nullopt, + const std::shared_ptr& predicate = nullptr) { ScanContextBuilder context_builder(table_path); context_builder.AddOption(Options::FILE_FORMAT, "orc") .AddOption(Options::MANIFEST_FORMAT, "orc") @@ -193,6 +195,9 @@ std::shared_ptr BuildScan(const std::string& table_path, if (bucket) { context_builder.SetBucketFilter(bucket.value()); } + if (predicate) { + context_builder.SetPredicate(predicate); + } EXPECT_OK_AND_ASSIGN(auto scan_context, context_builder.Finish()); EXPECT_OK_AND_ASSIGN(auto table_scan, TableScan::Create(std::move(scan_context))); auto typed_table_scan = dynamic_cast(table_scan.get()); @@ -202,6 +207,41 @@ std::shared_ptr BuildScan(const std::string& table_path, } // namespace +TEST(AppendOnlyFileStoreScanTest, TestDropStatsAfterFiltering) { + TimezoneGuard guard("Asia/Shanghai"); + std::string table_path = paimon::test::GetDataDir() + "/orc/append_09.db/append_09/"; + std::shared_ptr predicate = PredicateBuilder::Equal( + /*field_index=*/0, /*field_name=*/"f0", FieldType::STRING, + Literal(FieldType::STRING, "David", 5)); + + std::shared_ptr scan_with_stats = + BuildScan(table_path, /*cache=*/nullptr, /*bucket=*/std::nullopt, predicate); + ASSERT_OK_AND_ASSIGN(Snapshot snapshot, + scan_with_stats->GetSnapshotManager()->LoadSnapshot(/*snapshot_id=*/3)); + scan_with_stats->WithSnapshot(snapshot); + ASSERT_OK_AND_ASSIGN(std::shared_ptr plan_with_stats, + scan_with_stats->CreatePlan()); + std::vector entries_with_stats = plan_with_stats->Files(); + ASSERT_FALSE(entries_with_stats.empty()); + + std::shared_ptr scan_without_stats = + BuildScan(table_path, /*cache=*/nullptr, /*bucket=*/std::nullopt, predicate); + scan_without_stats->WithSnapshot(snapshot)->EnableDropStats(); + ASSERT_OK_AND_ASSIGN(std::shared_ptr plan_without_stats, + scan_without_stats->CreatePlan()); + std::vector entries_without_stats = plan_without_stats->Files(); + + ASSERT_EQ(entries_with_stats.size(), entries_without_stats.size()); + for (size_t i = 0; i < entries_with_stats.size(); ++i) { + ASSERT_EQ(entries_with_stats[i].CreateIdentifier(), + entries_without_stats[i].CreateIdentifier()); + ASSERT_FALSE(entries_with_stats[i].File()->value_stats == SimpleStats::EmptyStats()); + ASSERT_EQ(SimpleStats::EmptyStats(), entries_without_stats[i].File()->value_stats); + ASSERT_TRUE(entries_without_stats[i].File()->value_stats_cols.has_value()); + ASSERT_TRUE(entries_without_stats[i].File()->value_stats_cols->empty()); + } +} + TEST(AppendOnlyFileStoreScanTest, TestSnapshotLiveManifestCachePath) { TimezoneGuard guard("Asia/Shanghai"); std::string table_path = paimon::test::GetDataDir() + "/orc/append_09.db/append_09/"; diff --git a/src/paimon/core/operation/commit/commit_scanner.cpp b/src/paimon/core/operation/commit/commit_scanner.cpp index abee7695..5590dd02 100644 --- a/src/paimon/core/operation/commit/commit_scanner.cpp +++ b/src/paimon/core/operation/commit/commit_scanner.cpp @@ -96,7 +96,8 @@ Result> CommitScanner::ReadAllEntriesFromChangedParti PAIMON_ASSIGN_OR_RAISE(partition_filters, ToPartitionFilters(changed_partitions)); PAIMON_ASSIGN_OR_RAISE(std::unique_ptr scan, - NewScan(partition_filters, /*for_overwrite=*/false)); + NewScan(partition_filters, /*for_overwrite=*/false, + /*drop_stats=*/false)); PAIMON_ASSIGN_OR_RAISE(std::shared_ptr plan, scan->WithSnapshot(snapshot)->WithKind(ScanMode::ALL)->CreatePlan()); return plan->Files(); @@ -120,7 +121,9 @@ Result> CommitScanner::ReadIncrementalEntries( manifest_file_->Read(manifest_meta.FileName(), /*filter=*/nullptr, &manifest_entries)); for (const ManifestEntry& entry : manifest_entries) { if (changed_partition_set.find(entry.Partition()) != changed_partition_set.end()) { - incremental_entries.push_back(entry); + const bool drop_stats = core_options_.ManifestDeleteFileDropStats() && + entry.Kind() == FileKind::Delete(); + incremental_entries.push_back(drop_stats ? entry.CopyWithoutStats() : entry); } } } @@ -132,14 +135,16 @@ Result> CommitScanner::ReadAllEntriesFromPartitions( const Snapshot& snapshot, const std::vector>& partitions) const { PAIMON_ASSIGN_OR_RAISE(std::unique_ptr scan, - NewScan(partitions, /*for_overwrite=*/false)); + NewScan(partitions, /*for_overwrite=*/false, + /*drop_stats=*/false)); PAIMON_ASSIGN_OR_RAISE(std::shared_ptr plan, scan->WithSnapshot(snapshot)->WithKind(ScanMode::ALL)->CreatePlan()); return plan->Files(); } Result> CommitScanner::NewScan( - const std::vector>& partitions, bool for_overwrite) const { + const std::vector>& partitions, bool for_overwrite, + bool drop_stats) const { auto scan_filter = std::make_shared(/*predicate=*/nullptr, partitions, /*bucket_filter=*/std::nullopt); if (!scan_supplier_) { @@ -147,6 +152,9 @@ Result> CommitScanner::NewScan( } PAIMON_ASSIGN_OR_RAISE(std::unique_ptr scan, scan_supplier_(scan_filter)); + if (drop_stats && core_options_.ManifestDeleteFileDropStats()) { + scan->EnableDropStats(); + } if (for_overwrite && core_options_.GetBucket() != BucketModeDefine::POSTPONE_BUCKET) { scan->OnlyReadRealBuckets(); } @@ -203,7 +211,8 @@ std::shared_ptr CommitScanner::OverwriteChangesProvider( changes, index_entries, [this, partitions](const Snapshot& snapshot) -> Result> { PAIMON_ASSIGN_OR_RAISE(std::unique_ptr scan, - NewScan(partitions, /*for_overwrite=*/true)); + NewScan(partitions, /*for_overwrite=*/true, + /*drop_stats=*/true)); PAIMON_ASSIGN_OR_RAISE( std::shared_ptr plan, scan->WithSnapshot(snapshot)->WithKind(ScanMode::ALL)->CreatePlan()); diff --git a/src/paimon/core/operation/commit/commit_scanner.h b/src/paimon/core/operation/commit/commit_scanner.h index 552cc648..ef6f6a92 100644 --- a/src/paimon/core/operation/commit/commit_scanner.h +++ b/src/paimon/core/operation/commit/commit_scanner.h @@ -97,8 +97,8 @@ class CommitScanner { const std::vector& changed_partitions) const; Result> NewScan( - const std::vector>& partitions, - bool for_overwrite) const; + const std::vector>& partitions, bool for_overwrite, + bool drop_stats) const; private: std::shared_ptr snapshot_manager_; diff --git a/src/paimon/core/operation/commit/manifest_entry_changes.cpp b/src/paimon/core/operation/commit/manifest_entry_changes.cpp index 1ca54a4b..6b31a935 100644 --- a/src/paimon/core/operation/commit/manifest_entry_changes.cpp +++ b/src/paimon/core/operation/commit/manifest_entry_changes.cpp @@ -29,8 +29,8 @@ namespace paimon { -ManifestEntryChanges::ManifestEntryChanges(int32_t default_num_bucket) - : default_num_bucket_(default_num_bucket) {} +ManifestEntryChanges::ManifestEntryChanges(int32_t default_num_bucket, bool drop_delete_file_stats) + : default_num_bucket_(default_num_bucket), drop_delete_file_stats_(drop_delete_file_stats) {} Status ManifestEntryChanges::Collect(const std::shared_ptr& message) { auto commit_message = std::dynamic_pointer_cast(message); @@ -142,8 +142,10 @@ ManifestEntry ManifestEntryChanges::MakeEntry( int32_t total_buckets = commit_message->TotalBuckets() == std::nullopt ? default_num_bucket_ : commit_message->TotalBuckets().value(); + std::shared_ptr entry_file = + drop_delete_file_stats_ && kind == FileKind::Delete() ? file->CopyWithoutStats() : file; return ManifestEntry(kind, commit_message->Partition(), commit_message->Bucket(), total_buckets, - file); + entry_file); } } // namespace paimon diff --git a/src/paimon/core/operation/commit/manifest_entry_changes.h b/src/paimon/core/operation/commit/manifest_entry_changes.h index bafb8bf0..da2329fc 100644 --- a/src/paimon/core/operation/commit/manifest_entry_changes.h +++ b/src/paimon/core/operation/commit/manifest_entry_changes.h @@ -39,7 +39,11 @@ namespace paimon { /// Detailed changes from `CommitMessage`s. class ManifestEntryChanges { public: - explicit ManifestEntryChanges(int32_t default_num_bucket); + /// Create a change collector. + /// + /// @param default_num_bucket Bucket count used when a commit message omits it. + /// @param drop_delete_file_stats Whether DELETE data-file entries should omit value stats. + explicit ManifestEntryChanges(int32_t default_num_bucket, bool drop_delete_file_stats); Status Collect(const std::shared_ptr& message); @@ -70,6 +74,7 @@ class ManifestEntryChanges { private: int32_t default_num_bucket_; + bool drop_delete_file_stats_; }; } // namespace paimon 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 434f7b4c..8680ff2c 100644 --- a/src/paimon/core/operation/commit/manifest_entry_changes_test.cpp +++ b/src/paimon/core/operation/commit/manifest_entry_changes_test.cpp @@ -40,6 +40,7 @@ #include "paimon/data/timestamp.h" #include "paimon/defs.h" #include "paimon/memory/bytes.h" +#include "paimon/testing/utils/binary_row_generator.h" #include "paimon/testing/utils/testharness.h" namespace paimon::test { @@ -102,7 +103,8 @@ TEST_F(ManifestEntryChangesTest, TestCollectAndSummary) { std::shared_ptr message = std::make_shared( partition, /*bucket=*/0, /*total_buckets=*/4, data_increment, compact_increment); - ManifestEntryChanges changes(/*default_num_bucket=*/8); + ManifestEntryChanges changes(/*default_num_bucket=*/8, + /*drop_delete_file_stats=*/false); ASSERT_OK(changes.Collect(message)); ASSERT_EQ(2u, changes.append_table_files.size()); @@ -126,6 +128,52 @@ TEST_F(ManifestEntryChangesTest, TestCollectAndSummary) { ASSERT_NE(std::string::npos, summary.find("2 compact index files")); } +TEST_F(ManifestEntryChangesTest, TestDropStatsOnlyForDeleteEntries) { + SimpleStats value_stats = + BinaryRowGenerator::GenerateStats({1}, {8}, {0}, GetDefaultPool().get()); + std::shared_ptr before = std::make_shared( + "compact-file", /*file_size=*/1024, /*row_count=*/8, DataFileMeta::EmptyMinKey(), + DataFileMeta::EmptyMaxKey(), SimpleStats::EmptyStats(), value_stats, + /*min_sequence_number=*/16, /*max_sequence_number=*/32, /*schema_id=*/1, + /*level=*/0, /*extra_files=*/std::vector>(), + /*creation_time=*/Timestamp(0, 0), /*delete_row_count=*/std::nullopt, + /*embedded_index=*/nullptr, /*file_source=*/std::nullopt, + /*value_stats_cols=*/std::vector({"f0"}), + /*external_path=*/std::nullopt, /*first_row_id=*/std::nullopt, + /*write_cols=*/std::nullopt); + ASSERT_OK_AND_ASSIGN(std::shared_ptr after, before->Upgrade(/*new_level=*/1)); + + CompactIncrement compact_increment(/*compact_before=*/{before}, /*compact_after=*/{after}, + /*changelog_files=*/{}); + std::shared_ptr message = std::make_shared( + CreateIntRow(10), /*bucket=*/0, /*total_buckets=*/4, + DataIncrement(/*new_files=*/{}, /*deleted_files=*/{}, /*changelog_files=*/{}), + compact_increment); + + ManifestEntryChanges changes(/*default_num_bucket=*/8, + /*drop_delete_file_stats=*/true); + ASSERT_OK(changes.Collect(message)); + + ASSERT_EQ(2u, changes.compact_table_files.size()); + const ManifestEntry& delete_entry = changes.compact_table_files[0]; + ASSERT_EQ(FileKind::Delete(), delete_entry.Kind()); + ASSERT_EQ(SimpleStats::EmptyStats(), delete_entry.File()->value_stats); + ASSERT_TRUE(delete_entry.File()->value_stats_cols.has_value()); + ASSERT_TRUE(delete_entry.File()->value_stats_cols->empty()); + + const ManifestEntry& add_entry = changes.compact_table_files[1]; + ASSERT_EQ(FileKind::Add(), add_entry.Kind()); + ASSERT_EQ(value_stats, add_entry.File()->value_stats); + ASSERT_TRUE(add_entry.File()->value_stats_cols.has_value()); + ASSERT_EQ(std::vector({"f0"}), add_entry.File()->value_stats_cols.value()); + ASSERT_EQ(value_stats, before->value_stats); + + ManifestEntryChanges keep_stats(/*default_num_bucket=*/8, + /*drop_delete_file_stats=*/false); + ASSERT_OK(keep_stats.Collect(message)); + ASSERT_EQ(value_stats, keep_stats.compact_table_files[0].File()->value_stats); +} + TEST_F(ManifestEntryChangesTest, TestHasGlobalIndexFileAdditions) { const BinaryRow partition = CreateIntRow(10); @@ -141,14 +189,16 @@ TEST_F(ManifestEntryChangesTest, TestHasGlobalIndexFileAdditions) { std::shared_ptr message = std::make_shared( partition, /*bucket=*/0, /*total_buckets=*/4, data_increment, compact_increment); - ManifestEntryChanges changes(/*default_num_bucket=*/8); + ManifestEntryChanges changes(/*default_num_bucket=*/8, + /*drop_delete_file_stats=*/false); ASSERT_OK(changes.Collect(message)); ASSERT_TRUE(changes.HasGlobalIndexFileAdditions()); } TEST_F(ManifestEntryChangesTest, TestCollectInvalidCommitMessageType) { - ManifestEntryChanges changes(/*default_num_bucket=*/8); + ManifestEntryChanges changes(/*default_num_bucket=*/8, + /*drop_delete_file_stats=*/false); std::shared_ptr invalid_message = std::make_shared(); ASSERT_NOK_WITH_MSG(changes.Collect(invalid_message), "fail to cast commit message to commit message impl"); diff --git a/src/paimon/core/operation/file_store_commit_impl.cpp b/src/paimon/core/operation/file_store_commit_impl.cpp index d70f84e2..b9d88110 100644 --- a/src/paimon/core/operation/file_store_commit_impl.cpp +++ b/src/paimon/core/operation/file_store_commit_impl.cpp @@ -42,7 +42,6 @@ #include "paimon/common/utils/binary_row_partition_computer.h" #include "paimon/common/utils/date_time_utils.h" #include "paimon/common/utils/fields_comparator.h" -#include "paimon/common/utils/options_utils.h" #include "paimon/common/utils/path_util.h" #include "paimon/common/utils/scope_guard.h" #include "paimon/core/catalog/catalog_snapshot_commit.h" @@ -92,7 +91,6 @@ class MemoryPool; namespace { constexpr const char* kCommitStrictModeLastSafeSnapshot = "commit.strict-mode.last-safe-snapshot"; -constexpr const char* kManifestDeleteFileDropStats = "manifest.delete-file-drop-stats"; constexpr const char* kSequenceSnapshotOrdering = "sequence.snapshot-ordering"; constexpr const char* kPkClusteringOverride = "pk-clustering-override"; @@ -116,14 +114,6 @@ Status FileStoreCommitImpl::ValidateCommitOptions(const CoreOptions& options) { if (raw_options.find(kCommitStrictModeLastSafeSnapshot) != raw_options.end()) { unsupported_options.emplace_back(kCommitStrictModeLastSafeSnapshot); } - if (raw_options.find(kManifestDeleteFileDropStats) != raw_options.end()) { - PAIMON_ASSIGN_OR_RAISE( - bool manifest_delete_file_drop_stats, - OptionsUtils::GetValueFromMap(raw_options, kManifestDeleteFileDropStats)); - if (manifest_delete_file_drop_stats) { - unsupported_options.emplace_back(kManifestDeleteFileDropStats); - } - } if (raw_options.find(kSequenceSnapshotOrdering) != raw_options.end()) { unsupported_options.emplace_back(kSequenceSnapshotOrdering); } @@ -318,8 +308,10 @@ Result FileStoreCommitImpl::RollbackToAsLatest(int64_t target_snapshot_id) std::vector delta_files; for (const auto& entry : latest_entries) { if (target_identifiers.find(entry.CreateIdentifier()) == target_identifiers.end()) { - delta_files.emplace_back(FileKind::Delete(), entry.Partition(), entry.Bucket(), - entry.TotalBuckets(), entry.File()); + delta_files.emplace_back( + FileKind::Delete(), entry.Partition(), entry.Bucket(), entry.TotalBuckets(), + options_.ManifestDeleteFileDropStats() ? entry.File()->CopyWithoutStats() + : entry.File()); } } for (const auto& entry : target_entries) { @@ -1332,7 +1324,7 @@ std::shared_ptr FileStoreCommitImpl::CreateManifestCommitta Result FileStoreCommitImpl::CollectChanges( const std::vector>& commit_messages) { - ManifestEntryChanges changes(num_bucket_); + ManifestEntryChanges changes(num_bucket_, options_.ManifestDeleteFileDropStats()); for (const auto& message : commit_messages) { PAIMON_RETURN_NOT_OK(changes.Collect(message)); } diff --git a/src/paimon/core/operation/file_store_commit_impl_test.cpp b/src/paimon/core/operation/file_store_commit_impl_test.cpp index 309766dd..fadb1dd8 100644 --- a/src/paimon/core/operation/file_store_commit_impl_test.cpp +++ b/src/paimon/core/operation/file_store_commit_impl_test.cpp @@ -59,6 +59,7 @@ #include "paimon/core/manifest/index_manifest_file.h" #include "paimon/core/manifest/manifest_committable.h" #include "paimon/core/manifest/manifest_entry.h" +#include "paimon/core/manifest/manifest_file.h" #include "paimon/core/manifest/manifest_file_meta.h" #include "paimon/core/manifest/manifest_list.h" #include "paimon/core/operation/metrics/commit_metrics.h" @@ -2535,9 +2536,9 @@ TEST_F(FileStoreCommitImplTest, TestFixedBucketPKTableCommitAllowed) { } TEST_F(FileStoreCommitImplTest, ValidateCommitOptionsRejectsUnsupportedOptions) { - const std::vector unsupported_keys = { - "commit.strict-mode.last-safe-snapshot", "manifest.delete-file-drop-stats", - "sequence.snapshot-ordering", "pk-clustering-override"}; + const std::vector unsupported_keys = {"commit.strict-mode.last-safe-snapshot", + "sequence.snapshot-ordering", + "pk-clustering-override"}; for (const auto& key : unsupported_keys) { ASSERT_OK_AND_ASSIGN(CoreOptions options, CoreOptions::FromMap({{key, "true"}})); ASSERT_NOK_WITH_MSG(FileStoreCommitImpl::ValidateCommitOptions(options), @@ -2550,10 +2551,119 @@ TEST_F(FileStoreCommitImplTest, ValidateCommitOptionsRejectsUnsupportedOptions) ASSERT_OK(FileStoreCommitImpl::ValidateCommitOptions(ok_options)); } -TEST_F(FileStoreCommitImplTest, ValidateCommitOptionsAllowsDisabledManifestDeleteFileDropStats) { - ASSERT_OK_AND_ASSIGN(CoreOptions options, - CoreOptions::FromMap({{"manifest.delete-file-drop-stats", "false"}})); - ASSERT_OK(FileStoreCommitImpl::ValidateCommitOptions(options)); +TEST_F(FileStoreCommitImplTest, ValidateCommitOptionsAllowsManifestDeleteFileDropStats) { + for (const std::string value : {"false", "true"}) { + ASSERT_OK_AND_ASSIGN( + CoreOptions options, + CoreOptions::FromMap({{Options::MANIFEST_DELETE_FILE_DROP_STATS, value}})); + ASSERT_OK(FileStoreCommitImpl::ValidateCommitOptions(options)); + } +} + +TEST_F(FileStoreCommitImplTest, TestGetAllFilesKeepsValueStats) { + CommitContextBuilder context_builder(table_path_, "commit_user_1"); + ASSERT_OK_AND_ASSIGN(std::unique_ptr commit_context, + context_builder.AddOption(Options::MANIFEST_FORMAT, "orc") + .AddOption(Options::MANIFEST_TARGET_FILE_SIZE, "8mb") + .AddOption(Options::MANIFEST_DELETE_FILE_DROP_STATS, "true") + .AddOption(Options::FILE_SYSTEM, "local") + .Finish()); + ASSERT_OK_AND_ASSIGN(std::unique_ptr commit, + FileStoreCommit::Create(std::move(commit_context))); + auto commit_impl = dynamic_cast(commit.get()); + ASSERT_TRUE(commit_impl); + + std::vector> messages = + GetCommitMessages(paimon::test::GetDataDir() + + "/orc/append_09.db/append_09/commit_messages/commit_messages-01", + /*version=*/3); + ASSERT_OK(commit_impl->Commit(messages, /*commit_identifier=*/1)); + + ASSERT_OK_AND_ASSIGN(std::optional latest, + commit_impl->snapshot_manager_->LatestSnapshot()); + ASSERT_TRUE(latest.has_value()); + ASSERT_OK_AND_ASSIGN(std::vector entries, + commit_impl->GetAllFiles(latest.value(), /*partitions=*/{})); + ASSERT_FALSE(entries.empty()); + for (const ManifestEntry& entry : entries) { + ASSERT_FALSE(entry.File()->value_stats == SimpleStats::EmptyStats()); + } +} + +TEST_F(FileStoreCommitImplTest, TestOverwriteDropsDeleteFileStats) { + CommitContextBuilder context_builder(table_path_, "commit_user_1"); + ASSERT_OK_AND_ASSIGN(std::unique_ptr commit_context, + context_builder.AddOption(Options::MANIFEST_FORMAT, "orc") + .AddOption(Options::MANIFEST_TARGET_FILE_SIZE, "8mb") + .AddOption(Options::MANIFEST_DELETE_FILE_DROP_STATS, "true") + .AddOption(Options::FILE_SYSTEM, "local") + .Finish()); + ASSERT_OK_AND_ASSIGN(std::unique_ptr commit, + FileStoreCommit::Create(std::move(commit_context))); + auto commit_impl = dynamic_cast(commit.get()); + ASSERT_TRUE(commit_impl); + + std::vector> first_commit = + GetCommitMessages(paimon::test::GetDataDir() + + "/orc/append_09.db/append_09/commit_messages/commit_messages-01", + /*version=*/3); + ASSERT_OK(commit_impl->Commit(first_commit, /*commit_identifier=*/1)); + + std::vector> overwrite_commit = + GetCommitMessages(paimon::test::GetDataDir() + + "/orc/append_09.db/append_09/commit_messages/commit_messages-02", + /*version=*/3); + ASSERT_OK(commit_impl->Overwrite({}, overwrite_commit, /*commit_identifier=*/2)); + + ASSERT_OK_AND_ASSIGN(std::optional latest, + commit_impl->snapshot_manager_->LatestSnapshot()); + ASSERT_TRUE(latest.has_value()); + std::vector delta_manifests; + ASSERT_OK(commit_impl->manifest_list_->ReadDeltaManifests(latest.value(), &delta_manifests)); + std::vector delta_entries; + for (const ManifestFileMeta& manifest : delta_manifests) { + ASSERT_OK(commit_impl->manifest_file_->Read(manifest.FileName(), /*filter=*/nullptr, + &delta_entries)); + } + + int32_t add_count = 0; + int32_t delete_count = 0; + for (const ManifestEntry& entry : delta_entries) { + if (entry.Kind() == FileKind::Delete()) { + ++delete_count; + ASSERT_EQ(SimpleStats::EmptyStats(), entry.File()->value_stats); + ASSERT_TRUE(entry.File()->value_stats_cols.has_value()); + ASSERT_TRUE(entry.File()->value_stats_cols->empty()); + } else if (entry.Kind() == FileKind::Add()) { + ++add_count; + ASSERT_FALSE(entry.File()->value_stats == SimpleStats::EmptyStats()); + } + } + ASSERT_GT(delete_count, 0); + ASSERT_GT(add_count, 0); + + std::vector changed_partitions; + changed_partitions.reserve(delta_entries.size()); + for (const ManifestEntry& entry : delta_entries) { + changed_partitions.push_back(entry.Partition()); + } + ASSERT_OK_AND_ASSIGN( + std::vector incremental_entries, + commit_impl->commit_scanner_->ReadIncrementalEntries(latest.value(), changed_partitions)); + + add_count = 0; + delete_count = 0; + for (const ManifestEntry& entry : incremental_entries) { + if (entry.Kind() == FileKind::Delete()) { + ++delete_count; + ASSERT_EQ(SimpleStats::EmptyStats(), entry.File()->value_stats); + } else if (entry.Kind() == FileKind::Add()) { + ++add_count; + ASSERT_FALSE(entry.File()->value_stats == SimpleStats::EmptyStats()); + } + } + ASSERT_GT(delete_count, 0); + ASSERT_GT(add_count, 0); } TEST_F(FileStoreCommitImplTest, DropPartitionWithEmptyPartitionsFails) { diff --git a/src/paimon/core/operation/file_store_scan.cpp b/src/paimon/core/operation/file_store_scan.cpp index 681e772e..865e006f 100644 --- a/src/paimon/core/operation/file_store_scan.cpp +++ b/src/paimon/core/operation/file_store_scan.cpp @@ -187,6 +187,11 @@ Result> FileStoreScan::CreatePlan() cons } } } + if (drop_stats_) { + for (ManifestEntry& entry : manifest_entries) { + entry = entry.CopyWithoutStats(); + } + } const int64_t all_data_files = std::accumulate( all_manifest_file_metas.begin(), all_manifest_file_metas.end(), int64_t{0}, [](const int64_t sum, const ManifestFileMeta& manifest_file_meta) { diff --git a/src/paimon/core/operation/file_store_scan.h b/src/paimon/core/operation/file_store_scan.h index 53d7825a..669dc072 100644 --- a/src/paimon/core/operation/file_store_scan.h +++ b/src/paimon/core/operation/file_store_scan.h @@ -132,6 +132,14 @@ class FileStoreScan { return this; } + /// Drop value statistics from entries after all scan filters have been applied. + /// + /// @return This scan for chained configuration. + FileStoreScan* EnableDropStats() { + drop_stats_ = true; + return this; + } + const std::shared_ptr& GetSnapshotManager() const { return snapshot_manager_; } @@ -310,5 +318,6 @@ class FileStoreScan { std::optional specified_snapshot_; std::shared_ptr metrics_; std::string table_path_; + bool drop_stats_ = false; }; } // namespace paimon diff --git a/src/paimon/core/operation/file_system_write_restore.h b/src/paimon/core/operation/file_system_write_restore.h index 81bb6693..630ea83d 100644 --- a/src/paimon/core/operation/file_system_write_restore.h +++ b/src/paimon/core/operation/file_system_write_restore.h @@ -24,7 +24,6 @@ #include #include -#include "paimon/core/core_options.h" #include "paimon/core/index/index_file_handler.h" #include "paimon/core/operation/file_store_scan.h" #include "paimon/core/operation/restore_files.h" @@ -36,6 +35,11 @@ namespace paimon { /// `WriteRestore` to restore files directly from file system. class FileSystemWriteRestore : public WriteRestore { public: + /// Create a write restore backed by a file store scan. + /// + /// @param snapshot_manager Snapshot manager used to locate restore state. + /// @param scan Scan used to load existing files. + /// @param index_file_handler Handler used to restore deletion-vector indexes. FileSystemWriteRestore(const std::shared_ptr& snapshot_manager, std::unique_ptr&& scan, const std::shared_ptr& index_file_handler) diff --git a/src/paimon/core/operation/key_value_file_store_write_test.cpp b/src/paimon/core/operation/key_value_file_store_write_test.cpp index 60f2d95a..62dfd716 100644 --- a/src/paimon/core/operation/key_value_file_store_write_test.cpp +++ b/src/paimon/core/operation/key_value_file_store_write_test.cpp @@ -42,6 +42,8 @@ #include "paimon/common/table/special_fields.h" #include "paimon/common/utils/path_util.h" #include "paimon/core/io/data_file_meta.h" +#include "paimon/core/operation/restore_files.h" +#include "paimon/core/stats/simple_stats.h" #include "paimon/core/table/sink/commit_message_impl.h" #include "paimon/file_store_commit.h" #include "paimon/file_store_write.h" @@ -423,6 +425,58 @@ TEST_F(KeyValueFileStoreWriteTest, TestSpillSimple) { ASSERT_EQ(get_writer(1)->GetMemoryUsage(), 0); } +TEST_F(KeyValueFileStoreWriteTest, TestWriterRestoreKeepsValueStats) { + auto fields = {arrow::field("f0", arrow::utf8(), /*nullable=*/false)}; + arrow::Schema typed_schema(fields); + ::ArrowSchema schema; + ASSERT_TRUE(arrow::ExportSchema(typed_schema, &schema).ok()); + + auto dir = UniqueTestDirectory::Create(); + ASSERT_TRUE(dir); + std::map options = { + {Options::BUCKET, "1"}, + {Options::FILE_FORMAT, "orc"}, + {Options::MANIFEST_FORMAT, "orc"}, + {Options::MANIFEST_DELETE_FILE_DROP_STATS, "true"}}; + ASSERT_OK_AND_ASSIGN(auto catalog, Catalog::Create(dir->Str(), options)); + ASSERT_OK(catalog->CreateDatabase("foo", {}, /*ignore_if_exists=*/false)); + ASSERT_OK(catalog->CreateTable(Identifier("foo", "bar"), &schema, + /*partition_keys=*/{}, /*primary_keys=*/{"f0"}, options, + /*ignore_if_exists=*/false)); + + std::string table_path = PathUtil::JoinPath(dir->Str(), "foo.db/bar"); + WriteContextBuilder first_context_builder(table_path, "first-writer"); + first_context_builder.SetOptions(options); + ASSERT_OK_AND_ASSIGN(std::unique_ptr first_context, + first_context_builder.Finish()); + ASSERT_OK_AND_ASSIGN(auto first_write, FileStoreWrite::Create(std::move(first_context))); + ASSERT_OK(WriteSingleStringRow(first_write.get(), /*bucket=*/0, "alice")); + ASSERT_OK_AND_ASSIGN(auto commit_messages, + first_write->PrepareCommit(/*wait_compaction=*/false, 1)); + ASSERT_OK(first_write->Close()); + ASSERT_EQ(1, commit_messages.size()); + auto commit_message = std::dynamic_pointer_cast(commit_messages[0]); + ASSERT_NE(nullptr, commit_message); + ASSERT_EQ(1, commit_message->GetNewFilesIncrement().NewFiles().size()); + ASSERT_FALSE(commit_message->GetNewFilesIncrement().NewFiles()[0]->value_stats == + SimpleStats::EmptyStats()); + Commit(table_path, options, commit_messages); + + WriteContextBuilder restored_context_builder(table_path, "restored-writer"); + restored_context_builder.SetOptions(options); + ASSERT_OK_AND_ASSIGN(std::unique_ptr restored_context, + restored_context_builder.Finish()); + ASSERT_OK_AND_ASSIGN(auto restored_write, FileStoreWrite::Create(std::move(restored_context))); + auto key_value_write = dynamic_cast(restored_write.get()); + ASSERT_NE(nullptr, key_value_write); + ASSERT_OK_AND_ASSIGN(std::shared_ptr restore_files, + key_value_write->ScanExistingFileMetas(BinaryRow::EmptyRow(), + /*bucket=*/0)); + ASSERT_EQ(1, restore_files->DataFiles().size()); + ASSERT_FALSE(restore_files->DataFiles()[0]->value_stats == SimpleStats::EmptyStats()); + ASSERT_OK(restored_write->Close()); +} + TEST_F(KeyValueFileStoreWriteTest, TestSpillDiskQuotaExhaustedFallsBackToFlushDataFile) { auto fields = {arrow::field("f0", arrow::utf8(), /*nullable=*/false)}; arrow::Schema typed_schema(fields); diff --git a/test/inte/pk_compaction_inte_test.cpp b/test/inte/pk_compaction_inte_test.cpp index 8706a497..9103fbcc 100644 --- a/test/inte/pk_compaction_inte_test.cpp +++ b/test/inte/pk_compaction_inte_test.cpp @@ -384,6 +384,63 @@ class PkCompactionInteTest : public ::testing::Test, arrow::FieldVector fields_; }; +TEST_F(PkCompactionInteTest, TestMetadataOnlyLevelUpgradeKeepsValueStats) { + arrow::FieldVector fields = {arrow::field("id", arrow::int32()), + arrow::field("value", arrow::utf8())}; + std::map options = { + {Options::FILE_FORMAT, "parquet"}, + {Options::BUCKET, "1"}, + {Options::FILE_SYSTEM, "local"}, + {Options::MANIFEST_DELETE_FILE_DROP_STATS, "true"}}; + CreateTable(fields, /*partition_keys=*/{}, /*primary_keys=*/{"id"}, options); + const std::string table_path = TablePath(); + + auto array = arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), R"([ + [1, "Alice"], + [2, "Bob"] + ])") + .ValueOrDie(); + ASSERT_OK(WriteAndCommit(table_path, /*partition=*/{}, /*bucket=*/0, array, + /*commit_identifier=*/0)); + + ASSERT_OK_AND_ASSIGN(std::vector> compact_messages, + CompactAndCommit(table_path, /*partition=*/{}, /*bucket=*/0, + /*full_compaction=*/true, /*commit_identifier=*/1)); + ASSERT_EQ(1u, compact_messages.size()); + auto compact_message = std::dynamic_pointer_cast(compact_messages[0]); + ASSERT_NE(nullptr, compact_message); + const CompactIncrement& compact_increment = compact_message->GetCompactIncrement(); + ASSERT_EQ(1u, compact_increment.CompactBefore().size()); + ASSERT_EQ(1u, compact_increment.CompactAfter().size()); + + const std::shared_ptr& before = compact_increment.CompactBefore()[0]; + const std::shared_ptr& after = compact_increment.CompactAfter()[0]; + ASSERT_EQ(before->file_name, after->file_name) + << "Full compaction must use a metadata-only level upgrade"; + ASSERT_LT(before->level, after->level); + ASSERT_FALSE(before->value_stats == SimpleStats::EmptyStats()); + ASSERT_EQ(before->value_stats, after->value_stats); + + ScanContextBuilder scan_context_builder(table_path); + scan_context_builder.WithStreamingMode(false) + .AddOption(Options::FILE_SYSTEM, "local") + .AddOption(Options::SCAN_MODE, StartupMode::LatestFull().ToString()); + ASSERT_OK_AND_ASSIGN(std::unique_ptr scan_context, scan_context_builder.Finish()); + ASSERT_OK_AND_ASSIGN(std::unique_ptr table_scan, + TableScan::Create(std::move(scan_context))); + ASSERT_OK_AND_ASSIGN(std::shared_ptr plan, table_scan->CreatePlan()); + ASSERT_EQ(1u, plan->Splits().size()); + auto split = std::dynamic_pointer_cast(plan->Splits()[0]); + ASSERT_NE(nullptr, split); + ASSERT_EQ(1u, split->DataFiles().size()); + + const std::shared_ptr& active_file = split->DataFiles()[0]; + ASSERT_EQ(after->file_name, active_file->file_name); + ASSERT_EQ(after->level, active_file->level); + ASSERT_EQ(after->value_stats, active_file->value_stats); + ASSERT_FALSE(active_file->value_stats == SimpleStats::EmptyStats()); +} + // Verify shared-shredding MAP can be read correctly after PK full compaction. TEST_P(PkCompactionInteTest, TestKeyValueTableFullCompactionWithMapSharedShredding) { auto file_format = GetParam();