This is an automated email from the ASF dual-hosted git repository.
RussellSpitzer pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg.git
The following commit(s) were added to refs/heads/main by this push:
new 42034ffd10 Core: ensure snapshot timestamp is monotonically increasing
for V4 tables (#16293)
42034ffd10 is described below
commit 42034ffd10af29548fc73fe0c8f1833d74f9c792
Author: Steven Zhen Wu <[email protected]>
AuthorDate: Wed Oct 7 14:13:35 2026 -0700
Core: ensure snapshot timestamp is monotonically increasing for V4 tables
(#16293)
---
.../test/java/org/apache/iceberg/TestHelpers.java | 4 +
.../java/org/apache/iceberg/SnapshotProducer.java | 18 ++-
.../java/org/apache/iceberg/TableMetadata.java | 22 ++--
.../org/apache/iceberg/TestRowLineageMetadata.java | 4 +-
.../org/apache/iceberg/TestSnapshotProducer.java | 127 +++++++++++++++++++++
.../java/org/apache/iceberg/TestTableMetadata.java | 114 ++++++++++++++++++
6 files changed, 277 insertions(+), 12 deletions(-)
diff --git a/api/src/test/java/org/apache/iceberg/TestHelpers.java
b/api/src/test/java/org/apache/iceberg/TestHelpers.java
index 58d992d8df..ee9c169aa8 100644
--- a/api/src/test/java/org/apache/iceberg/TestHelpers.java
+++ b/api/src/test/java/org/apache/iceberg/TestHelpers.java
@@ -80,6 +80,10 @@ public class TestHelpers {
IntStream.rangeClosed(2,
MAX_FORMAT_VERSION).boxed().collect(Collectors.toUnmodifiableList());
public static final List<Integer> V3_AND_ABOVE =
IntStream.rangeClosed(3,
MAX_FORMAT_VERSION).boxed().collect(Collectors.toUnmodifiableList());
+ public static final List<Integer> V4_AND_ABOVE =
+ IntStream.rangeClosed(4,
MAX_FORMAT_VERSION).boxed().collect(Collectors.toUnmodifiableList());
+ public static final List<Integer> V3_AND_BELOW =
+ IntStream.rangeClosed(1,
3).boxed().collect(Collectors.toUnmodifiableList());
/** Wait in a tight check loop until system clock is past {@code
timestampMillis} */
public static long waitUntilAfter(long timestampMillis) {
diff --git a/core/src/main/java/org/apache/iceberg/SnapshotProducer.java
b/core/src/main/java/org/apache/iceberg/SnapshotProducer.java
index c735769223..06281188c3 100644
--- a/core/src/main/java/org/apache/iceberg/SnapshotProducer.java
+++ b/core/src/main/java/org/apache/iceberg/SnapshotProducer.java
@@ -35,6 +35,7 @@ import com.github.benmanes.caffeine.cache.Caffeine;
import com.github.benmanes.caffeine.cache.LoadingCache;
import java.io.IOException;
import java.math.RoundingMode;
+import java.time.Clock;
import java.util.Arrays;
import java.util.Collection;
import java.util.List;
@@ -121,6 +122,7 @@ abstract class SnapshotProducer<ThisT> implements
SnapshotUpdate<ThisT> {
private SnapshotAncestryValidator snapshotAncestryValidator =
SnapshotAncestryValidator.NON_VALIDATING;
+ private Clock clock = Clock.systemUTC();
private ExecutorService workerPool;
private ExecutorService writePool;
private int writePoolParallelism = ThreadPools.WORKER_THREAD_POOL_SIZE;
@@ -362,7 +364,7 @@ abstract class SnapshotProducer<ThisT> implements
SnapshotUpdate<ThisT> {
sequenceNumber,
snapshotId(),
parentSnapshotId,
- System.currentTimeMillis(),
+ snapshotTimestampMillis(parentSnapshot),
operation(),
summary(base),
base.currentSchemaId(),
@@ -694,6 +696,20 @@ abstract class SnapshotProducer<ThisT> implements
SnapshotUpdate<ThisT> {
return ManifestFiles.readDeleteManifest(manifest, ops.io(),
ops.current().specsById());
}
+ @VisibleForTesting
+ void setClock(Clock newClock) {
+ this.clock = newClock;
+ }
+
+ private long snapshotTimestampMillis(Snapshot parentSnapshot) {
+ long now = clock.millis();
+ if (parentSnapshot != null) {
+ return Math.max(now, parentSnapshot.timestampMillis() + 1);
+ }
+
+ return now;
+ }
+
protected long snapshotId() {
if (snapshotId == null) {
synchronized (this) {
diff --git a/core/src/main/java/org/apache/iceberg/TableMetadata.java
b/core/src/main/java/org/apache/iceberg/TableMetadata.java
index a43d414590..89427e4c69 100644
--- a/core/src/main/java/org/apache/iceberg/TableMetadata.java
+++ b/core/src/main/java/org/apache/iceberg/TableMetadata.java
@@ -59,6 +59,7 @@ public class TableMetadata implements Serializable {
static final int MIN_FORMAT_VERSION_ROW_LINEAGE = 3;
static final int MIN_FORMAT_VERSION_PARQUET_MANIFESTS = 4;
static final int MIN_FORMAT_VERSION_OPTIONAL_LOCATION = 4;
+ static final int MIN_FORMAT_VERSION_MONOTONIC_TIMESTAMPS = 4;
static final int INITIAL_SPEC_ID = 0;
static final int INITIAL_SORT_ORDER_ID = 1;
static final int INITIAL_SCHEMA_ID = 0;
@@ -378,15 +379,6 @@ public class TableMetadata implements Serializable {
}
last = logEntry;
}
- if (last != null) {
- Preconditions.checkArgument(
- // commits can happen concurrently from different machines.
- // A tolerance helps us avoid failure for small clock skew
- lastUpdatedMillis - last.timestampMillis() >= -ONE_MINUTE,
- "Invalid update timestamp %s: before last snapshot log entry at %s",
- lastUpdatedMillis,
- last.timestampMillis());
- }
MetadataLogEntry previous = null;
for (MetadataLogEntry metadataEntry : previousFiles) {
@@ -1270,6 +1262,18 @@ public class TableMetadata implements Serializable {
snapshot.sequenceNumber(),
lastSequenceNumber);
+ if (formatVersion >= MIN_FORMAT_VERSION_MONOTONIC_TIMESTAMPS &&
snapshot.parentId() != null) {
+ Snapshot parent = snapshotsById.get(snapshot.parentId());
+ if (parent != null) {
+ ValidationException.check(
+ snapshot.timestampMillis() > parent.timestampMillis(),
+ "Invalid snapshot timestamp %s: not after parent snapshot %s at
%s",
+ snapshot.timestampMillis(),
+ snapshot.parentId(),
+ parent.timestampMillis());
+ }
+ }
+
this.lastSequenceNumber = snapshot.sequenceNumber();
snapshots.add(snapshot);
snapshotsById.put(snapshot.snapshotId(), snapshot);
diff --git a/core/src/test/java/org/apache/iceberg/TestRowLineageMetadata.java
b/core/src/test/java/org/apache/iceberg/TestRowLineageMetadata.java
index 4fd4f9d361..4d861d5a81 100644
--- a/core/src/test/java/org/apache/iceberg/TestRowLineageMetadata.java
+++ b/core/src/test/java/org/apache/iceberg/TestRowLineageMetadata.java
@@ -127,7 +127,7 @@ public class TestRowLineageMetadata {
1,
2,
1L,
- 0,
+ 1,
DataOperations.APPEND,
null,
1,
@@ -163,7 +163,7 @@ public class TestRowLineageMetadata {
Snapshot invalidNewRows =
new BaseSnapshot(
- 1, 2, 1L, 0, DataOperations.APPEND, null, 1, "foo",
added.nextRowId() - 1, 10L, null);
+ 1, 2, 1L, 1, DataOperations.APPEND, null, 1, "foo",
added.nextRowId() - 1, 10L, null);
assertThatThrownBy(() ->
TableMetadata.buildFrom(added).addSnapshot(invalidNewRows))
.isInstanceOf(ValidationException.class)
diff --git a/core/src/test/java/org/apache/iceberg/TestSnapshotProducer.java
b/core/src/test/java/org/apache/iceberg/TestSnapshotProducer.java
index 1e6021a5c9..8d128b9b6b 100644
--- a/core/src/test/java/org/apache/iceberg/TestSnapshotProducer.java
+++ b/core/src/test/java/org/apache/iceberg/TestSnapshotProducer.java
@@ -27,9 +27,13 @@ import static org.assertj.core.api.Assumptions.assumeThat;
import java.io.File;
import java.io.IOException;
import java.nio.file.Paths;
+import java.time.Clock;
+import java.time.Instant;
+import java.time.ZoneOffset;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
import javax.annotation.Nonnull;
import org.apache.iceberg.exceptions.ValidationException;
@@ -276,4 +280,127 @@ public class TestSnapshotProducer extends TestBase {
executor.shutdownNow();
}
}
+
+ @TestTemplate
+ public void snapshotTimestampsAreMonotonicallyIncreasing() {
+ table.newFastAppend().appendFile(FILE_A).commit();
+ Snapshot first = table.currentSnapshot();
+
+ table.newFastAppend().appendFile(FILE_B).commit();
+ Snapshot second = table.currentSnapshot();
+
assertThat(second.timestampMillis()).isGreaterThan(first.timestampMillis());
+
+ table.newFastAppend().appendFile(FILE_C).commit();
+ Snapshot third = table.currentSnapshot();
+
assertThat(third.timestampMillis()).isGreaterThan(second.timestampMillis());
+ }
+
+ @TestTemplate
+ public void lamportClockFastForwardsDriftedClock() {
+ table.newFastAppend().appendFile(FILE_A).commit();
+ long firstTs = table.currentSnapshot().timestampMillis();
+
+ long driftedTime = firstTs - TimeUnit.MINUTES.toMillis(2);
+ Clock driftedClock = Clock.fixed(Instant.ofEpochMilli(driftedTime),
ZoneOffset.UTC);
+ AppendFiles append = table.newFastAppend().appendFile(FILE_B);
+ ((SnapshotProducer<?>) append).setClock(driftedClock);
+ append.commit();
+
+
assertThat(table.currentSnapshot().timestampMillis()).isGreaterThan(firstTs);
+ }
+
+ @TestTemplate
+ public void transactionCommitsProduceMonotonicTimestamps() {
+ table.newFastAppend().appendFile(FILE_A).commit();
+ Snapshot base = table.currentSnapshot();
+
+ Transaction txn = table.newTransaction();
+ txn.newFastAppend().appendFile(FILE_B).commit();
+ Snapshot first = txn.table().currentSnapshot();
+ txn.newFastAppend().appendFile(FILE_C).commit();
+ Snapshot second = txn.table().currentSnapshot();
+ txn.commitTransaction();
+
+ assertThat(first.parentId()).isEqualTo(base.snapshotId());
+ assertThat(first.timestampMillis()).isGreaterThan(base.timestampMillis());
+ assertThat(second.parentId()).isEqualTo(first.snapshotId());
+
assertThat(second.timestampMillis()).isGreaterThan(first.timestampMillis());
+
assertThat(table.currentSnapshot().snapshotId()).isEqualTo(second.snapshotId());
+ }
+
+ @TestTemplate
+ public void transactionRetryFastForwardsTimestampsPastConflictingSnapshot() {
+ table.updateProperties().set(TableProperties.COMMIT_NUM_RETRIES,
"1").commit();
+
+ table.newFastAppend().appendFile(FILE_A).commit();
+ long baseTs = table.currentSnapshot().timestampMillis();
+ Transaction txn = table.newTransaction();
+
+ long firstCommitTs = baseTs + 1;
+ Clock firstClock = Clock.fixed(Instant.ofEpochMilli(firstCommitTs),
ZoneOffset.UTC);
+ AppendFiles firstAppend = txn.newFastAppend().appendFile(FILE_B);
+ ((SnapshotProducer<?>) firstAppend).setClock(firstClock);
+ firstAppend.commit();
+
+ long secondCommitTs = firstCommitTs + 1;
+ Clock secondClock = Clock.fixed(Instant.ofEpochMilli(secondCommitTs),
ZoneOffset.UTC);
+ AppendFiles secondAppend = txn.newFastAppend().appendFile(FILE_C);
+ ((SnapshotProducer<?>) secondAppend).setClock(secondClock);
+ secondAppend.commit();
+
+ long conflictTs = baseTs + 1_000_000L;
+ AppendFiles conflictAppend = table.newFastAppend().appendFile(FILE_D);
+ ((SnapshotProducer<?>) conflictAppend)
+ .setClock(Clock.fixed(Instant.ofEpochMilli(conflictTs),
ZoneOffset.UTC));
+ conflictAppend.commit();
+ Snapshot conflict = table.currentSnapshot();
+ assertThat(conflict.timestampMillis()).isEqualTo(conflictTs);
+
assertThat(txn.table().currentSnapshot().timestampMillis()).isLessThan(conflictTs);
+
+ txn.commitTransaction();
+
+ Snapshot retriedSecond = table.currentSnapshot();
+ Snapshot retriedFirst = table.snapshot(retriedSecond.parentId());
+ assertThat(retriedFirst.parentId()).isEqualTo(conflict.snapshotId());
+
assertThat(retriedFirst.timestampMillis()).isGreaterThan(conflict.timestampMillis());
+
assertThat(retriedSecond.timestampMillis()).isGreaterThan(retriedFirst.timestampMillis());
+ }
+
+ @TestTemplate
+ public void monotonicityIsScopedToTargetBranch() {
+ table.newFastAppend().appendFile(FILE_A).commit();
+ long mainParentTs = table.currentSnapshot().timestampMillis();
+
+ // Create a branch pointing at the current main head.
+ String branchName = "test-branch";
+ table.manageSnapshots().createBranch(branchName).commit();
+
+ // Commit to the branch with a clock far in the future so the branch head's
+ // timestamp-ms is much higher than main's head timestamp-ms.
+ long branchFutureTs = mainParentTs + 1_000_000L;
+ Clock branchClock = Clock.fixed(Instant.ofEpochMilli(branchFutureTs),
ZoneOffset.UTC);
+ AppendFiles branchAppend =
table.newFastAppend().appendFile(FILE_B).toBranch(branchName);
+ ((SnapshotProducer<?>) branchAppend).setClock(branchClock);
+ branchAppend.commit();
+ long branchTs = table.snapshot(branchName).timestampMillis();
+ assertThat(branchTs)
+ .as("branch commit should adopt the simulated far-future wall-clock
value")
+ .isEqualTo(branchFutureTs);
+
+ // Commit to main with a clock just slightly after mainParentTs but far
below branchTs.
+ // The spec requires monotonicity "on the same branch", so main's new
timestamp must be
+ // constrained only by main's previous head, not by the unrelated future
timestamp on
+ // the other branch.
+ long mainNewTs = mainParentTs + 5;
+ Clock mainClock = Clock.fixed(Instant.ofEpochMilli(mainNewTs),
ZoneOffset.UTC);
+ AppendFiles mainAppend = table.newFastAppend().appendFile(FILE_C);
+ ((SnapshotProducer<?>) mainAppend).setClock(mainClock);
+ mainAppend.commit();
+
+ long actualMainTs = table.currentSnapshot().timestampMillis();
+ assertThat(actualMainTs)
+ .as("Main's monotonicity must be relative to main's parent, not the
branch head")
+ .isEqualTo(mainNewTs)
+ .isLessThan(branchTs);
+ }
}
diff --git a/core/src/test/java/org/apache/iceberg/TestTableMetadata.java
b/core/src/test/java/org/apache/iceberg/TestTableMetadata.java
index be06fecf85..02c2340503 100644
--- a/core/src/test/java/org/apache/iceberg/TestTableMetadata.java
+++ b/core/src/test/java/org/apache/iceberg/TestTableMetadata.java
@@ -52,6 +52,7 @@ import java.util.Random;
import java.util.Set;
import java.util.SortedSet;
import java.util.UUID;
+import java.util.concurrent.TimeUnit;
import java.util.stream.IntStream;
import java.util.stream.Stream;
import org.apache.iceberg.TableMetadata.MetadataLogEntry;
@@ -72,6 +73,7 @@ import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.FieldSource;
import org.junit.jupiter.params.provider.MethodSource;
import org.junit.jupiter.params.provider.ValueSource;
@@ -2184,4 +2186,116 @@ public class TestTableMetadata {
.build();
assertThat(withTag.ref("tag1").isTag()).isTrue();
}
+
+ @ParameterizedTest
+ @FieldSource("org.apache.iceberg.TestHelpers#V4_AND_ABOVE")
+ public void addSnapshotRejectsNonMonotonicTimestampAfterV4(int
formatVersion) {
+ TableMetadata base =
+ TableMetadata.newTableMetadata(
+ TEST_SCHEMA,
+ PartitionSpec.unpartitioned(),
+ TEST_LOCATION,
+ ImmutableMap.of(TableProperties.FORMAT_VERSION,
String.valueOf(formatVersion)));
+
+ Snapshot parent =
+ new BaseSnapshot(1, 1L, null, 1_000L, null, null, null,
"file:/s1.avro", 0L, 0L, null);
+ TableMetadata withParent =
TableMetadata.buildFrom(base).addSnapshot(parent).build();
+
+ Snapshot child =
+ new BaseSnapshot(2, 2L, 1L, 1_000L, null, null, null, "file:/s2.avro",
0L, 0L, null);
+
+ assertThatThrownBy(() ->
TableMetadata.buildFrom(withParent).addSnapshot(child))
+ .isInstanceOf(ValidationException.class)
+ .hasMessage("Invalid snapshot timestamp 1000: not after parent
snapshot 1 at 1000");
+ }
+
+ @ParameterizedTest
+ @FieldSource("org.apache.iceberg.TestHelpers#V3_AND_BELOW")
+ public void addSnapshotAllowsNonMonotonicTimestampBeforeV4(int
formatVersion) {
+ TableMetadata base =
+ TableMetadata.newTableMetadata(
+ TEST_SCHEMA,
+ PartitionSpec.unpartitioned(),
+ "location",
+ ImmutableMap.of(TableProperties.FORMAT_VERSION,
String.valueOf(formatVersion)));
+
+ // v1-v3 differ in sequence number and row-lineage fields.
+ long parentSequenceNumber = formatVersion == 1 ? 0L : 1L;
+ Long parentFirstRowId =
+ formatVersion >= TableMetadata.MIN_FORMAT_VERSION_ROW_LINEAGE ? 0L :
null;
+ Long parentAddedRows =
+ formatVersion >= TableMetadata.MIN_FORMAT_VERSION_ROW_LINEAGE ? 10L :
null;
+ Snapshot parent =
+ new BaseSnapshot(
+ parentSequenceNumber,
+ 1L,
+ null,
+ 1_000L,
+ null,
+ null,
+ null,
+ "file:/s1.avro",
+ parentFirstRowId,
+ parentAddedRows,
+ null);
+ TableMetadata withParent =
TableMetadata.buildFrom(base).addSnapshot(parent).build();
+
+ long childSequenceNumber = formatVersion == 1 ? 0L : 2L;
+ Long childFirstRowId =
+ formatVersion >= TableMetadata.MIN_FORMAT_VERSION_ROW_LINEAGE ?
parentAddedRows : null;
+ Long childAddedRows =
+ formatVersion >= TableMetadata.MIN_FORMAT_VERSION_ROW_LINEAGE ? 20L :
null;
+ Snapshot child =
+ new BaseSnapshot(
+ childSequenceNumber,
+ 2L,
+ 1L,
+ 999L,
+ null,
+ null,
+ null,
+ "file:/s2.avro",
+ childFirstRowId,
+ childAddedRows,
+ null);
+
+ TableMetadata withChild =
TableMetadata.buildFrom(withParent).addSnapshot(child).build();
+ assertThat(withChild.snapshot(2L).timestampMillis()).isEqualTo(999L);
+ }
+
+ @ParameterizedTest
+ @FieldSource("org.apache.iceberg.TestHelpers#ALL_VERSIONS")
+ public void snapshotTimestampMayBeAheadOfTableMetadataLastUpdatedMillis(int
formatVersion) {
+ TableMetadata base =
+ TableMetadata.newTableMetadata(
+ TEST_SCHEMA,
+ PartitionSpec.unpartitioned(),
+ TEST_LOCATION,
+ ImmutableMap.of(TableProperties.FORMAT_VERSION,
String.valueOf(formatVersion)));
+
+ long snapshotTs = System.currentTimeMillis() +
TimeUnit.MINUTES.toMillis(2);
+ // Versions differ in sequence number and row-lineage fields.
+ long sequenceNumber = formatVersion == 1 ? 0L : 1L;
+ Long firstRowId = formatVersion >=
TableMetadata.MIN_FORMAT_VERSION_ROW_LINEAGE ? 0L : null;
+ Long addedRows = formatVersion >=
TableMetadata.MIN_FORMAT_VERSION_ROW_LINEAGE ? 10L : null;
+ Snapshot snapshot =
+ new BaseSnapshot(
+ sequenceNumber,
+ 1L,
+ null,
+ snapshotTs,
+ null,
+ null,
+ null,
+ "file:/s1.avro",
+ firstRowId,
+ addedRows,
+ null);
+
+ TableMetadata updated =
+ TableMetadata.buildFrom(base).setBranchSnapshot(snapshot,
SnapshotRef.MAIN_BRANCH).build();
+
+
assertThat(updated.currentSnapshot().timestampMillis()).isEqualTo(snapshotTs);
+ assertThat(updated.lastUpdatedMillis()).isLessThan(snapshotTs);
+ }
}
