initial snapshot

Author: zikangh

kernel-spark/src/main/java/io/delta/kernel/spark/read/SparkMicroBatchStream.java

@@ -17,8 +17,11 @@
 
 import io.delta.kernel.CommitActions;
 import io.delta.kernel.CommitRange;
+import io.delta.kernel.Scan;
 import io.delta.kernel.Snapshot;
 import io.delta.kernel.data.ColumnarBatch;
+import io.delta.kernel.data.FilteredColumnarBatch;
+import io.delta.kernel.data.Row;
 import io.delta.kernel.defaults.engine.DefaultEngine;
 import io.delta.kernel.engine.Engine;
 import io.delta.kernel.exceptions.UnsupportedTableFeatureException;
@@ -35,6 +38,7 @@
 import java.io.IOException;
 import java.time.ZoneId;
 import java.util.*;
+import java.util.Comparator;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.spark.sql.SparkSession;
 import org.apache.spark.sql.connector.read.InputPartition;
@@ -154,9 +158,10 @@ public Offset initialOffset() {
       version = startingVersionOpt.get();
       isInitialSnapshot = false;
     } else {
-      // TODO(#5318): Support initial snapshot case (isInitialSnapshot == true)
-      throw new UnsupportedOperationException(
-          "initialOffset with initial snapshot is not supported yet");
+      // No starting version - create initial snapshot at latest version
+      Snapshot latestSnapshot = snapshotManager.loadLatestSnapshot();
+      version = latestSnapshot.getVersion();
+      isInitialSnapshot = true;
     }
 
     return DeltaSourceOffset.apply(
@@ -468,8 +473,16 @@ CloseableIterator<IndexedFile> getFileChanges(
     CloseableIterator<IndexedFile> result;
 
     if (isInitialSnapshot) {
-      // TODO(#5318): Implement initial snapshot
-      throw new UnsupportedOperationException("initial snapshot is not supported yet");
+      CloseableIterator<IndexedFile> snapshotFiles = getSnapshotFiles(fromVersion);
+      long latestVersion = snapshotAtSourceInit.getVersion();
+      if (latestVersion > fromVersion) {
+        // Start reading delta logs from fromVersion + 1 to avoid duplicating snapshot files
+        CloseableIterator<IndexedFile> deltaChanges = filterDeltaLogs(fromVersion + 1, endOffset);
+        // Lazily combine snapshot files and delta changes
+        result = snapshotFiles.combine(deltaChanges);
+      } else {
+        result = snapshotFiles;
+      }
     } else {
       result = filterDeltaLogs(fromVersion, endOffset);
     }
@@ -651,4 +664,66 @@ private long extractIndexedFilesFromBatch(
 
     return index;
   }
+
+  /**
+   * Get all files from a snapshot at the specified version, sorted by modificationTime and path,
+   * with indices assigned sequentially, and wrapped with BEGIN/END sentinels.
+   *
+   * <p>Mimics DeltaSourceSnapshot in DSv1.
+   *
+   * @param version The snapshot version to read
+   * @return An iterator of IndexedFile representing the snapshot files
+   */
+  private CloseableIterator<IndexedFile> getSnapshotFiles(long version) {
+    // Load snapshot at the specified version
+    Snapshot snapshot = snapshotManager.loadSnapshotAt(version);
+
+    // Build scan to get all files
+    Scan scan = snapshot.getScanBuilder().build();
+
+    // Collect all AddFile actions
+    List<AddFile> addFiles = new ArrayList<>();
+    try (CloseableIterator<FilteredColumnarBatch> filesIter = scan.getScanFiles(engine)) {
+      while (filesIter.hasNext()) {
+        FilteredColumnarBatch batch = filesIter.next();
+        try (CloseableIterator<Row> rowIter = batch.getRows()) {
+          while (rowIter.hasNext()) {
+            Row fileRow = rowIter.next();
+            // Extract AddFile from the "add" column (index 0)
+            if (!fileRow.isNullAt(0)) {
+              Row addFileRow = fileRow.getStruct(0);
+              AddFile addFile = new AddFile(addFileRow);
+              // Only include files with dataChange=true
+              if (addFile.getDataChange()) {
+                addFiles.add(addFile);
+              }
+            }
+          }
+        }
+      }
+    } catch (IOException e) {
+      throw new RuntimeException(
+          String.format("Failed to read snapshot files at version %d", version), e);
+    }
+
+    // CRITICAL: Sort by modificationTime, then path for deterministic ordering
+    addFiles.sort(
+        Comparator.comparing(AddFile::getModificationTime).thenComparing(AddFile::getPath));
+
+    // Build IndexedFile list with sentinels
+    List<IndexedFile> indexedFiles = new ArrayList<>();
+
+    // Add BEGIN sentinel
+    indexedFiles.add(new IndexedFile(version, DeltaSourceOffset.BASE_INDEX(), null));
+
+    // Add data files with sequential indices starting from 0
+    for (int i = 0; i < addFiles.size(); i++) {
+      indexedFiles.add(new IndexedFile(version, i, addFiles.get(i)));
+    }
+
+    // Add END sentinel
+    indexedFiles.add(new IndexedFile(version, DeltaSourceOffset.END_INDEX(), null));
+
+    return Utils.toCloseableIterator(indexedFiles.iterator());
+  }
 }

kernel-spark/src/test/java/io/delta/kernel/spark/Dsv2BasicTest.java

@@ -247,45 +247,43 @@ public void testStreamingReadWithRateLimit(@TempDir File deltaTablePath) throws
   }
 
   @Test
-  public void testStreamingReadWithoutStartingVersionThrowsException(@TempDir File deltaTablePath) {
+  public void testStreamingReadWithoutStartingVersion(@TempDir File deltaTablePath)
+      throws Exception {
     String tablePath = deltaTablePath.getAbsolutePath();
 
     // Write initial data
     List<Row> initialRows =
         Arrays.asList(RowFactory.create(1, "Alice", 100.0), RowFactory.create(2, "Bob", 200.0));
     writeInitialData(tablePath, initialRows);
 
-    // Try to create streaming DataFrame without startingVersion using DSv2 path
+    // Create streaming DataFrame without startingVersion using DSv2 path
     // Using dsv2.delta.`path` syntax to force DSv2 (SparkMicroBatchStream) instead of DSv1
     String dsv2TableRef = String.format("dsv2.delta.`%s`", tablePath);
 
-    // Should throw UnsupportedOperationException when trying to process
-    org.apache.spark.sql.streaming.StreamingQueryException exception =
-        assertThrows(
-            org.apache.spark.sql.streaming.StreamingQueryException.class,
-            () -> {
-              StreamingQuery query =
-                  spark
-                      .readStream()
-                      .table(dsv2TableRef)
-                      .writeStream()
-                      .format("memory")
-                      .queryName("test_no_starting_version")
-                      .outputMode("append")
-                      .start();
-              query.processAllAvailable();
-              query.stop();
-            });
-
-    // Verify the root cause is UnsupportedOperationException
-    Throwable rootCause = exception.getCause();
-    assertTrue(
-        rootCause instanceof UnsupportedOperationException,
-        "Root cause should be UnsupportedOperationException, but was: "
-            + (rootCause != null ? rootCause.getClass().getName() : "null"));
-    assertTrue(
-        rootCause.getMessage().contains("is not supported"),
-        "Exception message should indicate operation is not supported: " + rootCause.getMessage());
+    StreamingQuery query = null;
+    try {
+      query =
+          spark
+              .readStream()
+              .table(dsv2TableRef) // No startingVersion option
+              .writeStream()
+              .format("memory")
+              .queryName("test_no_starting_version")
+              .outputMode("append")
+              .start();
+
+      query.processAllAvailable();
+
+      // Should successfully read all data from initial snapshot
+      Dataset<Row> results = spark.sql("SELECT * FROM test_no_starting_version");
+      List<Row> actualRows = results.collectAsList();
+
+      assertStreamingDataEquals(actualRows, initialRows);
+    } finally {
+      if (query != null) {
+        query.stop();
+      }
+    }
   }
 
   //////////////////////