cshuo opened a new issue, #20289:
URL: https://github.com/apache/hudi/issues/20289
### Description
Flink data skipping can silently drop matching rows after expanding the
Column Stats Index (CSI) column list. An existing file may contain a non-null
value for a newly indexed column while still lacking statistics for that
column. When the query references both an originally indexed column and the
newly indexed column, the old file is incorrectly pruned.
This was reproduced through SQL writes and reads in Flink MiniCluster using
a new local regression test,
`ITTestHoodieDataSource.testDataSkippingWithMissingColumnStatsAfterIndexExpansion`.
### Environment
- Hudi: `master` at `3aa7fe2a9f70ef52064c977d28d28b4f4c95d00b`
(1.3.0-SNAPSHOT).
- Query engine: Flink 1.20.1, batch TableEnvironment.
- Table type: COPY_ON_WRITE.
- Metadata and CSI enabled; partition statistics disabled to isolate
file-level CSI pruning.
- Validation: the targeted JUnit IT was run with locally built dependencies
and relevant sources recompiled for Flink 1.20.1. The full Maven/IT suite was
not run.
### Steps to reproduce
Run the following in batch mode, using a fresh table path. Wait for each
INSERT job to finish before continuing.
```sql
SET 'execution.runtime-mode' = 'batch';
CREATE TABLE t1 (
uuid VARCHAR(20),
name VARCHAR(10),
age INT,
ts TIMESTAMP(3),
`partition` VARCHAR(10),
PRIMARY KEY (uuid) NOT ENFORCED
) PARTITIONED BY (`partition`) WITH (
'connector' = 'hudi',
'path' = '/tmp/hudi-csi-index-expansion',
'table.type' = 'COPY_ON_WRITE',
'hoodie.datasource.write.recordkey.field' = 'uuid',
'ordering.fields' = 'ts',
'metadata.enabled' = 'true',
'hoodie.metadata.index.column.stats.enable' = 'true',
'hoodie.metadata.index.partition.stats.enable' = 'false',
'hoodie.metadata.index.column.stats.column.list' = 'age',
'read.data.skipping.enabled' = 'false'
);
-- name exists in the data from the first write; only its statistics are
absent.
INSERT INTO t1 VALUES
('id1', 'alice', 25, TIMESTAMP '1970-01-01 00:00:01', 'par1');
ALTER TABLE t1 SET (
'hoodie.metadata.index.column.stats.column.list' = 'age,name'
);
-- A different partition leaves the old file untouched.
INSERT INTO t1 VALUES
('id2', 'bob', 30, TIMESTAMP '1970-01-01 00:00:02', 'par2');
-- Returns the alice row.
SELECT * FROM t1 WHERE age >= 0 AND name = 'alice';
ALTER TABLE t1 SET ('read.data.skipping.enabled' = 'true');
-- Still returns the alice row.
SELECT * FROM t1 WHERE name = 'alice';
-- Incorrectly returns no rows.
SELECT * FROM t1 WHERE age >= 0 AND name = 'alice';
```
### Expected and actual behavior
All three SELECTs above should return:
```text
+I[id1, alice, 25, 1970-01-01T00:00:01, par1]
```
The final SELECT returns an empty result. Adding the matching `age >= 0`
predicate changes the result incorrectly when data skipping is enabled.
The new local IT passes the skipping-disabled assertion and the name-only
assertion, then fails on the final combined-predicate assertion:
```text
Expected: is "[+I[id1, alice, 25, 1970-01-01T00:00:01, par1]]"
but: was "[]"
```
A separate index-level regression test,
`TestColumnStatsIndex.testCandidateFilesWithMissingColumnStatsAfterIndexExpansion`,
also verifies the metadata state: two files have three column-statistics
records, with only `age` recorded for the old file. Its combined-predicate
candidate set is empty. These regression tests are currently local and have not
been published in a PR.
### Root cause
1.
[`FileStatsIndex.foldRowsByFiles`](https://github.com/apache/hudi/blob/3aa7fe2a9f70ef52064c977d28d28b4f4c95d00b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/stats/FileStatsIndex.java#L307-L312)
pads a missing column-statistics record with `(null, null, valueCount)`,
treating it as though the data column contains only NULL values. Missing
statistics do not imply missing or all-null data.
2.
[`ExpressionEvaluators.EqualTo`](https://github.com/apache/hudi/blob/3aa7fe2a9f70ef52064c977d28d28b4f4c95d00b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/ExpressionEvaluators.java#L239-L248)
returns false for null min/max bounds. The old file therefore fails `name =
'alice'`. Changing only the padded null count would not address this comparison.
3.
[`candidatesInMetadataTable`](https://github.com/apache/hudi/blob/3aa7fe2a9f70ef52064c977d28d28b4f4c95d00b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/stats/FileStatsIndex.java#L185-L201)
adds back files absent from the transposed index. The old file already appears
because it has `age` statistics, so this fallback cannot recover it.
For the name-only query, the old file has no statistics in the requested
projection and is recovered by the unindexed-file fallback. Adding the age
predicate causes the file to appear in that projection and exposes the
incorrect pruning.
The reproduction queries both partitions so the fetched statistics include
the newer file's `name` statistics. It does not rely on schema evolution:
`name='alice'` is physically present in the old file.
### Suggested behavior
Preserve the distinction between unavailable statistics and known all-null
data. Retain a candidate whenever incomplete statistics prevent proving that
the query cannot match. A conservative fix can retain files with incomplete
statistics for the referenced columns; a more selective implementation can
propagate an explicit unknown-statistics state through predicate evaluation.
--
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]
