danny0405 commented on code in PR #19114:
URL: https://github.com/apache/hudi/pull/19114#discussion_r3497382397
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/format/cdc/CdcIterators.java:
##########
@@ -417,24 +494,31 @@ private static int[] computeRequiredPos(HoodieSchema
tableSchema, HoodieSchema r
.toArray();
}
+ private static boolean isNativeCdcFileSplit(HoodieCDCFileSplit fileSplit) {
+ boolean nativeCdc =
FSUtils.matchNativeLogFile(fileSplit.getCdcFiles().get(0)).isPresent();
Review Comment:
Fixed in 3d16791c1d49: Flink AS_IS CDC iterator creation now handles
null/empty CDC file lists by returning HoodieCDCLogRecordIterator.empty()
before checking the first path.
##########
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:
Fixed in 3d16791c1d49: native CDC reads in Flink now route through
HoodieIOFactory/HoodieRowDataFileReader using the native log file suffix to
select the format, instead of calling the parquet iterator directly. The
returned iterator also closes both the row iterator and the file reader.
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/cdc/CDCFileGroupIterator.scala:
##########
@@ -233,10 +225,21 @@ class CDCFileGroupIterator(split: HoodieCDCFileGroupSplit,
// images. Keyed by the record's schema id so schema evolution is handled
correctly.
private val cdcImageConverterMap: mutable.Map[Integer,
(UnaryOperator[InternalRow], InternalRowToJsonStringConverter)] =
mutable.Map.empty
+ private lazy val cdcDataSparkSchema: StructType =
HoodieSchemaConversionUtils.convertHoodieSchemaToStructType(
+ HoodieSchemaUtils.removeMetadataFields(schema))
+
+ private lazy val nativeCdcTableSchemaOpt = {
Review Comment:
Fixed in 3d16791c1d49: renamed nativeCdcTableSchemaOpt to
nativeCdcParquetSchemaOpt to make it clear this is the CDC Parquet MessageType,
not the table schema.
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/cdc/CDCFileGroupIterator.scala:
##########
@@ -558,6 +517,88 @@ class CDCFileGroupIterator(split: HoodieCDCFileGroupSplit,
CloseableIteratorListener.addListener(keyBasedFileGroupRecordBuffer.get().getLogRecordIterator).asScala
}
+ private def readNativeCdcFile(cdcFile: String): Iterator[InternalRow] = {
+ val absCDCPath = new StoragePath(basePath, cdcFile)
+ val fileStatus = storage.getPathInfo(absCDCPath)
+ val pf = sparkPartitionedFileUtils.createPartitionedFile(
+ InternalRow.empty, absCDCPath, 0, fileStatus.getLength)
+ baseFileReader.read(pf, cdcSparkSchema, new StructType(),
+ org.apache.hudi.common.util.Option.empty(), Seq.empty, conf,
nativeCdcTableSchemaOpt)
+ }
+
+ private val nativeCdcRecordAccessor = new
HoodieCDCEngineRecordAccessor[InternalRow] {
+ override def getOperation(record: InternalRow): String =
record.getString(0)
+ override def getRecordKey(record: InternalRow): String =
record.getString(1)
+ override def getImage(record: InternalRow, ordinal: Int, imageArity: Int):
InternalRow = {
+ if (record.isNullAt(ordinal)) null else record.getStruct(ordinal,
imageArity)
+ }
+ }
+
+ private def createCdcRecordIterator(fileSplit: HoodieCDCFileSplit):
HoodieCDCLogRecordIterator[_] = {
+ if (isNativeCdcFileSplit(fileSplit)) {
+ new HoodieCDCNativeLogRecordIterator[InternalRow](
+ fileSplit.getCdcFiles.iterator(),
+ cdcFile => ClosableIterator.wrap(readNativeCdcFile(cdcFile).asJava),
+ nativeCdcRecordAccessor)
+ } else {
+ val cdcLogFiles = fileSplit.getCdcFiles.asScala.map { cdcFile =>
+ new HoodieLogFile(storage.getPathInfo(new StoragePath(basePath,
cdcFile)))
+ }.toArray
+ new HoodieCDCInlineLogRecordIterator(storage, cdcLogFiles,
cdcHoodieSchema)
+ }
+ }
+
+ private def isNativeCdcFileSplit(fileSplit: HoodieCDCFileSplit): Boolean = {
+ val nativeFlags = fileSplit.getCdcFiles.asScala.map(path =>
FSUtils.matchNativeLogFile(path).isPresent)
+ assert(nativeFlags.forall(_ == nativeFlags.head), "CDC file split cannot
mix inline and native CDC log files")
+ nativeFlags.head
Review Comment:
Fixed in 3d16791c1d49: empty CDC file lists now return
HoodieCDCLogRecordIterator.empty(), and the mixed inline/native guard now uses
ValidationUtils.checkState so it is enforced in production.
##########
hudi-common/src/main/java/org/apache/hudi/common/table/log/HoodieCDCInlineLogRecordIterator.java:
##########
@@ -0,0 +1,169 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.common.table.log;
+
+import org.apache.hudi.common.model.HoodieAvroIndexedRecord;
+import org.apache.hudi.common.model.HoodieLogFile;
+import org.apache.hudi.common.model.HoodieRecord.HoodieRecordType;
+import org.apache.hudi.common.schema.HoodieSchema;
+import org.apache.hudi.common.table.log.block.HoodieDataBlock;
+import org.apache.hudi.common.util.collection.ClosableIterator;
+import org.apache.hudi.common.util.collection.CloseableMappingIterator;
+import org.apache.hudi.exception.HoodieIOException;
+import org.apache.hudi.storage.HoodieStorage;
+
+import org.apache.avro.generic.IndexedRecord;
+
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.Iterator;
+import java.util.NoSuchElementException;
+
+/**
+ * CDC log record iterator for inline CDC log blocks.
+ */
+public class HoodieCDCInlineLogRecordIterator implements
HoodieCDCLogRecordIterator<IndexedRecord> {
+
+ private final HoodieStorage storage;
+
+ private final HoodieSchema cdcSchema;
+
+ private final Iterator<HoodieLogFile> cdcLogFileIter;
+
+ private HoodieLogFormat.Reader reader;
+
+ private ClosableIterator<HoodieCDCLogRecord<IndexedRecord>> itr;
+
+ private HoodieCDCLogRecord<IndexedRecord> record;
+
+ public HoodieCDCInlineLogRecordIterator(HoodieStorage storage,
HoodieLogFile[] cdcLogFiles, HoodieSchema cdcSchema) {
+ this.storage = storage;
+ this.cdcSchema = cdcSchema;
+ this.cdcLogFileIter = Arrays.stream(cdcLogFiles).iterator();
+ }
+
+ @Override
+ public boolean hasNext() {
+ if (record != null) {
+ return true;
+ }
+ if (itr == null || !itr.hasNext()) {
+ if (reader == null || !reader.hasNext()) {
+ if (!loadReader()) {
+ return false;
+ }
+ }
+ if (!loadItr()) {
+ return false;
+ }
+ }
+ record = itr.next();
+ return true;
+ }
+
+ private boolean loadReader() {
+ try {
+ closeReader();
+ if (cdcLogFileIter.hasNext()) {
+ reader = new HoodieLogFileReader(
+ storage, cdcLogFileIter.next(), cdcSchema,
HoodieLogFileReader.DEFAULT_BUFFER_SIZE);
+ return reader.hasNext();
+ }
+ return false;
+ } catch (IOException e) {
+ throw new HoodieIOException(e.getMessage(), e);
+ }
+ }
+
+ private boolean loadItr() {
+ HoodieDataBlock dataBlock = (HoodieDataBlock) reader.next();
+ closeItr();
+ itr = new CloseableMappingIterator<>(
+ dataBlock.getRecordIterator(HoodieRecordType.AVRO),
+ record -> new InlineCDCLogRecord(((HoodieAvroIndexedRecord) (Object)
record).getData()));
+ return itr.hasNext();
Review Comment:
Fixed in 3d16791c1d49: added a short comment explaining why the cast goes
through Object for the AVRO record iterator.
--
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]
