hudi-agent commented on code in PR #19871:
URL: https://github.com/apache/hudi/pull/19871#discussion_r3965629278
##########
hudi-common/src/main/java/org/apache/hudi/common/schema/internal/utils/AvroSchemaEvolutionUtils.java:
##########
@@ -308,21 +308,26 @@ public static SchemaCompatibilityException
timestampPrecisionChangeError(String
* and target is long and since int can be promoted to long), colC will be
long data type in output schema.
*
*
+ * <p>Precedence when a side is degenerate (null, the NULL type, or no
fields): an absent source resolves to
Review Comment:
🤖 nit: this paragraph is a bit hard to follow for a reader who just wants
the contract. Could you trim it to the outcome ("an absent/empty source
resolves to target, checked before target") and move the "keep this check
first" rationale to a short inline comment on the source-null branch, where
someone reordering the code will actually see it?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-utilities/src/test/java/org/apache/hudi/utilities/streamer/TestHoodieIncrSourceE2E.java:
##########
@@ -520,4 +528,85 @@ public void testTargetCheckpointV2ForS3Gcs() {
assertFalse(StreamerCheckpointUtils.shouldTargetCheckpointV2(8,
GcsEventsHoodieIncrSource.class.getName()));
assertFalse(StreamerCheckpointUtils.shouldTargetCheckpointV2(6,
GcsEventsHoodieIncrSource.class.getName()));
}
-}
\ No newline at end of file
+
+ /**
+ * An empty batch from a source whose schema provider reports no schema at
all must still land an empty
+ * commit when the table's latest commit carries a schema without fields
(what a sync over column-less
+ * input leaves behind), so the checkpoint keeps advancing. With an
unchanged checkpoint the empty commit
+ * is gated by allowCommitOnNoCheckpointChange, as for any other source. The
non-empty batch that follows
+ * must then evolve the field-less table schema and land its rows.
+ */
+ @ParameterizedTest
+ @CsvSource({
+ "6, 80, false, true",
+ "8, 80, false, true",
+ "8, 70, false, false",
+ "8, 70, true, true"})
+ void testSyncE2EEmptyBatchWithAbsentSchemaAndFieldlessTableSchema(
+ String tableVersion, String returnedCheckpoint, boolean
allowCommitOnNoCheckpointChange, boolean expectNewCommit) throws Exception {
+ metaClient = getHoodieMetaClientWithTableVersion(storageConf(),
basePath(), tableVersion);
+ Schema fieldlessSchema = Schema.createRecord("hoodie_trips_record", null,
"hoodie.hoodie_trips", false, Collections.emptyList());
+ HoodieCommitMetadata seed = new HoodieCommitMetadata();
+ seed.setOperationType(WriteOperationType.INSERT);
+ seed.addMetadata(HoodieCommitMetadata.SCHEMA_KEY,
fieldlessSchema.toString());
+ seed.addMetadata(STREAMER_CHECKPOINT_KEY_V1, "70");
+
HoodieTestTable.of(metaClient).addCommit(metaClient.createNewInstantTime(false),
Option.of(seed));
+
+ TypedProperties props = setupBaseProperties(tableVersion);
+ props.put(OP_FETCH_NEXT_BATCH, OP_EMPTY_ROW_SET_NONE_NULL_CKP_V1_KEY);
+ props.put(RETURN_CHECKPOINT_KEY, returnedCheckpoint);
+ props.put(VAL_INPUT_CKP, VAL_NON_EMPTY_CKP_ALL_MEMBERS);
+ props.put(VAL_CKP_KEY_EQ_VAL, "70");
+ props.put(HoodieCommonConfig.SET_NULL_FOR_MISSING_COLUMNS.key(), "true");
+ props.put("hoodie.datasource.write.recordkey.field", "_row_key");
+ props.put("hoodie.datasource.write.partitionpath.field", "partition_path");
+
+ HoodieDeltaStreamer.Config cfg = createConfig(basePath(), null);
+ cfg.operation = WriteOperationType.UPSERT;
+ // an empty batch never reaches the transformer, but its presence selects
the row-based path in StreamSync
+ cfg.transformerClassNames =
Collections.singletonList(FlatteningTransformer.class.getName());
+ cfg.schemaProviderClassName = AbsentSchemaProvider.class.getName();
+ cfg.allowCommitOnNoCheckpointChange = allowCommitOnNoCheckpointChange;
+ new HoodieDeltaStreamer(cfg, jsc, Option.of(props)).sync();
+
+ metaClient.reloadActiveTimeline();
+ HoodieTimeline commits =
metaClient.getActiveTimeline().getCommitsTimeline().filterCompletedInstants();
+ assertEquals(expectNewCommit ? 2 : 1, commits.countInstants());
+ HoodieCommitMetadata last =
HoodieClientTestUtils.getCommitMetadataForInstant(metaClient,
commits.lastInstant().get()).get();
+ assertEquals(returnedCheckpoint,
last.getMetadata(STREAMER_CHECKPOINT_KEY_V1));
+ assertEquals(0, last.getWriteStats().size());
+ assertEquals(0, new
Schema.Parser().parse(last.getMetadata(HoodieCommitMetadata.SCHEMA_KEY)).getFields().size());
+
+ props.put(MOCK_ROWS_JSON,
"{\"_row_key\":\"k1\",\"partition_path\":\"p1\",\"timestamp\":1,\"name\":\"a\"}\n"
+ +
"{\"_row_key\":\"k2\",\"partition_path\":\"p1\",\"timestamp\":2,\"name\":\"b\"}");
+ props.put(RETURN_CHECKPOINT_KEY, "90");
+ props.put(VAL_CKP_KEY_EQ_VAL, returnedCheckpoint);
+ new HoodieDeltaStreamer(cfg, jsc, Option.of(props)).sync();
+
+ metaClient.reloadActiveTimeline();
+ commits =
metaClient.getActiveTimeline().getCommitsTimeline().filterCompletedInstants();
+ assertEquals(expectNewCommit ? 3 : 2, commits.countInstants());
+ last = HoodieClientTestUtils.getCommitMetadataForInstant(metaClient,
commits.lastInstant().get()).get();
+ assertEquals("90", last.getMetadata(STREAMER_CHECKPOINT_KEY_V1));
+ assertEquals(2, last.fetchTotalRecordsWritten());
+ Schema evolved = new
Schema.Parser().parse(last.getMetadata(HoodieCommitMetadata.SCHEMA_KEY));
+ for (String column : new String[] {"_row_key", "partition_path",
"timestamp", "name"}) {
Review Comment:
🤖 nit: `assertNotNull(evolved.getField(column), column)` reads more directly
than `assertFalse(... == null, column)`.
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
--
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]
