This is an automated email from the ASF dual-hosted git repository.
danny0405 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git
The following commit(s) were added to refs/heads/master by this push:
new eccbef97f339 feat(flink): support metadata field modes for COW tables
(#19858)
eccbef97f339 is described below
commit eccbef97f33941f7c2826a3f59f59bc235a82ad5
Author: Danny Chan <[email protected]>
AuthorDate: Wed Sep 9 12:06:49 2026 +0800
feat(flink): support metadata field modes for COW tables (#19858)
* feat(flink)!: support metadata field modes for COW tables
---
.../org/apache/hudi/config/HoodieWriteConfig.java | 7 +-
.../apache/hudi/io/HoodieAbstractMergeHandle.java | 7 --
.../org/apache/hudi/io/HoodieConcatHandle.java | 5 +-
.../org/apache/hudi/io/HoodieWriteMergeHandle.java | 8 +-
.../TestHoodieWriteConfigMetaFieldsMode.java | 17 ++++
.../hudi/client/model/HoodieFlinkRecord.java | 12 +++
.../java/org/apache/hudi/io/FlinkConcatHandle.java | 5 +-
.../hudi/io/FlinkIncrementalConcatHandle.java | 7 +-
.../java/org/apache/hudi/io/FlinkMergeHandle.java | 10 ++
.../io/storage/row/HoodieRowDataCreateHandle.java | 31 +++++--
.../row/HoodieRowDataFileWriterFactory.java | 9 +-
.../io/storage/row/HoodieRowDataLanceWriter.java | 28 +++---
.../io/storage/row/HoodieRowDataParquetWriter.java | 28 +++---
.../hudi/client/model/TestHoodieFlinkRecord.java | 37 ++++++++
.../storage/row/TestHoodieRowDataCreateHandle.java | 61 ++++++++++++
.../storage/row/TestHoodieRowDataLanceWriter.java | 41 +++++++-
.../TestBaseSparkInternalRowReaderContext.java | 2 +-
.../apache/hudi/common/engine/RecordContext.java | 12 +--
.../hudi/common/table/HoodieTableConfig.java | 3 +-
.../hudi/common/avro/TestAvroRecordContext.java | 28 ++++++
.../common/avro/TestHoodieAvroReaderContext.java | 10 +-
.../common/table/read/SchemaHandlerTestBase.java | 1 +
.../read/TestFileGroupReaderSchemaHandler.java | 5 +
.../TestParquetRowIndexBasedSchemaHandler.java | 1 +
.../buffer/TestFileGroupRecordBufferLoader.java | 1 +
.../buffer/TestKeyBasedFileGroupRecordBuffer.java | 9 ++
.../TestSortedKeyBasedFileGroupRecordBuffer.java | 1 +
...TestStreamingKeyBasedFileGroupRecordBuffer.java | 4 +
.../TestHoodieBackedTableMetadataDataCleanup.java | 2 +-
.../apache/hudi/configuration/OptionsResolver.java | 13 ++-
.../hudi/sink/bulk/BulkInsertWriterHelper.java | 12 ++-
.../java/org/apache/hudi/util/StreamerUtil.java | 2 +-
.../client/TestFlinkWriteClientFunctional.java | 103 ++++++++++++++++++++-
.../hudi/configuration/TestOptionsResolver.java | 12 +++
.../org/apache/hudi/sink/TestWriteCopyOnWrite.java | 92 +++++++++++++++++-
.../org/apache/hudi/sink/TestWriteMergeOnRead.java | 14 ++-
.../hudi/sink/TestWriteMergeOnReadWithCompact.java | 14 ++-
.../hudi/sink/bulk/TestBulkInsertWriteHelper.java | 78 ++++++++++++++++
.../org/apache/hudi/sink/utils/TestWriteBase.java | 10 ++
.../table/TestHoodieFileGroupReaderOnFlink.java | 10 +-
.../format/TestFlinkRowDataReaderContext.java | 2 +-
.../org/apache/hudi/utils/TestStreamerUtil.java | 17 ++++
.../hudi/hadoop/TestHiveHoodieReaderContext.java | 16 ++--
.../hudi/functional/TestBufferedRecordMerger.java | 3 +-
.../hudi/functional/TestMetaFieldsModeE2E.java | 68 ++++++++++++++
.../read/TestHoodieFileGroupReaderOnSpark.scala | 12 ++-
46 files changed, 761 insertions(+), 109 deletions(-)
diff --git
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieWriteConfig.java
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieWriteConfig.java
index dd93b0bd7faf..ffe2c0deb233 100644
---
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieWriteConfig.java
+++
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieWriteConfig.java
@@ -4045,10 +4045,9 @@ public class HoodieWriteConfig extends HoodieConfig {
+ "For MoR use %s=ALL or %s=NONE.",
HoodieTableConfig.META_FIELDS_MODE.key(), metaFieldsMode,
HoodieTableConfig.META_FIELDS_MODE.key(),
HoodieTableConfig.META_FIELDS_MODE.key()));
- // Selective meta-field modes are wired only for the Spark writer path
in this release. Flink
- // RowData / Java-client writers ignore the mode and would silently
produce NONE-mode output.
- checkArgument(!(engineType != EngineType.SPARK && isSelective),
- String.format("%s=%s is currently supported for the Spark writer
only. Support for engine=%s is a follow-up. "
+ // Java-client writers do not yet support selective meta-field
population.
+ checkArgument(!(engineType == EngineType.JAVA && isSelective),
+ String.format("%s=%s is currently supported for Spark and Flink
writers only. Support for engine=%s is a follow-up. "
+ "Use %s=ALL or %s=NONE.",
HoodieTableConfig.META_FIELDS_MODE.key(), metaFieldsMode,
engineType,
HoodieTableConfig.META_FIELDS_MODE.key(),
HoodieTableConfig.META_FIELDS_MODE.key()));
diff --git
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieAbstractMergeHandle.java
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieAbstractMergeHandle.java
index 54db1467a49f..36c4525a7cf2 100644
---
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieAbstractMergeHandle.java
+++
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieAbstractMergeHandle.java
@@ -26,7 +26,6 @@ import org.apache.hudi.common.model.HoodieRecord;
import org.apache.hudi.common.model.HoodieWriteStat;
import org.apache.hudi.common.model.IOType;
import org.apache.hudi.common.util.Option;
-import org.apache.hudi.common.util.ValidationUtils;
import org.apache.hudi.config.HoodieWriteConfig;
import org.apache.hudi.keygen.BaseKeyGenerator;
import org.apache.hudi.storage.StoragePath;
@@ -86,7 +85,6 @@ public abstract class HoodieAbstractMergeHandle<T, I, K, O>
extends HoodieWriteH
this.keyGeneratorOpt = keyGeneratorOpt;
initPartitionMetadataAndFilePaths(partitionPath);
initWriteStatus(fileId, partitionPath);
- validateAndSetAndKeyGenProps(keyGeneratorOpt, config.populateMetaFields());
}
/**
@@ -151,11 +149,6 @@ public abstract class HoodieAbstractMergeHandle<T, I, K,
O> extends HoodieWriteH
setWriteStatusPath();
}
- private void validateAndSetAndKeyGenProps(Option<BaseKeyGenerator>
keyGeneratorOpt, boolean populateMetaFields) {
- ValidationUtils.checkArgument(populateMetaFields ==
!keyGeneratorOpt.isPresent());
- this.keyGeneratorOpt = keyGeneratorOpt;
- }
-
protected String createNewFileName(String oldFileName) {
return FSUtils.makeBaseFileName(instantTime, writeToken, fileId,
hoodieTable.getBaseFileExtension());
}
diff --git
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieConcatHandle.java
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieConcatHandle.java
index a7a14cc4936e..1d8823c67183 100644
---
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieConcatHandle.java
+++
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieConcatHandle.java
@@ -23,6 +23,7 @@ import org.apache.hudi.common.model.HoodieBaseFile;
import org.apache.hudi.common.model.HoodieKey;
import org.apache.hudi.common.model.HoodieRecord;
import org.apache.hudi.common.model.HoodieRecordLocation;
+import org.apache.hudi.common.model.MetaFieldsMode;
import org.apache.hudi.common.schema.HoodieSchema;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.config.HoodieWriteConfig;
@@ -93,8 +94,8 @@ public class HoodieConcatHandle<T, I, K, O> extends
HoodieWriteMergeHandle<T, I,
*/
@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;
+ String key = getRecordKey(oldRecord, oldSchema);
try {
// NOTE: We're enforcing preservation of the record metadata to keep
existing semantic
writeToFile(new HoodieKey(key, partitionPath), oldRecord, oldSchema,
config.getPayloadConfig().getProps(), true);
diff --git
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieWriteMergeHandle.java
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieWriteMergeHandle.java
index fbd7b3d37872..b29746ec583e 100644
---
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieWriteMergeHandle.java
+++
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieWriteMergeHandle.java
@@ -112,7 +112,7 @@ public class HoodieWriteMergeHandle<T, I, K, O> extends
HoodieAbstractMergeHandl
// Read from the TABLE config, not the write config -- see the note on
BaseCreateHandle. Resolved
// once rather than per record: writeToFile consults it on the
preserve-metadata path, which runs
// for every record copied forward during a merge.
- private final MetaFieldsMode metaFieldsMode =
+ protected final MetaFieldsMode metaFieldsMode =
hoodieTable.getMetaClient().getTableConfig().getMetaFieldsMode();
protected long recordsWritten = 0;
@@ -374,7 +374,7 @@ public class HoodieWriteMergeHandle<T, I, K, O> extends
HoodieAbstractMergeHandl
HoodieSchema oldSchema = writeSchemaWithMetaFields;
HoodieSchema newSchema = getNewSchema();
boolean copyOldRecord = true;
- String key = oldRecord.getRecordKey(oldSchema, keyGeneratorOpt);
+ String key = getRecordKey(oldRecord, oldSchema);
TypedProperties props = config.getPayloadConfig().getProps();
if (keyToNewRecords.containsKey(key)) {
// If we have duplicate records that we are updating, then the hoodie
record will be deflated after
@@ -420,6 +420,10 @@ public class HoodieWriteMergeHandle<T, I, K, O> extends
HoodieAbstractMergeHandl
}
}
+ protected String getRecordKey(HoodieRecord<T> record, HoodieSchema schema) {
+ return record.getRecordKey(schema, keyGeneratorOpt);
+ }
+
protected void writeToFile(HoodieKey key, HoodieRecord<T> record,
HoodieSchema schema, Properties props, boolean shouldPreserveRecordMetadata)
throws IOException {
if (shouldPreserveRecordMetadata) {
// NOTE: `FILENAME_METADATA_FIELD` has to be rewritten to correctly
point to the
diff --git
a/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/config/TestHoodieWriteConfigMetaFieldsMode.java
b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/config/TestHoodieWriteConfigMetaFieldsMode.java
index f8becb176f83..b918acfaab7c 100644
---
a/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/config/TestHoodieWriteConfigMetaFieldsMode.java
+++
b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/config/TestHoodieWriteConfigMetaFieldsMode.java
@@ -19,11 +19,14 @@
package org.apache.hudi.config;
import org.apache.hudi.common.config.TypedProperties;
+import org.apache.hudi.common.engine.EngineType;
import org.apache.hudi.common.model.HoodieTableType;
import org.apache.hudi.common.model.MetaFieldsMode;
import org.apache.hudi.common.table.HoodieTableConfig;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.EnumSource;
import java.util.Properties;
@@ -41,6 +44,20 @@ import static org.junit.jupiter.api.Assertions.assertTrue;
*/
class TestHoodieWriteConfigMetaFieldsMode {
+ @ParameterizedTest
+ @EnumSource(MetaFieldsMode.class)
+ void flinkSupportsCopyOnWriteModes(MetaFieldsMode mode) {
+ assertEquals(mode, baseBuilder().withEngineType(EngineType.FLINK)
+ .withMetaFieldsMode(mode).build().getMetaFieldsMode());
+ if (mode.isSelective()) {
+ assertThrows(IllegalArgumentException.class, () -> baseBuilder()
+ .withEngineType(EngineType.FLINK).withMetaFieldsMode(mode)
+ .withProps(mergeOnReadProps()).build());
+ assertThrows(IllegalArgumentException.class, () -> baseBuilder()
+ .withEngineType(EngineType.JAVA).withMetaFieldsMode(mode).build());
+ }
+ }
+
private static HoodieWriteConfig.Builder baseBuilder() {
return
HoodieWriteConfig.newBuilder().withPath("file:///tmp/test_hudi_meta_fields_mode");
}
diff --git
a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/client/model/HoodieFlinkRecord.java
b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/client/model/HoodieFlinkRecord.java
index 913cd5e68350..c39f1fb08a46 100644
---
a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/client/model/HoodieFlinkRecord.java
+++
b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/client/model/HoodieFlinkRecord.java
@@ -20,6 +20,7 @@ package org.apache.hudi.client.model;
import org.apache.hudi.common.avro.HoodieAvroUtils;
import org.apache.hudi.common.config.HoodieStorageConfig;
+import org.apache.hudi.common.engine.RecordContext;
import org.apache.hudi.common.model.HoodieAvroIndexedRecord;
import org.apache.hudi.common.model.HoodieKey;
import org.apache.hudi.common.model.HoodieOperation;
@@ -125,6 +126,17 @@ public class HoodieFlinkRecord extends
HoodieRecord<RowData> {
return HoodieRecordType.FLINK;
}
+ /**
+ * Returns the cached record key, computing and caching it on the first
lookup if absent.
+ * The record context is not consulted when this record already carries a
key.
+ */
+ public String getRecordKey(HoodieSchema recordSchema, RecordContext<RowData>
recordContext) {
+ if (key == null) {
+ key = new HoodieKey(recordContext.getRecordKey(data, recordSchema),
null);
+ }
+ return getRecordKey();
+ }
+
@Override
public String getRecordKey(HoodieSchema recordSchema,
Option<BaseKeyGenerator> keyGeneratorOpt) {
if (key == null) {
diff --git
a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/io/FlinkConcatHandle.java
b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/io/FlinkConcatHandle.java
index 4aa8bcc19492..d3392e199b69 100644
---
a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/io/FlinkConcatHandle.java
+++
b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/io/FlinkConcatHandle.java
@@ -56,8 +56,9 @@ public class FlinkConcatHandle<T, I, K, O>
*/
@Override
public void write(HoodieRecord oldRecord) {
- HoodieSchema oldSchema = config.populateMetaFields() ?
writeSchemaWithMetaFields : writeSchema;
- String key = oldRecord.getRecordKey(oldSchema, keyGeneratorOpt);
+ // Match the schema supplied by HoodieMergeHelper: metadata columns are
present even in NONE mode.
+ HoodieSchema oldSchema = writeSchemaWithMetaFields;
+ String key = getRecordKey(oldRecord, oldSchema);
try {
fileWriter.write(key, oldRecord, oldSchema);
} catch (IOException | RuntimeException e) {
diff --git
a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/io/FlinkIncrementalConcatHandle.java
b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/io/FlinkIncrementalConcatHandle.java
index 1e3602b196a0..cc8679b8d885 100644
---
a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/io/FlinkIncrementalConcatHandle.java
+++
b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/io/FlinkIncrementalConcatHandle.java
@@ -57,10 +57,11 @@ public class FlinkIncrementalConcatHandle<T, I, K, O>
*/
@Override
public void write(HoodieRecord oldRecord) {
- HoodieSchema oldSchema = config.populateMetaFields() ?
writeSchemaWithMetaFields : writeSchema;
- String key = oldRecord.getRecordKey(oldSchema, keyGeneratorOpt);
+ // Match the schema supplied by HoodieMergeHelper: metadata columns are
present even in NONE mode.
+ HoodieSchema oldSchema = writeSchemaWithMetaFields;
+ String key = getRecordKey(oldRecord, oldSchema);
try {
- fileWriter.write(key, oldRecord, writeSchema);
+ fileWriter.write(key, oldRecord, oldSchema);
} catch (IOException | RuntimeException e) {
String errMsg = String.format(
"Failed to write old record into new file for key %s from old file
%s to new file %s with writerSchema %s",
diff --git
a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/io/FlinkMergeHandle.java
b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/io/FlinkMergeHandle.java
index b8f8c90310f6..5d5d496667ad 100644
---
a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/io/FlinkMergeHandle.java
+++
b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/io/FlinkMergeHandle.java
@@ -18,10 +18,13 @@
package org.apache.hudi.io;
+import org.apache.hudi.client.model.HoodieFlinkRecord;
+import org.apache.hudi.common.engine.RecordContext;
import org.apache.hudi.common.engine.TaskContextSupplier;
import org.apache.hudi.common.fs.FSUtils;
import org.apache.hudi.common.model.HoodieBaseFile;
import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.common.schema.HoodieSchema;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.config.HoodieWriteConfig;
import org.apache.hudi.exception.HoodieException;
@@ -31,6 +34,7 @@ import org.apache.hudi.table.marker.WriteMarkers;
import org.apache.hudi.table.marker.WriteMarkersFactory;
import lombok.extern.slf4j.Slf4j;
+import org.apache.flink.table.data.RowData;
import java.io.IOException;
import java.util.HashMap;
@@ -110,6 +114,12 @@ public class FlinkMergeHandle<T, I, K, O>
writeMarkers.createIfNotExists(partitionPath, dataFileName, getIOType());
}
+ @Override
+ @SuppressWarnings("unchecked")
+ protected String getRecordKey(HoodieRecord<T> record, HoodieSchema schema) {
+ return ((HoodieFlinkRecord) record).getRecordKey(schema,
(RecordContext<RowData>) readerContext.getRecordContext());
+ }
+
@Override
protected void initIncomingRecordsMap() {
log.info("Initialize on-heap keyToNewRecords for incoming records.");
diff --git
a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/io/storage/row/HoodieRowDataCreateHandle.java
b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/io/storage/row/HoodieRowDataCreateHandle.java
index 60ab015ef017..f5cd0c739930 100644
---
a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/io/storage/row/HoodieRowDataCreateHandle.java
+++
b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/io/storage/row/HoodieRowDataCreateHandle.java
@@ -29,6 +29,7 @@ import org.apache.hudi.common.model.HoodieRecordDelegate;
import org.apache.hudi.common.model.HoodieRecordLocation;
import org.apache.hudi.common.model.HoodieWriteStat;
import org.apache.hudi.common.model.IOType;
+import org.apache.hudi.common.model.MetaFieldsMode;
import org.apache.hudi.common.schema.HoodieSchema;
import org.apache.hudi.common.util.HoodieTimer;
import org.apache.hudi.common.util.Option;
@@ -80,6 +81,7 @@ public class HoodieRowDataCreateHandle implements
Serializable {
private final String fileId;
private final boolean preserveHoodieMetadata;
private final boolean skipMetadataWrite;
+ private final MetaFieldsMode metaFieldsMode;
// The schema (with meta fields appended when required) used to create the
file writer.
private final HoodieSchema writerSchema;
private final HoodieStorage storage;
@@ -105,6 +107,7 @@ public class HoodieRowDataCreateHandle implements
Serializable {
this.newRecordLocation = new HoodieRecordLocation(instantTime, fileId);
this.preserveHoodieMetadata = preserveHoodieMetadata;
this.skipMetadataWrite = skipMetadataWrite;
+ this.metaFieldsMode = writeConfig.getMetaFieldsMode();
this.writerSchema = schema;
this.currTimer = HoodieTimer.start();
this.storage = table.getStorage();
@@ -149,14 +152,26 @@ public class HoodieRowDataCreateHandle implements
Serializable {
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
+ ?
record.getString(HoodieRecord.COMMIT_TIME_METADATA_FIELD_ORD).toString()
+ : instantTime;
+ rowData = HoodieRowDataCreation.create(commitInstant, null, null,
null,
+ metaFieldsMode.isFileNamePopulated() ? path.getName() : null,
+ record, writeConfig.allowOperationMetadataField(),
preserveHoodieMetadata);
+ }
} else {
rowData = record;
}
diff --git
a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/io/storage/row/HoodieRowDataFileWriterFactory.java
b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/io/storage/row/HoodieRowDataFileWriterFactory.java
index 7b15cfaca90d..1f7323604d1c 100644
---
a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/io/storage/row/HoodieRowDataFileWriterFactory.java
+++
b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/io/storage/row/HoodieRowDataFileWriterFactory.java
@@ -93,7 +93,7 @@ public class HoodieRowDataFileWriterFactory extends
HoodieFileWriterFactory {
HoodieConfig config,
HoodieSchema schema,
TaskContextSupplier taskContextSupplier) throws IOException {
- boolean populateMetaFields =
MetaFieldsMode.resolve(config).toLegacyPopulateMetaFields();
+ MetaFieldsMode metaFieldsMode = MetaFieldsMode.resolve(config);
boolean withOperation =
config.getBooleanOrDefault(HoodieWriteConfig.ALLOW_OPERATION_METADATA_FIELD);
Pair<StorageConfiguration, HoodieConfig> injectedConfigs =
@@ -102,14 +102,14 @@ public class HoodieRowDataFileWriterFactory extends
HoodieFileWriterFactory {
HoodieConfig hoodieConfig = injectedConfigs.getRight();
Configuration conf = (Configuration)
storageConfiguration.unwrapAs(Configuration.class);
- BloomFilter filter = createBloomFilter(hoodieConfig);
+ BloomFilter filter = enableBloomFilter(metaFieldsMode, hoodieConfig) ?
createBloomFilter(hoodieConfig) : null;
HoodieRowDataParquetWriteSupport writeSupport =
(HoodieRowDataParquetWriteSupport) ReflectionUtils.loadClass(
hoodieConfig.getStringOrDefault(HoodieStorageConfig.HOODIE_PARQUET_FLINK_ROW_DATA_WRITE_SUPPORT_CLASS),
new Class<?>[] {Configuration.class, HoodieSchema.class,
BloomFilter.class},
conf, schema, filter);
return new HoodieRowDataParquetWriter(storagePath,
getParquetConfig(hoodieConfig, writeSupport),
- instantTime, taskContextSupplier, populateMetaFields, withOperation);
+ instantTime, taskContextSupplier, metaFieldsMode, withOperation);
}
@Override
@@ -120,7 +120,6 @@ public class HoodieRowDataFileWriterFactory extends
HoodieFileWriterFactory {
HoodieSchema schema,
TaskContextSupplier taskContextSupplier) {
MetaFieldsMode metaFieldsMode = MetaFieldsMode.resolve(config);
- boolean populateMetaFields = metaFieldsMode.toLegacyPopulateMetaFields();
boolean withOperation =
config.getBooleanOrDefault(HoodieWriteConfig.ALLOW_OPERATION_METADATA_FIELD);
Option<org.apache.hudi.common.bloom.BloomFilter> bloomFilter =
enableBloomFilter(metaFieldsMode, config)
? Option.of(createBloomFilter(config)) : Option.empty();
@@ -134,7 +133,7 @@ public class HoodieRowDataFileWriterFactory extends
HoodieFileWriterFactory {
config.getLongOrDefault(HoodieStorageConfig.LANCE_WRITE_ALLOCATOR_SIZE_BYTES),
config.getLongOrDefault(HoodieStorageConfig.LANCE_WRITE_FLUSH_BYTE_WATERMARK),
config.getBooleanOrDefault(HoodieStorageConfig.WRITE_UTC_TIMEZONE),
- populateMetaFields,
+ metaFieldsMode,
withOperation);
}
diff --git
a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/io/storage/row/HoodieRowDataLanceWriter.java
b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/io/storage/row/HoodieRowDataLanceWriter.java
index 238974cf3221..e77d8ab624a4 100644
---
a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/io/storage/row/HoodieRowDataLanceWriter.java
+++
b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/io/storage/row/HoodieRowDataLanceWriter.java
@@ -23,6 +23,7 @@ import org.apache.hudi.common.bloom.BloomFilter;
import org.apache.hudi.common.engine.TaskContextSupplier;
import org.apache.hudi.common.model.HoodieKey;
import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.common.model.MetaFieldsMode;
import org.apache.hudi.common.schema.HoodieSchema;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.common.util.ValidationUtils;
@@ -55,7 +56,7 @@ public class HoodieRowDataLanceWriter extends
HoodieBaseLanceWriter<RowData, Str
private final String instantTime;
private final long maxFileSize;
private final boolean utcTimestamp;
- private final boolean populateMetaFields;
+ private final MetaFieldsMode metaFieldsMode;
private final boolean withOperation;
private final Function<Long, String> seqIdGenerator;
private long recordCountForNextSizeCheck = MIN_RECORDS_FOR_SIZE_CHECK;
@@ -70,7 +71,7 @@ public class HoodieRowDataLanceWriter extends
HoodieBaseLanceWriter<RowData, Str
long allocatorSize,
long flushByteWatermark,
boolean utcTimestamp,
- boolean populateMetaFields,
+ MetaFieldsMode metaFieldsMode,
boolean withOperation) {
super(file, DEFAULT_BATCH_SIZE, allocatorSize, flushByteWatermark,
bloomFilterOpt.map(HoodieBloomFilterRowDataWriteSupport::new));
@@ -86,7 +87,7 @@ public class HoodieRowDataLanceWriter extends
HoodieBaseLanceWriter<RowData, Str
this.instantTime = instantTime;
this.maxFileSize = maxFileSize;
this.utcTimestamp = utcTimestamp;
- this.populateMetaFields = populateMetaFields;
+ this.metaFieldsMode = metaFieldsMode;
this.withOperation = withOperation;
this.seqIdGenerator = recordIndex -> {
Integer partitionId = taskContextSupplier.getPartitionIdSupplier().get();
@@ -118,12 +119,20 @@ public class HoodieRowDataLanceWriter extends
HoodieBaseLanceWriter<RowData, Str
@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) {
+ rowWithMeta = HoodieRowDataCreation.create(instantTime,
seqIdGenerator.apply(getWrittenRecordCount()),
+ key.getRecordKey(), key.getPartitionPath(), fileName, row,
withOperation, true);
+ } else if (metaFieldsMode == MetaFieldsMode.NONE) {
+ rowWithMeta = row;
} else {
- writeRow(key.getRecordKey(), row);
+ rowWithMeta = HoodieRowDataCreation.create(
+ metaFieldsMode.isCommitTimePopulated() ? instantTime : null,
+ null, null, null,
+ metaFieldsMode.isFileNamePopulated() ? fileName : null,
+ row, withOperation, true);
}
+ writeRow(key.getRecordKey(), rowWithMeta);
}
@Override
@@ -170,9 +179,4 @@ public class HoodieRowDataLanceWriter extends
HoodieBaseLanceWriter<RowData, Str
rowId = 0;
}
}
-
- private RowData updateRecordMetadata(RowData row, HoodieKey key, long
recordCount) {
- return HoodieRowDataCreation.create(instantTime,
seqIdGenerator.apply(recordCount),
- key.getRecordKey(), key.getPartitionPath(), fileName, row,
withOperation, true);
- }
}
diff --git
a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/io/storage/row/HoodieRowDataParquetWriter.java
b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/io/storage/row/HoodieRowDataParquetWriter.java
index 46e119610e7e..f9dd49d3c674 100644
---
a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/io/storage/row/HoodieRowDataParquetWriter.java
+++
b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/io/storage/row/HoodieRowDataParquetWriter.java
@@ -23,6 +23,7 @@ import org.apache.hudi.common.config.HoodieParquetConfig;
import org.apache.hudi.common.engine.TaskContextSupplier;
import org.apache.hudi.common.model.HoodieKey;
import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.common.model.MetaFieldsMode;
import org.apache.hudi.io.hadoop.HoodieBaseParquetWriter;
import org.apache.hudi.storage.StoragePath;
@@ -41,7 +42,7 @@ public class HoodieRowDataParquetWriter extends
HoodieBaseParquetWriter<RowData>
private final String fileName;
private final String instantTime;
- private final boolean populateMetaFields;
+ private final MetaFieldsMode metaFieldsMode;
private final boolean withOperation;
private final Function<Long, String> seqIdGenerator;
@@ -50,13 +51,13 @@ public class HoodieRowDataParquetWriter extends
HoodieBaseParquetWriter<RowData>
HoodieParquetConfig<HoodieRowDataParquetWriteSupport> parquetConfig,
String instantTime,
TaskContextSupplier taskContextSupplier,
- boolean populateMetaFields,
+ MetaFieldsMode metaFieldsMode,
boolean withOperation) throws IOException {
super(file, parquetConfig);
this.fileName = file.getName();
this.writeSupport = parquetConfig.getWriteSupport();
this.instantTime = instantTime;
- this.populateMetaFields = populateMetaFields;
+ this.metaFieldsMode = metaFieldsMode;
this.withOperation = withOperation;
this.seqIdGenerator = recordIndex -> {
Integer partitionId = taskContextSupplier.getPartitionIdSupplier().get();
@@ -77,16 +78,19 @@ public class HoodieRowDataParquetWriter extends
HoodieBaseParquetWriter<RowData>
@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) {
+ rowWithMeta = HoodieRowDataCreation.create(instantTime,
seqIdGenerator.apply(getWrittenRecordCount()),
+ key.getRecordKey(), key.getPartitionPath(), fileName, row,
withOperation, true);
+ } else if (metaFieldsMode == MetaFieldsMode.NONE) {
+ rowWithMeta = row;
} else {
- writeRow(key.getRecordKey(), row);
+ rowWithMeta = HoodieRowDataCreation.create(
+ metaFieldsMode.isCommitTimePopulated() ? instantTime : null,
+ null, null, null,
+ metaFieldsMode.isFileNamePopulated() ? fileName : null,
+ row, withOperation, true);
}
- }
-
- private RowData updateRecordMetadata(RowData row, HoodieKey key, long
recordCount) {
- return HoodieRowDataCreation.create(instantTime,
seqIdGenerator.apply(recordCount),
- key.getRecordKey(), key.getPartitionPath(), fileName, row,
withOperation, true);
+ writeRow(key.getRecordKey(), rowWithMeta);
}
}
diff --git
a/hudi-client/hudi-flink-client/src/test/java/org/apache/hudi/client/model/TestHoodieFlinkRecord.java
b/hudi-client/hudi-flink-client/src/test/java/org/apache/hudi/client/model/TestHoodieFlinkRecord.java
index 9753628da410..969bca214400 100644
---
a/hudi-client/hudi-flink-client/src/test/java/org/apache/hudi/client/model/TestHoodieFlinkRecord.java
+++
b/hudi-client/hudi-flink-client/src/test/java/org/apache/hudi/client/model/TestHoodieFlinkRecord.java
@@ -26,6 +26,7 @@ import org.apache.hudi.common.schema.HoodieSchemaField;
import org.apache.hudi.common.schema.HoodieSchemaType;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.common.util.OrderingValues;
+import org.apache.hudi.table.format.FlinkRecordContext;
import org.apache.flink.table.data.DecimalData;
import org.apache.flink.table.data.GenericRowData;
@@ -38,6 +39,7 @@ import java.math.BigDecimal;
import java.time.Instant;
import java.time.LocalDate;
import java.util.Arrays;
+import java.util.Collections;
import java.util.Properties;
import static org.junit.jupiter.api.Assertions.assertEquals;
@@ -47,6 +49,11 @@ import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertSame;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoInteractions;
+import static org.mockito.Mockito.verifyNoMoreInteractions;
+import static org.mockito.Mockito.when;
/**
* Unit tests for {@link HoodieFlinkRecord}.
@@ -192,6 +199,36 @@ public class TestHoodieFlinkRecord {
assertNotEquals(OrderingValues.getDefault(), orderingValue);
}
+ @Test
+ public void testGetRecordKeyCachesGeneratedKey() {
+ HoodieFlinkRecord record = new
HoodieFlinkRecord(GenericRowData.of(StringData.fromString("id1")));
+ HoodieSchema schema = HoodieSchema.createRecord("test", null, null,
Collections.singletonList(
+ HoodieSchemaField.of("id",
HoodieSchema.create(HoodieSchemaType.STRING), null, null)));
+ FlinkRecordContext recordContext = mock(FlinkRecordContext.class);
+ when(recordContext.getRecordKey(record.getData(),
schema)).thenReturn("id1");
+
+ assertNull(record.getKey());
+ assertEquals("id1", record.getRecordKey(schema, recordContext));
+ assertEquals("id1", record.getRecordKey(schema, recordContext));
+ verify(recordContext).getRecordKey(record.getData(), schema);
+ verifyNoMoreInteractions(recordContext);
+ assertEquals("id1", record.getRecordKey());
+ assertSame(record.getKey(), record.newInstance().getKey());
+ }
+
+ @Test
+ public void testGetRecordKeyPreservesExistingKey() {
+ HoodieKey key = new HoodieKey("id1", "partition1");
+ HoodieFlinkRecord record = new HoodieFlinkRecord(key,
HoodieOperation.INSERT, new GenericRowData(0));
+
+ HoodieSchema schema = HoodieSchema.createRecord("test", null, null,
Collections.emptyList());
+ FlinkRecordContext recordContext = mock(FlinkRecordContext.class);
+ assertEquals("id1", record.getRecordKey(schema, recordContext));
+ verifyNoInteractions(recordContext);
+ assertSame(key, record.getKey());
+ assertEquals("partition1", record.getPartitionPath());
+ }
+
@Test
public void testGetRecordKeyWithMetadataField() {
HoodieSchema schema = HoodieSchema.createRecord("test", null, null,
Arrays.asList(
diff --git
a/hudi-client/hudi-flink-client/src/test/java/org/apache/hudi/io/storage/row/TestHoodieRowDataCreateHandle.java
b/hudi-client/hudi-flink-client/src/test/java/org/apache/hudi/io/storage/row/TestHoodieRowDataCreateHandle.java
index b8d29dfb2c34..48a7b4b66a62 100644
---
a/hudi-client/hudi-flink-client/src/test/java/org/apache/hudi/io/storage/row/TestHoodieRowDataCreateHandle.java
+++
b/hudi-client/hudi-flink-client/src/test/java/org/apache/hudi/io/storage/row/TestHoodieRowDataCreateHandle.java
@@ -19,23 +19,35 @@
package org.apache.hudi.io.storage.row;
import org.apache.hudi.client.WriteStatus;
+import org.apache.hudi.client.model.HoodieRowDataCreation;
+import org.apache.hudi.common.engine.EngineType;
import org.apache.hudi.common.model.HoodiePayloadProps;
+import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.common.model.MetaFieldsMode;
+import org.apache.hudi.common.schema.HoodieSchema;
import org.apache.hudi.common.schema.HoodieSchemaUtils;
+import org.apache.hudi.common.util.ParquetUtils;
import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.storage.StoragePath;
import org.apache.hudi.table.HoodieFlinkTable;
import org.apache.hudi.testutils.HoodieFlinkClientTestHarness;
import org.apache.hudi.util.HoodieSchemaConverter;
+import org.apache.avro.generic.GenericRecord;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.data.GenericRowData;
+import org.apache.flink.table.data.RowData;
import org.apache.flink.table.data.StringData;
import org.apache.flink.table.types.DataType;
import org.apache.flink.table.types.logical.RowType;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.EnumSource;
import java.io.IOException;
+import java.util.Objects;
import java.util.Properties;
import static org.junit.jupiter.api.Assertions.assertEquals;
@@ -67,6 +79,55 @@ public class TestHoodieRowDataCreateHandle extends
HoodieFlinkClientTestHarness
cleanupResources();
}
+ @ParameterizedTest
+ @EnumSource(MetaFieldsMode.class)
+ void testMetaFieldsMode(MetaFieldsMode mode) throws Exception {
+ RowType rowType = (RowType) DataTypes.ROW(DataTypes.FIELD("id",
DataTypes.STRING())).notNull().getLogicalType();
+ HoodieSchema schema = HoodieSchemaConverter.convertToSchema(rowType);
+ HoodieWriteConfig config = HoodieWriteConfig.newBuilder()
+ .withPath(basePath)
+ .withEngineType(EngineType.FLINK)
+ .withMetaFieldsMode(mode)
+ .withEmbeddedTimelineServerEnabled(false)
+ .withSchema(schema.toString())
+ .build();
+ HoodieFlinkTable<?> table = HoodieFlinkTable.create(config, context,
metaClient);
+ for (boolean preserveMetadata : new boolean[] {false, true}) {
+ HoodieRowDataCreateHandle handle = new HoodieRowDataCreateHandle(
+ table, config, PARTITION_PATH, FILE_ID + preserveMetadata,
INSTANT_TIME,
+ TASK_PARTITION_ID, TASK_ID, TASK_EPOCH_ID,
+ HoodieSchemaUtils.addMetadataFields(schema, false),
preserveMetadata, false);
+ RowData row = GenericRowData.of(StringData.fromString("id1"));
+ if (preserveMetadata) {
+ row = HoodieRowDataCreation.create(
+ mode.isCommitTimePopulated() ? "old-instant" : null,
+ mode == MetaFieldsMode.ALL ? "old-sequence" : null,
+ mode.isRecordKeyPopulated() ? "id1" : null,
+ mode == MetaFieldsMode.ALL ? PARTITION_PATH : null,
+ mode.isFileNamePopulated() ? "old-file" : null,
+ row, false, false);
+ }
+ handle.write("id1", PARTITION_PATH, row);
+ WriteStatus status = handle.close();
+ assertEquals(0, status.getTotalErrorRecords());
+ StoragePath file = new StoragePath(basePath, status.getStat().getPath());
+ GenericRecord stored = new
ParquetUtils().readAvroRecords(metaClient.getStorage(), file).get(0);
+ assertEquals("id1", stored.get("id").toString());
+ assertEquals(mode.isCommitTimePopulated() ? (preserveMetadata ?
"old-instant" : INSTANT_TIME) : null,
+
Objects.toString(stored.get(HoodieRecord.COMMIT_TIME_METADATA_FIELD), null));
+ assertEquals(mode != MetaFieldsMode.ALL,
stored.get(HoodieRecord.COMMIT_SEQNO_METADATA_FIELD) == null);
+ if (preserveMetadata && mode == MetaFieldsMode.ALL) {
+ assertEquals("old-sequence",
stored.get(HoodieRecord.COMMIT_SEQNO_METADATA_FIELD).toString());
+ }
+ assertEquals(mode.isRecordKeyPopulated() ? "id1" : null,
+ Objects.toString(stored.get(HoodieRecord.RECORD_KEY_METADATA_FIELD),
null));
+ assertEquals(mode == MetaFieldsMode.ALL ? PARTITION_PATH : null,
+
Objects.toString(stored.get(HoodieRecord.PARTITION_PATH_METADATA_FIELD), null));
+ assertEquals(mode.isFileNamePopulated() ? file.getName() : null,
+ Objects.toString(stored.get(HoodieRecord.FILENAME_METADATA_FIELD),
null));
+ }
+ }
+
@Test
public void testEventTimeFieldIndexWithDoubleType() throws Exception {
// Schema with DOUBLE event_time field
diff --git
a/hudi-client/hudi-flink-client/src/test/java/org/apache/hudi/io/storage/row/TestHoodieRowDataLanceWriter.java
b/hudi-client/hudi-flink-client/src/test/java/org/apache/hudi/io/storage/row/TestHoodieRowDataLanceWriter.java
index b7303a61e93f..f66f3aaf66bd 100644
---
a/hudi-client/hudi-flink-client/src/test/java/org/apache/hudi/io/storage/row/TestHoodieRowDataLanceWriter.java
+++
b/hudi-client/hudi-flink-client/src/test/java/org/apache/hudi/io/storage/row/TestHoodieRowDataLanceWriter.java
@@ -18,8 +18,13 @@
package org.apache.hudi.io.storage.row;
+import org.apache.hudi.client.model.HoodieRowDataCreation;
+import org.apache.hudi.common.engine.LocalTaskContextSupplier;
import org.apache.hudi.common.engine.TaskContextSupplier;
+import org.apache.hudi.common.model.HoodieKey;
+import org.apache.hudi.common.model.MetaFieldsMode;
import org.apache.hudi.common.schema.HoodieSchema;
+import org.apache.hudi.common.schema.HoodieSchemaUtils;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.storage.StoragePath;
import org.apache.hudi.util.HoodieSchemaConverter;
@@ -33,6 +38,7 @@ import org.apache.arrow.vector.ipc.ArrowReader;
import org.apache.arrow.vector.types.pojo.ArrowType;
import org.apache.flink.table.data.GenericArrayData;
import org.apache.flink.table.data.GenericRowData;
+import org.apache.flink.table.data.RowData;
import org.apache.flink.table.types.logical.ArrayType;
import org.apache.flink.table.types.logical.FloatType;
import org.apache.flink.table.types.logical.IntType;
@@ -40,6 +46,8 @@ import org.apache.flink.table.types.logical.LogicalType;
import org.apache.flink.table.types.logical.RowType;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.EnumSource;
import org.lance.file.LanceFileReader;
import java.nio.file.Path;
@@ -56,6 +64,37 @@ public class TestHoodieRowDataLanceWriter {
@TempDir
Path tempDir;
+ @ParameterizedTest
+ @EnumSource(MetaFieldsMode.class)
+ void testMetaFieldsMode(MetaFieldsMode mode) throws Exception {
+ HoodieSchema schema =
HoodieSchemaUtils.addMetadataFields(HoodieSchemaConverter.convertToSchema(
+ RowType.of(new LogicalType[] {new IntType(false)}, new String[]
{"id"}).copy(false)), false);
+ StoragePath path = new
StoragePath(tempDir.resolve("metadata.lance").toUri());
+ try (HoodieRowDataLanceWriter writer = new HoodieRowDataLanceWriter(path,
schema, "001",
+ new LocalTaskContextSupplier(), Option.empty(), 128 * 1024 * 1024L,
+ 64 * 1024 * 1024L, 16 * 1024 * 1024L, true, mode, false)) {
+ RowData row = HoodieRowDataCreation.create(null, null, null, null, null,
+ GenericRowData.of(1), false, false);
+ writer.writeRowWithMetaData(new HoodieKey("key1", "partition"), row);
+ }
+ try (BufferAllocator allocator = new RootAllocator();
+ LanceFileReader reader = LanceFileReader.open(path.toString(),
allocator);
+ ArrowReader arrowReader = reader.readAll(null, null,
Integer.MAX_VALUE)) {
+ assertTrue(arrowReader.loadNextBatch());
+ VectorSchemaRoot root = arrowReader.getVectorSchemaRoot();
+ boolean[] populated = {mode.isCommitTimePopulated(), mode ==
MetaFieldsMode.ALL,
+ mode.isRecordKeyPopulated(), mode == MetaFieldsMode.ALL,
mode.isFileNamePopulated()};
+ String[] expected = {"001", null, "key1", "partition", path.getName()};
+ for (int i = 0; i < populated.length; i++) {
+ assertEquals(!populated[i], root.getVector(i).isNull(0));
+ if (populated[i] && expected[i] != null) {
+ assertEquals(expected[i], root.getVector(i).getObject(0).toString());
+ }
+ }
+ assertEquals(1, root.getVector("id").getObject(0));
+ }
+ }
+
@Test
public void testWritesVectorDataAndFooterMetadata() throws Exception {
RowType rowType = RowType.of(
@@ -75,7 +114,7 @@ public class TestHoodieRowDataLanceWriter {
64 * 1024 * 1024L,
16 * 1024 * 1024L,
true,
- false,
+ MetaFieldsMode.NONE,
false)) {
writer.writeRow("key1", GenericRowData.of(
1, new GenericArrayData(new Object[] {1.25F, 2.5F})));
diff --git
a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/TestBaseSparkInternalRowReaderContext.java
b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/TestBaseSparkInternalRowReaderContext.java
index 7edbdc3bf054..7792beafeaa7 100644
---
a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/TestBaseSparkInternalRowReaderContext.java
+++
b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/TestBaseSparkInternalRowReaderContext.java
@@ -70,7 +70,7 @@ class TestBaseSparkInternalRowReaderContext {
void setUp() {
storageconfig = mock(StorageConfiguration.class);
tableConfig = mock(HoodieTableConfig.class);
- when(tableConfig.populateMetaFields()).thenReturn(true);
+ when(tableConfig.isRecordKeyPopulated()).thenReturn(true);
when(tableConfig.getBaseFileFormat()).thenReturn(HoodieFileFormat.PARQUET);
when(tableConfig.getRecordKeyFields()).thenReturn(Option.of(new
String[]{"id"}));
diff --git
a/hudi-common/src/main/java/org/apache/hudi/common/engine/RecordContext.java
b/hudi-common/src/main/java/org/apache/hudi/common/engine/RecordContext.java
index 01f01a861c18..6a95e70a7b0e 100644
--- a/hudi-common/src/main/java/org/apache/hudi/common/engine/RecordContext.java
+++ b/hudi-common/src/main/java/org/apache/hudi/common/engine/RecordContext.java
@@ -74,8 +74,9 @@ public abstract class RecordContext<T> implements
Serializable {
protected RecordContext(HoodieTableConfig tableConfig, JavaTypeConverter
typeConverter) {
this.typeConverter = typeConverter;
- this.recordKeyExtractor = tableConfig.populateMetaFields() ?
metadataKeyExtractor() : virtualKeyExtractor(tableConfig.getRecordKeyFields()
- .orElseThrow(() -> new IllegalArgumentException("No record keys
specified and meta fields are not populated")));
+ this.recordKeyExtractor = tableConfig.isRecordKeyPopulated() ?
metadataKeyExtractor() : virtualKeyExtractor(tableConfig.getRecordKeyFields()
+ .orElseThrow(() -> new IllegalArgumentException("No record keys
specified and meta fields are not populated")),
+ tableConfig.getPartitionFields().map(fields ->
fields.length).orElse(0));
}
/**
@@ -440,11 +441,8 @@ public abstract class RecordContext<T> implements
Serializable {
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) {
return (record, schema) -> {
Object result = getValue(record, schema, recordKeyFields[0]);
if (result == null) {
diff --git
a/hudi-common/src/main/java/org/apache/hudi/common/table/HoodieTableConfig.java
b/hudi-common/src/main/java/org/apache/hudi/common/table/HoodieTableConfig.java
index 63ead8bcc087..f77960cd96a6 100644
---
a/hudi-common/src/main/java/org/apache/hudi/common/table/HoodieTableConfig.java
+++
b/hudi-common/src/main/java/org/apache/hudi/common/table/HoodieTableConfig.java
@@ -356,7 +356,8 @@ public class HoodieTableConfig extends HoodieConfig {
.key("hoodie.meta.fields.mode")
.noDefaultValue()
.withDocumentation("Which Hudi meta columns are physically populated on
disk. Allowed values are "
- + "ALL, NONE, COMMIT_TIME_ONLY, FILE_NAME_ONLY and
COMMIT_TIME_AND_FILE_NAME. This supersedes the "
+ + "ALL, NONE, COMMIT_TIME_ONLY, FILE_NAME_ONLY and
COMMIT_TIME_AND_FILE_NAME. "
+ + "Selective modes are supported by Spark and Flink COPY_ON_WRITE
writers. This supersedes the "
+ "deprecated hoodie.populate.meta.fields boolean, which is
consulted only when this property is unset "
+ "(true maps to ALL, false maps to NONE). Supported on any table
version 1.x can write, not just the "
+ "latest. Set only at table creation, via the hudi-cli, or during
table upgrade — the property is "
diff --git
a/hudi-common/src/test/java/org/apache/hudi/common/avro/TestAvroRecordContext.java
b/hudi-common/src/test/java/org/apache/hudi/common/avro/TestAvroRecordContext.java
index d1835dfbf7ae..6972b93fc735 100644
---
a/hudi-common/src/test/java/org/apache/hudi/common/avro/TestAvroRecordContext.java
+++
b/hudi-common/src/test/java/org/apache/hudi/common/avro/TestAvroRecordContext.java
@@ -19,6 +19,10 @@
package org.apache.hudi.common.avro;
+import org.apache.hudi.common.model.MetaFieldsMode;
+import org.apache.hudi.common.schema.HoodieSchema;
+import org.apache.hudi.common.table.HoodieTableConfig;
+
import org.apache.avro.Schema;
import org.apache.avro.generic.GenericData;
import org.apache.avro.generic.GenericRecord;
@@ -83,6 +87,30 @@ class TestAvroRecordContext {
return record;
}
+ @ParameterizedTest
+ @MethodSource("virtualRecordKeyParams")
+ void testVirtualRecordKey(String recordKeyFields, String partitionFields,
String expectedKey) {
+ HoodieTableConfig tableConfig = new HoodieTableConfig();
+ tableConfig.setValue(HoodieTableConfig.META_FIELDS_MODE,
MetaFieldsMode.NONE.name());
+ tableConfig.setValue(HoodieTableConfig.RECORDKEY_FIELDS, recordKeyFields);
+ if (partitionFields != null) {
+ tableConfig.setValue(HoodieTableConfig.PARTITION_FIELDS,
partitionFields);
+ }
+ AvroRecordContext context = new AvroRecordContext(tableConfig, null);
+ assertEquals(expectedKey, context.getRecordKey(buildRecord(),
HoodieSchema.fromAvroSchema(RECORD_SCHEMA)));
+ }
+
+ private static Stream<Arguments> virtualRecordKeyParams() {
+ return Stream.of(
+ Arguments.of("id", null, "1"),
+ Arguments.of("id", "", "1"),
+ Arguments.of("id", "name", "1"),
+ Arguments.of("id", "name,address.city", "id:1"),
+ Arguments.of("id,name", null, "id:1,name:alice"),
+ Arguments.of("id,name", "name", "id:1,name:alice"),
+ Arguments.of("id,name", "name,address.city", "id:1,name:alice"));
+ }
+
@Test
void testGetFieldValueTopLevel() {
GenericRecord record = buildRecord();
diff --git
a/hudi-common/src/test/java/org/apache/hudi/common/avro/TestHoodieAvroReaderContext.java
b/hudi-common/src/test/java/org/apache/hudi/common/avro/TestHoodieAvroReaderContext.java
index 8b37dcd5465d..927158d890dd 100644
---
a/hudi-common/src/test/java/org/apache/hudi/common/avro/TestHoodieAvroReaderContext.java
+++
b/hudi-common/src/test/java/org/apache/hudi/common/avro/TestHoodieAvroReaderContext.java
@@ -163,7 +163,7 @@ class TestHoodieAvroReaderContext {
@Test
void getNestedField() {
- when(tableConfig.populateMetaFields()).thenReturn(true);
+ when(tableConfig.isRecordKeyPopulated()).thenReturn(true);
HoodieAvroReaderContext avroReaderContext =
new HoodieAvroReaderContext(storageConfig, tableConfig,
Option.empty(), Option.empty());
IndexedRecord indexedRecord = createBaseRecord("compound", "field2", 3.2);
@@ -172,7 +172,8 @@ class TestHoodieAvroReaderContext {
@Test
void getRecordKeyWithSingleKey() {
- when(tableConfig.populateMetaFields()).thenReturn(false);
+ when(tableConfig.isRecordKeyPopulated()).thenReturn(false);
+ when(tableConfig.getPartitionFields()).thenReturn(Option.empty());
when(tableConfig.getRecordKeyFields()).thenReturn(Option.of(new
String[]{"skeleton_field_1"}));
HoodieAvroReaderContext avroReaderContext =
new HoodieAvroReaderContext(storageConfig, tableConfig,
Option.empty(), Option.empty());
@@ -183,7 +184,8 @@ class TestHoodieAvroReaderContext {
@Test
void getRecordKeyWithMultipleKeys() {
- when(tableConfig.populateMetaFields()).thenReturn(false);
+ when(tableConfig.isRecordKeyPopulated()).thenReturn(false);
+ when(tableConfig.getPartitionFields()).thenReturn(Option.empty());
when(tableConfig.getRecordKeyFields()).thenReturn(Option.of(new
String[]{"base_field_1", "base_field_3.nested_field"}));
HoodieAvroReaderContext avroReaderContext =
new HoodieAvroReaderContext(storageConfig, tableConfig,
Option.empty(), Option.empty());
@@ -249,7 +251,7 @@ class TestHoodieAvroReaderContext {
}
private HoodieAvroReaderContext getReaderContextWithMetaFields() {
- when(tableConfig.populateMetaFields()).thenReturn(true);
+ when(tableConfig.isRecordKeyPopulated()).thenReturn(true);
return new HoodieAvroReaderContext(storageConfig, tableConfig,
Option.empty(), Option.empty());
}
diff --git
a/hudi-common/src/test/java/org/apache/hudi/common/table/read/SchemaHandlerTestBase.java
b/hudi-common/src/test/java/org/apache/hudi/common/table/read/SchemaHandlerTestBase.java
index 047c08cdd873..9f6754d688c6 100644
---
a/hudi-common/src/test/java/org/apache/hudi/common/table/read/SchemaHandlerTestBase.java
+++
b/hudi-common/src/test/java/org/apache/hudi/common/table/read/SchemaHandlerTestBase.java
@@ -278,6 +278,7 @@ public abstract class SchemaHandlerTestBase {
static void setupMORTable(RecordMergeMode mergeMode, boolean hasPrecombine,
HoodieTableConfig hoodieTableConfig) {
when(hoodieTableConfig.populateMetaFields()).thenReturn(true);
+ when(hoodieTableConfig.isRecordKeyPopulated()).thenReturn(true);
when(hoodieTableConfig.getRecordMergeMode()).thenReturn(mergeMode);
when(hoodieTableConfig.getTableVersion()).thenReturn(HoodieTableVersion.current());
if (hasPrecombine) {
diff --git
a/hudi-common/src/test/java/org/apache/hudi/common/table/read/TestFileGroupReaderSchemaHandler.java
b/hudi-common/src/test/java/org/apache/hudi/common/table/read/TestFileGroupReaderSchemaHandler.java
index 8e9026cd10f7..1f997412d734 100644
---
a/hudi-common/src/test/java/org/apache/hudi/common/table/read/TestFileGroupReaderSchemaHandler.java
+++
b/hudi-common/src/test/java/org/apache/hudi/common/table/read/TestFileGroupReaderSchemaHandler.java
@@ -74,6 +74,7 @@ public class TestFileGroupReaderSchemaHandler extends
SchemaHandlerTestBase {
@Test
public void testCow() {
when(hoodieTableConfig.populateMetaFields()).thenReturn(true);
+ when(hoodieTableConfig.isRecordKeyPopulated()).thenReturn(true);
HoodieReaderContext<String> readerContext =
createReaderContext(hoodieTableConfig, false, false, false, false, null);
HoodieSchema requestedSchema = DATA_SCHEMA;
FileGroupReaderSchemaHandler schemaHandler =
createSchemaHandler(readerContext, DATA_SCHEMA, requestedSchema, false);
@@ -89,6 +90,7 @@ public class TestFileGroupReaderSchemaHandler extends
SchemaHandlerTestBase {
@Test
public void testCowBootstrap() {
when(hoodieTableConfig.populateMetaFields()).thenReturn(true);
+ when(hoodieTableConfig.isRecordKeyPopulated()).thenReturn(true);
HoodieReaderContext<String> readerContext =
createReaderContext(hoodieTableConfig, false, false, true, false, null);
HoodieSchema requestedSchema = generateProjectionSchema("begin_lat",
"tip_history", "_hoodie_record_key", "rider");
@@ -105,6 +107,7 @@ public class TestFileGroupReaderSchemaHandler extends
SchemaHandlerTestBase {
@Test
void testGetRequiredSchemaForFileAndRenameColumns() {
when(hoodieTableConfig.populateMetaFields()).thenReturn(true);
+ when(hoodieTableConfig.isRecordKeyPopulated()).thenReturn(true);
HoodieReaderContext<String> readerContext =
createReaderContext(hoodieTableConfig, false, false, true, false, null);
HoodieSchema requestedSchema =
generateProjectionSchema("_hoodie_record_key", "timestamp", "rider");
@@ -244,6 +247,7 @@ public class TestFileGroupReaderSchemaHandler extends
SchemaHandlerTestBase {
when(hoodieTableConfig.getRecordMergeMode()).thenReturn(mergeMode);
when(hoodieTableConfig.populateMetaFields()).thenReturn(true);
+ when(hoodieTableConfig.isRecordKeyPopulated()).thenReturn(true);
when(hoodieTableConfig.getOrderingFieldsStr()).thenReturn(Option.of(setPrecombine
? preCombineField : StringUtils.EMPTY_STRING));
when(hoodieTableConfig.getOrderingFields()).thenReturn(setPrecombine ?
Collections.singletonList(preCombineField) : Collections.emptyList());
when(hoodieTableConfig.getTableVersion()).thenReturn(tableVersion);
@@ -343,6 +347,7 @@ public class TestFileGroupReaderSchemaHandler extends
SchemaHandlerTestBase {
when(hoodieTableConfig.getPayloadClass()).thenReturn(payloadClass);
when(hoodieTableConfig.getRecordMergeStrategyId()).thenReturn(null);
when(hoodieTableConfig.populateMetaFields()).thenReturn(true);
+ when(hoodieTableConfig.isRecordKeyPopulated()).thenReturn(true);
if (orderingField != null) {
when(hoodieTableConfig.getOrderingFieldsStr()).thenReturn(Option.of(orderingField));
when(hoodieTableConfig.getOrderingFields()).thenReturn(Collections.singletonList(orderingField));
diff --git
a/hudi-common/src/test/java/org/apache/hudi/common/table/read/TestParquetRowIndexBasedSchemaHandler.java
b/hudi-common/src/test/java/org/apache/hudi/common/table/read/TestParquetRowIndexBasedSchemaHandler.java
index f6cd3426290e..49537dc0e919 100644
---
a/hudi-common/src/test/java/org/apache/hudi/common/table/read/TestParquetRowIndexBasedSchemaHandler.java
+++
b/hudi-common/src/test/java/org/apache/hudi/common/table/read/TestParquetRowIndexBasedSchemaHandler.java
@@ -48,6 +48,7 @@ public class TestParquetRowIndexBasedSchemaHandler extends
SchemaHandlerTestBase
@Test
public void testCowBootstrapWithPositionMerge() {
when(hoodieTableConfig.populateMetaFields()).thenReturn(true);
+ when(hoodieTableConfig.isRecordKeyPopulated()).thenReturn(true);
HoodieReaderContext<String> readerContext =
createReaderContext(hoodieTableConfig, true, false, true, false, null);
HoodieSchema requestedSchema = generateProjectionSchema("begin_lat",
"tip_history", "_hoodie_record_key", "rider");
FileGroupReaderSchemaHandler schemaHandler =
createSchemaHandler(readerContext, DATA_SCHEMA, requestedSchema, true);
diff --git
a/hudi-common/src/test/java/org/apache/hudi/common/table/read/buffer/TestFileGroupRecordBufferLoader.java
b/hudi-common/src/test/java/org/apache/hudi/common/table/read/buffer/TestFileGroupRecordBufferLoader.java
index 1ed45d7e7bda..f57fe11adca5 100644
---
a/hudi-common/src/test/java/org/apache/hudi/common/table/read/buffer/TestFileGroupRecordBufferLoader.java
+++
b/hudi-common/src/test/java/org/apache/hudi/common/table/read/buffer/TestFileGroupRecordBufferLoader.java
@@ -67,6 +67,7 @@ public class TestFileGroupRecordBufferLoader extends
BaseTestFileGroupRecordBuff
when(tableConfig.getPartialUpdateMode()).thenReturn(Option.empty());
when(tableConfig.getOrderingFieldsStr()).thenReturn(Option.empty());
when(tableConfig.getRecordKeyFields()).thenReturn(Option.of(new String[]
{"record_key"}));
+ when(tableConfig.getPartitionFields()).thenReturn(Option.empty());
StorageConfiguration<?> storageConfiguration =
mock(StorageConfiguration.class);
HoodieReaderContext<IndexedRecord> readerContext = new
HoodieAvroReaderContext(storageConfiguration, tableConfig, Option.empty(),
Option.empty());
readerContext.initRecordMerger(new TypedProperties());
diff --git
a/hudi-common/src/test/java/org/apache/hudi/common/table/read/buffer/TestKeyBasedFileGroupRecordBuffer.java
b/hudi-common/src/test/java/org/apache/hudi/common/table/read/buffer/TestKeyBasedFileGroupRecordBuffer.java
index b2138b9a495c..a5dbe4c9d86f 100644
---
a/hudi-common/src/test/java/org/apache/hudi/common/table/read/buffer/TestKeyBasedFileGroupRecordBuffer.java
+++
b/hudi-common/src/test/java/org/apache/hudi/common/table/read/buffer/TestKeyBasedFileGroupRecordBuffer.java
@@ -110,6 +110,7 @@ class TestKeyBasedFileGroupRecordBuffer extends
BaseTestFileGroupRecordBuffer {
HoodieReadStats readStats = new HoodieReadStats();
HoodieTableConfig tableConfig = mock(HoodieTableConfig.class);
when(tableConfig.getRecordKeyFields()).thenReturn(Option.of(new String[]
{"record_key"}));
+ when(tableConfig.getPartitionFields()).thenReturn(Option.empty());
StorageConfiguration<?> storageConfiguration =
mock(StorageConfiguration.class);
HoodieReaderContext<IndexedRecord> readerContext = new
HoodieAvroReaderContext(storageConfiguration, tableConfig, Option.empty(),
Option.empty());
KeyBasedFileGroupRecordBuffer<IndexedRecord> fileGroupRecordBuffer =
buildKeyBasedFileGroupRecordBuffer(readerContext, tableConfig, readStats, null,
@@ -137,6 +138,7 @@ class TestKeyBasedFileGroupRecordBuffer extends
BaseTestFileGroupRecordBuffer {
HoodieReadStats readStats = new HoodieReadStats();
HoodieTableConfig tableConfig = mock(HoodieTableConfig.class);
when(tableConfig.getRecordKeyFields()).thenReturn(Option.of(new String[]
{"record_key"}));
+ when(tableConfig.getPartitionFields()).thenReturn(Option.empty());
StorageConfiguration<?> storageConfiguration =
mock(StorageConfiguration.class);
HoodieReaderContext<IndexedRecord> readerContext = new
HoodieAvroReaderContext(storageConfiguration, tableConfig, Option.empty(),
Option.empty());
KeyBasedFileGroupRecordBuffer<IndexedRecord> fileGroupRecordBuffer =
buildKeyBasedFileGroupRecordBuffer(readerContext, tableConfig, readStats, null,
@@ -178,6 +180,7 @@ class TestKeyBasedFileGroupRecordBuffer extends
BaseTestFileGroupRecordBuffer {
when(tableConfig.getPartialUpdateMode()).thenReturn(Option.empty());
when(tableConfig.getTableVersion()).thenReturn(HoodieTableVersion.current());
when(tableConfig.getRecordKeyFields()).thenReturn(Option.of(new String[]
{"record_key"}));
+ when(tableConfig.getPartitionFields()).thenReturn(Option.empty());
StorageConfiguration<?> storageConfiguration =
mock(StorageConfiguration.class);
HoodieReaderContext<IndexedRecord> readerContext = new
HoodieAvroReaderContext(storageConfiguration, tableConfig, Option.empty(),
Option.empty());
readerContext.setHasLogFiles(false);
@@ -210,6 +213,7 @@ class TestKeyBasedFileGroupRecordBuffer extends
BaseTestFileGroupRecordBuffer {
HoodieReadStats readStats = new HoodieReadStats();
HoodieTableConfig tableConfig = mock(HoodieTableConfig.class);
when(tableConfig.getRecordKeyFields()).thenReturn(Option.of(new String[]
{"record_key"}));
+ when(tableConfig.getPartitionFields()).thenReturn(Option.empty());
StorageConfiguration<?> storageConfiguration =
mock(StorageConfiguration.class);
HoodieReaderContext<IndexedRecord> readerContext = new
HoodieAvroReaderContext(storageConfiguration, tableConfig, Option.empty(),
Option.empty());
KeyBasedFileGroupRecordBuffer<IndexedRecord> fileGroupRecordBuffer =
buildKeyBasedFileGroupRecordBuffer(readerContext, tableConfig, readStats, null,
@@ -246,6 +250,7 @@ class TestKeyBasedFileGroupRecordBuffer extends
BaseTestFileGroupRecordBuffer {
when(tableConfig.getPartialUpdateMode()).thenReturn(Option.empty());
when(tableConfig.getTableVersion()).thenReturn(HoodieTableVersion.current());
when(tableConfig.getRecordKeyFields()).thenReturn(Option.of(new String[]
{"record_key"}));
+ when(tableConfig.getPartitionFields()).thenReturn(Option.empty());
StorageConfiguration<?> storageConfiguration =
mock(StorageConfiguration.class);
HoodieReaderContext<IndexedRecord> readerContext = new
HoodieAvroReaderContext(storageConfiguration, tableConfig, Option.empty(),
Option.empty());
readerContext.setHasLogFiles(false);
@@ -277,6 +282,7 @@ class TestKeyBasedFileGroupRecordBuffer extends
BaseTestFileGroupRecordBuffer {
HoodieTableConfig tableConfig = mock(HoodieTableConfig.class);
when(tableConfig.getPayloadClass()).thenReturn(CustomPayload.class.getName());
when(tableConfig.getRecordKeyFields()).thenReturn(Option.of(new String[]
{"record_key"}));
+ when(tableConfig.getPartitionFields()).thenReturn(Option.empty());
when(tableConfig.getRecordMergeMode()).thenReturn(RecordMergeMode.CUSTOM);
when(tableConfig.getPartialUpdateMode()).thenReturn(Option.empty());
StorageConfiguration<?> storageConfiguration =
mock(StorageConfiguration.class);
@@ -322,6 +328,7 @@ class TestKeyBasedFileGroupRecordBuffer extends
BaseTestFileGroupRecordBuffer {
HoodieTableConfig tableConfig = mock(HoodieTableConfig.class);
when(tableConfig.getPayloadClass()).thenReturn(TestKeyBasedFileGroupRecordBuffer.CustomPayload.class.getName());
when(tableConfig.getRecordKeyFields()).thenReturn(Option.of(new String[]
{"record_key"}));
+ when(tableConfig.getPartitionFields()).thenReturn(Option.empty());
when(tableConfig.getRecordMergeMode()).thenReturn(RecordMergeMode.CUSTOM);
when(tableConfig.getPartialUpdateMode()).thenReturn(Option.empty());
when(tableConfig.getRecordMergeStrategyId()).thenReturn(HoodieRecordMerger.PAYLOAD_BASED_MERGE_STRATEGY_UUID);
@@ -355,6 +362,7 @@ class TestKeyBasedFileGroupRecordBuffer extends
BaseTestFileGroupRecordBuffer {
HoodieTableConfig tableConfig = mock(HoodieTableConfig.class);
when(tableConfig.getPayloadClass()).thenReturn(CustomPayload.class.getName());
when(tableConfig.getRecordKeyFields()).thenReturn(Option.of(new String[]
{"record_key"}));
+ when(tableConfig.getPartitionFields()).thenReturn(Option.empty());
StorageConfiguration<?> storageConfiguration =
mock(StorageConfiguration.class);
HoodieReaderContext<IndexedRecord> readerContext = new
HoodieAvroReaderContext(storageConfiguration, tableConfig, Option.empty(),
Option.empty());
KeyBasedFileGroupRecordBuffer<IndexedRecord> fileGroupRecordBuffer =
buildKeyBasedFileGroupRecordBuffer(readerContext, tableConfig, readStats, new
CustomMerger(),
@@ -396,6 +404,7 @@ class TestKeyBasedFileGroupRecordBuffer extends
BaseTestFileGroupRecordBuffer {
HoodieTableConfig tableConfig = mock(HoodieTableConfig.class);
when(tableConfig.getPayloadClass()).thenReturn(CustomPayload.class.getName());
when(tableConfig.getRecordKeyFields()).thenReturn(Option.of(new String[]
{"record_key"}));
+ when(tableConfig.getPartitionFields()).thenReturn(Option.empty());
when(tableConfig.getRecordMergeMode()).thenReturn(RecordMergeMode.CUSTOM);
when(tableConfig.getPartialUpdateMode()).thenReturn(Option.empty());
when(tableConfig.getRecordMergeStrategyId()).thenReturn(HoodieRecordMerger.PAYLOAD_BASED_MERGE_STRATEGY_UUID);
diff --git
a/hudi-common/src/test/java/org/apache/hudi/common/table/read/buffer/TestSortedKeyBasedFileGroupRecordBuffer.java
b/hudi-common/src/test/java/org/apache/hudi/common/table/read/buffer/TestSortedKeyBasedFileGroupRecordBuffer.java
index 4dde1b53c37c..fbde79157951 100644
---
a/hudi-common/src/test/java/org/apache/hudi/common/table/read/buffer/TestSortedKeyBasedFileGroupRecordBuffer.java
+++
b/hudi-common/src/test/java/org/apache/hudi/common/table/read/buffer/TestSortedKeyBasedFileGroupRecordBuffer.java
@@ -119,6 +119,7 @@ class TestSortedKeyBasedFileGroupRecordBuffer extends
BaseTestFileGroupRecordBuf
properties.setProperty(DELETE_MARKER, "3");
HoodieTableConfig tableConfig = mock(HoodieTableConfig.class);
when(tableConfig.getRecordKeyFields()).thenReturn(Option.of(new String[]
{"record_key"}));
+ when(tableConfig.getPartitionFields()).thenReturn(Option.empty());
when(tableConfig.getTableVersion()).thenReturn(HoodieTableVersion.current());
when(tableConfig.getRecordMergeMode()).thenReturn(RecordMergeMode.EVENT_TIME_ORDERING);
StorageConfiguration<?> storageConfiguration =
mock(StorageConfiguration.class);
diff --git
a/hudi-common/src/test/java/org/apache/hudi/common/table/read/buffer/TestStreamingKeyBasedFileGroupRecordBuffer.java
b/hudi-common/src/test/java/org/apache/hudi/common/table/read/buffer/TestStreamingKeyBasedFileGroupRecordBuffer.java
index 1a650771e46a..d03be553194a 100644
---
a/hudi-common/src/test/java/org/apache/hudi/common/table/read/buffer/TestStreamingKeyBasedFileGroupRecordBuffer.java
+++
b/hudi-common/src/test/java/org/apache/hudi/common/table/read/buffer/TestStreamingKeyBasedFileGroupRecordBuffer.java
@@ -78,6 +78,7 @@ class TestStreamingKeyBasedFileGroupRecordBuffer extends
BaseTestFileGroupRecord
when(tableConfig.getPartialUpdateMode()).thenReturn(Option.empty());
when(tableConfig.getTableVersion()).thenReturn(HoodieTableVersion.current());
when(tableConfig.getRecordKeyFields()).thenReturn(Option.of(new String[]
{"record_key"}));
+ when(tableConfig.getPartitionFields()).thenReturn(Option.empty());
StorageConfiguration<?> storageConfiguration =
mock(StorageConfiguration.class);
HoodieReaderContext<IndexedRecord> readerContext = new
HoodieAvroReaderContext(storageConfiguration, tableConfig, Option.empty(),
Option.empty());
readerContext.setHasLogFiles(false);
@@ -116,6 +117,7 @@ class TestStreamingKeyBasedFileGroupRecordBuffer extends
BaseTestFileGroupRecord
when(tableConfig.getPartialUpdateMode()).thenReturn(Option.empty());
when(tableConfig.getTableVersion()).thenReturn(HoodieTableVersion.current());
when(tableConfig.getRecordKeyFields()).thenReturn(Option.of(new String[]
{"record_key"}));
+ when(tableConfig.getPartitionFields()).thenReturn(Option.empty());
StorageConfiguration<?> storageConfiguration =
mock(StorageConfiguration.class);
HoodieReaderContext<IndexedRecord> readerContext = new
HoodieAvroReaderContext(storageConfiguration, tableConfig, Option.empty(),
Option.empty());
readerContext.setHasLogFiles(false);
@@ -152,6 +154,7 @@ class TestStreamingKeyBasedFileGroupRecordBuffer extends
BaseTestFileGroupRecord
HoodieTableConfig tableConfig = mock(HoodieTableConfig.class);
when(tableConfig.getPayloadClass()).thenReturn(CustomPayload.class.getName());
when(tableConfig.getRecordKeyFields()).thenReturn(Option.of(new String[]
{"record_key"}));
+ when(tableConfig.getPartitionFields()).thenReturn(Option.empty());
when(tableConfig.getRecordMergeMode()).thenReturn(RecordMergeMode.CUSTOM);
when(tableConfig.getPartialUpdateMode()).thenReturn(Option.empty());
when(tableConfig.getRecordMergeStrategyId()).thenReturn(HoodieRecordMerger.PAYLOAD_BASED_MERGE_STRATEGY_UUID);
@@ -189,6 +192,7 @@ class TestStreamingKeyBasedFileGroupRecordBuffer extends
BaseTestFileGroupRecord
HoodieTableConfig tableConfig = mock(HoodieTableConfig.class);
when(tableConfig.getPayloadClass()).thenReturn(CustomPayload.class.getName());
when(tableConfig.getRecordKeyFields()).thenReturn(Option.of(new String[]
{"record_key"}));
+ when(tableConfig.getPartitionFields()).thenReturn(Option.empty());
when(tableConfig.getRecordMergeMode()).thenReturn(RecordMergeMode.CUSTOM);
when(tableConfig.getPartialUpdateMode()).thenReturn(Option.empty());
when(tableConfig.getRecordMergeStrategyId()).thenReturn(HoodieRecordMerger.PAYLOAD_BASED_MERGE_STRATEGY_UUID);
diff --git
a/hudi-common/src/test/java/org/apache/hudi/metadata/TestHoodieBackedTableMetadataDataCleanup.java
b/hudi-common/src/test/java/org/apache/hudi/metadata/TestHoodieBackedTableMetadataDataCleanup.java
index c6c97c66bb7e..27a49bf897f7 100644
---
a/hudi-common/src/test/java/org/apache/hudi/metadata/TestHoodieBackedTableMetadataDataCleanup.java
+++
b/hudi-common/src/test/java/org/apache/hudi/metadata/TestHoodieBackedTableMetadataDataCleanup.java
@@ -602,7 +602,7 @@ public class TestHoodieBackedTableMetadataDataCleanup {
when(timeline.filterCompletedInstants()).thenReturn(timeline);
when(timeline.lastInstant()).thenReturn(Option.empty());
HoodieTableConfig tableConfig = mock(HoodieTableConfig.class);
- when(tableConfig.populateMetaFields()).thenReturn(true);
+ when(tableConfig.isRecordKeyPopulated()).thenReturn(true);
when(tableConfig.getBaseFileFormat()).thenReturn(HoodieFileFormat.PARQUET);
when(metadataMetaClient.getTableConfig()).thenReturn(tableConfig);
diff --git
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/configuration/OptionsResolver.java
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/configuration/OptionsResolver.java
index f848b3ae4343..a0e1c0db4e3a 100644
---
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/configuration/OptionsResolver.java
+++
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/configuration/OptionsResolver.java
@@ -26,6 +26,7 @@ import org.apache.hudi.common.config.HoodieCommonConfig;
import org.apache.hudi.common.config.HoodieStorageConfig;
import org.apache.hudi.common.model.DefaultHoodieRecordPayload;
import org.apache.hudi.common.model.HoodieFailedWritesCleaningPolicy;
+import org.apache.hudi.common.model.MetaFieldsMode;
import org.apache.hudi.common.model.WriteConcurrencyMode;
import org.apache.hudi.common.model.WriteOperationType;
import org.apache.hudi.common.table.HoodieTableConfig;
@@ -565,10 +566,14 @@ public class OptionsResolver {
* 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());
}
/**
diff --git
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/bulk/BulkInsertWriterHelper.java
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/bulk/BulkInsertWriterHelper.java
index 367f86085613..4fa933d75db6 100644
---
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/bulk/BulkInsertWriterHelper.java
+++
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/bulk/BulkInsertWriterHelper.java
@@ -20,6 +20,7 @@ package org.apache.hudi.sink.bulk;
import org.apache.hudi.client.WriteStatus;
import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.common.model.MetaFieldsMode;
import org.apache.hudi.common.schema.HoodieSchema;
import org.apache.hudi.common.schema.HoodieSchemaUtils;
import org.apache.hudi.common.util.Option;
@@ -71,6 +72,7 @@ public class BulkInsertWriterHelper implements AutoCloseable {
// table schema handed to the write handles.
protected final HoodieSchema writerSchema;
protected final boolean preserveHoodieMetadata;
+ private final boolean preserveRecordKey;
protected final boolean isAppendMode;
// used for Append mode only, if true then only initial row data without
metacolumns is written
protected final boolean populateMetaFields;
@@ -106,7 +108,8 @@ public class BulkInsertWriterHelper implements
AutoCloseable {
this.totalSubtaskNum = totalSubtaskNum;
this.taskEpochId = taskEpochId;
this.isAppendMode = OptionsResolver.isAppendMode(conf);
- this.populateMetaFields = writeConfig.populateMetaFields();
+ MetaFieldsMode metaFieldsMode = writeConfig.getMetaFieldsMode();
+ this.populateMetaFields = metaFieldsMode != MetaFieldsMode.NONE;
HoodieSchema schema = HoodieSchemaConverter.convertToSchema(
rowType,
HoodieSchemaUtils.getRecordQualifiedName(conf.get(FlinkOptions.TABLE_NAME)),
@@ -115,20 +118,21 @@ public class BulkInsertWriterHelper implements
AutoCloseable {
? schema
: HoodieSchemaUtils.addMetadataFields(schema,
writeConfig.allowOperationMetadataField());
this.preserveHoodieMetadata = preserveHoodieMetadata;
+ this.preserveRecordKey = preserveHoodieMetadata &&
metaFieldsMode.isRecordKeyPopulated();
this.isInputSorted = OptionsResolver.isBulkInsertOperation(conf)
&& (conf.get(FlinkOptions.WRITE_BULK_INSERT_SORT_INPUT)
|| OptionsResolver.isLsmTreeStorageLayout(conf));
this.fileIdPrefix = UUID.randomUUID().toString();
- this.keyGen = preserveHoodieMetadata ? null :
RowDataKeyGens.instance(conf, rowType, taskPartitionId, instantTime);
+ this.keyGen = preserveRecordKey ? null : RowDataKeyGens.instance(conf,
rowType, taskPartitionId, instantTime);
this.writeMetrics = writeMetrics;
}
public void write(RowData record) throws IOException {
try {
- String recordKey = preserveHoodieMetadata
+ String recordKey = preserveRecordKey
? record.getString(HoodieRecord.RECORD_KEY_META_FIELD_ORD).toString()
: keyGen.getRecordKey(record);
- String partitionPath = preserveHoodieMetadata
+ String partitionPath = preserveRecordKey // only ALL mode populates the
partition path meta field
?
record.getString(HoodieRecord.PARTITION_PATH_META_FIELD_ORD).toString()
: keyGen.getPartitionPath(record);
writeRecord(recordKey, partitionPath, record);
diff --git
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/util/StreamerUtil.java
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/util/StreamerUtil.java
index 4da31daddece..470099723090 100644
---
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/util/StreamerUtil.java
+++
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/util/StreamerUtil.java
@@ -361,7 +361,7 @@ public class StreamerUtil {
.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))
.initTable(HadoopFSUtils.getStorageConfWithCopy(hadoopConf),
basePath);
log.info("Table initialized under base path {}", basePath);
} else {
diff --git
a/hudi-client/hudi-flink-client/src/test/java/org/apache/hudi/client/TestFlinkWriteClientFunctional.java
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/client/TestFlinkWriteClientFunctional.java
similarity index 79%
rename from
hudi-client/hudi-flink-client/src/test/java/org/apache/hudi/client/TestFlinkWriteClientFunctional.java
rename to
hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/client/TestFlinkWriteClientFunctional.java
index 4d5e2670db14..9a127f96e5f8 100644
---
a/hudi-client/hudi-flink-client/src/test/java/org/apache/hudi/client/TestFlinkWriteClientFunctional.java
+++
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/client/TestFlinkWriteClientFunctional.java
@@ -22,8 +22,10 @@ package org.apache.hudi.client;
import org.apache.hudi.avro.model.HoodieClusteringPlan;
import
org.apache.hudi.client.clustering.plan.strategy.FlinkSizeBasedClusteringPlanStrategyRecently;
import org.apache.hudi.client.common.HoodieFlinkEngineContext;
+import org.apache.hudi.client.model.EventTimeFlinkRecordMerger;
import org.apache.hudi.client.model.HoodieFlinkRecord;
import org.apache.hudi.common.config.HoodieMetadataConfig;
+import org.apache.hudi.common.config.HoodieStorageConfig;
import org.apache.hudi.common.engine.EngineType;
import org.apache.hudi.common.fs.FSUtils;
import org.apache.hudi.common.model.HoodieKey;
@@ -31,6 +33,7 @@ import org.apache.hudi.common.model.HoodieOperation;
import org.apache.hudi.common.model.HoodieRecord;
import org.apache.hudi.common.model.HoodieRecordLocation;
import org.apache.hudi.common.model.HoodieTableType;
+import org.apache.hudi.common.model.MetaFieldsMode;
import org.apache.hudi.common.model.WriteOperationType;
import org.apache.hudi.common.table.HoodieTableConfig;
import org.apache.hudi.common.table.HoodieTableMetaClient;
@@ -38,6 +41,7 @@ import
org.apache.hudi.common.table.cdc.HoodieCDCSupplementalLoggingMode;
import org.apache.hudi.common.table.timeline.HoodieInstant;
import org.apache.hudi.common.util.ClusteringUtils;
import org.apache.hudi.common.util.Option;
+import org.apache.hudi.common.util.ParquetUtils;
import org.apache.hudi.config.HoodieClusteringConfig;
import org.apache.hudi.config.HoodieWriteConfig;
import org.apache.hudi.io.FlinkCreateHandle;
@@ -49,8 +53,10 @@ import org.apache.hudi.table.HoodieFlinkTable;
import org.apache.hudi.table.action.HoodieWriteMetadata;
import org.apache.hudi.table.action.commit.BucketInfo;
import org.apache.hudi.table.action.commit.BucketType;
+import org.apache.hudi.table.format.HoodieFlinkIOFactory;
import org.apache.hudi.testutils.HoodieFlinkClientTestHarness;
+import org.apache.avro.generic.GenericRecord;
import org.apache.flink.table.data.GenericRowData;
import org.apache.flink.table.data.StringData;
import org.junit.jupiter.api.AfterEach;
@@ -58,6 +64,7 @@ import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.EnumSource;
import org.junit.jupiter.params.provider.MethodSource;
import org.mockito.MockedStatic;
import org.mockito.Mockito;
@@ -69,6 +76,7 @@ import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
+import java.util.Objects;
import java.util.Properties;
import java.util.function.Supplier;
import java.util.stream.Stream;
@@ -90,8 +98,7 @@ import static org.mockito.Mockito.when;
* Functional coverage for the Flink client write boundary.
*
* <p>The datasource bucket assigner hands this client records that already
carry a target file group.
- * These tests construct that same input directly so client and handle
behavior can be exercised without
- * depending on the datasource module.
+ * These tests construct that same input directly and use the datasource's
Flink readers for base files.
*/
class TestFlinkWriteClientFunctional extends HoodieFlinkClientTestHarness {
@@ -112,6 +119,8 @@ class TestFlinkWriteClientFunctional extends
HoodieFlinkClientTestHarness {
void setUp() {
initPath();
initFileSystem();
+ storageConf.set(HoodieStorageConfig.HOODIE_IO_FACTORY_CLASS.key(),
HoodieFlinkIOFactory.class.getName());
+ context = new HoodieFlinkEngineContext(storageConf.unwrap());
}
@AfterEach
@@ -178,9 +187,89 @@ class TestFlinkWriteClientFunctional extends
HoodieFlinkClientTestHarness {
}
}
+ @ParameterizedTest
+ @EnumSource(MetaFieldsMode.class)
+ void testCopyOnWriteMetaFieldsMode(MetaFieldsMode mode) throws IOException {
+ initWriteClient(HoodieTableType.COPY_ON_WRITE, false, false, mode);
+ String insertInstant = writeClient.startCommit();
+ transitionToInflight(insertInstant);
+ List<WriteStatus> statuses = writeClient.insert(Arrays.asList(
+ insertRecord("id1", "one", 1L), insertRecord("id2", "two", 2L)),
insertInstant);
+ assertWriteStatuses(statuses, 2);
+ statuses =
writeClient.insert(Collections.singletonList(insertRecord("id3", "three", 3L)),
insertInstant);
+ assertWriteStatuses(statuses, 3);
+ assertTrue(writeClient.commit(insertInstant, statuses));
+ assertMetaFields(statuses, mode, insertInstant, insertInstant);
+
+ writeClient.cleanHandles();
+ String updateInstant = writeClient.startCommit();
+ transitionToInflight(updateInstant);
+ statuses = writeClient.upsert(Arrays.asList(
+ updateRecord("id1", "updated", 4L, insertInstant),
+ deleteRecord("id2", 5L, insertInstant)), updateInstant);
+ assertWriteStatuses(statuses, 2);
+ assertTrue(writeClient.commit(updateInstant, statuses));
+ assertMetaFields(statuses, mode, insertInstant, updateInstant);
+ }
+
+ @ParameterizedTest
+ @EnumSource(MetaFieldsMode.class)
+ void testCopyOnWriteConcatMetaFieldsMode(MetaFieldsMode mode) throws
IOException {
+ initWriteClient(HoodieTableType.COPY_ON_WRITE, false, false, mode);
+
writeConfig.setValue(HoodieWriteConfig.MERGE_ALLOW_DUPLICATE_ON_INSERTS_ENABLE,
"true");
+ String insertInstant = writeClient.startCommit();
+ transitionToInflight(insertInstant);
+ writeClient.insert(Collections.singletonList(insertRecord("id1", "one",
1L)), insertInstant);
+ List<WriteStatus> statuses = writeClient.insert(
+ Collections.singletonList(insertRecord("id1", "one", 1L)),
insertInstant);
+ assertWriteStatuses(statuses, 2);
+ assertTrue(writeClient.commit(insertInstant, statuses));
+ assertMetaFields(statuses, mode, insertInstant, insertInstant);
+
+ writeClient.cleanHandles();
+ String nextInstant = writeClient.startCommit();
+ transitionToInflight(nextInstant);
+ // Target the existing file group while keeping insert semantics.
+ statuses = writeClient.insert(Collections.singletonList(
+ updateRecord("id3", "three", 3L, insertInstant)), nextInstant);
+ assertWriteStatuses(statuses, 3);
+ assertTrue(writeClient.commit(nextInstant, statuses));
+ for (WriteStatus status : statuses) {
+ StoragePath path = new StoragePath(basePath, status.getStat().getPath());
+ for (GenericRecord row : new
ParquetUtils().readAvroRecords(metaClient.getStorage(), path)) {
+ assertEquals(mode.isFileNamePopulated(),
row.get(HoodieRecord.FILENAME_METADATA_FIELD) != null);
+ String expectedInstant = row.get("id").toString().equals("id3") ?
nextInstant : insertInstant;
+ assertEquals(mode.isCommitTimePopulated() ? expectedInstant : null,
+ Objects.toString(row.get(HoodieRecord.COMMIT_TIME_METADATA_FIELD),
null));
+ }
+ }
+ }
+
+ private void assertMetaFields(List<WriteStatus> statuses, MetaFieldsMode
mode,
+ String insertInstant, String updateInstant) {
+ for (WriteStatus status : statuses) {
+ StoragePath path = new StoragePath(basePath, status.getStat().getPath());
+ for (GenericRecord row : new
ParquetUtils().readAvroRecords(metaClient.getStorage(), path)) {
+ String id = row.get("id").toString();
+ String expectedInstant = id.equals("id1") ? updateInstant :
insertInstant;
+ assertEquals(mode.isCommitTimePopulated() ? expectedInstant : null,
+ Objects.toString(row.get(HoodieRecord.COMMIT_TIME_METADATA_FIELD),
null));
+ assertEquals(mode != MetaFieldsMode.ALL,
row.get(HoodieRecord.COMMIT_SEQNO_METADATA_FIELD) == null);
+ assertEquals(mode.isRecordKeyPopulated() ? id : null,
+ Objects.toString(row.get(HoodieRecord.RECORD_KEY_METADATA_FIELD),
null));
+ assertEquals(mode == MetaFieldsMode.ALL ? PARTITION_PATH : null,
+
Objects.toString(row.get(HoodieRecord.PARTITION_PATH_METADATA_FIELD), null));
+ assertEquals(mode.isFileNamePopulated(),
row.get(HoodieRecord.FILENAME_METADATA_FIELD) != null);
+ if (id.equals("id1")) {
+ assertEquals(insertInstant.equals(updateInstant) ? "one" :
"updated", row.get("name").toString());
+ }
+ }
+ }
+ }
+
@Test
void testCopyOnWriteCleansRetryFiles() throws IOException {
- context = new HoodieFlinkEngineContext(
+ context = new HoodieFlinkEngineContext(storageConf,
new HoodieFlinkEngineContext.DefaultTaskContextSupplier() {
@Override
public Supplier<Long> getAttemptIdSupplier() {
@@ -408,7 +497,13 @@ class TestFlinkWriteClientFunctional extends
HoodieFlinkClientTestHarness {
private void initWriteClient(
HoodieTableType tableType, boolean cdcEnabled, boolean
useRecentClusteringStrategy)
throws IOException {
+ initWriteClient(tableType, cdcEnabled, useRecentClusteringStrategy,
MetaFieldsMode.ALL);
+ }
+
+ private void initWriteClient(HoodieTableType tableType, boolean cdcEnabled,
+ boolean useRecentClusteringStrategy,
MetaFieldsMode mode) throws IOException {
Properties tableProperties = new Properties();
+ tableProperties.setProperty(HoodieTableConfig.META_FIELDS_MODE.key(),
mode.name());
tableProperties.setProperty(HoodieTableConfig.CDC_ENABLED.key(),
Boolean.toString(cdcEnabled));
tableProperties.setProperty(
HoodieTableConfig.CDC_SUPPLEMENTAL_LOGGING_MODE.key(),
@@ -427,6 +522,8 @@ class TestFlinkWriteClientFunctional extends
HoodieFlinkClientTestHarness {
HoodieWriteConfig.Builder builder = HoodieWriteConfig.newBuilder()
.withPath(basePath)
.withEngineType(EngineType.FLINK)
+ .withRecordMergeImplClasses(EventTimeFlinkRecordMerger.class.getName())
+
.withRecordMergeStrategyId(EventTimeFlinkRecordMerger.EVENT_TIME_BASED_MERGE_STRATEGY_UUID)
.withSchema(SCHEMA)
.withProperties(tableProperties)
.withMetadataConfig(HoodieMetadataConfig.newBuilder().enable(false).build())
diff --git
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/configuration/TestOptionsResolver.java
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/configuration/TestOptionsResolver.java
index 15ff8110dcbb..965e947ab73c 100644
---
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/configuration/TestOptionsResolver.java
+++
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/configuration/TestOptionsResolver.java
@@ -25,6 +25,7 @@ import org.apache.hudi.common.config.HoodieMetadataConfig;
import org.apache.hudi.common.model.DefaultHoodieRecordPayload;
import org.apache.hudi.common.model.HoodieFailedWritesCleaningPolicy;
import org.apache.hudi.common.model.HoodieTableType;
+import org.apache.hudi.common.model.MetaFieldsMode;
import org.apache.hudi.common.model.WriteConcurrencyMode;
import org.apache.hudi.common.model.WriteOperationType;
import org.apache.hudi.common.table.HoodieTableConfig;
@@ -60,6 +61,17 @@ import static org.junit.jupiter.api.Assertions.assertTrue;
*/
public class TestOptionsResolver {
+ @Test
+ void testMetaFieldsModeLegacyFallback() {
+ Configuration conf = new Configuration();
+ assertEquals(MetaFieldsMode.ALL, OptionsResolver.getMetaFieldsMode(conf));
+ conf.setString(HoodieTableConfig.POPULATE_META_FIELDS.key(), "false");
+ assertEquals(MetaFieldsMode.NONE, OptionsResolver.getMetaFieldsMode(conf));
+ conf.setString(HoodieTableConfig.META_FIELDS_MODE.key(),
"commit_time_only");
+ assertEquals(MetaFieldsMode.COMMIT_TIME_ONLY,
OptionsResolver.getMetaFieldsMode(conf));
+ assertFalse(OptionsResolver.isPopulateMetaFields(conf));
+ }
+
@Test
void testTableStorageLayoutDefaultsByOperation() {
Configuration conf = new Configuration();
diff --git
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/TestWriteCopyOnWrite.java
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/TestWriteCopyOnWrite.java
index 008494fade54..a9e38d0d8e48 100644
---
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/TestWriteCopyOnWrite.java
+++
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/TestWriteCopyOnWrite.java
@@ -24,9 +24,12 @@ import org.apache.hudi.common.config.HoodieMetadataConfig;
import org.apache.hudi.common.data.HoodieListData;
import org.apache.hudi.common.model.HoodieFailedWritesCleaningPolicy;
import org.apache.hudi.common.model.HoodieKey;
+import org.apache.hudi.common.model.HoodieRecord;
import org.apache.hudi.common.model.HoodieRecordGlobalLocation;
import org.apache.hudi.common.model.HoodieTableType;
+import org.apache.hudi.common.model.MetaFieldsMode;
import org.apache.hudi.common.model.WriteConcurrencyMode;
+import org.apache.hudi.common.table.HoodieTableConfig;
import org.apache.hudi.common.table.HoodieTableMetaClient;
import org.apache.hudi.common.table.marker.MarkerType;
import org.apache.hudi.common.table.timeline.HoodieTimeline;
@@ -66,10 +69,12 @@ import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.stream.Collectors;
import java.util.stream.Stream;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertSame;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -340,6 +345,73 @@ public class TestWriteCopyOnWrite extends TestWriteBase {
.end();
}
+ @ParameterizedTest
+ @MethodSource("mergeWithMetaFieldsModeParams")
+ public void testMergeWithMetaFieldsMode(MetaFieldsMode mode, String
storageLayout, String mergeHandleClass) throws Exception {
+ conf.setString(HoodieTableConfig.TABLE_STORAGE_LAYOUT.key(),
storageLayout);
+ conf.setString(HoodieTableConfig.META_FIELDS_MODE.key(), mode.name());
+ conf.setString(HoodieWriteConfig.MERGE_HANDLE_CLASS_NAME.key(),
mergeHandleClass);
+ preparePipeline(conf)
+ .consume(TestData.DATA_SET_INSERT.subList(0, 2))
+ .checkpoint(1)
+ .assertNextEvent()
+ .checkpointComplete(1)
+ .consume(TestData.DATA_SET_UPDATE_INSERT.subList(0, 1))
+ .checkpoint(2)
+ .assertNextEvent()
+ .checkpointComplete(2)
+ .end();
+ TestData.checkWrittenData(tempFile, Collections.singletonMap("par1",
"[id1:24, id2:33]"), 1,
+ record -> record.get("uuid") + ":" + record.get("age"));
+ }
+
+ private static Stream<Arguments> mergeWithMetaFieldsModeParams() {
+ return Arrays.stream(MetaFieldsMode.values()).flatMap(mode -> Stream.of(
+ Arguments.of(mode, "DEFAULT", HoodieWriteMergeHandle.class.getName()),
+ Arguments.of(mode, "DEFAULT",
FileGroupReaderBasedMergeHandle.class.getName()),
+ Arguments.of(mode, "LSM_TREE",
FileGroupReaderBasedMergeHandle.class.getName())));
+ }
+
+ @ParameterizedTest
+ @MethodSource("mergeWithMultiplePartitionFieldsParams")
+ public void testMergeWithMultiplePartitionFieldsAndMetaFieldsMode(
+ MetaFieldsMode mode, String storageLayout, String mergeHandleClass)
throws Exception {
+ conf.set(FlinkOptions.OPERATION, "upsert");
+ conf.set(FlinkOptions.RECORD_KEY_FIELD, "uuid");
+ conf.set(FlinkOptions.PARTITION_PATH_FIELD, "partition,name");
+ conf.setString(HoodieTableConfig.TABLE_STORAGE_LAYOUT.key(),
storageLayout);
+ conf.setString(HoodieTableConfig.META_FIELDS_MODE.key(), mode.name());
+ conf.setString(HoodieWriteConfig.MERGE_HANDLE_CLASS_NAME.key(),
mergeHandleClass);
+ conf.setString(HoodieWriteConfig.COMPLEX_KEYGEN_NEW_ENCODING.key(),
"false");
+ preparePipeline(conf)
+ .consume(TestData.DATA_SET_INSERT.subList(0, 2))
+ .checkpoint(1)
+ .assertNextEvent()
+ .checkpointComplete(1)
+ .consume(TestData.DATA_SET_UPDATE_INSERT.subList(0, 1))
+ .checkpoint(2)
+ .assertNextEvent()
+ .checkpointComplete(2)
+ .end();
+
+ Map<String, String> expected = new HashMap<>();
+ expected.put("par1/Danny", "[id1:24]");
+ expected.put("par1/Stephen", "[id2:33]");
+ // Both leaf partitions share the same top-level directory. The updated
key must occur only once.
+ TestData.checkWrittenData(tempFile, expected, 1, record -> {
+ if (mode == MetaFieldsMode.NONE) {
+
assertNull(record.getSchema().getField(HoodieRecord.RECORD_KEY_METADATA_FIELD));
+ } else {
+ assertNull(record.get(HoodieRecord.RECORD_KEY_METADATA_FIELD));
+ }
+ return record.get("uuid") + ":" + record.get("age");
+ });
+ }
+
+ private static Stream<Arguments> mergeWithMultiplePartitionFieldsParams() {
+ return mergeWithMetaFieldsModeParams().filter(args -> args.get()[0] !=
MetaFieldsMode.ALL);
+ }
+
@Test
public void testInsertDuplicates() throws Exception {
// reset the config option
@@ -503,9 +575,19 @@ public class TestWriteCopyOnWrite extends TestWriteBase {
* The test is almost same with {@link #testInsertWithSmallBufferSize}
except that
* it is with insert clustering mode.
*/
- @Test
- public void testInsertClustering() throws Exception {
+ @ParameterizedTest
+ @EnumSource(MetaFieldsMode.class)
+ public void testInsertClustering(MetaFieldsMode mode) throws Exception {
+ List<String> expected = Arrays.asList(
+ "id1,Danny,23,1970-01-01 00:00:00.0,par1",
+ "id1,Danny,23,1970-01-01 00:00:00.001,par1",
+ "id1,Danny,23,1970-01-01 00:00:00.002,par1",
+ "id1,Danny,23,1970-01-01 00:00:00.003,par1",
+ "id1,Danny,23,1970-01-01 00:00:00.004,par1");
+ List<String> expectedDuplicates = Stream.concat(expected.stream(),
expected.stream()).sorted().collect(Collectors.toList());
+
// reset the config option
+ conf.setString(HoodieTableConfig.META_FIELDS_MODE.key(), mode.name());
conf.set(FlinkOptions.OPERATION, "insert");
conf.set(FlinkOptions.INSERT_CLUSTER, true);
conf.set(FlinkOptions.WRITE_MEMORY_SEGMENT_PAGE_SIZE, 64);
@@ -521,13 +603,13 @@ public class TestWriteCopyOnWrite extends TestWriteBase {
.allDataFlushed()
.handleEvents(2)
.checkpointComplete(1)
- .checkWrittenData(EXPECTED4, 1)
- // insert duplicates again
+ .checkWrittenDataNoMeta(Collections.singletonMap("par1",
expected.toString()), 1)
+ // Insert duplicates again, exercising concat across checkpoints as
well as mini-batches.
.consume(TestData.DATA_SET_INSERT_SAME_KEY)
.checkpoint(2)
.handleEvents(2)
.checkpointComplete(2)
- .checkWrittenDataCOW(EXPECTED5)
+ .checkWrittenDataNoMeta(Collections.singletonMap("par1",
expectedDuplicates.toString()), 1)
.end();
}
diff --git
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/TestWriteMergeOnRead.java
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/TestWriteMergeOnRead.java
index d7572b4b4209..64cdd0405518 100644
---
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/TestWriteMergeOnRead.java
+++
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/TestWriteMergeOnRead.java
@@ -25,6 +25,7 @@ import org.apache.hudi.common.model.HoodieCommitMetadata;
import org.apache.hudi.common.model.HoodieKey;
import org.apache.hudi.common.model.HoodieTableType;
import org.apache.hudi.common.model.HoodieWriteStat;
+import org.apache.hudi.common.model.MetaFieldsMode;
import org.apache.hudi.common.table.HoodieTableMetaClient;
import org.apache.hudi.common.table.timeline.HoodieInstant;
import org.apache.hudi.config.HoodieClusteringConfig;
@@ -171,10 +172,21 @@ public class TestWriteMergeOnRead extends
TestWriteCopyOnWrite {
}
@Override
- public void testInsertClustering() {
+ public void testInsertClustering(MetaFieldsMode mode) {
// insert clustering is only valid for cow table.
}
+ @Override
+ public void testMergeWithMetaFieldsMode(MetaFieldsMode mode, String
storageLayout, String mergeHandleClass) {
+ // These tests exercise COW merge handles; MOR does not support selective
metadata modes.
+ }
+
+ @Override
+ public void testMergeWithMultiplePartitionFieldsAndMetaFieldsMode(
+ MetaFieldsMode mode, String storageLayout, String mergeHandleClass) {
+ // These tests exercise COW merge handles; MOR does not support selective
metadata modes.
+ }
+
@Test
public void testInsertAsyncClustering() {
// insert async clustering is only valid for cow table.
diff --git
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/TestWriteMergeOnReadWithCompact.java
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/TestWriteMergeOnReadWithCompact.java
index b71a4611da93..d1daaa312457 100644
---
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/TestWriteMergeOnReadWithCompact.java
+++
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/TestWriteMergeOnReadWithCompact.java
@@ -20,6 +20,7 @@ package org.apache.hudi.sink;
import org.apache.hudi.client.HoodieFlinkWriteClient;
import org.apache.hudi.common.model.HoodieTableType;
+import org.apache.hudi.common.model.MetaFieldsMode;
import org.apache.hudi.common.model.PartialUpdateAvroPayload;
import org.apache.hudi.common.model.WriteConcurrencyMode;
import org.apache.hudi.common.table.HoodieTableConfig;
@@ -99,10 +100,21 @@ public class TestWriteMergeOnReadWithCompact extends
TestWriteCopyOnWrite {
}
@Override
- public void testInsertClustering() {
+ public void testInsertClustering(MetaFieldsMode mode) {
// insert clustering is only valid for cow table.
}
+ @Override
+ public void testMergeWithMetaFieldsMode(MetaFieldsMode mode, String
storageLayout, String mergeHandleClass) {
+ // These tests exercise COW merge handles; MOR does not support selective
metadata modes.
+ }
+
+ @Override
+ public void testMergeWithMultiplePartitionFieldsAndMetaFieldsMode(
+ MetaFieldsMode mode, String storageLayout, String mergeHandleClass) {
+ // These tests exercise COW merge handles; MOR does not support selective
metadata modes.
+ }
+
@Test
public void testInsertAsyncClustering() {
// insert async clustering is only valid for cow table.
diff --git
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/bulk/TestBulkInsertWriteHelper.java
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/bulk/TestBulkInsertWriteHelper.java
index c79f33123769..da26b9d88b4c 100644
---
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/bulk/TestBulkInsertWriteHelper.java
+++
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/bulk/TestBulkInsertWriteHelper.java
@@ -20,7 +20,14 @@ package org.apache.hudi.sink.bulk;
import org.apache.hudi.client.WriteClientTestUtils;
import org.apache.hudi.client.WriteStatus;
+import org.apache.hudi.client.model.HoodieRowDataCreation;
+import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.common.model.MetaFieldsMode;
+import org.apache.hudi.common.table.HoodieTableConfig;
+import org.apache.hudi.common.util.ParquetUtils;
import org.apache.hudi.common.util.StringUtils;
+import org.apache.hudi.configuration.FlinkOptions;
+import org.apache.hudi.storage.StoragePath;
import org.apache.hudi.table.HoodieFlinkTable;
import org.apache.hudi.util.DataTypeUtils;
import org.apache.hudi.util.FlinkTables;
@@ -37,17 +44,25 @@ import org.apache.flink.table.types.logical.RowType;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
import java.io.File;
import java.io.IOException;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.Objects;
import java.util.stream.Collectors;
+import java.util.stream.Stream;
import static org.hamcrest.CoreMatchers.is;
import static org.hamcrest.MatcherAssert.assertThat;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertThrows;
/**
@@ -65,6 +80,69 @@ public class TestBulkInsertWriteHelper {
StreamerUtil.initTableIfNotExists(conf);
}
+ @ParameterizedTest
+ @MethodSource("metaFieldsModeParams")
+ void testMetaFieldsMode(MetaFieldsMode mode, boolean preserveMetadata)
throws Exception {
+ for (String operation : new String[] {"insert", "bulk_insert"}) {
+ File path = new File(tempFile, operation);
+ Configuration modeConf =
TestConfigurations.getDefaultConf(path.getAbsolutePath(),
TestConfigurations.ROW_DATA_TYPE);
+ modeConf.set(FlinkOptions.TABLE_TYPE, "COPY_ON_WRITE");
+ modeConf.set(FlinkOptions.OPERATION, operation);
+ modeConf.set(FlinkOptions.INSERT_CLUSTER, false);
+ modeConf.setString(HoodieTableConfig.TABLE_STORAGE_LAYOUT.key(),
"DEFAULT");
+ modeConf.setString(HoodieTableConfig.META_FIELDS_MODE.key(),
mode.name());
+ StreamerUtil.initTableIfNotExists(modeConf);
+ HoodieFlinkTable<?> table = FlinkTables.createTable(modeConf);
+ String instant = WriteClientTestUtils.createNewInstantTime();
+ String expectedCommitTime = preserveMetadata ? "old-instant" : instant;
+ RowType rowType = preserveMetadata ?
DataTypeUtils.addMetadataFields(TestConfigurations.ROW_TYPE, false) :
TestConfigurations.ROW_TYPE;
+ BulkInsertWriterHelper helper = new BulkInsertWriterHelper(modeConf,
table, table.getConfig(),
+ instant, 1, 1, 0, rowType, preserveMetadata);
+ for (RowData row : TestData.DATA_SET_INSERT) {
+ if (preserveMetadata) {
+ // Clustering reads rows with metadata columns, but selective modes
leave keys null.
+ row = HoodieRowDataCreation.create(mode.isCommitTimePopulated() ?
expectedCommitTime : null,
+ mode == MetaFieldsMode.ALL ? "old-sequence" : null,
+ mode.isRecordKeyPopulated() ? row.getString(0).toString() : null,
+ mode == MetaFieldsMode.ALL ? row.getString(4).toString() : null,
+ mode.isFileNamePopulated() ? "old.parquet" : null, row, false,
false);
+ }
+ helper.write(row);
+ }
+ List<WriteStatus> statuses = helper.getWriteStatuses(1);
+ assertWriteStatus(statuses);
+ assertEquals(TestData.DATA_SET_INSERT.size(),
statuses.stream().mapToLong(status -> status.getStat().getNumWrites()).sum());
+ for (WriteStatus status : statuses) {
+ assertFalse(status.hasErrors());
+ StoragePath file = new StoragePath(path.getAbsolutePath(),
status.getStat().getPath());
+ for (GenericRecord row : new
ParquetUtils().readAvroRecords(table.getStorage(), file)) {
+ if (mode == MetaFieldsMode.NONE && operation.equals("insert") &&
!preserveMetadata) {
+ assertEquals(TestConfigurations.ROW_TYPE.getFieldCount(),
row.getSchema().getFields().size());
+ for (String field :
HoodieRecord.HOODIE_META_COLUMNS_WITH_OPERATION) {
+ assertEquals(null, row.getSchema().getField(field), field);
+ }
+ continue;
+ }
+ assertEquals(row.get("partition").toString(),
status.getStat().getPartitionPath());
+ assertEquals(mode.isCommitTimePopulated() ? expectedCommitTime :
null,
+
Objects.toString(row.get(HoodieRecord.COMMIT_TIME_METADATA_FIELD), null));
+ assertEquals(mode != MetaFieldsMode.ALL,
row.get(HoodieRecord.COMMIT_SEQNO_METADATA_FIELD) == null);
+ assertEquals(mode.isRecordKeyPopulated() ?
row.get("uuid").toString() : null,
+
Objects.toString(row.get(HoodieRecord.RECORD_KEY_METADATA_FIELD), null));
+ assertEquals(mode == MetaFieldsMode.ALL ?
status.getStat().getPartitionPath() : null,
+
Objects.toString(row.get(HoodieRecord.PARTITION_PATH_METADATA_FIELD), null));
+ assertEquals(mode.isFileNamePopulated() ? file.getName() : null,
+ Objects.toString(row.get(HoodieRecord.FILENAME_METADATA_FIELD),
null));
+ }
+ }
+ }
+ }
+
+ private static Stream<Arguments> metaFieldsModeParams() {
+ return Arrays.stream(MetaFieldsMode.values()).flatMap(mode -> Stream.of(
+ Arguments.of(mode, false), Arguments.of(mode, true)));
+ }
+
@Test
void testWrite() throws Exception {
HoodieFlinkTable<?> table = FlinkTables.createTable(conf);
diff --git
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/utils/TestWriteBase.java
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/utils/TestWriteBase.java
index b051cad77a6e..623a99704951 100644
---
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/utils/TestWriteBase.java
+++
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/utils/TestWriteBase.java
@@ -620,6 +620,16 @@ public class TestWriteBase {
return this;
}
+ /**
+ * Checks base-file records without Hudi metadata fields.
+ */
+ public TestHarness checkWrittenDataNoMeta(
+ Map<String, String> expected,
+ int partitions) throws IOException {
+ TestData.checkWrittenData(this.baseFile, expected, partitions,
TestData::filterOutVariablesWithoutHudiMetadata);
+ return this;
+ }
+
public TestHarness checkWrittenAllData(Map<String, String> expected, int
partitions) throws IOException {
TestData.checkWrittenAllData(baseFile, expected, partitions);
return this;
diff --git
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/TestHoodieFileGroupReaderOnFlink.java
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/TestHoodieFileGroupReaderOnFlink.java
index 84e1136bc296..2d0258c17a16 100644
---
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/TestHoodieFileGroupReaderOnFlink.java
+++
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/TestHoodieFileGroupReaderOnFlink.java
@@ -211,7 +211,7 @@ public class TestHoodieFileGroupReaderOnFlink extends
TestHoodieFileGroupReaderB
@Test
public void testGetOrderingValue() {
HoodieTableConfig tableConfig = Mockito.mock(HoodieTableConfig.class);
- when(tableConfig.populateMetaFields()).thenReturn(true);
+ when(tableConfig.isRecordKeyPopulated()).thenReturn(true);
FlinkRowDataReaderContext readerContext =
new FlinkRowDataReaderContext(getStorageConf(), () ->
InternalSchemaManager.DISABLED, Collections.emptyList(), tableConfig,
Option.empty());
HoodieSchema schema = HoodieSchema.createRecord("test", null, null,
@@ -228,7 +228,7 @@ public class TestHoodieFileGroupReaderOnFlink extends
TestHoodieFileGroupReaderB
@Test
public void getRecordKeyFromMetadataFields() {
HoodieTableConfig tableConfig = Mockito.mock(HoodieTableConfig.class);
- when(tableConfig.populateMetaFields()).thenReturn(true);
+ when(tableConfig.isRecordKeyPopulated()).thenReturn(true);
FlinkRowDataReaderContext readerContext =
new FlinkRowDataReaderContext(getStorageConf(), () ->
InternalSchemaManager.DISABLED, Collections.emptyList(), tableConfig,
Option.empty());
HoodieSchema schema = HoodieSchema.createRecord("test", null, null,
@@ -244,7 +244,8 @@ public class TestHoodieFileGroupReaderOnFlink extends
TestHoodieFileGroupReaderB
@Test
public void getRecordKeySingleKey() {
HoodieTableConfig tableConfig = Mockito.mock(HoodieTableConfig.class);
- when(tableConfig.populateMetaFields()).thenReturn(false);
+ when(tableConfig.isRecordKeyPopulated()).thenReturn(false);
+ when(tableConfig.getPartitionFields()).thenReturn(Option.empty());
when(tableConfig.getRecordKeyFields()).thenReturn(Option.of(new String[]
{"field1"}));
FlinkRowDataReaderContext readerContext =
new FlinkRowDataReaderContext(getStorageConf(), () ->
InternalSchemaManager.DISABLED, Collections.emptyList(), tableConfig,
Option.empty());
@@ -261,7 +262,8 @@ public class TestHoodieFileGroupReaderOnFlink extends
TestHoodieFileGroupReaderB
@Test
public void getRecordKeyWithMultipleKeys() {
HoodieTableConfig tableConfig = Mockito.mock(HoodieTableConfig.class);
- when(tableConfig.populateMetaFields()).thenReturn(false);
+ when(tableConfig.isRecordKeyPopulated()).thenReturn(false);
+ when(tableConfig.getPartitionFields()).thenReturn(Option.empty());
when(tableConfig.getRecordKeyFields()).thenReturn(Option.of(new String[]
{"field1", "field2"}));
FlinkRowDataReaderContext readerContext =
new FlinkRowDataReaderContext(getStorageConf(), () ->
InternalSchemaManager.DISABLED, Collections.emptyList(), tableConfig,
Option.empty());
diff --git
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/format/TestFlinkRowDataReaderContext.java
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/format/TestFlinkRowDataReaderContext.java
index 751f63d6de75..f89b367441a4 100644
---
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/format/TestFlinkRowDataReaderContext.java
+++
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/format/TestFlinkRowDataReaderContext.java
@@ -62,7 +62,7 @@ class TestFlinkRowDataReaderContext {
@BeforeEach
void setUp() {
List<ExpressionPredicates.Predicate> predicates = new ArrayList<>();
- when(tableConfig.populateMetaFields()).thenReturn(true);
+ when(tableConfig.isRecordKeyPopulated()).thenReturn(true);
when(tableConfig.getBaseFileFormat()).thenReturn(HoodieFileFormat.PARQUET);
when(tableConfig.getRecordKeyFields()).thenReturn(Option.of(new
String[]{"id"}));
readerContext = new FlinkRowDataReaderContext(
diff --git
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/utils/TestStreamerUtil.java
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/utils/TestStreamerUtil.java
index becc48b64786..f6fd78802e0d 100644
---
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/utils/TestStreamerUtil.java
+++
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/utils/TestStreamerUtil.java
@@ -28,6 +28,7 @@ import org.apache.hudi.common.config.HoodieStorageConfig;
import org.apache.hudi.common.config.RecordMergeMode;
import org.apache.hudi.common.config.TypedProperties;
import org.apache.hudi.common.model.EventTimeAvroPayload;
+import org.apache.hudi.common.model.MetaFieldsMode;
import org.apache.hudi.common.model.OverwriteWithLatestAvroPayload;
import org.apache.hudi.common.model.PartialUpdateAvroPayload;
import org.apache.hudi.common.model.WriteOperationType;
@@ -62,6 +63,8 @@ import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.EnumSource;
import org.mockito.Mockito;
import java.io.File;
@@ -92,6 +95,20 @@ class TestStreamerUtil {
@TempDir
File tempFile;
+ @ParameterizedTest
+ @EnumSource(MetaFieldsMode.class)
+ void testInitTableMetaFieldsMode(MetaFieldsMode mode) throws IOException {
+ Configuration conf =
TestConfigurations.getDefaultConf(tempFile.getAbsolutePath());
+ conf.set(FlinkOptions.TABLE_TYPE, "COPY_ON_WRITE");
+ conf.set(FlinkOptions.WRITE_TABLE_VERSION,
HoodieTableVersion.TEN.versionCode());
+ conf.setString(HoodieTableConfig.META_FIELDS_MODE.key(), mode.name());
+ // The explicit mode must win over the legacy default.
+ conf.setString(HoodieTableConfig.POPULATE_META_FIELDS.key(), "true");
+ HoodieTableMetaClient metaClient = StreamerUtil.initTableIfNotExists(conf);
+ assertEquals(mode, metaClient.getTableConfig().getMetaFieldsMode());
+ assertEquals(mode.toLegacyPopulateMetaFields(),
metaClient.getTableConfig().populateMetaFields());
+ }
+
@Test
void testMetadataConfigIncludesMetadataTableBloomFilterSettings() {
Configuration conf =
TestConfigurations.getDefaultConf(tempFile.getAbsolutePath());
diff --git
a/hudi-hadoop-mr/src/test/java/org/apache/hudi/hadoop/TestHiveHoodieReaderContext.java
b/hudi-hadoop-mr/src/test/java/org/apache/hudi/hadoop/TestHiveHoodieReaderContext.java
index 9b01becfd81c..9a484c6edf10 100644
---
a/hudi-hadoop-mr/src/test/java/org/apache/hudi/hadoop/TestHiveHoodieReaderContext.java
+++
b/hudi-hadoop-mr/src/test/java/org/apache/hudi/hadoop/TestHiveHoodieReaderContext.java
@@ -81,7 +81,8 @@ class TestHiveHoodieReaderContext {
@Test
void getRecordKeyWithSingleKey() {
- when(tableConfig.populateMetaFields()).thenReturn(false);
+ when(tableConfig.isRecordKeyPopulated()).thenReturn(false);
+ when(tableConfig.getPartitionFields()).thenReturn(Option.empty());
when(tableConfig.getRecordKeyFields()).thenReturn(Option.of(new
String[]{"field_1"}));
HiveHoodieReaderContext avroReaderContext = new
HiveHoodieReaderContext(readerCreator, Collections.emptyList(),
storageConfiguration, tableConfig);
ArrayWritable row = new ArrayWritable(Writable.class, new Writable[]{new
Text("value1"), new Text("value2"), new ArrayWritable(new String[]{"value3"})});
@@ -91,7 +92,8 @@ class TestHiveHoodieReaderContext {
@Test
void getRecordKeyWithMultipleKeys() {
- when(tableConfig.populateMetaFields()).thenReturn(false);
+ when(tableConfig.isRecordKeyPopulated()).thenReturn(false);
+ when(tableConfig.getPartitionFields()).thenReturn(Option.empty());
when(tableConfig.getRecordKeyFields()).thenReturn(Option.of(new
String[]{"field_1", "field_3.nested_field"}));
HiveHoodieReaderContext avroReaderContext = new
HiveHoodieReaderContext(readerCreator, Collections.emptyList(),
storageConfiguration, tableConfig);
ArrayWritable row = new ArrayWritable(Writable.class, new Writable[]{new
Text("value1"), new Text("value2"), new ArrayWritable(new String[]{"value3"})});
@@ -101,7 +103,7 @@ class TestHiveHoodieReaderContext {
@Test
void getNestedField() {
- when(tableConfig.populateMetaFields()).thenReturn(true);
+ when(tableConfig.isRecordKeyPopulated()).thenReturn(true);
HiveHoodieReaderContext avroReaderContext = new
HiveHoodieReaderContext(readerCreator, Collections.emptyList(),
storageConfiguration, tableConfig);
ArrayWritable row = new ArrayWritable(Writable.class, new Writable[]{new
Text("value1"), new Text("value2"), new ArrayWritable(new String[]{"value3"})});
@@ -110,7 +112,7 @@ class TestHiveHoodieReaderContext {
@Test
void testConstructEngineRecordWithFieldValues() {
- when(tableConfig.populateMetaFields()).thenReturn(true);
+ when(tableConfig.isRecordKeyPopulated()).thenReturn(true);
HiveHoodieReaderContext avroReaderContext = new HiveHoodieReaderContext(
readerCreator, Collections.emptyList(), storageConfiguration,
tableConfig);
Object[] fieldVals = new Writable[]{
@@ -126,7 +128,7 @@ class TestHiveHoodieReaderContext {
@Test
void testConstructEngineRecordWithNoUpdates() {
- when(tableConfig.populateMetaFields()).thenReturn(true);
+ when(tableConfig.isRecordKeyPopulated()).thenReturn(true);
HiveHoodieReaderContext avroReaderContext = new HiveHoodieReaderContext(
readerCreator, Collections.emptyList(), storageConfiguration,
tableConfig);
@@ -147,7 +149,7 @@ class TestHiveHoodieReaderContext {
@Test
void testConstructEngineRecordWithUpdates() {
- when(tableConfig.populateMetaFields()).thenReturn(true);
+ when(tableConfig.isRecordKeyPopulated()).thenReturn(true);
HiveHoodieReaderContext avroReaderContext = new HiveHoodieReaderContext(
readerCreator, Collections.emptyList(), storageConfiguration,
tableConfig);
@@ -389,7 +391,7 @@ class TestHiveHoodieReaderContext {
}
private HiveHoodieReaderContext newReaderContext() {
- when(tableConfig.populateMetaFields()).thenReturn(true);
+ when(tableConfig.isRecordKeyPopulated()).thenReturn(true);
HiveHoodieReaderContext readerContext =
new HiveHoodieReaderContext(readerCreator, Collections.emptyList(),
storageConfiguration, tableConfig);
readerContext.setNeedsBootstrapMerge(false);
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/functional/TestBufferedRecordMerger.java
b/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/functional/TestBufferedRecordMerger.java
index cb6999cb62b6..d074cff217ff 100644
---
a/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/functional/TestBufferedRecordMerger.java
+++
b/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/functional/TestBufferedRecordMerger.java
@@ -119,7 +119,8 @@ class TestBufferedRecordMerger extends
SparkClientFunctionalTestHarness {
storageConfig = mock(StorageConfiguration.class);
when(tableConfig.getPayloadClass()).thenReturn(
"org.apache.hudi.common.model.DefaultHoodieRecordPayload");
- when(tableConfig.populateMetaFields()).thenReturn(false);
+ when(tableConfig.isRecordKeyPopulated()).thenReturn(false);
+ when(tableConfig.getPartitionFields()).thenReturn(Option.empty());
when(tableConfig.getRecordKeyFields()).thenReturn(Option.of(new
String[]{"id"}));
// Create reader context.
props = new TypedProperties();
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/functional/TestMetaFieldsModeE2E.java
b/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/functional/TestMetaFieldsModeE2E.java
index 76cfbe687ed2..cb5f185b784c 100644
---
a/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/functional/TestMetaFieldsModeE2E.java
+++
b/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/functional/TestMetaFieldsModeE2E.java
@@ -30,6 +30,7 @@ import org.apache.hudi.common.table.HoodieTableMetaClient;
import org.apache.hudi.common.table.HoodieTableVersion;
import org.apache.hudi.common.table.timeline.HoodieInstant;
import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.io.FileGroupReaderBasedMergeHandle;
import org.apache.hudi.testutils.SparkClientFunctionalTestHarness;
import org.apache.spark.api.java.function.VoidFunction2;
@@ -157,6 +158,39 @@ class TestMetaFieldsModeE2E extends
SparkClientFunctionalTestHarness {
}
}
+ @ParameterizedTest
+ @EnumSource(value = MetaFieldsMode.class, names = "NONE", mode =
EnumSource.Mode.EXCLUDE)
+ void concatPreservesRecordsWithTheSparkRecordType(MetaFieldsMode mode) {
+ Map<String, String> options = baseOptions();
+ options.put(HoodieTableConfig.META_FIELDS_MODE.key(), mode.name());
+ options.put(DataSourceWriteOptions.OPERATION().key(),
DataSourceWriteOptions.INSERT_OPERATION_OPT_VAL());
+ options.put("hoodie.write.record.merge.custom.implementation.classes",
"org.apache.hudi.DefaultSparkRecordMerger");
+
options.put(HoodieWriteConfig.MERGE_ALLOW_DUPLICATE_ON_INSERTS_ENABLE.key(),
"true");
+ options.put("hoodie.parquet.small.file.limit", "10485760");
+
+ writeSampleAndGetTableConfig(options, basePath());
+ // Route a duplicate insert into the existing small file, exercising
HoodieConcatHandle.
+ writeRows(Collections.singletonList(RowFactory.create("k1", "p1", "v3")),
+ simpleSchema(), options, basePath(), SaveMode.Append);
+
+ Dataset<Row> latest = spark().read().format("hudi").load(basePath())
+ .withColumn("actual_file", functions.input_file_name());
+ assertEquals(1, latest.select("actual_file").distinct().count(), "the
inserts must share one file group");
+ assertEquals(Arrays.asList("v1", "v2", "v3"),
latest.select("column3").collectAsList().stream()
+ .map(row -> row.getString(0)).sorted().collect(Collectors.toList()),
+ "concat must preserve both existing records and the duplicate insert");
+ assertMetaColumn(latest, 3, HoodieRecord.COMMIT_TIME_METADATA_FIELD,
mode.isCommitTimePopulated(), mode);
+ assertMetaColumn(latest, 3, HoodieRecord.COMMIT_SEQNO_METADATA_FIELD, mode
== MetaFieldsMode.ALL, mode);
+ assertMetaColumn(latest, 3, HoodieRecord.RECORD_KEY_METADATA_FIELD,
mode.isRecordKeyPopulated(), mode);
+ assertMetaColumn(latest, 3, HoodieRecord.PARTITION_PATH_METADATA_FIELD,
mode == MetaFieldsMode.ALL, mode);
+ assertMetaColumn(latest, 3, HoodieRecord.FILENAME_METADATA_FIELD,
mode.isFileNamePopulated(), mode);
+ if (mode.isFileNamePopulated()) {
+ for (Row row : latest.collectAsList()) {
+ assertTrue(row.<String>getAs("actual_file").endsWith("/" +
row.getAs(HoodieRecord.FILENAME_METADATA_FIELD)));
+ }
+ }
+ }
+
@Test
void allModePersistsAndPopulatesAllColumns() {
Map<String, String> options = baseOptions();
@@ -676,6 +710,40 @@ class TestMetaFieldsModeE2E extends
SparkClientFunctionalTestHarness {
// the one write path that stamps a meta column outside the mode-aware
writers.
//
---------------------------------------------------------------------------------------------
+ @ParameterizedTest
+ @EnumSource(value = MetaFieldsMode.class, names = "ALL", mode =
EnumSource.Mode.EXCLUDE)
+ void upsertWithSingleRecordKeyAndMultiplePartitionFields(MetaFieldsMode
mode) {
+ Map<String, String> options = baseOptions();
+ options.put(HoodieTableConfig.META_FIELDS_MODE.key(), mode.name());
+ options.put(DataSourceWriteOptions.PARTITIONPATH_FIELD().key(),
"column2,column4");
+ options.put(DataSourceWriteOptions.OPERATION().key(),
DataSourceWriteOptions.UPSERT_OPERATION_OPT_VAL());
+ options.put("hoodie.write.record.merge.custom.implementation.classes",
"org.apache.hudi.DefaultSparkRecordMerger");
+ options.put(HoodieWriteConfig.MERGE_HANDLE_CLASS_NAME.key(),
FileGroupReaderBasedMergeHandle.class.getName());
+ options.put(HoodieWriteConfig.COMPLEX_KEYGEN_NEW_ENCODING.key(), "false");
+ StructType schema = simpleSchema().add("column4", DataTypes.StringType,
true);
+
+ writeRows(Arrays.asList(
+ RowFactory.create("k1", "p1", "v1", "a"),
+ RowFactory.create("k2", "p1", "v2", "a"),
+ RowFactory.create("k3", "p1", "v3", "b")),
+ schema, options, basePath(), SaveMode.Overwrite);
+ writeRows(Collections.singletonList(RowFactory.create("k1", "p1", "v4",
"a")),
+ schema, options, basePath(), SaveMode.Append);
+
+ Dataset<Row> latest = spark().read().format("hudi").load(basePath());
+ assertEquals(Arrays.asList(
+ RowFactory.create("k1", "p1", "v4", "a"),
+ RowFactory.create("k2", "p1", "v2", "a"),
+ RowFactory.create("k3", "p1", "v3", "b")),
+ latest.select("column1", "column2", "column3",
"column4").orderBy("column1").collectAsList(),
+ "upsert must replace the existing key exactly once and preserve
untouched records in both partitions");
+ if (mode == MetaFieldsMode.NONE) {
+
assertFalse(Arrays.asList(latest.columns()).contains(HoodieRecord.RECORD_KEY_METADATA_FIELD));
+ } else {
+ assertMetaColumn(latest, 3, HoodieRecord.RECORD_KEY_METADATA_FIELD,
false, mode);
+ }
+ }
+
/**
* An upsert rewrites the whole file group: the updated record goes through
the normal write path,
* while untouched records are copied forward through
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/common/table/read/TestHoodieFileGroupReaderOnSpark.scala
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/common/table/read/TestHoodieFileGroupReaderOnSpark.scala
index db2488c7c8e9..550c4c456531 100644
---
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/common/table/read/TestHoodieFileGroupReaderOnSpark.scala
+++
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/common/table/read/TestHoodieFileGroupReaderOnSpark.scala
@@ -202,7 +202,7 @@ class TestHoodieFileGroupReaderOnSpark extends
TestHoodieFileGroupReaderBase[Int
def testGetOrderingValue(): Unit = {
val reader = Mockito.mock(classOf[SparkColumnarFileReader])
val tableConfig = Mockito.mock(classOf[HoodieTableConfig])
- Mockito.when(tableConfig.populateMetaFields()).thenReturn(true)
+ Mockito.when(tableConfig.isRecordKeyPopulated()).thenReturn(true)
val sparkReaderContext = new
SparkFileFormatInternalRowReaderContext(reader, Seq.empty, Seq.empty,
getStorageConf, tableConfig)
val orderingFieldName = "col2"
val avroSchema = new Schema.Parser().parse(
@@ -346,7 +346,7 @@ class TestHoodieFileGroupReaderOnSpark extends
TestHoodieFileGroupReaderBase[Int
val reader = Mockito.mock(classOf[SparkColumnarFileReader])
val tableConfig = Mockito.mock(classOf[HoodieTableConfig])
val storageConf = Mockito.mock(classOf[StorageConfiguration[_]])
- when(tableConfig.populateMetaFields()).thenReturn(true)
+ when(tableConfig.isRecordKeyPopulated()).thenReturn(true)
val sparkReaderContext = new
SparkFileFormatInternalRowReaderContext(reader, Seq.empty, Seq.empty,
storageConf, tableConfig)
val schema = SchemaBuilder.builder()
.record("test")
@@ -363,7 +363,8 @@ class TestHoodieFileGroupReaderOnSpark extends
TestHoodieFileGroupReaderBase[Int
def getRecordKeySingleKey(): Unit = {
val reader = Mockito.mock(classOf[SparkColumnarFileReader])
val tableConfig = Mockito.mock(classOf[HoodieTableConfig])
- when(tableConfig.populateMetaFields()).thenReturn(false)
+ when(tableConfig.isRecordKeyPopulated()).thenReturn(false)
+
when(tableConfig.getPartitionFields()).thenReturn(HOption.empty[Array[String]]())
when(tableConfig.getRecordKeyFields).thenReturn(HOption.of(Array("field1")))
val storageConf = Mockito.mock(classOf[StorageConfiguration[_]])
val props = new TypedProperties
@@ -385,7 +386,8 @@ class TestHoodieFileGroupReaderOnSpark extends
TestHoodieFileGroupReaderBase[Int
def getRecordKeyWithMultipleKeys(): Unit = {
val reader = Mockito.mock(classOf[SparkColumnarFileReader])
val tableConfig = Mockito.mock(classOf[HoodieTableConfig])
- when(tableConfig.populateMetaFields()).thenReturn(false)
+ when(tableConfig.isRecordKeyPopulated()).thenReturn(false)
+
when(tableConfig.getPartitionFields()).thenReturn(HOption.empty[Array[String]]())
when(tableConfig.getRecordKeyFields).thenReturn(HOption.of(Array("outer1.field1",
"outer1.field2", "outer1.field3")))
val storageConf = Mockito.mock(classOf[StorageConfiguration[_]])
val sparkReaderContext = new
SparkFileFormatInternalRowReaderContext(reader, Seq.empty, Seq.empty,
storageConf, tableConfig)
@@ -400,7 +402,7 @@ class TestHoodieFileGroupReaderOnSpark extends
TestHoodieFileGroupReaderBase[Int
def getNestedValue(): Unit = {
val reader = Mockito.mock(classOf[SparkColumnarFileReader])
val tableConfig = Mockito.mock(classOf[HoodieTableConfig])
- when(tableConfig.populateMetaFields()).thenReturn(true)
+ when(tableConfig.isRecordKeyPopulated()).thenReturn(true)
val storageConf = Mockito.mock(classOf[StorageConfiguration[_]])
val sparkReaderContext = new
SparkFileFormatInternalRowReaderContext(reader, Seq.empty, Seq.empty,
storageConf, tableConfig)
val schema: Schema = buildMultiLevelSchema
