yihua commented on a change in pull request #4948:
URL: https://github.com/apache/hudi/pull/4948#discussion_r823184442
##########
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): _*)
Review comment:
Got it, this is fine for now. I'm thinking from the perspective of
whether this can be reused for index on the write path.
--
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]
