hudi-agent commented on code in PR #18949:
URL: https://github.com/apache/hudi/pull/18949#discussion_r3867697924
##########
hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/TestS3EventsHoodieIncrSource.java:
##########
@@ -335,16 +346,87 @@ public void testSplitSnapshotLoad(String
snapshotCheckPoint, String exptected1,
// Verify the partitions being passed in getCloudObjectDataDF are correct.
ArgumentCaptor<Integer> argumentCaptor =
ArgumentCaptor.forClass(Integer.class);
ArgumentCaptor<Integer> argumentCaptorForMetrics =
ArgumentCaptor.forClass(Integer.class);
- verify(mockCloudObjectsSelectorCommon,
atLeastOnce()).loadAsDataset(Mockito.any(), Mockito.any(), Mockito.any(),
Mockito.eq(schemaProvider), argumentCaptor.capture());
+ verify(mockCloudObjectsSelectorCommon, atLeastOnce()).loadAsDataset(any(),
any(), any(), eq(schemaProvider), argumentCaptor.capture());
verify(metrics,
atLeastOnce()).updateStreamerSourceParallelism(argumentCaptorForMetrics.capture());
List<Integer> numPartitions;
if (snapshotCheckPoint.equals("1") || snapshotCheckPoint.equals("2")) {
numPartitions = Arrays.asList(12, 3, sourcePartitions);
} else {
numPartitions = Arrays.asList(23, sourcePartitions);
}
- Assertions.assertEquals(numPartitions, argumentCaptor.getAllValues());
- Assertions.assertEquals(numPartitions,
argumentCaptorForMetrics.getAllValues());
+ assertEquals(numPartitions, argumentCaptor.getAllValues());
+ assertEquals(numPartitions, argumentCaptorForMetrics.getAllValues());
+ }
+
+ /**
+ * Resume from `commit#fileKey` must re-include the start commit; runs on v6
and v8, COW and MOR
+ * source meta-tables since cloud event sources always use V1/requested-time
regardless of version.
+ */
+ @ParameterizedTest
+ @CsvSource({"6,COPY_ON_WRITE", "8,COPY_ON_WRITE", "6,MERGE_ON_READ",
"8,MERGE_ON_READ"})
+ void testRealQueryRunnerResumesMidCommitPagination(String
sourceTableVersion, HoodieTableType tableType) throws IOException {
+ Properties tableProps = new Properties();
+ tableProps.put(HoodieTableConfig.POPULATE_META_FIELDS.key(),
String.valueOf(true));
+ tableProps.put("hoodie.datasource.write.recordkey.field", "_row_key");
+ tableProps.put("hoodie.datasource.write.partitionpath.field", "");
+ tableProps.put(HoodieTableConfig.RECORDKEY_FIELDS.key(), "_row_key");
+ tableProps.put(HoodieTableConfig.PARTITION_FIELDS.key(), "");
+ tableProps.put(WRITE_TABLE_VERSION.key(), sourceTableVersion);
+ metaClient = getHoodieMetaClient(storageConf(), basePath(), tableProps,
tableType);
+
+ // timestamp-format instants: the incremental read normalizes
START_COMMIT/END_COMMIT
+ // through HoodieSqlCommonUtils.formatIncrementalInstant, which rejects
other formats
+ String startCommit = "20260601000001";
+ String laterCommit = "20260601000002";
+ writeS3MetadataRecords(startCommit, Arrays.asList(
+ Pair.of("path/to/file-01.json", 100L),
+ Pair.of("path/to/file-02.json", 100L),
+ Pair.of("path/to/file-03.json", 100L),
+ Pair.of("path/to/file-04.json", 100L),
+ Pair.of("path/to/file-05.json", 100L)));
+ // the second commit re-writes an existing key (an S3 re-upload), landing
in a log file on MOR
+ writeS3MetadataRecords(laterCommit,
Arrays.asList(Pair.of("path/to/file-05.json", 100L)));
+ if (tableType == HoodieTableType.MERGE_ON_READ) {
+ boolean hasLogFiles = Arrays.stream(fs().listStatus(new
Path(basePath())))
+ .anyMatch(f -> f.getPath().getName().contains(".log."));
+ assertTrue(hasLogFiles, "Expected log files in the MOR source
meta-table");
+ }
+
+ TypedProperties props = setProps(READ_UPTO_LATEST_COMMIT);
+ props.setProperty(CloudSourceConfig.ENABLE_EXISTS_CHECK.key(), "false");
+ when(mockCloudObjectsSelectorCommon.loadAsDataset(
+ any(), any(), any(), eq(schemaProvider), anyInt()))
+ .thenReturn(Option.empty());
+ when(sourceProfileSupplier.getSourceProfile()).thenReturn(null);
+
+ // Real QueryRunner so the actual Spark incremental read against the
on-disk meta-table runs.
+ S3EventsHoodieIncrSource incrSource = new S3EventsHoodieIncrSource(
+ props, jsc(), spark(),
+ new QueryRunner(spark(), props),
+ new CloudDataFetcher(props, jsc(), spark(), metrics,
mockCloudObjectsSelectorCommon),
+ new DefaultStreamContext(schemaProvider.orElse(null),
Option.of(sourceProfileSupplier)));
+
+ // Resume mid-commit at file-02; sourceLimit=250B fits file-03+file-04,
file-05 would exceed.
+ Checkpoint resumeFrom = new StreamerCheckpointV1(startCommit +
"#path/to/file-02.json");
+ Pair<Option<Dataset<Row>>, Checkpoint> result =
incrSource.fetchNextBatch(Option.of(resumeFrom), 250L);
+
+ assertEquals(
+ new StreamerCheckpointV1(startCommit + "#path/to/file-04.json"),
+ result.getRight(),
+ "Next batch must continue within the start commit, not advance to a
bare instant.");
+
+ // Filter must pass exactly file-03 and file-04 to downstream loading.
+ @SuppressWarnings("unchecked")
+ ArgumentCaptor<List<CloudObjectMetadata>> captor =
ArgumentCaptor.forClass((Class) List.class);
+ verify(mockCloudObjectsSelectorCommon).loadAsDataset(
+ any(), captor.capture(), any(), eq(schemaProvider), anyInt());
+ List<String> selectedPaths = captor.getValue().stream()
+ .map(CloudObjectMetadata::getPath)
+ .sorted()
+ .collect(java.util.stream.Collectors.toList());
+ assertEquals(2, selectedPaths.size(), "Expected file-03 and file-04, got:
" + selectedPaths);
+ assertTrue(selectedPaths.get(0).endsWith("/path/to/file-03.json"),
selectedPaths.get(0));
+ assertTrue(selectedPaths.get(1).endsWith("/path/to/file-04.json"),
selectedPaths.get(1));
}
Review Comment:
🤖 nit: `java.util.stream.Collectors.toList()` is fully-qualified here while
the GCS counterpart imports `Collectors` and uses the short form — could you
add the import and align with the GCS test style?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/HoodieFileGroupReaderBasedFileFormat.scala:
##########
@@ -590,6 +600,18 @@ class HoodieFileGroupReaderBasedFileFormat(tablePath:
String,
*/
private def isNestedPartitionField(name: String): Boolean =
name.contains(".")
+ /**
+ * Projects to `to` only when the read schema was augmented with filter-only
columns;
Review Comment:
🤖 nit: `projectIfNeeded` compares only field *names* via `sameElements` to
decide whether to skip projection — have you considered whether a comment
noting this intentional name-only check (e.g. types are guaranteed identical
when augmenting with filter-only columns) would help a future reader who
wonders why types aren't compared?
<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]
