hudi-agent commented on code in PR #19858:
URL: https://github.com/apache/hudi/pull/19858#discussion_r3951421520
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/util/StreamerUtil.java:
##########
@@ -361,7 +361,7 @@ public static HoodieTableMetaClient initTableIfNotExists(
.setUrlEncodePartitioning(conf.get(FlinkOptions.URL_ENCODE_PARTITIONING))
.setCDCEnabled(conf.get(FlinkOptions.CDC_ENABLED))
.setCDCSupplementalLoggingMode(conf.get(FlinkOptions.SUPPLEMENTAL_LOGGING_MODE))
- .setPopulateMetaFields(OptionsResolver.isPopulateMetaFields(conf))
+ .setMetaFieldsMode(OptionsResolver.getMetaFieldsMode(conf))
Review Comment:
🤖 Since MOR + selective mode is only rejected later in `HoodieWriteConfig`
validation, `initTableIfNotExists` (and `HoodieCatalog.createTable`, which
calls it with no write client at all) will already have persisted e.g.
`hoodie.meta.fields.mode=COMMIT_TIME_ONLY` on a MERGE_ON_READ table before the
coordinator's `createWriteClient` throws. Given the mode can only be changed
via hudi-cli/upgrade afterwards, would it make sense to reject that combination
here (or in `OptionsResolver.getMetaFieldsMode`) before `initTable` runs?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/configuration/OptionsResolver.java:
##########
@@ -565,10 +566,14 @@ public static boolean readCDCFromChangelog(Configuration
conf) {
* Returns whether to populate meta fields or not
*/
public static boolean isPopulateMetaFields(Configuration conf) {
- return Boolean.parseBoolean(
- conf.getString(
- HoodieTableConfig.POPULATE_META_FIELDS.key(),
- HoodieTableConfig.POPULATE_META_FIELDS.defaultValue().toString()));
+ return getMetaFieldsMode(conf).toLegacyPopulateMetaFields();
+ }
+
+ /**
+ * Resolves meta-field population, including the legacy boolean fallback.
+ */
+ public static MetaFieldsMode getMetaFieldsMode(Configuration conf) {
+ return MetaFieldsMode.resolve(conf.toMap());
Review Comment:
🤖 For an existing table that already carries a selective mode (e.g. created
by Spark with COMMIT_TIME_ONLY), `HoodieTableFactory.setupTableOptions` doesn't
copy `hoodie.meta.fields.mode` from hoodie.properties the way it does for
record key / ordering fields / hive-style, so this resolves to ALL and
`validateAgainstTableProperties` rejects the writer unless the user restates
the option. Is that intended, or should the table's mode be picked up
automatically like the other table-level properties?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-common/src/main/java/org/apache/hudi/common/engine/RecordContext.java:
##########
@@ -440,11 +441,8 @@ private SerializableBiFunction<T, HoodieSchema, String>
metadataKeyExtractor() {
return (record, schema) -> getValue(record, schema,
RECORD_KEY_METADATA_FIELD).toString();
}
- private SerializableBiFunction<T, HoodieSchema, String>
virtualKeyExtractor(String[] recordKeyFields) {
- if (recordKeyFields.length == 1) {
- // there might be consistency for record key encoding when partition
fields are multiple for cow merging,
- // currently the incoming records are using the keys from HoodieRecord
which utilities the write config and by default encodes the field name with the
value
- // while here the field names are ignored, this function would be used
to extract record keys from old base file.
+ private SerializableBiFunction<T, HoodieSchema, String>
virtualKeyExtractor(String[] recordKeyFields, int numPartitionFields) {
+ if (recordKeyFields.length == 1 && numPartitionFields <= 1) {
Review Comment:
🤖 Related to the CustomKeyGenerator thread but a different trigger: for
table version ≤ 8 with `hoodie.write.complex.keygen.new.encoding=true`,
`KeyGenUtils.encodeSingleKeyFieldNameForComplexKeyGen` (and `RowDataKeyGen` via
`useComplexKeygenNewEncoding`) emit the bare value for a single key + multiple
partition fields, while this extractor now always emits `field:value`. For a
virtual-key table that would make Flink COW merges miss updates and MOR delete
blocks miss base records. Should this be gated on
`tableConfig.getTableVersion()` too, or is v8 + new-encoding considered out of
scope here?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/io/storage/row/HoodieRowDataParquetWriter.java:
##########
@@ -77,16 +78,19 @@ public void addFooterMetadata(Map<String, String>
footerMetadata) {
@Override
public void writeRowWithMetaData(HoodieKey key, RowData row) throws
IOException {
- if (populateMetaFields) {
- RowData rowWithMeta = updateRecordMetadata(row, key,
getWrittenRecordCount());
- writeRow(key.getRecordKey(), rowWithMeta);
+ RowData rowWithMeta;
+ if (metaFieldsMode == MetaFieldsMode.ALL) {
Review Comment:
🤖 nit: this ALL / NONE / selective dispatch is duplicated verbatim in
`HoodieRowDataLanceWriter.writeRowWithMetaData` (and a close variant in
`HoodieRowDataCreateHandle`). Could it live once as a helper on
`HoodieRowDataCreation`, e.g. `create(MetaFieldsMode, instant, seqId, key,
fileName, row, withOperation, ...)`, so the next mode change only has to touch
one place?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieConcatHandle.java:
##########
@@ -93,8 +94,8 @@ public HoodieConcatHandle(HoodieWriteConfig config, String
instantTime, HoodieTa
*/
@Override
public void write(HoodieRecord oldRecord) {
- HoodieSchema oldSchema = config.populateMetaFields() ?
writeSchemaWithMetaFields : writeSchema;
- String key = oldRecord.getRecordKey(oldSchema, keyGeneratorOpt);
+ HoodieSchema oldSchema = metaFieldsMode == MetaFieldsMode.NONE ?
writeSchema : writeSchemaWithMetaFields;
Review Comment:
🤖 Is there a reason NONE still picks `writeSchema` here? `HoodieMergeHelper`
always hands the old record over in `writerSchemaWithMetaFields` (pure
projection or rewritten), so with `HoodieSparkRecord` + a key generator this
indexes the row against a struct that is 5 columns short. I notice
`concatPreservesRecordsWithTheSparkRecordType` excludes NONE — the Flink
handles were switched to always use `writeSchemaWithMetaFields`; could the
common handle do the same?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/io/storage/row/HoodieRowDataCreateHandle.java:
##########
@@ -149,14 +152,26 @@ public void write(String recordKey, String partitionPath,
RowData record) throws
String commitInstant;
RowData rowData;
if (!skipMetadataWrite) {
- seqId = preserveHoodieMetadata
- ?
record.getString(HoodieRecord.COMMIT_SEQNO_METADATA_FIELD_ORD).toString()
- : HoodieRecord.generateSequenceId(instantTime, taskPartitionId,
SEQGEN.getAndIncrement());
- commitInstant = preserveHoodieMetadata
- ?
record.getString(HoodieRecord.COMMIT_TIME_METADATA_FIELD_ORD).toString()
- : instantTime;
- rowData = HoodieRowDataCreation.create(commitInstant, seqId,
recordKey, partitionPath, path.getName(),
- record, writeConfig.allowOperationMetadataField(),
preserveHoodieMetadata);
+ if (metaFieldsMode == MetaFieldsMode.ALL) {
+ seqId = preserveHoodieMetadata
+ ?
record.getString(HoodieRecord.COMMIT_SEQNO_METADATA_FIELD_ORD).toString()
+ : HoodieRecord.generateSequenceId(instantTime, taskPartitionId,
SEQGEN.getAndIncrement());
+ commitInstant = preserveHoodieMetadata
+ ?
record.getString(HoodieRecord.COMMIT_TIME_METADATA_FIELD_ORD).toString()
+ : instantTime;
+ rowData = HoodieRowDataCreation.create(commitInstant, seqId,
recordKey, partitionPath, path.getName(),
+ record, writeConfig.allowOperationMetadataField(),
preserveHoodieMetadata);
+ } else if (metaFieldsMode == MetaFieldsMode.NONE) {
+ rowData = HoodieRowDataCreation.create(null, null, null, null, null,
+ record, writeConfig.allowOperationMetadataField(),
preserveHoodieMetadata);
+ } else {
+ commitInstant = !metaFieldsMode.isCommitTimePopulated() ? null :
preserveHoodieMetadata
Review Comment:
🤖 nit: the negated outer condition plus nested ternary is a bit hard to
parse; could you flip it to `metaFieldsMode.isCommitTimePopulated() ?
(preserveHoodieMetadata ? ... : instantTime) : null`, or pull the
preserve-vs-instantTime choice into a small `resolveCommitInstant()` helper
shared with the ALL branch above?
<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]
