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 316c7df2839e fix(flink): respect data skipping settings for record
index pruning (#20502)
316c7df2839e is described below
commit 316c7df2839e28aea462ecb9b6d1b60ca8102655
Author: Shuo Cheng <[email protected]>
AuthorDate: Wed Nov 4 10:08:42 2026 +0800
fix(flink): respect data skipping settings for record index pruning (#20502)
---
.../java/org/apache/hudi/source/FileIndex.java | 2 +-
.../java/org/apache/hudi/source/TestFileIndex.java | 50 ++++++++++++++++++++++
2 files changed, 51 insertions(+), 1 deletion(-)
diff --git
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/FileIndex.java
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/FileIndex.java
index 1248e5078969..8feb50a9c6c1 100644
---
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/FileIndex.java
+++
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/FileIndex.java
@@ -93,7 +93,7 @@ public class FileIndex implements Serializable, AutoCloseable
{
this.partitionPruner = Option.ofNullable(partitionPruner);
this.fileStatsIndex = new FileStatsIndex(path.toString(), rowType, conf,
metaClient);
this.partitionBucketIdFunc = partitionBucketIdFunc;
- List<ExpressionEvaluators.Evaluator> evaluators =
Option.ofNullable(colStatsProbe).map(ColumnStatsProbe::getEvaluators).orElse(Collections.emptyList());
+ List<ExpressionEvaluators.Evaluator> evaluators =
Option.ofNullable(this.colStatsProbe).map(ColumnStatsProbe::getEvaluators).orElse(Collections.emptyList());
this.recordLevelIndex = BaseRecordLevelIndex.create(path.toString(), conf,
metaClient, evaluators, rowType);
this.metaClient = metaClient;
}
diff --git
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/source/TestFileIndex.java
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/source/TestFileIndex.java
index 3ad096a96870..8222a586ee6d 100644
---
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/source/TestFileIndex.java
+++
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/source/TestFileIndex.java
@@ -29,6 +29,7 @@ import org.apache.hudi.configuration.FlinkOptions;
import org.apache.hudi.keygen.NonpartitionedAvroKeyGenerator;
import org.apache.hudi.source.prune.ColumnStatsProbe;
import org.apache.hudi.source.prune.PartitionPruners;
+import org.apache.hudi.source.stats.BaseRecordLevelIndex;
import org.apache.hudi.source.stats.FileStatsIndex;
import org.apache.hudi.storage.StoragePath;
import org.apache.hudi.storage.StoragePathInfo;
@@ -89,6 +90,7 @@ import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyList;
import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.CALLS_REAL_METHODS;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.mockConstruction;
import static org.mockito.Mockito.mockStatic;
@@ -355,6 +357,54 @@ public class TestFileIndex {
assertEquals(Arrays.asList("par3"), p);
}
+ @Test
+ void testRecordLevelIndexRespectsDataSkipping() throws Exception {
+ DataType dataType = TestConfigurations.ROW_DATA_TYPE_WITH_ATOMIC_TYPES;
+ Configuration conf =
TestConfigurations.getDefaultConf(tempFile.getAbsolutePath(), dataType);
+ conf.set(FlinkOptions.TABLE_TYPE, FlinkOptions.TABLE_TYPE_COPY_ON_WRITE);
+ conf.set(FlinkOptions.METADATA_ENABLED, true);
+ conf.set(FlinkOptions.RECORD_KEY_FIELD, "f_str");
+
conf.setString(HoodieMetadataConfig.GLOBAL_RECORD_LEVEL_INDEX_ENABLE_PROP.key(),
"true");
+ TestData.writeData(TestData.DATA_SET_WITH_ATOMIC_TYPES, conf);
+
+ HoodieTableMetaClient metaClient = StreamerUtil.createMetaClient(conf);
+ CallExpression filter = CallExpression.permanent(
+ BuiltInFunctionDefinitions.EQUALS,
+ Arrays.asList(
+ new FieldReferenceExpression("f_str", DataTypes.STRING(), 0, 0),
+ new ValueLiteralExpression("str1", DataTypes.STRING().notNull())),
+ DataTypes.BOOLEAN());
+ ColumnStatsProbe probe =
ColumnStatsProbe.newInstance(Collections.singletonList(filter));
+
+ for (boolean dataSkipping : new boolean[] {true, false}) {
+ conf.set(FlinkOptions.READ_DATA_SKIPPING_ENABLED, dataSkipping);
+ try (FileIndex fileIndex = FileIndex.builder()
+ .path(new StoragePath(tempFile.getAbsolutePath()))
+ .conf(conf)
+ .rowType((RowType) dataType.getLogicalType())
+ .metaClient(metaClient)
+ .columnStatsProbe(probe)
+ .build()) {
+ assertThat(getFilteredFileSlices(metaClient, fileIndex).size(),
is(dataSkipping ? 1 : 3));
+ }
+ }
+
+ conf.set(FlinkOptions.READ_DATA_SKIPPING_ENABLED, true);
+ conf.set(FlinkOptions.METADATA_ENABLED, false);
+ try (MockedStatic<BaseRecordLevelIndex> recordIndex =
mockStatic(BaseRecordLevelIndex.class, CALLS_REAL_METHODS);
+ FileIndex fileIndex = FileIndex.builder()
+ .path(new StoragePath(tempFile.getAbsolutePath()))
+ .conf(conf)
+ .rowType((RowType) dataType.getLogicalType())
+ .metaClient(metaClient)
+ .columnStatsProbe(probe)
+ .build()) {
+ recordIndex.verify(() -> BaseRecordLevelIndex.create(
+ eq(tempFile.getAbsolutePath()), eq(conf), eq(metaClient),
eq(Collections.emptyList()), eq((RowType) dataType.getLogicalType())));
+ assertThat(getFilteredFileSlices(metaClient, fileIndex).size(), is(3));
+ }
+ }
+
@ParameterizedTest
@MethodSource("filtersAndResults")
void testFileListingWithRecordLevelIndex(String recordFields,
ColumnStatsProbe probe, int maxKeyCnt, int expectedCnt) throws Exception {
