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 <noreply@openai.com>
AI-Model: gpt-5
Co-Authored-By: Claude Code <noreply@anthropic.com>
AI-Contributed/Feature: 120/124
AI-Contributed/UT: 267/321

Author: 3 people

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.
+    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[];

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";

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());

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));

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;
 }

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;

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());

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,

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();

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(),