wombatu-kun commented on code in PR #19114:
URL: https://github.com/apache/hudi/pull/19114#discussion_r3497771679
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/cdc/CDCFileGroupIterator.scala:
##########
@@ -558,6 +517,91 @@ 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,
nativeCdcParquetSchemaOpt)
+ }
+
+ 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 (fileSplit.getCdcFiles == null || fileSplit.getCdcFiles.isEmpty) {
+ HoodieCDCLogRecordIterator.empty()
+ } else if (isNativeCdcFileSplit(fileSplit)) {
+ new HoodieCDCNativeLogRecordIterator[InternalRow](
Review Comment:
The native CDC branch wraps each per-file reader with ClosableIterator.wrap,
whose close() is a no-op, so HoodieCDCNativeLogRecordIterator.close() does not
close the InternalRow iterator returned by readNativeCdcFile (a Closeable
RecordReaderIterator for parquet). On a full drain the reader self-closes, but
an early-terminated CDC read (a LIMIT, or an exception mid-scan) leaks the
current cdc file's handle and, for vectorized reads, its off-heap column-vector
memory. This same PR closes the reader explicitly on the Flink native path via
closeReaderWithIterator and on the inline path, so the Spark native branch is
the only one that drops it. Suggest wrapping readNativeCdcFile's result in a
ClosableIterator whose close() closes the underlying reader iterator rather
than the no-op ClosableIterator.wrap.
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/cdc/CDCFileGroupIterator.scala:
##########
@@ -558,6 +517,91 @@ 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,
nativeCdcParquetSchemaOpt)
+ }
+
+ 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 (fileSplit.getCdcFiles == null || fileSplit.getCdcFiles.isEmpty) {
+ HoodieCDCLogRecordIterator.empty()
+ } else 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)
+ }
+ }
Review Comment:
The native-vs-inline classification (the cannot-mix guard and the empty/null
fallback) is duplicated here and in CdcIterators on the Flink side. It depends
only on getCdcFiles() and FSUtils.matchNativeLogFile, both already in
hudi-common, so the copies can drift - this PR already saw that when the Spark
guard used assert (compiled out under -ea) while the Flink guard used
ValidationUtils.checkState. Suggest hoisting the flag and the mixed-mode guard
onto HoodieCDCFileSplit or HoodieCDCUtils and calling it from both engines;
createCdcRecordIterator stays engine-specific.
--
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]
