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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
