Fix comment

Author: JingsongLi

crates/integrations/datafusion/tests/pk_tables.rs

@@ -25,7 +25,8 @@ use std::sync::Arc;
 
 use datafusion::arrow::array::{Int32Array, StringArray};
 use datafusion::prelude::SessionContext;
-use paimon::{CatalogOptions, FileSystemCatalog, Options};
+use paimon::catalog::Identifier;
+use paimon::{Catalog, CatalogOptions, FileSystemCatalog, Options};
 use paimon_datafusion::{PaimonCatalogProvider, PaimonRelationPlanner, PaimonSqlHandler};
 use tempfile::TempDir;
 
@@ -1146,3 +1147,112 @@ async fn test_pk_multiple_value_columns() {
         ]
     );
 }
+
+// ======================= FirstRow Engine: INSERT OVERWRITE =======================
+
+/// INSERT OVERWRITE on a partitioned FirstRow-engine PK table should delete
+/// level-0 files. Before the fix, `skip_level_zero` was applied in the overwrite
+/// scan path, causing level-0 files to survive the overwrite.
+///
+/// Verifies via TableScan (scan_all_files) that the overwrite correctly produces
+/// delete entries for level-0 files, leaving only the new file per partition.
+#[tokio::test]
+async fn test_pk_first_row_insert_overwrite() {
+    let (_tmp, catalog) = create_test_env();
+    let handler = create_handler(catalog.clone());
+    handler
+        .sql("CREATE SCHEMA paimon.test_db")
+        .await
+        .expect("CREATE SCHEMA failed");
+
+    handler
+        .sql(
+            "CREATE TABLE paimon.test_db.t_fr_ow (
+                dt STRING, id INT NOT NULL, name STRING,
+                PRIMARY KEY (dt, id)
+            ) PARTITIONED BY (dt STRING)
+            WITH ('bucket' = '1', 'merge-engine' = 'first-row')",
+        )
+        .await
+        .unwrap();
+
+    // First commit: two partitions, creates level-0 files
+    handler
+        .sql(
+            "INSERT INTO paimon.test_db.t_fr_ow VALUES \
+             ('2024-01-01', 1, 'alice'), ('2024-01-01', 2, 'bob'), \
+             ('2024-01-02', 3, 'carol')",
+        )
+        .await
+        .unwrap()
+        .collect()
+        .await
+        .unwrap();
+
+    // Verify via scan_all_files: 2 level-0 files (one per partition)
+    let table = catalog
+        .get_table(&Identifier::new("test_db", "t_fr_ow"))
+        .await
+        .unwrap();
+    let plan = table
+        .new_read_builder()
+        .new_scan()
+        .with_scan_all_files()
+        .plan()
+        .await
+        .unwrap();
+    let file_count: usize = plan.splits().iter().map(|s| s.data_files().len()).sum();
+    assert_eq!(file_count, 2, "After INSERT: 2 level-0 files (one per partition)");
+
+    // INSERT OVERWRITE partition 2024-01-01 — must delete old level-0 file
+    handler
+        .sql("INSERT OVERWRITE paimon.test_db.t_fr_ow VALUES ('2024-01-01', 10, 'new_alice')")
+        .await
+        .unwrap()
+        .collect()
+        .await
+        .unwrap();
+
+    let table = catalog
+        .get_table(&Identifier::new("test_db", "t_fr_ow"))
+        .await
+        .unwrap();
+    let plan = table
+        .new_read_builder()
+        .new_scan()
+        .with_scan_all_files()
+        .plan()
+        .await
+        .unwrap();
+    let file_count: usize = plan.splits().iter().map(|s| s.data_files().len()).sum();
+    assert_eq!(
+        file_count, 2,
+        "After OVERWRITE: 2 files (1 replaced for 2024-01-01 + 1 unchanged for 2024-01-02)"
+    );
+
+    // Second overwrite on the same partition — no stale files should accumulate
+    handler
+        .sql("INSERT OVERWRITE paimon.test_db.t_fr_ow VALUES ('2024-01-01', 20, 'newer_alice')")
+        .await
+        .unwrap()
+        .collect()
+        .await
+        .unwrap();
+
+    let table = catalog
+        .get_table(&Identifier::new("test_db", "t_fr_ow"))
+        .await
+        .unwrap();
+    let plan = table
+        .new_read_builder()
+        .new_scan()
+        .with_scan_all_files()
+        .plan()
+        .await
+        .unwrap();
+    let file_count: usize = plan.splits().iter().map(|s| s.data_files().len()).sum();
+    assert_eq!(
+        file_count, 2,
+        "After second OVERWRITE: still 2 files (no stale level-0 files accumulated)"
+    );
+}

crates/paimon/src/table/table_commit.rs

@@ -477,7 +477,8 @@ impl TableCommit {
                 None,
                 None,
                 None,
-            );
+            )
+            .with_scan_all_files();
             let current_entries = scan.plan_manifest_entries(snap).await?;
             for entry in current_entries {
                 entries.push(entry.with_kind(FileKind::Delete));
@@ -567,7 +568,8 @@ impl TableCommit {
         };
 
         // Read all current files from the latest snapshot.
-        let scan = TableScan::new(&self.table, None, vec![], None, None, None);
+        let scan =
+            TableScan::new(&self.table, None, vec![], None, None, None).with_scan_all_files();
         let existing_entries = scan.plan_manifest_entries(snap).await?;
 
         // Build index: (partition, bucket, first_row_id, row_count)

crates/paimon/src/table/table_scan.rs

@@ -308,6 +308,10 @@ pub struct TableScan<'a> {
     /// When set, the scan will try to return only enough splits to satisfy the limit.
     limit: Option<usize>,
     row_ranges: Option<Vec<RowRange>>,
+    /// When true, disables level-0 filtering so all files are visible.
+    /// Used by non-read paths (overwrite, truncate, writer restore) that need
+    /// the complete file set. Normal read scans leave this as `false`.
+    scan_all_files: bool,
 }
 
 impl<'a> TableScan<'a> {
@@ -326,9 +330,19 @@ impl<'a> TableScan<'a> {
             bucket_predicate,
             limit,
             row_ranges,
+            scan_all_files: false,
         }
     }
 
+    /// Disable level-0 filtering so all files are visible.
+    ///
+    /// Used by non-read paths (overwrite, truncate, writer restore) that need
+    /// the complete file set regardless of merge engine or DV settings.
+    pub fn with_scan_all_files(mut self) -> Self {
+        self.scan_all_files = true;
+        self
+    }
+
     /// Set row ranges for scan-time filtering.
     ///
     /// This replaces any existing row_ranges. Typically used to inject
@@ -467,7 +481,12 @@ impl<'a> TableScan<'a> {
         // - DV mode: level-0 files are unmerged, DV handles dedup at higher levels
         // - FirstRow engine without DV: reads go through DataFileReader (no merge),
         //   so only compacted (level > 0) files are safe to read directly
-        let skip_level_zero = if has_primary_keys {
+        //
+        // Non-read paths (overwrite, truncate, writer restore) set scan_all_files=true
+        // to see all files including level-0, matching Java's CommitScanner behavior.
+        let skip_level_zero = if self.scan_all_files {
+            false
+        } else if has_primary_keys {
             deletion_vectors_enabled
                 || core_options
                     .merge_engine()

crates/paimon/src/table/table_write.rs

@@ -239,7 +239,8 @@ impl TableWrite {
         let mut bucket_seq: HashMap<i32, i64> = HashMap::new();
         if let Some(snapshot) = latest_snapshot {
             let partition_predicate = Self::build_partition_predicate(table, partition_bytes)?;
-            let scan = TableScan::new(table, partition_predicate, vec![], None, None, None);
+            let scan = TableScan::new(table, partition_predicate, vec![], None, None, None)
+                .with_scan_all_files();
             let entries = scan.plan_manifest_entries(&snapshot).await?;
             for entry in &entries {
                 let bucket = entry.bucket();