This is an automated email from the ASF dual-hosted git repository. voonhous pushed a commit to branch release-1.2.1 in repository https://gitbox.apache.org/repos/asf/hudi.git
commit d36f87f74ae3ea765274a4d77112abbd2f07622d Author: Shuo Cheng <[email protected]> AuthorDate: Thu Jun 4 17:03:18 2026 +0800 refactor(core): Unify record key/index key splitting and extraction (#18842) (cherry picked from commit cb14ca7275833e9e76f1c3116ba96cf31af6f8e9) --- .../org/apache/hudi/config/HoodieIndexConfig.java | 11 +++++---- .../apache/hudi/index/bucket/BucketIdentifier.java | 2 +- .../hudi/index/bucket/HoodieBucketIndex.java | 4 ++-- .../apache/hudi/keygen/CustomAvroKeyGenerator.java | 6 +---- .../java/org/apache/hudi/keygen/KeyGenUtils.java | 26 +++++++++++++--------- .../table/BucketIndexBulkInsertPartitioner.java | 4 ++-- .../hudi/client/TestBaseHoodieWriteClient.java | 5 +++-- .../apache/hudi/config/TestHoodieWriteConfig.java | 20 +++++++++++++++++ .../org/apache/hudi/keygen/TestKeyGenUtils.java | 21 +++++++++++++++++ ...arkConsistentBucketDuplicateUpdateStrategy.java | 4 ++-- .../org/apache/hudi/keygen/CustomKeyGenerator.java | 8 +------ .../apache/hudi/configuration/OptionsResolver.java | 13 ++++------- .../bucket/ConsistentBucketAssignFunction.java | 2 +- .../ConsistentBucketStreamWriteFunction.java | 4 ++-- .../org/apache/hudi/table/HoodieTableSource.java | 2 +- .../hudi/configuration/TestOptionsResolver.java | 4 ++-- .../scala/org/apache/hudi/BucketIndexSupport.scala | 4 ++-- .../hudi/command/DeleteHoodieTableCommand.scala | 5 ++++- .../hudi/command/DeleteHoodieTableCommand.scala | 5 ++++- 19 files changed, 94 insertions(+), 56 deletions(-) diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieIndexConfig.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieIndexConfig.java index 9cedac1be74d..1d2c6bb2b426 100644 --- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieIndexConfig.java +++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieIndexConfig.java @@ -29,6 +29,7 @@ import org.apache.hudi.exception.HoodieIndexException; import org.apache.hudi.exception.HoodieNotSupportedException; import org.apache.hudi.index.HoodieIndex; import org.apache.hudi.index.bucket.partition.PartitionBucketIndexRule; +import org.apache.hudi.keygen.KeyGenUtils; import org.apache.hudi.keygen.constant.KeyGeneratorOptions; import lombok.Getter; @@ -39,9 +40,8 @@ import javax.annotation.concurrent.Immutable; import java.io.File; import java.io.FileReader; import java.io.IOException; -import java.util.Arrays; +import java.util.List; import java.util.Properties; -import java.util.stream.Collectors; import static org.apache.hudi.common.config.HoodieStorageConfig.BLOOM_FILTER_DYNAMIC_MAX_ENTRIES; import static org.apache.hudi.common.config.HoodieStorageConfig.BLOOM_FILTER_FPP_VALUE; @@ -777,10 +777,9 @@ public class HoodieIndexConfig extends HoodieConfig { hoodieIndexConfig.setValue(BUCKET_INDEX_HASH_FIELD, hoodieIndexConfig.getString(KeyGeneratorOptions.RECORDKEY_FIELD_NAME)); } else { - boolean valid = Arrays - .stream(hoodieIndexConfig.getString(KeyGeneratorOptions.RECORDKEY_FIELD_NAME).split(",")) - .collect(Collectors.toSet()) - .containsAll(Arrays.asList(hoodieIndexConfig.getString(BUCKET_INDEX_HASH_FIELD).split(","))); + List<String> recordKeyFields = KeyGenUtils.getRecordKeyFields(hoodieIndexConfig.getString(KeyGeneratorOptions.RECORDKEY_FIELD_NAME)); + List<String> indexKeyFields = KeyGenUtils.getIndexKeyFields(hoodieIndexConfig.getString(BUCKET_INDEX_HASH_FIELD)); + boolean valid = recordKeyFields.containsAll(indexKeyFields); if (!valid) { throw new HoodieIndexException("Bucket index key (if configured) must be subset of record key."); } diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/bucket/BucketIdentifier.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/bucket/BucketIdentifier.java index eed3ab39599c..2bde3aec815b 100644 --- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/bucket/BucketIdentifier.java +++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/bucket/BucketIdentifier.java @@ -42,7 +42,7 @@ public class BucketIdentifier implements Serializable { } protected static List<String> getHashKeys(String recordKey, String indexKeyFields) { - return getHashKeysUsingIndexFields(recordKey, Arrays.asList(indexKeyFields.split(","))); + return getHashKeysUsingIndexFields(recordKey, KeyGenUtils.getIndexKeyFields(indexKeyFields)); } protected static List<String> getHashKeys(String recordKey, List<String> indexKeyFields) { diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/bucket/HoodieBucketIndex.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/bucket/HoodieBucketIndex.java index 38c7cb5319a3..7b9c28b46216 100644 --- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/bucket/HoodieBucketIndex.java +++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/bucket/HoodieBucketIndex.java @@ -29,13 +29,13 @@ import org.apache.hudi.common.util.Option; import org.apache.hudi.config.HoodieWriteConfig; import org.apache.hudi.exception.HoodieIndexException; import org.apache.hudi.index.HoodieIndex; +import org.apache.hudi.keygen.KeyGenUtils; import org.apache.hudi.table.HoodieTable; import lombok.Getter; import lombok.extern.slf4j.Slf4j; import java.io.Serializable; -import java.util.Arrays; import java.util.HashMap; import java.util.List; import java.util.Map; @@ -57,7 +57,7 @@ public abstract class HoodieBucketIndex extends HoodieIndex<Object, Object> { super(config); this.numBuckets = config.getBucketIndexNumBuckets(); - this.indexKeyFields = Arrays.asList(config.getBucketIndexHashField().split(",")); + this.indexKeyFields = KeyGenUtils.getIndexKeyFields(config.getBucketIndexHashField()); log.info("Use bucket index, numBuckets = " + numBuckets + ", indexFields: " + indexKeyFields); } diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/keygen/CustomAvroKeyGenerator.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/keygen/CustomAvroKeyGenerator.java index beeb4947acd4..6ab6378367ae 100644 --- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/keygen/CustomAvroKeyGenerator.java +++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/keygen/CustomAvroKeyGenerator.java @@ -62,11 +62,7 @@ public class CustomAvroKeyGenerator extends BaseKeyGenerator { public CustomAvroKeyGenerator(TypedProperties props) { super(props); - this.recordKeyFields = Option.ofNullable(props.getString(KeyGeneratorOptions.RECORDKEY_FIELD_NAME.key(), null)) - .map(recordKeyConfigValue -> - Arrays.stream(recordKeyConfigValue.split(FIELD_SEPARATOR)) - .map(String::trim).collect(Collectors.toList()) - ).orElse(Collections.emptyList()); + this.recordKeyFields = KeyGenUtils.getRecordKeyFields(props); this.partitionPathFields = Arrays.stream(props.getString(KeyGeneratorOptions.PARTITIONPATH_FIELD_NAME.key()).split(FIELD_SEPARATOR)).map(String::trim).collect(Collectors.toList()); this.recordKeyGenerator = getRecordKeyFieldNames().size() == 1 ? new SimpleAvroKeyGenerator(config) : new ComplexAvroKeyGenerator(config); this.partitionKeyGenerators = getPartitionKeyGenerators(this.partitionPathFields, config); diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/keygen/KeyGenUtils.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/keygen/KeyGenUtils.java index 292f261c2223..81faa0cdb2ba 100644 --- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/keygen/KeyGenUtils.java +++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/keygen/KeyGenUtils.java @@ -39,11 +39,9 @@ import org.apache.avro.generic.GenericRecord; import java.io.IOException; import java.util.ArrayList; -import java.util.Arrays; import java.util.Collections; import java.util.List; import java.util.function.BiFunction; -import java.util.stream.Collectors; import static org.apache.hudi.config.HoodieWriteConfig.COMPLEX_KEYGEN_NEW_ENCODING; import static org.apache.hudi.config.HoodieWriteConfig.WRITE_TABLE_VERSION; @@ -72,7 +70,7 @@ public class KeyGenUtils { */ public static KeyGeneratorType inferKeyGeneratorType( Option<String> recordsKeyFields, String partitionFields) { - int numRecordKeyFields = recordsKeyFields.map(fields -> fields.split(",").length).orElse(0); + int numRecordKeyFields = recordsKeyFields.map(keyStr -> getRecordKeyFields(keyStr).size()).orElse(0); KeyGeneratorType partitionKeyGeneratorType = inferKeyGeneratorTypeFromPartitionFields(partitionFields); if (numRecordKeyFields <= 1) { return partitionKeyGeneratorType; @@ -331,13 +329,21 @@ public class KeyGenUtils { } public static List<String> getRecordKeyFields(TypedProperties props) { - return Option.ofNullable(props.getString(KeyGeneratorOptions.RECORDKEY_FIELD_NAME.key(), null)) - .map(recordKeyConfigValue -> - Arrays.stream(recordKeyConfigValue.split(",")) - .map(String::trim) - .filter(s -> !s.isEmpty()) - .collect(Collectors.toList()) - ).orElse(Collections.emptyList()); + return getRecordKeyFields(props.getString(KeyGeneratorOptions.RECORDKEY_FIELD_NAME.key(), null)); + } + + public static List<String> getRecordKeyFields(String recordKeys) { + return getKeyFields(recordKeys); + } + + public static List<String> getIndexKeyFields(String indexKeys) { + return getKeyFields(indexKeys); + } + + private static List<String> getKeyFields(String keys) { + return Option.ofNullable(keys) + .map(value -> StringUtils.split(value, ",")) + .orElse(Collections.emptyList()); } /** diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/BucketIndexBulkInsertPartitioner.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/BucketIndexBulkInsertPartitioner.java index 7d8d7c6900ed..5fd3909f0562 100644 --- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/BucketIndexBulkInsertPartitioner.java +++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/BucketIndexBulkInsertPartitioner.java @@ -25,12 +25,12 @@ import org.apache.hudi.exception.HoodieNotSupportedException; import org.apache.hudi.io.AppendHandleFactory; import org.apache.hudi.io.SingleFileHandleCreateFactory; import org.apache.hudi.io.WriteHandleFactory; +import org.apache.hudi.keygen.KeyGenUtils; import org.apache.logging.log4j.LogManager; import org.apache.logging.log4j.Logger; import java.util.ArrayList; -import java.util.Arrays; import java.util.List; /** @@ -50,7 +50,7 @@ public abstract class BucketIndexBulkInsertPartitioner<T> extends BucketSortBulk public BucketIndexBulkInsertPartitioner(HoodieTable table, String sortString, boolean preserveHoodieMetadata) { super(table, sortString); - this.indexKeyFields = Arrays.asList(table.getConfig().getBucketIndexHashField().split(",")); + this.indexKeyFields = KeyGenUtils.getIndexKeyFields(table.getConfig().getBucketIndexHashField()); this.consistentLogicalTimestampEnabled = table.getConfig().isConsistentLogicalTimestampEnabled(); this.preserveHoodieMetadata = preserveHoodieMetadata; // Multiple bulk inserts into COW using `BucketIndexBulkInsertPartitioner` is restricted, otherwise AppendHandleFactory will produce MOR log files diff --git a/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/client/TestBaseHoodieWriteClient.java b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/client/TestBaseHoodieWriteClient.java index 377a54eedcd8..9190dde5ab1e 100644 --- a/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/client/TestBaseHoodieWriteClient.java +++ b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/client/TestBaseHoodieWriteClient.java @@ -46,6 +46,7 @@ import org.apache.hudi.config.HoodieWriteConfig; import org.apache.hudi.index.HoodieIndex; import org.apache.hudi.index.simple.HoodieSimpleIndex; import org.apache.hudi.keygen.ComplexAvroKeyGenerator; +import org.apache.hudi.keygen.KeyGenUtils; import org.apache.hudi.keygen.constant.KeyGeneratorOptions; import org.apache.hudi.table.BulkInsertPartitioner; import org.apache.hudi.table.HoodieTable; @@ -242,7 +243,7 @@ class TestBaseHoodieWriteClient extends HoodieCommonTestHarness { if (tableVersion <= 8 && enableComplexKeyGeneratorValidation && (ComplexAvroKeyGenerator.class.getCanonicalName().equals(keyGeneratorClass) || "org.apache.hudi.keygen.ComplexKeyGenerator".equals(keyGeneratorClass)) - && recordKeyFields.split(",").length == 1) { + && KeyGenUtils.getRecordKeyFields(recordKeyFields).size() == 1) { assertComplexKeyGeneratorValidationThrows(() -> writeClient.initTable(WriteOperationType.INSERT, Option.empty()), "ingestion"); } else { writeClient.initTable(WriteOperationType.INSERT, Option.empty()); @@ -399,4 +400,4 @@ class TestBaseHoodieWriteClient extends HoodieCommonTestHarness { } } -} \ No newline at end of file +} diff --git a/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/config/TestHoodieWriteConfig.java b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/config/TestHoodieWriteConfig.java index 6c1d1418b190..044e41ed89d2 100644 --- a/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/config/TestHoodieWriteConfig.java +++ b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/config/TestHoodieWriteConfig.java @@ -37,6 +37,7 @@ import org.apache.hudi.common.table.marker.MarkerType; import org.apache.hudi.common.table.view.FileSystemViewStorageConfig; import org.apache.hudi.common.util.CollectionUtils; import org.apache.hudi.config.HoodieWriteConfig.Builder; +import org.apache.hudi.exception.HoodieIndexException; import org.apache.hudi.index.HoodieIndex; import org.apache.hudi.keygen.constant.KeyGeneratorOptions; @@ -619,6 +620,25 @@ public class TestHoodieWriteConfig { assertEquals("org.apache.hudi.table.action.commit.UpsertPartitioner", overwritePartitioner.getString(HoodieLayoutConfig.LAYOUT_PARTITIONER_CLASS_NAME)); } + @Test + public void testBucketIndexKeyFieldValidationTrimsFields() { + Properties props = new Properties(); + props.setProperty(KeyGeneratorOptions.RECORDKEY_FIELD_NAME.key(), "uuid, name"); + + HoodieIndexConfig indexConfig = HoodieIndexConfig.newBuilder() + .fromProperties(props) + .withIndexType(HoodieIndex.IndexType.BUCKET) + .withIndexKeyField(" name") + .build(); + assertEquals(" name", indexConfig.getString(HoodieIndexConfig.BUCKET_INDEX_HASH_FIELD)); + + assertThrows(HoodieIndexException.class, () -> HoodieIndexConfig.newBuilder() + .fromProperties(props) + .withIndexType(HoodieIndex.IndexType.BUCKET) + .withIndexKeyField("missing") + .build()); + } + @Test void testBloomIndexFileIdKeySortingConfig() { Properties props = new Properties(); diff --git a/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/keygen/TestKeyGenUtils.java b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/keygen/TestKeyGenUtils.java index 777c4792901f..cd837c0608e8 100644 --- a/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/keygen/TestKeyGenUtils.java +++ b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/keygen/TestKeyGenUtils.java @@ -18,9 +18,11 @@ package org.apache.hudi.keygen; +import org.apache.hudi.common.config.TypedProperties; import org.apache.hudi.common.table.HoodieTableConfig; import org.apache.hudi.common.util.Option; import org.apache.hudi.exception.HoodieKeyException; +import org.apache.hudi.keygen.constant.KeyGeneratorOptions; import org.apache.hudi.keygen.constant.KeyGeneratorType; import org.apache.avro.Schema; @@ -31,6 +33,7 @@ import org.junit.jupiter.api.Test; import java.util.ArrayList; import java.util.Arrays; +import java.util.Collections; import java.util.List; import static org.apache.hudi.common.table.HoodieTableConfig.KEY_GENERATOR_TYPE; @@ -121,6 +124,24 @@ public class TestKeyGenUtils { KeyGenUtils.inferKeyGeneratorTypeFromPartitionFields(null)); } + @Test + public void testGetRecordKeyFields() { + assertEquals(Collections.emptyList(), KeyGenUtils.getRecordKeyFields((String) null)); + assertEquals(Collections.emptyList(), KeyGenUtils.getRecordKeyFields("")); + assertEquals(Arrays.asList("id", "ts", "name"), KeyGenUtils.getRecordKeyFields(" id,ts, name,, ")); + + TypedProperties props = new TypedProperties(); + props.setProperty(KeyGeneratorOptions.RECORDKEY_FIELD_NAME.key(), " id,ts "); + assertEquals(Arrays.asList("id", "ts"), KeyGenUtils.getRecordKeyFields(props)); + } + + @Test + public void testGetIndexKeyFields() { + assertEquals(Collections.emptyList(), KeyGenUtils.getIndexKeyFields(null)); + assertEquals(Collections.emptyList(), KeyGenUtils.getIndexKeyFields("")); + assertEquals(Arrays.asList("id", "ts", "name"), KeyGenUtils.getIndexKeyFields(" id,ts, name,, ")); + } + @Test public void testExtractRecordKeys() { // if for recordKey one column only is used, then there is no added column name before value diff --git a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/clustering/update/strategy/SparkConsistentBucketDuplicateUpdateStrategy.java b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/clustering/update/strategy/SparkConsistentBucketDuplicateUpdateStrategy.java index 2d8247de92e7..eb4b308a668a 100644 --- a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/clustering/update/strategy/SparkConsistentBucketDuplicateUpdateStrategy.java +++ b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/clustering/update/strategy/SparkConsistentBucketDuplicateUpdateStrategy.java @@ -28,11 +28,11 @@ import org.apache.hudi.common.model.HoodieRecordLocation; import org.apache.hudi.common.util.Option; import org.apache.hudi.common.util.collection.Pair; import org.apache.hudi.index.bucket.ConsistentBucketIdentifier; +import org.apache.hudi.keygen.KeyGenUtils; import org.apache.hudi.table.HoodieTable; import org.apache.hudi.table.action.cluster.strategy.UpdateStrategy; import org.apache.hudi.table.action.cluster.util.ConsistentHashingUpdateStrategyUtils; -import java.util.Arrays; import java.util.Collections; import java.util.HashSet; import java.util.List; @@ -74,7 +74,7 @@ public class SparkConsistentBucketDuplicateUpdateStrategy<T> extends UpdateStrat ConsistentHashingUpdateStrategyUtils.constructPartitionToIdentifier(partitions, table); // Produce records tagged with new record location - List<String> indexKeyFields = Arrays.asList(table.getConfig().getBucketIndexHashField().split(",")); + List<String> indexKeyFields = KeyGenUtils.getIndexKeyFields(table.getConfig().getBucketIndexHashField()); HoodieData<HoodieRecord<T>> redirectedRecordsRDD = filteredRecordsRDD.map(r -> { Pair<String, ConsistentBucketIdentifier> identifierPair = partitionToIdentifier.get(r.getPartitionPath()); ConsistentHashingNode node = identifierPair.getValue().getBucket(r.getKey(), indexKeyFields); diff --git a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/keygen/CustomKeyGenerator.java b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/keygen/CustomKeyGenerator.java index 954af2000c1b..45f77ff9ce28 100644 --- a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/keygen/CustomKeyGenerator.java +++ b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/keygen/CustomKeyGenerator.java @@ -63,12 +63,7 @@ public class CustomKeyGenerator extends BuiltinKeyGenerator { // NOTE: We have to strip partition-path configuration, since it could only be interpreted by // this key-gen super(stripPartitionPathConfig(props)); - this.recordKeyFields = Option.ofNullable(props.getString(KeyGeneratorOptions.RECORDKEY_FIELD_NAME.key(), null)) - .map(recordKeyConfigValue -> - Arrays.stream(recordKeyConfigValue.split(",")) - .map(String::trim) - .collect(Collectors.toList()) - ).orElse(Collections.emptyList()); + this.recordKeyFields = KeyGenUtils.getRecordKeyFields(props); String partitionPathFields = props.getString(KeyGeneratorOptions.PARTITIONPATH_FIELD_NAME.key()); this.partitionPathFields = partitionPathFields == null ? Collections.emptyList() @@ -167,4 +162,3 @@ public class CustomKeyGenerator extends BuiltinKeyGenerator { return filtered; } } - 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 74c5257cf44f..cc32351e62cf 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 @@ -40,6 +40,7 @@ import org.apache.hudi.config.HoodieWriteConfig; import org.apache.hudi.exception.HoodieException; import org.apache.hudi.index.HoodieIndex; import org.apache.hudi.index.bucket.partition.PartitionBucketIndexUtils; +import org.apache.hudi.keygen.KeyGenUtils; import org.apache.hudi.keygen.constant.KeyGeneratorOptions; import org.apache.hudi.sink.buffer.BufferMemoryType; import org.apache.hudi.sink.overwrite.PartitionOverwriteMode; @@ -168,21 +169,15 @@ public class OptionsResolver { */ public static String[] getRecordKeys(Configuration conf) { final String recordKeyStr = conf.get(FlinkOptions.RECORD_KEY_FIELD); - if (StringUtils.isNullOrEmpty(recordKeyStr)) { - return new String[]{}; - } - return recordKeyStr.split(","); + return KeyGenUtils.getRecordKeyFields(recordKeyStr).toArray(new String[0]); } /** * Return the bucket index keys as an array. */ public static String[] getBucketIndexKeys(Configuration conf) { - final String indexKeyStr = conf.get(FlinkOptions.INDEX_KEY_FIELD); - if (StringUtils.isNullOrEmpty(indexKeyStr)) { - return new String[]{}; - } - return indexKeyStr.split(","); + final String indexKeyStr = getIndexKeyField(conf); + return KeyGenUtils.getIndexKeyFields(indexKeyStr).toArray(new String[0]); } /** diff --git a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/bucket/ConsistentBucketAssignFunction.java b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/bucket/ConsistentBucketAssignFunction.java index ec2fa104c58e..089224ac985d 100644 --- a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/bucket/ConsistentBucketAssignFunction.java +++ b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/bucket/ConsistentBucketAssignFunction.java @@ -68,7 +68,7 @@ public class ConsistentBucketAssignFunction extends ProcessFunctionAdapter<Hoodi public ConsistentBucketAssignFunction(Configuration conf) { this.config = conf; - this.indexKeyFields = Arrays.asList(OptionsResolver.getIndexKeyField(conf).split(",")); + this.indexKeyFields = Arrays.asList(OptionsResolver.getBucketIndexKeys(conf)); this.bucketNum = conf.get(FlinkOptions.BUCKET_INDEX_NUM_BUCKETS); } diff --git a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/bucket/ConsistentBucketStreamWriteFunction.java b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/bucket/ConsistentBucketStreamWriteFunction.java index 40fd32f2e9c6..91f1fd91e61a 100644 --- a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/bucket/ConsistentBucketStreamWriteFunction.java +++ b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/bucket/ConsistentBucketStreamWriteFunction.java @@ -23,7 +23,7 @@ import org.apache.hudi.common.model.HoodieFileGroupId; import org.apache.hudi.common.model.HoodieRecord; import org.apache.hudi.common.util.collection.MappingIterator; import org.apache.hudi.common.util.collection.Pair; -import org.apache.hudi.configuration.FlinkOptions; +import org.apache.hudi.configuration.OptionsResolver; import org.apache.hudi.sink.StreamWriteFunction; import org.apache.hudi.sink.buffer.RowDataBucket; import org.apache.hudi.sink.clustering.update.strategy.ConsistentBucketUpdateStrategy; @@ -65,7 +65,7 @@ public class ConsistentBucketStreamWriteFunction extends StreamWriteFunction { @Override public void open(Configuration parameters) throws IOException { super.open(parameters); - List<String> indexKeyFields = Arrays.asList(config.get(FlinkOptions.INDEX_KEY_FIELD).split(",")); + List<String> indexKeyFields = Arrays.asList(OptionsResolver.getBucketIndexKeys(config)); this.updateStrategy = new ConsistentBucketUpdateStrategy(this.writeClient, indexKeyFields); log.info("Create update strategy with index key fields: {}", indexKeyFields); } diff --git a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/HoodieTableSource.java b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/HoodieTableSource.java index d1b13ff7e375..b5fe77b8e082 100644 --- a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/HoodieTableSource.java +++ b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/HoodieTableSource.java @@ -494,7 +494,7 @@ public class HoodieTableSource extends FileIndexReader implements if (!OptionsResolver.isBucketIndexType(conf) || dataFilters.isEmpty()) { return Option.empty(); } - Set<String> indexKeyFields = Arrays.stream(OptionsResolver.getIndexKeyField(conf).split(",")).collect(Collectors.toSet()); + Set<String> indexKeyFields = Arrays.stream(OptionsResolver.getBucketIndexKeys(conf)).collect(Collectors.toSet()); List<ResolvedExpression> indexKeyFilters = dataFilters.stream().filter(expr -> ExpressionUtils.isEqualsLitExpr(expr, indexKeyFields)).collect(Collectors.toList()); if (!ExpressionUtils.isFilteringByAllFields(indexKeyFilters, indexKeyFields)) { return Option.empty(); 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 aef93b7d10fb..0ed8e61fd350 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 @@ -66,7 +66,7 @@ public class TestOptionsResolver { assertArrayEquals(new String[]{}, OptionsResolver.getRecordKeys(conf)); conf.set(FlinkOptions.RECORD_KEY_FIELD, "uuid, name"); - assertArrayEquals(new String[]{"uuid", " name"}, OptionsResolver.getRecordKeys(conf)); + assertArrayEquals(new String[]{"uuid", "name"}, OptionsResolver.getRecordKeys(conf)); } @Test @@ -78,7 +78,7 @@ public class TestOptionsResolver { assertArrayEquals(new String[]{}, OptionsResolver.getBucketIndexKeys(conf)); conf.set(FlinkOptions.INDEX_KEY_FIELD, "uuid, name"); - assertArrayEquals(new String[]{"uuid", " name"}, OptionsResolver.getBucketIndexKeys(conf)); + assertArrayEquals(new String[]{"uuid", "name"}, OptionsResolver.getBucketIndexKeys(conf)); } @Test diff --git a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/BucketIndexSupport.scala b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/BucketIndexSupport.scala index 71c0ac4d379a..e9aee3ab78b4 100644 --- a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/BucketIndexSupport.scala +++ b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/BucketIndexSupport.scala @@ -27,6 +27,7 @@ import org.apache.hudi.index.HoodieIndex import org.apache.hudi.index.HoodieIndex.IndexType import org.apache.hudi.index.bucket.BucketIdentifier import org.apache.hudi.keygen.KeyGenerator +import org.apache.hudi.keygen.KeyGenUtils import org.apache.hudi.keygen.factory.HoodieSparkKeyGeneratorFactory import org.apache.avro.generic.GenericData @@ -212,7 +213,7 @@ class BucketIndexSupport(spark: SparkSession, if (bucketHashFields == null || bucketHashFields.isEmpty) { Option.apply(null) } else { - Option.apply(JavaConverters.seqAsJavaListConverter(bucketHashFields.split(",")).asJava) + Option.apply(KeyGenUtils.getIndexKeyFields(bucketHashFields)) } } @@ -224,4 +225,3 @@ class BucketIndexSupport(spark: SparkSession, object BucketIndexSupport { val INDEX_NAME = "BUCKET" } - diff --git a/hudi-spark-datasource/hudi-spark3-common/src/main/scala/org/apache/spark/sql/hudi/command/DeleteHoodieTableCommand.scala b/hudi-spark-datasource/hudi-spark3-common/src/main/scala/org/apache/spark/sql/hudi/command/DeleteHoodieTableCommand.scala index b03b6fec3b8f..bd4ca0c751e7 100644 --- a/hudi-spark-datasource/hudi-spark3-common/src/main/scala/org/apache/spark/sql/hudi/command/DeleteHoodieTableCommand.scala +++ b/hudi-spark-datasource/hudi-spark3-common/src/main/scala/org/apache/spark/sql/hudi/command/DeleteHoodieTableCommand.scala @@ -20,6 +20,7 @@ package org.apache.spark.sql.hudi.command import org.apache.hudi.{HoodieSparkSqlWriter, SparkAdapterSupport} import org.apache.hudi.DataSourceWriteOptions.{SPARK_SQL_OPTIMIZED_WRITES, SPARK_SQL_WRITES_PREPPED_KEY} import org.apache.hudi.common.table.HoodieTableConfig +import org.apache.hudi.keygen.KeyGenUtils import org.apache.spark.sql._ import org.apache.spark.sql.SparkSession @@ -34,6 +35,8 @@ import org.apache.spark.sql.hudi.ProvidesHoodieConfig import org.apache.spark.sql.hudi.command.HoodieCommandMetrics.updateCommitMetrics import org.apache.spark.sql.hudi.command.HoodieLeafRunnableCommand.stripMetaFieldAttributes +import scala.collection.JavaConverters._ + case class DeleteHoodieTableCommand(catalogTable: HoodieCatalogTable, query: LogicalPlan, config: Map[String, String]) extends DataWritingCommand with SparkAdapterSupport with ProvidesHoodieConfig { @@ -76,7 +79,7 @@ object DeleteHoodieTableCommand extends SparkAdapterSupport with ProvidesHoodieC } val recordKeysStr = config.getOrElse(HoodieTableConfig.RECORDKEY_FIELDS.key(), "") - val recordKeys = recordKeysStr.split(",").filter(_.nonEmpty) + val recordKeys = KeyGenUtils.getRecordKeyFields(recordKeysStr).asScala.toSeq // get all columns which are used in condition val conditionColumns = if (condition == null) { diff --git a/hudi-spark-datasource/hudi-spark4-common/src/main/scala/org/apache/spark/sql/hudi/command/DeleteHoodieTableCommand.scala b/hudi-spark-datasource/hudi-spark4-common/src/main/scala/org/apache/spark/sql/hudi/command/DeleteHoodieTableCommand.scala index e40cd2e4840b..b05d826fbb1b 100644 --- a/hudi-spark-datasource/hudi-spark4-common/src/main/scala/org/apache/spark/sql/hudi/command/DeleteHoodieTableCommand.scala +++ b/hudi-spark-datasource/hudi-spark4-common/src/main/scala/org/apache/spark/sql/hudi/command/DeleteHoodieTableCommand.scala @@ -20,6 +20,7 @@ package org.apache.spark.sql.hudi.command import org.apache.hudi.{HoodieSparkSqlWriter, SparkAdapterSupport} import org.apache.hudi.DataSourceWriteOptions.{SPARK_SQL_OPTIMIZED_WRITES, SPARK_SQL_WRITES_PREPPED_KEY} import org.apache.hudi.common.table.HoodieTableConfig +import org.apache.hudi.keygen.KeyGenUtils import org.apache.spark.sql import org.apache.spark.sql._ @@ -35,6 +36,8 @@ import org.apache.spark.sql.hudi.ProvidesHoodieConfig import org.apache.spark.sql.hudi.command.HoodieCommandMetrics.updateCommitMetrics import org.apache.spark.sql.hudi.command.HoodieLeafRunnableCommand.stripMetaFieldAttributes +import scala.collection.JavaConverters._ + case class DeleteHoodieTableCommand(catalogTable: HoodieCatalogTable, query: LogicalPlan, config: Map[String, String]) extends DataWritingCommand with SparkAdapterSupport with ProvidesHoodieConfig { @@ -81,7 +84,7 @@ object DeleteHoodieTableCommand extends SparkAdapterSupport with ProvidesHoodieC } val recordKeysStr = config.getOrElse(HoodieTableConfig.RECORDKEY_FIELDS.key(), "") - val recordKeys = recordKeysStr.split(",").filter(_.nonEmpty) + val recordKeys = KeyGenUtils.getRecordKeyFields(recordKeysStr).asScala.toSeq // get all columns which are used in condition val conditionColumns = if (condition == null) {
