[RLC] Case 1 — conflict-time data skipping (disjoint data ranges)

protocol_rfcs/iceberg-v4-metadata.md

@@ -424,7 +424,7 @@ Summary information for `content_type` = DATA_MANIFEST entries. Includes file/ro
 
 ### Manifest Deletion Vectors (MDVs)
 
-MDVs track which entries in a leaf manifest have been deleted or replaced. The `dv` field in `manifest_info` contains a Roaring bitmap of all invalidated positions (cumulative, used for reading).  The `deleted_positions` and `replaced_positions` fields in `tracking` represent what changed in the current commit only (used for CDF):
+MDVs track which entries in a leaf manifest have been deleted or replaced. The `dv` field in `manifest_info` contains all invalidated positions (cumulative, used for reading). The `deleted_positions` and `replaced_positions` fields in `tracking` represent what changed in the current commit only (used for CDF). All three bitmaps are encoded using the [Mumbling bitmap](https://iceberg.apache.org/mumbling-spec/) spec:
 
 - **`manifest_info.dv`**: Cumulative set of all invalidated positions. Grows
   monotonically. Used by readers to skip entries.

spark-unified/src/test/java/io/delta/spark/internal/v2/DeltaV2TestBase.java

@@ -15,16 +15,22 @@
  */
 package io.delta.spark.internal.v2;
 
-import io.delta.kernel.defaults.engine.DefaultEngine;
 import io.delta.kernel.engine.Engine;
+import io.delta.spark.internal.v2.kernel.KernelContext;
+import io.delta.spark.internal.v2.utils.ScalaUtils;
+import java.util.HashMap;
+import java.util.Map;
+import org.apache.hadoop.conf.Configuration;
 import org.apache.spark.sql.SparkSession;
+import org.apache.spark.sql.delta.storage.LogStore;
 import org.junit.jupiter.api.AfterAll;
 import org.junit.jupiter.api.BeforeAll;
 
 public abstract class DeltaV2TestBase {
 
   protected static SparkSession spark;
   protected static Engine defaultEngine;
+  protected static KernelContext defaultKernelContext;
 
   @BeforeAll
   public static void setUpSparkAndEngine() {
@@ -39,7 +45,18 @@ public static void setUpSparkAndEngine() {
             .config("spark.sql.catalog.dsv2", "io.delta.spark.internal.v2.catalog.TestCatalog")
             .config("spark.sql.catalog.dsv2.base_path", System.getProperty("java.io.tmpdir"))
             .getOrCreate();
-    defaultEngine = DefaultEngine.create(spark.sessionState().newHadoopConf());
+    defaultKernelContext = createKernelContext();
+    defaultEngine = defaultKernelContext.getDefaultEngine();
+  }
+
+  protected static KernelContext createKernelContext() {
+    return createKernelContext(spark.sessionState().newHadoopConf());
+  }
+
+  protected static KernelContext createKernelContext(Configuration hadoopConf) {
+    Map<String, String> options = new HashMap<>();
+    hadoopConf.forEach(entry -> options.put(entry.getKey(), entry.getValue()));
+    return new KernelContext(ScalaUtils.toScalaMap(options), LogStore.apply(spark));
   }
 
   @AfterAll

spark-unified/src/test/scala-shims/spark-4.2/io/delta/spark/internal/v2/read/changelog/DeltaV2ChangelogCatalogIntegrationTest.java

@@ -107,7 +107,7 @@ private java.sql.Timestamp commitTimestamp(String tableLocation, long version) {
     // Reads through the kernel snapshot manager, so it works inside the STRICT block the test body
     // runs in. tableLocation is resolved by withHistoryTable before STRICT is set.
     DeltaV2SnapshotManager snapshotManager =
-        SnapshotManagerFactory.create(tableLocation, defaultEngine, Optional.empty());
+        SnapshotManagerFactory.create(tableLocation, defaultKernelContext, Optional.empty());
     long millis = snapshotManager.loadSnapshotAt(version).timestamp();
     return new java.sql.Timestamp(millis);
   }

spark-unified/src/test/scala-shims/spark-4.2/io/delta/spark/internal/v2/read/changelog/DeltaV2ChangelogDirectBatchExecutionTest.java

@@ -77,7 +77,7 @@ public void testDirectBatchExecutionWithExplicitExpectedRows(boolean enableDvs)
           spark.sql(String.format("DELETE FROM %s WHERE id = 1", tableName));
 
           DeltaV2SnapshotManager snapshotManager =
-              SnapshotManagerFactory.create(tablePath, defaultEngine, Optional.empty());
+              SnapshotManagerFactory.create(tablePath, defaultKernelContext, Optional.empty());
           StructType dataSchema = spark.table(tableName).schema();
           long latestVersion = snapshotManager.loadLatestSnapshot().version();
           Map<Long, Long> commitTimestampsMicros = loadCommitTimestampsMicros(tableName);
@@ -149,7 +149,7 @@ public void testRowTrackingMetadataReferences() throws Exception {
           spark.sql(String.format("INSERT INTO %s VALUES (1)", tableName));
 
           DeltaV2SnapshotManager snapshotManager =
-              SnapshotManagerFactory.create(tablePath, defaultEngine, Optional.empty());
+              SnapshotManagerFactory.create(tablePath, defaultKernelContext, Optional.empty());
           StructType dataSchema = spark.table(tableName).schema();
           long latestVersion = snapshotManager.loadLatestSnapshot().version();
 
@@ -200,7 +200,7 @@ public void testUpdateProducesPairedDeleteAndInsert(boolean enableDvs) throws Ex
           spark.sql(String.format("UPDATE %s SET name = 'AliceX' WHERE id = 1", tableName));
 
           DeltaV2SnapshotManager snapshotManager =
-              SnapshotManagerFactory.create(tablePath, defaultEngine, Optional.empty());
+              SnapshotManagerFactory.create(tablePath, defaultKernelContext, Optional.empty());
           StructType dataSchema = spark.table(tableName).schema();
           long latestVersion = snapshotManager.loadLatestSnapshot().version();
           Map<Long, Long> commitTimestampsMicros = loadCommitTimestampsMicros(tableName);
@@ -273,7 +273,7 @@ public void testRangeSlicingNonZeroStart() throws Exception {
           spark.sql(String.format("INSERT INTO %s VALUES (3, 'Charlie')", tableName));
 
           DeltaV2SnapshotManager snapshotManager =
-              SnapshotManagerFactory.create(tablePath, defaultEngine, Optional.empty());
+              SnapshotManagerFactory.create(tablePath, defaultKernelContext, Optional.empty());
           StructType dataSchema = spark.table(tableName).schema();
           Map<Long, Long> commitTimestampsMicros = loadCommitTimestampsMicros(tableName);
 
@@ -342,7 +342,7 @@ public void testChangelogWithReservedCharsInPath() throws Exception {
           spark.sql(String.format("INSERT INTO %s VALUES (1, 'Alice'), (2, 'Bob')", tableName));
 
           DeltaV2SnapshotManager snapshotManager =
-              SnapshotManagerFactory.create(tablePath, defaultEngine, Optional.empty());
+              SnapshotManagerFactory.create(tablePath, defaultKernelContext, Optional.empty());
           long latestVersion = snapshotManager.loadLatestSnapshot().version();
           Map<Long, Long> commitTimestampsMicros = loadCommitTimestampsMicros(tableName);
 
@@ -407,7 +407,7 @@ public void testChangelogWithReservedCharsInDataFileNames() throws Exception {
           spark.sql(String.format("INSERT INTO %s VALUES (1, 'Alice'), (2, 'Bob')", tableName));
 
           DeltaV2SnapshotManager snapshotManager =
-              SnapshotManagerFactory.create(tablePath, defaultEngine, Optional.empty());
+              SnapshotManagerFactory.create(tablePath, defaultKernelContext, Optional.empty());
           long latestVersion = snapshotManager.loadLatestSnapshot().version();
           Map<Long, Long> commitTimestampsMicros = loadCommitTimestampsMicros(tableName);
 

spark-unified/src/test/scala-shims/spark-4.2/io/delta/spark/internal/v2/read/changelog/DeltaV2ChangelogTestBase.java

@@ -15,7 +15,6 @@
  */
 package io.delta.spark.internal.v2.read.changelog;
 
-import io.delta.kernel.defaults.engine.DefaultEngine;
 import io.delta.spark.internal.v2.DeltaV2TestBase;
 import org.apache.spark.sql.SparkSession;
 import org.junit.jupiter.api.BeforeAll;
@@ -57,6 +56,7 @@ public static void setUpChangelogSparkAndEngine() {
             .config("spark.sql.catalog.dsv2.base_path", System.getProperty("java.io.tmpdir"))
             .config("spark.databricks.delta.changelogV2.enabled", "true")
             .getOrCreate();
-    defaultEngine = DefaultEngine.create(spark.sessionState().newHadoopConf());
+    defaultKernelContext = createKernelContext();
+    defaultEngine = defaultKernelContext.getDefaultEngine();
   }
 }

spark-unified/src/test/scala/io/delta/internal/ApplyV2ReadOptionsSuite.scala

@@ -34,6 +34,7 @@ import org.apache.spark.sql.delta.sources.{DeltaSourceMetadataTrackingLog, Delta
 import org.apache.spark.sql.delta.test.DeltaSQLCommandTest
 import org.apache.spark.sql.util.CaseInsensitiveStringMap
 import io.delta.spark.internal.v2.catalog.DeltaV2Table
+import io.delta.spark.internal.v2.kernel.KernelContext
 import io.delta.spark.internal.v2.snapshot.PathBasedSnapshotManager
 import io.delta.storage.commit.uccommitcoordinator.UCCommitCoordinatorClient
 
@@ -270,7 +271,7 @@ class ApplyV2ReadOptionsSuite extends DeltaSQLCommandTest {
   private def seedSchemaLogWithExtraColumn(tablePath: String, schemaLogPath: String): Unit = {
     val deltaLog = DeltaLog.forTable(spark, tablePath)
     val snapshotManager =
-      new PathBasedSnapshotManager(tablePath, deltaLog.newDeltaHadoopConf())
+      new PathBasedSnapshotManager(tablePath, new KernelContext(deltaLog.options, deltaLog.store))
     val tableId = snapshotManager.loadLatestSnapshot.metadata.getId
     val trackingLog = DeltaSourceMetadataTrackingLog.create(
       spark, schemaLogPath, tableId, tablePath, parameters = Map.empty[String, String])

spark-unified/src/test/scala/io/delta/spark/internal/v2/read/DeltaV2LimitPushdownE2ESuite.scala

@@ -303,6 +303,29 @@ private[read] trait DeltaV2LimitPushdownE2ETests {
     }
   }
 
+  test("disabling Delta limit pushdown keeps LIMIT off the scan") {
+    val table = "v2_scan_e2e_limit_pushdown_disabled"
+    withTable(table) {
+      withV1Mode {
+        sql(s"CREATE TABLE $table (id INT) USING $tableProvider")
+        // One row per file so that a pushed limit would select fewer files than the table has.
+        (0 until 5).foreach { id =>
+          sql(s"INSERT INTO $table VALUES ($id)")
+        }
+      }
+
+      withSQLConf(DeltaSQLConf.DELTA_LIMIT_PUSHDOWN_ENABLED.key -> "false") {
+        val df = sql(s"SELECT * FROM $table LIMIT 2")
+        assert(df.collect().length == 2, "LIMIT 2 should return exactly 2 rows")
+        assertExpectedNoPushedLimit(df, "LIMIT with Delta limit pushdown disabled")
+        assertRouteSpecificInputFileCount(
+          df,
+          expected = 5,
+          context = "LIMIT with Delta limit pushdown disabled reads every file")
+      }
+    }
+  }
+
   test("a non-deterministic filter prevents LIMIT pushdown") {
     val table = "v2_scan_e2e_limit_nondeterministic_filter"
     withTable(table) {

spark-unified/src/test/scala/io/delta/spark/internal/v2/read/DeltaV2ReadE2ESuite.scala

@@ -266,6 +266,28 @@ private[read] trait DeltaV2ReadE2ETests {
     }
   }
 
+  test("partition names with spaces and backticks under column mapping prune on V2") {
+    val table = "v2_scan_e2e_spaced_partitions"
+    withTable(table) {
+      withV1Mode {
+        sql(
+          s"CREATE TABLE $table (id LONG, `order date` STRING, `a``b` STRING) " +
+            s"USING $tableProvider PARTITIONED BY (`order date`, `a``b`) " +
+            "TBLPROPERTIES ('delta.columnMapping.mode' = 'name')")
+        Seq("(1, 'd1', 'x')", "(2, 'd2', 'x')", "(3, 'd1', 'y')").foreach { values =>
+          sql(s"INSERT INTO $table VALUES $values")
+        }
+      }
+      // Spark 4.0 keeps the partition filter above the scan, so the scan reads all 3 files.
+      checkPartitionPrunedReadMatchesV1(
+        s"SELECT id, `order date`, `a``b` FROM $table " +
+          "WHERE `order date` = 'd1' AND `a``b` = 'y'",
+        expectedRows = Seq(Row(3L, "d1", "y")),
+        expectedSelectedFiles = if (spark.version.startsWith("4.0")) 3 else 1,
+        context = "space and backtick partition-column filter")
+    }
+  }
+
   test("deletion vectors with a middle partition column preserve rows and DDL order") {
     val table = "v2_scan_e2e_dv_partition_middle"
     withTable(table) {
@@ -422,6 +444,25 @@ private[read] trait DeltaV2ReadE2ETests {
       }
     }
   }
+
+  private def checkPartitionPrunedReadMatchesV1(
+      query: String,
+      expectedRows: Seq[Row],
+      expectedSelectedFiles: Int,
+      context: String): Unit = {
+    withV1Mode {
+      checkAnswer(sql(query), expectedRows)
+    }
+    checkRead(context)(sql(query)) { df =>
+      checkAnswer(df, expectedRows)
+    }
+    withDeltaV2ScanForSelectedFiles(sql(query), context) { scan =>
+      val numSelectedFiles = scan.preparedScan().files.size
+      assert(
+        numSelectedFiles == expectedSelectedFiles,
+        s"expected $expectedSelectedFiles selected files for $context, got $numSelectedFiles")
+    }
+  }
 }
 
 /** Covers read, schema, column-mapping, and deletion-vector semantics shared across Delta V2

spark-unified/src/test/scala/org/apache/spark/sql/delta/v2/interop/DeltaV2OptimisticTransactionSuite.scala

@@ -21,8 +21,9 @@ import java.nio.file.{Files, StandardCopyOption}
 
 import org.apache.spark.sql.delta.{DeltaColumnMapping, DeltaOperations, OptimisticTransaction, OptimisticTransactionSuite}
 import org.apache.spark.sql.delta.actions.{AddFile, Metadata, Protocol, SetTransaction}
+import org.apache.spark.sql.delta.storage.LogStore
 import org.apache.spark.sql.delta.test.V2ForceTest
-import io.delta.spark.internal.v2.kernel.KernelEngineFactory
+import io.delta.spark.internal.v2.kernel.KernelContext
 import org.apache.hadoop.fs.Path
 import io.delta.kernel.{Table => KernelTable}
 import io.delta.kernel.internal.{SnapshotImpl => KernelSnapshotImpl}
@@ -48,15 +49,14 @@ class DeltaV2OptimisticTransactionSuite
    * inherited V1 conflict test cases.
    */
   override protected def startTestTransaction(dataPath: Path): OptimisticTransaction = {
-    // scalastyle:off deltahadoopconfiguration
-    val kernelEngine = KernelEngineFactory.createDefaultEngine(spark.sessionState.newHadoopConf())
-    // scalastyle:on deltahadoopconfiguration
+    val kernelContext = new KernelContext(Map.empty, LogStore(spark))
+    val kernelEngine = kernelContext.getDefaultEngine()
     val kernelSnap = KernelTable
       .forPath(kernelEngine, dataPath.toString)
       .getLatestSnapshot(kernelEngine)
       .asInstanceOf[KernelSnapshotImpl]
-    val deltaV2Snapshot = new DeltaV2Snapshot(kernelSnap)
-    new DeltaV2OptimisticTransaction(catalogTable = None, deltaV2Snapshot, kernelEngine)
+    val deltaV2Snapshot = new DeltaV2Snapshot(kernelSnap, kernelContext)
+    new DeltaV2OptimisticTransaction(catalogTable = None, deltaV2Snapshot, kernelContext)
   }
 
   override protected def testDefaultMetadata(): Metadata =
@@ -93,27 +93,24 @@ class DeltaV2OptimisticTransactionSuite
 
   /** Builds a Kernel-backed transaction over the latest snapshot of the table at `dir`. */
   private def startKernelTxn(dir: File): DeltaV2OptimisticTransaction = {
-    // scalastyle:off deltahadoopconfiguration
-    // No DeltaLog here (the snapshot is loaded via Kernel), so use the session Hadoop conf.
-    val kernelEngine = KernelEngineFactory.createDefaultEngine(spark.sessionState.newHadoopConf())
-    // scalastyle:on deltahadoopconfiguration
+    val kernelContext = new KernelContext(Map.empty, LogStore(spark))
+    val kernelEngine = kernelContext.getDefaultEngine()
     val kernelSnap = KernelTable
       .forPath(kernelEngine, dir.getCanonicalPath)
       .getLatestSnapshot(kernelEngine)
       .asInstanceOf[KernelSnapshotImpl]
-    val deltaV2Snapshot = new DeltaV2Snapshot(kernelSnap)
-    new DeltaV2OptimisticTransaction(catalogTable = None, deltaV2Snapshot, kernelEngine)
+    val deltaV2Snapshot = new DeltaV2Snapshot(kernelSnap, kernelContext)
+    new DeltaV2OptimisticTransaction(catalogTable = None, deltaV2Snapshot, kernelContext)
   }
 
   private def latestKernelSnapshot(dir: File): DeltaV2Snapshot = {
-    // scalastyle:off deltahadoopconfiguration
-    val kernelEngine = KernelEngineFactory.createDefaultEngine(spark.sessionState.newHadoopConf())
-    // scalastyle:on deltahadoopconfiguration
+    val kernelContext = new KernelContext(Map.empty, LogStore(spark))
+    val kernelEngine = kernelContext.getDefaultEngine()
     val kernelSnap = KernelTable
       .forPath(kernelEngine, dir.getCanonicalPath)
       .getLatestSnapshot(kernelEngine)
       .asInstanceOf[KernelSnapshotImpl]
-    new DeltaV2Snapshot(kernelSnap)
+    new DeltaV2Snapshot(kernelSnap, kernelContext)
   }
 
   /** Seeds a simple (unpartitioned) V1 Delta table at `dir`. */