voonhous commented on code in PR #19388:
URL: https://github.com/apache/hudi/pull/19388#discussion_r3666972370
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/ConcurrentSchemaEvolutionTableSchemaGetter.java:
##########
@@ -160,9 +170,11 @@ Option<Pair<HoodieInstant, HoodieSchema>>
getLastCommitMetadataWithValidSchemaFr
// 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())))
Review Comment:
This makes the shared bounded-lookup primitive silently drop its bound
whenever the target has no ordering time. Only one caller actually wants that
-- the `isSchemaNull()` branch -- but every future caller inherits it, and in a
conflict-*detection* path a fail-open direction masks conflicts rather than
surfacing them.
Since `SimpleSchemaConflictResolutionStrategy` is the only production
caller, the decision could live there and this primitive could keep a strict
contract:
```java
if (writerSchemaOfTxn.isSchemaNull()) {
// A txn writing no data does not evolve the schema; adopt the table
schema as of its owner
// instant. On layout v2 an inflight owner has no completion time, so
resolve the latest.
HoodieInstant owner = currTxnOwnerInstant.get();
return getTableSchemaAtInstant(schemaResolver,
StringUtils.isNullOrEmpty(schemaResolver.getOrderingTime(owner)) ?
Option.empty() : Option.of(owner));
}
```
Behaviour is identical -- `Option.empty()` walks the same unbounded path
through `getTableSchemaFromTimelineWithCache` -- but the fail-open becomes a
deliberate call-site choice instead of a property of the primitive. Not
blocking, your call.
##########
hudi-client/hudi-client-common/src/test/java/org/apache/hudi/client/transaction/TestConcurrentSchemaEvolutionTableSchemaGetter.java:
##########
@@ -403,6 +410,76 @@ void testGetTableSchema(HoodieSchema inputSchema, boolean
includeMetadataFields,
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();
Review Comment:
Worth asserting that this fixture actually took effect.
The commits are written `001` then `009`, so the *natural* mtime order
already agrees with requested order. The assertions below only discriminate
requested-vs-completion ordering *because* of this inversion. If
`setLastModifiedTime` ever stops sticking -- mtime granularity, a different
storage impl under the harness, someone reordering the `addCommit` calls --
both orderings agree again and the whole test goes green against code that
never got fixed.
One assertion after the reload closes it:
```java
// the inversion has to stick, otherwise the assertions below also hold
under completion-time ordering
Map<String, String> completionTimes =
metaClient.getActiveTimeline().getInstantsAsStream()
.collect(Collectors.toMap(HoodieInstant::requestedTime,
HoodieInstant::getCompletionTime));
assertTrue(compareTimestamps(completionTimes.get("001"), GREATER_THAN,
completionTimes.get("009")));
```
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/SimpleSchemaConflictResolutionStrategy.java:
##########
@@ -77,7 +75,7 @@ public Option<HoodieSchema> resolveConcurrentSchemaEvolution(
// schema and writer schema.
HoodieInstant lastCompletedInstantAtTxnStart =
lastCompletedTxnOwnerInstant.isPresent()
? getInstantInTimelineImmediatelyPriorToTimestamp(
- lastCompletedTxnOwnerInstant.get().getCompletionTime(),
schemaResolver.computeSchemaEvolutionTimelineInReverseOrder()).orElse(null)
+ schemaResolver.getOrderingTime(lastCompletedTxnOwnerInstant.get()),
schemaResolver).orElse(null)
Review Comment:
Flagging the blast radius on this line. On table version 6 the bound is now
a requested time and the stream underneath is sorted by requested time, so
`lastCompletedInstantAtTxnStart` -- and `lastCompletedInstantAtTxnValidation`
just below -- can now select a different instant than before. That's every
RFC-82 case on v6, not only the null-writer-schema path this PR set out to fix.
I think the change is right (v1 completion times are mtime-derived, see the
other thread). But the added v6 coverage is one hand-picked case,
`testNoConflictBackwardsCompatible1TableVersionSix`. Parameterizing
`setupInstants` over table version so the existing suite runs on both would
actually exercise this -- same suggestion I left on the test file.
##########
hudi-common/src/main/java/org/apache/hudi/common/table/timeline/versioning/v1/InstantComparatorV1.java:
##########
@@ -74,4 +74,14 @@ public Comparator<HoodieInstant>
requestedTimeOrderedComparator() {
public Comparator<HoodieInstant> completionTimeOrderedComparator() {
return COMPLETION_TIME_BASED_COMPARATOR;
}
+
+ @Override
+ public Comparator<HoodieInstant> orderingComparator() {
+ return REQUESTED_TIME_BASED_COMPARATOR;
+ }
Review Comment:
Minor, on the framing rather than the code.
The PR description says requested-time ordering here "restores 0.x parity".
But concurrent schema evolution conflict detection landed in `e0af47fe97df`
([HUDI-8219] #12781, Mar 2025) and used completion-time ordering from its very
first commit -- there's no 0.x behaviour to restore, this code never existed
there.
The actual justification is stronger and I'd rather see it in the commit
message: v1 completion times aren't recorded on disk at all, they're
synthesized per-read from file mtime (`InstantGeneratorV1:79-80`), so they were
never a durable ordering key. Worth rewording so whoever does archaeology on
this line later gets the real reason.
##########
hudi-client/hudi-client-common/src/test/java/org/apache/hudi/client/transaction/TestSimpleSchemaConflictResolutionStrategy.java:
##########
@@ -156,6 +164,66 @@ void testNullTypeWriterSchema() throws Exception {
assertEquals(HoodieSchema.parse(SCHEMA1), result);
}
+ @Test
+ void testNullTypeWriterSchemaCurrTxnInstantWithoutCompletionTime() throws
Exception {
+ setupInstants(SCHEMA1, SCHEMA2, NULL_SCHEMA, true, false);
+ // 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);
+ }
+
+ @Test
+ void testNullTypeWriterSchemaTableVersionSixBoundedByRequestedTime() throws
Exception {
+ setupInstants(SCHEMA1, SCHEMA2, NULL_SCHEMA, true, false,
tableVersionSixProperties());
+ // Table version 6 orders the schema evolution timeline by requested time:
a curr txn owner
+ // instant requested between the two commits adopts the table schema of
the earlier commit.
+ Option<HoodieInstant> currTxnOwnerInstant = Option.of(
+ metaClient.createNewInstant(HoodieInstant.State.INFLIGHT,
COMMIT_ACTION, "0015"));
+ HoodieSchema result = strategy.resolveConcurrentSchemaEvolution(
+ table, config, lastCompletedTxnOwnerInstant,
currTxnOwnerInstant).get();
+ assertEquals(HoodieSchema.parse(SCHEMA1), result);
+ }
+
+ @Test
+ void testNullTypeWriterSchemaTableVersionSixAfterAllCommits() throws
Exception {
+ setupInstants(SCHEMA1, SCHEMA2, NULL_SCHEMA, true, false,
tableVersionSixProperties());
+ 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);
+ }
+
+ @Test
+ void testNullTypeWriterSchemaTableVersionSixBeforeAllCommits() throws
Exception {
+ setupInstants(SCHEMA1, SCHEMA2, NULL_SCHEMA, true, false,
tableVersionSixProperties());
+ // No commit is requested at or before the curr txn owner instant, so the
lookup falls back
+ // to the table create schema.
+ Option<HoodieInstant> currTxnOwnerInstant = Option.of(
+ metaClient.createNewInstant(HoodieInstant.State.INFLIGHT,
COMMIT_ACTION, "0005"));
+ HoodieSchema result = strategy.resolveConcurrentSchemaEvolution(
+ table, config, lastCompletedTxnOwnerInstant,
currTxnOwnerInstant).get();
+ assertEquals(HoodieSchema.parse(SCHEMA1), result);
Review Comment:
This one can't fail either way. `setupInstants` passes
`setTableCreateSchema(SCHEMA1)` and commit `0010` also carries SCHEMA1, so the
assertion holds whether the lookup fell back to the create schema (what the
comment claims is being tested) or simply matched `0010`.
Threading a distinct create schema through -- SCHEMA3, say -- would make it
actually test the fallback.
--
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]
