vinishjail97 commented on code in PR #992: URL: https://github.com/apache/incubator-xtable/pull/992#discussion_r4394449688
########## 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: [P1] Bind the lookup to the index version that passed the freshness check The freshness check and lookup open separate Hudi views. A refresh can complete between them. I reproduced this with a file rewrite that preserved ID 7. After REFRESH, lookup returned the replacement file while candidates still referenced the previous snapshot. Their intersection was empty, although both snapshots contained the row. Would you pin lookup to the checked Hudi instant, or validate a stable generation through collection? A deterministic refresh test would protect this path. I consider this blocking because it silently drops rows. This reproduces the concurrency concern raised in https://github.com/apache/incubator-xtable/pull/933#issuecomment-5743299051. ########## 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: It works with spark 3.3 and 3.5 as well so let's say 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 +`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: `Then add the extensions after Iceberg's` What does this mean? Can you fix the text? ########## 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: + +```properties +spark.sql.extensions=org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions,org.apache.xtable.spark.extensions.XTableSparkSessionExtensions +spark.serializer=org.apache.spark.serializer.KryoSerializer +``` + +The extensions work with any Iceberg catalog that Spark is configured with. + +## Managing indexes + +```sql +-- builds the index from the current snapshot of the table +CREATE INDEX [IF NOT EXISTS] email_idx ON prod.db.customer USING xtable (email) + [OPTIONS ('xtable.hudi.target.metadata.record.index.min.filegroup.count' = '64')]; + +-- syncs the index with the current snapshot of the table +REFRESH INDEX email_idx ON prod.db.customer; + +DROP INDEX [IF EXISTS] email_idx ON prod.db.customer; +``` + +- An index covers one top level column of type `string`, `int` or `bigint`. A table can have one + index, and its column does not change. To index another column, drop the index and create a new + one. +- The index definition is stored in the table properties as `xtable.index.<name>.column`, and + options as `xtable.index.<name>.option.<key>`. `SHOW TBLPROPERTIES` lists them. +- Options are passed to the Hudi target that stores the index, for example the + `xtable.hudi.target.metadata.record.index.*` file group counts. +- The index files are stored under `.hoodie/` in the data folder of the table. Dropping the index + deletes that folder and keeps the data files. + +## Keeping the index current + +A sync job that already runs XTable keeps the index current. Add a Hudi target with the index +settings to the `--datasetconfig` of `XTableSparkSync` in the runtime jar, or of `RunSync`: + +```yaml +sourceFormat: ICEBERG +targetFormats: + - HUDI +datasets: + - tableBasePath: s3://bucket/warehouse/db.db/customer + tableDataPath: s3://bucket/warehouse/db.db/customer/data + tableName: customer + namespace: db + targetProperties: + HUDI: + xtable.hudi.target.table_version: '9' + xtable.hudi.target.execution.engine: spark + xtable.hudi.target.secondary.index.column: c_email_address +``` + +The column must match the column of `CREATE INDEX`, otherwise the sync fails. Without a sync job, +run `REFRESH INDEX` after writes. + +## Querying with an index + +Queries do not change. When a filter compares an indexed column with literals, for example +`email = 'x'` or `email IN ('x', 'y')`, possibly combined with other filters by `AND`, the extensions +look the values up in the index and read only the files that hold them. The query returns the same +rows as without the index. + +The index is not used when: + +- the table has new snapshots that the index was not refreshed with. The query then reads the table Review Comment: Is there a warning log that the index is stale and REFRESH INDEX needs to be used? ########## 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: How will this query understand the index is present and it needs to be used? Can we create a spark session property for it? Will this module work only for Spark SQL queries or data frame reads work too? Shouldn't it work for both of them? ########## 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: [P2] Keep staged tasks available while a scan can still use them The registry evicts the oldest set even when a live scan still needs it. Iceberg fetches these tasks lazily during partition planning. I reproduced `ValidationException` after constructing one scan and staging 1000 additional sets before planning its partitions. The failure occurs outside the pruning rule's fallback. Would you retain tasks for live scans and decline new staging when capacity is exhausted? I consider this blocking because unrelated planning can fail an existing query. ########## 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: [P1] Preserve incremental scan boundaries Iceberg incremental reads carry `start-snapshot-id` and `end-snapshot-id` in relation options. This guard checks neither option. I reproduced a read after S1 where S2 appended new rows. The ordinary scan returned nothing for `id = 7` from S1. The indexed scan returned row 7 because staging selected files from the full current snapshot. Would you skip incremental reads here and add a row comparison test? I consider this blocking because the rule changes the requested scan range. ########## 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: Let's show 3.5 in examples as that's the more popular spark version? ########## 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: + +```properties +spark.sql.extensions=org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions,org.apache.xtable.spark.extensions.XTableSparkSessionExtensions +spark.serializer=org.apache.spark.serializer.KryoSerializer +``` + +The extensions work with any Iceberg catalog that Spark is configured with. + +## Managing indexes + +```sql +-- builds the index from the current snapshot of the table +CREATE INDEX [IF NOT EXISTS] email_idx ON prod.db.customer USING xtable (email) + [OPTIONS ('xtable.hudi.target.metadata.record.index.min.filegroup.count' = '64')]; + +-- syncs the index with the current snapshot of the table +REFRESH INDEX email_idx ON prod.db.customer; + +DROP INDEX [IF EXISTS] email_idx ON prod.db.customer; +``` + +- An index covers one top level column of type `string`, `int` or `bigint`. A table can have one + index, and its column does not change. To index another column, drop the index and create a new + one. +- The index definition is stored in the table properties as `xtable.index.<name>.column`, and + options as `xtable.index.<name>.option.<key>`. `SHOW TBLPROPERTIES` lists them. +- Options are passed to the Hudi target that stores the index, for example the + `xtable.hudi.target.metadata.record.index.*` file group counts. +- The index files are stored under `.hoodie/` in the data folder of the table. Dropping the index + deletes that folder and keeps the data files. + +## Keeping the index current + +A sync job that already runs XTable keeps the index current. Add a Hudi target with the index +settings to the `--datasetconfig` of `XTableSparkSync` in the runtime jar, or of `RunSync`: + +```yaml +sourceFormat: ICEBERG +targetFormats: + - HUDI +datasets: + - tableBasePath: s3://bucket/warehouse/db.db/customer + tableDataPath: s3://bucket/warehouse/db.db/customer/data + tableName: customer + namespace: db + targetProperties: + HUDI: + xtable.hudi.target.table_version: '9' + xtable.hudi.target.execution.engine: spark + xtable.hudi.target.secondary.index.column: c_email_address +``` + +The column must match the column of `CREATE INDEX`, otherwise the sync fails. Without a sync job, +run `REFRESH INDEX` after writes. + +## Querying with an index Review Comment: Given that it's being packaged as part of xtable-spark-runtime jar...should there be an iceberg reader property for using the index? Otherwise users need to remove the extensions jar to disable reading from the index in case it's being repaired because of some reason and it's not caught as part of the automatic detection we have below. ########## website/sidebars.js: ########## @@ -20,7 +20,8 @@ module.exports = { items: [ 'how-to', 'how-to-catalog-sync', - 'how-to-spark-runtime' + 'how-to-spark-runtime', + 'how-to-index-sql' Review Comment: Drop the sql as you can read/write indexes using spark jobs as well? ########## 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: + +```properties +spark.sql.extensions=org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions,org.apache.xtable.spark.extensions.XTableSparkSessionExtensions +spark.serializer=org.apache.spark.serializer.KryoSerializer +``` + +The extensions work with any Iceberg catalog that Spark is configured with. + +## Managing indexes + +```sql +-- builds the index from the current snapshot of the table +CREATE INDEX [IF NOT EXISTS] email_idx ON prod.db.customer USING xtable (email) + [OPTIONS ('xtable.hudi.target.metadata.record.index.min.filegroup.count' = '64')]; + +-- syncs the index with the current snapshot of the table +REFRESH INDEX email_idx ON prod.db.customer; + +DROP INDEX [IF EXISTS] email_idx ON prod.db.customer; +``` + +- An index covers one top level column of type `string`, `int` or `bigint`. A table can have one + index, and its column does not change. To index another column, drop the index and create a new + one. +- The index definition is stored in the table properties as `xtable.index.<name>.column`, and + options as `xtable.index.<name>.option.<key>`. `SHOW TBLPROPERTIES` lists them. +- Options are passed to the Hudi target that stores the index, for example the + `xtable.hudi.target.metadata.record.index.*` file group counts. +- The index files are stored under `.hoodie/` in the data folder of the table. Dropping the index + deletes that folder and keeps the data files. + +## Keeping the index current + +A sync job that already runs XTable keeps the index current. Add a Hudi target with the index +settings to the `--datasetconfig` of `XTableSparkSync` in the runtime jar, or of `RunSync`: + +```yaml +sourceFormat: ICEBERG +targetFormats: + - HUDI +datasets: + - tableBasePath: s3://bucket/warehouse/db.db/customer + tableDataPath: s3://bucket/warehouse/db.db/customer/data + tableName: customer + namespace: db + targetProperties: + HUDI: + xtable.hudi.target.table_version: '9' + xtable.hudi.target.execution.engine: spark + xtable.hudi.target.secondary.index.column: c_email_address +``` + +The column must match the column of `CREATE INDEX`, otherwise the sync fails. Without a sync job, +run `REFRESH INDEX` after writes. + +## Querying with an index + +Queries do not change. When a filter compares an indexed column with literals, for example +`email = 'x'` or `email IN ('x', 'y')`, possibly combined with other filters by `AND`, the extensions +look the values up in the index and read only the files that hold them. The query returns the same +rows as without the index. + +The index is not used when: + +- the table has new snapshots that the index was not refreshed with. The query then reads the table + as usual, so results stay correct; run `REFRESH INDEX` to use the index again. +- the query reads an older snapshot (`VERSION AS OF`, `TIMESTAMP AS OF`) or a branch. +- the scan is small. A lookup runs a Spark job, which costs more than reading a few files. The + index is used only when Iceberg's own partition and min/max pruning leave at least + `spark.xtable.index.pruning.minCandidateFiles` files or `spark.xtable.index.pruning.minCandidateBytes` + bytes to read. +- the filter has more than `spark.xtable.index.pruning.maxKeys` values. + +The driver log shows for each query whether the index was used and how many files it selected. + +| Setting | Default | Meaning | +|---|---|---| +| `spark.xtable.index.pruning.enabled` | `true` | Use indexes for queries | Review Comment: `spark.xtable.index.pruning.enabled` Is by default true/false? ########## 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: + +```properties +spark.sql.extensions=org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions,org.apache.xtable.spark.extensions.XTableSparkSessionExtensions +spark.serializer=org.apache.spark.serializer.KryoSerializer +``` + +The extensions work with any Iceberg catalog that Spark is configured with. + +## Managing indexes + +```sql +-- builds the index from the current snapshot of the table +CREATE INDEX [IF NOT EXISTS] email_idx ON prod.db.customer USING xtable (email) + [OPTIONS ('xtable.hudi.target.metadata.record.index.min.filegroup.count' = '64')]; + +-- syncs the index with the current snapshot of the table +REFRESH INDEX email_idx ON prod.db.customer; + +DROP INDEX [IF EXISTS] email_idx ON prod.db.customer; +``` + +- An index covers one top level column of type `string`, `int` or `bigint`. A table can have one + index, and its column does not change. To index another column, drop the index and create a new + one. +- The index definition is stored in the table properties as `xtable.index.<name>.column`, and + options as `xtable.index.<name>.option.<key>`. `SHOW TBLPROPERTIES` lists them. +- Options are passed to the Hudi target that stores the index, for example the + `xtable.hudi.target.metadata.record.index.*` file group counts. +- The index files are stored under `.hoodie/` in the data folder of the table. Dropping the index + deletes that folder and keeps the data files. + +## Keeping the index current Review Comment: This doc is conflicting...this user doc should describe how to use indexes using both SQL and Jobs in separate sections it's all over the place right now and there's no structure. -- 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]
