This is an automated email from the ASF dual-hosted git repository.
voonhous 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 0b696f915b67 fix(spark): retain matching files with missing column
statistics (#20328)
0b696f915b67 is described below
commit 0b696f915b6748b56ca114e21f7444489b60802c
Author: Shuo Cheng <[email protected]>
AuthorDate: Fri Nov 6 13:26:09 2026 +0800
fix(spark): retain matching files with missing column statistics (#20328)
* fix(spark): retain matching files with missing column statistics
* fix comments
* add tests
* test(spark): cover data skipping after adding a column
---
.../hudi/metadata/HoodieMetadataWriteUtils.java | 10 +-
.../partitionstats/PartitionStatsIndexer.java | 67 +++--
.../metadata/TestHoodieMetadataWriteUtils.java | 70 +++++
.../partitionstats/TestPartitionStatsIndexer.java | 91 ++++++
.../hudi/metadata/HoodieTableMetadataUtil.java | 95 +++---
.../hudi/metadata/TestHoodieTableMetadataUtil.java | 75 ++++-
.../hudi/metadata/TestHoodieTableMetadataUtil.java | 30 ++
.../apache/hudi/PartitionStatsIndexSupport.scala | 44 ++-
.../org/apache/hudi/SparkBaseIndexSupport.scala | 68 ++++-
.../spark/sql/hudi/dataskipping/TermRange.scala | 3 +-
.../org/apache/hudi/TestDataSkippingUtils.scala | 165 ++++++++++-
.../hudi/TestPartitionStatsIndexSupport.scala | 145 ++++++++++
.../sql/hudi/feature/TestDataSkippingQuery.scala | 317 +++++++++++++++++++++
13 files changed, 1027 insertions(+), 153 deletions(-)
diff --git
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/HoodieMetadataWriteUtils.java
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/HoodieMetadataWriteUtils.java
index 9ff6f1a06cc3..cdd4032bff6a 100644
---
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/HoodieMetadataWriteUtils.java
+++
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/HoodieMetadataWriteUtils.java
@@ -78,7 +78,6 @@ import
org.apache.hudi.core.io.storage.HoodieFileWriterFactory;
import org.apache.hudi.core.transaction.lock.InProcessLockProvider;
import org.apache.hudi.exception.HoodieException;
import org.apache.hudi.exception.HoodieMetadataException;
-import org.apache.hudi.metadata.stats.HoodieColumnRangeMetadata;
import org.apache.hudi.storage.HoodieStorage;
import org.apache.hudi.storage.StoragePath;
import org.apache.hudi.storage.StoragePathInfo;
@@ -101,7 +100,6 @@ import java.util.Set;
import java.util.stream.Collectors;
import java.util.stream.Stream;
-import static java.util.stream.Collectors.toList;
import static
org.apache.hudi.common.config.HoodieMetadataConfig.DEFAULT_METADATA_ASYNC_CLEAN;
import static
org.apache.hudi.common.config.HoodieMetadataConfig.DEFAULT_METADATA_CLEANER_COMMITS_RETAINED;
import static
org.apache.hudi.common.config.HoodieMetadataConfig.DEFAULT_METADATA_POPULATE_META_FIELDS;
@@ -109,7 +107,6 @@ import static
org.apache.hudi.common.model.LogExtensions.DATA_LOG_EXTENSION;
import static org.apache.hudi.common.util.StringUtils.nonEmpty;
import static org.apache.hudi.common.util.ValidationUtils.checkState;
import static
org.apache.hudi.metadata.HoodieTableMetadata.METADATA_TABLE_NAME_SUFFIX;
-import static
org.apache.hudi.metadata.HoodieTableMetadataUtil.translateWriteStatToFileStats;
/**
* Metadata table write utils.
@@ -506,18 +503,17 @@ public class HoodieMetadataWriteUtils {
tableMetadata, dataMetaClient, dataMetaClient.getActiveTimeline(),
consolidatedPathInfos, dataWriteConfig.getViewStorageConfig(),
dataWriteConfig.getCommonConfig());
- // Collect column metadata for each file part of the latest merged file
slice before the current instant time
- List<HoodieColumnRangeMetadata<Comparable>> fileColumnMetadata =
partitionedWriteStat.stream()
- .flatMap(writeStat -> translateWriteStatToFileStats(writeStat,
dataMetaClient, colsToIndex,
partitionStatsIndexVersion).stream()).collect(toList());
Set<String> fileGroupIdsToReplace =
fileGroupIdsToReplaceMap.getOrDefault(partitionName, Collections.emptySet());
Set<String> filesWithColumnStats = partitionedWriteStat.stream()
.map(stat -> new
StoragePath(stat.getPath()).getName()).collect(Collectors.toSet());
// Collect column metadata of each file that does not have column stats
provided by the write stat in the commit metadata
return
consolidatedFileSystemView.getLatestMergedFileSlicesBeforeOrOnIncludingInflight(partitionName,
maxInstantTime, instantTime)
+ // Inflight replacement commits can still expose the old groups in the
view.
+ .filter(fileSlice ->
!fileGroupIdsToReplace.contains(fileSlice.getFileId()))
.flatMap(fileSlice -> Stream.concat(
Stream.of(fileSlice.getBaseFile().map(HoodieBaseFile::getFileName).orElse(null)),
fileSlice.getLogFiles().map(HoodieLogFile::getFileName)))
- .filter(e -> Objects.nonNull(e) && !filesWithColumnStats.contains(e)
&& !fileGroupIdsToReplace.contains(e))
+ .filter(e -> Objects.nonNull(e) && !filesWithColumnStats.contains(e))
.collect(Collectors.toSet());
}
diff --git
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/index/partitionstats/PartitionStatsIndexer.java
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/index/partitionstats/PartitionStatsIndexer.java
index 035e3ce902af..73fe8bcba980 100644
---
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/index/partitionstats/PartitionStatsIndexer.java
+++
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/index/partitionstats/PartitionStatsIndexer.java
@@ -22,7 +22,6 @@ package org.apache.hudi.metadata.index.partitionstats;
import org.apache.hudi.common.config.HoodieMetadataConfig;
import org.apache.hudi.common.data.HoodieData;
import org.apache.hudi.common.data.HoodieListData;
-import org.apache.hudi.common.data.HoodiePairData;
import org.apache.hudi.common.engine.HoodieEngineContext;
import org.apache.hudi.common.model.HoodieCommitMetadata;
import org.apache.hudi.common.model.HoodieRecord;
@@ -36,7 +35,6 @@ import org.apache.hudi.common.table.HoodieTableMetaClient;
import org.apache.hudi.common.util.Lazy;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.common.util.VisibleForTesting;
-import org.apache.hudi.common.util.collection.Pair;
import org.apache.hudi.config.HoodieWriteConfig;
import org.apache.hudi.exception.HoodieException;
import org.apache.hudi.metadata.HoodieIndexVersion;
@@ -51,6 +49,7 @@ import
org.apache.hudi.metadata.index.model.IndexInitializationPlan;
import org.apache.hudi.metadata.index.model.IndexPartitionAndRecords;
import org.apache.hudi.metadata.index.model.IndexUpdateContext;
import org.apache.hudi.metadata.stats.HoodieColumnRangeMetadata;
+import org.apache.hudi.storage.StoragePath;
import lombok.extern.slf4j.Slf4j;
@@ -171,9 +170,8 @@ public class PartitionStatsIndexer extends BaseIndexer {
});
}
- // In this function we fetch column range metadata for all new files
part of commit metadata along with all the other files
- // of the affected partitions. The column range metadata is grouped by
partition name to generate HoodiePairData of partition name
- // and list of column range metadata for that partition files. This pair
data is then used to generate partition stat records.
+ // Gather statistics and the expected file inventory independently for
each affected partition.
+ // Missing file/column records must invalidate that column's partition
statistics.
List<HoodieWriteStat> allWriteStats =
commitMetadata.getPartitionToWriteStats().values().stream()
.flatMap(Collection::stream).collect(Collectors.toList());
if (allWriteStats.isEmpty()) {
@@ -194,36 +192,35 @@ public class PartitionStatsIndexer extends BaseIndexer {
int parallelism = Math.max(Math.min(partitionedWriteStats.size(),
metadataConfig.getPartitionStatsIndexParallelism()), 1);
String maxInstantTime = getMaxInstantTime(dataMetaClient, instantTime);
- HoodiePairData<String, List<HoodieColumnRangeMetadata<Comparable>>>
columnRangeMetadata =
- engineContext.parallelize(partitionedWriteStats,
parallelism).mapToPair(partitionedWriteStat -> {
- final String partitionName =
partitionedWriteStat.get(0).getPartitionPath();
- checkState(tableMetadata != null, "tableMetadata should not be
null when scanning metadata table");
-
- // Collect column metadata for each file part of the latest merged
file slice before the current instant time
- List<HoodieColumnRangeMetadata<Comparable>> fileColumnMetadata =
partitionedWriteStat.stream()
- .flatMap(writeStat -> translateWriteStatToFileStats(writeStat,
dataMetaClient, colsToIndex,
partitionStatsIndexVersion).stream()).collect(toList());
- // Collect column metadata of each file that does not have column
stats provided by the write stat in the commit metadata
- Set<String> filesToFetchColumnStats =
getFilesToFetchColumnStats(partitionedWriteStat, dataMetaClient, tableMetadata,
dataWriteConfig, partitionName, maxInstantTime,
- instantTime, fileGroupIdsToReplaceMap, colsToIndex,
partitionStatsIndexVersion);
- // Fetch metadata table COLUMN_STATS partition records for the
above files
- List<HoodieColumnRangeMetadata<Comparable>>
partitionColumnMetadata = tableMetadata
- .getRecordsByKeyPrefixes(
- HoodieListData.lazy(generateColumnStatsKeys(colsToIndex,
partitionName)),
- MetadataPartitionType.COLUMN_STATS.getPartitionPath(),
false)
- // schema and properties are ignored in getInsertValue, so
simply pass as null
- .map(record -> ((HoodieMetadataPayload)
record.getData()).getColumnStatMetadata())
- .filter(Option::isPresent)
- .map(colStatsOpt -> colStatsOpt.get())
- .filter(stats ->
filesToFetchColumnStats.contains(stats.getFileName()))
-
.map(HoodieColumnRangeMetadata::fromColumnStats).collectAsList();
- // fileColumnMetadata already contains stats for the files from
the current inflight commit.
- // Here it adds the stats for the committed files part of the
latest merged file slices
- fileColumnMetadata.addAll(partitionColumnMetadata);
- return Pair.of(partitionName, fileColumnMetadata);
- });
-
- return HoodieTableMetadataUtil.convertMetadataToPartitionStatsRecords(
- columnRangeMetadata, dataMetaClient, columnsToIndexSchemaMap,
partitionStatsIndexVersion);
+ return engineContext.parallelize(partitionedWriteStats,
parallelism).flatMap(partitionedWriteStat -> {
+ final String partitionName =
partitionedWriteStat.get(0).getPartitionPath();
+ checkState(tableMetadata != null, "tableMetadata should not be null
when scanning metadata table");
+
+ // Collect column metadata for each file part of the latest merged
file slice before the current instant time
+ List<HoodieColumnRangeMetadata<Comparable>> fileColumnMetadata =
partitionedWriteStat.stream()
+ .flatMap(writeStat -> translateWriteStatToFileStats(writeStat,
dataMetaClient, colsToIndex,
partitionStatsIndexVersion).stream()).collect(toList());
+ // Collect column metadata of each file that does not have column
stats provided by the write stat in the commit metadata
+ Set<String> filesToFetchColumnStats =
getFilesToFetchColumnStats(partitionedWriteStat, dataMetaClient, tableMetadata,
dataWriteConfig, partitionName, maxInstantTime,
+ instantTime, fileGroupIdsToReplaceMap, colsToIndex,
partitionStatsIndexVersion);
+ // Fetch metadata table COLUMN_STATS partition records for the above
files
+ List<HoodieColumnRangeMetadata<Comparable>> partitionColumnMetadata =
tableMetadata
+ .getRecordsByKeyPrefixes(
+ HoodieListData.lazy(generateColumnStatsKeys(colsToIndex,
partitionName)),
+ MetadataPartitionType.COLUMN_STATS.getPartitionPath(), false)
+ // schema and properties are ignored in getInsertValue, so simply
pass as null
+ .map(record -> ((HoodieMetadataPayload)
record.getData()).getColumnStatMetadata())
+ .filter(Option::isPresent)
+ .map(colStatsOpt -> colStatsOpt.get())
+ .filter(stats ->
filesToFetchColumnStats.contains(stats.getFileName()))
+ .map(HoodieColumnRangeMetadata::fromColumnStats).collectAsList();
+ // fileColumnMetadata already contains stats for the files from the
current inflight commit.
+ // Here it adds the stats for the committed files part of the latest
merged file slices
+ fileColumnMetadata.addAll(partitionColumnMetadata);
+ Set<String> expectedFileNames = new HashSet<>(filesToFetchColumnStats);
+ partitionedWriteStat.forEach(stat -> expectedFileNames.add(new
StoragePath(stat.getPath()).getName()));
+ return HoodieTableMetadataUtil.convertMetadataToPartitionStatsRecords(
+ partitionName, fileColumnMetadata, expectedFileNames,
columnsToIndexSchemaMap, partitionStatsIndexVersion).iterator();
+ });
} catch (Exception e) {
throw new HoodieException("Failed to generate partition stats records
for metadata table", e);
}
diff --git
a/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/metadata/TestHoodieMetadataWriteUtils.java
b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/metadata/TestHoodieMetadataWriteUtils.java
index abc5e5a42a77..f8b2ada5e969 100644
---
a/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/metadata/TestHoodieMetadataWriteUtils.java
+++
b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/metadata/TestHoodieMetadataWriteUtils.java
@@ -29,17 +29,24 @@ import
org.apache.hudi.common.engine.HoodieLocalEngineContext;
import org.apache.hudi.common.engine.TaskContextSupplier;
import org.apache.hudi.common.fs.FSUtils;
import org.apache.hudi.common.model.ActionType;
+import org.apache.hudi.common.model.FileSlice;
+import org.apache.hudi.common.model.HoodieBaseFile;
import org.apache.hudi.common.model.HoodieCleaningPolicy;
import org.apache.hudi.common.model.HoodieFailedWritesCleaningPolicy;
import org.apache.hudi.common.model.HoodieFileFormat;
import org.apache.hudi.common.model.HoodieLogFile;
+import org.apache.hudi.common.model.HoodieWriteStat;
import org.apache.hudi.common.model.WriteConcurrencyMode;
import org.apache.hudi.common.schema.HoodieSchema;
import org.apache.hudi.common.schema.HoodieSchemaUtils;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
import org.apache.hudi.common.table.HoodieTableVersion;
import org.apache.hudi.common.table.log.HoodieLogFormat;
import org.apache.hudi.common.table.log.NativeLogFooterMetadata;
import
org.apache.hudi.common.table.log.block.HoodieLogBlock.HeaderMetadataType;
+import org.apache.hudi.common.table.view.FileSystemViewManager;
+import org.apache.hudi.common.table.view.SpillableMapBasedFileSystemView;
+import org.apache.hudi.common.table.view.SyncableFileSystemView;
import org.apache.hudi.common.util.HoodieStorageUtils;
import org.apache.hudi.config.HoodieCleanConfig;
import org.apache.hudi.config.HoodieLockConfig;
@@ -57,11 +64,18 @@ import org.junit.jupiter.api.io.TempDir;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.EnumSource;
import org.junit.jupiter.params.provider.ValueSource;
+import org.mockito.MockedConstruction;
+import org.mockito.MockedStatic;
import java.nio.file.Path;
+import java.util.Arrays;
import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
import java.util.Map;
import java.util.Properties;
+import java.util.Set;
+import java.util.stream.Stream;
import static
org.apache.hudi.common.testutils.HoodieTestUtils.getDefaultStorageConf;
import static org.junit.jupiter.api.Assertions.assertEquals;
@@ -70,9 +84,65 @@ import static
org.junit.jupiter.api.Assertions.assertNotEquals;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.mockConstruction;
+import static org.mockito.Mockito.mockStatic;
+import static org.mockito.Mockito.when;
public class TestHoodieMetadataWriteUtils {
+ @Test
+ void testGetFilesToFetchColumnStatsExcludesReplacedFileGroups() {
+ String partition = "p1";
+ String instant = "002";
+ FileSlice replaced = new FileSlice(partition, "001", "replaced");
+ replaced.setBaseFile(new
HoodieBaseFile("/table/p1/replaced_1-0-1_001.parquet"));
+ replaced.addLogFile(new
HoodieLogFile("/table/p1/.replaced_001.log.1_1-0-1"));
+ FileSlice untouched = new FileSlice(partition, "001", "untouched");
+ untouched.setBaseFile(new
HoodieBaseFile("/table/p1/untouched_1-0-1_001.parquet"));
+ untouched.addLogFile(new
HoodieLogFile("/table/p1/.untouched_001.log.1_1-0-1"));
+ FileSlice current = new FileSlice(partition, instant, "current");
+ current.setBaseFile(new
HoodieBaseFile("/table/p1/current_1-0-1_002.parquet"));
+ current.addLogFile(new
HoodieLogFile("/table/p1/.current_002.log.1_1-0-1"));
+ HoodieWriteStat baseWrite = new HoodieWriteStat();
+ baseWrite.setPath(partition + "/" +
current.getBaseFile().get().getFileName());
+ HoodieWriteStat logWrite = new HoodieWriteStat();
+ logWrite.setPath(partition + "/" +
current.getLogFiles().findFirst().get().getFileName());
+ List<HoodieWriteStat> writes = Arrays.asList(baseWrite, logWrite);
+ HoodieTableMetaClient metaClient = mock(HoodieTableMetaClient.class);
+ when(metaClient.getBasePath()).thenReturn(new StoragePath("/table"));
+ doReturn(getDefaultStorageConf()).when(metaClient).getStorageConf();
+ HoodieWriteConfig config =
HoodieWriteConfig.newBuilder().withPath("/table").build();
+ HoodieTableMetadata metadata = mock(HoodieTableMetadata.class);
+ FileSystemViewManager manager = mock(FileSystemViewManager.class);
+ SyncableFileSystemView committedView = mock(SyncableFileSystemView.class);
+ when(manager.getFileSystemView(metaClient)).thenReturn(committedView);
+ when(committedView.getLatestMergedFileSlicesBeforeOrOn(partition, instant))
+ .thenReturn(Stream.of(replaced, untouched));
+
+ try (MockedStatic<FileSystemViewManager> viewManager =
mockStatic(FileSystemViewManager.class);
+ MockedStatic<HoodieTableMetadataUtil> metadataUtil =
mockStatic(HoodieTableMetadataUtil.class);
+ MockedConstruction<SpillableMapBasedFileSystemView> consolidatedViews
= mockConstruction(
+ SpillableMapBasedFileSystemView.class, (view, context) ->
+
when(view.getLatestMergedFileSlicesBeforeOrOnIncludingInflight(partition,
instant, instant))
+ .thenReturn(Stream.of(replaced, untouched, current)))) {
+ viewManager.when(() -> FileSystemViewManager.createViewManager(any(),
any(), any(), any(), any()))
+ .thenReturn(manager);
+ metadataUtil.when(() ->
HoodieTableMetadataUtil.translateWriteStatToFileStats(any(), any(), any(),
any()))
+ .thenReturn(Collections.emptyList());
+
+ Set<String> files = HoodieMetadataWriteUtils.getFilesToFetchColumnStats(
+ writes, metaClient, metadata, config, partition, instant, instant,
+ Collections.singletonMap(partition,
Collections.singleton(replaced.getFileId())),
+ Collections.singletonList("age"), HoodieIndexVersion.V1);
+
+ assertEquals(new HashSet<>(Arrays.asList("untouched_1-0-1_001.parquet",
".untouched_001.log.1_1-0-1")), files);
+ assertEquals(1, consolidatedViews.constructed().size());
+ }
+ }
+
@Test
void testCreateEmptyNativeLogFile(@TempDir Path tempDir) throws Exception {
String instantTime = "20260717120000000";
diff --git
a/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/metadata/index/partitionstats/TestPartitionStatsIndexer.java
b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/metadata/index/partitionstats/TestPartitionStatsIndexer.java
index 7c02f9c6e26f..6a734291923a 100644
---
a/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/metadata/index/partitionstats/TestPartitionStatsIndexer.java
+++
b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/metadata/index/partitionstats/TestPartitionStatsIndexer.java
@@ -5,8 +5,10 @@
package org.apache.hudi.metadata.index.partitionstats;
+import org.apache.hudi.avro.model.HoodieMetadataColumnStats;
import org.apache.hudi.common.config.HoodieMetadataConfig;
import org.apache.hudi.common.data.HoodieData;
+import org.apache.hudi.common.data.HoodieListData;
import org.apache.hudi.common.engine.HoodieEngineContext;
import org.apache.hudi.common.engine.HoodieLocalEngineContext;
import org.apache.hudi.common.model.HoodieCommitMetadata;
@@ -14,8 +16,11 @@ import org.apache.hudi.common.model.HoodieRecord;
import org.apache.hudi.common.model.HoodieRecordMerger;
import org.apache.hudi.common.model.HoodieWriteStat;
import org.apache.hudi.common.model.WriteOperationType;
+import org.apache.hudi.common.schema.HoodieSchema;
+import org.apache.hudi.common.schema.HoodieSchemaType;
import org.apache.hudi.common.table.HoodieTableConfig;
import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.table.HoodieTableVersion;
import org.apache.hudi.common.table.view.HoodieTableFileSystemView;
import org.apache.hudi.common.util.Lazy;
import org.apache.hudi.common.util.Option;
@@ -23,6 +28,8 @@ import org.apache.hudi.config.HoodieWriteConfig;
import org.apache.hudi.metadata.HoodieBackedTableMetadata;
import org.apache.hudi.metadata.HoodieIndexVersion;
import org.apache.hudi.metadata.HoodieMetadataPayload;
+import org.apache.hudi.metadata.HoodieMetadataWriteUtils;
+import org.apache.hudi.metadata.HoodieTableMetadata;
import org.apache.hudi.metadata.HoodieTableMetadataUtil;
import org.apache.hudi.metadata.MetadataPartitionType;
import org.apache.hudi.metadata.index.model.IndexInitializationContext;
@@ -30,16 +37,22 @@ import
org.apache.hudi.metadata.index.model.IndexInitializationPlan;
import org.apache.hudi.metadata.index.model.IndexPartitionAndRecords;
import org.apache.hudi.metadata.index.model.IndexUpdateContext;
import org.apache.hudi.metadata.stats.HoodieColumnRangeMetadata;
+import org.apache.hudi.metadata.stats.ValueMetadata;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.EnumSource;
import org.junit.jupiter.params.provider.NullSource;
+import org.junit.jupiter.params.provider.ValueSource;
import org.mockito.MockedStatic;
import java.io.IOException;
+import java.util.ArrayList;
import java.util.Collections;
+import java.util.HashMap;
import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
import static
org.apache.hudi.common.testutils.HoodieTestUtils.getDefaultStorageConf;
import static org.junit.jupiter.api.Assertions.assertEquals;
@@ -48,12 +61,90 @@ import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyBoolean;
+import static org.mockito.Mockito.CALLS_REAL_METHODS;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.mockStatic;
import static org.mockito.Mockito.when;
class TestPartitionStatsIndexer {
+ @ParameterizedTest
+ @ValueSource(strings = {"old", "current", "all", "complete"})
+ @SuppressWarnings("unchecked")
+ void testPartitionStatsRequireEveryFileForEachColumn(String missingStats) {
+ HoodieEngineContext engineContext = new
HoodieLocalEngineContext(getDefaultStorageConf());
+ HoodieTableMetaClient metaClient = mock(HoodieTableMetaClient.class);
+ HoodieTableConfig tableConfig = mock(HoodieTableConfig.class);
+ HoodieTableMetadata tableMetadata = mock(HoodieTableMetadata.class);
+ HoodieWriteConfig writeConfig = mock(HoodieWriteConfig.class);
+ HoodieMetadataConfig metadataConfig =
HoodieMetadataConfig.newBuilder().enable(true)
+
.withMetadataIndexColumnStats(true).withColumnStatsIndexForColumns("age,name").build();
+ when(metaClient.getTableConfig()).thenReturn(tableConfig);
+
when(tableConfig.isMetadataPartitionAvailable(MetadataPartitionType.COLUMN_STATS)).thenReturn(true);
+ when(metaClient.getIndexMetadata()).thenReturn(Option.empty());
+ when(tableConfig.getTableVersion()).thenReturn(HoodieTableVersion.NINE);
+
+ HoodieCommitMetadata commit = new HoodieCommitMetadata();
+ commit.addMetadata(HoodieCommitMetadata.SCHEMA_KEY,
+
"{\"type\":\"record\",\"name\":\"row\",\"fields\":[{\"name\":\"age\",\"type\":\"int\"},{\"name\":\"name\",\"type\":\"string\"}]}");
+ HoodieWriteStat writeStat = new HoodieWriteStat();
+ writeStat.setPartitionPath("p1");
+ writeStat.setPath("p1/new.parquet");
+ commit.addWriteStat("p1", writeStat);
+ Map<String, HoodieSchema> columns = new HashMap<>();
+ columns.put("age", HoodieSchema.create(HoodieSchemaType.INT));
+ columns.put("name", HoodieSchema.create(HoodieSchemaType.STRING));
+ List<HoodieColumnRangeMetadata<Comparable>> oldStats = new ArrayList<>();
+ List<HoodieColumnRangeMetadata<Comparable>> newStats = new ArrayList<>();
+ if (!"all".equals(missingStats)) {
+ oldStats.add(range("old.parquet", "age", 25, columns));
+ newStats.add(range("p1/new.parquet", "age", 30, columns));
+ if (!"old".equals(missingStats)) {
+ oldStats.add(range("old.parquet", "name", "alice", columns));
+ }
+ if (!"current".equals(missingStats)) {
+ newStats.add(range("p1/new.parquet", "name", "bob", columns));
+ }
+ }
+ List<HoodieRecord> storedStats =
HoodieMetadataPayload.createColumnStatsRecords("p1", oldStats, false)
+ .collect(Collectors.toList());
+ when(tableMetadata.getRecordsByKeyPrefixes(any(), any(), anyBoolean()))
+ .thenReturn((HoodieData) HoodieListData.eager(storedStats));
+
+ try (MockedStatic<HoodieTableMetadataUtil> util =
mockStatic(HoodieTableMetadataUtil.class, CALLS_REAL_METHODS);
+ MockedStatic<HoodieMetadataWriteUtils> writeUtils =
mockStatic(HoodieMetadataWriteUtils.class)) {
+ util.when(() -> HoodieTableMetadataUtil.getColumnsToIndex(any(), any(),
any(), anyBoolean(), any(), any()))
+ .thenReturn(columns);
+ util.when(() ->
HoodieTableMetadataUtil.translateWriteStatToFileStats(any(), any(), any(),
any())).thenReturn(newStats);
+ writeUtils.when(() ->
HoodieMetadataWriteUtils.getMaxInstantTime(metaClient,
"002")).thenReturn("002");
+ writeUtils.when(() ->
HoodieMetadataWriteUtils.getFilesToFetchColumnStats(any(), any(), any(), any(),
any(), any(), any(), any(), any(), any()))
+ .thenReturn(Collections.singleton("old.parquet"));
+
+ List<HoodieRecord> records =
PartitionStatsIndexer.convertMetadataToPartitionStatsRecords(
+ commit, "002", engineContext, writeConfig, metaClient,
tableMetadata, metadataConfig, Option.empty(), false).collectAsList();
+ Map<String, HoodieMetadataColumnStats> result = records.stream()
+ .map(record -> ((HoodieMetadataPayload)
record.getData()).getColumnStatMetadata().get())
+ .collect(Collectors.toMap(HoodieMetadataColumnStats::getColumnName,
stats -> stats));
+ assertEquals(columns.keySet(), result.keySet());
+ assertEquals("all".equals(missingStats),
result.get("age").getIsDeleted());
+ assertEquals(!"complete".equals(missingStats),
result.get("name").getIsDeleted());
+ if (!"all".equals(missingStats)) {
+ assertEquals(2L, result.get("age").getValueCount());
+ assertEquals(25,
HoodieColumnRangeMetadata.fromColumnStats(result.get("age")).getMinValue());
+ assertEquals(30,
HoodieColumnRangeMetadata.fromColumnStats(result.get("age")).getMaxValue());
+ }
+ if ("complete".equals(missingStats)) {
+ assertEquals("alice",
HoodieColumnRangeMetadata.fromColumnStats(result.get("name")).getMinValue());
+ assertEquals("bob",
HoodieColumnRangeMetadata.fromColumnStats(result.get("name")).getMaxValue());
+ }
+ }
+ }
+
+ private static HoodieColumnRangeMetadata<Comparable> range(String file,
String column, Comparable value, Map<String, HoodieSchema> columns) {
+ return HoodieColumnRangeMetadata.create(file, column, value, value, 0, 1,
1, 1,
+ ValueMetadata.getValueMetadata(columns.get(column),
HoodieIndexVersion.V1));
+ }
+
@Test
void testSkipForNonPartitionedTable() throws IOException {
HoodieEngineContext engineContext = mock(HoodieEngineContext.class);
diff --git
a/hudi-common/src/main/java/org/apache/hudi/metadata/HoodieTableMetadataUtil.java
b/hudi-common/src/main/java/org/apache/hudi/metadata/HoodieTableMetadataUtil.java
index 7509aafc645c..af736595531c 100644
---
a/hudi-common/src/main/java/org/apache/hudi/metadata/HoodieTableMetadataUtil.java
+++
b/hudi-common/src/main/java/org/apache/hudi/metadata/HoodieTableMetadataUtil.java
@@ -2131,44 +2131,6 @@ public class HoodieTableMetadataUtil {
};
}
- private static Stream<HoodieRecord> collectAndProcessColumnMetadata(
- List<List<HoodieColumnRangeMetadata<Comparable>>> fileColumnMetadata,
- String partitionPath, boolean isTightBound,
- Map<String, HoodieSchema> colsToIndexSchemaMap,
- HoodieIndexVersion partitionStatsIndexVersion
- ) {
- return collectAndProcessColumnMetadata(partitionPath, isTightBound,
Option.empty(), fileColumnMetadata.stream().flatMap(List::stream),
colsToIndexSchemaMap, partitionStatsIndexVersion);
- }
-
- private static Stream<HoodieRecord>
collectAndProcessColumnMetadata(Iterable<HoodieColumnRangeMetadata<Comparable>>
fileColumnMetadataIterable, String partitionPath,
- boolean
isTightBound, Option<String> indexPartitionOpt,
-
Map<String, HoodieSchema> colsToIndexSchemaMap,
-
HoodieIndexVersion partitionStatsIndexVersion
- ) {
-
- List<HoodieColumnRangeMetadata<Comparable>> fileColumnMetadata = new
ArrayList<>();
- fileColumnMetadataIterable.forEach(fileColumnMetadata::add);
- // Group by Column Name
- return collectAndProcessColumnMetadata(partitionPath, isTightBound,
indexPartitionOpt, fileColumnMetadata.stream(), colsToIndexSchemaMap,
partitionStatsIndexVersion);
- }
-
- private static Stream<HoodieRecord> collectAndProcessColumnMetadata(String
partitionPath, boolean isTightBound, Option<String> indexPartitionOpt,
-
Stream<HoodieColumnRangeMetadata<Comparable>> fileColumnMetadata,
-
Map<String, HoodieSchema> colsToIndexSchemaMap,
-
HoodieIndexVersion partitionStatsIndexVersion
- ) {
- // Group by Column Name
- Map<String, List<HoodieColumnRangeMetadata<Comparable>>> columnMetadataMap
=
-
fileColumnMetadata.collect(Collectors.groupingBy(HoodieColumnRangeMetadata::getColumnName,
Collectors.toList()));
-
- // Aggregate Column Ranges
- Stream<HoodieColumnRangeMetadata<Comparable>> partitionStatsRangeMetadata
= columnMetadataMap.entrySet().stream()
- .map(entry -> FileFormatUtils.getColumnRangeInPartition(partitionPath,
entry.getKey(), entry.getValue(), colsToIndexSchemaMap,
partitionStatsIndexVersion));
-
- // Create Partition Stats Records
- return HoodieMetadataPayload.createPartitionStatsRecords(partitionPath,
partitionStatsRangeMetadata.collect(Collectors.toList()), false, isTightBound,
indexPartitionOpt);
- }
-
public static HoodieData<HoodieRecord>
collectAndProcessExprIndexPartitionStatRecords(HoodiePairData<String,
HoodieColumnRangeMetadata<Comparable>> fileColumnMetadata,
boolean isTightBound, Option<String> indexPartitionOpt) {
// Step 1: Group by partition name
@@ -2221,12 +2183,12 @@ public class HoodieTableMetadataUtil {
return engineContext.parallelize(partitionToFileNames,
parallelism).flatMap(partitionInfo -> {
final String partitionPath = partitionInfo.getKey();
// Step 1: Collect Column Metadata for Each File
- List<List<HoodieColumnRangeMetadata<Comparable>>> fileColumnMetadata =
partitionInfo.getValue().stream()
- .map(fileName -> getFileStatsRangeMetadata(partitionPath, fileName,
dataTableMetaClient, new ArrayList<>(columnsToIndexSchemaMap.keySet()), false,
- metadataConfig.getMaxReaderBufferSize(),
partitionStatsIndexVersion))
+ List<HoodieColumnRangeMetadata<Comparable>> fileColumnMetadata =
partitionInfo.getValue().stream()
+ .flatMap(fileName -> getFileStatsRangeMetadata(partitionPath,
fileName, dataTableMetaClient, new
ArrayList<>(columnsToIndexSchemaMap.keySet()), false,
+ metadataConfig.getMaxReaderBufferSize(),
partitionStatsIndexVersion).stream())
.collect(Collectors.toList());
- return collectAndProcessColumnMetadata(fileColumnMetadata,
partitionPath, true, columnsToIndexSchemaMap,
partitionStatsIndexVersion).iterator();
+ return convertMetadataToPartitionStatsRecords(partitionPath,
fileColumnMetadata, partitionInfo.getValue(), columnsToIndexSchemaMap,
partitionStatsIndexVersion).iterator();
});
}
@@ -2253,27 +2215,34 @@ public class HoodieTableMetadataUtil {
return readColumnRangeMetadataFrom(partitionPath, fileName,
datasetMetaClient, columnsToIndex, maxBufferSize, indexVersion);
}
- public static HoodieData<HoodieRecord>
convertMetadataToPartitionStatsRecords(HoodiePairData<String,
List<HoodieColumnRangeMetadata<Comparable>>> columnRangeMetadataPartitionPair,
-
HoodieTableMetaClient dataMetaClient,
-
Map<String, HoodieSchema> colsToIndexSchemaMap,
-
HoodieIndexVersion partitionStatsIndexVersion
- ) {
- try {
- return columnRangeMetadataPartitionPair
- .flatMapValues(List::iterator)
- .groupByKey()
- .map(pair -> {
- final String partitionName = pair.getLeft();
- return collectAndProcessColumnMetadata(pair.getRight(),
partitionName, isShouldScanColStatsForTightBound(dataMetaClient),
Option.empty(), colsToIndexSchemaMap, partitionStatsIndexVersion);
- })
- .flatMap(recordStream -> recordStream.iterator());
- } catch (Exception e) {
- throw new HoodieException("Failed to generate column stats records for
metadata table", e);
- }
- }
-
- public static boolean
isShouldScanColStatsForTightBound(HoodieTableMetaClient dataMetaClient) {
- return
MetadataPartitionType.COLUMN_STATS.isMetadataPartitionAvailable(dataMetaClient);
+ /**
+ * Generates partition statistics only for columns with statistics covering
every expected base and log file.
+ * Incomplete columns produce partition stats delete records to invalidate
any previously stored bounds for those columns.
+ * The expected file names must come from the file inventory, independently
of statistics extraction.
+ */
+ public static Stream<HoodieRecord>
convertMetadataToPartitionStatsRecords(String partitionPath,
+
List<HoodieColumnRangeMetadata<Comparable>> fileColumnMetadata,
+
Set<String> expectedFileNames,
+
Map<String, HoodieSchema> colsToIndexSchemaMap,
+
HoodieIndexVersion partitionStatsIndexVersion) {
+ Map<String, List<HoodieColumnRangeMetadata<Comparable>>> columnMetadataMap
= fileColumnMetadata.stream()
+ // Negative counts belong to missing/deleted statistics, whereas
all-null and empty files have valid counts.
+ .filter(range -> range.getValueCount() >= 0 && range.getNullCount() >=
0)
+
.collect(Collectors.groupingBy(HoodieColumnRangeMetadata::getColumnName));
+ return colsToIndexSchemaMap.keySet().stream().flatMap(column -> {
+ List<HoodieColumnRangeMetadata<Comparable>> ranges =
columnMetadataMap.getOrDefault(column, Collections.emptyList());
+ Set<String> coveredFiles = ranges.stream()
+ .map(range -> new StoragePath(range.getFilePath()).getName())
+ .collect(Collectors.toSet());
+ if (expectedFileNames.isEmpty() ||
!coveredFiles.containsAll(expectedFileNames)) {
+ // Omitting an incomplete update would leave the old, potentially
narrower bounds in use.
+ return HoodieMetadataPayload.createPartitionStatsRecords(partitionPath,
+
Collections.singletonList(HoodieColumnRangeMetadata.stub(partitionPath, column,
partitionStatsIndexVersion)), true, true, Option.empty());
+ }
+ HoodieColumnRangeMetadata<Comparable> partitionRange =
FileFormatUtils.getColumnRangeInPartition(
+ partitionPath, column, ranges, colsToIndexSchemaMap,
partitionStatsIndexVersion);
+ return HoodieMetadataPayload.createPartitionStatsRecords(partitionPath,
Collections.singletonList(partitionRange), false, true, Option.empty());
+ });
}
public static HoodieIndexDefinition getHoodieIndexDefinition(String
indexName, HoodieTableMetaClient metaClient) {
diff --git
a/hudi-common/src/test/java/org/apache/hudi/metadata/TestHoodieTableMetadataUtil.java
b/hudi-common/src/test/java/org/apache/hudi/metadata/TestHoodieTableMetadataUtil.java
index 36256f50ae90..247ded3b6e2a 100644
---
a/hudi-common/src/test/java/org/apache/hudi/metadata/TestHoodieTableMetadataUtil.java
+++
b/hudi-common/src/test/java/org/apache/hudi/metadata/TestHoodieTableMetadataUtil.java
@@ -26,7 +26,6 @@ import org.apache.hudi.common.avro.HoodieAvroUtils;
import org.apache.hudi.common.config.HoodieMetadataConfig;
import org.apache.hudi.common.data.HoodieData;
import org.apache.hudi.common.data.HoodieListData;
-import org.apache.hudi.common.data.HoodiePairData;
import org.apache.hudi.common.engine.HoodieEngineContext;
import org.apache.hudi.common.engine.HoodieLocalEngineContext;
import org.apache.hudi.common.engine.HoodieReaderContext;
@@ -793,18 +792,70 @@ class TestHoodieTableMetadataUtil {
}
@Test
- @SuppressWarnings("unchecked")
- void testPartitionStatsConversionFailureIsWrapped() {
- HoodiePairData<String, List<HoodieColumnRangeMetadata<Comparable>>>
pairData =
- mock(HoodiePairData.class);
- when(pairData.flatMapValues(any())).thenThrow(new
RuntimeException("conversion failed"));
+ void testPartitionStatsCoverageAndInvalidation() {
+ Map<String, HoodieSchema> columns = new HashMap<>();
+ columns.put("age", HoodieSchema.create(HoodieSchemaType.INT));
+ columns.put("name", HoodieSchema.create(HoodieSchemaType.STRING));
+ columns.put("absent", HoodieSchema.create(HoodieSchemaType.STRING));
+ List<HoodieColumnRangeMetadata<Comparable>> ranges = Arrays.asList(
+ partitionStatsRange("/table/p1/old.parquet", "age", 25, 0, columns),
+ partitionStatsRange("p1/new.parquet", "age", 30, 0, columns),
+ partitionStatsRange("new.parquet", "name", "bob", 0, columns),
+ partitionStatsRange("new.parquet", "name", "bob", 0, columns));
+ Map<String, HoodieMetadataColumnStats> result =
partitionStatsByColumn(ranges, columns);
+ assertEquals(columns.keySet(), result.keySet());
+ assertFalse(result.get("age").getIsDeleted());
+ assertTrue(result.get("age").getIsTightBound());
+ assertEquals(2L, result.get("age").getValueCount());
+ assertEquals(25,
HoodieColumnRangeMetadata.fromColumnStats(result.get("age")).getMinValue());
+ assertEquals(30,
HoodieColumnRangeMetadata.fromColumnStats(result.get("age")).getMaxValue());
+ // Duplicate contributions from one file cannot stand in for another
file's missing statistics.
+ assertTrue(result.get("name").getIsDeleted());
+ assertTrue(result.get("absent").getIsDeleted());
+ assertEquals("p1", result.get("name").getFileName());
+
+ HoodieMetadataColumnStats previous =
HoodieMetadataPayload.createPartitionStatsRecords("p1",
+ Collections.singletonList(partitionStatsRange("p1", "name", "bob", 0,
columns)), false, true, Option.empty())
+ .map(record -> ((HoodieMetadataPayload)
record.getData()).getColumnStatMetadata().get()).findFirst().get();
+ assertEquals(result.get("name"),
HoodieTableMetadataUtil.mergeColumnStatsRecords(previous, result.get("name")));
+ List<HoodieColumnRangeMetadata<Comparable>> complete = Arrays.asList(
+ partitionStatsRange("old.parquet", "name", "alice", 0, columns),
+ partitionStatsRange("new.parquet", "name", "bob", 0, columns));
+ HoodieMetadataColumnStats restored = partitionStatsByColumn(complete,
columns).get("name");
+ assertFalse(restored.getIsDeleted());
+ assertEquals(restored,
HoodieTableMetadataUtil.mergeColumnStatsRecords(result.get("name"), restored));
+ }
- assertThrows(org.apache.hudi.exception.HoodieException.class,
- () -> HoodieTableMetadataUtil.convertMetadataToPartitionStatsRecords(
- pairData,
- mock(HoodieTableMetaClient.class),
- Collections.emptyMap(),
- HoodieIndexVersion.V1));
+ @Test
+ void testPartitionStatsWithAllMissingOrAllNullStatistics() {
+ Map<String, HoodieSchema> columns = Collections.singletonMap("name",
HoodieSchema.create(HoodieSchemaType.STRING));
+ assertTrue(partitionStatsByColumn(Collections.emptyList(),
columns).get("name").getIsDeleted());
+ List<HoodieColumnRangeMetadata<Comparable>> ranges = Arrays.asList(
+ partitionStatsRange("old.parquet", "name", null, 1, columns),
+ HoodieColumnRangeMetadata.createEmpty("new.parquet", "name",
HoodieIndexVersion.V1));
+ HoodieMetadataColumnStats allNull = partitionStatsByColumn(ranges,
columns).get("name");
+ assertFalse(allNull.getIsDeleted());
+ assertNull(allNull.getMinValue());
+ assertNull(allNull.getMaxValue());
+ assertEquals(1L, allNull.getNullCount());
+ assertEquals(1L, allNull.getValueCount());
+
+ ranges = Arrays.asList(ranges.get(0),
HoodieColumnRangeMetadata.stub("new.parquet", "name", HoodieIndexVersion.V1));
+ assertTrue(partitionStatsByColumn(ranges,
columns).get("name").getIsDeleted());
+ }
+
+ private static Map<String, HoodieMetadataColumnStats>
partitionStatsByColumn(List<HoodieColumnRangeMetadata<Comparable>> ranges,
+
Map<String, HoodieSchema> columns) {
+ return
HoodieTableMetadataUtil.convertMetadataToPartitionStatsRecords("p1", ranges,
+ new java.util.HashSet<>(Arrays.asList("old.parquet", "new.parquet")),
columns, HoodieIndexVersion.V1)
+ .map(record -> ((HoodieMetadataPayload)
record.getData()).getColumnStatMetadata().get())
+
.collect(java.util.stream.Collectors.toMap(HoodieMetadataColumnStats::getColumnName,
stats -> stats));
+ }
+
+ private static HoodieColumnRangeMetadata<Comparable>
partitionStatsRange(String file, String column, Comparable value, long
nullCount,
+
Map<String, HoodieSchema> columns) {
+ return HoodieColumnRangeMetadata.create(file, column, value, value,
nullCount, 1, 1, 1,
+
org.apache.hudi.metadata.stats.ValueMetadata.getValueMetadata(columns.get(column),
HoodieIndexVersion.V1));
}
@Test
diff --git
a/hudi-hadoop-common/src/test/java/org/apache/hudi/metadata/TestHoodieTableMetadataUtil.java
b/hudi-hadoop-common/src/test/java/org/apache/hudi/metadata/TestHoodieTableMetadataUtil.java
index 6b34762d97fc..cc03685b2ce3 100644
---
a/hudi-hadoop-common/src/test/java/org/apache/hudi/metadata/TestHoodieTableMetadataUtil.java
+++
b/hudi-hadoop-common/src/test/java/org/apache/hudi/metadata/TestHoodieTableMetadataUtil.java
@@ -66,6 +66,7 @@ import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.Arguments;
import org.junit.jupiter.params.provider.MethodSource;
+import org.junit.jupiter.params.provider.ValueSource;
import java.io.IOException;
import java.net.URI;
@@ -192,6 +193,35 @@ public class TestHoodieTableMetadataUtil extends
HoodieCommonTestHarness {
validatePartitionStats(result, instant1, instant2);
}
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void testPartitionStatsInitializationWithUnreadableFile(boolean
logFile) throws Exception {
+ HoodieLocalEngineContext engineContext = new
HoodieLocalEngineContext(metaClient.getStorageConf());
+ String instant = "20230918120000000";
+ hoodieTestTable = hoodieTestTable.addCommit(instant);
+ String partition = DATE_PARTITIONS.get(0);
+ String fileId = UUID.randomUUID().toString();
+ StoragePath path = new
StoragePath(hoodieTestTable.getBaseFilePath(partition, fileId).toUri());
+ writeParquetFile(instant, path,
dataGen.generateInsertsForPartition(instant, 10, partition), metaClient,
engineContext);
+ FileSlice readable = new FileSlice(partition, instant, fileId);
+ readable.setBaseFile(new HoodieBaseFile(path.toString()));
+ FileSlice unreadable = new FileSlice(partition, instant, "missing");
+ if (logFile) {
+ unreadable.addLogFile(new HoodieLogFile(new
StoragePath(path.getParent(), ".missing_" + instant + ".log.1_1-0-1")));
+ } else {
+ unreadable.setBaseFile(new HoodieBaseFile(new
StoragePath(path.getParent(), "missing_1-0-1_" + instant +
".parquet").toString()));
+ }
+
+ List<HoodieRecord> records =
HoodieTableMetadataUtil.convertFilesToPartitionStatsRecords(
+ engineContext, Arrays.asList(FileSliceAndPartition.of(partition,
readable), FileSliceAndPartition.of(partition, unreadable)),
+
HoodieMetadataConfig.newBuilder().enable(true).withMetadataIndexColumnStats(true)
+
.withColumnStatsIndexForColumns("rider,driver").withPartitionStatsIndexParallelism(1).build(),
+ metaClient,
Lazy.eagerly(Option.of(HOODIE_SCHEMA_WITH_METADATA_FIELDS)),
Option.empty()).collectAsList();
+ assertFalse(records.isEmpty());
+ records.forEach(record -> assertTrue(((HoodieMetadataPayload)
record.getData()).getColumnStatMetadata().get().getIsDeleted(),
+ "A readable subset must not become partition statistics"));
+ }
+
@Test
public void testReadRecordKeysFromBaseFilesWithValidRecords() throws
Exception {
HoodieLocalEngineContext engineContext = new
HoodieLocalEngineContext(metaClient.getStorageConf());
diff --git
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/PartitionStatsIndexSupport.scala
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/PartitionStatsIndexSupport.scala
index 5e7d6748f706..94779f454288 100644
---
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/PartitionStatsIndexSupport.scala
+++
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/PartitionStatsIndexSupport.scala
@@ -38,8 +38,7 @@ import org.apache.hudi.util.{JFunction}
import org.apache.spark.internal.Logging
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.catalyst.expressions.{And, AttributeReference,
DateAdd, DateFormatClass, DateSub, EqualNullSafe, Expression, FromUnixTime,
IsNotNull, IsNull, Literal, Not, Or, ParseToDate, ParseToTimestamp,
RegExpExtract, RegExpReplace, StringSplit, StringTrim, StringTrimLeft,
StringTrimRight, Substring, UnaryExpression, UnixTimestamp}
-import org.apache.spark.sql.catalyst.expressions.Literal.TrueLiteral
-import
org.apache.spark.sql.hudi.DataSkippingUtils.{getTargetIndexedColumnName,
translateIntoColumnStatsIndexFilterExpr}
+import org.apache.spark.sql.hudi.DataSkippingUtils.getTargetIndexedColumnName
import org.apache.spark.sql.hudi.HoodieSqlCommonUtils
import org.apache.spark.sql.types.StructType
import org.apache.spark.storage.StorageLevel
@@ -121,30 +120,27 @@ class PartitionStatsIndexSupport(spark: SparkSession,
transposedPartitionStatsDF => {
try {
transposedPartitionStatsDF.persist(StorageLevel.MEMORY_AND_DISK_SER)
- val allPartitions =
transposedPartitionStatsDF.select(HoodieMetadataPayload.COLUMN_STATS_FIELD_FILE_NAME)
- .collect()
- .map(_.getString(0))
- .toSet
- if (allPartitions.nonEmpty) {
- // PARTITION_STATS index exist for all or some columns in the
filters
- // NOTE: [[translateIntoColumnStatsIndexFilterExpr]] has
covered the case where the
- // column in a filter does not have the stats available,
by making sure such a
- // filter does not prune any partition.
- // to be fixed. HUDI-8836.
- val indexFilter =
filteredQueryFilters.map(translateIntoColumnStatsIndexFilterExpr(_, indexedCols
= filteredIndexedCols)).reduce(And)
- if (indexFilter.equals(TrueLiteral)) {
- // if there are any non indexed cols or we can't translate
source expr, we cannot prune partitions based on col stats lookup.
- Some(allPartitions)
- } else {
-
Some(transposedPartitionStatsDF.where(sparkAdapter.createColumnFromExpression(indexFilter))
- .select(HoodieMetadataPayload.COLUMN_STATS_FIELD_FILE_NAME)
+ // Use the file index's query-path inventory: partitions
entirely absent from the
+ // projected stats must survive, just like files absent from the
column stats index.
+ lazy val allPartitions =
fileIndex.getAllQueryPartitionPaths.asScala.map(_.getPath).toSet
+ buildCandidateFilter(transposedPartitionStatsDF,
filteredQueryFilters, filteredIndexedCols) match {
+ case None => Some(allPartitions)
+ case Some(candidateFilter) =>
+ val indexedPartitions =
transposedPartitionStatsDF.select(HoodieMetadataPayload.COLUMN_STATS_FIELD_FILE_NAME)
.collect()
.map(_.getString(0))
- .toSet)
- }
- } else {
- // PARTITION_STATS index does not exist for any column in the
filters, skip the pruning
- Option.empty
+ .toSet
+ if (indexedPartitions.nonEmpty) {
+ val matchingPartitions =
transposedPartitionStatsDF.where(candidateFilter)
+
.select(HoodieMetadataPayload.COLUMN_STATS_FIELD_FILE_NAME)
+ .collect()
+ .map(_.getString(0))
+ .toSet
+ Some((matchingPartitions ++ (allPartitions --
indexedPartitions)).intersect(allPartitions))
+ } else {
+ // PARTITION_STATS index does not exist for any column in
the filters, skip the pruning
+ None
+ }
}
} finally {
transposedPartitionStatsDF.unpersist()
diff --git
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/SparkBaseIndexSupport.scala
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/SparkBaseIndexSupport.scala
index 6714bdcbf295..f4ef5ccf4fa4 100644
---
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/SparkBaseIndexSupport.scala
+++
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/SparkBaseIndexSupport.scala
@@ -18,6 +18,7 @@
package org.apache.hudi
+import org.apache.hudi.ColumnStatsIndexSupport.{getMaxColumnNameFor,
getMinColumnNameFor, getNullCountColumnNameFor}
import org.apache.hudi.HoodieFileIndex.DataSkippingFailureMode
import org.apache.hudi.client.common.HoodieSparkEngineContext
import org.apache.hudi.common.config.HoodieMetadataConfig
@@ -29,9 +30,11 @@ import org.apache.hudi.metadata.{HoodieMetadataPayload,
HoodieTableMetadata}
import
org.apache.hudi.metadata.HoodieTableMetadataUtil.PARTITION_NAME_COLUMN_STATS
import org.apache.spark.api.java.JavaSparkContext
-import org.apache.spark.sql.{DataFrame, SparkSession}
-import org.apache.spark.sql.catalyst.expressions.{And, Expression}
+import org.apache.spark.sql.{Column, DataFrame, SparkSession}
+import org.apache.spark.sql.catalyst.analysis.UnresolvedAttribute
+import org.apache.spark.sql.catalyst.expressions.{Expression, PredicateHelper}
import org.apache.spark.sql.catalyst.expressions.Literal.TrueLiteral
+import org.apache.spark.sql.catalyst.util.quoteIdentifier
import
org.apache.spark.sql.hudi.DataSkippingUtils.translateIntoColumnStatsIndexFilterExpr
import scala.collection.JavaConverters._
@@ -113,14 +116,13 @@ abstract class SparkBaseIndexSupport(spark: SparkSession,
}
val validIndexedColumns = getValidIndexedColumnsFunc.apply(indexDefinition)
- val indexFilter =
queryFilters.map(translateIntoColumnStatsIndexFilterExpr(_, isExpressionIndex,
validIndexedColumns)).reduce(And)
- if (indexFilter.equals(TrueLiteral)) {
- // if there are any non indexed cols or we can't translate source expr,
we have to read all files and may not benefit from col stats lookup.
- fileNamesFromPrunedPartitions
+ val candidateFilter = buildCandidateFilter(indexDf, queryFilters,
validIndexedColumns, isExpressionIndex)
+ if (candidateFilter.isEmpty) {
+ // No translated predicate can prune files, so avoid scanning the index.
+ fileNamesFromPrunedPartitions
} else {
- // only lookup in col stats if all filters are eligible to be looked up
in col stats index in MDT
val prunedCandidateFileNames =
- indexDf.where(sparkAdapter.createColumnFromExpression(indexFilter))
+ indexDf.where(candidateFilter.get)
.select(HoodieMetadataPayload.COLUMN_STATS_FIELD_FILE_NAME)
.collect()
.map(_.getString(0))
@@ -144,6 +146,51 @@ abstract class SparkBaseIndexSupport(spark: SparkSession,
}
}
+ /**
+ * Builds a conservative filter shared by file and partition statistics
indexes.
+ * A null null-count marks missing statistics, whereas null min/max values
can also describe an all-null column.
+ */
+ protected def buildCandidateFilter(indexDf: DataFrame, queryFilters:
Seq[Expression], validIndexedColumns: Seq[String],
+ isExpressionIndex: Boolean = false):
Option[Column] = {
+ // Only split top-level AND predicates; keep OR and NOT subtrees intact.
+ val indexFilters =
queryFilters.flatMap(SparkBaseIndexSupport.splitConjuncts)
+ .map(translateIntoColumnStatsIndexFilterExpr(_, isExpressionIndex,
validIndexedColumns))
+ .filterNot(_.equals(TrueLiteral))
+ if (indexFilters.isEmpty) {
+ None
+ } else {
+ val projectedColumns = indexDf.columns.toSet
+ val statsToNullCount = validIndexedColumns.flatMap { column =>
+ val nullCountColumn = getNullCountColumnNameFor(column)
+ if (projectedColumns.contains(nullCountColumn)) {
+ Seq(getMinColumnNameFor(column), getMaxColumnNameFor(column),
nullCountColumn)
+ .map(_ -> nullCountColumn)
+ } else {
+ Seq.empty
+ }
+ }.toMap
+ // Guard each independent conjunct using only the statistics it reads. A
complete conjunct
+ // can still exclude a candidate when another conjunct has missing
statistics.
+ // For example, age > 100 AND name = 'alice' becomes:
+ // (age_maxValue > 100 OR age_nullCount IS NULL)
+ // AND ((name_minValue <= 'alice' AND name_maxValue >= 'alice') OR
name_nullCount IS NULL)
+ // A file with age_maxValue = 30 and age_nullCount = 0 is still excluded
even if name statistics are missing.
+ Some(indexFilters.map { conjunct =>
+ val referencedStats = conjunct.references.map {
+ // UnresolvedAttribute.name adds SQL quoting; the index schema holds
literal field names.
+ case attribute: UnresolvedAttribute =>
attribute.nameParts.mkString(".")
+ case attribute => attribute.name
+ }.toSet
+ val nullCountColumns = referencedStats.flatMap(statsToNullCount.get)
+ // Keep the guard outside the whole translated conjunct, including
NOT. Substituting true
+ // for a missing-statistics leaf before negation could incorrectly
exclude a candidate.
+
nullCountColumns.foldLeft(sparkAdapter.createColumnFromExpression(conjunct)) {
(filter, column) =>
+ filter.or(indexDf.col(quoteIdentifier(column)).isNull)
+ }
+ }.reduce(_.and(_)))
+ }
+ }
+
/**
* Determines whether it would be more optimal to read Column Stats Index a)
in-memory of the invoking process,
* or b) executing it on-cluster via Spark [[Dataset]] and [[RDD]] APIs
@@ -240,3 +287,8 @@ abstract class SparkBaseIndexSupport(spark: SparkSession,
Option(recordKeysOpt.orElse(null)).filter(_.nonEmpty)
}
}
+
+// Keep PredicateHelper's Logging trait out of the index support class
hierarchy.
+private object SparkBaseIndexSupport extends PredicateHelper {
+ def splitConjuncts(expression: Expression): Seq[Expression] =
splitConjunctivePredicates(expression)
+}
diff --git
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/dataskipping/TermRange.scala
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/dataskipping/TermRange.scala
index 8c26151c7f86..811d38190df2 100644
---
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/dataskipping/TermRange.scala
+++
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/dataskipping/TermRange.scala
@@ -21,6 +21,7 @@ import
org.apache.hudi.ColumnStatsIndexSupport.{getMaxColumnNameFor, getMinColum
import org.apache.hudi.SparkAdapterSupport.sparkAdapter
import org.apache.spark.sql.catalyst.expressions.{AttributeReference,
Expression}
+import org.apache.spark.sql.catalyst.util.quoteIdentifier
import org.apache.spark.sql.functions.col
import org.apache.spark.sql.hudi.DataSkippingUtils.getTargetIndexedColumnName
@@ -75,7 +76,7 @@ case class ColumnStatsSource(indexedCols: Seq[String])
extends RangeSource {
}
/** A column of the transposed column stats table, for ex `qty_minValue`. */
- private def statsColumn(name: String): Expression =
sparkAdapter.getExpressionFromColumn(col(name))
+ private def statsColumn(name: String): Expression =
sparkAdapter.getExpressionFromColumn(col(quoteIdentifier(name)))
}
object TermRange {
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/TestDataSkippingUtils.scala
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/TestDataSkippingUtils.scala
index e18b9a419e78..d0c4e0933dff 100644
---
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/TestDataSkippingUtils.scala
+++
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/TestDataSkippingUtils.scala
@@ -19,15 +19,17 @@ package org.apache.hudi
import org.apache.hudi.ColumnStatsIndexSupport.composeIndexSchema
import org.apache.hudi.SparkAdapterSupport.sparkAdapter
+import org.apache.hudi.common.config.HoodieMetadataConfig
+import org.apache.hudi.common.model.HoodieIndexDefinition
import org.apache.hudi.testutils.{HoodieDummyExpressionHolder,
HoodieSparkClientTestBase}
import org.apache.spark.sql.{Row, SparkSession}
import org.apache.spark.sql.HoodieCatalystExpressionUtils.resolveExpr
import org.apache.spark.sql.catalyst.analysis.UnresolvedAttribute
import org.apache.spark.sql.catalyst.encoders.DummyExpressionHolder
-import org.apache.spark.sql.catalyst.expressions.{Expression, InSet, Not,
ParseToDate, ParseToTimestamp}
+import org.apache.spark.sql.catalyst.expressions.{And, Expression,
GetStructField, InSet, Not, ParseToDate, ParseToTimestamp}
import org.apache.spark.sql.catalyst.optimizer.OptimizeIn
-import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
+import org.apache.spark.sql.catalyst.plans.logical.{Filter, LogicalPlan}
import org.apache.spark.sql.catalyst.rules.Rule
import org.apache.spark.sql.functions.{col, date_format, lit, lower, upper}
import org.apache.spark.sql.hudi.DataSkippingUtils
@@ -37,7 +39,7 @@ import org.junit.jupiter.api.{BeforeEach, Test}
import org.junit.jupiter.api.Assertions.{assertEquals, assertTrue}
import org.junit.jupiter.api.Assumptions.assumeTrue
import org.junit.jupiter.params.ParameterizedTest
-import org.junit.jupiter.params.provider.{Arguments, MethodSource}
+import org.junit.jupiter.params.provider.{Arguments, MethodSource, ValueSource}
import org.junit.jupiter.params.provider.Arguments.arguments
import java.sql.Timestamp
@@ -89,6 +91,129 @@ class TestDataSkippingUtils extends
HoodieSparkClientTestBase with SparkAdapterS
val (indexSchema: StructType, targetIndexedColumns: Seq[String]) =
composeIndexSchema(indexedCols, indexedCols, sourceTableSchema)
+ @ParameterizedTest
+ @MethodSource(Array("missingColumnStatsPredicates"))
+ def testCandidateFilesWithMissingColumnStats(predicate: String, expected:
Set[String], separateFilters: Boolean): Unit = {
+ val input = Seq(
+ IndexRow("missing", A_minValue = 25, A_maxValue = 25, A_nullCount = 0),
+ IndexRow("all_null", A_minValue = 25, A_maxValue = 25, A_nullCount = 0,
B_nullCount = 1),
+ IndexRow("matching", A_minValue = 25, A_maxValue = 25, A_nullCount = 0,
+ B_minValue = "alice", B_maxValue = "alice", B_nullCount = 0),
+ IndexRow("nonmatching", A_minValue = 25, A_maxValue = 25, A_nullCount =
0,
+ B_minValue = "aaa", B_maxValue = "aaa", B_nullCount = 0)
+ )
+ // C is indexed globally but not referenced by this query, so it is absent
from the projection.
+ val indexDf = spark.createDataFrame(input.map(_.toRow).asJava, indexSchema)
+ .drop("C_minValue", "C_maxValue", "C_nullCount")
+ val definition =
HoodieIndexDefinition.newBuilder().withIndexName("column_stats")
+
.withIndexType("column_stats").withSourceFields(indexedCols.asJava).build()
+ val resolvedPredicate = resolveExpr(spark, predicate, sourceTableSchema)
+ val filters = resolvedPredicate match {
+ case And(left, right) if separateFilters => Seq(left, right)
+ case other => Seq(other)
+ }
+ val support = new ColumnStatsIndexSupport(spark, sourceTableSchema, null,
+ HoodieMetadataConfig.newBuilder().build(), null) {
+ def candidates: Set[String] = getCandidateFiles(indexDf, filters,
+ input.map(_.fileName).toSet + "unindexed", _ => indexedCols,
+ indexDefinitionOpt = Some(definition))
+ }
+ assertEquals(expected + "unindexed", support.candidates)
+ }
+
+ @ParameterizedTest
+ @MethodSource(Array("dottedColumnStatsPredicates"))
+ def testCandidateFilesWithDottedColumnStats(predicate: String, expected:
Set[String],
+ separateFilters: Boolean,
structNested: Boolean): Unit = {
+ val nestedField = if (structNested) {
+ StructField("profile", StructType(Seq(StructField("name", StringType))))
+ } else {
+ StructField("profile.name", StringType)
+ }
+ val nestedSchema = StructType(Seq(StructField("A", LongType), nestedField,
StructField("C", TimestampType)))
+ // Literal dotted source attributes have a quoted canonical index name,
including the backticks.
+ val indexedName = if (structNested) "profile.name" else "`profile.name`"
+ val nestedIndexedCols = Seq("A", indexedName, "C")
+ val nestedIndexSchema = StructType(indexSchema.fields.map { field =>
+ if (field.name.startsWith("B_")) field.copy(name = indexedName +
field.name.stripPrefix("B")) else field
+ })
+ val input = Seq(
+ IndexRow("missing", A_minValue = 25, A_maxValue = 25, A_nullCount = 0),
+ IndexRow("missing_pruned_by_A", A_minValue = -1, A_maxValue = -1,
A_nullCount = 0),
+ IndexRow("all_null", A_minValue = 25, A_maxValue = 25, A_nullCount = 0,
B_nullCount = 1),
+ IndexRow("matching", A_minValue = 25, A_maxValue = 25, A_nullCount = 0,
+ B_minValue = "alice", B_maxValue = "alice", B_nullCount = 0),
+ IndexRow("nonmatching", A_minValue = 25, A_maxValue = 25, A_nullCount =
0,
+ B_minValue = "aaa", B_maxValue = "aaa", B_nullCount = 0))
+ // The index stores nested paths as literal top-level names containing
dots.
+ val indexDf =
spark.createDataFrame(spark.sparkContext.parallelize(input.map(_.toRow), 1),
nestedIndexSchema)
+ .drop("C_minValue", "C_maxValue", "C_nullCount")
+ assertTrue(indexDf.columns.contains(s"${indexedName}_nullCount"))
+
+ // Analyze the raw source DataFrame to retain the actual Catalyst shape of
both field references.
+ val sourceDf =
spark.createDataFrame(java.util.Collections.emptyList[Row](), nestedSchema)
+ def analyzePredicate(sql: String): Expression =
sourceDf.where(sql).queryExecution.analyzed
+ .asInstanceOf[Filter].condition
+ val resolvedPredicate = analyzePredicate(predicate)
+ assertEquals(structNested,
resolvedPredicate.exists(_.isInstanceOf[GetStructField]))
+ if (!structNested) {
+ val nestedLookup =
DataSkippingUtils.translateIntoColumnStatsIndexFilterExpr(
+ analyzePredicate("`profile.name` = 'alice'"), indexedCols =
nestedIndexedCols)
+ assertEquals(Set(Seq(s"${indexedName}_minValue"),
Seq(s"${indexedName}_maxValue")),
+ nestedLookup.collect { case attribute: UnresolvedAttribute =>
attribute.nameParts }.toSet,
+ "Dotted predicates must translate to literal statistics names instead
of silently disabling pruning")
+ }
+ val filters = resolvedPredicate match {
+ case And(left, right) if separateFilters => Seq(left, right)
+ case other => Seq(other)
+ }
+ val definition =
HoodieIndexDefinition.newBuilder().withIndexName("column_stats")
+
.withIndexType("column_stats").withSourceFields(nestedIndexedCols.asJava).build()
+ val support = new ColumnStatsIndexSupport(spark, nestedSchema, null,
+ HoodieMetadataConfig.newBuilder().build(), null) {
+ def candidates: Set[String] = getCandidateFiles(indexDf, filters,
+ input.map(_.fileName).toSet + "unindexed", _ => nestedIndexedCols,
+ indexDefinitionOpt = Some(definition))
+ }
+ val jobGroup = s"nested-column-stats-$predicate-$separateFilters"
+ spark.sparkContext.setJobGroup(jobGroup, "Nested column statistics must
prune file candidates")
+ try {
+ assertEquals(expected + "unindexed", support.candidates)
+
assertTrue(spark.sparkContext.statusTracker.getJobIdsForGroup(jobGroup).nonEmpty,
+ "Candidate pruning must execute against the index")
+ } finally {
+ spark.sparkContext.clearJobGroup()
+ }
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = Array(false, true))
+ def testUnsupportedPredicatesDoNotScanIndex(separateFilters: Boolean): Unit
= {
+ val input = Seq(IndexRow("indexed"))
+ val indexDf =
spark.createDataFrame(spark.sparkContext.parallelize(input.map(_.toRow), 1),
indexSchema)
+ val definition =
HoodieIndexDefinition.newBuilder().withIndexName("column_stats")
+
.withIndexType("column_stats").withSourceFields(indexedCols.asJava).build()
+ val predicate = resolveExpr(spark, "A % 2 = 1 AND length(B) > 2",
sourceTableSchema)
+ val filters = predicate match {
+ case And(left, right) if separateFilters => Seq(left, right)
+ case other => Seq(other)
+ }
+ val expected = Set("indexed", "unindexed")
+ val support = new ColumnStatsIndexSupport(spark, sourceTableSchema, null,
+ HoodieMetadataConfig.newBuilder().build(), null) {
+ def candidates: Set[String] = getCandidateFiles(indexDf, filters,
expected, _ => indexedCols,
+ indexDefinitionOpt = Some(definition))
+ }
+ val jobGroup = s"unsupported-column-stats-predicates-$separateFilters"
+ spark.sparkContext.setJobGroup(jobGroup, "Unsupported predicates must not
scan the index")
+ try {
+ assertEquals(expected, support.candidates)
+ assertEquals(0,
spark.sparkContext.statusTracker.getJobIdsForGroup(jobGroup).length)
+ } finally {
+ spark.sparkContext.clearJobGroup()
+ }
+ }
+
@ParameterizedTest
@MethodSource(Array(
"testBasicLookupFilterExpressionsSource",
@@ -318,6 +443,40 @@ object TestDataSkippingUtils {
private val NyFallBackEdt0130 = new Timestamp(1793511000000L) // 2026-11-01
05:30Z
private val NyFallBackEst0110 = new Timestamp(1793513400000L) // 2026-11-01
06:10Z
+ def dottedColumnStatsPredicates(): java.util.stream.Stream[Arguments] = {
+ val cases = Seq(
+ ("A >= 0 AND `profile.name` = 'alice'", Set("missing", "matching"),
false),
+ ("A >= 0 AND NOT (`profile.name` = 'aaa')", Set("missing", "matching"),
false),
+ ("A > 100 AND `profile.name` = 'alice'", Set.empty[String], false),
+ // GetStructField is currently unsupported, but its independent complete
conjunct can still prune.
+ ("A >= 0 AND profile.name = 'alice'", Set("missing", "all_null",
"matching", "nonmatching"), true),
+ ("A > 100 AND profile.name = 'alice'", Set.empty[String], true))
+ cases.flatMap { case (predicate, expected, structNested) =>
+ Seq(false, true).map(separate => arguments(predicate, expected,
Boolean.box(separate), Boolean.box(structNested)))
+ }.asJava.stream()
+ }
+
+ def missingColumnStatsPredicates(): java.util.stream.Stream[Arguments] = {
+ val matching = Set("missing", "matching")
+ val cases = Seq(
+ "A >= 0 AND B = 'alice'" -> matching,
+ "A >= 0 AND B > 'aaa'" -> matching,
+ "A >= 0 AND B IN ('alice', 'ann')" -> matching,
+ "A >= 0 AND NOT (B = 'aaa')" -> matching,
+ "A >= 0 AND (B = 'alice' OR B = 'ann')" -> matching,
+ "A > 100 AND B = 'alice'" -> Set.empty[String],
+ "A > 100 AND NOT (B = 'alice')" -> Set.empty[String],
+ "A > 100 AND length(B) > 2" -> Set.empty[String],
+ "length(B) > 2 AND A > 100" -> Set.empty[String],
+ "A > 100 OR B = 'alice'" -> matching,
+ "NOT (A > 100 OR B = 'aaa')" -> matching,
+ "NOT (A > 100 AND B = 'alice')" -> Set("missing", "matching",
"nonmatching", "all_null")
+ )
+ cases.flatMap { case (predicate, expected) =>
+ Seq(false, true).map(separate => arguments(predicate, expected,
Boolean.box(separate)))
+ }.asJava.stream()
+ }
+
def testStringsLookupFilterExpressionsSource():
java.util.stream.Stream[Arguments] = {
java.util.stream.Stream.of(
arguments(
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/TestPartitionStatsIndexSupport.scala
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/TestPartitionStatsIndexSupport.scala
new file mode 100644
index 000000000000..3f1920e1b237
--- /dev/null
+++
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/TestPartitionStatsIndexSupport.scala
@@ -0,0 +1,145 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi
+
+import org.apache.hudi.ColumnStatsIndexSupport.composeIndexSchema
+import org.apache.hudi.common.config.HoodieMetadataConfig
+import org.apache.hudi.common.model.{HoodieIndexDefinition,
HoodieIndexMetadata}
+import org.apache.hudi.common.table.{HoodieTableConfig, HoodieTableMetaClient,
HoodieTableVersion}
+import org.apache.hudi.common.util.{Option => HOption}
+import org.apache.hudi.core.read.BaseHoodieTableFileIndex.PartitionPath
+import org.apache.hudi.metadata.HoodieIndexVersion
+import org.apache.hudi.testutils.HoodieSparkClientTestBase
+
+import org.apache.spark.sql.{DataFrame, Row, SparkSession}
+import org.apache.spark.sql.HoodieCatalystExpressionUtils.resolveExpr
+import org.apache.spark.sql.catalyst.expressions.{And, Expression}
+import org.apache.spark.sql.types.{LongType, StringType, StructField,
StructType}
+import org.junit.jupiter.api.Assertions.{assertEquals, assertTrue}
+import org.junit.jupiter.api.BeforeEach
+import org.junit.jupiter.params.ParameterizedTest
+import org.junit.jupiter.params.provider.ValueSource
+import org.mockito.Mockito.{mock, never, verify, when}
+
+import scala.collection.JavaConverters._
+
+class TestPartitionStatsIndexSupport extends HoodieSparkClientTestBase {
+
+ private var spark: SparkSession = _
+ private val indexedColumns = Seq("A", "B")
+ private val sourceSchema = StructType(Seq(
+ StructField("A", LongType), StructField("B", StringType), StructField("D",
LongType)))
+ private val (indexSchema, _) = composeIndexSchema(indexedColumns,
indexedColumns, sourceSchema)
+ private val allPartitions = Set("missing", "missing_low", "all_null",
"matching", "nonmatching", "unindexed")
+
+ @BeforeEach
+ override def setUp(): Unit = {
+ initSparkContexts()
+ spark = sqlContext.sparkSession
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = Array(false, true))
+ def testMissingPartitionStatistics(readInMemory: Boolean): Unit = {
+ val indexDf = createIndex(readInMemory, Seq(
+ Row("missing", 1L, 125L, 125L, 0L, null, null, null),
+ Row("missing_low", 1L, 25L, 25L, 0L, null, null, null),
+ Row("all_null", 1L, 125L, 125L, 0L, null, null, 1L),
+ Row("matching", 1L, 125L, 125L, 0L, "alice", "alice", 0L),
+ Row("nonmatching", 1L, 125L, 125L, 0L, "aaa", "aaa", 0L),
+ // Stats can contain partitions outside the query's input paths.
+ Row("outside_query", 1L, 125L, 125L, 0L, "alice", "alice", 0L)))
+ val (support, fileIndex) = createSupport(indexDf, readInMemory)
+ val cases = Seq(
+ "B = 'alice'" -> Set("missing", "missing_low", "matching", "unindexed"),
+ "A > 100L AND B = 'alice'" -> Set("missing", "matching", "unindexed"),
+ "A > 100L OR B = 'alice'" -> allPartitions,
+ "A > 100L AND D = 1L" -> Set("missing", "all_null", "matching",
"nonmatching", "unindexed"))
+ cases.foreach { case (predicate, expected) =>
+ val filter = resolveExpr(spark, predicate, sourceSchema)
+ val filterSequences = filter match {
+ case And(left, right) => Seq(Seq(filter), Seq(left, right))
+ case _ => Seq(Seq(filter))
+ }
+ filterSequences.foreach { filters =>
+ assertEquals(Some(expected), support.prunePartitions(fileIndex,
filters), predicate)
+ }
+ }
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = Array(false, true))
+ def testAbsentPartitionsAndUntranslatedFilters(readInMemory: Boolean): Unit
= {
+ val frames = Seq(
+ createIndex(readInMemory, Seq.empty),
+ createIndex(readInMemory, Seq(Row("nonmatching", 1L, 25L, 25L, 0L,
"aaa", "aaa", 0L))))
+ frames.foreach { indexDf =>
+ val (support, fileIndex) = createSupport(indexDf, readInMemory)
+ val expected = if (indexDf.isEmpty) None else Some(allPartitions -
"nonmatching")
+ assertEquals(expected, support.prunePartitions(fileIndex,
+ Seq(resolveExpr(spark, "B = 'alice'", sourceSchema))))
+ if (expected.isEmpty) {
+ verify(fileIndex, never()).getAllQueryPartitionPaths
+ }
+ // This predicate references indexed A but translates to TrueLiteral.
+ assertEquals(Some(allPartitions), support.prunePartitions(fileIndex,
+ Seq(resolveExpr(spark, "A % 2L = 1L", sourceSchema))))
+ // Empty/constant filters return None, allowing the caller to use its
full inventory.
+ assertTrue(support.prunePartitions(fileIndex, Seq.empty).isEmpty)
+ assertTrue(support.prunePartitions(fileIndex, Seq(resolveExpr(spark,
"true", sourceSchema))).isEmpty)
+ assertTrue(support.prunePartitions(fileIndex, Seq(resolveExpr(spark,
"NOT (B = 'alice')", sourceSchema))).isEmpty)
+ }
+ }
+
+ private def createIndex(readInMemory: Boolean, rows: Seq[Row]): DataFrame = {
+ if (readInMemory) {
+ spark.createDataFrame(rows.asJava, indexSchema)
+ } else {
+ spark.createDataFrame(spark.sparkContext.parallelize(rows, 1),
indexSchema)
+ }
+ }
+
+ private def createSupport(indexDf: DataFrame, readInMemory: Boolean):
(PartitionStatsIndexSupport, HoodieFileIndex) = {
+ val metaClient = mock(classOf[HoodieTableMetaClient])
+ val tableConfig = mock(classOf[HoodieTableConfig])
+ val definition = HoodieIndexDefinition.newBuilder()
+ .withIndexName("column_stats").withIndexType("column_stats")
+
.withVersion(HoodieIndexVersion.V2).withSourceFields(indexedColumns.asJava).build()
+ when(metaClient.getIndexMetadata).thenReturn(HOption.of(
+ new HoodieIndexMetadata(Map("column_stats" -> definition).asJava)))
+ when(metaClient.getTableConfig).thenReturn(tableConfig)
+
when(tableConfig.getTableInitialVersion).thenReturn(HoodieTableVersion.NINE)
+ val fileIndex = mock(classOf[HoodieFileIndex])
+ when(fileIndex.getAllQueryPartitionPaths).thenReturn(
+ allPartitions.toSeq.map(path => new PartitionPath(path,
Array.empty[AnyRef])).asJava)
+ val support = new PartitionStatsIndexSupport(spark, sourceSchema, null,
+ HoodieMetadataConfig.newBuilder().build(), metaClient) {
+ override def isIndexAvailable: Boolean = true
+
+ override protected def shouldReadInMemory(fileIndex: HoodieFileIndex,
columns: Seq[String], threshold: Integer): Boolean = readInMemory
+
+ override def loadTransposed[T](targetColumns: Seq[String],
shouldReadInMemory: Boolean,
+ prunedPartitions: Option[Set[String]],
prunedFileNamesOpt: Option[Set[String]])(block: DataFrame => T): T = {
+ assertEquals(readInMemory, shouldReadInMemory)
+ block(indexDf)
+ }
+ }
+ (support, fileIndex)
+ }
+}
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/feature/TestDataSkippingQuery.scala
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/feature/TestDataSkippingQuery.scala
index 69f45fb0018f..bcef6459e144 100644
---
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/feature/TestDataSkippingQuery.scala
+++
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/feature/TestDataSkippingQuery.scala
@@ -19,8 +19,20 @@
package org.apache.spark.sql.hudi.feature
+import org.apache.hudi.{ColumnStatsIndexSupport, PartitionStatsIndexSupport}
+import org.apache.hudi.common.config.HoodieMetadataConfig
+import org.apache.hudi.common.fs.FSUtils
+import org.apache.hudi.common.model.HoodieLogFile
+import org.apache.hudi.common.table.TableSchemaResolver
+import org.apache.hudi.common.table.log.HoodieLogFormat
+import org.apache.hudi.common.table.log.block.HoodieDeleteBlock
+import org.apache.hudi.storage.StoragePath
+import org.apache.hudi.testutils.HoodieClientTestUtils.createMetaClient
+
import org.apache.spark.sql.hudi.common.HoodieSparkSqlTestBase
+import scala.collection.JavaConverters._
+
class TestDataSkippingQuery extends HoodieSparkSqlTestBase {
override protected def beforeAll(): Unit = {
@@ -128,6 +140,311 @@ class TestDataSkippingQuery extends
HoodieSparkSqlTestBase {
}
}
+ test("Data skipping retains files with missing column statistics after index
expansion") {
+ for (tableType <- Seq("cow", "mor"); partitionStats <- Seq(false, true)) {
+ withTempDir { tmp =>
+ val tableName = generateTableName
+ withSQLConf(
+ "hoodie.metadata.enable" -> "true",
+ "hoodie.metadata.index.column.stats.enable" -> "true",
+ "hoodie.metadata.index.partition.stats.enable" ->
partitionStats.toString,
+ "hoodie.fileIndex.dataSkippingFailureMode" -> "strict",
+ "hoodie.metadata.index.column.stats.column.list" -> "age",
+ "hoodie.enable.data.skipping" -> "false"
+ ) {
+ spark.sql(
+ s"""
+ |create table $tableName (
+ | id int,
+ | name string,
+ | age int,
+ | ts long,
+ | dt string
+ |) using hudi
+ | tblproperties (primaryKey = 'id', orderingFields = 'ts', type
= '$tableType')
+ | partitioned by (dt)
+ | location '${tmp.getCanonicalPath}'
+ |""".stripMargin)
+ // The data column exists from the first write, but only age is
indexed.
+ spark.sql(s"insert into $tableName values (1, 'alice', 25, 1,
'par1')")
+
+ withSQLConf("hoodie.metadata.index.column.stats.column.list" ->
"age,name") {
+ // A separate partition leaves the old file and its statistics
untouched.
+ spark.sql(s"insert into $tableName values (2, 'bob', 30, 2,
'par2')")
+ val metaClient = createMetaClient(spark, tmp.getCanonicalPath)
+ val indexedColumns =
metaClient.getIndexMetadata.get().getIndexDefinitions.get("column_stats").getSourceFields
+ assert(indexedColumns.contains("age") &&
indexedColumns.contains("name"), indexedColumns.toString)
+ val metadataConfig =
HoodieMetadataConfig.newBuilder().enable(true).build()
+ val index = new ColumnStatsIndexSupport(spark,
spark.table(tableName).schema,
+ new TableSchemaResolver(metaClient).getTableSchema,
metadataConfig, metaClient)
+ val stats = index.loadColumnStatsIndexRecords(Seq("age", "name"),
shouldReadInMemory = true)
+ .collectAsList().asScala
+ assert(stats.count(_.getColumnName == "age") == 2)
+ assert(stats.count(_.getColumnName == "name") == 1)
+
assert(stats.groupBy(_.getFileName).values.map(_.map(_.getColumnName).toSet).toSet
==
+ Set(Set("age"), Set("age", "name")))
+ val query = s"select id, name, age from $tableName where age >= 0
and name = 'alice'"
+ checkAnswer(query)(Seq(1, "alice", 25))
+
+ withSQLConf("hoodie.enable.data.skipping" -> "true") {
+ spark.sql(s"alter table $tableName set tblproperties
('hoodie.enable.data.skipping' = 'true')")
+ spark.catalog.refreshTable(tableName)
+ withClue("Name-only predicate with data skipping enabled: ") {
+ checkAnswer(s"select id, name, age from $tableName where name
= 'alice'")(Seq(1, "alice", 25))
+ }
+ // Adding a matching age predicate must not discard the
partially indexed file.
+ withClue("Combined age/name predicate with data skipping
enabled: ") {
+ checkAnswer(query)(Seq(1, "alice", 25))
+ }
+ }
+ }
+ }
+ }
+ }
+ }
+
+ test("Partition statistics invalidate incomplete columns after adding
indexed columns") {
+ Seq("cow", "mor").foreach { tableType =>
+ withTempDir { tmp =>
+ val tableName = generateTableName
+ withSQLConf(
+ "hoodie.metadata.enable" -> "true",
+ "hoodie.metadata.index.column.stats.enable" -> "true",
+ "hoodie.metadata.index.partition.stats.enable" -> "true",
+ "hoodie.metadata.index.column.stats.column.list" -> "age",
+ "hoodie.parquet.small.file.limit" -> "0",
+ "hoodie.enable.data.skipping" -> "false"
+ ) {
+ spark.sql(
+ s"""
+ |create table $tableName (id int, name string, age int, ts
long, dt string) using hudi
+ | tblproperties (primaryKey = 'id', orderingFields = 'ts', type
= '$tableType')
+ | partitioned by (dt)
+ | location '${tmp.getCanonicalPath}'
+ |""".stripMargin)
+ spark.sql(s"insert into $tableName values (1, 'alice', 25, 1,
'par1')")
+ withSQLConf("hoodie.metadata.index.column.stats.column.list" ->
"age,name") {
+ // Keep both files live in the same partition; rewriting the old
file would hide missing coverage.
+ spark.sql(s"insert into $tableName values (2, 'bob', 30, 2,
'par1')")
+ assert(spark.sql(s"select input_file_name() from
$tableName").collect().map(_.getString(0)).toSet.size == 2)
+ val metaClient = createMetaClient(spark, tmp.getCanonicalPath)
+ val metadataConfig =
HoodieMetadataConfig.newBuilder().enable(true).build()
+ val hoodieSchema = new
TableSchemaResolver(metaClient).getTableSchema
+ val index = new ColumnStatsIndexSupport(spark,
spark.table(tableName).schema, hoodieSchema, metadataConfig, metaClient)
+ val fileStats = index.loadColumnStatsIndexRecords(Seq("age",
"name"), shouldReadInMemory = true).collectAsList().asScala
+ assert(fileStats.count(_.getColumnName == "age") == 2)
+ assert(fileStats.count(_.getColumnName == "name") == 1)
+
assert(fileStats.groupBy(_.getFileName).values.map(_.map(_.getColumnName).toSet).toSet
==
+ Set(Set("age"), Set("age", "name")))
+ val partitions = new PartitionStatsIndexSupport(spark,
spark.table(tableName).schema, hoodieSchema, metadataConfig, metaClient)
+ val partitionStats =
partitions.loadColumnStatsIndexRecords(Seq("age", "name"), shouldReadInMemory =
true)
+ .collectAsList().asScala.filterNot(_.getIsDeleted)
+ assert(partitionStats.map(_.getColumnName).toSet == Set("age"),
partitionStats.toString)
+ assert(partitionStats.head.getValueCount == 2)
+ checkAnswer(s"select id, name, age from $tableName where name =
'alice'")(Seq(1, "alice", 25))
+ withSQLConf("hoodie.enable.data.skipping" -> "true",
"hoodie.fileIndex.dataSkippingFailureMode" -> "strict") {
+ spark.sql(s"alter table $tableName set tblproperties
('hoodie.enable.data.skipping' = 'true')")
+ spark.catalog.refreshTable(tableName)
+ checkAnswer(s"select id, name, age from $tableName where name =
'alice'")(Seq(1, "alice", 25))
+ checkAnswer(s"select id, name, age from $tableName where age >=
0 and name = 'alice'")(Seq(1, "alice", 25))
+ }
+ }
+ }
+ }
+ }
+ }
+
+ test("Data skipping retains old files after adding a column") {
+ Seq("cow", "mor").foreach { tableType =>
+ withTempDir { tmp =>
+ val tableName = generateTableName
+ withSQLConf(
+ "hoodie.metadata.enable" -> "true",
+ "hoodie.metadata.index.column.stats.enable" -> "true",
+ "hoodie.metadata.index.partition.stats.enable" -> "true",
+ "hoodie.parquet.small.file.limit" -> "0",
+ "hoodie.enable.data.skipping" -> "false",
+ "hoodie.fileIndex.dataSkippingFailureMode" -> "strict"
+ ) {
+ spark.sql(
+ s"""
+ |create table $tableName (id int, age int, ts long, dt string)
using hudi
+ | tblproperties (primaryKey = 'id', orderingFields = 'ts', type
= '$tableType')
+ | partitioned by (dt)
+ | location '${tmp.getCanonicalPath}'
+ |""".stripMargin)
+ spark.sql(s"insert into $tableName values (1, 25, 1, 'par1')")
+ spark.sql(s"alter table $tableName add columns (name string)")
+ // Keep the old file live alongside a new file written with the
evolved schema.
+ spark.sql(s"insert into $tableName values (2, 30, 2, 'bob', 'par1')")
+ assert(spark.sql(s"select input_file_name() from
$tableName").collect().map(_.getString(0)).toSet.size == 2)
+
+ val metaClient = createMetaClient(spark, tmp.getCanonicalPath)
+ val indexedColumns =
metaClient.getIndexMetadata.get().getIndexDefinitions.get("column_stats").getSourceFields
+ assert(indexedColumns.contains("age") &&
indexedColumns.contains("name"), indexedColumns.toString)
+ val metadataConfig =
HoodieMetadataConfig.newBuilder().enable(true).build()
+ val hoodieSchema = new TableSchemaResolver(metaClient).getTableSchema
+ val index = new ColumnStatsIndexSupport(spark,
spark.table(tableName).schema, hoodieSchema, metadataConfig, metaClient)
+ val fileStats = index.loadColumnStatsIndexRecords(Seq("age",
"name"), shouldReadInMemory = true).collectAsList().asScala
+
assert(fileStats.groupBy(_.getFileName).values.map(_.map(_.getColumnName).toSet).toSet
==
+ Set(Set("age"), Set("age", "name")), fileStats.toString)
+ val partitions = new PartitionStatsIndexSupport(spark,
spark.table(tableName).schema, hoodieSchema, metadataConfig, metaClient)
+ val partitionStats =
partitions.loadColumnStatsIndexRecords(Seq("age", "name"), shouldReadInMemory =
true)
+ .collectAsList().asScala.filterNot(_.getIsDeleted)
+ assert(partitionStats.map(_.getColumnName).toSet == Set("age"),
partitionStats.toString)
+ assert(partitionStats.head.getValueCount == 2)
+
+ def checkQueries(): Unit = {
+ checkAnswer(s"select id, age, name from $tableName where name is
null")(Seq(1, 25, null))
+ checkAnswer(s"select id, age, name from $tableName where name =
'bob'")(Seq(2, 30, "bob"))
+ checkAnswer(s"select id, age, name from $tableName where age >= 0
and (name is null or name = 'bob')")(
+ Seq(1, 25, null), Seq(2, 30, "bob"))
+ }
+ checkQueries()
+ withSQLConf("hoodie.enable.data.skipping" -> "true") {
+ spark.sql(s"alter table $tableName set tblproperties
('hoodie.enable.data.skipping' = 'true')")
+ spark.catalog.refreshTable(tableName)
+ checkQueries()
+ }
+ }
+ }
+ }
+ }
+
+ test("Data skipping retains MOR log files with missing column statistics") {
+ Seq(false, true).foreach { partitionStats =>
+ withTempDir { tmp =>
+ val tableName = generateTableName
+ withSQLConf(
+ "hoodie.metadata.enable" -> "true",
+ "hoodie.metadata.index.column.stats.enable" -> "true",
+ "hoodie.metadata.index.partition.stats.enable" ->
partitionStats.toString,
+ "hoodie.metadata.index.column.stats.column.list" -> "age",
+ "hoodie.compact.inline" -> "false",
+ "hoodie.compact.schedule.inline" -> "false",
+ "hoodie.enable.data.skipping" -> "false",
+ "hoodie.fileIndex.dataSkippingFailureMode" -> "strict"
+ ) {
+ spark.sql(
+ s"""
+ |create table $tableName (id int, name string, age int, ts
long, dt string) using hudi
+ | tblproperties (primaryKey = 'id', orderingFields = 'ts', type
= 'mor')
+ | partitioned by (dt)
+ | location '${tmp.getCanonicalPath}'
+ |""".stripMargin)
+ spark.sql(s"insert into $tableName values (1, 'before', 20, 1,
'par1')")
+ // Updating the existing key writes alice to a log before name is
indexed.
+ spark.sql(s"update $tableName set name = 'alice', age = 25, ts = 2
where id = 1")
+ withSQLConf("hoodie.metadata.index.column.stats.column.list" ->
"age,name") {
+ spark.sql(s"insert into $tableName values (2, 'bob', 30, 3,
'par2')")
+ val metaClient = createMetaClient(spark, tmp.getCanonicalPath)
+ val metadataConfig =
HoodieMetadataConfig.newBuilder().enable(true).build()
+ val index = new ColumnStatsIndexSupport(spark,
spark.table(tableName).schema,
+ new TableSchemaResolver(metaClient).getTableSchema,
metadataConfig, metaClient)
+ val stats = index.loadColumnStatsIndexRecords(Seq("age", "name"),
shouldReadInMemory = true).collectAsList().asScala
+ val partialLogs = stats.groupBy(_.getFileName).filter { case
(fileName, columns) =>
+ FSUtils.isLogFile(fileName) &&
columns.map(_.getColumnName).toSet == Set("age")
+ }
+ assert(partialLogs.nonEmpty, stats.toString)
+ val query = s"select id, name, age from $tableName where age >= 0
and name = 'alice'"
+ // The base file contains before, so this result proves the log
contributes the matching row.
+ checkAnswer(query)(Seq(1, "alice", 25))
+ withSQLConf("hoodie.enable.data.skipping" -> "true") {
+ spark.sql(s"alter table $tableName set tblproperties
('hoodie.enable.data.skipping' = 'true')")
+ spark.catalog.refreshTable(tableName)
+ checkAnswer(s"select id, name, age from $tableName where name =
'alice'")(Seq(1, "alice", 25))
+ checkAnswer(query)(Seq(1, "alice", 25))
+ }
+ }
+ }
+ }
+ }
+ }
+
+ test("Partition statistics survive MOR delete-only inline log appends") {
+ withTempDir { tmp =>
+ val tableName = generateTableName
+ withSQLConf(
+ "hoodie.metadata.enable" -> "true",
+ "hoodie.metadata.index.column.stats.enable" -> "true",
+ "hoodie.metadata.index.partition.stats.enable" -> "true",
+ "hoodie.metadata.index.column.stats.column.list" -> "age,name",
+ // Version 10 uses native logs; this regression must exercise
HoodieInlineLogAppendHandle.
+ "hoodie.write.table.version" -> "9",
+ "hoodie.compact.inline" -> "false",
+ "hoodie.compact.schedule.inline" -> "false",
+ "hoodie.insert.shuffle.parallelism" -> "1",
+ "hoodie.enable.data.skipping" -> "true"
+ ) {
+ spark.sql(
+ s"""
+ |create table $tableName (id int, name string, age int, ts long,
dt string) using hudi
+ | tblproperties (primaryKey = 'id', orderingFields = 'ts', type =
'mor',
+ | hoodie.write.table.version = '9')
+ | partitioned by (dt)
+ | location '${tmp.getCanonicalPath}'
+ |""".stripMargin)
+ spark.sql(s"insert into $tableName values (1, 'alice', 25, 1, 'par1'),
" +
+ "(2, 'bob', 30, 1, 'par1'), (3, 'carol', 35, 1, 'par1')")
+ val locations = spark.sql(s"select _hoodie_file_name,
_hoodie_partition_path from $tableName").collect()
+ assert(locations.length == 3 && locations.toSet.size == 1,
locations.toSeq.toString)
+ val baseFile = locations.head.getString(0)
+ val partitionPath = locations.head.getString(1)
+ assert(!FSUtils.isLogFile(baseFile))
+ val metadataConfig =
HoodieMetadataConfig.newBuilder().enable(true).build()
+ val indexedColumns = Seq("age", "name")
+ def loadPartitionStats() = {
+ val metaClient = createMetaClient(spark, tmp.getCanonicalPath)
+ val partitions = new PartitionStatsIndexSupport(spark,
spark.table(tableName).schema,
+ new TableSchemaResolver(metaClient).getTableSchema,
metadataConfig, metaClient)
+ partitions.loadColumnStatsIndexRecords(indexedColumns,
shouldReadInMemory = true)
+ .collectAsList().asScala.toSeq
+ }
+ val initialStats = loadPartitionStats()
+ assert(initialStats.size == 2 &&
initialStats.map(_.getColumnName).toSet == indexedColumns.toSet,
initialStats.toString)
+ initialStats.foreach { stats =>
+ assert(!stats.getIsDeleted)
+ assert(stats.getValueCount == 3 && stats.getNullCount == 0,
stats.toString)
+ assert(stats.getMinValue != null && stats.getMaxValue != null,
stats.toString)
+ }
+
+ spark.sql(s"delete from $tableName where id = 2")
+ val metaClient = createMetaClient(spark, tmp.getCanonicalPath)
+ val partitionFiles = metaClient.getStorage.listDirectEntries(new
StoragePath(tmp.getCanonicalPath, partitionPath)).asScala
+ val logs = partitionFiles.filter(info =>
FSUtils.isLogFile(info.getPath.getName)).map(new HoodieLogFile(_))
+ assert(logs.size == 1, partitionFiles.toString)
+ val log = logs.head
+ assert(!log.isNativeLogFile)
+ assert(log.getFileId == FSUtils.getFileId(baseFile))
+ val hoodieSchema = new TableSchemaResolver(metaClient).getTableSchema
+ val reader = HoodieLogFormat.newReader(metaClient, log, hoodieSchema)
+ try {
+ assert(reader.hasNext)
+ val block = reader.next()
+ assert(block.isInstanceOf[HoodieDeleteBlock],
block.getBlockType.toString)
+
assert(block.asInstanceOf[HoodieDeleteBlock].getRecordsToDelete.map(_.getRecordKey).toSet
== Set("2"))
+ assert(!reader.hasNext, "A delete-only append must not contain a
data block")
+ } finally {
+ reader.close()
+ }
+
+ // Empty stats must cover every indexed column, so the log does not
invalidate partition coverage.
+ val index = new ColumnStatsIndexSupport(spark,
spark.table(tableName).schema, hoodieSchema, metadataConfig, metaClient)
+ val logStats = index.loadColumnStatsIndexRecords(indexedColumns,
shouldReadInMemory = true)
+ .collectAsList().asScala.filter(_.getFileName == log.getFileName)
+ assert(logStats.size == 2 && logStats.map(_.getColumnName).toSet ==
indexedColumns.toSet, logStats.toString)
+ logStats.foreach { stats =>
+ assert(!stats.getIsDeleted)
+ assert(stats.getValueCount == 0 && stats.getNullCount == 0,
stats.toString)
+ assert(stats.getMinValue == null && stats.getMaxValue == null,
stats.toString)
+ }
+ assert(loadPartitionStats().sortBy(_.getColumnName) ==
initialStats.sortBy(_.getColumnName))
+ checkAnswer(s"select id, name, age from $tableName where age >= 0 and
name in ('alice', 'bob', 'carol')")(
+ Seq(1, "alice", 25), Seq(3, "carol", 35))
+ }
+ }
+ }
+
test("bucket index query") {
Seq("cow", "mor").foreach { tableType =>
// table bucket prop can not be read in the query sql now, so need to
set these configs
