vinishjail97 commented on code in PR #903:
URL: https://github.com/apache/incubator-xtable/pull/903#discussion_r4010718413
##########
xtable-core/src/test/java/org/apache/xtable/ITConversionController.java:
##########
@@ -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);
Review Comment:
[P2] Cover updates and deletes in the Kernel target matrix
Could we exercise unchanged-schema upserts and deletes here before testing
schema evolution? `runVariousOperationsTest` validates both operations through
dataset equivalence, but this target variant only performs two inserts before
expecting a schema error. The new Kernel source cases exclude DELTA from their
target list, so they do not cover this target path either. Incorrect removal
handling can therefore leave deleted rows or old row values visible without
failing the new target matrix. I would close this coverage gap before treating
prerequisite 1 of #886 as complete, using the existing upsert/delete assertions
across both sync modes and partitioning cases.
##########
xtable-core/src/test/java/org/apache/xtable/ITDeltaLogTruncationSafetyCheck.java:
##########
@@ -0,0 +1,316 @@
+/*
+ * 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 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.Properties;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+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 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);
+ org.junit.jupiter.api.Assertions.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.
+ logSafetyCheckResult(
+ "Standalone", new DeltaConversionSourceProvider(), sourceTable,
instantAfterV0);
+ logSafetyCheckResult(
+ "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.
+ @Test
+ public void testSafetyCheckRaceWithConcurrentCleanup_Standalone() throws
Exception {
+ raceCheckThenReadAfterConcurrentDeletion("Standalone", new
DeltaConversionSourceProvider());
+ }
+
+ @Test
+ public void testSafetyCheckRaceWithConcurrentCleanup_Kernel() throws
Exception {
+ raceCheckThenReadAfterConcurrentDeletion("Kernel", new
DeltaKernelConversionSourceProvider());
+ }
+
+ private void 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);
+ org.junit.jupiter.api.Assertions.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);
+
+ if (safeBeforeDeletion) {
Review Comment:
[P2] Assert the truncation safety outcomes
Could we replace the logging-only checks with assertions before merging?
`safeBeforeDeletion == false` skips the entire race check, while the catch
blocks swallow every exception. The non-race helper also accepts either safety
result. I temporarily made the Kernel safety check return `false` for every
non-null instant, and all three tests still passed. The catch blocks also allow
unexpected exceptions and incomplete backlogs to pass. Would it work to assert
`true` before deletion, `false` after deletion, and the specific expected
exception for the Kernel cleanup race? The Standalone case could assert its
observed behavior separately.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
