You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by st...@apache.org on 2024/01/23 03:14:18 UTC
(hudi) branch master updated: [HUDI-7310] Optimize Column Stats Partition Pruning for Non-Partition Pruning Queries (#10528)
This is an automated email from the ASF dual-hosted git repository.
stream2000 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 9d829c37e82 [HUDI-7310] Optimize Column Stats Partition Pruning for Non-Partition Pruning Queries (#10528)
9d829c37e82 is described below
commit 9d829c37e82a88b3676983938271b7131d3a3897
Author: majian <47...@users.noreply.github.com>
AuthorDate: Tue Jan 23 11:14:11 2024 +0800
[HUDI-7310] Optimize Column Stats Partition Pruning for Non-Partition Pruning Queries (#10528)
* [HUDI-7310] Optimize Column Stats Partition Pruning for Non-Partition Pruning Queries
* [HUDI-7310] Optimize Column Stats Partition Pruning for Non-Partition Pruning Queries
* [HUDI-7310] Optimize Column Stats Partition Pruning for Non-Partition Pruning Queries
* [HUDI-7310] Optimize Column Stats Partition Pruning for Non-Partition Pruning Queries
* [HUDI-7310] Optimize Column Stats Partition Pruning for Non-Partition Pruning Queries
* [HUDI-7310] Optimize Column Stats Partition Pruning for Non-Partition Pruning Queries
* [HUDI-7310] Optimize Column Stats Partition Pruning for Non-Partition Pruning Queries
---
.../org/apache/hudi/ColumnStatsIndexSupport.scala | 23 +++++----
.../scala/org/apache/hudi/HoodieFileIndex.scala | 15 ++++--
.../hudi/functional/TestColumnStatsIndex.scala | 59 ++++++++++++++++++++++
3 files changed, 82 insertions(+), 15 deletions(-)
diff --git a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/ColumnStatsIndexSupport.scala b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/ColumnStatsIndexSupport.scala
index 0a6cf437a6f..dc15a3e8c8c 100644
--- a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/ColumnStatsIndexSupport.scala
+++ b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/ColumnStatsIndexSupport.scala
@@ -107,22 +107,23 @@ class ColumnStatsIndexSupport(spark: SparkSession,
*
* Please check out scala-doc of the [[transpose]] method explaining this view in more details
*/
- def loadTransposed[T](targetColumns: Seq[String], shouldReadInMemory: Boolean, prunedFileNames: Set[String] = Set.empty)(block: DataFrame => T): T = {
+ def loadTransposed[T](targetColumns: Seq[String],
+ shouldReadInMemory: Boolean,
+ prunedFileNamesOpt: Option[Set[String]] = None)(block: DataFrame => T): T = {
cachedColumnStatsIndexViews.get(targetColumns) match {
case Some(cachedDF) =>
block(cachedDF)
-
case None =>
- val colStatsRecords: HoodieData[HoodieMetadataColumnStats] = if (prunedFileNames.isEmpty) {
- // NOTE: Because some tests directly check this method and don't get prunedPartitionsAndFileSlices, we need to make sure these tests are correct.
- loadColumnStatsIndexRecords(targetColumns, shouldReadInMemory)
- } else {
- val filterFunction = new SerializableFunction[HoodieMetadataColumnStats, java.lang.Boolean] {
- override def apply(r: HoodieMetadataColumnStats): java.lang.Boolean = {
- prunedFileNames.contains(r.getFileName)
+ val colStatsRecords: HoodieData[HoodieMetadataColumnStats] = prunedFileNamesOpt match {
+ case Some(prunedFileNames) =>
+ val filterFunction = new SerializableFunction[HoodieMetadataColumnStats, java.lang.Boolean] {
+ override def apply(r: HoodieMetadataColumnStats): java.lang.Boolean = {
+ prunedFileNames.contains(r.getFileName)
+ }
}
- }
- loadColumnStatsIndexRecords(targetColumns, shouldReadInMemory).filter(filterFunction)
+ loadColumnStatsIndexRecords(targetColumns, shouldReadInMemory).filter(filterFunction)
+ case None =>
+ loadColumnStatsIndexRecords(targetColumns, shouldReadInMemory)
}
withPersistedData(colStatsRecords, StorageLevel.MEMORY_ONLY) {
diff --git a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieFileIndex.scala b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieFileIndex.scala
index db8525be3d1..3444feaecff 100644
--- a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieFileIndex.scala
+++ b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieFileIndex.scala
@@ -233,8 +233,9 @@ case class HoodieFileIndex(spark: SparkSession,
// - Col-Stats Index is present
// - Record-level Index is present
// - List of predicates (filters) is present
+ val shouldPushDownFilesFilter = !partitionFilters.isEmpty
val candidateFilesNamesOpt: Option[Set[String]] =
- lookupCandidateFilesInMetadataTable(dataFilters, prunedPartitionsAndFileSlices) match {
+ lookupCandidateFilesInMetadataTable(dataFilters, prunedPartitionsAndFileSlices, shouldPushDownFilesFilter) match {
case Success(opt) => opt
case Failure(e) =>
logError("Failed to lookup candidate files in File Index", e)
@@ -328,7 +329,9 @@ case class HoodieFileIndex(spark: SparkSession,
* @param queryFilters list of original data filters passed down from querying engine
* @return list of pruned (data-skipped) candidate base-files and log files' names
*/
- private def lookupCandidateFilesInMetadataTable(queryFilters: Seq[Expression], prunedPartitionsAndFileSlices: Seq[(Option[BaseHoodieTableFileIndex.PartitionPath], Seq[FileSlice])]): Try[Option[Set[String]]] = Try {
+ private def lookupCandidateFilesInMetadataTable(queryFilters: Seq[Expression],
+ prunedPartitionsAndFileSlices: Seq[(Option[BaseHoodieTableFileIndex.PartitionPath], Seq[FileSlice])],
+ shouldPushDownFilesFilter: Boolean): Try[Option[Set[String]]] = Try {
// NOTE: For column stats, Data Skipping is only effective when it references columns that are indexed w/in
// the Column Stats Index (CSI). Following cases could not be effectively handled by Data Skipping:
// - Expressions on top-level column's fields (ie, for ex filters like "struct.field > 0", since
@@ -361,9 +364,13 @@ case class HoodieFileIndex(spark: SparkSession,
// For that we use a simple-heuristic to determine whether we should read and process CSI in-memory or
// on-cluster: total number of rows of the expected projected portion of the index has to be below the
// threshold (of 100k records)
- val prunedFileNames = getPrunedFileNames(prunedPartitionsAndFileSlices)
val shouldReadInMemory = columnStatsIndex.shouldReadInMemory(this, queryReferencedColumns)
- columnStatsIndex.loadTransposed(queryReferencedColumns, shouldReadInMemory, prunedFileNames) { transposedColStatsDF =>
+ val prunedFileNames = getPrunedFileNames(prunedPartitionsAndFileSlices)
+ // NOTE: If partition pruning doesn't prune any files, then there's no need to apply file filters
+ // when loading the Column Statistics Index
+ val prunedFileNamesOpt = if (shouldPushDownFilesFilter) Some(prunedFileNames) else None
+
+ columnStatsIndex.loadTransposed(queryReferencedColumns, shouldReadInMemory, prunedFileNamesOpt) { transposedColStatsDF =>
Some(getCandidateFiles(transposedColStatsDF, queryFilters, prunedFileNames))
}
}
diff --git a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestColumnStatsIndex.scala b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestColumnStatsIndex.scala
index ac83cf81918..3c80a7faf5b 100644
--- a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestColumnStatsIndex.scala
+++ b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestColumnStatsIndex.scala
@@ -139,6 +139,65 @@ class TestColumnStatsIndex extends ColumnStatIndexTestBase {
}
}
+ @ParameterizedTest
+ @ValueSource(booleans = Array(true, false))
+ def testMetadataColumnStatsWithFilesFilter(shouldReadInMemory: Boolean): Unit = {
+ val targetColumnsToIndex = Seq("c1", "c2", "c3")
+
+ val metadataOpts = Map(
+ HoodieMetadataConfig.ENABLE.key -> "true",
+ HoodieMetadataConfig.ENABLE_METADATA_INDEX_COLUMN_STATS.key -> "true",
+ HoodieMetadataConfig.COLUMN_STATS_INDEX_FOR_COLUMNS.key -> targetColumnsToIndex.mkString(",")
+ )
+
+ val opts = Map(
+ "hoodie.insert.shuffle.parallelism" -> "4",
+ "hoodie.upsert.shuffle.parallelism" -> "4",
+ HoodieWriteConfig.TBL_NAME.key -> "hoodie_test",
+ RECORDKEY_FIELD.key -> "c1",
+ PRECOMBINE_FIELD.key -> "c1",
+ HoodieTableConfig.POPULATE_META_FIELDS.key -> "true",
+ HoodieCommonConfig.RECONCILE_SCHEMA.key -> "true"
+ ) ++ metadataOpts
+
+ val sourceJSONTablePath = getClass.getClassLoader.getResource("index/colstats/input-table-json").toString
+
+ // NOTE: Schema here is provided for validation that the input date is in the appropriate format
+ val inputDF = spark.read.schema(sourceTableSchema).json(sourceJSONTablePath)
+
+ inputDF
+ .sort("c1")
+ .repartition(4, new Column("c1"))
+ .write
+ .format("hudi")
+ .options(opts)
+ .option(HoodieStorageConfig.PARQUET_MAX_FILE_SIZE.key, 10 * 1024)
+ .option(DataSourceWriteOptions.OPERATION.key, DataSourceWriteOptions.INSERT_OPERATION_OPT_VAL)
+ .mode(SaveMode.Overwrite)
+ .save(basePath)
+
+ metaClient = HoodieTableMetaClient.reload(metaClient)
+
+ val metadataConfig = HoodieMetadataConfig.newBuilder()
+ .fromProperties(toProperties(metadataOpts))
+ .build()
+ val columnStatsIndex = new ColumnStatsIndexSupport(spark, sourceTableSchema, metadataConfig, metaClient)
+ val requestedColumns = Seq("c1")
+ // get all file names
+ val stringEncoder: Encoder[String] = org.apache.spark.sql.Encoders.STRING
+ val fileNameSet: Set[String] = columnStatsIndex.loadTransposed(requestedColumns, shouldReadInMemory) { df =>
+ val fileNames: Array[String] = df.select("fileName").as[String](stringEncoder).collect()
+ val newFileNameSet: Set[String] = fileNames.toSet
+ newFileNameSet
+ }
+ val targetFileName = fileNameSet.take(2)
+ columnStatsIndex.loadTransposed(requestedColumns, shouldReadInMemory, Some(targetFileName)) { df =>
+ assertEquals(2, df.collect().length)
+ val targetDFFileNameSet: Set[String] = df.select("fileName").as[String](stringEncoder).collect().toSet
+ assertEquals(targetFileName, targetDFFileNameSet)
+ }
+ }
+
@ParameterizedTest
@ValueSource(booleans = Array(true, false))
def testMetadataColumnStatsIndexPartialProjection(shouldReadInMemory: Boolean): Unit = {