hudi-agent commented on code in PR #19013: URL: https://github.com/apache/hudi/pull/19013#discussion_r3579562069
########## 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]) { (fileIdToPartitionMap, location) => Review Comment: 🤖 nit: the `foldLeft` accumulator parameter is also named `fileIdToPartitionMap`, the same as the `val` being assigned on line 106. Could you rename the accumulator to `acc` (or `map`) to make it obvious which `fileIdToPartitionMap.put(...)` on the next line refers to the accumulator and not the outer binding? <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/GlobalRecordLevelIndexSupport.scala: ########## @@ -0,0 +1,54 @@ +/* + * 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 +import org.apache.hudi.common.config.HoodieMetadataConfig +import org.apache.hudi.common.data.HoodieListData +import org.apache.hudi.common.model.FileSlice +import org.apache.hudi.common.table.HoodieTableMetaClient +import org.apache.hudi.core.read.BaseHoodieTableFileIndex + +import org.apache.spark.sql.SparkSession + +import scala.collection.JavaConverters + +/** + * Data skipping based on a global Record Level Index (RLI), where a single set of file groups indexes + * the record keys across the whole table. All record keys are resolved with one metadata table lookup. + */ +class GlobalRecordLevelIndexSupport(spark: SparkSession, + metadataConfig: HoodieMetadataConfig, + metaClient: HoodieTableMetaClient) + extends RecordLevelIndexSupport(spark, metadataConfig, metaClient) { + + override protected def lookupCandidateFilesForRecordKeys(fileIndex: HoodieFileIndex, + prunedPartitionsAndFileSlices: Seq[(Option[BaseHoodieTableFileIndex.PartitionPath], Seq[FileSlice])], + recordKeys: List[String]): Option[Set[String]] = { + val prunedStoragePaths = getPrunedStoragePaths(prunedPartitionsAndFileSlices, fileIndex) + val recordIndexData = metadataTable.readRecordIndexLocationsWithKeys( + HoodieListData.eager(JavaConverters.seqAsJavaListConverter(recordKeys).asJava)) Review Comment: 🤖 nit: `JavaConverters.seqAsJavaListConverter(recordKeys).asJava` is the verbose form — sibling files (`RecordLevelIndexSupport.scala`, `PartitionedRecordLevelIndexSupport.scala`) use `import scala.collection.JavaConverters._` and just write `recordKeys.asJava`. Could you align the import style here? <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/PartitionedRecordLevelIndexSupport.scala: ########## @@ -0,0 +1,77 @@ +/* + * 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 +import org.apache.hudi.common.config.HoodieMetadataConfig +import org.apache.hudi.common.data.HoodieListData +import org.apache.hudi.common.model.FileSlice +import org.apache.hudi.common.table.HoodieTableMetaClient +import org.apache.hudi.common.util.{Option => HOption} + +import org.apache.spark.internal.Logging +import org.apache.spark.sql.SparkSession + +import scala.collection.{mutable, JavaConverters} + +/** + * 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 maxDataSkippingPartitions} + * (see [[DataSourceReadOptions.RECORD_INDEX_MAX_DATA_SKIPPING_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, + maxDataSkippingPartitions: Int) + 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 > maxDataSkippingPartitions) { + logInfo(s"The number of candidate partitions ${partitions.size} exceeds the partitioned record level index " + + s"lookup threshold $maxDataSkippingPartitions. Skipping record level index pruning.") + Option.empty + } else { + val keys = HoodieListData.eager(JavaConverters.seqAsJavaListConverter(recordKeys).asJava) + val fileIdToPartitionMap: mutable.Map[String, String] = mutable.Map.empty + for (partition <- partitions) { + val recordIndexData = metadataTable.readRecordIndexLocationsWithKeys(keys, HOption.of(partition)) Review Comment: 🤖 I think the current `lookupRecordKeys` actually handles the multi-file-group case: it builds `(partition, recordKey)` pairs, keys them by the global shard id (`partitionOffsetIndexes(partition) + mapRecordKeyToFileGroupIndex(recordKey, fileGroupCount)`), `partitionBy`s on that, and each Spark partition then calls `PartitionedRecordIndexFileGroupLookupFunction` with a single partitionName. Since the offsets are disjoint per partition, every shard maps to exactly one `(partition, fileGroupIndex)`, so keys that hash to different file groups within a partition land in different lookups — the `findFirst` shard assumption in the MDT read holds. So this mirrors the write path rather than forwarding all keys to one shard. A test with `RECORD_LEVEL_INDEX_MIN_FILE_GROUP_COUNT_PROP = 2+` and a multi-key `IN` would still be valuable to lock the behavior in. ########## 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 per-query rather than per-task — it caps the number of candidate data-table partitions for a single query (`partitions.size > MAX_PARTITIONS` in `PartitionedRecordLevelIndexSupport`), above which RLI pruning is skipped because each candidate partition needs its own metadata-table lookup. One thing worth noting: it's currently hard-coded to 10 in `RecordLevelIndexSupport.MAX_PARTITIONS` and not read from any config in the current revision, even though the PR description mentions a `hoodie.datasource.read.record.index.max.partitions` option. ########## hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/PartitionedRecordLevelIndexSupport.scala: ########## @@ -0,0 +1,119 @@ +/* + * 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.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) } Review Comment: 🤖 Agree this looks like real over-pruning risk, and one detail sharpens it: the `Option.empty` fallback only fires when `fileGroupCountPerDataPartition.isEmpty` — i.e. when *none* of the candidate partitions are indexed. In the mixed case (some partitions indexed, some not yet), line 82's filter silently drops the un-indexed ones, the lookup never queries them, and `filterCandidateFiles` then prunes all their files. Matching Flink's behavior — bailing to `Option.empty` if *any* candidate partition is absent from the bucketized map — would avoid dropping records during a backfill/skew window. ########## hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/RecordLevelIndexSupport.scala: ########## @@ -132,6 +148,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: 🤖 Confirmed the wiring is missing in the current revision — `MAX_PARTITIONS` is hard-coded to 10 in `RecordLevelIndexSupport`, and a grep for `record.index.max.partitions` / `RECORD_INDEX_MAX_DATA_SKIPPING_PARTITIONS` turns up no `ConfigProperty` definition in `DataSourceOptions` at all, so nothing reads it. Given Flink hard-codes its analogue to 3 and Spark to 10, exposing this as a tunable seems worthwhile for wide-partition tables. -- 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]
