cshuo commented on code in PR #19114:
URL: https://github.com/apache/hudi/pull/19114#discussion_r3497112575
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/format/cdc/CdcIterators.java:
##########
@@ -390,20 +421,66 @@ protected BaseImageIterator(
this.requiredPos = computeRequiredPos(tableSchema, requiredSchema);
this.recordBuilder = new
GenericRecordBuilder(requiredSchema.getAvroSchema());
this.avroToRowDataConverter =
AvroToRowDataConverters.createRowConverter(requiredSchema, requiredRowType,
true);
+ this.nativeCdcImageProjection =
RowDataProjection.instance(requiredRowType, requiredPos);
+ this.nativeCdcImageArity =
HoodieSchemaUtils.removeMetadataFields(tableSchema).getFields().size();
+ this.cdcItr = createCdcRecordIterator(
+ conf, hadoopConf, tablePath, cdcSchema, fileSplit);
+ }
- StoragePath hadoopTablePath = new StoragePath(tablePath);
- HoodieStorage storage = HoodieStorageUtils.getStorage(
- tablePath, HadoopFSUtils.getStorageConf(hadoopConf));
- HoodieLogFile[] cdcLogFiles = fileSplit.getCdcFiles().stream()
- .map(cdcFile -> {
- try {
- return new HoodieLogFile(storage.getPathInfo(new
StoragePath(hadoopTablePath, cdcFile)));
- } catch (IOException e) {
- throw new HoodieIOException("Failed to get file status for CDC
log: " + cdcFile, e);
- }
- })
- .toArray(HoodieLogFile[]::new);
- this.cdcItr = new HoodieCDCLogRecordIterator(storage, cdcLogFiles,
cdcSchema);
+ private static HoodieCDCLogRecordIterator<?> createCdcRecordIterator(
+ org.apache.flink.configuration.Configuration conf,
+ org.apache.hadoop.conf.Configuration hadoopConf,
+ String tablePath,
+ HoodieSchema cdcSchema,
+ HoodieCDCFileSplit fileSplit) {
+ if (isNativeCdcFileSplit(fileSplit)) {
+ return new HoodieCDCNativeLogRecordIterator<>(
+ fileSplit.getCdcFiles().iterator(),
+ cdcFile -> getNativeCdcFileIterator(conf, hadoopConf, tablePath,
cdcFile, cdcSchema),
+ ROW_DATA_CDC_RECORD_ACCESSOR);
+ } else {
+ StoragePath hadoopTablePath = new StoragePath(tablePath);
+ HoodieStorage storage = HoodieStorageUtils.getStorage(
+ tablePath, HadoopFSUtils.getStorageConf(hadoopConf));
+ HoodieLogFile[] cdcLogFiles = fileSplit.getCdcFiles().stream()
+ .map(cdcFile -> {
+ try {
+ return new HoodieLogFile(storage.getPathInfo(new
StoragePath(hadoopTablePath, cdcFile)));
+ } catch (IOException e) {
+ throw new HoodieIOException("Failed to get file status for CDC
log: " + cdcFile, e);
+ }
+ })
+ .toArray(HoodieLogFile[]::new);
+ return new HoodieCDCInlineLogRecordIterator(storage, cdcLogFiles,
cdcSchema);
+ }
+ }
+
+ private static ClosableIterator<RowData> getNativeCdcFileIterator(
+ org.apache.flink.configuration.Configuration conf,
+ org.apache.hadoop.conf.Configuration hadoopConf,
+ String tablePath,
+ String cdcFile,
+ HoodieSchema cdcSchema) {
+ try {
+ DataType cdcDataType =
HoodieSchemaConverter.convertToDataType(cdcSchema);
+ RowType cdcRowType = (RowType) cdcDataType.getLogicalType();
+ return RecordIterators.getParquetRecordIterator(
Review Comment:
`getNativeCdcFileIterator` currently hardcodes
`RecordIterators.getParquetRecordIterator(...)`, so native CDC reads only work
for `.cdc.parquet`. Since native log filenames already carry the file-format
suffix, can we make this path format-agnostic and route through
`HoodieIOFactory` / `HoodieFileReaderFactory` instead?
--
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]
