fix!: preserve unindexed fragments during zonemap pruning

integration-tests/test_lance_spark.py

@@ -915,6 +915,41 @@ def test_create_distributed_bitmap_index(self, spark):
         )
         assert spark.sql("SELECT * FROM default.test_table").count() == 4
 
+    def test_zonemap_partial_coverage_after_append(self, spark):
+        """A fragment appended after index creation must remain visible to filtered scans."""
+        spark.sql("""
+            CREATE TABLE default.test_table (
+                id INT,
+                name STRING,
+                value DOUBLE
+            )
+        """)
+
+        initial = [(i, f"Name{i}", float(i)) for i in range(10)]
+        spark.createDataFrame(initial, ["id", "name", "value"]).writeTo(
+            "default.test_table"
+        ).append()
+
+        spark.sql("""
+            ALTER TABLE default.test_table
+            CREATE INDEX idx_id_zonemap USING zonemap (id)
+            WITH (rows_per_zone = 4)
+        """).collect()
+
+        spark.createDataFrame(
+            [(1000, "Appended", 1000.0)], ["id", "name", "value"]
+        ).writeTo("default.test_table").append()
+
+        rows = spark.sql("""
+            SELECT id, name, value
+            FROM default.test_table
+            WHERE id = 1000
+        """).collect()
+
+        assert len(rows) == 1
+        assert rows[0].id == 1000
+        assert rows[0].name == "Appended"
+
     def test_create_btree_index_on_nested_literal_dot_field(self, spark):
         """Test CREATE INDEX on nested struct fields, including literal dots."""
         spark.sql("""

lance-spark-base_2.12/src/main/java/org/lance/spark/read/LanceScan.java

@@ -51,6 +51,7 @@
 import java.io.Serializable;
 import java.util.Arrays;
 import java.util.Collections;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Objects;
 import java.util.Set;
@@ -85,6 +86,9 @@ public class LanceScan
    */
   private final java.util.Map<String, List<ZoneStats>> zonemapStats;
 
+  /** Live fragment IDs from the same Dataset snapshot as {@link #zonemapStats}. */
+  private final Set<Integer> liveFragmentIds;
+
   /**
    * Pre-computed surviving fragment IDs from zonemap pruning in LanceScanBuilder. When non-null,
    * {@link #pruneByZonemapStats} skips re-computing and uses these directly.
@@ -138,6 +142,7 @@ public LanceScan(
       Predicate[] pushedPredicates,
       LanceStatistics statistics,
       java.util.Map<String, List<ZoneStats>> zonemapStats,
+      Set<Integer> liveFragmentIds,
       Set<Integer> survivingFragmentIds,
       List<LanceSplit> precomputedSplits,
       java.util.Map<Integer, Long> precomputedFragmentRowCounts,
@@ -159,6 +164,9 @@ public LanceScan(
             : new Predicate[0];
     this.statistics = statistics;
     this.zonemapStats = zonemapStats != null ? zonemapStats : Collections.emptyMap();
+    this.liveFragmentIds =
+        Collections.unmodifiableSet(
+            new HashSet<>(Objects.requireNonNull(liveFragmentIds, "liveFragmentIds")));
     this.cachedSurvivingFragmentIds = survivingFragmentIds;
     this.precomputedSplits = precomputedSplits;
     this.precomputedFragmentRowCounts =
@@ -371,7 +379,8 @@ private List<LanceSplit> pruneByZonemapStats(List<LanceSplit> allSplits) {
       allowedIds = cachedSurvivingFragmentIds;
     } else if (!zonemapStats.isEmpty()) {
       allowedIds =
-          ZonemapFragmentPruner.pruneFragments(pushedPredicates, zonemapStats).orElse(null);
+          ZonemapFragmentPruner.pruneFragments(pushedPredicates, zonemapStats, liveFragmentIds)
+              .orElse(null);
     } else {
       return allSplits;
     }

lance-spark-base_2.12/src/main/java/org/lance/spark/read/LanceScanBuilder.java

@@ -191,6 +191,12 @@ public Scan build() {
           SparkLanceShardingUtils.isEmpty(shardingSpec)
               ? SparkLanceShardingUtils.firstShardingSpec(dataset)
               : shardingSpec;
+
+      // Plan splits before zonemap analysis so live fragment IDs, zonemap stats, and the splits
+      // shipped to workers all come from the same Dataset snapshot.
+      LanceSplit.ScanPlanResult scanPlan = LanceSplit.planScan(dataset, readOptions);
+      Set<Integer> liveFragmentIds = new HashSet<>(scanPlan.getFragmentRowCounts().keySet());
+
       for (ShardingField field : SparkLanceShardingUtils.fields(activeShardingSpec)) {
         columnsToLoad.add(SparkLanceShardingUtils.columnName(field, lanceSchema));
       }
@@ -215,7 +221,8 @@ public Scan build() {
           continue;
         }
         java.util.Optional<Map<Integer, Object>> keys =
-            SparkLanceShardingUtils.detectFragmentKeys(field, lanceSchema, colStats);
+            SparkLanceShardingUtils.detectFragmentKeys(
+                field, lanceSchema, colStats, liveFragmentIds);
         if (keys.isPresent()) {
           fragmentShardingKeys = keys.get();
           activeShardingExpression = SparkLanceShardingUtils.toSparkExpression(field, lanceSchema);
@@ -234,18 +241,23 @@ public Scan build() {
       Set<Integer> survivingFragmentIds = null;
       if (pushedPredicates.length > 0 && !zonemapStats.isEmpty()) {
         survivingFragmentIds =
-            ZonemapFragmentPruner.pruneFragments(pushedPredicates, zonemapStats).orElse(null);
+            ZonemapFragmentPruner.pruneFragments(pushedPredicates, zonemapStats, liveFragmentIds)
+                .orElse(null);
       }
 
       // Scale rows and full size by the zonemap fragment-pruning ratio first, then let
       // LanceStatistics.estimateProjected apply the column-width ratio on top
       // (when the projected schema is narrower than the full schema).
       long projectedRows = summary.getTotalRows();
       long projectedFullSize = summary.getTotalFilesSize();
-      if (survivingFragmentIds != null && summary.getTotalFragments() > 0) {
-        double ratio = (double) survivingFragmentIds.size() / summary.getTotalFragments();
-        projectedRows = (long) (projectedRows * ratio);
-        projectedFullSize = (long) (projectedFullSize * ratio);
+      if (survivingFragmentIds != null && !liveFragmentIds.isEmpty()) {
+        long survivingRows =
+            survivingFragmentIds.stream().mapToLong(scanPlan.getFragmentRowCounts()::get).sum();
+        LanceStatistics postPruning =
+            LanceStatistics.estimatePostPruningByRows(
+                summary.getTotalRows(), summary.getTotalFilesSize(), survivingRows);
+        projectedRows = postPruning.numRows().getAsLong();
+        projectedFullSize = postPruning.sizeInBytes().getAsLong();
       }
       LanceStatistics statistics =
           LanceStatistics.estimateProjected(projectedRows, projectedFullSize, fullSchema, schema);
@@ -254,19 +266,13 @@ public Scan build() {
             "Scan statistics after pruning: {} of {} fragments survive,"
                 + " estimatedSize={}, estimatedRows={} (full: size={}, rows={})",
             survivingFragmentIds.size(),
-            summary.getTotalFragments(),
+            liveFragmentIds.size(),
             statistics.sizeInBytes(),
             statistics.numRows(),
             summary.getTotalFilesSize(),
             summary.getTotalRows());
       }
 
-      // Pre-compute splits and per-fragment row counts from the same Dataset handle that we
-      // already opened above. This consolidates two driver-side opens into one and lets us pin
-      // the resolved version onto the read options shipped to workers, providing snapshot
-      // isolation across all tasks of this query. The version is kept as a long end-to-end so
-      // long-lived high-write-frequency datasets do not silently truncate to a wrong version.
-      LanceSplit.ScanPlanResult scanPlan = LanceSplit.planScan(dataset, readOptions);
       LanceSparkReadOptions resolvedReadOptions = readOptions.withRef(scanPlan.getRef());
 
       Optional<String> whereCondition =
@@ -282,6 +288,7 @@ public Scan build() {
           pushedPredicates,
           statistics,
           zonemapStats,
+          liveFragmentIds,
           survivingFragmentIds,
           scanPlan.getSplits(),
           scanPlan.getFragmentRowCounts(),

lance-spark-base_2.12/src/main/java/org/lance/spark/read/LanceStatistics.java

@@ -67,6 +67,30 @@ public static LanceStatistics estimatePostPruning(
     return new LanceStatistics((long) (totalRows * ratio), (long) (totalFilesSize * ratio));
   }
 
+  /**
+   * Estimate post-pruning statistics from the exact row count of surviving fragments.
+   *
+   * <p>The row count is exact for the fragments selected by planning. File size remains an estimate
+   * because the scan plan does not carry per-fragment byte sizes, so it is scaled by the
+   * surviving-row ratio. Invalid or non-selective inputs conservatively retain full-table stats.
+   *
+   * @param totalRows total rows in the dataset
+   * @param totalFilesSize total file size in bytes
+   * @param survivingRows exact row count across surviving fragments
+   * @return row-weighted post-pruning statistics
+   */
+  static LanceStatistics estimatePostPruningByRows(
+      long totalRows, long totalFilesSize, long survivingRows) {
+    if (totalRows <= 0 || survivingRows >= totalRows) {
+      return new LanceStatistics(totalRows, totalFilesSize);
+    }
+    if (survivingRows <= 0) {
+      return new LanceStatistics(0, 0);
+    }
+    double ratio = (double) survivingRows / totalRows;
+    return new LanceStatistics(survivingRows, (long) (totalFilesSize * ratio));
+  }
+
   /**
    * Estimate post-projection size using {@code sizeInBytes × (projectedWidths / fullWidths)}, the
    * same formula Spark's DSv2 {@code FileScan.estimateStatistics} applies after column pruning (see