diff --git a/docs/generated/core_configuration.html b/docs/generated/core_configuration.html index 0fc8c42fd261..151675ee3e33 100644 --- a/docs/generated/core_configuration.html +++ b/docs/generated/core_configuration.html @@ -1566,6 +1566,12 @@ String When a batch job queries from a table, if a partition does not exist in the current branch, the reader will try to get this partition from this fallback branch. + +
scan.fallback-branch.allow-earliest-snapshot
+ false + Boolean + Whether to allow reading the earliest available snapshot of a fallback or primary branch when no snapshot exists at or before the current branch's requested snapshot time. This applies to snapshot-ID time travel, including numeric scan.version. By default, such reads fail unless the other branch has no snapshots. Enabling this option may read data committed after the requested time. +
scan.fallback-branch.read-fail-fast
false diff --git a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java index c3202188dc09..098ab6bfdc40 100644 --- a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java +++ b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java @@ -2396,6 +2396,17 @@ public String toString() { + "falls through to the current branch, which can mask data issues. " + "Set this to true to surface fallback branch errors to the caller instead."); + public static final ConfigOption SCAN_FALLBACK_BRANCH_ALLOW_EARLIEST_SNAPSHOT = + key("scan.fallback-branch.allow-earliest-snapshot") + .booleanType() + .defaultValue(false) + .withDescription( + "Whether to allow reading the earliest available snapshot of a fallback or primary branch " + + "when no snapshot exists at or before the current branch's requested snapshot time. " + + "This applies to snapshot-ID time travel, including numeric scan.version. " + + "By default, such reads fail unless the other branch has no snapshots. " + + "Enabling this option may read data committed after the requested time."); + public static final ConfigOption SCAN_PRIMARY_BRANCH = key("scan.primary-branch") .stringType() diff --git a/paimon-core/src/main/java/org/apache/paimon/table/FallbackReadFileStoreTable.java b/paimon-core/src/main/java/org/apache/paimon/table/FallbackReadFileStoreTable.java index 02624408f2fb..eaf7abf49955 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/FallbackReadFileStoreTable.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/FallbackReadFileStoreTable.java @@ -216,13 +216,34 @@ protected Map rewriteOtherOptions(Map options) { long id = Long.parseLong(scanSnapshotId); long millis = wrapped.snapshotManager().snapshot(id).timeMillis(); Snapshot otherSnapshot = other.snapshotManager().earlierOrEqualTimeMills(millis); - long otherId; if (otherSnapshot == null) { - otherId = Snapshot.FIRST_SNAPSHOT_ID; - } else { - otherId = otherSnapshot.id(); + otherSnapshot = other.snapshotManager().earliestSnapshot(); + if (otherSnapshot != null) { + Preconditions.checkArgument( + new Options(wrapped.options(), options) + .get(CoreOptions.SCAN_FALLBACK_BRANCH_ALLOW_EARLIEST_SNAPSHOT), + "No snapshot at or before %s is available in branch '%s'. " + + "The earliest available snapshot is %s (time %s). " + + "Set '%s' to true to allow reading it.", + millis, + other.coreOptions().branch(), + otherSnapshot.id(), + otherSnapshot.timeMillis(), + CoreOptions.SCAN_FALLBACK_BRANCH_ALLOW_EARLIEST_SNAPSHOT.key()); + LOG.warn( + "No snapshot at or before {} is available in branch '{}'; reading earliest " + + "snapshot {} (time {}) because '{}' is enabled.", + millis, + other.coreOptions().branch(), + otherSnapshot.id(), + otherSnapshot.timeMillis(), + CoreOptions.SCAN_FALLBACK_BRANCH_ALLOW_EARLIEST_SNAPSHOT.key()); + } } - result.put(scanSnapshotIdOptionKey, String.valueOf(otherId)); + // An empty branch is skipped by newFallbackScan instead of loading this ID. + result.put( + scanSnapshotIdOptionKey, + otherSnapshot == null ? scanSnapshotId : String.valueOf(otherSnapshot.id())); } // bucket number of main branch and other branch are very likely different, @@ -246,7 +267,53 @@ public DataTableScan newFallbackScan(Function sca validateSchema(); FileStoreTable first = wrappedFirst ? wrapped : other; FileStoreTable second = wrappedFirst ? other : wrapped; - return new FallbackReadScan(first, second, wrapped.schema(), scanCreator); + boolean otherIsEmpty = + other.coreOptions().scanSnapshotId() != null + && other.snapshotManager().latestSnapshotId() == null; + if (otherIsEmpty) { + LOG.warn( + "Branch '{}' has no snapshots; skipping it when reading branch '{}'.", + other.coreOptions().branch(), + wrapped.coreOptions().branch()); + } + return new FallbackReadScan( + first, + second, + wrapped.schema(), + table -> { + if (table == other && otherIsEmpty) { + return new EmptyBranchScan(); + } + return scanCreator.apply(table); + }); + } + + private static class EmptyBranchScan implements DataTableScan { + + @Override + public DataTableScan withShard(int indexOfThisSubtask, int numberOfParallelSubtasks) { + return this; + } + + @Override + public InnerTableScan withFilter(Predicate predicate) { + return this; + } + + @Override + public InnerTableScan withInternalPartitionFilter(PartitionPredicate partitionPredicate) { + return this; + } + + @Override + public Plan plan() { + return new DataFilePlan<>(Collections.emptyList()); + } + + @Override + public List listPartitionEntries() { + return Collections.emptyList(); + } } protected void validateSchema() { diff --git a/paimon-core/src/test/java/org/apache/paimon/table/FallbackReadFileStoreTableTest.java b/paimon-core/src/test/java/org/apache/paimon/table/FallbackReadFileStoreTableTest.java index 5c161a1a7088..fd8d74a10ebd 100644 --- a/paimon-core/src/test/java/org/apache/paimon/table/FallbackReadFileStoreTableTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/table/FallbackReadFileStoreTableTest.java @@ -70,7 +70,9 @@ import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; +import java.util.HashMap; import java.util.List; +import java.util.Map; import java.util.UUID; import java.util.function.Consumer; import java.util.stream.Collectors; @@ -368,6 +370,141 @@ public void testScanVersionForFallbackBranch() throws Exception { .containsExactlyInAnyOrder(Pair.of(1, 10), Pair.of(2, 20)); } + @ParameterizedTest + @CsvSource({"false,false,false", "false,true,true", "true,false,true", "true,true,false"}) + public void testTimeTravelWithEmptyOtherBranch( + boolean wrappedFirst, boolean scanVersion, boolean allowEarliest) throws Exception { + FileStoreTable mainTable = createTable(); + writeDataIntoTable(mainTable, 0, rowData(1, 10)); + mainTable.createBranch("empty"); + FileStoreTable branchTable = createTableFromBranch(mainTable, "empty"); + assertThat(branchTable.snapshotManager().latestSnapshotId()).isNull(); + Map options = new HashMap<>(); + options.put( + scanVersion ? CoreOptions.SCAN_VERSION.key() : CoreOptions.SCAN_SNAPSHOT_ID.key(), + "1"); + options.put( + CoreOptions.SCAN_FALLBACK_BRANCH_ALLOW_EARLIEST_SNAPSHOT.key(), + String.valueOf(allowEarliest)); + + FallbackReadFileStoreTable latest = + new FallbackReadFileStoreTable(mainTable, branchTable, wrappedFirst); + assertThat(readAndCollect(latest, scan -> {})).containsExactly(Pair.of(1, 10)); + FallbackReadFileStoreTable table = (FallbackReadFileStoreTable) latest.copy(options); + + assertThat(readAndCollect(table, scan -> {})).containsExactly(Pair.of(1, 10)); + assertThat(table.newScan().listPartitionEntries()) + .map(entry -> entry.partition().getInt(0)) + .containsExactly(1); + } + + @ParameterizedTest + @CsvSource({"false,false", "false,true", "true,false", "true,true"}) + public void testTimeTravelBeforeOtherSnapshots(boolean wrappedFirst, boolean expired) + throws Exception { + FileStoreTable mainTable = createTable(); + writeDataIntoTable(mainTable, 0, rowData(1, 10)); + long snapshotTime = mainTable.snapshotManager().snapshot(1).timeMillis(); + mainTable.createBranch("bc"); + FileStoreTable branchTable = createTableFromBranch(mainTable, "bc"); + while (System.currentTimeMillis() <= snapshotTime) { + Thread.yield(); + } + writeDataIntoTable(branchTable, 0, rowData(1, 100), rowData(2, 20)); + if (expired) { + writeDataIntoTable(branchTable, 1, rowData(3, 30)); + branchTable.snapshotManager().deleteSnapshot(1); + branchTable.snapshotManager().commitEarliestHint(2); + } + // The latest snapshot must differ from earliest; its extra row must not be read. + writeDataIntoTable(branchTable, 2, rowData(4, 40)); + if (expired) { + branchTable + .schemaManager() + .commitChanges(SchemaChange.addColumn("extra", DataTypes.STRING())); + branchTable = branchTable.copyWithLatestSchema(); + } + FallbackReadFileStoreTable table = + new FallbackReadFileStoreTable(mainTable, branchTable, wrappedFirst); + Map options = new HashMap<>(); + options.put(CoreOptions.SCAN_SNAPSHOT_ID.key(), "1"); + assertThatThrownBy(() -> table.copy(options)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("No snapshot at or before"); + + options.put(CoreOptions.SCAN_FALLBACK_BRANCH_ALLOW_EARLIEST_SNAPSHOT.key(), "true"); + FallbackReadFileStoreTable historical = (FallbackReadFileStoreTable) table.copy(options); + long earliestId = expired ? 2L : 1L; + assertThat(historical.other().coreOptions().scanSnapshotId()).isEqualTo(earliestId); + assertThat(historical.other().rowType()).isEqualTo(mainTable.rowType()); + List> expected = + new ArrayList<>(Arrays.asList(Pair.of(1, wrappedFirst ? 10 : 100), Pair.of(2, 20))); + if (expired) { + expected.add(Pair.of(3, 30)); + } + assertThat(readAndCollect(historical, scan -> {})) + .containsExactlyInAnyOrderElementsOf(expected); + + options.remove(CoreOptions.SCAN_FALLBACK_BRANCH_ALLOW_EARLIEST_SNAPSHOT.key()); + assertThat( + ((FallbackReadFileStoreTable) historical.copy(options)) + .other() + .coreOptions() + .scanSnapshotId()) + .isEqualTo(earliestId); + for (String override : Arrays.asList("false", null)) { + options.put(CoreOptions.SCAN_FALLBACK_BRANCH_ALLOW_EARLIEST_SNAPSHOT.key(), override); + assertThatThrownBy(() -> historical.copy(options)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("No snapshot at or before"); + } + } + + @ParameterizedTest + @ValueSource(booleans = {false, true}) + public void testExplicitMissingSnapshotIsNotReplaced(boolean expired) throws Exception { + FileStoreTable table = createTable(); + if (expired) { + writeDataIntoTable(table, 0, rowData(1, 10)); + writeDataIntoTable(table, 1, rowData(2, 20)); + table.snapshotManager().deleteSnapshot(1); + table.snapshotManager().commitEarliestHint(2); + } + Map options = new HashMap<>(); + options.put(CoreOptions.SCAN_SNAPSHOT_ID.key(), "1"); + options.put(CoreOptions.SCAN_FALLBACK_BRANCH_ALLOW_EARLIEST_SNAPSHOT.key(), "true"); + + assertThatThrownBy(() -> table.copy(options).newScan().plan()) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining( + expired + ? "out of available snapshotId range" + : "There is currently no snapshot."); + } + + @Test + public void testTagTimeTravelWithoutActiveOtherSnapshots() throws Exception { + FileStoreTable mainTable = createTable(); + writeDataIntoTable(mainTable, 0, rowData(1, 10)); + mainTable.createTag("base"); + mainTable.createBranch("bc"); + FileStoreTable branchTable = createTableFromBranch(mainTable, "bc"); + writeDataIntoTable(branchTable, 0, rowData(2, 20)); + branchTable.createTag("base"); + branchTable.snapshotManager().deleteSnapshot(1); + branchTable.snapshotManager().deleteLatestHint(); + assertThat(branchTable.snapshotManager().latestSnapshotId()).isNull(); + FallbackReadFileStoreTable table = + (FallbackReadFileStoreTable) + new FallbackReadFileStoreTable(mainTable, branchTable, true) + .copy( + Collections.singletonMap( + CoreOptions.SCAN_TAG_NAME.key(), "base")); + + assertThat(readAndCollect(table, scan -> {})) + .containsExactlyInAnyOrder(Pair.of(1, 10), Pair.of(2, 20)); + } + private DataTableScan queryAuthScan(DataTableScan delegate, TableQueryAuthResult authResult) { DataTableScan scan = Mockito.mock(DataTableScan.class, AdditionalAnswers.delegatesTo(delegate));