You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by "wombatu-kun (via GitHub)" <gi...@apache.org> on 2026/10/14 04:35:37 UTC
[I] [BUG] flink: Source V2 query filtering a bucket index table by its keys fails with NotSerializableException [hudi]
wombatu-kun opened a new issue, #20317:
URL: https://github.com/apache/hudi/issues/20317
### Bug Description
**What happened:**
With the Flink Source V2 reader (`read.source-v2.enabled=true`), a query on a bucket index table fails while Flink builds the job graph when its filter has an equality predicate on every bucket index key field (`hoodie.bucket.index.hash.field`, which defaults to the record key fields): `java.io.NotSerializableException: org.apache.hudi.source.prune.PrimaryKeyPruners$$Lambda$...`. The same query succeeds without the filter, and the filtered query succeeds on the legacy source. Batch and streaming reads fail alike, on both COPY_ON_WRITE and MERGE_ON_READ tables.
`HoodieTableSource#applyFilters` turns such a filter into a bucket id function for bucket pruning with `PrimaryKeyPruners.getBucketIdFunc`, which returns a plain `java.util.function.Function` lambda that is not serializable. Flink serializes the FLIP-27 `HoodieSource` into the job graph, and the source reaches the lambda in two ways:
- `HoodieSource.readerFunctionSupplier` is a lambda created in `HoodieTableSource#createHoodieSource` that reads instance fields, so it captures the whole `HoodieTableSource`, including its `dataBucketFunc` field. The serialization fails on this path (trace below), which 1.2.1 has too.
- `HoodieSource.scanContext` holds a `PartitionBucketIdFunc` wrapping the same lambda, for bucket pruning in the split enumerator (added by #18716, master only).
The legacy source is not affected because none of the objects it ships to the cluster holds the function.
**What you expected:**
The filtered query returns the matching row, as it does on the legacy source.
**Steps to reproduce:**
```sql
CREATE TABLE t1 (
uuid VARCHAR(20),
name VARCHAR(10),
age INT,
ts TIMESTAMP(3),
`partition` VARCHAR(20),
PRIMARY KEY (uuid) NOT ENFORCED
) PARTITIONED BY (`partition`) WITH (
'connector' = 'hudi',
'path' = '/tmp/hudi-source-v2-bucket-pruning',
'ordering.fields' = 'ts',
'index.type' = 'BUCKET',
'read.source-v2.enabled' = 'true'
);
INSERT INTO t1 VALUES
('id1', 'Danny', 23, TIMESTAMP '1970-01-01 00:00:01', 'par1'),
('id2', 'Stephen', 33, TIMESTAMP '1970-01-01 00:00:02', 'par1'),
('id3', 'Julian', 53, TIMESTAMP '1970-01-01 00:00:03', 'par2'),
('id4', 'Fabian', 31, TIMESTAMP '1970-01-01 00:00:04', 'par2'),
('id5', 'Sophia', 18, TIMESTAMP '1970-01-01 00:00:05', 'par3'),
('id6', 'Emma', 20, TIMESTAMP '1970-01-01 00:00:06', 'par3'),
('id7', 'Bob', 44, TIMESTAMP '1970-01-01 00:00:07', 'par4'),
('id8', 'Han', 56, TIMESTAMP '1970-01-01 00:00:08', 'par4');
SELECT * FROM t1; -- returns all 8 rows
SELECT * FROM t1 WHERE uuid = 'id5'; -- fails with NotSerializableException
```
Reproducing test: the existing `ITTestHoodieDataSource#testBucketPruning`, also run with Source V2. Its Source V2 cases fail on master for both table types at the first filtered query, and its legacy cases pass.
```java
@ParameterizedTest
@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);
execInsertSql(tableEnv, TestSQL.INSERT_T1);
List<Row> result1 = CollectionUtil.iterableToList(
() -> tableEnv.sqlQuery("select * from t1").execute().collect());
assertRowsEquals(result1, TestData.DATA_SET_SOURCE_INSERT);
// apply filters
List<Row> result2 = CollectionUtil.iterableToList(
() -> tableEnv.sqlQuery("select * from t1 where uuid = 'id5' and age < 20").execute().collect());
assertRowsEquals(result2, "[+I[id5, Sophia, 18, 1970-01-01T00:00:05, par3]]");
// filter by timestamp
List<Row> result3 = CollectionUtil.iterableToList(
() -> tableEnv.sqlQuery("select * from t1 where uuid = 'id7' and ts > TIMESTAMP '1970-01-01 00:00:05'").execute().collect());
assertRowsEquals(result3, "[+I[id7, Bob, 44, 1970-01-01T00:00:07, par4]]");
// filter by in expression
List<Row> result4 = CollectionUtil.iterableToList(
() -> tableEnv.sqlQuery("select * from t1 where name in ('Danny', 'Julian') and uuid='id1'").execute().collect());
assertRowsEquals(result4, "[+I[id1, Danny, 23, 1970-01-01T00:00:01, par1]]");
}
```
### Environment
**Hudi version:** master (1.3.0-SNAPSHOT) at `092e4c40c504`, and 1.2.1. By code inspection 1.2.0 is not affected: its source holds the split reader function itself rather than a supplier that captures the table source, and it has no bucket pruning.
**Query engine:** Flink 1.20, batch and streaming SQL
**Relevant configs:** `read.source-v2.enabled=true`, `index.type=BUCKET`
### Logs and Stack Trace
Exception chain from the reproducing test on master (stack frames omitted), with the serialization path printed by `-Dsun.io.serialization.extendedDebugInfo=true`:
```
org.apache.flink.table.api.TableException: Failed to execute sql
Caused by: org.apache.flink.util.FlinkRuntimeException: Error in serialization.
Caused by: java.util.concurrent.ExecutionException: org.apache.flink.streaming.runtime.tasks.StreamTaskException: Could not serialize object for key serializedUDF.
Caused by: org.apache.flink.streaming.runtime.tasks.StreamTaskException: Could not serialize object for key serializedUDF.
Caused by: java.io.NotSerializableException: org.apache.hudi.source.prune.PrimaryKeyPruners$$Lambda$4471/0x000074b6912e0d60
- field (class "org.apache.hudi.common.util.Option", name: "val", type: "class java.lang.Object")
- object (class "org.apache.hudi.common.util.Option", Option{val=org.apache.hudi.source.prune.PrimaryKeyPruners$$Lambda$4471/0x000074b6912e0d60@74e8040b})
- field (class "org.apache.hudi.table.HoodieTableSource", name: "dataBucketFunc", type: "class org.apache.hudi.common.util.Option")
- object (class "org.apache.hudi.table.HoodieTableSource", org.apache.hudi.table.HoodieTableSource@7ff1f88f)
- element of array (index: 0)
- array (class "[Ljava.lang.Object;", size: 4)
- field (class "java.lang.invoke.SerializedLambda", name: "capturedArgs", type: "class [Ljava.lang.Object;")
- object (class "java.lang.invoke.SerializedLambda", SerializedLambda[capturingClass=class org.apache.hudi.table.HoodieTableSource, functionalInterfaceMethod=org/apache/hudi/common/function/SerializableSupplier.get:()Ljava/lang/Object;, implementation=invokeSpecial org/apache/hudi/table/HoodieTableSource.lambda$createHoodieSource$ccdb1231$1:(Lorg/apache/hudi/common/schema/HoodieSchema;Lorg/apache/hudi/common/schema/HoodieSchema;Z)Lorg/apache/hudi/source/reader/function/SplitReaderFunction;, instantiatedMethodType=()Lorg/apache/hudi/source/reader/function/SplitReaderFunction;, numCaptured=4])
- field (class "org.apache.hudi.source.HoodieSource", name: "readerFunctionSupplier", type: "interface org.apache.hudi.common.function.SerializableSupplier")
- object (class "org.apache.hudi.source.HoodieSource", org.apache.hudi.source.HoodieSource@1388e6f4)
- field (class "org.apache.flink.streaming.api.operators.SourceOperatorFactory", name: "source", type: "interface org.apache.flink.api.connector.source.Source")
- root object (class "org.apache.flink.streaming.api.operators.SourceOperatorFactory", org.apache.flink.streaming.api.operators.SourceOperatorFactory@64a5fdef)
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] [BUG] flink: Source V2 query filtering a bucket index table by its keys fails with NotSerializableException [hudi]
Posted by "danny0405 (via GitHub)" <gi...@apache.org>.
danny0405 closed issue #20317: [BUG] flink: Source V2 query filtering a bucket index table by its keys fails with NotSerializableException
URL: https://github.com/apache/hudi/issues/20317
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org