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 61dde8631136 fix(flink): make the bucket pruning function serializable
for source V2 (#20318)
61dde8631136 is described below
commit 61dde86311368a7d373e582f40b941db55aa60a4
Author: Vova Kolmakov <[email protected]>
AuthorDate: Wed Oct 14 14:25:11 2026 +0700
fix(flink): make the bucket pruning function serializable for source V2
(#20318)
Co-authored-by: Vova Kolmakov <[email protected]>
---
.../org/apache/hudi/source/prune/PrimaryKeyPruners.java | 4 ++--
.../apache/hudi/source/prune/TestPrimaryKeyPruners.java | 17 +++++++++++++++++
.../org/apache/hudi/table/ITTestHoodieDataSource.java | 5 +++--
3 files changed, 22 insertions(+), 4 deletions(-)
diff --git
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/prune/PrimaryKeyPruners.java
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/prune/PrimaryKeyPruners.java
index 45dea1455312..b84239cf3b5d 100644
---
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/prune/PrimaryKeyPruners.java
+++
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/prune/PrimaryKeyPruners.java
@@ -30,10 +30,10 @@ import org.apache.flink.table.expressions.Expression;
import org.apache.flink.table.expressions.FieldReferenceExpression;
import org.apache.flink.table.expressions.ResolvedExpression;
import org.apache.flink.table.expressions.ValueLiteralExpression;
+import org.apache.flink.util.function.SerializableFunction;
import java.util.Arrays;
import java.util.List;
-import java.util.function.Function;
import java.util.stream.Collectors;
/**
@@ -42,7 +42,7 @@ import java.util.stream.Collectors;
@Slf4j
public class PrimaryKeyPruners {
- public static Function<Integer, Integer>
getBucketIdFunc(List<ResolvedExpression> hashKeyFilters, Configuration conf) {
+ public static SerializableFunction<Integer, Integer>
getBucketIdFunc(List<ResolvedExpression> hashKeyFilters, Configuration conf) {
List<String> pkFields = Arrays.asList(OptionsResolver.getRecordKeys(conf));
// step1: resolve the hash key values
final boolean logicalTimestamp =
OptionsResolver.isConsistentLogicalTimestampEnabled(conf);
diff --git
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/source/prune/TestPrimaryKeyPruners.java
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/source/prune/TestPrimaryKeyPruners.java
index f96574aec723..278330e41678 100644
---
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/source/prune/TestPrimaryKeyPruners.java
+++
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/source/prune/TestPrimaryKeyPruners.java
@@ -36,6 +36,8 @@ import java.util.Arrays;
import java.util.List;
import java.util.function.Function;
+import static org.apache.flink.util.InstantiationUtil.deserializeObject;
+import static org.apache.flink.util.InstantiationUtil.serializeObject;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.mockito.Mockito.mock;
@@ -62,6 +64,21 @@ class TestPrimaryKeyPruners {
assertNull(PartitionBucketIdFunc.create(Option.empty(),
mock(HoodieTableMetaClient.class), 8));
}
+ @Test
+ void testBucketIdFunctionIsSerializable() throws Exception {
+ Configuration conf = new Configuration();
+ conf.set(FlinkOptions.RECORD_KEY_FIELD, "id,tenant");
+ Function<Integer, Integer> bucketId = PrimaryKeyPruners.getBucketIdFunc(
+ Arrays.asList(equality("tenant", "tenant-a", true), equality("id",
"id-1", false)), conf);
+
+ // the source V2 ships the function to the job manager within the job graph
+ Function<Integer, Integer> deserialized =
deserializeObject(serializeObject(bucketId), getClass().getClassLoader());
+
+ List<String> orderedValues = Arrays.asList("id-1", "tenant-a");
+ assertEquals(BucketIdentifier.getBucketId(orderedValues, 8),
deserialized.apply(8));
+ assertEquals(BucketIdentifier.getBucketId(orderedValues, 16),
deserialized.apply(16));
+ }
+
private static ResolvedExpression equality(String field, String value,
boolean literalFirst) {
FieldReferenceExpression fieldReference =
new FieldReferenceExpression(field, DataTypes.STRING(), 0, 0);
diff --git
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/ITTestHoodieDataSource.java
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/ITTestHoodieDataSource.java
index 69b22e266b6a..5740a5e2e6d3 100644
---
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/ITTestHoodieDataSource.java
+++
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/ITTestHoodieDataSource.java
@@ -3196,14 +3196,15 @@ public class ITTestHoodieDataSource {
}
@ParameterizedTest
- @EnumSource(value = HoodieTableType.class)
- void testBucketPruning(HoodieTableType tableType) {
+ @MethodSource("tableTypeAndBooleanTrueFalseParams")
+ void testBucketPruning(HoodieTableType tableType, boolean useSourceV2) {
TableEnvironment tableEnv = batchTableEnv;
String hoodieTableDDL = sql("t1")
.option(FlinkOptions.PATH, tempFile.getAbsolutePath())
.options(getDefaultKeys())
.option(FlinkOptions.INDEX_TYPE, "BUCKET")
.option(FlinkOptions.TABLE_TYPE, tableType)
+ .option(FlinkOptions.READ_SOURCE_V2_ENABLED, useSourceV2)
.end();
tableEnv.executeSql(hoodieTableDDL);
