[886] Add Delta Kernel source/target coverage to ITConversionController

xtable-core/src/main/java/org/apache/xtable/kernel/DeltaKernelConversionTarget.java

@@ -59,6 +59,7 @@
 import io.delta.kernel.utils.CloseableIterable;
 
 import org.apache.xtable.conversion.TargetTable;
+import org.apache.xtable.delta.DeltaConversionTargetConfig;
 import org.apache.xtable.exception.NotSupportedException;
 import org.apache.xtable.exception.UpdateException;
 import org.apache.xtable.model.InternalTable;
@@ -102,8 +103,14 @@
  *   <li><strong>Commit Tags:</strong> Delta Kernel 4.0.0 does not support commit tags in commitInfo
  *       (e.g., XTABLE_METADATA tags). This affects source-to-target commit identifier mapping.
  *       Tracked in: https://github.com/apache/incubator-xtable/issues/819
- *   <li><strong>Schema Evolution:</strong> Schema changes are handled through Delta Kernel's
- *       transaction API, which may have different semantics compared to Delta Standalone.
+ *   <li><strong>Schema Evolution:</strong> Delta Kernel 4.0.0 only applies a schema via {@code
+ *       TransactionBuilder.withSchema()} on {@code CREATE_TABLE}, not on {@code WRITE} to an
+ *       existing table. {@link #syncSchema(InternalSchema)} detects when a sync's schema differs
+ *       from an existing table's current schema and throws {@link NotSupportedException} rather
+ *       than silently committing a stale schema. Set {@link
+ *       org.apache.xtable.delta.DeltaConversionTargetConfig#USE_KERNEL} to {@code false} to fall
+ *       back to the Delta Standalone target, which does support this. Tracked upstream:
+ *       https://github.com/delta-io/delta/issues/4305
  *   <li><strong>Internal API Usage:</strong> This implementation casts to internal classes
  *       (SnapshotImpl, TableImpl) to access metadata and commit history, as Delta Kernel 4.0.0
  *       lacks public APIs for these operations. These casts are brittle and may break on version
@@ -229,6 +236,30 @@ public void beginSync(InternalTable table) {
 
   @Override
   public void syncSchema(InternalSchema schema) {
+    if (transactionState.isTableExists()) {
+      // Compare in Kernel's own StructType, not InternalSchema, to avoid false positives from
+      // any asymmetry in the InternalSchema<->StructType round trip. transactionState's current
+      // latestSchema still holds the schema loaded from the existing snapshot at this point,
+      // since it is only overwritten by the setLatestSchema() call below.
+      StructType existingSchema = transactionState.getLatestSchema();
+      StructType newSchema = schemaExtractor.fromInternalSchema(schema);
+      if (!newSchema.equals(existingSchema)) {
+        // LIMITATION: Delta Kernel 4.0.0's TransactionBuilder.withSchema() only takes effect for
+        // CREATE_TABLE operations (see commitTransaction() below), so there is no supported way
+        // to commit an evolved schema for an existing table today. Committing anyway would write
+        // AddFile actions for files that may contain the new columns/fields while leaving the
+        // table's registered schema stale, silently making the new data unreadable rather than
+        // failing. Fail fast instead. Tracked upstream:
+        // https://github.com/delta-io/delta/issues/4305
+        throw new NotSupportedException(
+            "Schema evolution on an existing Delta table is not supported by "
+                + "DeltaKernelConversionTarget (see https://github.com/delta-io/delta/issues/4305). "
+                + "Set "
+                + DeltaConversionTargetConfig.USE_KERNEL
+                + "=false on the target's additional properties to use the Delta Standalone "
+                + "target instead.");
+      }
+    }
     transactionState.setLatestSchema(schema);
   }
 
@@ -377,6 +408,14 @@ void addPartitionColumn(String columnName) {
       partitionColumns.add(columnName);
     }
 
+    /**
+     * Whether the target table already existed at the start of this sync. Package-private to allow
+     * access from outer class.
+     */
+    boolean isTableExists() {
+      return tableExists;
+    }
+
     /**
      * Gets the cached snapshot. Returns null if no snapshot was cached (new table). Package-private
      * to allow access from outer class.
@@ -415,11 +454,11 @@ private void commitTransaction() {
           table.createTransactionBuilder(engine, "XTable Delta Sync", operation);
 
       // LIMITATION: Schema evolution for existing tables is NOT supported in Delta Kernel 4.0.0.
-      // The withSchema() method only works during CREATE_TABLE operations. For existing tables:
-      // - AddFile/RemoveFile actions are created using the old schema from existing snapshot
-      // - If source schema has evolved (columns added/removed/type changed), the Delta table
-      //   will have mismatched metadata and data, causing query failures or incorrect results
-      // This is a known Delta Kernel limitation: https://github.com/delta-io/delta/issues/4305
+      // The withSchema() method only works during CREATE_TABLE operations. syncSchema() already
+      // guards against this by throwing NotSupportedException as soon as an evolved schema is
+      // detected for an existing table, so by the time we get here, an existing table's schema
+      // is guaranteed to be unchanged from the existing snapshot. This is a known Delta Kernel
+      // limitation: https://github.com/delta-io/delta/issues/4305
       if (!tableExists) {
         txnBuilder = txnBuilder.withSchema(engine, latestSchema);
 

xtable-core/src/test/java/org/apache/xtable/ITConversionController.java

@@ -28,6 +28,7 @@
 import static org.apache.xtable.model.storage.TableFormat.PARQUET;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
 
 import java.net.URI;
 import java.nio.ByteBuffer;
@@ -100,12 +101,16 @@
 import org.apache.xtable.conversion.SourceTable;
 import org.apache.xtable.conversion.TargetTable;
 import org.apache.xtable.delta.DeltaConversionSourceProvider;
+import org.apache.xtable.delta.DeltaConversionTargetConfig;
 import org.apache.xtable.hudi.HudiConversionSourceProvider;
 import org.apache.xtable.hudi.HudiTestUtil;
 import org.apache.xtable.iceberg.IcebergConversionSourceProvider;
 import org.apache.xtable.iceberg.TestIcebergDataHelper;
+import org.apache.xtable.kernel.DeltaKernelConversionSourceProvider;
 import org.apache.xtable.model.storage.TableFormat;
 import org.apache.xtable.model.sync.SyncMode;
+import org.apache.xtable.model.sync.SyncResult;
+import org.apache.xtable.model.sync.SyncStatusCode;
 import org.apache.xtable.paimon.PaimonConversionSourceProvider;
 
 public class ITConversionController {
@@ -160,6 +165,16 @@ private static Stream<Arguments> generateTestParametersForFormatsSyncModesAndPar
     return arguments.stream();
   }
 
+  private static Stream<Arguments> generateTestParametersForSyncModesAndPartitioning() {
+    List<Arguments> arguments = new ArrayList<>();
+    for (SyncMode syncMode : SyncMode.values()) {
+      for (boolean isPartitioned : new boolean[] {true, false}) {
+        arguments.add(Arguments.of(syncMode, isPartitioned));
+      }
+    }
+    return arguments.stream();
+  }
+
   private static Stream<Arguments> generateTestParametersForUUID() {
     List<Arguments> arguments = new ArrayList<>();
     for (SyncMode syncMode : SyncMode.values()) {
@@ -226,6 +241,116 @@ private ConversionSourceProvider<?> getConversionSourceProvider(String sourceTab
   @MethodSource("generateTestParametersForFormatsSyncModesAndPartitioning")
   public void testVariousOperations(
       String sourceTableFormat, SyncMode syncMode, boolean isPartitioned) {
+    runVariousOperationsTest(
+        sourceTableFormat, syncMode, isPartitioned, getConversionSourceProvider(sourceTableFormat));
+  }
+
+  // Runs the same equivalence checks as testVariousOperations above, but forces the Delta Kernel
+  // conversion source rather than resolving the provider from sourceTableFormat. This covers the
+  // Kernel source path against the same assertions the Standalone source is validated with today,
+  // per https://github.com/apache/incubator-xtable/issues/886. A targeted (sync mode x
+  // partitioning) subset is used here rather than folding DELTA_KERNEL into
+  // generateTestParametersForFormatsSyncModesAndPartitioning, since sourceTableFormat there also
+  // drives GenericTable selection, Spark format-name reads, and target-format exclusion for every
+  // other format, none of which differ for Kernel vs Standalone.
+  @ParameterizedTest
+  @MethodSource("generateTestParametersForSyncModesAndPartitioning")
+  public void testVariousOperationsDeltaKernelSource(SyncMode syncMode, boolean isPartitioned) {
+    ConversionSourceProvider<Long> deltaKernelConversionSourceProvider =
+        new DeltaKernelConversionSourceProvider();
+    deltaKernelConversionSourceProvider.init(jsc.hadoopConfiguration());
+    runVariousOperationsTest(DELTA, syncMode, isPartitioned, deltaKernelConversionSourceProvider);
+  }
+
+  // Validates the DELTA target routed through DeltaKernelConversionTarget (via
+  // DeltaConversionTargetConfig.USE_KERNEL on that target's properties, using
+  // ConversionTargetFactory's normal dispatch path). Source is fixed to HUDI, a simple,
+  // always-available source unrelated to what's under test here (the target-side dispatch).
+  // Covers the "as a target" half of prerequisite #1 in
+  // https://github.com/apache/incubator-xtable/issues/886.
+  //
+  // DeltaKernelConversionTarget does not support schema evolution on an already-existing table
+  // (Delta Kernel 4.0.0 only applies withSchema() at CREATE_TABLE; see
+  // https://github.com/delta-io/delta/issues/4305) and throws NotSupportedException rather than
+  // silently committing a stale schema. ConversionController.sync() catches that per-target
+  // (TableFormatSync#buildResultForError) rather than propagating it, so the exception surfaces
+  // as a SyncStatusCode.ERROR entry in the returned result map, not as a thrown exception from
+  // sync() itself. This test therefore doesn't reuse runVariousOperationsTest: it syncs an
+  // initial snapshot (expected to succeed, verified via the normal equivalence check), resyncs
+  // the same unchanged schema to an existing table (expected to still succeed -- this guards
+  // against a false positive in syncSchema()'s drift check, since it compares in Kernel's
+  // StructType rather than InternalSchema specifically to avoid InternalSchema<->StructType
+  // round-trip asymmetry causing an unchanged schema to look "evolved"), then evolves the schema
+  // and syncs again, asserting the DELTA target's sync result is ERROR with the expected message,
+  // and that the target's data was left at its last-good state rather than partially written.
+  @ParameterizedTest
+  @MethodSource("generateTestParametersForSyncModesAndPartitioning")
+  public void testVariousOperationsDeltaKernelTarget(SyncMode syncMode, boolean isPartitioned) {
+    String tableName = getTableName();
+    ConversionSourceProvider<?> conversionSourceProvider = getConversionSourceProvider(HUDI);
+    String partitionConfig = isPartitioned ? "level:VALUE" : null;
+    List<String> targetTableFormats = Collections.singletonList(DELTA);
+
+    try (GenericTable table =
+        GenericTable.getInstance(tableName, tempDir, sparkSession, jsc, HUDI, isPartitioned)) {
+      table.insertRows(100);
+      ConversionConfig conversionConfig =
+          getTableSyncConfig(
+              HUDI, syncMode, tableName, table, targetTableFormats, partitionConfig, null, true);
+      Map<String, SyncResult> results =
+          conversionController.sync(conversionConfig, conversionSourceProvider);
+      assertEquals(
+          SyncStatusCode.SUCCESS, results.get(DELTA).getTableFormatSyncStatus().getStatusCode());
+      checkDatasetEquivalence(HUDI, table, targetTableFormats, 100);
+
+      // Resync the same table with its unchanged schema. The table now exists, so this exercises
+      // syncSchema()'s drift check on the existing-table path without any real schema evolution.
+      table.insertRows(50);
+      results = conversionController.sync(conversionConfig, conversionSourceProvider);
+      assertEquals(
+          SyncStatusCode.SUCCESS,
+          results.get(DELTA).getTableFormatSyncStatus().getStatusCode(),
+          "An unchanged-schema resync of an existing table must not be flagged as schema drift");
+      checkDatasetEquivalence(HUDI, table, targetTableFormats, 150);
+    }
+
+    try (GenericTable tableWithUpdatedSchema =
+        GenericTable.getInstanceWithAdditionalColumns(
+            tableName, tempDir, sparkSession, jsc, HUDI, isPartitioned)) {
+      tableWithUpdatedSchema.insertRows(100);
+      ConversionConfig conversionConfig =
+          getTableSyncConfig(
+              HUDI,
+              syncMode,
+              tableName,
+              tableWithUpdatedSchema,
+              targetTableFormats,
+              partitionConfig,
+              null,
+              true);
+      Map<String, SyncResult> results =
+          conversionController.sync(conversionConfig, conversionSourceProvider);
+      SyncResult.SyncStatus deltaStatus = results.get(DELTA).getTableFormatSyncStatus();
+      assertEquals(SyncStatusCode.ERROR, deltaStatus.getStatusCode());
+      assertTrue(
+          deltaStatus
+              .getErrorDetails()
+              .getErrorMessage()
+              .contains("https://github.com/delta-io/delta/issues/4305"));
+
+      // The target should be left exactly at its last successfully-synced state (150 rows, old
+      // schema), not partially written with some but not all of the new rows/columns.
+      long targetRowCount =
+          sparkSession.read().format("delta").load(tableWithUpdatedSchema.getDataPath()).count();
+      assertEquals(150, targetRowCount);
+    }
+  }
+
+  private void runVariousOperationsTest(
+      String sourceTableFormat,
+      SyncMode syncMode,
+      boolean isPartitioned,
+      ConversionSourceProvider<?> conversionSourceProvider) {
     String tableName = getTableName();
     List<String> targetTableFormats = getOtherFormats(sourceTableFormat);
     if (sourceTableFormat.equals(PAIMON)) {
@@ -237,8 +362,6 @@ public void testVariousOperations(
     if (isPartitioned) {
       partitionConfig = "level:VALUE";
     }
-    ConversionSourceProvider<?> conversionSourceProvider =
-        getConversionSourceProvider(sourceTableFormat);
     List<?> insertRecords;
     try (GenericTable table =
         GenericTable.getInstance(
@@ -1194,6 +1317,32 @@ private static ConversionConfig getTableSyncConfig(
       List<String> targetTableFormats,
       String partitionConfig,
       Duration metadataRetention) {
+    return getTableSyncConfig(
+        sourceTableFormat,
+        syncMode,
+        tableName,
+        table,
+        targetTableFormats,
+        partitionConfig,
+        metadataRetention,
+        false);
+  }
+
+  // Same as above, but when useDeltaKernelTarget is true, routes any DELTA target through
+  // ConversionTargetFactory's DeltaKernelConversionTarget rather than the default Standalone one,
+  // by setting DeltaConversionTargetConfig.USE_KERNEL on that target's additional properties. This
+  // lets a test validate the Kernel target writer through the same dispatch path (and the same
+  // equivalence assertions) a Standalone target is validated with, per
+  // https://github.com/apache/incubator-xtable/issues/886.
+  private static ConversionConfig getTableSyncConfig(
+      String sourceTableFormat,
+      SyncMode syncMode,
+      String tableName,
+      GenericTable table,
+      List<String> targetTableFormats,
+      String partitionConfig,
+      Duration metadataRetention,
+      boolean useDeltaKernelTarget) {
     Properties sourceProperties = new Properties();
     if (partitionConfig != null) {
       sourceProperties.put(PARTITION_FIELD_SPEC_CONFIG, partitionConfig);
@@ -1210,15 +1359,20 @@ private static ConversionConfig getTableSyncConfig(
     List<TargetTable> targetTables =
         targetTableFormats.stream()
             .map(
-                formatName ->
-                    TargetTable.builder()
-                        .name(tableName)
-                        .formatName(formatName)
-                        // set the metadata path to the data path as the default (required by Hudi)
-                        .basePath(table.getDataPath())
-                        .metadataRetention(metadataRetention)
-                        .additionalProperties(new TypedProperties())
-                        .build())
+                formatName -> {
+                  TypedProperties targetProperties = new TypedProperties();
+                  if (useDeltaKernelTarget && formatName.equals(DELTA)) {
+                    targetProperties.setProperty(DeltaConversionTargetConfig.USE_KERNEL, "true");
+                  }
+                  return TargetTable.builder()
+                      .name(tableName)
+                      .formatName(formatName)
+                      // set the metadata path to the data path as the default (required by Hudi)
+                      .basePath(table.getDataPath())
+                      .metadataRetention(metadataRetention)
+                      .additionalProperties(targetProperties)
+                      .build();
+                })
             .collect(Collectors.toList());
 
     return ConversionConfig.builder()