alexeykudinkin commented on a change in pull request #4948:
URL: https://github.com/apache/hudi/pull/4948#discussion_r822166241
##########
File path:
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieFileIndex.scala
##########
@@ -194,77 +192,102 @@ 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' names
*/
- private def lookupCandidateFilesInColStatsIndex(queryFilters:
Seq[Expression]): Try[Option[Set[String]]] = Try {
- val indexPath = metaClient.getColumnStatsIndexPath
+ private def lookupCandidateFilesInMetadataTable(queryFilters:
Seq[Expression]): Try[Option[Set[String]]] = Try {
val fs = metaClient.getFs
+ val metadataTablePath =
HoodieTableMetadata.getMetadataTableBasePath(basePath)
- if (!enableDataSkipping() || !fs.exists(new Path(indexPath)) ||
queryFilters.isEmpty) {
- // scalastyle:off return
- return Success(Option.empty)
- // scalastyle:on return
- }
-
- val completedCommits =
getActiveTimeline.filterCompletedInstants().getInstants.iterator.asScala.toList.map(_.getTimestamp)
-
- // Collect all index tables present in `.zindex` folder
- val candidateIndexTables =
- fs.listStatus(new Path(indexPath))
- .filter(_.isDirectory)
- .map(_.getPath.getName)
- .filter(completedCommits.contains(_))
- .sortBy(x => x)
-
- if (candidateIndexTables.isEmpty) {
- // scalastyle:off return
- return Success(Option.empty)
- // scalastyle:on return
- }
-
- val dataFrameOpt = try {
- Some(spark.read.load(new Path(indexPath,
candidateIndexTables.last).toString))
- } catch {
- case t: Throwable =>
- logError("Failed to read col-stats index; skipping", t)
- None
+ if (!isDataSkippingEnabled() || !fs.exists(new Path(metadataTablePath)) ||
queryFilters.isEmpty) {
+ Option.empty
+ } else {
+ val targetColStatsIndexColumns = Seq(
+ HoodieMetadataPayload.COLUMN_STATS_FIELD_FILE_NAME,
+ HoodieMetadataPayload.COLUMN_STATS_FIELD_MIN_VALUE,
+ HoodieMetadataPayload.COLUMN_STATS_FIELD_MAX_VALUE,
+ HoodieMetadataPayload.COLUMN_STATS_FIELD_NULL_COUNT)
+
+ val requiredMetadataIndexColumns =
+ (targetColStatsIndexColumns :+
HoodieMetadataPayload.COLUMN_STATS_FIELD_COLUMN_NAME).map(colName =>
+ s"${HoodieMetadataPayload.SCHEMA_FIELD_ID_COLUMN_STATS}.${colName}")
+
+ // Read Metadata Table's Column Stats Index into Spark's [[DataFrame]]
+ val metadataTableDF = spark.read.format("org.apache.hudi")
+
.load(s"$metadataTablePath/${MetadataPartitionType.COLUMN_STATS.getPartitionPath}")
+
+ // TODO filter on (column, partition) prefix
+ val colStatsDF =
metadataTableDF.where(col(HoodieMetadataPayload.SCHEMA_FIELD_ID_COLUMN_STATS).isNotNull)
+ .select(requiredMetadataIndexColumns.map(col): _*)
+
+ val queryReferencedColumns = collectReferencedColumns(spark,
queryFilters, schema)
+
+ // Persist DF to avoid re-computing column statistics unraveling
+ withPersistence(colStatsDF) {
+ // Metadata Table bears rows in the following format
+ //
+ //
+---------------------------+------------+------------+------------+-------------+
+ // | fileName | columnName | minValue | maxValue
| num_nulls |
+ //
+---------------------------+------------+------------+------------+-------------+
+ // | one_base_file.parquet | A | 1 | 10
| 0 |
+ // | another_base_file.parquet | A | -10 | 0
| 5 |
+ //
+---------------------------+------------+------------+------------+-------------+
+ //
+ // While Data Skipping utils are expecting following (transposed)
format, where per-column stats are
+ // essentially transposed (from rows to columns):
+ //
+ //
+---------------------------+------------+------------+-------------+
+ // | file | A_minValue | A_maxValue |
A_num_nulls |
+ //
+---------------------------+------------+------------+-------------+
+ // | one_base_file.parquet | 1 | 10 |
0 |
+ // | another_base_file.parquet | -10 | 0 |
5 |
+ //
+---------------------------+------------+------------+-------------+
+ //
+ // NOTE: Column Stats Index might potentially contain statistics for
many columns (if not all), while
+ // query at hand might only be referencing a handful of those.
As such, we collect all the
+ // column references from the filtering expressions, and only
transpose records corresponding to the
+ // columns referenced in those
+ val transposedColStatsDF =
Review comment:
Correct. It also makes code much simpler (otherwise you need to do rows
intersection for every column, that makes code more involved)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
