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 276d120b839a fix(flink): retain files with missing column statistics
(#20295)
276d120b839a is described below
commit 276d120b839a3d7b6ec64bf32b9132fab9f88a6d
Author: Shuo Cheng <[email protected]>
AuthorDate: Wed Oct 14 12:57:32 2026 +0800
fix(flink): retain files with missing column statistics (#20295)
* fix(flink): retain files with missing column statistics
Preserve unknown column statistics during transposition and retain
candidates with missing statistics before evaluating predicates. Add probe,
index expansion, and SQL integration regression coverage.
* test(flink): cover incomplete statistics in partition pruning
Document conservative file retention when another conjunct excludes the
file, and verify partition candidates after expanding indexed columns.
---
.../apache/hudi/source/prune/ColumnStatsProbe.java | 8 ++
.../apache/hudi/source/stats/FileStatsIndex.java | 7 +-
.../hudi/source/prune/TestColumnStatsProbe.java | 28 +++++
.../hudi/source/stats/TestColumnStatsIndex.java | 115 +++++++++++++++++++++
.../apache/hudi/table/ITTestHoodieDataSource.java | 35 +++++++
5 files changed, 189 insertions(+), 4 deletions(-)
diff --git
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/prune/ColumnStatsProbe.java
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/prune/ColumnStatsProbe.java
index 5fbb9ff938fa..6769128cd93e 100644
---
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/prune/ColumnStatsProbe.java
+++
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/prune/ColumnStatsProbe.java
@@ -60,6 +60,14 @@ public class ColumnStatsProbe implements Serializable {
* @return true if the index row should be considered as a candidate
*/
public boolean test(RowData indexRow, RowType.RowField[] queryFields) {
+ // Retain candidates with missing column statistics rather than treating
unknown values as nulls.
+ for (int i = 0; i < queryFields.length; i++) {
+ // The row starts with file name and value count, followed by (min, max,
null count) per column.
+ // Skip the two leading fields and i triples, then use offset 2 to check
this column's null count.
+ if (indexRow.isNullAt(2 + i * 3 + 2)) {
+ return true;
+ }
+ }
Map<String, ColumnStats> columnStatsMap = convertColumnStats(indexRow,
queryFields);
for (ExpressionEvaluators.Evaluator evaluator : evaluators) {
if (!evaluator.eval(columnStatsMap)) {
diff --git
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/stats/FileStatsIndex.java
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/stats/FileStatsIndex.java
index 99c22402a02f..a18d52ff6a5d 100644
---
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/stats/FileStatsIndex.java
+++
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/stats/FileStatsIndex.java
@@ -306,10 +306,9 @@ public class FileStatsIndex implements ColumnStatsIndex {
List<Tuple3> columnStats =
alignedColumnRowsMap.values().stream().map(row -> {
if (row == null) {
- // NOTE: Since we're assuming missing column to essentially contain
exclusively
- // null values, we set null-count to be equal to value-count
(this behavior is
- // consistent with reading non-existent columns from Parquet)
- return Tuple3.of(null, null, valueCount);
+ // Missing statistics do not imply all-null data. A null null-count
marks
+ // unknown statistics so the probe can conservatively retain this
candidate.
+ return Tuple3.of(null, null, null);
} else {
GenericRowData gr = (GenericRowData) row;
return Tuple3.of(gr.getField(ORD_MIN_VAL), gr.getField(ORD_MAX_VAL),
gr.getField(ORD_NULL_CNT));
diff --git
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/source/prune/TestColumnStatsProbe.java
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/source/prune/TestColumnStatsProbe.java
index 01e302acc5ff..b11aec662eac 100644
---
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/source/prune/TestColumnStatsProbe.java
+++
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/source/prune/TestColumnStatsProbe.java
@@ -23,14 +23,21 @@ import org.apache.hudi.source.stats.ColumnStats;
import org.apache.hudi.utils.TestData;
import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.data.GenericRowData;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.data.StringData;
+import org.apache.flink.table.expressions.CallExpression;
+import org.apache.flink.table.expressions.FieldReferenceExpression;
+import org.apache.flink.table.expressions.ValueLiteralExpression;
+import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
import org.apache.flink.table.types.DataType;
import org.apache.flink.table.types.logical.RowType;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.Arguments;
import org.junit.jupiter.params.provider.MethodSource;
+import java.util.Arrays;
+import java.util.Collections;
import java.util.Map;
import java.util.stream.Stream;
@@ -54,6 +61,27 @@ class TestColumnStatsProbe {
assertEquals(maxValue, stats1.get("field").getMaxVal());
}
+ @ParameterizedTest
+ @MethodSource("columnStatsCompleteness")
+ void testProbeWithIncompleteColumnStats(Integer minValue, Integer maxValue,
Long nullCount, boolean expected) {
+ CallExpression filter =
CallExpression.permanent(BuiltInFunctionDefinitions.EQUALS,
+ Arrays.asList(new FieldReferenceExpression("field", DataTypes.INT(),
0, 0), new ValueLiteralExpression(3)),
+ DataTypes.BOOLEAN());
+ ColumnStatsProbe probe =
ColumnStatsProbe.newInstance(Collections.singletonList(filter));
+ RowData indexRow = GenericRowData.of(StringData.fromString("f1"), 10L,
minValue, maxValue, nullCount);
+ RowType.RowField[] fields =
getDataFields(getDataRowDataType(DataTypes.INT()));
+ assertEquals(expected, probe.test(indexRow, fields));
+ }
+
+ static Stream<Arguments> columnStatsCompleteness() {
+ return Stream.of(
+ arguments(null, null, null, true), // Missing statistics must retain
the file.
+ arguments(null, null, 10L, false), // Known all-null data cannot match
equality.
+ arguments(1, 5, 0L, true),
+ arguments(4, 5, 0L, false)
+ );
+ }
+
static Stream<Arguments> testTypes() {
return Stream.of(
arguments(DataTypes.INT(), 1, 5),
diff --git
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/source/stats/TestColumnStatsIndex.java
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/source/stats/TestColumnStatsIndex.java
index 02c6f9cea77b..59e0f8c26fe0 100644
---
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/source/stats/TestColumnStatsIndex.java
+++
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/source/stats/TestColumnStatsIndex.java
@@ -28,14 +28,22 @@ import org.apache.hudi.utils.TestConfigurations;
import org.apache.hudi.utils.TestData;
import org.apache.flink.configuration.Configuration;
+import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.data.GenericRowData;
import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.StringData;
+import org.apache.flink.table.data.TimestampData;
+import org.apache.flink.table.expressions.CallExpression;
+import org.apache.flink.table.expressions.FieldReferenceExpression;
+import org.apache.flink.table.expressions.ValueLiteralExpression;
+import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.ValueSource;
import java.io.File;
+import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.Comparator;
@@ -164,6 +172,113 @@ public class TestColumnStatsIndex {
}
}
+ @Test
+ void testCandidateFilesWithMissingColumnStatsAfterIndexExpansion() throws
Exception {
+ final String path = tempFile.getAbsolutePath();
+ Configuration conf = TestConfigurations.getDefaultConf(path);
+ conf.set(FlinkOptions.TABLE_TYPE, "COPY_ON_WRITE");
+ conf.set(FlinkOptions.METADATA_ENABLED, true);
+ conf.set(FlinkOptions.READ_DATA_SKIPPING_ENABLED, true);
+
conf.setString(HoodieMetadataConfig.ENABLE_METADATA_INDEX_COLUMN_STATS.key(),
"true");
+
conf.setString(HoodieMetadataConfig.ENABLE_METADATA_INDEX_PARTITION_STATS.key(),
"false");
+ conf.setString(HoodieMetadataConfig.COLUMN_STATS_INDEX_FOR_COLUMNS.key(),
"age");
+
+ // The old file contains a non-null name, but only age has column
statistics.
+ TestData.writeData(Collections.singletonList(TestData.insertRow(
+ StringData.fromString("id1"), StringData.fromString("alice"), 25,
+ TimestampData.fromEpochMillis(1), StringData.fromString("par1"))),
conf);
+
+ // Write a different partition so expanding the index does not rewrite the
old file.
+ conf.setString(HoodieMetadataConfig.COLUMN_STATS_INDEX_FOR_COLUMNS.key(),
"age,name");
+ TestData.writeData(Collections.singletonList(TestData.insertRow(
+ StringData.fromString("id2"), StringData.fromString("bob"), 30,
+ TimestampData.fromEpochMillis(2), StringData.fromString("par2"))),
conf);
+
+ try (FileStatsIndex index = new FileStatsIndex(path,
TestConfigurations.ROW_TYPE, conf, StreamerUtil.createMetaClient(conf))) {
+ String[] columns = {"age", "name"};
+ List<RowData> oldStats = index.readColumnStatsIndexByColumns(columns,
Collections.singletonList("par1"));
+ assertEquals(1, oldStats.size());
+ assertEquals("age",
oldStats.get(0).getString(ColumnStatsSchemas.ORD_COL_NAME).toString());
+ String oldFile =
oldStats.get(0).getString(ColumnStatsSchemas.ORD_FILE_NAME).toString();
+
+ List<RowData> allStats = index.readColumnStatsIndexByColumns(columns,
Collections.emptyList());
+ assertEquals(3, allStats.size());
+ List<String> allFiles = allStats.stream()
+ .map(row ->
row.getString(ColumnStatsSchemas.ORD_FILE_NAME).toString()).distinct().collect(Collectors.toList());
+ assertEquals(2, allFiles.size());
+
+ CallExpression ageFilter = CallExpression.permanent(
+ BuiltInFunctionDefinitions.GREATER_THAN_OR_EQUAL,
+ Arrays.asList(new FieldReferenceExpression("age", DataTypes.INT(),
0, 2), new ValueLiteralExpression(0)),
+ DataTypes.BOOLEAN());
+ CallExpression nameFilter = CallExpression.permanent(
+ BuiltInFunctionDefinitions.EQUALS,
+ Arrays.asList(new FieldReferenceExpression("name",
DataTypes.STRING(), 0, 1), new ValueLiteralExpression("alice")),
+ DataTypes.BOOLEAN());
+
+ // With only name referenced, the old file is retained by the
unindexed-file fallback.
+ assertEquals(Collections.singleton(oldFile), index.computeCandidateFiles(
+ ColumnStatsProbe.newInstance(Collections.singletonList(nameFilter)),
new ArrayList<>(allFiles), Collections.emptyList()));
+
+ // Adding a matching age predicate must not turn missing name statistics
into all-null data.
+ assertEquals(Collections.singleton(oldFile), index.computeCandidateFiles(
+ ColumnStatsProbe.newInstance(Arrays.asList(ageFilter, nameFilter)),
new ArrayList<>(allFiles), Collections.emptyList()));
+
+ CallExpression nonMatchingAgeFilter = CallExpression.permanent(
+ BuiltInFunctionDefinitions.GREATER_THAN,
+ Arrays.asList(new FieldReferenceExpression("age", DataTypes.INT(),
0, 2), new ValueLiteralExpression(100)),
+ DataTypes.BOOLEAN());
+ assertEquals(Collections.emptySet(), index.computeCandidateFiles(
+
ColumnStatsProbe.newInstance(Collections.singletonList(nonMatchingAgeFilter)),
new ArrayList<>(allFiles), Collections.emptyList()));
+ // Missing name statistics conservatively retain the old file even when
age alone excludes it.
+ // The fully indexed new file can still be pruned.
+ assertEquals(Collections.singleton(oldFile), index.computeCandidateFiles(
+ ColumnStatsProbe.newInstance(Arrays.asList(nonMatchingAgeFilter,
nameFilter)), new ArrayList<>(allFiles), Collections.emptyList()));
+ }
+ }
+
+ @Test
+ void testCandidatePartitionsWithMissingColumnStatsAfterIndexExpansion()
throws Exception {
+ final String path = tempFile.getAbsolutePath();
+ Configuration conf = TestConfigurations.getDefaultConf(path);
+ conf.set(FlinkOptions.TABLE_TYPE, "COPY_ON_WRITE");
+ conf.set(FlinkOptions.METADATA_ENABLED, true);
+ conf.set(FlinkOptions.READ_DATA_SKIPPING_ENABLED, true);
+
conf.setString(HoodieMetadataConfig.ENABLE_METADATA_INDEX_COLUMN_STATS.key(),
"true");
+
conf.setString(HoodieMetadataConfig.ENABLE_METADATA_INDEX_PARTITION_STATS.key(),
"true");
+ conf.setString(HoodieMetadataConfig.COLUMN_STATS_INDEX_FOR_COLUMNS.key(),
"age");
+ TestData.writeData(Collections.singletonList(TestData.insertRow(
+ StringData.fromString("id1"), StringData.fromString("alice"), 25,
+ TimestampData.fromEpochMillis(1), StringData.fromString("par1"))),
conf);
+
+ conf.setString(HoodieMetadataConfig.COLUMN_STATS_INDEX_FOR_COLUMNS.key(),
"age,name");
+ TestData.writeData(Collections.singletonList(TestData.insertRow(
+ StringData.fromString("id2"), StringData.fromString("bob"), 30,
+ TimestampData.fromEpochMillis(2), StringData.fromString("par2"))),
conf);
+
+ try (PartitionStatsIndex index = new PartitionStatsIndex(path,
TestConfigurations.ROW_TYPE, conf, StreamerUtil.createMetaClient(conf))) {
+ List<RowData> stats = index.readColumnStatsIndexByColumns(new String[]
{"age", "name"}, Collections.emptyList());
+ List<String> actualStats =
stats.stream().map(Object::toString).sorted().collect(Collectors.toList());
+ // Verify the old partition is partially indexed, rather than entirely
absent from the index.
+ assertEquals(Arrays.asList(
+ "+I(par1,+I(25),+I(25),0,1,age)",
+ "+I(par2,+I(30),+I(30),0,1,age)",
+ "+I(par2,+I(bob),+I(bob),0,1,name)"), actualStats);
+
+ CallExpression ageFilter = CallExpression.permanent(
+ BuiltInFunctionDefinitions.GREATER_THAN_OR_EQUAL,
+ Arrays.asList(new FieldReferenceExpression("age", DataTypes.INT(),
0, 2), new ValueLiteralExpression(0)),
+ DataTypes.BOOLEAN());
+ CallExpression nameFilter = CallExpression.permanent(
+ BuiltInFunctionDefinitions.EQUALS,
+ Arrays.asList(new FieldReferenceExpression("name",
DataTypes.STRING(), 0, 1), new ValueLiteralExpression("alice")),
+ DataTypes.BOOLEAN());
+ // Partition pruning must retain the matching old partition before file
pruning runs.
+ assertEquals(Collections.singleton("par1"),
index.computeCandidatePartitions(
+ ColumnStatsProbe.newInstance(Arrays.asList(ageFilter, nameFilter)),
new ArrayList<>(Arrays.asList("par1", "par2"))));
+ }
+ }
+
private static List<RowData> filterOutFileNames(List<RowData> indexRows) {
return indexRows.stream().map(row -> {
GenericRowData gr = (GenericRowData) row;
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 ef017d7ed088..69b22e266b6a 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
@@ -2942,6 +2942,41 @@ public class ITTestHoodieDataSource {
+ "+I[id3, id3, Julian, 43, 1970-01-01T00:00:03, par1]]");
}
+ @Test
+ void testDataSkippingWithMissingColumnStatsAfterIndexExpansion() throws
Exception {
+ TableEnvironment tableEnv = batchTableEnv;
+ String hoodieTableDDL = sql("t1")
+ .option(FlinkOptions.PATH, tempFile.getAbsolutePath())
+ .options(getDefaultKeys())
+ .option(FlinkOptions.TABLE_TYPE, COPY_ON_WRITE)
+ .option(FlinkOptions.METADATA_ENABLED, true)
+ .option(HoodieMetadataConfig.ENABLE_METADATA_INDEX_COLUMN_STATS.key(),
true)
+
.option(HoodieMetadataConfig.ENABLE_METADATA_INDEX_PARTITION_STATS.key(), false)
+ .option(HoodieMetadataConfig.COLUMN_STATS_INDEX_FOR_COLUMNS.key(),
"age")
+ .option(FlinkOptions.READ_DATA_SKIPPING_ENABLED, false)
+ .end();
+ tableEnv.executeSql(hoodieTableDDL);
+
+ // The name column exists in the old data file, but only age is indexed.
+ tableEnv.executeSql("insert into t1 values "
+ + "('id1', 'alice', 25, TIMESTAMP '1970-01-01 00:00:01',
'par1')").await();
+
+ tableEnv.executeSql("alter table t1 set
('hoodie.metadata.index.column.stats.column.list' = 'age,name')");
+ // A new partition leaves the old file and its incomplete column
statistics untouched.
+ tableEnv.executeSql("insert into t1 values "
+ + "('id2', 'bob', 30, TIMESTAMP '1970-01-01 00:00:02',
'par2')").await();
+
+ String query = "select * from t1 where age >= 0 and name = 'alice'";
+ String expected = "[+I[id1, alice, 25, 1970-01-01T00:00:01, par1]]";
+ List<Row> withoutSkipping = execSelectSql(tableEnv, query);
+ assertRowsEquals(withoutSkipping, expected);
+
+ tableEnv.executeSql("alter table t1 set ('read.data.skipping.enabled' =
'true')");
+ assertRowsEquals(execSelectSql(tableEnv, "select * from t1 where name =
'alice'"), expected);
+ // Adding a matching age predicate must not cause the old file to be
skipped.
+ assertRowsEquals(execSelectSql(tableEnv, query), expected);
+ }
+
@ParameterizedTest
@EnumSource(value = HoodieTableType.class)
void testDataSkippingByFilteringFileSlice(HoodieTableType tableType) {
