This is an automated email from the ASF dual-hosted git repository.
SteNicholas pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/paimon-cpp.git
The following commit(s) were added to refs/heads/main by this push:
new e0db538 feat(commit): support dropping delete file stats (#190)
e0db538 is described below
commit e0db538a7ee7e930d3c1a3bea15a64f4f93c6a76
Author: Joey <[email protected]>
AuthorDate: Tue Aug 11 13:54:26 2026 +0800
feat(commit): support dropping delete file stats (#190)
* feat(commit): support dropping delete file stats
Honor manifest.delete-file-drop-stats during commit scans and manifest
entry creation while preserving statistics for ADD entries and writer restore.
Co-Authored-By: Codex <[email protected]>
AI-Model: gpt-5
Co-Authored-By: Claude Code <[email protected]>
* fix(commit): refine delete file stats support
Require callers to pass drop_delete_file_stats explicitly, document the
old-reader compatibility default, and cover PK metadata-only level upgrades end
to end.
Co-Authored-By: Codex <[email protected]>
AI-Model: gpt-5
---------
Co-authored-by: Codex <[email protected]>
Co-authored-by: Claude Code <[email protected]>
---
include/paimon/defs.h | 4 +
src/paimon/common/defs.cpp | 1 +
.../core/append/append_compact_coordinator.cpp | 3 +
.../append/append_compact_coordinator_test.cpp | 17 ++-
src/paimon/core/core_options.cpp | 8 ++
src/paimon/core/core_options.h | 6 +
src/paimon/core/core_options_test.cpp | 3 +
src/paimon/core/io/data_file_meta.cpp | 8 ++
src/paimon/core/io/data_file_meta.h | 5 +
src/paimon/core/io/data_file_meta_test.cpp | 35 ++++++
src/paimon/core/manifest/manifest_entry.cpp | 4 +
src/paimon/core/manifest/manifest_entry.h | 5 +
.../core/operation/abstract_file_store_write.cpp | 4 +
.../operation/append_only_file_store_scan_test.cpp | 42 ++++++-
.../core/operation/commit/commit_scanner.cpp | 19 +++-
src/paimon/core/operation/commit/commit_scanner.h | 4 +-
.../operation/commit/manifest_entry_changes.cpp | 8 +-
.../core/operation/commit/manifest_entry_changes.h | 7 +-
.../commit/manifest_entry_changes_test.cpp | 56 +++++++++-
.../core/operation/file_store_commit_impl.cpp | 18 +--
.../core/operation/file_store_commit_impl_test.cpp | 124 +++++++++++++++++++--
src/paimon/core/operation/file_store_scan.cpp | 5 +
src/paimon/core/operation/file_store_scan.h | 9 ++
.../core/operation/file_system_write_restore.h | 6 +-
.../operation/key_value_file_store_write_test.cpp | 54 +++++++++
test/inte/pk_compaction_inte_test.cpp | 57 ++++++++++
26 files changed, 473 insertions(+), 39 deletions(-)
diff --git a/include/paimon/defs.h b/include/paimon/defs.h
index ff5b466..46b20c3 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 2d9a876..53f18c3 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 37693a5..f840e9c 100644
--- a/src/paimon/core/append/append_compact_coordinator.cpp
+++ b/src/paimon/core/append/append_compact_coordinator.cpp
@@ -244,6 +244,9 @@ Result<LinkedHashMap<BinaryRow,
std::vector<std::shared_ptr<DataFileMeta>>>> 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<FileStoreScan::RawPlan> plan,
scan->CreatePlan());
std::vector<ManifestEntry> 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 f1b82a9..2d1ff12 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<CommitMessage>& 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<CommitMessageImpl*>(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 965c3b1..e3366cd 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<int64_t>::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 3367f45..ebf4ece 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<int64_t> GetScanSnapshotId() const;
diff --git a/src/paimon/core/core_options_test.cpp
b/src/paimon/core/core_options_test.cpp
index 0fea674..ab6fc69 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 86b4d62..7ce767f 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>
DataFileMeta::CopyWithExtraFiles(
first_row_id, write_cols);
}
+std::shared_ptr<DataFileMeta> DataFileMeta::CopyWithoutStats() const {
+ return std::make_shared<DataFileMeta>(
+ 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<std::string>(), 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 443e9c4..98aaa0e 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<DataFileMeta> CopyWithExtraFiles(
const std::vector<std::optional<std::string>>& 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<DataFileMeta> CopyWithoutStats() const;
+
std::optional<int64_t> AddRowCount() const {
return delete_row_count == std::nullopt ? std::optional<int64_t>()
: 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 78b27b8..f7283f1 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<MemoryPool> 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<DataFileMeta>(
+ "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<std::optional<std::string>>(),
+ /*creation_time=*/Timestamp(1737111915429ll, 0),
+ /*delete_row_count=*/2, /*embedded_index=*/nullptr,
FileSource::Append(),
+ /*value_stats_cols=*/std::vector<std::string>({"f0", "f1"}),
+ /*external_path=*/"file:/tmp/bucket-0/data-0.orc",
/*first_row_id=*/100,
+ /*write_cols=*/std::vector<std::string>({"f0"}));
+
+ std::shared_ptr<DataFileMeta> 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<std::string>();
+ ASSERT_EQ(expected, *result);
+ ASSERT_EQ(value_stats, file_meta->value_stats);
+ ASSERT_EQ(std::vector<std::string>({"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<DataFileMeta> 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 c0fbac0..849407b 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<arrow::DataType>& ManifestEntry::DataType() {
static std::shared_ptr<arrow::DataType> 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 e39fa51..ca6e8ff 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 d7dafa1..653e6d9 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<std::shared_ptr<RestoreFiles>>
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<RestoreFiles> 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 1fe7533..f319498 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<FileStoreScan> BuildScan(const std::string& table_path,
const std::shared_ptr<Cache>& cache,
- const std::optional<int32_t>& bucket
= std::nullopt) {
+ const std::optional<int32_t>& bucket
= std::nullopt,
+ const std::shared_ptr<Predicate>&
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<FileStoreScan> 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<AbstractTableScan*>(table_scan.get());
@@ -202,6 +207,41 @@ std::shared_ptr<FileStoreScan> 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> predicate = PredicateBuilder::Equal(
+ /*field_index=*/0, /*field_name=*/"f0", FieldType::STRING,
+ Literal(FieldType::STRING, "David", 5));
+
+ std::shared_ptr<FileStoreScan> 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<FileStoreScan::RawPlan>
plan_with_stats,
+ scan_with_stats->CreatePlan());
+ std::vector<ManifestEntry> entries_with_stats = plan_with_stats->Files();
+ ASSERT_FALSE(entries_with_stats.empty());
+
+ std::shared_ptr<FileStoreScan> 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<FileStoreScan::RawPlan>
plan_without_stats,
+ scan_without_stats->CreatePlan());
+ std::vector<ManifestEntry> 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 abee769..5590dd0 100644
--- a/src/paimon/core/operation/commit/commit_scanner.cpp
+++ b/src/paimon/core/operation/commit/commit_scanner.cpp
@@ -96,7 +96,8 @@ Result<std::vector<ManifestEntry>>
CommitScanner::ReadAllEntriesFromChangedParti
PAIMON_ASSIGN_OR_RAISE(partition_filters,
ToPartitionFilters(changed_partitions));
PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<FileStoreScan> scan,
- NewScan(partition_filters,
/*for_overwrite=*/false));
+ NewScan(partition_filters, /*for_overwrite=*/false,
+ /*drop_stats=*/false));
PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<FileStoreScan::RawPlan> plan,
scan->WithSnapshot(snapshot)->WithKind(ScanMode::ALL)->CreatePlan());
return plan->Files();
@@ -120,7 +121,9 @@ Result<std::vector<ManifestEntry>>
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<std::vector<ManifestEntry>>
CommitScanner::ReadAllEntriesFromPartitions(
const Snapshot& snapshot,
const std::vector<std::map<std::string, std::string>>& partitions) const {
PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<FileStoreScan> scan,
- NewScan(partitions, /*for_overwrite=*/false));
+ NewScan(partitions, /*for_overwrite=*/false,
+ /*drop_stats=*/false));
PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<FileStoreScan::RawPlan> plan,
scan->WithSnapshot(snapshot)->WithKind(ScanMode::ALL)->CreatePlan());
return plan->Files();
}
Result<std::unique_ptr<FileStoreScan>> CommitScanner::NewScan(
- const std::vector<std::map<std::string, std::string>>& partitions, bool
for_overwrite) const {
+ const std::vector<std::map<std::string, std::string>>& partitions, bool
for_overwrite,
+ bool drop_stats) const {
auto scan_filter = std::make_shared<ScanFilter>(/*predicate=*/nullptr,
partitions,
/*bucket_filter=*/std::nullopt);
if (!scan_supplier_) {
@@ -147,6 +152,9 @@ Result<std::unique_ptr<FileStoreScan>>
CommitScanner::NewScan(
}
PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<FileStoreScan> 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<CommitChangesProvider>
CommitScanner::OverwriteChangesProvider(
changes, index_entries,
[this, partitions](const Snapshot& snapshot) ->
Result<std::vector<ManifestEntry>> {
PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<FileStoreScan> scan,
- NewScan(partitions,
/*for_overwrite=*/true));
+ NewScan(partitions, /*for_overwrite=*/true,
+ /*drop_stats=*/true));
PAIMON_ASSIGN_OR_RAISE(
std::shared_ptr<FileStoreScan::RawPlan> 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 552cc64..ef6f6a9 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<BinaryRow>& changed_partitions) const;
Result<std::unique_ptr<FileStoreScan>> NewScan(
- const std::vector<std::map<std::string, std::string>>& partitions,
- bool for_overwrite) const;
+ const std::vector<std::map<std::string, std::string>>& partitions,
bool for_overwrite,
+ bool drop_stats) const;
private:
std::shared_ptr<SnapshotManager> 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 1ca54a4..6b31a93 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<CommitMessage>&
message) {
auto commit_message =
std::dynamic_pointer_cast<CommitMessageImpl>(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<DataFileMeta> 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 bafb8bf..da2329f 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<CommitMessage>& 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 434f7b4..8680ff2 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<CommitMessage> message =
std::make_shared<CommitMessageImpl>(
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<DataFileMeta> before = std::make_shared<DataFileMeta>(
+ "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<std::optional<std::string>>(),
+ /*creation_time=*/Timestamp(0, 0), /*delete_row_count=*/std::nullopt,
+ /*embedded_index=*/nullptr, /*file_source=*/std::nullopt,
+ /*value_stats_cols=*/std::vector<std::string>({"f0"}),
+ /*external_path=*/std::nullopt, /*first_row_id=*/std::nullopt,
+ /*write_cols=*/std::nullopt);
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<DataFileMeta> after,
before->Upgrade(/*new_level=*/1));
+
+ CompactIncrement compact_increment(/*compact_before=*/{before},
/*compact_after=*/{after},
+ /*changelog_files=*/{});
+ std::shared_ptr<CommitMessage> message =
std::make_shared<CommitMessageImpl>(
+ 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<std::string>({"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<CommitMessage> message =
std::make_shared<CommitMessageImpl>(
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<CommitMessage> invalid_message =
std::make_shared<CommitMessage>();
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 d70f84e..b9d8811 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<bool>(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<bool>
FileStoreCommitImpl::RollbackToAsLatest(int64_t target_snapshot_id)
std::vector<ManifestEntry> 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<ManifestCommittable>
FileStoreCommitImpl::CreateManifestCommitta
Result<ManifestEntryChanges> FileStoreCommitImpl::CollectChanges(
const std::vector<std::shared_ptr<CommitMessage>>& 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 309766d..fadb1dd 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<std::string> unsupported_keys = {
- "commit.strict-mode.last-safe-snapshot",
"manifest.delete-file-drop-stats",
- "sequence.snapshot-ordering", "pk-clustering-override"};
+ const std::vector<std::string> 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<CommitContext> 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<FileStoreCommit> commit,
+ FileStoreCommit::Create(std::move(commit_context)));
+ auto commit_impl = dynamic_cast<FileStoreCommitImpl*>(commit.get());
+ ASSERT_TRUE(commit_impl);
+
+ std::vector<std::shared_ptr<CommitMessage>> 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<Snapshot> latest,
+ commit_impl->snapshot_manager_->LatestSnapshot());
+ ASSERT_TRUE(latest.has_value());
+ ASSERT_OK_AND_ASSIGN(std::vector<ManifestEntry> 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<CommitContext> 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<FileStoreCommit> commit,
+ FileStoreCommit::Create(std::move(commit_context)));
+ auto commit_impl = dynamic_cast<FileStoreCommitImpl*>(commit.get());
+ ASSERT_TRUE(commit_impl);
+
+ std::vector<std::shared_ptr<CommitMessage>> 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<std::shared_ptr<CommitMessage>> 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<Snapshot> latest,
+ commit_impl->snapshot_manager_->LatestSnapshot());
+ ASSERT_TRUE(latest.has_value());
+ std::vector<ManifestFileMeta> delta_manifests;
+ ASSERT_OK(commit_impl->manifest_list_->ReadDeltaManifests(latest.value(),
&delta_manifests));
+ std::vector<ManifestEntry> 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<BinaryRow> 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<ManifestEntry> 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 681e772..865e006 100644
--- a/src/paimon/core/operation/file_store_scan.cpp
+++ b/src/paimon/core/operation/file_store_scan.cpp
@@ -187,6 +187,11 @@ Result<std::shared_ptr<FileStoreScan::RawPlan>>
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 53d7825..669dc07 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<SnapshotManager>& GetSnapshotManager() const {
return snapshot_manager_;
}
@@ -310,5 +318,6 @@ class FileStoreScan {
std::optional<Snapshot> specified_snapshot_;
std::shared_ptr<Metrics> 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 81bb669..630ea83 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 <string>
#include <vector>
-#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<SnapshotManager>&
snapshot_manager,
std::unique_ptr<FileStoreScan>&& scan,
const std::shared_ptr<IndexFileHandler>&
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 60f2d95..62dfd71 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<std::string, std::string> 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<WriteContext> 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<CommitMessageImpl>(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<WriteContext> 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<KeyValueFileStoreWrite*>(restored_write.get());
+ ASSERT_NE(nullptr, key_value_write);
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<RestoreFiles> 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 8706a49..9103fbc 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<std::string, std::string> 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<std::shared_ptr<CommitMessage>>
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<CommitMessageImpl>(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<DataFileMeta>& before =
compact_increment.CompactBefore()[0];
+ const std::shared_ptr<DataFileMeta>& 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<ScanContext> scan_context,
scan_context_builder.Finish());
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<TableScan> table_scan,
+ TableScan::Create(std::move(scan_context)));
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<Plan> plan, table_scan->CreatePlan());
+ ASSERT_EQ(1u, plan->Splits().size());
+ auto split = std::dynamic_pointer_cast<DataSplitImpl>(plan->Splits()[0]);
+ ASSERT_NE(nullptr, split);
+ ASSERT_EQ(1u, split->DataFiles().size());
+
+ const std::shared_ptr<DataFileMeta>& 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();
