You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by si...@apache.org on 2024/10/19 14:43:29 UTC
(hudi) branch master updated: [HUDI-8348] Whenever any new file slice is added, do the full computation of stats for the partitions touched (#12121)
This is an automated email from the ASF dual-hosted git repository.
sivabalan 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 070c600d79c [HUDI-8348] Whenever any new file slice is added, do the full computation of stats for the partitions touched (#12121)
070c600d79c is described below
commit 070c600d79ce5714c8787d423d913615add60fcc
Author: Lokesh Jain <lj...@apache.org>
AuthorDate: Sat Oct 19 20:13:22 2024 +0530
[HUDI-8348] Whenever any new file slice is added, do the full computation of stats for the partitions touched (#12121)
* [HUDI-8348] Whenever any new file slice is added, do the full computation of stats for the partitions touched
---------
Co-authored-by: sivabalan <n....@gmail.com>
Co-authored-by: Sagar Sumit <sa...@gmail.com>
---
.../hudi/common/config/HoodieMetadataConfig.java | 20 ++--
.../hudi/metadata/HoodieTableMetadataUtil.java | 29 +++++-
.../TestPartitionStatsIndexWithSql.scala | 101 ++++++++++++++++++++-
.../utilities/HoodieMetadataTableValidator.java | 4 +-
.../TestHoodieMetadataTableValidator.java | 10 +-
5 files changed, 144 insertions(+), 20 deletions(-)
diff --git a/hudi-common/src/main/java/org/apache/hudi/common/config/HoodieMetadataConfig.java b/hudi-common/src/main/java/org/apache/hudi/common/config/HoodieMetadataConfig.java
index c84723db8ed..332c26f6908 100644
--- a/hudi-common/src/main/java/org/apache/hudi/common/config/HoodieMetadataConfig.java
+++ b/hudi-common/src/main/java/org/apache/hudi/common/config/HoodieMetadataConfig.java
@@ -358,14 +358,14 @@ public final class HoodieMetadataConfig extends HoodieConfig {
.sinceVersion("1.0.0")
.withDocumentation("Parallelism to use, when generating partition stats index.");
- public static final ConfigProperty<Boolean> ENABLE_PARTITION_STATS_INDEX_TIGHT_BOUND = ConfigProperty
- .key(METADATA_PREFIX + ".index.partition.stats.tightBound.enable")
- .defaultValue(true)
+ public static final ConfigProperty<Boolean> PARTITION_STATS_INDEX_CONSOLIDATE_ON_EVERY_WRITE = ConfigProperty
+ .key(METADATA_PREFIX + ".index.partition.stats.consolidate.on.every.write")
+ .defaultValue(false)
.sinceVersion("1.0.0")
- .withDocumentation("Enable tight bound for the min/max value for each column at the storage partition level. "
- + "Typically, the min/max range for each column can become wider (i.e. the minValue is <= all valid values "
+ .withDocumentation("When enabled, partition stats is consolidated is computed on every commit for the min/max value of every column "
+ + "at the storage partition level. Typically, the min/max range for each column can become wider (i.e. the minValue is <= all valid values "
+ "in the file, and the maxValue >= all valid values in the file) with updates and deletes. If this config is enabled, "
- + "the min/max range will be updated to the tight bound of the valid values in the latest snapshot of the partition.");
+ + "the min/max range will be updated to the tight bound of the valid values after every commit for the partitions touched.");
public static final ConfigProperty<Boolean> SECONDARY_INDEX_ENABLE_PROP = ConfigProperty
.key(METADATA_PREFIX + ".index.secondary.enable")
@@ -534,8 +534,8 @@ public final class HoodieMetadataConfig extends HoodieConfig {
return getInt(PARTITION_STATS_INDEX_PARALLELISM);
}
- public boolean isPartitionStatsIndexTightBoundEnabled() {
- return getBooleanOrDefault(ENABLE_PARTITION_STATS_INDEX_TIGHT_BOUND);
+ public boolean isPartitionStatsIndexConsolidationEnabledOnEveryWrite() {
+ return getBooleanOrDefault(PARTITION_STATS_INDEX_CONSOLIDATE_ON_EVERY_WRITE);
}
public boolean isSecondaryIndexEnabled() {
@@ -749,8 +749,8 @@ public final class HoodieMetadataConfig extends HoodieConfig {
return this;
}
- public Builder withPartitionStatsIndexTightBound(boolean enable) {
- metadataConfig.setValue(ENABLE_PARTITION_STATS_INDEX_TIGHT_BOUND, String.valueOf(enable));
+ public Builder withPartitionStatsIndexConsolidationEnabledOnEveryWrite(boolean enable) {
+ metadataConfig.setValue(PARTITION_STATS_INDEX_CONSOLIDATE_ON_EVERY_WRITE, String.valueOf(enable));
return this;
}
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 21d7518669d..68042be8cfd 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
@@ -2201,7 +2201,7 @@ public class HoodieTableMetadataUtil {
int parallelism = Math.max(Math.min(partitionedWriteStats.size(), metadataConfig.getPartitionStatsIndexParallelism()), 1);
boolean shouldScanColStatsForTightBound = MetadataPartitionType.COLUMN_STATS.isMetadataPartitionAvailable(dataMetaClient)
- && metadataConfig.isPartitionStatsIndexTightBoundEnabled() && WriteOperationType.isPartitionStatsTightBoundRequired(commitMetadata.getOperationType());
+ && (metadataConfig.isPartitionStatsIndexConsolidationEnabledOnEveryWrite() || shouldConsolidatePartitionStats(commitMetadata, dataMetaClient));
HoodieTableMetadata tableMetadata;
if (shouldScanColStatsForTightBound) {
tableMetadata = HoodieTableMetadata.create(engineContext, dataMetaClient.getStorage(), metadataConfig, dataMetaClient.getBasePath().toString());
@@ -2218,7 +2218,7 @@ public class HoodieTableMetadataUtil {
checkState(tableMetadata != null, "tableMetadata should not be null when scanning metadata table");
// Collect Column Metadata for Each File part of active file system view of latest snapshot
// Get all file names, including log files, in a set from the file slices
- Set<String> fileNames = getPartitionFileSlices(dataMetaClient, Option.empty(), partitionName, true).stream()
+ Set<String> fileNames = getPartitionLatestFileSlicesIncludingInflight(dataMetaClient, Option.empty(), partitionName).stream()
.flatMap(fileSlice -> Stream.concat(
Stream.of(fileSlice.getBaseFile().map(HoodieBaseFile::getFileName).orElse(null)),
fileSlice.getLogFiles().map(HoodieLogFile::getFileName)))
@@ -2246,6 +2246,31 @@ public class HoodieTableMetadataUtil {
}
}
+ private static boolean shouldConsolidatePartitionStats(HoodieCommitMetadata commitMetadata, HoodieTableMetaClient metaClient) {
+ if (WriteOperationType.isPartitionStatsTightBoundRequired(commitMetadata.getOperationType())) {
+ return true;
+ }
+ // if not for compaction or clustering, lets check for any new base file added.
+ HoodieTableFileSystemView fsv = getFileSystemView(metaClient);
+ try {
+ fsv.loadPartitions(new ArrayList<>(commitMetadata.getPartitionToWriteStats().keySet()));
+ // Collect already existing filegroupIds.
+ Map<String, List<String>> partitionToExistingFileIds = new HashMap<>();
+ commitMetadata.getPartitionToWriteStats().keySet().forEach(partition -> {
+ partitionToExistingFileIds.put(partition, fsv.getLatestBaseFiles(partition).map(baseFile -> baseFile.getFileId()).collect(Collectors.toList()));
+ });
+ // check if new base file is added to any of existing file groups.
+ // as of now, if any one partition has a new base file added, we are triggering tighter bound computation for all partitions touched in this commit.
+ return commitMetadata.getPartitionToWriteStats().entrySet().stream().filter(entry -> {
+ List<String> existingFileIds = partitionToExistingFileIds.get(entry.getKey());
+ return entry.getValue().stream().filter(writeStat -> writeStat.getPath().contains(".parquet"))
+ .filter(writeStat -> existingFileIds.contains(writeStat.getFileId())).findAny().isPresent();
+ }).findAny().isPresent();
+ } finally {
+ fsv.close();
+ }
+ }
+
/**
* Generate key prefixes for each combination of column name in {@param columnsToIndex} and {@param partitionName}.
*/
diff --git a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestPartitionStatsIndexWithSql.scala b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestPartitionStatsIndexWithSql.scala
index 338600b8204..05b480c3542 100644
--- a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestPartitionStatsIndexWithSql.scala
+++ b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestPartitionStatsIndexWithSql.scala
@@ -278,6 +278,99 @@ class TestPartitionStatsIndexWithSql extends HoodieSparkSqlTestBase {
}
}
+ /**
+ * Test the partition stats consolidation:
+ *
+ * 1. Insert values: Initially insert records with values that create wide bounds for the price field across different partitions.
+ * 2. Update to widen the bounds: Perform updates to increase the price values, causing the min/max stats to widen.
+ * 3. Update to remove or lower the max value: Delete or update the record that holds the max value to simulate a
+ * scenario where the bounds need to be tightened.
+ * 4. Trigger stats recomputation: Assuming the config for tighter bounds is enabled, validate that the partition stats
+ * have adjusted correctly after scanning and recomputing accurate stats.
+ */
+ test("Test partition stats index with tight bound") {
+ Seq("cow", "mor").foreach { tableType =>
+ Seq("true", "false").foreach { isPartitionStatsIndexConsolidationEnabledOnEveryWrite =>
+ withTempDir { tmp =>
+ val tableName = generateTableName + s"_tight_bound_$isPartitionStatsIndexConsolidationEnabledOnEveryWrite"
+ val tablePath = s"${tmp.getCanonicalPath}/$tableName"
+ spark.sql(
+ s"""
+ |create table $tableName (
+ | id int,
+ | name string,
+ | price int,
+ | ts long
+ |) using hudi
+ |partitioned by (ts)
+ |tblproperties (
+ | type = '$tableType',
+ | primaryKey = 'id',
+ | preCombineField = 'price',
+ | hoodie.metadata.index.partition.stats.enable = 'true',
+ | hoodie.metadata.index.partition.stats.consolidate.on.every.write = '$isPartitionStatsIndexConsolidationEnabledOnEveryWrite',
+ | hoodie.metadata.index.column.stats.enable = 'true',
+ | hoodie.metadata.index.column.stats.column.list = 'price'
+ |)
+ |location '$tablePath'
+ |""".stripMargin
+ )
+
+ /**
+ * Insert: Insert values for price across multiple partitions (ts=10, ts=20, ts=30):
+ *
+ * Partition ts=10: price = 1000, 1500
+ * Partition ts=20: price = 2000, 2500
+ * Partition ts=30: price = 3000, 3500
+ *
+ * This will initialize the partition stats with bounds like [1000, 1500], [2000, 2500], and [3000, 3500].
+ */
+ spark.sql(s"insert into $tableName values (1, 'a1', 1000, 10), (2, 'a2', 1500, 10)")
+ spark.sql(s"insert into $tableName values (3, 'a3', 2000, 20), (4, 'a4', 2500, 20)")
+ spark.sql(s"insert into $tableName values (5, 'a5', 3000, 30), (6, 'a6', 3500, 30)")
+
+ // validate partition stats initialization
+ checkAnswer(s"select key, ColumnStatsMetadata.minValue.member1.value, ColumnStatsMetadata.maxValue.member1.value from hudi_metadata('$tableName') where type=${MetadataPartitionType.PARTITION_STATS.getRecordType} and ColumnStatsMetadata.columnName='price'")(
+ Seq(getPartitionStatsIndexKey("ts=10", "price"), 1000, 1500),
+ Seq(getPartitionStatsIndexKey("ts=20", "price"), 2000, 2500),
+ Seq(getPartitionStatsIndexKey("ts=30", "price"), 3000, 3500)
+ )
+
+ // First Update (widen the bounds): Update the price in partition ts=30, where price = 4000 for id=6.
+ // This will widen the max bounds in ts=30 from 3500 to 4000.
+ spark.sql(s"update $tableName set price = 4000 where id = 6")
+ // Validate widened stats
+ checkAnswer(s"select key, ColumnStatsMetadata.minValue.member1.value, ColumnStatsMetadata.maxValue.member1.value, ColumnStatsMetadata.isTightBound from hudi_metadata('$tableName') where type=${MetadataPartitionType.PARTITION_STATS.getRecordType} and ColumnStatsMetadata.columnName='price'")(
+ Seq(getPartitionStatsIndexKey("ts=10", "price"), 1000, 1500, true),
+ Seq(getPartitionStatsIndexKey("ts=20", "price"), 2000, 2500, true),
+ // for COW table, that stats are consolidated on every write as there is a new slice for the same filegroup
+ Seq(getPartitionStatsIndexKey("ts=30", "price"), 3000, 4000, isPartitionStatsIndexConsolidationEnabledOnEveryWrite.toBoolean || tableType == "cow")
+ )
+ // verify file pruning
+ var metaClient = HoodieTableMetaClient.builder()
+ .setBasePath(tablePath)
+ .setConf(HoodieTestUtils.getDefaultStorageConf)
+ .build()
+ verifyFilePruning(Map.apply(DataSourceReadOptions.ENABLE_DATA_SKIPPING.key -> "true", HoodieMetadataConfig.ENABLE.key -> "true"),
+ GreaterThan(AttributeReference("price", IntegerType)(), Literal(3000)), metaClient, isDataSkippingExpected = true)
+
+ // Second update (reduce max value)
+ spark.sql(s"delete from $tableName where id = 6")
+ // Validate that stats have recomputed and tightened
+ checkAnswer(s"select key, ColumnStatsMetadata.minValue.member1.value, ColumnStatsMetadata.maxValue.member1.value, ColumnStatsMetadata.isTightBound from hudi_metadata('$tableName') where type=${MetadataPartitionType.PARTITION_STATS.getRecordType} and ColumnStatsMetadata.columnName='price'")(
+ Seq(getPartitionStatsIndexKey("ts=10", "price"), 1000, 1500, true),
+ Seq(getPartitionStatsIndexKey("ts=20", "price"), 2000, 2500, true),
+ Seq(getPartitionStatsIndexKey("ts=30", "price"), 3000, 3000, true) // tighter bound, note that record with prev max was deleted
+ )
+ // verify file pruning
+ metaClient = HoodieTableMetaClient.reload(metaClient)
+ verifyFilePruning(Map.apply(DataSourceReadOptions.ENABLE_DATA_SKIPPING.key -> "true", HoodieMetadataConfig.ENABLE.key -> "true"),
+ GreaterThan(AttributeReference("price", IntegerType)(), Literal(3000)), metaClient, isDataSkippingExpected = true, isNoScanExpected = true)
+ }
+ }
+ }
+ }
+
/**
* 1. Create MOR table with compaction enabled.
* 2. Do an insert and validate the partition stats index initialization.
@@ -352,8 +445,8 @@ class TestPartitionStatsIndexWithSql extends HoodieSparkSqlTestBase {
// validate partition stats index
checkAnswer(s"select key, ColumnStatsMetadata.minValue.member1.value, ColumnStatsMetadata.maxValue.member1.value, ColumnStatsMetadata.isTightBound from hudi_metadata('$tableName') where type=${MetadataPartitionType.PARTITION_STATS.getRecordType} and ColumnStatsMetadata.columnName='price'")(
- Seq(getPartitionStatsIndexKey("ts=10", "price"), 1000, 2000, false),
- Seq(getPartitionStatsIndexKey("ts=20", "price"), 2000, 3000, false),
+ Seq(getPartitionStatsIndexKey("ts=10", "price"), 1000, 2000, true),
+ Seq(getPartitionStatsIndexKey("ts=20", "price"), 2000, 3000, true),
Seq(getPartitionStatsIndexKey("ts=30", "price"), 3000, 4003, false)
)
}
@@ -398,8 +491,8 @@ class TestPartitionStatsIndexWithSql extends HoodieSparkSqlTestBase {
if (tableType == "mor" && shouldCompact) {
// check partition stats records with tightBound
checkAnswer(s"select key, ColumnStatsMetadata.minValue.member1.value, ColumnStatsMetadata.maxValue.member1.value, ColumnStatsMetadata.isTightBound from hudi_metadata('$tableName') where type=${MetadataPartitionType.PARTITION_STATS.getRecordType} and ColumnStatsMetadata.columnName='price'")(
- Seq(getPartitionStatsIndexKey("ts=10", "price"), 1000, 2000, false),
- Seq(getPartitionStatsIndexKey("ts=20", "price"), 2000, 3000, false),
+ Seq(getPartitionStatsIndexKey("ts=10", "price"), 1000, 2000, true),
+ Seq(getPartitionStatsIndexKey("ts=20", "price"), 2000, 3000, true),
Seq(getPartitionStatsIndexKey("ts=30", "price"), 3000, 4001, true)
)
}
diff --git a/hudi-utilities/src/main/java/org/apache/hudi/utilities/HoodieMetadataTableValidator.java b/hudi-utilities/src/main/java/org/apache/hudi/utilities/HoodieMetadataTableValidator.java
index 4cee0815e46..5459e3359a2 100644
--- a/hudi-utilities/src/main/java/org/apache/hudi/utilities/HoodieMetadataTableValidator.java
+++ b/hudi-utilities/src/main/java/org/apache/hudi/utilities/HoodieMetadataTableValidator.java
@@ -980,7 +980,9 @@ public class HoodieMetadataTableValidator implements Serializable {
AvroConversionUtils.convertAvroSchemaToStructType(metadataTableBasedContext.getSchema()), metadataTableBasedContext.getMetadataConfig(),
metaClient, false);
HoodieData<HoodieMetadataColumnStats> partitionStats =
- partitionStatsIndexSupport.loadColumnStatsIndexRecords(JavaConverters.asScalaBufferConverter(metadataTableBasedContext.allColumnNameList).asScala().toSeq(), scala.Option.empty(), false);
+ partitionStatsIndexSupport.loadColumnStatsIndexRecords(JavaConverters.asScalaBufferConverter(metadataTableBasedContext.allColumnNameList).asScala().toSeq(), scala.Option.empty(), false)
+ // set isTightBound to false since partition stats generated using column stats does not contain the field
+ .map(colStat -> HoodieMetadataColumnStats.newBuilder(colStat).setIsTightBound(false).build());
JavaRDD<HoodieMetadataColumnStats> diffRDD = HoodieJavaRDD.getJavaRDD(partitionStats).subtract(HoodieJavaRDD.getJavaRDD(partitionStatsUsingColStats));
if (!diffRDD.isEmpty()) {
List<HoodieMetadataColumnStats> diff = diffRDD.collect();
diff --git a/hudi-utilities/src/test/java/org/apache/hudi/utilities/TestHoodieMetadataTableValidator.java b/hudi-utilities/src/test/java/org/apache/hudi/utilities/TestHoodieMetadataTableValidator.java
index e0b0255a6f8..9fad1385400 100644
--- a/hudi-utilities/src/test/java/org/apache/hudi/utilities/TestHoodieMetadataTableValidator.java
+++ b/hudi-utilities/src/test/java/org/apache/hudi/utilities/TestHoodieMetadataTableValidator.java
@@ -288,13 +288,17 @@ public class TestHoodieMetadataTableValidator extends HoodieSparkClientTestBase
return rows;
}
- @Test
- public void testPartitionStatsValidation() {
+ @ParameterizedTest
+ @ValueSource(strings = {"MERGE_ON_READ", "COPY_ON_WRITE"})
+ public void testPartitionStatsValidation(String tableType) {
// TODO: Add validation for compaction and clustering cases
Map<String, String> writeOptions = new HashMap<>();
writeOptions.put(DataSourceWriteOptions.TABLE_NAME().key(), "test_table");
writeOptions.put("hoodie.table.name", "test_table");
- writeOptions.put(DataSourceWriteOptions.TABLE_TYPE().key(), "MERGE_ON_READ");
+ writeOptions.put(DataSourceWriteOptions.TABLE_TYPE().key(), tableType);
+ if (tableType.equals("COPY_ON_WRITE")) {
+ writeOptions.put(HoodieMetadataConfig.PARTITION_STATS_INDEX_CONSOLIDATE_ON_EVERY_WRITE.key(), "true");
+ }
writeOptions.put(DataSourceWriteOptions.RECORDKEY_FIELD().key(), "_row_key");
writeOptions.put(DataSourceWriteOptions.PRECOMBINE_FIELD().key(), "timestamp");
writeOptions.put(DataSourceWriteOptions.PARTITIONPATH_FIELD().key(), "partition_path");