apache · liujiayi771 · Aug 9, 2026 · Aug 9, 2026 · lxy-9602 · Aug 9, 2026
diff --git 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
@@ -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
@@ -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
@@ -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
@@ -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
@@ -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
@@ -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
@@ -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
@@ -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
@@ -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
@@ -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
@@ -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
@@ -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
@@ -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/";