vinishjail97 commented on code in PR #992: URL: https://github.com/apache/incubator-xtable/pull/992#discussion_r4394927697
########## 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: Fixed in b57e1b83. The rule reads the synced snapshot again after the lookup collects and falls back if a sync committed in between; `syncDuringLookupKeepsTheRows` runs a sync between the check and the lookup and fails without the fix. ########## 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 => Review Comment: Fixed in b57e1b83. Relations with Iceberg snapshot read options (`start-snapshot-id`, `end-snapshot-id`, `snapshot-id`, `as-of-timestamp`, `branch`, `tag` and the timestamp variants) are skipped; covered by `incrementalReadsDoNotUseIndex`. ########## xtable-spark-extensions/src/main/scala/org/apache/xtable/spark/extensions/optimizer/StagedScans.scala: ########## @@ -0,0 +1,100 @@ +/* + * 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.UUID + +import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan +import org.apache.spark.sql.execution.QueryExecution +import org.apache.spark.sql.execution.datasources.v2.{ + DataSourceV2Relation, + DataSourceV2ScanRelation +} +import org.apache.spark.sql.util.QueryExecutionListener + +import org.apache.iceberg.{FileScanTask, Table} +import org.apache.iceberg.spark.ScanTaskSetManager + +import scala.collection.JavaConverters._ +import scala.util.Try + +/** + * Stages the file scan tasks that an index lookup selects. Iceberg reads a staged task set when a + * scan has the {@code scan-task-set-id} option. Staged tasks live in a process wide registry until + * they are removed, so this object removes them when the query that uses them ends, and removes the + * oldest sets when too many are live (for example from plans that are only explained). + */ +object StagedScans { + val ScanTaskSetIdOption = "scan-task-set-id" + + private val MaxLiveSets = 1000 + + // insertion ordered, so the oldest set is evicted first + private val liveSets = new java.util.LinkedHashMap[String, Table]() + + def stage(table: Table, tasks: Seq[FileScanTask]): String = { + val setId = UUID.randomUUID().toString + ScanTaskSetManager.get().stageTasks(table, setId, tasks.asJava) + val evicted = liveSets.synchronized { + liveSets.put(setId, table) + val oldest = liveSets.keySet().asScala.take(math.max(0, liveSets.size() - MaxLiveSets)).toList Review Comment: Fixed in b57e1b83. No eviction anymore: once 1000 sets are live the rule stops staging and the scan reads the table as usual; `stagedScansStayLiveUntilTheirQueryEnds` covers it. ########## website/docs/how-to-index-sql.md: ########## @@ -0,0 +1,125 @@ +--- +sidebar_position: 4 +title: "Index Iceberg tables with Spark SQL" +--- + +# Index Iceberg tables with Spark SQL + +Apache XTable™ (Incubating) can keep a secondary index on a column of an Apache Iceberg table. The index maps each value +of the column to the data files that hold it, and is stored as an Apache Hudi metadata table next to +the table's data. The XTable Spark SQL extensions let you manage the index with SQL, and they use it +automatically for point lookups: + +```sql +CREATE INDEX email_idx ON prod.db.customer USING xtable (email); + +-- reads only the files that hold these emails +SELECT * FROM prod.db.customer WHERE email IN ('[email protected]', '[email protected]'); +``` + +## Setting up Spark + +The extensions need Spark 3.4 with Scala 2.12, the Iceberg Spark runtime and the Hudi Spark bundle. Review Comment: Done in b57e1b83, the doc now says Spark 3.x. ########## website/docs/how-to-index-sql.md: ########## @@ -0,0 +1,125 @@ +--- +sidebar_position: 4 +title: "Index Iceberg tables with Spark SQL" +--- + +# Index Iceberg tables with Spark SQL + +Apache XTable™ (Incubating) can keep a secondary index on a column of an Apache Iceberg table. The index maps each value +of the column to the data files that hold it, and is stored as an Apache Hudi metadata table next to +the table's data. The XTable Spark SQL extensions let you manage the index with SQL, and they use it +automatically for point lookups: + +```sql +CREATE INDEX email_idx ON prod.db.customer USING xtable (email); + +-- reads only the files that hold these emails +SELECT * FROM prod.db.customer WHERE email IN ('[email protected]', '[email protected]'); +``` + +## Setting up Spark + +The extensions need Spark 3.4 with Scala 2.12, the Iceberg Spark runtime and the Hudi Spark bundle. +They ship in the [XTable Spark runtime jar](how-to-spark-runtime), so no other XTable jar is needed. +Until a release includes them, build the jar from source: + +```shell +./mvnw -pl xtable-spark-runtime -am package -DskipTests +``` + +Put these jars on the classpath of the driver and the executors. Use Review Comment: Done in b57e1b83, the examples use the Spark 3.5 Hudi and Iceberg bundles. ########## website/docs/how-to-index-sql.md: ########## @@ -0,0 +1,125 @@ +--- +sidebar_position: 4 +title: "Index Iceberg tables with Spark SQL" +--- + +# Index Iceberg tables with Spark SQL + +Apache XTable™ (Incubating) can keep a secondary index on a column of an Apache Iceberg table. The index maps each value +of the column to the data files that hold it, and is stored as an Apache Hudi metadata table next to +the table's data. The XTable Spark SQL extensions let you manage the index with SQL, and they use it +automatically for point lookups: + +```sql +CREATE INDEX email_idx ON prod.db.customer USING xtable (email); + +-- reads only the files that hold these emails +SELECT * FROM prod.db.customer WHERE email IN ('[email protected]', '[email protected]'); +``` + +## Setting up Spark + +The extensions need Spark 3.4 with Scala 2.12, the Iceberg Spark runtime and the Hudi Spark bundle. +They ship in the [XTable Spark runtime jar](how-to-spark-runtime), so no other XTable jar is needed. +Until a release includes them, build the jar from source: + +```shell +./mvnw -pl xtable-spark-runtime -am package -DskipTests +``` + +Put these jars on the classpath of the driver and the executors. Use +`spark.driver.extraClassPath` and `spark.executor.extraClassPath` rather than `--jars`, so that +XTable and Hudi are loaded by the same class loader: + +- `xtable-spark-runtime/target/xtable-spark-runtime_2.12-<version>.jar` +- `org.apache.hudi:hudi-spark3.4-bundle_2.12:1.2.1` +- `org.apache.iceberg:iceberg-spark-runtime-3.4_2.12:1.9.2` + +Then add the extensions after Iceberg's, and use Kryo, which Hudi needs: Review Comment: Reworded in b57e1b83: `spark.sql.extensions` lists the Iceberg extensions first and the XTable extensions second. ########## website/docs/how-to-index-sql.md: ########## @@ -0,0 +1,125 @@ +--- +sidebar_position: 4 +title: "Index Iceberg tables with Spark SQL" +--- + +# Index Iceberg tables with Spark SQL + +Apache XTable™ (Incubating) can keep a secondary index on a column of an Apache Iceberg table. The index maps each value +of the column to the data files that hold it, and is stored as an Apache Hudi metadata table next to +the table's data. The XTable Spark SQL extensions let you manage the index with SQL, and they use it +automatically for point lookups: + +```sql +CREATE INDEX email_idx ON prod.db.customer USING xtable (email); + +-- reads only the files that hold these emails Review Comment: The rule finds the index from the Hudi metadata in the data folder of the table, so queries need no hint, and it runs as an optimizer rule, so DataFrame reads use it too (`dataFrameReadsUseIndex` in b57e1b83). `spark.xtable.index.pruning.enabled` already turns it off per session; the doc now shows both. -- 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]
