hudi-agent commented on code in PR #19013:
URL: https://github.com/apache/hudi/pull/19013#discussion_r3584039374
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/DataSourceOptions.scala:
##########
@@ -190,6 +190,16 @@ object DataSourceReadOptions {
.withDocumentation("Enables data-skipping allowing queries to leverage
indexes to reduce the search space by " +
"skipping over files")
+ val RECORD_INDEX_MAX_DATA_SKIPPING_PARTITIONS: ConfigProperty[Integer] =
ConfigProperty
+ .key("hoodie.datasource.read.record.index.max.partitions")
+ .defaultValue(Integer.valueOf(3))
+ .markAdvanced()
+ .sinceVersion("1.3.0")
+ .withDocumentation("The maximum number of candidate data table partitions
that can be queried through the " +
+ "partitioned record level index for data skipping optimization. Each
candidate partition requires a separate " +
+ "metadata table lookup. If the number of candidate partitions exceeds
this value, the reader skips record " +
Review Comment:
🤖 It's not per Spark task — it's the number of candidate (pruned) data-table
partitions for a single query, evaluated on the driver at planning time. In
`PartitionedRecordLevelIndexSupport.filterQueriesWithRecordKey`, `partitions`
is derived from `prunedPartitionsAndFileSlices`, and if `partitions.size >
MAX_PARTITIONS` the RLI lookup is skipped (returns None) and the reader falls
back to other indexes. The threshold exists because each candidate partition
needs a separate metadata-table lookup, so it bounds the number of MDT lookups
per query resolution. Right now it's hard-coded to 10 in
`RecordLevelIndexSupport` rather than being read from a config.
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/PartitionedRecordLevelIndexSupport.scala:
##########
@@ -0,0 +1,120 @@
+/*
+ * 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.RecordLevelIndexSupport.{getPrunedStoragePaths,
MAX_PARTITIONS}
+import org.apache.hudi.common.config.HoodieMetadataConfig
+import org.apache.hudi.common.model.FileSlice
+import org.apache.hudi.common.table.HoodieTableMetaClient
+import org.apache.hudi.common.util.ValidationUtils
+import org.apache.hudi.common.util.collection.Pair
+import org.apache.hudi.core.read.BaseHoodieTableFileIndex
+import org.apache.hudi.index.PartitionedRecordIndexFileGroupLookupFunction
+import org.apache.hudi.metadata.{BucketizedMetadataTableFileGroupIndexParser,
HoodieTableMetadataUtil, MetadataPartitionType}
+
+import org.apache.spark.Partitioner
+import org.apache.spark.internal.Logging
+import org.apache.spark.sql.SparkSession
+
+import scala.collection.JavaConverters._
+import scala.collection.mutable
+
+/**
+ * Data skipping based on a partitioned Record Level Index (RLI), where the
file groups indexing the
+ * record keys are sharded per data-table partition. The metadata lookup must
therefore be scoped to each
+ * candidate partition.
+ *
+ * The candidate partitions are derived from the already pruned partitions.
Because each partition requires a
+ * separate metadata table lookup, if the number of candidate partitions
exceeds {@code MAX_PARTITIONS} the
+ * record index filtering is skipped (returns [[None]]) and the reader falls
back to other indexes.
+ */
+class PartitionedRecordLevelIndexSupport(spark: SparkSession,
+ metadataConfig: HoodieMetadataConfig,
+ metaClient: HoodieTableMetaClient)
+ extends RecordLevelIndexSupport(spark, metadataConfig, metaClient) with
Logging {
+
+ override protected def lookupCandidateFilesForRecordKeys(fileIndex:
HoodieFileIndex,
+
prunedPartitionsAndFileSlices:
Seq[(Option[BaseHoodieTableFileIndex.PartitionPath], Seq[FileSlice])],
+ recordKeys:
List[String]): Option[Set[String]] = {
+ val partitions = prunedPartitionsAndFileSlices.flatMap { case
(partitionPathOpt, _) =>
+ partitionPathOpt.map(_.getPath)
+ }.toSet
+ if (partitions.isEmpty) {
+ // Cannot resolve candidate partitions, fall back to other indexes
rather than over-pruning
+ Option.empty
+ } else if (partitions.size > MAX_PARTITIONS) {
+ logInfo(s"The number of candidate partitions ${partitions.size} exceeds
the partitioned record level index " +
+ s"lookup threshold $MAX_PARTITIONS. Skipping record level index
pruning.")
+ Option.empty
+ } else {
+ lookupRecordKeys(partitions, recordKeys) match {
+ case Some(fileIdToPartitionMap) =>
+ val prunedStoragePaths =
getPrunedStoragePaths(prunedPartitionsAndFileSlices, fileIndex)
+ Option.apply(filterCandidateFiles(prunedStoragePaths,
fileIdToPartitionMap))
+ case None =>
+ // None of the candidate partitions are indexed by the partitioned
RLI (e.g. partitions
+ // not yet indexed), so we cannot determine the matching files. Fall
back to other indexes
+ // rather than over-pruning to an empty candidate set.
+ Option.empty
+ }
+ }
+ }
+
+ private def lookupRecordKeys(partitions: Set[String],
+ recordKeys: List[String]):
Option[mutable.Map[String, String]] = {
+ val fileGroups =
metadataTable.getBucketizedFileGroupsForPartitionedRLI(MetadataPartitionType.RECORD_INDEX)
+ val fileGroupCountPerDataPartition = fileGroups.asScala
+ .filter { case (partition, _) => partitions.contains(partition) }
+ .map { case (partition, slices) => partition ->
Integer.valueOf(slices.size()) }
+ .toMap
+ if (fileGroupCountPerDataPartition.isEmpty) {
+ None
+ } else {
+ val numFileGroups =
BucketizedMetadataTableFileGroupIndexParser.calculateNumberOfFileGroups(fileGroupCountPerDataPartition.asJava)
+ val partitionOffsetIndexes =
BucketizedMetadataTableFileGroupIndexParser.generatePartitionToBaseIndexOffsets(fileGroupCountPerDataPartition.asJava).asScala
+ // Like SparkMetadataTableRecordLevelIndex#lookupRecords: build
(partition, recordKey)
+ // pairs, key by the global partitioned-RLI shard id, and let each Spark
partition look up
+ // one record-index shard through
PartitionedRecordIndexFileGroupLookupFunction.
+ val partitionRecordKeys =
fileGroupCountPerDataPartition.keys.toSeq.flatMap { partition =>
+ recordKeys.map { recordKey => Pair.of(partition, recordKey) }
+ }
+ val partitionedKeyRDD =
spark.sparkContext.parallelize(partitionRecordKeys, numFileGroups)
+ .keyBy { partitionRecordKey =>
+ val partition = partitionRecordKey.getLeft
+ partitionOffsetIndexes(partition).intValue() +
+
HoodieTableMetadataUtil.mapRecordKeyToFileGroupIndex(partitionRecordKey.getRight,
fileGroupCountPerDataPartition(partition).intValue())
+ }
+ .partitionBy(new PartitionIdPassthrough(numFileGroups))
+ .map(_._2)
+ .toJavaRDD()
+ ValidationUtils.checkState(partitionedKeyRDD.getNumPartitions <=
numFileGroups)
+ val fileIdToPartitionMap = partitionedKeyRDD.mapPartitionsToPair(new
PartitionedRecordIndexFileGroupLookupFunction(metadataTable))
+ .collect()
+ .asScala
+ .foldLeft(mutable.Map.empty[String, String]) { (acc, location) =>
+ acc.put(location._2.getFileId, location._2.getPartitionPath)
Review Comment:
🤖 nit: the `foldLeft` with `acc.put(…); acc` is a bit Java-ish for Scala —
`acc.put` returns `Option[V]` (the evicted value) which is silently discarded,
and then `acc` must be manually returned. Have you considered `.map(loc =>
loc._2.getFileId -> loc._2.getPartitionPath).to(mutable.Map)` for a more
idiomatic one-liner?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/RecordLevelIndexSupport.scala:
##########
@@ -56,48 +64,56 @@ class RecordLevelIndexSupport(spark: SparkSession,
shouldPushDownFilesFilter: Boolean
): Option[Set[String]] = {
lazy val (_, recordKeys) = filterQueriesWithRecordKey(queryFilters)
- val prunedStoragePaths =
getPrunedStoragePaths(prunedPartitionsAndFileSlices, fileIndex)
if (recordKeys.nonEmpty) {
- Option.apply(getCandidateFilesForRecordKeys(prunedStoragePaths,
recordKeys))
+ lookupCandidateFilesForRecordKeys(fileIndex,
prunedPartitionsAndFileSlices, recordKeys)
} else {
Option.empty
}
}
+ /**
+ * Looks up the candidate files which may store the provided record keys
from the record level index.
+ * Implemented differently for a global vs a partitioned RLI.
+ *
+ * @param fileIndex the file index of the query
+ * @param prunedPartitionsAndFileSlices already pruned partitions and file
slices
+ * @param recordKeys the record key literals extracted
from the query filters
+ * @return the set of candidate file names, or [[None]] if the index could
not be used and pruning
+ * should be skipped (falling back to other indexes).
+ */
+ protected def lookupCandidateFilesForRecordKeys(fileIndex: HoodieFileIndex,
+
prunedPartitionsAndFileSlices:
Seq[(Option[BaseHoodieTableFileIndex.PartitionPath], Seq[FileSlice])],
+ recordKeys: List[String]):
Option[Set[String]]
+
override def invalidateCaches(): Unit = {
// no caches for this index type, do nothing
}
/**
- * Returns the list of candidate files which store the provided record keys
based on Metadata Table Record Index.
- *
- * @param allFiles - List of all files which needs to be considered for
the query
- * @param recordKeys - List of record keys.
- * @return Sequence of file names which need to be queried
+ * Builds a map from fileId to data-table partition path from record index
lookup results.
*/
- private def getCandidateFilesForRecordKeys(allFiles: Seq[StoragePath],
recordKeys: List[String]): Set[String] = {
- val recordIndexData = metadataTable.readRecordIndexLocationsWithKeys(
-
HoodieListData.eager(JavaConverters.seqAsJavaListConverter(recordKeys).asJava))
- try {
- val recordKeyLocationsList =
HoodieDataUtils.dedupeAndCollectAsList(recordIndexData)
- val fileIdToPartitionMap: mutable.Map[String, String] = mutable.Map.empty
- val candidateFiles: mutable.Set[String] = mutable.Set.empty
- for (recordKeyLocation <- recordKeyLocationsList.asScala) {
- val location = recordKeyLocation.getValue
- fileIdToPartitionMap.put(location.getFileId, location.getPartitionPath)
- }
- for (file <- allFiles) {
- val fileId = FSUtils.getFileIdFromFilePath(file)
- val partitionOpt = fileIdToPartitionMap.get(fileId)
- if (partitionOpt.isDefined) {
- candidateFiles += file.getName
- }
+ protected def collectFileIdToPartitionMap(recordIndexData:
HoodiePairData[String, HoodieRecordGlobalLocation]): mutable.Map[String,
String] = {
+ val recordKeyLocationsList =
HoodieDataUtils.dedupeAndCollectAsList(recordIndexData)
+ val fileIdToPartitionMap: mutable.Map[String, String] = mutable.Map.empty
+ for (recordKeyLocation <- recordKeyLocationsList.asScala) {
+ val location = recordKeyLocation.getValue
+ fileIdToPartitionMap.put(location.getFileId, location.getPartitionPath)
+ }
+ fileIdToPartitionMap
+ }
+
+ /**
+ * Filters the input files, keeping only those whose fileId is present in
the record index lookup results.
+ */
+ protected def filterCandidateFiles(allFiles: Seq[StoragePath],
fileIdToPartitionMap: mutable.Map[String, String]): Set[String] = {
+ val candidateFiles: mutable.Set[String] = mutable.Set.empty
+ for (file <- allFiles) {
+ val fileId = FSUtils.getFileIdFromFilePath(file)
Review Comment:
🤖 nit: `fileIdToPartitionMap` is only read inside this method (`.contains`),
so declaring the parameter as `mutable.Map` leaks an implementation detail and
would prevent callers from passing an immutable `Map` in the future. Could you
widen it to `Map[String, String]`?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/RecordLevelIndexSupport.scala:
##########
@@ -133,6 +149,34 @@ class RecordLevelIndexSupport(spark: SparkSession,
object RecordLevelIndexSupport {
val INDEX_NAME = "RECORD_LEVEL"
+ /**
+ * Upper bound on the number of candidate data-table partitions eligible for
a partitioned RLI lookup.
+ *
+ * Unlike the global RLI (a single lookup over all keys), the partitioned
variant performs one metadata-table read
+ * per candidate partition. When a query does not filter on the partition
column the candidate set can span many
+ * partitions, and fanning out a lookup to each one can add latency that
outweighs the skipping benefit. Once the
+ * candidate partition count exceeds this threshold, pruning is skipped.
+ */
+ private[hudi] val MAX_PARTITIONS = 10
Review Comment:
🤖 nit: `MAX_PARTITIONS` is a bit generic when read in isolation — at the
call site it appears as `RecordLevelIndexSupport.MAX_PARTITIONS`, which doesn't
hint at what kind of partitions or why there's a max. Something like
`MAX_CANDIDATE_PARTITIONS` (or `PARTITIONED_RLI_PARTITION_THRESHOLD`) would
make it self-describing without needing to consult the Scaladoc.
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
--
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]
