vinishjail97 commented on code in PR #992: URL: https://github.com/apache/incubator-xtable/pull/992#discussion_r4395063819
########## xtable-spark-extensions/src/main/scala/org/apache/xtable/spark/extensions/optimizer/IndexPruningRule.scala: ########## @@ -0,0 +1,258 @@ +/* + * 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.xtable.spark.extensions.optimizer + +import java.util.concurrent.atomic.AtomicBoolean + +import org.apache.hadoop.fs.Path +import org.apache.spark.sql.{Encoders, SparkSession} +import org.apache.spark.sql.catalyst.expressions.{ + And, + Attribute, + AttributeReference, + EqualTo, + Expression, + In, + InSet, + Literal +} +import org.apache.spark.sql.catalyst.plans.logical.{Filter, LocalRelation, LogicalPlan} +import org.apache.spark.sql.catalyst.rules.Rule +import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Relation +import org.apache.spark.sql.internal.SQLConf +import org.apache.spark.sql.util.CaseInsensitiveStringMap + +import org.apache.iceberg.{FileScanTask, Table} +import org.apache.iceberg.expressions.{Expression => IcebergExpression, Expressions} +import org.apache.iceberg.spark.source.SparkTable +import org.apache.iceberg.types.Type.TypeID + +import scala.collection.JavaConverters._ +import scala.util.control.NonFatal + +import org.apache.xtable.hudi.index.HudiBackedSecondaryIndexReader +import org.apache.xtable.spark.extensions.{IndexDefinition, XTableIndexConf} + +/** + * Uses an XTable index to read only the files that hold the keys of a filter such as + * {@code col = 'x'} or {@code col IN ('x', 'y')} on an indexed column of an Iceberg table. + * + * The rule looks the keys up in the index, keeps the files that Iceberg's own partition and min/max + * pruning also keeps, and stages them as the task set of the scan. The filter stays in the plan, so + * the query returns the same rows as without the rule. The rule leaves the plan unchanged when the + * index is behind the table, when the query reads an older snapshot or a branch, or when the scan + * is too small for a lookup to pay off. + */ +case class IndexPruningRule(spark: SparkSession) extends Rule[LogicalPlan] { + private val listenerRegistered = new AtomicBoolean(false) + + override def apply(plan: LogicalPlan): LogicalPlan = { + if (!XTableIndexConf.pruningEnabled(conf)) { + return plan + } + plan.transformDown { + case filter @ Filter( + condition, + relation @ DataSourceV2Relation(table: SparkTable, _, _, _, options)) + if !options.containsKey(StagedScans.ScanTaskSetIdOption) && + table.snapshotId() == null && table.branch() == null => + try { + prune(filter, condition, relation, table).getOrElse(filter) + } catch { + case NonFatal(e) => + logWarning(s"Not using the XTable index for ${table.name()}", e) + filter + } + } + } + + private def prune( + filter: Filter, + condition: Expression, + relation: DataSourceV2Relation, + sparkTable: SparkTable): Option[LogicalPlan] = { + val definitions = IndexDefinition.fromProperties(sparkTable.properties()) + val table = sparkTable.table() + for { + (definition, keys) <- IndexPruningRule + .findIndexedKeys(condition, relation.output, definitions, conf.resolver) + if withinKeyLimit(table, definition, keys) + snapshot <- Option(table.currentSnapshot()) + index = IndexDefinition.newIndex( + spark, + table, + relation.catalog.map(_.name()), + relation.identifier, + definition) + reader = index.reader + if isCurrent(table, definition, reader.getLastSyncedSourceIdentifier, snapshot.snapshotId()) Review Comment: The second check fixes the ordinary forward-sync race, but it does not identify a stable index version. I reproduced another missing row with this sequence. 1. The first check sees S1. 2. A rewrite and sync publish S2 before lookup. 3. Lookup returns S2 files. 4. The source rolls back to S1, and the documented delete-and-rebuild procedure restores an S1 index. 5. The second check accepts S1 again, but S1 candidates and S2 lookup files have an empty intersection. The query returned no rows instead of ID 7. I used actual Iceberg writes and Hudi syncs. A test-only callback after collection made the interleaving deterministic. I kept this thread open because the original stable-version requirement remains unresolved. Would you validate an index generation and completed commit identity, including rebuilds, or pin the lookup to that generation? The source snapshot ID alone does not identify an index generation. -- 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]
