[902] Move the build to Spark 3.5 and Delta 3.3

demo/jupyter/predef.sc

@@ -50,7 +50,7 @@ def xtableVersion(key: String): String =
     ("org.apache.spark", s"spark-hive_$scalaBinaryVersion", xtableVersion("spark.version")),
     ("org.apache.hudi", s"hudi-spark$sparkVersionPrefix-bundle_$scalaBinaryVersion", xtableVersion("hudi.version")),
     ("org.apache.hudi", "hudi-java-client", xtableVersion("hudi.version")),
-    ("io.delta", s"delta-core_$scalaBinaryVersion", xtableVersion("delta.version")),
+    ("io.delta", s"delta-spark_$scalaBinaryVersion", xtableVersion("delta.version")),
     ("io.delta", "delta-kernel-api", xtableVersion("delta.kernel.version")),
     ("io.delta", "delta-kernel-defaults", xtableVersion("delta.kernel.version")),
     ("org.apache.iceberg", "iceberg-hive-runtime", xtableVersion("iceberg.hive.runtime.version")),

demo/jupyter/prefetch_dependencies.sh

@@ -60,7 +60,7 @@ coursier fetch --quiet \
   "org.apache.spark:spark-hive_${SCALA_BINARY_VERSION}:${SPARK_VERSION}" \
   "org.apache.hudi:hudi-spark${SPARK_VERSION_PREFIX}-bundle_${SCALA_BINARY_VERSION}:${HUDI_VERSION}" \
   "org.apache.hudi:hudi-java-client:${HUDI_VERSION}" \
-  "io.delta:delta-core_${SCALA_BINARY_VERSION}:${DELTA_VERSION}" \
+  "io.delta:delta-spark_${SCALA_BINARY_VERSION}:${DELTA_VERSION}" \
   "io.delta:delta-kernel-api:${DELTA_KERNEL_VERSION}" \
   "io.delta:delta-kernel-defaults:${DELTA_KERNEL_VERSION}" \
   "org.apache.iceberg:iceberg-hive-runtime:${ICEBERG_HIVE_RUNTIME_VERSION}" \

pom.xml

@@ -85,11 +85,11 @@
         <scala13.version>2.13.15</scala13.version>
         <scala.version>${scala12.version}</scala.version>
         <scala.binary.version>2.12</scala.binary.version>
-        <spark.version>3.4.2</spark.version>
-        <spark.version.prefix>3.4</spark.version.prefix>
+        <spark.version>3.5.9</spark.version>
+        <spark.version.prefix>3.5</spark.version.prefix>
         <iceberg.version>1.9.2</iceberg.version>
         <iceberg.hive.runtime.version>1.7.2</iceberg.hive.runtime.version>
-        <delta.version>2.4.0</delta.version>
+        <delta.version>3.3.3</delta.version>
         <paimon.version>1.3.1</paimon.version>
         <jackson.version>2.18.2</jackson.version>
         <spotless.version>2.46.1</spotless.version>
@@ -328,7 +328,7 @@
             <!-- Delta -->
             <dependency>
                 <groupId>io.delta</groupId>
-                <artifactId>delta-core_${scala.binary.version}</artifactId>
+                <artifactId>delta-spark_${scala.binary.version}</artifactId>
                 <version>${delta.version}</version>
             </dependency>
             <dependency>
@@ -756,6 +756,9 @@
                     <skip>${skipUTs}</skip>
                     <redirectTestOutputToFile>true</redirectTestOutputToFile>
                     <trimStackTrace>false</trimStackTrace>
+                    <!-- Hudi stamps its timeline in UTC, so a JVM on any other zone reads its own
+                         commits as not yet completed. Pin the tests to UTC. -->
+                    <argLine>-Duser.timezone=UTC</argLine>
                     <forkedProcessExitTimeoutInSeconds>120</forkedProcessExitTimeoutInSeconds>
                 </configuration>
             </plugin>
@@ -776,7 +779,7 @@
                     <reuseForks>false</reuseForks>
                     <forkCount>6</forkCount>
                     <trimStackTrace>false</trimStackTrace>
-                    <argLine>-Xmx1500m</argLine>
+                    <argLine>-Xmx1500m -Duser.timezone=UTC</argLine>
                     <forkedProcessExitTimeoutInSeconds>120</forkedProcessExitTimeoutInSeconds>
                 </configuration>
             </plugin>

website/docs/hms.md

@@ -148,7 +148,7 @@ SELECT * FROM <database_name>.<table_name>;
 <TabItem value="delta">
 
 ```shell md title="shell"
-spark-sql --packages io.delta:delta-core_2.12:2.0.0 \
+spark-sql --packages io.delta:delta-spark_2.12:3.3.3 \
 --conf "spark.sql.extensions=io.delta.sql.DeltaSparkSessionExtension" \
 --conf "spark.sql.catalog.spark_catalog=org.apache.spark.sql.delta.catalog.DeltaCatalog" \
 --conf "spark.sql.catalogImplementation=hive"

website/docs/how-to.md

@@ -65,7 +65,7 @@ pyspark \
 
 ```shell md title="shell"
 pyspark \
-  --packages io.delta:delta-core_2.12:2.1.0 \
+  --packages io.delta:delta-spark_2.12:3.3.3 \
   --conf "spark.sql.extensions=io.delta.sql.DeltaSparkSessionExtension" \
   --conf "spark.sql.catalog.spark_catalog=org.apache.spark.sql.delta.catalog.DeltaCatalog"
 ```

xtable-core/pom.xml

@@ -106,7 +106,7 @@
         <!-- Delta dependencies -->
         <dependency>
             <groupId>io.delta</groupId>
-            <artifactId>delta-core_${scala.binary.version}</artifactId>
+            <artifactId>delta-spark_${scala.binary.version}</artifactId>
             <scope>provided</scope>
         </dependency>
         <dependency>

xtable-core/src/main/java/org/apache/xtable/delta/DeltaConversionSource.java

@@ -73,7 +73,7 @@
   @Builder.Default
   private final DeltaTableExtractor tableExtractor = DeltaTableExtractor.builder().build();
 
   private Optional<DeltaIncrementalChangesState> deltaIncrementalChangesState = Optional.empty();
 
   private final SparkSession sparkSession;
   private final DeltaLog deltaLog;
@@ -129,7 +129,8 @@
               getChangesState().getCommitTimestamp(versionNumber));
       fileFormat = actionsConverter.convertToFileFormat(metadataAtVersion.format().provider());
     } else {
-      Snapshot snapshotAtVersion = deltaLog.getSnapshotAt(versionNumber, Option.empty());
+      Snapshot snapshotAtVersion =
+          deltaLog.getSnapshotAt(versionNumber, Option.empty(), Option.empty());
       tableAtVersion = tableExtractor.table(snapshotAtVersion, tableName);
       fileFormat =
           actionsConverter.convertToFileFormat(snapshotAtVersion.metadata().format().provider());
@@ -262,7 +263,8 @@
     }
     if (cachedMetadata == null) {
       // baseline; the snapshot is read for its metadata and not retained
-      cachedMetadata = deltaLog.getSnapshotAt(versionNumber, Option.empty()).metadata();
+      cachedMetadata =
+          deltaLog.getSnapshotAt(versionNumber, Option.empty(), Option.empty()).metadata();
     } else if (metadataInCommit != null) {
       cachedMetadata = metadataInCommit;
     }

xtable-core/src/main/java/org/apache/xtable/delta/DeltaConversionTarget.java

@@ -130,7 +130,7 @@ private void _init(
     DeltaLog deltaLog = DeltaLog.forTable(sparkSession, tableDataPath);
     boolean deltaTableExists = deltaLog.tableExists();
     if (!deltaTableExists) {
-      deltaLog.ensureLogDirectoryExist();
+      deltaLog.createLogDirectoriesIfNotExists();
     }
     this.schemaExtractor = schemaExtractor;
     this.partitionExtractor = partitionExtractor;

xtable-core/src/main/java/org/apache/xtable/delta/DeltaDataFileUpdatesExtractor.java

@@ -34,6 +34,7 @@
 import org.apache.spark.sql.delta.actions.Action;
 import org.apache.spark.sql.delta.actions.AddFile;
 
+import scala.Option;
 import scala.collection.JavaConverters;
 import scala.collection.Seq;
 
@@ -125,7 +126,12 @@ private Stream<AddFile> createAddFileAction(
             true,
             getColumnStats(schema, dataFile.getRecordCount(), dataFile.getColumnStats()),
             null,
-            null));
+            null,
+            // baseRowId, defaultRowCommitVersion and clusteringProvider: row tracking and
+            // clustering are not written by the sync, so all three are left unset.
+            Option.empty(),
+            Option.empty(),
+            Option.empty()));
   }
 
   private String getColumnStats(

xtable-core/src/main/java/org/apache/xtable/delta/DeltaTableExtractor.java

@@ -44,7 +44,7 @@ public class DeltaTableExtractor {
   private static final DeltaSchemaExtractor schemaExtractor = DeltaSchemaExtractor.getInstance();
 
   public InternalTable table(DeltaLog deltaLog, String tableName, Long version) {
-    Snapshot snapshot = deltaLog.getSnapshotAt(version, Option.empty());
+    Snapshot snapshot = deltaLog.getSnapshotAt(version, Option.empty(), Option.empty());
     return table(snapshot, tableName);
   }
 

xtable-core/src/test/java/org/apache/xtable/delta/ITDeltaConversionSource.java

@@ -26,9 +26,7 @@
 import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.mockito.ArgumentMatchers.any;
-import static org.mockito.ArgumentMatchers.anyBoolean;
 import static org.mockito.ArgumentMatchers.anyLong;
-import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.spy;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
@@ -887,7 +885,7 @@ public void getTableChangeForCommitReconstructsSnapshotOnceForAppendOnlyBacklog(
     assertTrue(
         commitsBacklog.getCommitsToProcess().size() >= 2,
         "backlog must span multiple commits for this assertion to be meaningful");
-    verify(spiedDeltaLog, times(1)).getSnapshotAt(anyLong(), any());
+    verify(spiedDeltaLog, times(1)).getSnapshotAt(anyLong(), any(), any());
   }
 
   @Test
@@ -943,7 +941,7 @@ public void schemaChangeMidBacklogRefreshesCachedMetadataWithoutSnapshotReload()
     }
     // The refresh comes from the commit's own metaData action, not a snapshot reload: the only
     // snapshot read is the baseline at the first commit.
-    verify(spiedDeltaLog, times(1)).getSnapshotAt(anyLong(), any());
+    verify(spiedDeltaLog, times(1)).getSnapshotAt(anyLong(), any(), any());
   }
 
   @Test
@@ -984,10 +982,11 @@ public void reuseMetadataDisabledReconstructsSnapshotPerCommit() {
 
     ValidationTestHelper.validateTableChanges(allActiveFiles, allTableChanges);
     // With reuse disabled the pre-optimization behaviour is restored in full: one snapshot
-    // reconstruction per commit, and no second listing of the log for commit-file mtimes.
+    // reconstruction per commit. The absence of the extra mtime listing is no longer assertable
+    // here, because DeltaLog.getChanges itself delegates to getChangeLogFiles as of Delta 3.x,
+    // so the spy cannot tell that call apart from one this class makes.
     verify(spiedDeltaLog, times(commitsBacklog.getCommitsToProcess().size()))
-        .getSnapshotAt(anyLong(), any());
-    verify(spiedDeltaLog, never()).getChangeLogFiles(anyLong(), anyBoolean());
+        .getSnapshotAt(anyLong(), any(), any());
   }
 
   @Test

xtable-core/src/test/java/org/apache/xtable/delta/ITDeltaDeleteVectorConvert.java

@@ -104,6 +104,10 @@ public void testInsertsUpsertsAndDeletes() {
             "ALTER TABLE "
                 + tableName
                 + " SET TBLPROPERTIES ('delta.enableDeletionVectors' = true)");
+    // The DeltaTable handle resolved at construction still carries the pre-ALTER protocol, and
+    // as of Delta 3.x a merge planned against it does not see deletion vectors as readable, so
+    // the row_index metadata column the DV write path needs is never exposed.
+    testSparkDeltaTable.reload();
 
     List<List<String>> allActiveFiles = new ArrayList<>();
     List<TableChange> allTableChanges = new ArrayList<>();
@@ -185,7 +189,7 @@ public void testInsertsUpsertsAndDeletes() {
   private void validateDeletedRecordCount(
       DeltaLog deltaLog, int version, int deleteVectorFileCount, int deletionRecordCount) {
     List<AddFile> allFiles =
-        deltaLog.getSnapshotAt(version, Option.empty()).allFiles().collectAsList();
+        deltaLog.getSnapshotAt(version, Option.empty(), Option.empty()).allFiles().collectAsList();
     List<AddFile> filesWithDeletionVectors =
         allFiles.stream().filter(f -> f.deletionVector() != null).collect(Collectors.toList());
 

xtable-core/src/test/java/org/apache/xtable/delta/TestDeltaActionsConverter.java

@@ -51,15 +51,37 @@ void extractDeletionVector() throws URISyntaxException {
 
     DeletionVectorDescriptor deletionVector = null;
     AddFile addFileAction =
-        new AddFile(filePath, null, size, time, dataChange, stats, null, deletionVector);
+        new AddFile(
+            filePath,
+            null,
+            size,
+            time,
+            dataChange,
+            stats,
+            null,
+            deletionVector,
+            Option.empty(),
+            Option.empty(),
+            Option.empty());
     Assertions.assertNull(actionsConverter.extractDeletionVectorFile(snapshot, addFileAction));
 
     deletionVector =
         DeletionVectorDescriptor.onDiskWithAbsolutePath(
             filePath, size, 42, Option.empty(), Option.empty());
 
     addFileAction =
-        new AddFile(filePath, null, size, time, dataChange, stats, null, deletionVector);
+        new AddFile(
+            filePath,
+            null,
+            size,
+            time,
+            dataChange,
+            stats,
+            null,
+            deletionVector,
+            Option.empty(),
+            Option.empty(),
+            Option.empty());
 
     Assertions.assertEquals(
         filePath, actionsConverter.extractDeletionVectorFile(snapshot, addFileAction));

xtable-core/src/test/java/org/apache/xtable/delta/TestDeltaStatsExtractor.java

@@ -36,6 +36,8 @@
 
 import org.apache.spark.sql.delta.actions.AddFile;
 
+import scala.Option;
+
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.ObjectMapper;
 
@@ -132,7 +134,19 @@ void roundTripStatsConversion() throws IOException {
     String stats =
         DeltaStatsExtractor.getInstance()
             .convertStatsToDeltaFormat(schema, numRecords, columnStats);
-    AddFile addFile = new AddFile("file://path/to/file", null, 0, 0, true, stats, null, null);
+    AddFile addFile =
+        new AddFile(
+            "file://path/to/file",
+            null,
+            0,
+            0,
+            true,
+            stats,
+            null,
+            null,
+            Option.empty(),
+            Option.empty(),
+            Option.empty());
     DeltaStatsExtractor extractor = DeltaStatsExtractor.getInstance();
     FileStats actual = extractor.getColumnStatsForFile(addFile, fields);
     List<ColumnStat> actualColumStats = actual.getColumnStats();
@@ -168,7 +182,19 @@ void convertStatsToInternalRepresentation() throws IOException {
     deltaStats.put("tightBounds", Boolean.TRUE);
     deltaStats.put("nonExisting", minValues);
     String stats = MAPPER.writeValueAsString(deltaStats);
-    AddFile addFile = new AddFile("file://path/to/file", null, 0, 0, true, stats, null, null);
+    AddFile addFile =
+        new AddFile(
+            "file://path/to/file",
+            null,
+            0,
+            0,
+            true,
+            stats,
+            null,
+            null,
+            Option.empty(),
+            Option.empty(),
+            Option.empty());
     DeltaStatsExtractor extractor = DeltaStatsExtractor.getInstance();
     FileStats actual = extractor.getColumnStatsForFile(addFile, fields);
     List<ColumnStat> actualColumStats = actual.getColumnStats();
@@ -211,7 +237,19 @@ void convertStatsToInternalRepresentation() throws IOException {
   @Test
   void convertNullStatsToInternalRepresentation() {
     List<InternalField> fields = getSchemaFields();
-    AddFile addFile = new AddFile("file://path/to/file", null, 0, 0, true, null, null, null);
+    AddFile addFile =
+        new AddFile(
+            "file://path/to/file",
+            null,
+            0,
+            0,
+            true,
+            null,
+            null,
+            null,
+            Option.empty(),
+            Option.empty(),
+            Option.empty());
     DeltaStatsExtractor extractor = DeltaStatsExtractor.getInstance();
     FileStats actual = extractor.getColumnStatsForFile(addFile, fields);
     List<ColumnStat> actualColumStats = actual.getColumnStats();

xtable-core/src/test/java/org/apache/xtable/delta/TestDeltaSync.java

@@ -358,7 +358,7 @@ public void testTimestampPartitioning(PartitionTransformType transformType) thro
             .expr();
     org.apache.spark.sql.delta.DeltaLog deltaLog =
         org.apache.spark.sql.delta.DeltaLog.forTable(sparkSession, basePath.toString());
-    org.apache.spark.sql.delta.Snapshot snapshot = deltaLog.getSnapshotAtInit().snapshot();
+    org.apache.spark.sql.delta.Snapshot snapshot = deltaLog.snapshot();
     Seq<org.apache.spark.sql.catalyst.expressions.Expression> expressionSeq =
         scala.collection.JavaConversions.asScalaBuffer(Collections.singletonList(expression));
     Seq<org.apache.spark.sql.catalyst.expressions.Expression> translatedExpression =

xtable-hudi-support/xtable-hudi-support-extensions/pom.xml

@@ -133,7 +133,7 @@
         </dependency>
         <dependency>
             <groupId>io.delta</groupId>
-            <artifactId>delta-core_${scala.binary.version}</artifactId>
+            <artifactId>delta-spark_${scala.binary.version}</artifactId>
             <scope>test</scope>
         </dependency>
         <dependency>

xtable-service/pom.xml

@@ -216,7 +216,7 @@
         <!-- Delta test dependencies -->
         <dependency>
             <groupId>io.delta</groupId>
-            <artifactId>delta-core_${scala.binary.version}</artifactId>
+            <artifactId>delta-spark_${scala.binary.version}</artifactId>
             <scope>test</scope>
         </dependency>
         <dependency>

xtable-utilities/pom.xml

@@ -196,7 +196,7 @@
         </dependency>
         <dependency>
             <groupId>io.delta</groupId>
-            <artifactId>delta-core_${scala.binary.version}</artifactId>
+            <artifactId>delta-spark_${scala.binary.version}</artifactId>
         </dependency>
         <dependency>
             <groupId>io.delta</groupId>
@@ -406,7 +406,7 @@
                             <include>commons-pool:commons-pool</include>
                             <include>dnsjava:dnsjava</include>
                             <include>io.airlift:aircompressor</include>
-                            <include>io.delta:delta-core_${scala.binary.version}</include>
+                            <include>io.delta:delta-spark_${scala.binary.version}</include>
                             <include>io.delta:delta-hive_${scala.binary.version}</include>
                             <include>io.delta:delta-kernel-api</include>
                             <include>io.delta:delta-kernel-defaults</include>