This is an automated email from the ASF dual-hosted git repository.
vinishjail97 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/incubator-xtable.git
The following commit(s) were added to refs/heads/main by this push:
new 2b475a59 [886] Add Delta Kernel source/target coverage to
ITConversionController (#903)
2b475a59 is described below
commit 2b475a59809d508d08061327c5ec140d5cf0737e
Author: Vaibhav <[email protected]>
AuthorDate: Sat Oct 31 02:27:37 2026 +0530
[886] Add Delta Kernel source/target coverage to ITConversionController
(#903)
* [886] Add Delta Kernel source/target coverage to ITConversionController
- testVariousOperationsDeltaKernelSource runs the same dataset-equivalence
checks as testVariousOperations, forcing
DeltaKernelConversionSourceProvider
across both sync modes and both partitioning cases.
- testVariousOperationsDeltaKernelTarget does the same for the DELTA target
leg, routing through DeltaKernelConversionTarget via
DeltaConversionTargetConfig.USE_KERNEL, using ConversionTargetFactory's
normal dispatch path.
- testVariousOperations body extracted into runVariousOperationsTest so both
new tests share the same assertions the Standalone paths are validated
with; existing Standalone matrix is unchanged.
The target-side test currently fails: DeltaKernelConversionTarget only calls
withSchema() on table creation, so schema evolution on an existing table
(e.g. a new nested field added inside an array column) never reaches the
Delta log's Metadata action. Traced to open upstream gap
delta-io/delta#4305. Follow-up fail-fast guard tracked separately.
* [886] Fail fast on Delta Kernel schema evolution for existing tables
DeltaKernelConversionTarget.commitTransaction() only applies withSchema()
on CREATE_TABLE (Delta Kernel 4.0.0 limitation, delta-io/delta#4305), so
schema evolution on an already-existing table was silently dropped: rows
kept syncing, but a newly-added nested field (e.g. inside a struct that is
an array column's element type) never reached the Delta log's registered
schema, making it unreadable even though the underlying Parquet had it.
syncSchema() now compares the incoming schema (as a Kernel StructType, to
avoid any InternalSchema<->StructType round-trip noise) against the
existing snapshot's schema, and throws NotSupportedException as soon as
they differ for an existing table, instead of committing a stale schema.
ConversionController/TableFormatSync catches this per-target and records
it as a SyncStatusCode.ERROR, leaving the target at its last-good state
rather than partially written.
testVariousOperationsDeltaKernelTarget updated accordingly: it no longer
reuses runVariousOperationsTest's full lifecycle (which assumes success),
since the exception surfaces as a SyncResult status rather than a thrown
exception from ConversionController.sync(). It now syncs an initial
snapshot (expected SUCCESS), evolves the schema, syncs again (expected
ERROR with the delta-io/delta#4305 reference), and confirms the target's
row count didn't move.
* [886] Remove incidental schema evolution from TestDeltaKernelSync tests
testCreateSnapshotControlFlow and testTimestampNtz each did two syncSnapshot
calls to the same existing table, with the second call's schema silently
adding a field on top of the first (schema2 = schema1 + float_field). This
is exactly the schema-evolution-on-an-existing-table scenario
DeltaKernelConversionTarget now rejects (see the syncSchema guard added
earlier). Before that guard existed, Kernel silently accepted the second
sync without registering the new schema, and these tests only checked
physical file presence/size, not schema correctness, so they never caught
that the second sync's schema change was being dropped.
Neither test is actually about schema evolution:
testCreateSnapshotControlFlow
is about the file add/remove control flow across two sequential snapshot
syncs, and testTimestampNtz is about TIMESTAMP_NTZ handling. Fixed both to
reuse the same schema for their second snapshot, matching their real intent.
Added testSchemaEvolutionOnExistingTableFailsSync as a dedicated test for
the guarded behavior: first sync succeeds, second sync (with an evolved
schema) returns SyncStatusCode.ERROR referencing delta-io/delta#4305, and
the table is left at its last-good state rather than partially written.
* [886] Minor formatting/unused-import cleanup
* [886] Add unchanged-schema resync assertion to
testVariousOperationsDeltaKernelTarget
Per PR #903 review feedback: nothing exercised syncing an already-existing
table twice with an unchanged schema through the Kernel target, which is
exactly the path the syncSchema() drift guard could false-positive on if
there's any asymmetry in the InternalSchema<->StructType round trip.
Added a second sync (50 more rows, same schema) to the first block of
testVariousOperationsDeltaKernelTarget, asserting SyncStatusCode.SUCCESS,
before moving on to the schema-evolution block that expects ERROR. Covers
both partitioned and unpartitioned cases via the existing parameterization.
* [886] Add repro test for #779 log truncation safety check on Kernel vs
Standalone
ITDeltaLogTruncationSafetyCheck reproduces the #779 scenario against a
protocol-realistic setup: enough commits to force a real Delta checkpoint
(per PROTOCOL.md's 'Metadata cleanup', retention is required to always
leave a checkpoint before deleting older commit files), then deletes only
the pre-checkpoint JSON files, leaving the checkpoint intact.
With that realistic setup, both Standalone and Kernel correctly return
false from isIncrementalSyncSafeFrom -- no bug in either for that case.
A second pair of tests targets the actual gap: a TOCTOU race between the
safety check passing and the subsequent read, since retention cleanup can
run concurrently between the two. Kernel throws immediately with a clear,
specific exception when this happens. Standalone does not: getCommitsBacklog
silently returns a truncated commit list (a handful of commits instead of
the ~13 expected), meaning early commits are dropped with no error and the
sync would appear to succeed while quietly missing file actions for the
truncated range.
Filed as a standalone repro (not a fix) per the bar set when #782 (an
earlier, incomplete attempt at this) was closed: a clear description of
the failure scenario plus a reproducing test, before any fix is proposed.
The bug lives in getCommitsBacklog, not isIncrementalSyncSafeFrom itself,
so brishi19791's mustBeRecreatable suggestion on #782 would not address it.
* [886] Assert truncation safety outcomes instead of only logging them
Converts the log-only observations in ITDeltaLogTruncationSafetyCheck
into real JUnit assertions, so the suite can actually fail on a
regression instead of just narrating what happened:
- assertSafetyCheckDetectsUnsafe now asserts isIncrementalSyncSafeFrom
returns false post-truncation for both Standalone and Kernel.
- The check-then-read race helper now returns a RaceOutcome captured
before/after the concurrent deletion, so each of the two @Test
methods can assert its own engine-specific expectation: Kernel
throws KernelException, Standalone silently narrows the backlog to
commits after the checkpoint.
Verified all three tests still pass against real Standalone/Kernel
behavior (JDK 11; JDK 17.0.15 hits an unrelated Spark/Hadoop
UserGroupInformation incompatibility).
Addresses
https://github.com/apache/incubator-xtable/pull/903/changes#r4010718408
* [886] Cover updates and deletes in the Kernel target matrix
testVariousOperationsDeltaKernelTarget previously only inserted rows
before exercising schema evolution, so update/delete correctness
against a Kernel-written Delta target was never checked. Adds an
unchanged-schema upsert and delete step, using the existing
checkDatasetEquivalence/checkDatasetEquivalenceWithFilter helpers, so
incorrect removal handling can no longer leave deleted rows or stale
values visible without failing the matrix. Carries the new row count
(130, after the delete) through to the last-good-state assertion in
the schema-evolution section.
Verified across all 4 parameterized cases (2 sync modes x 2
partitioning) on JDK 11.
Addresses
https://github.com/apache/incubator-xtable/pull/903/changes#r4010718413
* [886] Assert non-empty backlog to avoid vacuous allMatch pass
allMatch on an empty stream returns true, so the Standalone race
assertion would still pass if Standalone regressed to silently
dropping every commit in the backlog, not just the pre-checkpoint
ones. Adds an explicit non-empty check so that regression fails
loudly instead of passing by vacuous truth. Also links #779 in the
comment above the race test, so a later Standalone fix reads as the
bug going away rather than as a regression.
Addresses
https://github.com/apache/incubator-xtable/pull/903/changes#r4108593854
* [886] Flip the Delta target default to Kernel
DeltaConversionTargetConfig.USE_KERNEL now defaults to true instead of
false, routing Delta targets through DeltaKernelConversionTarget by
default via ConversionTargetFactory's existing ServiceLoader-based
dispatch. Setting USE_KERNEL=false falls back to Delta Standalone --
the escape hatch for the one known regression on the Kernel path
(schema evolution on an existing table; delta-io/delta#4305).
ITConversionController's getTableSyncConfig previously only set
USE_KERNEL when explicitly requesting Kernel, leaving it unset (and
therefore defaulting) otherwise. That would have silently flipped
every existing "Standalone" test case in testVariousOperations* over
to Kernel. Now it always pins the property explicitly to true or
false based on the test's intent, so the existing Standalone-vs-Kernel
test matrix keeps exercising exactly what it did before.
TestConversionTargetFactory's default-resolution test is renamed and
flipped to assert Kernel instead of Standalone.
Verified with the Maven build cache explicitly disabled (this repo's
build cache was found to silently serve stale surefire results for a
full-class run with no test filter):
- TestConversionTargetFactory: 9/9 pass.
- ITConversionController (full class, every source format x sync mode
x partitioning combination, including the Standalone-pinned cases):
52/52 pass.
* [886] Flip the Delta source default to Kernel
Delta-as-source had no equivalent of DeltaConversionTargetConfig's
USE_KERNEL at all -- selection was hardcoded to
DeltaConversionSourceProvider (Standalone) in three independent
places, none of which went through the ServiceLoader-based factory
the target side uses:
- xtable-conversion-defaults.yaml's DELTA.conversionSourceProviderClass,
read via reflection by both RunSync and RunCatalogSync (confirmed
RunCatalogSync reads the same merged TableFormatConverters).
- ConversionService.initSourceProviders, a separate hardcoded Java map
in the Quarkus REST service layer.
Flips both to DeltaKernelConversionSourceProvider by default:
- xtable-conversion-defaults.yaml: changed the default class. The
escape hatch is the existing --convertersConfig override mechanism,
verified end-to-end against the real bundled CLI jar (built a
genuine Delta table, ran the sync to Iceberg with a
--convertersConfig pointing DELTA.conversionSourceProviderClass back
at DeltaConversionSourceProvider, and separately with no override;
both produced identical row counts, and the Kernel run's log carried
io.delta.kernel's own SnapshotReport/ScanReport entries, confirming
Kernel's engine actually executed rather than silently falling back).
- ConversionService/ConversionServiceConfig: added a new
xtable.delta.source.use_kernel Quarkus @ConfigProperty (default
true), mirroring USE_KERNEL's shape, since this call site has no
config-driven mechanism of its own to piggyback on.
TestRunSync's default-config assertion updated to expect Kernel.
ITConversionService, TestConversionService, and TestConversionResource
all construct ConversionService via its @VisibleForTesting
constructor with a manually pre-built provider map, bypassing
initSourceProviders and the new flag entirely by design -- confirmed
unaffected, no changes needed.
Verified: xtable-utilities and xtable-service both compile cleanly on
their respective required JDKs (11 and 17). TestRunSync: 8/8 pass
(build cache disabled). The xtable-service unit tests
(TestConversionService, TestConversionResource) currently fail in
this environment's @BeforeEach Mockito setup before any test body
runs (JDK 17.0.15 / Mockito inline-mock-maker incompatibility) --
pre-existing and unrelated to this change, since it fails before ever
reaching the modified code path.
* [886] Apply spotless formatting to ConversionService
Wraps a comment line that exceeded the configured line length.
* [886] Pin ITParquetConversionSource's Delta target to Standalone
Found via a real CI run of this branch: testSync's two partitioned
cases failed after the default flip. The MONTH-transform partition
this test uses is implemented as a Delta generated column, and Delta
Kernel 4.0.0 does not support the generatedColumns writer table
feature at all -- writing throws KernelException ("Unsupported Delta
writer feature"), caught per-target by TableFormatSync and surfaced as
SyncStatusCode.ERROR rather than corrupting the table. The test never
checked that result before reading the table back, so it silently
"passed" under the old Standalone default and only surfaced once
Kernel became the default.
This was missed by the earlier getTableSyncConfig pinning fix in
ITConversionController because this is a different, pre-existing test
file that calls ConversionController.sync() directly rather than
naming ConversionTargetFactory -- the grep-based audit that found
ITConversionController's call sites didn't catch it. Pin this test's
Delta target to Standalone the same way, and document the new
limitation in DeltaKernelConversionTarget's Javadoc alongside the
existing schema-evolution one.
Verified: ITParquetConversionSource 6/6 pass (build cache disabled),
including both previously-failing partitioned cases.
* [886] Pin XTableSyncTool's own use_kernel flag explicitly
Found via a real CI run of this branch: TestXTableSyncTool.testSync
failed with "Unable to sync to InternalTable for formats: DELTA" after
the default flip. This is production code, not a test -- a bigger
deal than the earlier test-only fixes.
XTableSyncTool already has its own documented config property,
hoodie.xtable.delta.target.use_kernel (default false, "instead of the
default Delta Standalone implementation"), predating this PR. It only
set DeltaConversionTargetConfig.USE_KERNEL on the target's
additionalProperties when its own flag was true, leaving the property
unset otherwise -- falling through to whatever
ConversionTargetFactory's own default was. That was harmless while
the global default was Standalone (false and unset both meant
Standalone), but now that the global default is Kernel, leaving the
property unset when this flag is false silently means Kernel too --
making the flag's documented "default to Standalone" behavior a
no-op for exactly the users relying on it.
Always set the property explicitly to the flag's actual value (true
or false), so this tool's own flag keeps meaning what it has always
documented, independent of whatever ConversionTargetFactory's global
default is.
This was missed by the earlier searches because they only covered
xtable-core, xtable-service, and xtable-utilities test directories --
never xtable-hudi-support/xtable-hudi-support-extensions, which this
production code lives in.
Verified: TestXTableSyncTool 5/5 pass (build cache disabled),
including both the default-flag case and the existing
testSyncWithDeltaKernel case that explicitly opts into Kernel.
* [886] Add missing delta-kernel-defaults dep and a Delta target use_kernel
toggle to xtable-service
xtable-service's ITConversionService hit NoClassDefFoundError for
io/delta/kernel/defaults/engine/DefaultEngine because the module declared
delta-kernel-api but never delta-kernel-defaults (xtable-core only has these
as provided-scope, which isn't transitive into xtable-service's test
classpath).
Separately, now that Kernel is the default Delta target, ConversionService
had no escape hatch to pin a Delta target to Standalone, which is needed
because Kernel does not yet support schema evolution against an existing
Delta table. Added xtable.delta.target.use_kernel (default true) mirroring
the existing source-side toggle, and pin it explicitly on the target's
additionalProperties rather than only setting it when true.
* [886] Back out the Delta default flip; land coverage, guard and
truncation test first
Per review, the default flip is deferred to a follow-up PR until two
Kernel-target failures pass an IT:
1. An unchanged resync of a Standalone-written Delta table fails the
schema-change guard for Hudi/Iceberg sources (columnMapping.id and
nested-nullability asymmetry in fromInternalSchema).
2. A DAY/MONTH transform partition fails with the unsupported
generatedColumns writer feature.
Reverts d47d916, 0c12db8, c4bb487, f551397, 525b7e6 and fc82864, so the
CLI, service, Hudi sync tool and core all default to Delta Standalone
again, with Kernel still available via the existing opt-in flag.
ITConversionController keeps upstream's hudiTargetProperties() and goes
back to only setting USE_KERNEL when a case requests Kernel.
The generatedColumns Known Limitations note in DeltaKernelConversionTarget
is kept, since it documents finding 2 independently of the default.
---
.../xtable/kernel/DeltaKernelConversionTarget.java | 61 +++-
.../org/apache/xtable/ITConversionController.java | 229 ++++++++++++-
.../xtable/ITDeltaLogTruncationSafetyCheck.java | 353 +++++++++++++++++++++
.../apache/xtable/kernel/TestDeltaKernelSync.java | 72 +++--
4 files changed, 677 insertions(+), 38 deletions(-)
diff --git
a/xtable-core/src/main/java/org/apache/xtable/kernel/DeltaKernelConversionTarget.java
b/xtable-core/src/main/java/org/apache/xtable/kernel/DeltaKernelConversionTarget.java
index 70384257..b98b4216 100644
---
a/xtable-core/src/main/java/org/apache/xtable/kernel/DeltaKernelConversionTarget.java
+++
b/xtable-core/src/main/java/org/apache/xtable/kernel/DeltaKernelConversionTarget.java
@@ -59,6 +59,7 @@ import io.delta.kernel.types.StructType;
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,12 +103,26 @@ import org.apache.xtable.spi.sync.ConversionTarget;
* <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
* upgrades. Public API alternatives should be used when available.
+ * <li><strong>Generated Columns:</strong> Delta Kernel 4.0.0 does not
support the {@code
+ * generatedColumns} writer table feature, which XTable's
schema/partition extraction emits
+ * for a derived (transform-based, e.g. a date truncation) partition
column. Writing such a
+ * schema throws {@link io.delta.kernel.exceptions.KernelException}
("Unsupported Delta writer
+ * feature"), caught per-target by {@code TableFormatSync} and surfaced
as a {@code
+ * SyncStatusCode.ERROR} entry rather than corrupting the table. Delta
Standalone does support
+ * this. Set {@link
org.apache.xtable.delta.DeltaConversionTargetConfig#USE_KERNEL} to {@code
+ * false} to fall back to it.
* </ul>
*
* <p><strong>Implementation Choice:</strong> Delta Kernel API was chosen over
Delta Standalone to:
@@ -229,6 +244,30 @@ public class DeltaKernelConversionTarget implements
ConversionTarget {
@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 +416,14 @@ public class DeltaKernelConversionTarget implements
ConversionTarget {
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 +462,11 @@ public class DeltaKernelConversionTarget implements
ConversionTarget {
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);
diff --git
a/xtable-core/src/test/java/org/apache/xtable/ITConversionController.java
b/xtable-core/src/test/java/org/apache/xtable/ITConversionController.java
index d282cd0a..c6b097d8 100644
--- a/xtable-core/src/test/java/org/apache/xtable/ITConversionController.java
+++ b/xtable-core/src/test/java/org/apache/xtable/ITConversionController.java
@@ -28,6 +28,7 @@ import static
org.apache.xtable.model.storage.TableFormat.PAIMON;
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;
@@ -101,13 +102,17 @@ import
org.apache.xtable.conversion.ConversionSourceProvider;
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.HudiTargetConfig;
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 {
@@ -180,6 +185,16 @@ public class ITConversionController {
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()) {
@@ -249,14 +264,148 @@ public class ITConversionController {
SyncMode syncMode,
boolean isPartitioned,
HoodieTableVersion hudiTargetVersion) {
+ runVariousOperationsTest(
+ sourceTableFormat,
+ syncMode,
+ isPartitioned,
+ hudiTargetVersion,
+ 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, null,
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
exercises upserts
+ // and deletes against the Kernel target with that same unchanged schema
(since incorrect
+ // removal/update handling could otherwise leave deleted rows or stale
values visible without
+ // failing this target's matrix), 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)) {
+ List<?> insertRecords = 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);
+
+ // Cover updates and deletes against the Kernel target before moving on
to schema evolution
+ // below -- still with an unchanged schema, so this isolates
removal/update handling from
+ // the schema-drift behavior exercised above and below.
+ table.upsertRows(insertRecords.subList(0, 20));
+ results = conversionController.sync(conversionConfig,
conversionSourceProvider);
+ assertEquals(
+ SyncStatusCode.SUCCESS,
results.get(DELTA).getTableFormatSyncStatus().getStatusCode());
+ checkDatasetEquivalence(HUDI, table, targetTableFormats, 150);
+
+ table.deleteRows(insertRecords.subList(30, 50));
+ results = conversionController.sync(conversionConfig,
conversionSourceProvider);
+ assertEquals(
+ SyncStatusCode.SUCCESS,
results.get(DELTA).getTableFormatSyncStatus().getStatusCode());
+ checkDatasetEquivalence(HUDI, table, targetTableFormats, 130);
+ checkDatasetEquivalenceWithFilter(
+ HUDI, table, targetTableFormats, table.getFilterQuery(),
Collections.emptyMap());
+ }
+
+ 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 (130 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(130, targetRowCount);
+ }
+ }
+
+ private void runVariousOperationsTest(
+ String sourceTableFormat,
+ SyncMode syncMode,
+ boolean isPartitioned,
+ HoodieTableVersion hudiTargetVersion,
+ ConversionSourceProvider<?> conversionSourceProvider) {
String tableName = getTableName();
List<String> targetTableFormats = getOtherFormats(sourceTableFormat);
String partitionConfig = null;
if (isPartitioned) {
partitionConfig = "level:VALUE";
}
- ConversionSourceProvider<?> conversionSourceProvider =
- getConversionSourceProvider(sourceTableFormat);
List<?> insertRecords;
try (GenericTable table =
GenericTable.getInstance(
@@ -1412,7 +1561,8 @@ public class ITConversionController {
targetTableFormats,
partitionConfig,
metadataRetention,
- null);
+ null,
+ false);
}
private static ConversionConfig getTableSyncConfig(
@@ -1424,6 +1574,55 @@ public class ITConversionController {
String partitionConfig,
Duration metadataRetention,
HoodieTableVersion hudiTargetVersion) {
+ return getTableSyncConfig(
+ sourceTableFormat,
+ syncMode,
+ tableName,
+ table,
+ targetTableFormats,
+ partitionConfig,
+ metadataRetention,
+ hudiTargetVersion,
+ 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) {
+ return getTableSyncConfig(
+ sourceTableFormat,
+ syncMode,
+ tableName,
+ table,
+ targetTableFormats,
+ partitionConfig,
+ metadataRetention,
+ null,
+ useDeltaKernelTarget);
+ }
+
+ private static ConversionConfig getTableSyncConfig(
+ String sourceTableFormat,
+ SyncMode syncMode,
+ String tableName,
+ GenericTable table,
+ List<String> targetTableFormats,
+ String partitionConfig,
+ Duration metadataRetention,
+ HoodieTableVersion hudiTargetVersion,
+ boolean useDeltaKernelTarget) {
Properties sourceProperties = new Properties();
if (partitionConfig != null) {
sourceProperties.put(PARTITION_FIELD_SPEC_CONFIG, partitionConfig);
@@ -1440,15 +1639,21 @@ public class ITConversionController {
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(hudiTargetProperties(formatName,
hudiTargetVersion))
- .build())
+ formatName -> {
+ TypedProperties targetProperties =
+ hudiTargetProperties(formatName, hudiTargetVersion);
+ 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()
diff --git
a/xtable-core/src/test/java/org/apache/xtable/ITDeltaLogTruncationSafetyCheck.java
b/xtable-core/src/test/java/org/apache/xtable/ITDeltaLogTruncationSafetyCheck.java
new file mode 100644
index 00000000..fd1da82c
--- /dev/null
+++
b/xtable-core/src/test/java/org/apache/xtable/ITDeltaLogTruncationSafetyCheck.java
@@ -0,0 +1,353 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.xtable;
+
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.net.URI;
+import java.nio.file.DirectoryStream;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.time.Instant;
+import java.util.List;
+import java.util.Properties;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+import lombok.Value;
+import lombok.extern.log4j.Log4j2;
+
+import org.apache.spark.api.java.JavaSparkContext;
+import org.apache.spark.sql.SparkSession;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import io.delta.kernel.exceptions.KernelException;
+
+import org.apache.xtable.conversion.ConversionSourceProvider;
+import org.apache.xtable.conversion.SourceTable;
+import org.apache.xtable.delta.DeltaConversionSourceProvider;
+import org.apache.xtable.hudi.HudiTestUtil;
+import org.apache.xtable.kernel.DeltaKernelConversionSourceProvider;
+import org.apache.xtable.model.CommitsBacklog;
+import org.apache.xtable.model.InstantsForIncrementalSync;
+import org.apache.xtable.model.storage.TableFormat;
+import org.apache.xtable.spi.extractor.ConversionSource;
+
+/**
+ * Empirical repro for https://github.com/apache/incubator-xtable/issues/779:
does {@code
+ * isIncrementalSyncSafeFrom} correctly report "not safe" once the underlying
_delta_log commit
+ * files backing an old instant have been physically removed, for both the
Standalone and Kernel
+ * Delta sources?
+ *
+ * <p>Per the Delta protocol spec (PROTOCOL.md, "Metadata cleanup"), real log
retention cleanup is
+ * REQUIRED to leave a checkpoint covering the oldest kept version before
deleting older JSON commit
+ * files -- readers should never be able to reach a state with commit files
missing and no
+ * checkpoint to fall back on. To reproduce that realistic, protocol-compliant
state rather than an
+ * artificial one, this test inserts enough rows to force Delta's default
checkpoint interval (10
+ * commits) to actually create a checkpoint, then deletes only the JSON commit
files strictly older
+ * than that checkpoint -- leaving the checkpoint itself and everything after
it intact.
+ */
+@Log4j2
+public class ITDeltaLogTruncationSafetyCheck {
+ // Delta's default delta.checkpointInterval is 10 commits; insert enough
rows to comfortably
+ // pass that so a checkpoint actually gets written.
+ private static final int NUM_COMMITS = 14;
+
+ @TempDir public static Path tempDir;
+
+ private static JavaSparkContext jsc;
+ private static SparkSession sparkSession;
+
+ @BeforeAll
+ public static void setupOnce() {
+ sparkSession =
SparkSession.builder().config(HudiTestUtil.getSparkConf(tempDir)).getOrCreate();
+ jsc = JavaSparkContext.fromSparkContext(sparkSession.sparkContext());
+ }
+
+ @AfterAll
+ public static void teardown() {
+ if (jsc != null) {
+ jsc.close();
+ }
+ if (sparkSession != null) {
+ sparkSession.close();
+ }
+ }
+
+ @Test
+ public void testIsIncrementalSyncSafeFromAfterLogTruncation() throws
Exception {
+ String tableName = GenericTable.getTableName();
+ Instant instantAfterV0;
+ String basePath;
+ try (TestSparkDeltaTable table =
+ new TestSparkDeltaTable(tableName, tempDir, sparkSession, null,
false)) {
+ table.insertRows(10); // version 0
+ instantAfterV0 = Instant.now();
+ Thread.sleep(1100);
+
+ for (int i = 1; i < NUM_COMMITS; i++) {
+ table.insertRows(10);
+ Thread.sleep(200);
+ }
+ basePath = table.getBasePath();
+ }
+
+ Path deltaLogDir = Paths.get(URI.create(basePath)).resolve("_delta_log");
+ long checkpointVersion = findLatestCheckpointVersion(deltaLogDir);
+ log.info("Latest checkpoint version found: {}", checkpointVersion);
+ assertTrue(
+ checkpointVersion > 0,
+ "Expected a checkpoint to have been created after "
+ + NUM_COMMITS
+ + " commits; increase NUM_COMMITS if delta.checkpointInterval has
changed.");
+
+ // Delete every JSON commit file strictly older than the checkpoint --
exactly what
+ // protocol-compliant log retention cleanup does, per PROTOCOL.md's
"Metadata cleanup"
+ // section -- leaving the checkpoint (and everything from its version
onward) intact.
+ int deletedCount = 0;
+ for (long version = 0; version < checkpointVersion; version++) {
+ Path commitFile = deltaLogDir.resolve(String.format("%020d.json",
version));
+ if (Files.deleteIfExists(commitFile)) {
+ deletedCount++;
+ }
+ }
+ log.info(
+ "Deleted {} commit file(s) older than checkpoint version {}",
+ deletedCount,
+ checkpointVersion);
+
+ SourceTable sourceTable =
+ SourceTable.builder()
+ .name(tableName)
+ .formatName(TableFormat.DELTA)
+ .basePath(basePath)
+ .dataPath(basePath)
+ .additionalProperties(new Properties())
+ .build();
+
+ // instantAfterV0 is before the checkpoint, and its backing commit file
has been deleted,
+ // while a valid checkpoint covering that deletion exists -- the
realistic, protocol-compliant
+ // version of the "log truncated past the requested instant" scenario in
#779.
+ assertSafetyCheckDetectsUnsafe(
+ "Standalone", new DeltaConversionSourceProvider(), sourceTable,
instantAfterV0);
+ assertSafetyCheckDetectsUnsafe(
+ "Kernel", new DeltaKernelConversionSourceProvider(), sourceTable,
instantAfterV0);
+ }
+
+ // Both isIncrementalSyncSafeFrom implementations correctly return false
once a valid
+ // checkpoint already covers the deleted commits (see
+ // testIsIncrementalSyncSafeFromAfterLogTruncation
+ // above). Delta's own docs on
getEarliestDeltaFile/getEarliestRecreatableCommit explicitly warn
+ // that "this version isn't guaranteed to exist when performing an action as
a concurrent
+ // operation can delete the file during cleanup" -- i.e. a TOCTOU race
between the safety check
+ // and the actual incremental read that follows it. This test targets that
race directly: call
+ // the safety check BEFORE any deletion (so it should say "safe"), then
delete the pre-checkpoint
+ // commit files (simulating retention cleanup running concurrently, right
after the check
+ // passed), then attempt the actual incremental read the check just approved.
+ //
+ // The two implementations react differently once the race is hit, so each
gets its own
+ // assertions on the observed behavior rather than a shared expectation:
+ // - Kernel resolves the sync instant to a version eagerly and throws a
KernelException once
+ // that version's backing file is gone.
+ // - Standalone does not throw; it silently narrows the backlog to only the
commits at or
+ // after the checkpoint, silently dropping the earlier commits it had
just approved reading.
+ // If https://github.com/apache/incubator-xtable/issues/779 is ever fixed
on the Standalone
+ // path, this assertion should start failing because commitsToProcess
becomes empty (or the
+ // read starts throwing) rather than narrowed -- that's the bug going
away, not a regression.
+ @Test
+ public void testSafetyCheckRaceWithConcurrentCleanup_Standalone() throws
Exception {
+ RaceOutcome outcome =
+ raceCheckThenReadAfterConcurrentDeletion("Standalone", new
DeltaConversionSourceProvider());
+ assertTrue(
+ outcome.isSafeBeforeDeletion(),
+ "Standalone isIncrementalSyncSafeFrom should report safe before the
concurrent cleanup"
+ + " runs");
+ assertNull(
+ outcome.getThrown(),
+ "Standalone's read is not expected to throw after the race; it is
expected to instead"
+ + " silently narrow the backlog -- see the commitsToProcess
assertion below. If this"
+ + " now throws, Standalone's behavior has changed and this test
needs updating: "
+ + outcome.getThrown());
+ assertNotNull(outcome.getCommitsToProcess());
+ // allMatch alone would pass vacuously on an empty backlog, which would
hide Standalone
+ // silently dropping every commit instead of just the pre-checkpoint ones
-- assert non-empty
+ // too so that regression fails loudly instead of passing by vacuous truth.
+ assertFalse(
+ outcome.getCommitsToProcess().isEmpty(),
+ "Expected Standalone's backlog to still contain the post-checkpoint
commits, not be"
+ + " empty");
+ assertTrue(
+ outcome.getCommitsToProcess().stream().allMatch(c -> c >
outcome.getCheckpointVersion()),
+ "Expected Standalone to silently drop every commit at or before the
checkpoint version ("
+ + outcome.getCheckpointVersion()
+ + ") once their backing files were deleted mid-race, even though
the safety check"
+ + " had approved syncing from before those commits; got "
+ + outcome.getCommitsToProcess());
+ }
+
+ @Test
+ public void testSafetyCheckRaceWithConcurrentCleanup_Kernel() throws
Exception {
+ RaceOutcome outcome =
+ raceCheckThenReadAfterConcurrentDeletion(
+ "Kernel", new DeltaKernelConversionSourceProvider());
+ assertTrue(
+ outcome.isSafeBeforeDeletion(),
+ "Kernel isIncrementalSyncSafeFrom should report safe before the
concurrent cleanup runs");
+ assertNotNull(
+ outcome.getThrown(),
+ "Expected Kernel's read to throw once concurrent cleanup deletes the
commit files"
+ + " backing the sync instant the safety check had just approved");
+ assertInstanceOf(
+ KernelException.class,
+ outcome.getThrown(),
+ "Expected a KernelException from Kernel's timestamp-to-version
resolution, got: "
+ + outcome.getThrown());
+ }
+
+ @Value
+ private static class RaceOutcome {
+ boolean safeBeforeDeletion;
+ Exception thrown;
+ List<Long> commitsToProcess;
+ long checkpointVersion;
+ }
+
+ private RaceOutcome raceCheckThenReadAfterConcurrentDeletion(
+ String label, ConversionSourceProvider<Long> provider) throws Exception {
+ // Each implementation gets its own freshly-built table, so one
implementation's deletion of
+ // pre-checkpoint files can't contaminate the other's "before deletion"
baseline.
+ String tableName = GenericTable.getTableName();
+ Instant instantAfterV0;
+ String basePath;
+ try (TestSparkDeltaTable table =
+ new TestSparkDeltaTable(tableName, tempDir, sparkSession, null,
false)) {
+ table.insertRows(10); // version 0
+ instantAfterV0 = Instant.now();
+ Thread.sleep(1100);
+
+ for (int i = 1; i < NUM_COMMITS; i++) {
+ table.insertRows(10);
+ Thread.sleep(200);
+ }
+ basePath = table.getBasePath();
+ }
+
+ Path deltaLogDir = Paths.get(URI.create(basePath)).resolve("_delta_log");
+ long checkpointVersion = findLatestCheckpointVersion(deltaLogDir);
+ log.info("[{}] Latest checkpoint version found: {}", label,
checkpointVersion);
+ assertTrue(checkpointVersion > 0, "Expected a checkpoint");
+
+ SourceTable sourceTable =
+ SourceTable.builder()
+ .name(tableName)
+ .formatName(TableFormat.DELTA)
+ .basePath(basePath)
+ .dataPath(basePath)
+ .additionalProperties(new Properties())
+ .build();
+
+ provider.init(jsc.hadoopConfiguration());
+ try (ConversionSource<Long> source =
provider.getConversionSourceInstance(sourceTable)) {
+ boolean safeBeforeDeletion =
source.isIncrementalSyncSafeFrom(instantAfterV0);
+ log.info(
+ "[{}] isIncrementalSyncSafeFrom({}) BEFORE concurrent cleanup = {}",
+ label,
+ instantAfterV0,
+ safeBeforeDeletion);
+
+ // Simulate retention cleanup running concurrently, immediately after
the check passed.
+ int deletedCount = 0;
+ for (long version = 0; version < checkpointVersion; version++) {
+ Path commitFile = deltaLogDir.resolve(String.format("%020d.json",
version));
+ if (Files.deleteIfExists(commitFile)) {
+ deletedCount++;
+ }
+ }
+ log.info("[{}] Deleted {} commit file(s) after the safety check ran",
label, deletedCount);
+
+ List<Long> commitsToProcess = null;
+ Exception thrown = null;
+ try {
+ CommitsBacklog<Long> backlog =
+ source.getCommitsBacklog(
+
InstantsForIncrementalSync.builder().lastSyncInstant(instantAfterV0).build());
+ commitsToProcess = backlog.getCommitsToProcess();
+ log.info(
+ "[{}] getCommitsBacklog after concurrent cleanup returned
commitsToProcess={},"
+ + " expected commits 1..{} (checkpoint covers 0..{}) if
nothing was silently"
+ + " skipped",
+ label,
+ commitsToProcess,
+ NUM_COMMITS - 1,
+ checkpointVersion - 1);
+ for (Long commit : commitsToProcess) {
+ source.getTableChangeForCommit(commit);
+ log.info("[{}] getTableChangeForCommit({}) succeeded", label,
commit);
+ }
+ } catch (Exception e) {
+ thrown = e;
+ log.info("[{}] read after concurrent cleanup threw: {}", label,
e.toString());
+ }
+ return new RaceOutcome(safeBeforeDeletion, thrown, commitsToProcess,
checkpointVersion);
+ }
+ }
+
+ /** Returns the highest checkpoint version found in the log directory, or -1
if none exist. */
+ private long findLatestCheckpointVersion(Path deltaLogDir) throws Exception {
+ Pattern checkpointPattern =
Pattern.compile("^(\\d{20})\\.checkpoint(\\..*)?\\.parquet$");
+ long latest = -1;
+ try (DirectoryStream<Path> stream = Files.newDirectoryStream(deltaLogDir))
{
+ for (Path entry : stream) {
+ Matcher matcher =
checkpointPattern.matcher(entry.getFileName().toString());
+ if (matcher.matches()) {
+ latest = Math.max(latest, Long.parseLong(matcher.group(1)));
+ }
+ }
+ }
+ return latest;
+ }
+
+ private void assertSafetyCheckDetectsUnsafe(
+ String label,
+ ConversionSourceProvider<Long> provider,
+ SourceTable sourceTable,
+ Instant instant)
+ throws Exception {
+ provider.init(jsc.hadoopConfiguration());
+ try (ConversionSource<Long> source =
provider.getConversionSourceInstance(sourceTable)) {
+ boolean safe = source.isIncrementalSyncSafeFrom(instant);
+ log.info("[{}] isIncrementalSyncSafeFrom({}) = {}", label, instant,
safe);
+ assertFalse(
+ safe,
+ "["
+ + label
+ + "] isIncrementalSyncSafeFrom should report false once the
commit files backing "
+ + instant
+ + " have been removed by protocol-compliant log retention
cleanup");
+ }
+ }
+}
diff --git
a/xtable-core/src/test/java/org/apache/xtable/kernel/TestDeltaKernelSync.java
b/xtable-core/src/test/java/org/apache/xtable/kernel/TestDeltaKernelSync.java
index b67e645f..06f569d7 100644
---
a/xtable-core/src/test/java/org/apache/xtable/kernel/TestDeltaKernelSync.java
+++
b/xtable-core/src/test/java/org/apache/xtable/kernel/TestDeltaKernelSync.java
@@ -77,6 +77,8 @@ import org.apache.xtable.model.storage.FileFormat;
import org.apache.xtable.model.storage.InternalDataFile;
import org.apache.xtable.model.storage.PartitionFileGroup;
import org.apache.xtable.model.storage.TableFormat;
+import org.apache.xtable.model.sync.SyncResult;
+import org.apache.xtable.model.sync.SyncStatusCode;
import org.apache.xtable.spi.sync.TableFormatSync;
/**
@@ -115,6 +117,36 @@ public class TestDeltaKernelSync {
@Test
public void testCreateSnapshotControlFlow() throws Exception {
+ // Both snapshots use the same schema: this test is about the file
add/remove control flow
+ // across two sequential snapshot syncs to an existing table, not schema
evolution (which
+ // DeltaKernelConversionTarget does not support for existing tables; see
+ // testSchemaEvolutionOnExistingTableFailsSync below).
+ InternalSchema schema = getInternalSchema();
+ InternalTable table1 = getInternalTable(tableName, basePath, schema, null,
LAST_COMMIT_TIME);
+ InternalTable table2 = getInternalTable(tableName, basePath, schema, null,
LAST_COMMIT_TIME);
+
+ InternalDataFile dataFile1 = getDataFile(1, Collections.emptyList(),
basePath);
+ InternalDataFile dataFile2 = getDataFile(2, Collections.emptyList(),
basePath);
+ InternalDataFile dataFile3 = getDataFile(3, Collections.emptyList(),
basePath);
+
+ InternalSnapshot snapshot1 = buildSnapshot(table1, "0", dataFile1,
dataFile2);
+ InternalSnapshot snapshot2 = buildSnapshot(table2, "1", dataFile2,
dataFile3);
+
+ TableFormatSync.getInstance()
+ .syncSnapshot(Collections.singletonList(conversionTarget), snapshot1);
+ validateDeltaTable(basePath, new HashSet<>(Arrays.asList(dataFile1,
dataFile2)));
+
+ TableFormatSync.getInstance()
+ .syncSnapshot(Collections.singletonList(conversionTarget), snapshot2);
+ validateDeltaTable(basePath, new HashSet<>(Arrays.asList(dataFile2,
dataFile3)));
+ }
+
+ @Test
+ public void testSchemaEvolutionOnExistingTableFailsSync() throws Exception {
+ // DeltaKernelConversionTarget.commitTransaction() only applies
withSchema() on CREATE_TABLE
+ // (Delta Kernel 4.0.0 limitation, see
https://github.com/delta-io/delta/issues/4305), so a
+ // second sync that evolves the schema of an already-existing table must
fail the sync rather
+ // than silently commit files without registering the new schema in the
Delta log.
InternalSchema schema1 = getInternalSchema();
List<InternalField> fields2 = new ArrayList<>(schema1.getFields());
fields2.add(
@@ -127,7 +159,7 @@ public class TestDeltaKernelSync {
.isNullable(true)
.build())
.build());
- InternalSchema schema2 =
getInternalSchema().toBuilder().fields(fields2).build();
+ InternalSchema schema2 = schema1.toBuilder().fields(fields2).build();
InternalTable table1 = getInternalTable(tableName, basePath, schema1,
null, LAST_COMMIT_TIME);
InternalTable table2 = getInternalTable(tableName, basePath, schema2,
null, LAST_COMMIT_TIME);
@@ -142,9 +174,19 @@ public class TestDeltaKernelSync {
.syncSnapshot(Collections.singletonList(conversionTarget), snapshot1);
validateDeltaTable(basePath, new HashSet<>(Arrays.asList(dataFile1,
dataFile2)));
- TableFormatSync.getInstance()
- .syncSnapshot(Collections.singletonList(conversionTarget), snapshot2);
- validateDeltaTable(basePath, new HashSet<>(Arrays.asList(dataFile2,
dataFile3)));
+ Map<String, SyncResult> results =
+ TableFormatSync.getInstance()
+ .syncSnapshot(Collections.singletonList(conversionTarget),
snapshot2);
+ SyncResult.SyncStatus status =
results.get(TableFormat.DELTA).getTableFormatSyncStatus();
+ assertEquals(SyncStatusCode.ERROR, status.getStatusCode());
+ assertTrue(
+
status.getErrorDetails().getErrorMessage().contains("delta/issues/4305"),
+ "Expected error message to reference the upstream Kernel limitation,
but was: "
+ + status.getErrorDetails().getErrorMessage());
+
+ // The table must be left exactly as the first sync left it: the second
sync's files must not
+ // have been committed alongside a stale schema.
+ validateDeltaTable(basePath, new HashSet<>(Arrays.asList(dataFile1,
dataFile2)));
}
@Test
@@ -569,21 +611,13 @@ public class TestDeltaKernelSync {
@Test
public void testTimestampNtz() throws Exception {
- InternalSchema schema1 = getInternalSchemaWithTimestampNtz();
- List<InternalField> fields2 = new ArrayList<>(schema1.getFields());
- fields2.add(
- InternalField.builder()
- .name("float_field")
- .schema(
- InternalSchema.builder()
- .name("float")
- .dataType(InternalType.FLOAT)
- .isNullable(true)
- .build())
- .build());
- InternalSchema schema2 =
getInternalSchema().toBuilder().fields(fields2).build();
- InternalTable table1 = getInternalTable(tableName, basePath, schema1,
null, LAST_COMMIT_TIME);
- InternalTable table2 = getInternalTable(tableName, basePath, schema2,
null, LAST_COMMIT_TIME);
+ // Both snapshots use the same TIMESTAMP_NTZ-inclusive schema: this test
is about that type
+ // being handled correctly across two sequential snapshot syncs to an
existing table, not
+ // schema evolution (which DeltaKernelConversionTarget does not support
for existing tables;
+ // see testSchemaEvolutionOnExistingTableFailsSync).
+ InternalSchema schema = getInternalSchemaWithTimestampNtz();
+ InternalTable table1 = getInternalTable(tableName, basePath, schema, null,
LAST_COMMIT_TIME);
+ InternalTable table2 = getInternalTable(tableName, basePath, schema, null,
LAST_COMMIT_TIME);
InternalDataFile dataFile1 = getDataFile(1, Collections.emptyList(),
basePath);
InternalDataFile dataFile2 = getDataFile(2, Collections.emptyList(),
basePath);
