vaibhavk1992 commented on code in PR #903: URL: https://github.com/apache/incubator-xtable/pull/903#discussion_r4174139092
########## xtable-core/src/test/java/org/apache/xtable/ITDeltaLogTruncationSafetyCheck.java: ########## @@ -0,0 +1,343 @@ +/* + * 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. + @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()); Review Comment: Fixed in commit 79bfec9: -- 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]
