This is an automated email from the ASF dual-hosted git repository.
voonhous pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git
The following commit(s) were added to refs/heads/master by this push:
new b4756518990e fix(client): fix NPE in schema conflict resolution on
commits with null writer schema (#19388)
b4756518990e is described below
commit b4756518990e0091e6a2a312f2887350d0f896d7
Author: Y Ethan Guo <[email protected]>
AuthorDate: Wed Jul 29 00:40:04 2026 -0700
fix(client): fix NPE in schema conflict resolution on commits with null
writer schema (#19388)
* fix(client): fix NPE in schema conflict resolution on commits with null
writer schema
When schema conflict resolution is enabled and a writer commits a batch
whose
writer schema is the Avro null schema (an ingestion round that writes no
data),
the resolution strategy resolves the table schema at the current
transaction's
owner instant. At pre-commit time that instant is inflight with no
completion
time, so the completion-time filter in the schema getter calls
String.compareTo(null) and the commit fails with an NPE.
- Expose the per-timeline-version instant ordering as a first-class API on
InstantComparator: orderingComparator() / getOrderingTime(instant),
requested-time based in v1 and completion-time based in v2.
- ConcurrentSchemaEvolutionTableSchemaGetter sorts and bounds the schema
evolution timeline with that ordering (completion time for table version 8
and above, requested time for earlier versions, matching 0.x). A target
instant without an ordering time no longer bounds the lookup instead of
throwing.
- SimpleSchemaConflictResolutionStrategy adopts the table schema as of the
owner instant on the null-writer-schema path, using the
version-appropriate
ordering time.
- Add regression coverage in TestSimpleSchemaConflictResolutionStrategy,
TestConcurrentSchemaEvolutionTableSchemaGetter, and
TestInstantComparators.
* test(client): parameterize schema conflict resolution over table version
6 and 8; docs(common): ordering-key invariant and upgrade-boundary completion
time
---
...ConcurrentSchemaEvolutionTableSchemaGetter.java | 26 +++--
.../SimpleSchemaConflictResolutionStrategy.java | 10 +-
...ConcurrentSchemaEvolutionTableSchemaGetter.java | 97 ++++++++++++++++
...TestSimpleSchemaConflictResolutionStrategy.java | 129 ++++++++++++++++-----
.../common/table/timeline/InstantComparator.java | 16 +++
.../versioning/v1/InstantComparatorV1.java | 10 ++
.../versioning/v2/InstantComparatorV2.java | 14 +++
.../table/timeline/TestInstantComparators.java | 35 ++++++
8 files changed, 295 insertions(+), 42 deletions(-)
diff --git
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/ConcurrentSchemaEvolutionTableSchemaGetter.java
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/ConcurrentSchemaEvolutionTableSchemaGetter.java
index bc6f9b36df90..b2315869ab2a 100644
---
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/ConcurrentSchemaEvolutionTableSchemaGetter.java
+++
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/ConcurrentSchemaEvolutionTableSchemaGetter.java
@@ -25,7 +25,7 @@ import org.apache.hudi.common.table.HoodieTableMetaClient;
import org.apache.hudi.common.table.TableSchemaResolver;
import org.apache.hudi.common.table.timeline.HoodieActiveTimeline;
import org.apache.hudi.common.table.timeline.HoodieInstant;
-import org.apache.hudi.common.table.timeline.TimelineLayout;
+import org.apache.hudi.common.table.timeline.InstantComparator;
import org.apache.hudi.common.util.ClusteringUtils;
import org.apache.hudi.common.util.Lazy;
import org.apache.hudi.common.util.Option;
@@ -60,6 +60,8 @@ class ConcurrentSchemaEvolutionTableSchemaGetter {
private final Lazy<ConcurrentHashMap<HoodieInstant, HoodieSchema>>
tableSchemaCache;
+ private final InstantComparator instantComparator;
+
private Option<HoodieInstant> latestCommitWithValidSchema = Option.empty();
@VisibleForTesting
@@ -69,10 +71,18 @@ class ConcurrentSchemaEvolutionTableSchemaGetter {
public ConcurrentSchemaEvolutionTableSchemaGetter(HoodieTableMetaClient
metaClient) {
this.metaClient = metaClient;
+ this.instantComparator =
metaClient.getTimelineLayout().getInstantComparator();
// Unbounded sized map. Should replace with some caching library.
this.tableSchemaCache = Lazy.lazily(ConcurrentHashMap::new);
}
+ /**
+ * Returns the timestamp ordering the instant in the schema evolution
timeline.
+ */
+ String getOrderingTime(HoodieInstant instant) {
+ return instantComparator.getOrderingTime(instant);
+ }
+
/**
* Handles partition column logic for a given schema.
*
@@ -160,9 +170,11 @@ class ConcurrentSchemaEvolutionTableSchemaGetter {
// the timeline finding a completed instant containing a valid schema.
ConcurrentHashMap<HoodieInstant, HoodieSchema> tableSchemaAtInstant = new
ConcurrentHashMap<>();
Option<HoodieInstant> instantWithTableSchema =
Option.fromJavaOptional(reversedTimelineStream
- // If a completion time is specified, find the first eligible instant
in the schema evolution timeline.
- // Should switch to completion time based.
- .filter(s -> instant.isEmpty() ||
compareTimestamps(s.getCompletionTime(), LESSER_THAN_OR_EQUALS,
instant.get().getCompletionTime()))
+ // Find the first eligible instant whose ordering time is no later
than the target instant's;
+ // a target instant without an ordering time (not completed yet, on
table version 8 and above)
+ // does not bound the lookup.
+ .filter(s -> instant.isEmpty() ||
StringUtils.isNullOrEmpty(getOrderingTime(instant.get()))
+ || compareTimestamps(getOrderingTime(s), LESSER_THAN_OR_EQUALS,
getOrderingTime(instant.get())))
// Make sure the commit metadata has a valid schema inside. Same
caching the result for expensive operation.
.filter(s -> {
try {
@@ -193,6 +205,8 @@ class ConcurrentSchemaEvolutionTableSchemaGetter {
/**
* Get timeline in REVERSE order that only contains completed instants which
POTENTIALLY evolve the table schema.
+ * The stream follows the timeline layout's instant ordering, newest first
(completion time for
+ * layout v2, requested time for v1).
* For types of instants that are included and not reflecting table schema
at their instant completion time please refer
* comments inside the code.
*/
@@ -214,9 +228,7 @@ class ConcurrentSchemaEvolutionTableSchemaGetter {
}
// We only care committed instant when it comes to table schema.
- TimelineLayout timelineLayout = metaClient.getTimelineLayout();
- // Table schema getter is completion time based ordering.
- Comparator<HoodieInstant> reversedComparator =
timelineLayout.getInstantComparator().completionTimeOrderedComparator().reversed();
+ Comparator<HoodieInstant> reversedComparator =
instantComparator.orderingComparator().reversed();
// The timeline still contains DELTA_COMMIT_ACTION/COMMIT_ACTION which
might not contain a valid schema
// field in their commit metadata.
diff --git
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/SimpleSchemaConflictResolutionStrategy.java
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/SimpleSchemaConflictResolutionStrategy.java
index cfcd26362552..523b21356094 100644
---
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/SimpleSchemaConflictResolutionStrategy.java
+++
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/SimpleSchemaConflictResolutionStrategy.java
@@ -30,8 +30,6 @@ import org.apache.hudi.table.HoodieTable;
import lombok.extern.slf4j.Slf4j;
-import java.util.stream.Stream;
-
import static
org.apache.hudi.client.transaction.SchemaConflictResolutionStrategy.throwConcurrentSchemaEvolutionException;
import static
org.apache.hudi.common.table.timeline.HoodieTimeline.COMPACTION_ACTION;
import static
org.apache.hudi.common.table.timeline.InstantComparison.LESSER_THAN_OR_EQUALS;
@@ -77,7 +75,7 @@ public class SimpleSchemaConflictResolutionStrategy
implements SchemaConflictRes
// schema and writer schema.
HoodieInstant lastCompletedInstantAtTxnStart =
lastCompletedTxnOwnerInstant.isPresent()
? getInstantInTimelineImmediatelyPriorToTimestamp(
- lastCompletedTxnOwnerInstant.get().getCompletionTime(),
schemaResolver.computeSchemaEvolutionTimelineInReverseOrder()).orElse(null)
+ schemaResolver.getOrderingTime(lastCompletedTxnOwnerInstant.get()),
schemaResolver).orElse(null)
: null;
// If lastCompletedInstantAtTxnValidation is null there are 2
possibilities:
// - No committed txn at validation starts
@@ -157,9 +155,9 @@ public class SimpleSchemaConflictResolutionStrategy
implements SchemaConflictRes
}
private Option<HoodieInstant>
getInstantInTimelineImmediatelyPriorToTimestamp(
- String timestamp, Stream<HoodieInstant> reverseOrderTimeline) {
- return Option.fromJavaOptional(reverseOrderTimeline
- .filter(s -> compareTimestamps(s.getCompletionTime(),
LESSER_THAN_OR_EQUALS, timestamp))
+ String timestamp, ConcurrentSchemaEvolutionTableSchemaGetter
schemaResolver) {
+ return
Option.fromJavaOptional(schemaResolver.computeSchemaEvolutionTimelineInReverseOrder()
+ .filter(s -> compareTimestamps(schemaResolver.getOrderingTime(s),
LESSER_THAN_OR_EQUALS, timestamp))
.findFirst());
}
diff --git
a/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/client/transaction/TestConcurrentSchemaEvolutionTableSchemaGetter.java
b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/client/transaction/TestConcurrentSchemaEvolutionTableSchemaGetter.java
index 909ca863bfec..6178a9847469 100644
---
a/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/client/transaction/TestConcurrentSchemaEvolutionTableSchemaGetter.java
+++
b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/client/transaction/TestConcurrentSchemaEvolutionTableSchemaGetter.java
@@ -36,6 +36,7 @@ import org.apache.hudi.common.schema.HoodieSchema;
import org.apache.hudi.common.schema.HoodieSchemaUtils;
import org.apache.hudi.common.table.HoodieTableConfig;
import org.apache.hudi.common.table.timeline.HoodieInstant;
+import org.apache.hudi.common.table.timeline.versioning.TimelineLayoutVersion;
import
org.apache.hudi.common.table.timeline.versioning.clean.CleanPlanV2MigrationHandler;
import org.apache.hudi.common.testutils.HoodieCommonTestHarness;
import org.apache.hudi.common.testutils.HoodieTestDataGenerator;
@@ -52,10 +53,17 @@ import org.junit.jupiter.params.provider.MethodSource;
import org.mockito.Mockito;
import java.io.IOException;
+import java.net.URI;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.nio.file.attribute.FileTime;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
+import java.util.Map;
import java.util.Properties;
+import java.util.stream.Collectors;
import java.util.stream.Stream;
import static org.apache.hudi.common.table.HoodieTableConfig.PARTITION_FIELDS;
@@ -70,6 +78,7 @@ import static
org.apache.hudi.common.testutils.FileCreateUtils.createRequestedDe
import static
org.apache.hudi.common.testutils.HoodieTestDataGenerator.TRIP_SCHEMA;
import static
org.apache.hudi.common.testutils.HoodieTestUtils.getDefaultStorageConf;
import static org.apache.hudi.common.util.CommitUtils.buildMetadata;
+import static org.apache.hudi.config.HoodieWriteConfig.WRITE_TABLE_VERSION;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -403,6 +412,82 @@ public class
TestConcurrentSchemaEvolutionTableSchemaGetter extends HoodieCommon
includeMetadataFields, Option.of(instant)).get());
}
+ @Test
+ void testTableVersionEightAndAboveOrdersByCompletionTime() throws Exception {
+ metaClient =
HoodieTestUtils.getMetaClientBuilder(HoodieTableType.COPY_ON_WRITE, new
Properties(), "")
+ .initTable(getDefaultStorageConf(), basePath);
+ // The ordering is driven by the timeline layout version.
+ assertEquals(TimelineLayoutVersion.VERSION_2,
metaClient.getTimelineLayoutVersion().getVersion());
+ testTable = HoodieTestTable.of(metaClient);
+
+ // Completion order inverts requested order: requested 001 completes last
(at 100) with
+ // schema 2, requested 009 completes first (at 050) with schema 1.
+ testTable.addCommit("001", Option.of("100"), Option.of(buildMetadata(
+ Collections.emptyList(), Collections.emptyMap(), Option.empty(),
WriteOperationType.UNKNOWN,
+ SCHEMA_WITHOUT_METADATA_STR2, COMMIT_ACTION)));
+ testTable.addCommit("009", Option.of("050"), Option.of(buildMetadata(
+ Collections.emptyList(), Collections.emptyMap(), Option.empty(),
WriteOperationType.UNKNOWN,
+ SCHEMA_WITHOUT_METADATA_STR, COMMIT_ACTION)));
+
+ ConcurrentSchemaEvolutionTableSchemaGetter resolver = new
ConcurrentSchemaEvolutionTableSchemaGetter(metaClient);
+ // The latest table schema follows completion time: schema 2 of requested
001, completed 100.
+ assertEquals(SCHEMA_WITHOUT_METADATA2.toString(),
+ resolver.getTableSchemaIfPresent(false,
Option.empty()).get().toString());
+ // A target completed at 075 only sees the commit completed at 050
(requested 009, schema 1).
+ assertEquals(SCHEMA_WITHOUT_METADATA.toString(),
+ resolver.getTableSchemaIfPresent(false,
+ Option.of(metaClient.getInstantGenerator().createNewInstant(
+ HoodieInstant.State.COMPLETED, COMMIT_ACTION, "005",
"075"))).get().toString());
+ }
+
+ @Test
+ void testTableVersionSixOrdersByRequestedTime() throws Exception {
+ Properties properties = new Properties();
+ properties.setProperty(WRITE_TABLE_VERSION.key(), "6");
+ metaClient =
HoodieTestUtils.getMetaClientBuilder(HoodieTableType.COPY_ON_WRITE, properties,
"")
+ .initTable(getDefaultStorageConf(), basePath);
+ // The ordering is driven by the timeline layout version.
+ assertEquals(TimelineLayoutVersion.VERSION_1,
metaClient.getTimelineLayoutVersion().getVersion());
+ testTable = HoodieTestTable.of(metaClient);
+
+ // Same layout as the table-version-8 test: requested 001 carries schema
2, requested 009
+ // carries schema 1. The completion times below are ignored by the
table-version-6
+ // (timeline layout v1) instant file naming.
+ testTable.addCommit("001", Option.of("100"), Option.of(buildMetadata(
+ Collections.emptyList(), Collections.emptyMap(), Option.empty(),
WriteOperationType.UNKNOWN,
+ SCHEMA_WITHOUT_METADATA_STR2, COMMIT_ACTION)));
+ testTable.addCommit("009", Option.of("050"), Option.of(buildMetadata(
+ Collections.emptyList(), Collections.emptyMap(), Option.empty(),
WriteOperationType.UNKNOWN,
+ SCHEMA_WITHOUT_METADATA_STR, COMMIT_ACTION)));
+ // Invert the file modification times so that the mtime-derived completion
order disagrees
+ // with the requested order, mirroring the table-version-8 fixture above.
+ Path timelinePath =
Paths.get(metaClient.getTimelinePath().makeQualified(new
URI("file:///")).toUri());
+ Files.setLastModifiedTime(timelinePath.resolve("001.commit"),
FileTime.fromMillis(2_000_000_000_000L));
+ Files.setLastModifiedTime(timelinePath.resolve("009.commit"),
FileTime.fromMillis(1_000_000_000_000L));
+ metaClient.reloadActiveTimeline();
+
+ // The mtime inversion must stick, otherwise the assertions below also
hold under completion-time
+ // ordering and the test would pass against unfixed code.
+ Map<String, String> completionTimeByRequestedTime =
metaClient.getActiveTimeline().getInstantsAsStream()
+ .collect(Collectors.toMap(HoodieInstant::requestedTime,
HoodieInstant::getCompletionTime));
+
assertTrue(completionTimeByRequestedTime.get("001").compareTo(completionTimeByRequestedTime.get("009"))
> 0);
+
+ ConcurrentSchemaEvolutionTableSchemaGetter resolver = new
ConcurrentSchemaEvolutionTableSchemaGetter(metaClient);
+ // The latest table schema follows requested time (schema 1 of requested
009), not the
+ // mtime-derived completion order which would pick schema 2 of requested
001.
+ assertEquals(SCHEMA_WITHOUT_METADATA.toString(),
+ resolver.getTableSchemaIfPresent(false,
Option.empty()).get().toString());
+ // An inflight target bounds the lookup by its requested time.
+ assertEquals(SCHEMA_WITHOUT_METADATA2.toString(),
+ resolver.getTableSchemaIfPresent(false,
+ Option.of(metaClient.getInstantGenerator().createNewInstant(
+ HoodieInstant.State.INFLIGHT, COMMIT_ACTION,
"005"))).get().toString());
+ assertEquals(SCHEMA_WITHOUT_METADATA.toString(),
+ resolver.getTableSchemaIfPresent(false,
+ Option.of(metaClient.getInstantGenerator().createNewInstant(
+ HoodieInstant.State.INFLIGHT, COMMIT_ACTION,
"999"))).get().toString());
+ }
+
private static Stream<Arguments> partitionColumnSchemaTestParams() {
return Stream.of(
Arguments.of(false, SCHEMA_WITHOUT_METADATA), // Schema with
metadata fields, don't include metadata
@@ -528,6 +613,18 @@ public class
TestConcurrentSchemaEvolutionTableSchemaGetter extends HoodieCommon
assertTrue(schema2Option.isPresent());
assertEquals(schema2.toString(), schema2Option.get().toString());
+ // A target instant without a completion time (e.g., an inflight instant
at pre-commit time)
+ // does not bound the lookup; the latest table schema is returned.
+ String inflightTimestamp =
padWithLeadingZeros(Integer.toString(startCommitTime), REQUEST_TIME_LENGTH);
+ Option<HoodieSchema> schemaAtInstantWithoutCompletionTime =
resolver.getTableSchemaIfPresent(
+ false,
+ Option.of(metaClient.getInstantGenerator().createNewInstant(
+ HoodieInstant.State.INFLIGHT,
+ tableType.equals(HoodieTableType.COPY_ON_WRITE) ? COMMIT_ACTION :
DELTA_COMMIT_ACTION,
+ inflightTimestamp)));
+ assertTrue(schemaAtInstantWithoutCompletionTime.isPresent());
+ assertEquals(schema2.toString(),
schemaAtInstantWithoutCompletionTime.get().toString());
+
// Now follow with more disqualified instants and try to get table schema
with their request time, we should back track to instant 2.
int endCommitTime = createExhaustiveDisqualifiedInstants(startCommitTime,
tableType);
metaClient.reloadActiveTimeline();
diff --git
a/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/client/transaction/TestSimpleSchemaConflictResolutionStrategy.java
b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/client/transaction/TestSimpleSchemaConflictResolutionStrategy.java
index 0796950d624d..f29a98293d48 100644
---
a/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/client/transaction/TestSimpleSchemaConflictResolutionStrategy.java
+++
b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/client/transaction/TestSimpleSchemaConflictResolutionStrategy.java
@@ -33,6 +33,7 @@ import org.apache.hudi.common.model.HoodieWriteStat;
import org.apache.hudi.common.model.WriteOperationType;
import org.apache.hudi.common.schema.HoodieSchema;
import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.table.HoodieTableVersion;
import org.apache.hudi.common.table.timeline.HoodieInstant;
import org.apache.hudi.common.table.view.FileSystemViewManager;
import org.apache.hudi.common.testutils.HoodieTestTable;
@@ -45,12 +46,17 @@ import org.apache.hudi.table.TestBaseHoodieTable;
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.MethodSource;
+import org.junit.jupiter.params.provider.ValueSource;
import org.mockito.Mock;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Properties;
+import java.util.stream.Stream;
import static
org.apache.hudi.common.table.timeline.HoodieTimeline.CLUSTERING_ACTION;
import static
org.apache.hudi.common.table.timeline.HoodieTimeline.COMMIT_ACTION;
@@ -60,6 +66,7 @@ import static
org.apache.hudi.common.testutils.HoodieCommonTestHarness.incTimest
import static
org.apache.hudi.common.testutils.HoodieTestUtils.getDefaultStorageConf;
import static org.apache.hudi.common.util.CommitUtils.buildMetadata;
import static
org.apache.hudi.config.HoodieWriteConfig.ENABLE_SCHEMA_CONFLICT_RESOLUTION;
+import static org.apache.hudi.config.HoodieWriteConfig.WRITE_TABLE_VERSION;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertThrows;
@@ -92,9 +99,25 @@ public class TestSimpleSchemaConflictResolutionStrategy {
private static final String NULL_SCHEMA = "{\"type\":\"null\"}";
private void setupInstants(String tableSchemaAtTxnStart, String
tableSchemaAtTxnValidation,
- String writerSchemaOfTxn, Boolean
enableResolution, boolean setupLegacyClustering) throws Exception {
- metaClient =
HoodieTestUtils.getMetaClientBuilder(HoodieTableType.COPY_ON_WRITE, new
Properties(), "")
- .setTableCreateSchema(SCHEMA1)
+ String writerSchemaOfTxn, boolean
enableResolution, boolean setupLegacyClustering) throws Exception {
+ setupInstants(SCHEMA1, tableSchemaAtTxnStart, tableSchemaAtTxnValidation,
writerSchemaOfTxn,
+ enableResolution, setupLegacyClustering,
HoodieTableVersion.current().versionCode());
+ }
+
+ private void setupInstants(String tableSchemaAtTxnStart, String
tableSchemaAtTxnValidation,
+ String writerSchemaOfTxn, boolean
enableResolution, boolean setupLegacyClustering,
+ int writeTableVersion) throws Exception {
+ setupInstants(SCHEMA1, tableSchemaAtTxnStart, tableSchemaAtTxnValidation,
writerSchemaOfTxn,
+ enableResolution, setupLegacyClustering, writeTableVersion);
+ }
+
+ private void setupInstants(String tableCreateSchema, String
tableSchemaAtTxnStart, String tableSchemaAtTxnValidation,
+ String writerSchemaOfTxn, boolean
enableResolution, boolean setupLegacyClustering,
+ int writeTableVersion) throws Exception {
+ Properties tableProperties = new Properties();
+ tableProperties.setProperty(WRITE_TABLE_VERSION.key(),
String.valueOf(writeTableVersion));
+ metaClient =
HoodieTestUtils.getMetaClientBuilder(HoodieTableType.COPY_ON_WRITE,
tableProperties, "")
+ .setTableCreateSchema(tableCreateSchema)
.initTable(getDefaultStorageConf(), basePath.toString());
dummyInstantGenerator = HoodieTestTable.of(metaClient);
@@ -133,71 +156,119 @@ public class TestSimpleSchemaConflictResolutionStrategy {
strategy = new SimpleSchemaConflictResolutionStrategy();
}
- @Test
- void testNoConflictFirstCommit() throws Exception {
- setupInstants(null, null, SCHEMA1, true, false);
+ // The schema evolution timeline ordering follows the table version
(requested time for table
+ // version 6, completion time for 8 and above), so the RFC-82 resolution
cases run on both. The
+ // fixture's requested and completion orders agree, so the outcomes are the
same on both versions.
+ @ParameterizedTest
+ @ValueSource(ints = {6, 8})
+ void testNoConflictFirstCommit(int writeTableVersion) throws Exception {
+ setupInstants(null, null, SCHEMA1, true, false, writeTableVersion);
HoodieSchema result = strategy.resolveConcurrentSchemaEvolution(
table, config, Option.empty(), nonTableCompactionInstant).get();
assertEquals(HoodieSchema.parse(SCHEMA1), result);
}
- @Test
- void testNullWriterSchema() throws Exception {
- setupInstants(SCHEMA1, SCHEMA1, "", true, false);
+ @ParameterizedTest
+ @ValueSource(ints = {6, 8})
+ void testNullWriterSchema(int writeTableVersion) throws Exception {
+ setupInstants(SCHEMA1, SCHEMA1, "", true, false, writeTableVersion);
assertFalse(strategy.resolveConcurrentSchemaEvolution(
table, config, lastCompletedTxnOwnerInstant,
nonTableCompactionInstant).isPresent());
}
- @Test
- void testNullTypeWriterSchema() throws Exception {
- setupInstants(SCHEMA1, SCHEMA1, NULL_SCHEMA, true, false);
+ @ParameterizedTest
+ @ValueSource(ints = {6, 8})
+ void testNullTypeWriterSchema(int writeTableVersion) throws Exception {
+ setupInstants(SCHEMA1, SCHEMA1, NULL_SCHEMA, true, false,
writeTableVersion);
HoodieSchema result = strategy.resolveConcurrentSchemaEvolution(
table, config, lastCompletedTxnOwnerInstant,
nonTableCompactionInstant).get();
assertEquals(HoodieSchema.parse(SCHEMA1), result);
}
@Test
- void testConflictSecondCommitDifferentSchema() throws Exception {
- setupInstants(null, SCHEMA1, SCHEMA2, true, false);
+ void testNullTypeWriterSchemaCurrTxnInstantWithoutCompletionTime() throws
Exception {
+ setupInstants(SCHEMA1, SCHEMA2, NULL_SCHEMA, true, false, 8);
+ // At pre-commit time the curr txn owner instant is inflight and has no
completion time;
+ // on table version 8 and above the resolution falls back to the latest
table schema.
+ Option<HoodieInstant> currTxnOwnerInstant = Option.of(
+ metaClient.createNewInstant(HoodieInstant.State.INFLIGHT,
COMMIT_ACTION, "0040"));
+ HoodieSchema result = strategy.resolveConcurrentSchemaEvolution(
+ table, config, lastCompletedTxnOwnerInstant,
currTxnOwnerInstant).get();
+ assertEquals(HoodieSchema.parse(SCHEMA2), result);
+ }
+
+ @ParameterizedTest
+ @MethodSource("tableVersionSixNullSchemaCases")
+ void testNullTypeWriterSchemaTableVersionSix(String currTxnInstantTime,
String expectedSchema) throws Exception {
+ // Table version 6 orders the schema evolution timeline by requested time,
so the curr txn owner
+ // instant's requested time bounds the null-writer-schema lookup. A create
schema (SCHEMA3)
+ // distinct from the commit schemas makes the before-all-commits fallback
observable.
+ setupInstants(SCHEMA3, SCHEMA1, SCHEMA2, NULL_SCHEMA, true, false, 6);
+ Option<HoodieInstant> currTxnOwnerInstant = Option.of(
+ metaClient.createNewInstant(HoodieInstant.State.INFLIGHT,
COMMIT_ACTION, currTxnInstantTime));
+ HoodieSchema result = strategy.resolveConcurrentSchemaEvolution(
+ table, config, lastCompletedTxnOwnerInstant,
currTxnOwnerInstant).get();
+ assertEquals(HoodieSchema.parse(expectedSchema), result);
+ }
+
+ private static Stream<Arguments> tableVersionSixNullSchemaCases() {
+ return Stream.of(
+ // curr txn requested between the two commits: adopt the earlier
commit's schema
+ Arguments.of("0015", SCHEMA1),
+ // curr txn requested after all commits: adopt the latest commit's
schema
+ Arguments.of("0040", SCHEMA2),
+ // curr txn requested before all commits: fall back to the table
create schema
+ Arguments.of("0005", SCHEMA3));
+ }
+
+ @ParameterizedTest
+ @ValueSource(ints = {6, 8})
+ void testConflictSecondCommitDifferentSchema(int writeTableVersion) throws
Exception {
+ setupInstants(null, SCHEMA1, SCHEMA2, true, false, writeTableVersion);
assertThrows(HoodieSchemaEvolutionConflictException.class,
() -> strategy.resolveConcurrentSchemaEvolution(table, config,
Option.empty(), nonTableCompactionInstant));
}
- @Test
- void testConflictSecondCommitSameSchema() throws Exception {
- setupInstants(null, SCHEMA1, SCHEMA1, true, false);
+ @ParameterizedTest
+ @ValueSource(ints = {6, 8})
+ void testConflictSecondCommitSameSchema(int writeTableVersion) throws
Exception {
+ setupInstants(null, SCHEMA1, SCHEMA1, true, false, writeTableVersion);
HoodieSchema result = strategy.resolveConcurrentSchemaEvolution(
table, config, Option.empty(), nonTableCompactionInstant).get();
assertEquals(HoodieSchema.parse(SCHEMA1), result);
}
- @Test
- void testNoConflictSameSchema() throws Exception {
- setupInstants(SCHEMA1, SCHEMA1, SCHEMA1, true, false);
+ @ParameterizedTest
+ @ValueSource(ints = {6, 8})
+ void testNoConflictSameSchema(int writeTableVersion) throws Exception {
+ setupInstants(SCHEMA1, SCHEMA1, SCHEMA1, true, false, writeTableVersion);
HoodieSchema result = strategy.resolveConcurrentSchemaEvolution(
table, config, lastCompletedTxnOwnerInstant,
nonTableCompactionInstant).get();
assertEquals(HoodieSchema.parse(SCHEMA1), result);
}
- @Test
- void testNoConflictBackwardsCompatible1() throws Exception {
- setupInstants(SCHEMA1, SCHEMA2, SCHEMA1, true, false);
+ @ParameterizedTest
+ @ValueSource(ints = {6, 8})
+ void testNoConflictBackwardsCompatible1(int writeTableVersion) throws
Exception {
+ setupInstants(SCHEMA1, SCHEMA2, SCHEMA1, true, false, writeTableVersion);
HoodieSchema result = strategy.resolveConcurrentSchemaEvolution(
table, config, lastCompletedTxnOwnerInstant,
nonTableCompactionInstant).get();
assertEquals(HoodieSchema.parse(SCHEMA2), result);
}
- @Test
- void testNoConflictBackwardsCompatible2() throws Exception {
- setupInstants(SCHEMA1, SCHEMA1, SCHEMA2, true, false);
+ @ParameterizedTest
+ @ValueSource(ints = {6, 8})
+ void testNoConflictBackwardsCompatible2(int writeTableVersion) throws
Exception {
+ setupInstants(SCHEMA1, SCHEMA1, SCHEMA2, true, false, writeTableVersion);
HoodieSchema result = strategy.resolveConcurrentSchemaEvolution(
table, config, lastCompletedTxnOwnerInstant,
nonTableCompactionInstant).get();
assertEquals(HoodieSchema.parse(SCHEMA2), result);
}
- @Test
- void testNoConflictConcurrentEvolutionSameSchema() throws Exception {
- setupInstants(SCHEMA1, SCHEMA2, SCHEMA2, true, false);
+ @ParameterizedTest
+ @ValueSource(ints = {6, 8})
+ void testNoConflictConcurrentEvolutionSameSchema(int writeTableVersion)
throws Exception {
+ setupInstants(SCHEMA1, SCHEMA2, SCHEMA2, true, false, writeTableVersion);
HoodieSchema result = strategy.resolveConcurrentSchemaEvolution(
table, config, lastCompletedTxnOwnerInstant,
nonTableCompactionInstant).get();
assertEquals(HoodieSchema.parse(SCHEMA2), result);
diff --git
a/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/InstantComparator.java
b/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/InstantComparator.java
index 4be5f941b9d8..282486670626 100644
---
a/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/InstantComparator.java
+++
b/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/InstantComparator.java
@@ -37,4 +37,20 @@ public interface InstantComparator extends Serializable {
* @return {@link Comparator<HoodieInstant>} that orders primarily based on
completion time and secondary ordering based on {@link
#requestedTimeOrderedComparator()}.
*/
Comparator<HoodieInstant> completionTimeOrderedComparator();
+
+ /**
+ * Returns the comparator implementing the instant ordering of this timeline
version:
+ * completion-time based for v2, requested-time based for v1.
+ *
+ * <p>Implementations must keep this consistent with {@link
#getOrderingTime(HoodieInstant)},
+ * which returns the primary timestamp this comparator orders by: a timeline
walk that sorts by
+ * this comparator and then bounds instants by {@code getOrderingTime}
relies on the two agreeing.
+ */
+ Comparator<HoodieInstant> orderingComparator();
+
+ /**
+ * Returns the timestamp ordering the given instant in this timeline
version: completion time
+ * for v2 (null if the instant is not completed yet), requested time for v1.
+ */
+ String getOrderingTime(HoodieInstant instant);
}
diff --git
a/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/versioning/v1/InstantComparatorV1.java
b/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/versioning/v1/InstantComparatorV1.java
index 8b21e672b797..2c4c150a1416 100644
---
a/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/versioning/v1/InstantComparatorV1.java
+++
b/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/versioning/v1/InstantComparatorV1.java
@@ -74,4 +74,14 @@ public class InstantComparatorV1 implements Serializable,
InstantComparator {
public Comparator<HoodieInstant> completionTimeOrderedComparator() {
return COMPLETION_TIME_BASED_COMPARATOR;
}
+
+ @Override
+ public Comparator<HoodieInstant> orderingComparator() {
+ return REQUESTED_TIME_BASED_COMPARATOR;
+ }
+
+ @Override
+ public String getOrderingTime(HoodieInstant instant) {
+ return instant.requestedTime();
+ }
}
diff --git
a/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/versioning/v2/InstantComparatorV2.java
b/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/versioning/v2/InstantComparatorV2.java
index ea05f70918f4..071bcabb5fa9 100644
---
a/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/versioning/v2/InstantComparatorV2.java
+++
b/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/versioning/v2/InstantComparatorV2.java
@@ -70,4 +70,18 @@ public class InstantComparatorV2 implements Serializable,
InstantComparator {
public Comparator<HoodieInstant> completionTimeOrderedComparator() {
return COMPLETION_TIME_BASED_COMPARATOR;
}
+
+ @Override
+ public Comparator<HoodieInstant> orderingComparator() {
+ return COMPLETION_TIME_BASED_COMPARATOR;
+ }
+
+ // On tables upgraded from version 6, completion times of instants before
the upgrade boundary are
+ // backfilled from the meta file modification time and are not guaranteed
durable ordering keys.
+ // This is safe: the upgrade runs a full compaction with no concurrent
writers, so those pre-upgrade
+ // completion times no longer affect concurrency or file-slicing decisions.
+ @Override
+ public String getOrderingTime(HoodieInstant instant) {
+ return instant.getCompletionTime();
+ }
}
diff --git
a/hudi-common/src/test/java/org/apache/hudi/common/table/timeline/TestInstantComparators.java
b/hudi-common/src/test/java/org/apache/hudi/common/table/timeline/TestInstantComparators.java
index f8b51f79ff69..c01cf1cf681f 100644
---
a/hudi-common/src/test/java/org/apache/hudi/common/table/timeline/TestInstantComparators.java
+++
b/hudi-common/src/test/java/org/apache/hudi/common/table/timeline/TestInstantComparators.java
@@ -20,6 +20,7 @@
package org.apache.hudi.common.table.timeline;
import
org.apache.hudi.common.table.timeline.versioning.common.InstantComparators;
+import org.apache.hudi.common.table.timeline.versioning.v1.InstantComparatorV1;
import org.apache.hudi.common.table.timeline.versioning.v2.InstantComparatorV2;
import org.junit.jupiter.api.Test;
@@ -30,6 +31,7 @@ import java.util.Comparator;
import java.util.List;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
class TestInstantComparators {
@Test
@@ -46,6 +48,39 @@ class TestInstantComparators {
assertEquals(Arrays.asList(instant1, instant3, instant2, instant4,
instant5), instants);
}
+ @Test
+ void testOrderingComparatorPerTimelineVersion() {
+ // Completion order (001 completes at 005, 002 completes at 003) inverts
requested order.
+ HoodieInstant instant1 = createCompletedHoodieInstant("001", "005");
+ HoodieInstant instant2 = createCompletedHoodieInstant("002", "003");
+
+ // Timeline layout v1 orders by requested time.
+ List<HoodieInstant> instants = Arrays.asList(instant2, instant1);
+ instants.sort(new InstantComparatorV1().orderingComparator());
+ assertEquals(Arrays.asList(instant1, instant2), instants);
+
+ // Timeline layout v2 orders by completion time.
+ instants = Arrays.asList(instant1, instant2);
+ instants.sort(new InstantComparatorV2().orderingComparator());
+ assertEquals(Arrays.asList(instant2, instant1), instants);
+ }
+
+ @Test
+ void testGetOrderingTimePerTimelineVersion() {
+ HoodieInstant completed = createCompletedHoodieInstant("001", "005");
+ HoodieInstant inflight = createInflightHoodieInstant("002");
+
+ // Timeline layout v1 orders by requested time.
+ InstantComparator comparatorV1 = new InstantComparatorV1();
+ assertEquals("001", comparatorV1.getOrderingTime(completed));
+ assertEquals("002", comparatorV1.getOrderingTime(inflight));
+
+ // Timeline layout v2 orders by completion time, which an inflight instant
does not have yet.
+ InstantComparator comparatorV2 = new InstantComparatorV2();
+ assertEquals("005", comparatorV2.getOrderingTime(completed));
+ assertNull(comparatorV2.getOrderingTime(inflight));
+ }
+
private static HoodieInstant createCompletedHoodieInstant(String
requestedTime, String completionTime) {
return new HoodieInstant(HoodieInstant.State.COMPLETED,
HoodieTimeline.COMMIT_ACTION, requestedTime, completionTime,
InstantComparatorV2.COMPLETION_TIME_BASED_COMPARATOR);
}
