hudi-agent commented on code in PR #19903:
URL: https://github.com/apache/hudi/pull/19903#discussion_r4020884528
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/format/mor/MergeOnReadInputFormat.java:
##########
@@ -272,32 +278,43 @@ private void mayShiftInputSplit(MergeOnReadInputSplit
split) throws IOException
protected ClosableIterator<RowData> getBaseFileIterator(String path) throws
IOException {
if (path.endsWith(HoodieFileFormat.LANCE.getFileExtension())) {
- return FormatUtils.getLanceRecordIterator(path, fieldNames, fieldTypes,
requiredPos, hadoopConf);
+ if (requiredSchema == null) {
+ requiredSchema =
HoodieSchemaCache.intern(HoodieSchema.parse(tableState.getRequiredSchema()));
+ }
+ return FormatUtils.getLanceRecordIterator(path, requiredSchema,
hadoopConf);
+ }
+
+ if (readFieldTypes == null) {
+ HoodieSchema tableSchema =
HoodieSchema.parse(tableState.getTableSchema());
+ vectorColumnInfo = VectorConversionUtils.detectVectorColumns(fieldNames,
requiredPos, tableSchema);
+ readFieldTypes =
VectorConversionUtils.getParquetReadFieldTypes(fieldNames, fieldTypes,
tableSchema);
}
LinkedHashMap<String, Object> partObjects =
FilePathUtils.generatePartitionSpecs(
path,
Review Comment:
🤖 nit: `tableSchema` here is parsed with `HoodieSchema.parse(...)` directly,
but the equivalent code in `HoodieCdcSplitReaderFunction.getBaseFileIterator`
wraps it with `HoodieSchemaCache.intern(...)`. Worth making these consistent so
schema instances get interned the same way in both readers.
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-common/src/main/java/org/apache/hudi/common/table/cdc/HoodieCDCUtils.java:
##########
@@ -84,6 +86,32 @@ public static HoodieSchema schemaBySupplementalLoggingMode(
}
}
+ /**
+ * Validates that the table schema is compatible with the CDC supplemental
logging mode.
+ *
+ * <p>DATA_BEFORE and DATA_BEFORE_AFTER embed the table schema in the CDC
record's
+ * {@code before}/{@code after} fields. This makes any VECTOR column a
nested field,
+ * which is unsupported because VECTOR schemas and readers currently require
top-level fields.
+ * OP_KEY_ONLY stores only the operation and record key, so images can be
reconstructed
+ * from the original file slices without nesting VECTOR columns in the CDC
schema.
+ *
+ * @param tableConfig table configuration containing the CDC settings
+ * @param tableSchema table schema to validate
+ * @throws HoodieNotSupportedException if CDC image logging is enabled for a
schema containing VECTOR columns
+ */
+ public static void validateCdcSchema(HoodieTableConfig tableConfig,
HoodieSchema tableSchema) {
+ if (!tableConfig.isCDCEnabled() ||
tableConfig.cdcSupplementalLoggingMode() ==
HoodieCDCSupplementalLoggingMode.OP_KEY_ONLY) {
+ return;
+ }
Review Comment:
🤖 This looks resolved in the current revision: the check now lives in
`createCDCSchema` (`validateCdcImageSchema`), which is only reached from the
CDC logger/reader constructors with the record data schema after
`removeMetadataFields`, so `initTable`/bootstrap with `NULL_SCHEMA` no longer
hits `getFields()`. One trade-off worth noting for the PR description (which
still mentions table-creation/`commitTableChange` validation): the rejection
now surfaces at the first CDC write rather than at table creation.
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/reader/function/HoodieCdcSplitReaderFunction.java:
##########
@@ -280,12 +285,19 @@ private ClosableIterator<HoodieRecord<RowData>>
getFileSliceHoodieRecordIterator
/** Reads a CDC base file returning required-schema records. */
private ClosableIterator<RowData> getBaseFileIterator(String path) throws
IOException {
if (path.endsWith(HoodieFileFormat.LANCE.getFileExtension())) {
- return FormatUtils.getLanceRecordIterator(
- path, tableState.getRowType().getFieldNames(), fieldTypes,
tableState.getRequiredPositions(), getHadoopConf());
+ if (requiredSchema == null) {
+ requiredSchema =
HoodieSchemaCache.intern(HoodieSchema.parse(tableState.getRequiredSchema()));
+ }
+ return FormatUtils.getLanceRecordIterator(path, requiredSchema,
getHadoopConf());
+ }
+
+ if (readFieldTypes == null) {
+ HoodieSchema tableSchema =
HoodieSchemaCache.intern(HoodieSchema.parse(tableState.getTableSchema()));
+ String[] fullFieldNames =
tableState.getRowType().getFieldNames().toArray(new String[0]);
+ vectorColumnInfo =
VectorConversionUtils.detectVectorColumns(fullFieldNames,
tableState.getRequiredPositions(), tableSchema);
+ readFieldTypes =
VectorConversionUtils.getParquetReadFieldTypes(fullFieldNames,
fieldTypes.toArray(new DataType[0]), tableSchema);
Review Comment:
🤖 nit: this lazy-init block (readFieldTypes/vectorColumnInfo computed once,
cached in transient fields) is duplicated almost verbatim in
`MergeOnReadInputFormat.getBaseFileIterator`. Might be worth pulling the
"compute-once-and-cache" logic into a small shared helper (e.g. in
`VectorConversionUtils`) to avoid maintaining two copies.
<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]
