This is an automated email from the ASF dual-hosted git repository.
vinishjail97 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/incubator-xtable.git
The following commit(s) were added to refs/heads/main by this push:
new b74b2332 [887] Add Spark SQL extensions for the XTable secondary index
(#992)
b74b2332 is described below
commit b74b2332f4648de937b24e77e604ab2793e25966
Author: Vinish Reddy <[email protected]>
AuthorDate: Sat Oct 31 01:43:21 2026 +0530
[887] Add Spark SQL extensions for the XTable secondary index (#992)
* Add Spark SQL extensions for the XTable secondary index
Adds the xtable-spark-extensions module with XTableSparkSessionExtensions:
- CREATE INDEX ... USING xtable, DROP INDEX and REFRESH INDEX for Iceberg
tables. CREATE and REFRESH sync the table to a Hudi target at its data
path with xtable.hudi.target.secondary.index.column set, the same way
RunSync builds the index. DROP deletes the Hudi metadata and keeps the
data files. The index definition is kept in the table properties.
- A rule that uses the index for col = literal and col IN (...) filters.
It looks the keys up with HudiBackedSecondaryIndexReader and stages only
the files that hold them as the scan's task set (scan-task-set-id),
since Iceberg does not prune files on _file. It skips stale indexes,
time travel and branch reads, and scans below a size threshold.
- A how-to page for the website.
The Catalyst code is in Scala, so the root pom manages scala-maven-plugin
and Spotless checks Scala sources with scalafmt.
* Bundle the Spark SQL extensions in xtable-spark-runtime
Users add the runtime jar and name XTableSparkSessionExtensions in
spark.sql.extensions, so no other XTable jar is needed. The extensions keep
their package in the bundle, and ITXTableSparkRuntimeBundle checks that.
* Pass target properties from the XTableSparkSync dataset config
A table entry of --datasetconfig takes targetProperties keyed by the target
format, the same as RunSync, so a sync job of the runtime jar can keep the
secondary index current with xtable.hudi.target.secondary.index.column.
Before, XTableSparkSync ignored the key in a RunSync config.
* Address review feedback: read-only index rule, consistent lookups,
incremental reads, staged scan capacity
* Address review feedback: index version check, indexed field identity,
Hudi bundle per Spark version in CI
- Compare the index version (latest completion time on the Hudi timeline
and the synced snapshot)
before and after the lookup, so a rollback and rebuild to the same
snapshot falls back.
- Skip the index when a schema of the table reuses the name of the indexed
column for another
field, renames the field or changes its type.
- Pass the Hudi Spark bundle that matches the Spark of spark-submit in the
bundle IT, and raise the
Maven CI timeout to 60 minutes.
* Fail the query when the indexed column was renamed after the index was
synced
Replace the check of every retained schema with one comparison of the
indexed field id in the
schema of the synced snapshot and in the current schema. A mismatch throws
IndexedColumnRenamedException instead of falling back to a full scan.
* Address review feedback: skip metadata tables and snapshots without a
schema id, look declined keys up once per query, release staged scans only
after optimization
---
.github/workflows/spark-runtime-validation.yml | 3 +-
pom.xml | 31 +
style/scalafmt.conf | 26 +
website/docs/how-to-index.md | 146 ++++
website/docs/how-to-spark-runtime.md | 3 +
website/sidebars.js | 3 +-
.../hudi/index/HudiBackedSecondaryIndexReader.java | 10 +
.../hudi/index/HudiBackedSecondaryIndexUtils.java | 47 +-
xtable-spark-extensions/pom.xml | 135 ++++
.../xtable/spark/extensions/XTableIndex.scala | 148 ++++
.../xtable/spark/extensions/XTableIndexConf.scala | 55 ++
.../extensions/XTableSparkSessionExtensions.scala | 41 ++
.../extensions/optimizer/IndexPruningRule.scala | 391 +++++++++++
.../spark/extensions/optimizer/StagedScans.scala | 107 +++
.../spark/extensions/ITXTableIndexExtensions.java | 742 +++++++++++++++++++++
.../xtable/spark/extensions/TestXTableIndex.java | 117 ++++
xtable-spark-runtime/pom.xml | 36 +-
.../org/apache/xtable/spark/TableSyncSpec.java | 7 +
.../org/apache/xtable/spark/XTableSparkSync.java | 16 +
.../org/apache/xtable/spark/XTableSyncService.java | 60 +-
.../xtable/spark/ITXTableSparkRuntimeBundle.java | 118 +++-
.../apache/xtable/spark/XTableSparkSyncTest.java | 58 ++
22 files changed, 2262 insertions(+), 38 deletions(-)
diff --git a/.github/workflows/spark-runtime-validation.yml
b/.github/workflows/spark-runtime-validation.yml
index 9ae242ae..20c743b5 100644
--- a/.github/workflows/spark-runtime-validation.yml
+++ b/.github/workflows/spark-runtime-validation.yml
@@ -83,4 +83,5 @@ jobs:
- name: Run xtable-spark-runtime bundle IT
env:
SPARK_LOCAL_IP: "127.0.0.1"
- run: ./mvnw verify -ntp -B -pl xtable-spark-runtime
+ # spark-submit gets the Hudi bundle built for its own Spark version,
as users pass it
+ run: ./mvnw verify -ntp -B -pl xtable-spark-runtime
-Dengine.spark.version.prefix="${SPARK_VERSION%.*}"
diff --git a/pom.xml b/pom.xml
index 5c8c234e..72965182 100644
--- a/pom.xml
+++ b/pom.xml
@@ -50,6 +50,7 @@
<module>xtable-api</module>
<module>xtable-hudi-support</module>
<module>xtable-core</module>
+ <module>xtable-spark-extensions</module>
<module>xtable-utilities</module>
<module>xtable-aws</module>
<module>xtable-hive-metastore</module>
@@ -94,6 +95,8 @@
<paimon.version>1.3.1</paimon.version>
<jackson.version>2.18.2</jackson.version>
<spotless.version>2.46.1</spotless.version>
+ <scala-maven-plugin.version>4.9.2</scala-maven-plugin.version>
+ <scalafmt.version>3.8.3</scalafmt.version>
<apache.rat.version>0.16.1</apache.rat.version>
<google.java.format.version>1.10.0</google.java.format.version>
<delta.standalone.version>3.3.0</delta.standalone.version>
@@ -679,6 +682,21 @@
<build>
<pluginManagement>
<plugins>
+ <plugin>
+ <groupId>net.alchim31.maven</groupId>
+ <artifactId>scala-maven-plugin</artifactId>
+ <version>${scala-maven-plugin.version}</version>
+ <configuration>
+ <scalaVersion>${scala.version}</scalaVersion>
+ <args>
+ <arg>-unchecked</arg>
+ <arg>-deprecation</arg>
+ <arg>-feature</arg>
+ <arg>-release</arg>
+ <arg>${javaVersion}</arg>
+ </args>
+ </configuration>
+ </plugin>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-shade-plugin</artifactId>
@@ -934,6 +952,19 @@
<file>style/license-header</file>
</licenseHeader>
</java>
+ <scala>
+ <includes>
+ <include>src/main/scala/**/*.scala</include>
+ <include>src/test/scala/**/*.scala</include>
+ </includes>
+ <scalafmt>
+ <version>${scalafmt.version}</version>
+ <file>style/scalafmt.conf</file>
+ </scalafmt>
+ <licenseHeader>
+ <file>style/license-header</file>
+ </licenseHeader>
+ </scala>
<python>
<includes>
<include>**/*.py</include>
diff --git a/style/scalafmt.conf b/style/scalafmt.conf
new file mode 100644
index 00000000..e29a1930
--- /dev/null
+++ b/style/scalafmt.conf
@@ -0,0 +1,26 @@
+#
+# 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.
+#
+version = 3.8.3
+runner.dialect = scala212
+maxColumn = 100
+docstrings.style = Asterisk
+docstrings.wrap = no
+rewrite.rules = [Imports]
+rewrite.imports.sort = ascii
+align.preset = none
+danglingParentheses.preset = false
diff --git a/website/docs/how-to-index.md b/website/docs/how-to-index.md
new file mode 100644
index 00000000..1a8c517d
--- /dev/null
+++ b/website/docs/how-to-index.md
@@ -0,0 +1,146 @@
+---
+sidebar_position: 4
+title: "Index Iceberg tables"
+---
+
+# Index Iceberg tables
+
+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 under `.hoodie/` in the data folder of the table. Using an
index takes two parts:
+
+1. An XTable sync job [builds the index and keeps it
current](#building-the-index-with-a-sync-job).
+2. Spark sessions with the XTable Spark SQL extensions [use the
index](#querying-with-the-index) for
+ point lookups, in SQL queries and DataFrame reads alike. Queries do not
change.
+
+## Building the index with a sync job
+
+Any XTable sync of the table builds the index when it has a Hudi target with
the index settings:
+`RunSync` from the utilities jar, `XTableSparkSync` from the
+[Spark runtime jar](how-to-spark-runtime), or a Spark job that calls the
XTable sync API. Add the
+Hudi target to the `--datasetconfig` of the job:
+
+```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 first sync builds the index from the current snapshot of the table, and
each later sync adds
+the changes of the new snapshots.
+
+- An index covers one top level column of type `string`, `int` or `bigint`,
and its column does not
+ change. A sync with another column fails before it commits.
+- Other `xtable.hudi.target.*` settings tune the index, for example the
+ `xtable.hudi.target.metadata.record.index.*` file group counts.
+- To remove the index, remove the Hudi target from the sync job and delete the
`.hoodie/` folder in
+ the data folder of the table. The data files of the table are not changed. A
later sync with
+ another column builds a new index.
+
+## Querying with the index
+
+### Setting up Spark
+
+The extensions need Spark 3.x 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. The versions below are
for Spark 3.5; use the
+bundles that match your Spark version.
+
+- `xtable-spark-runtime/target/xtable-spark-runtime_2.12-<version>.jar`
+- `org.apache.hudi:hudi-spark3.5-bundle_2.12:1.2.1`
+- `org.apache.iceberg:iceberg-spark-runtime-3.5_2.12:1.9.2`
+
+`spark.sql.extensions` must list the Iceberg extensions first and the XTable
extensions second,
+and Hudi needs the Kryo serializer:
+
+```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.
+
+### Running queries
+
+The extensions find the index by themselves: when a query filters an Iceberg
table, they look for
+the Hudi metadata in the data folder of the table. 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.
+
+```sql
+SELECT * FROM prod.db.customer WHERE c_email_address IN ('[email protected]',
'[email protected]');
+```
+
+DataFrame reads use the index the same way:
+
+```python
+from pyspark.sql.functions import col
+
+spark.table("prod.db.customer").where(col("c_email_address") ==
"[email protected]")
+spark.read.format("iceberg").load("prod.db.customer").where(col("c_email_address")
== "[email protected]")
+```
+
+The driver log shows for each query whether the index was used and how many
files it selected.
+
+### When the index is not used
+
+The query reads the table as usual, so results stay correct, when:
+
+- the index was not synced with the current snapshot of the table. The driver
log has a warning
+ that names the snapshot of the index and of the table; run the sync job to
use the index again.
+- a sync, a rollback or a rebuild changes the index while the query looks keys
up in it.
+- the query reads an older snapshot (`VERSION AS OF`, `TIMESTAMP AS OF`,
`snapshot-id`,
+ `as-of-timestamp`), a branch, a tag, or the changes between snapshots
(`start-snapshot-id`,
+ `end-snapshot-id`).
+- 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.
+
+### When a query fails
+
+A query with a filter on the indexed column fails when the column was renamed,
or its name given to
+another field, after the index was synced. The index keys the column by name,
so it would select
+the files of the old field. Remove the index and rebuild it after the next
commit to the table, or
+turn the index off for the session.
+
+### Settings
+
+| Setting | Default | Meaning |
+|---|---|---|
+| `spark.xtable.index.pruning.enabled` | `true` | Use indexes for queries |
+| `spark.xtable.index.pruning.minCandidateFiles` | `32` | Minimum files a scan
reads before an index is used |
+| `spark.xtable.index.pruning.minCandidateBytes` | `1073741824` | Minimum
bytes a scan reads before an index is used |
+| `spark.xtable.index.pruning.maxKeys` | `10000` | Maximum values of a filter
that is looked up |
+
+To read tables without their index, for example while an index is repaired,
turn the index off for
+a session without removing the jar:
+
+```sql
+SET spark.xtable.index.pruning.enabled=false;
+```
+
+## Limitations
+
+- Only Spark sessions with the extensions use the index. Other engines read
the table as usual.
+- Joins, subqueries, `MERGE`, `UPDATE` and `DELETE` do not use the index.
diff --git a/website/docs/how-to-spark-runtime.md
b/website/docs/how-to-spark-runtime.md
index bafa2caf..82f2b663 100644
--- a/website/docs/how-to-spark-runtime.md
+++ b/website/docs/how-to-spark-runtime.md
@@ -173,6 +173,9 @@ $SPARK_HOME/bin/spark-submit \
`--datasetconfig` and `--basepath` are mutually exclusive — pass one or the
other.
+A table entry can also set `targetProperties`, keyed by target format, for
example the
+[secondary index settings](how-to-index#building-the-index-with-a-sync-job) of
a Hudi target.
+
| Option | Description |
| --- | --- |
| `--basepath` | Base path of the source table. |
diff --git a/website/sidebars.js b/website/sidebars.js
index 79cb498d..729e99ec 100644
--- a/website/sidebars.js
+++ b/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'
],
},
{
diff --git
a/xtable-core/src/main/java/org/apache/xtable/hudi/index/HudiBackedSecondaryIndexReader.java
b/xtable-core/src/main/java/org/apache/xtable/hudi/index/HudiBackedSecondaryIndexReader.java
index b74fc0d6..891b83ef 100644
---
a/xtable-core/src/main/java/org/apache/xtable/hudi/index/HudiBackedSecondaryIndexReader.java
+++
b/xtable-core/src/main/java/org/apache/xtable/hudi/index/HudiBackedSecondaryIndexReader.java
@@ -186,6 +186,16 @@ public class HudiBackedSecondaryIndexReader
dataPath, javaSparkContext.hadoopConfiguration());
}
+ /**
+ * Returns the version of the index. A caller that compares the source
commit of the index with
+ * its own view of the table can read the version before and after a lookup,
and trust the lookup
+ * only when both versions are equal.
+ */
+ public Optional<HudiBackedSecondaryIndexUtils.SyncedVersion>
getLastSyncedVersion() {
+ return HudiBackedSecondaryIndexUtils.getLastSyncedVersion(
+ dataPath, javaSparkContext.hadoopConfiguration());
+ }
+
@Override
public Dataset<Row> lookup(SourceTable table, Dataset<Row> keys, String
columnName) {
HudiBackedSecondaryIndexUtils.checkIndexedTable(tableLocation, table);
diff --git
a/xtable-core/src/main/java/org/apache/xtable/hudi/index/HudiBackedSecondaryIndexUtils.java
b/xtable-core/src/main/java/org/apache/xtable/hudi/index/HudiBackedSecondaryIndexUtils.java
index 088844b3..2ebc5dc2 100644
---
a/xtable-core/src/main/java/org/apache/xtable/hudi/index/HudiBackedSecondaryIndexUtils.java
+++
b/xtable-core/src/main/java/org/apache/xtable/hudi/index/HudiBackedSecondaryIndexUtils.java
@@ -30,10 +30,12 @@ import java.util.stream.Collectors;
import lombok.AccessLevel;
import lombok.NoArgsConstructor;
+import lombok.Value;
import org.apache.hadoop.conf.Configuration;
import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.table.timeline.HoodieActiveTimeline;
import org.apache.hudi.common.table.timeline.HoodieInstant;
import org.apache.hudi.common.table.timeline.TimelineUtils;
import org.apache.hudi.exception.TableNotFoundException;
@@ -142,11 +144,24 @@ public final class HudiBackedSecondaryIndexUtils {
}
/**
- * Returns the source commit identifier in the sync metadata of the last
completed commit, or
- * empty when there is no Hudi table or no completed commit yet. A sync
without a new source
- * commit makes no Hudi commit, so the identifier stays at the last synced
source commit.
+ * A version of the index: the latest completion time on the timeline of the
Hudi target, and the
+ * source commit that the last completed commit synced. Every action on the
target completes at a
+ * new time, including a sync, a rollback, a restore and a rebuild of a
deleted target, so two
+ * reads that return equal versions read the same index, even when a rebuilt
index syncs a source
+ * commit that an earlier version synced as well.
*/
- static Optional<String> getLastSyncedSourceIdentifier(
+ @Value
+ public static class SyncedVersion {
+ String latestCompletionTime;
+ String sourceIdentifier;
+ }
+
+ /**
+ * Returns the version of the index, or empty when there is no Hudi table,
no completed commit or
+ * no sync metadata yet. A sync without a new source commit makes no Hudi
commit, so the source
+ * identifier stays at the last synced source commit.
+ */
+ static Optional<SyncedVersion> getLastSyncedVersion(
String dataPath, Configuration configuration) {
HoodieTableMetaClient metaClient;
try {
@@ -154,20 +169,36 @@ public final class HudiBackedSecondaryIndexUtils {
} catch (TableNotFoundException e) {
return Optional.empty();
}
+ // one load of the timeline for both fields, so they describe the same
version
+ HoodieActiveTimeline timeline = metaClient.getActiveTimeline();
Optional<HoodieInstant> lastCommit =
-
metaClient.getCommitsTimeline().filterCompletedInstants().lastInstant().toJavaOptional();
- if (!lastCommit.isPresent()) {
+
timeline.getCommitsTimeline().filterCompletedInstants().lastInstant().toJavaOptional();
+ Optional<String> latestCompletionTime =
+
timeline.filterCompletedInstants().getLatestCompletionTime().toJavaOptional();
+ if (!lastCommit.isPresent() || !latestCompletionTime.isPresent()) {
return Optional.empty();
}
try {
String syncMetadata =
- TimelineUtils.getCommitMetadata(lastCommit.get(),
metaClient.getActiveTimeline())
+ TimelineUtils.getCommitMetadata(lastCommit.get(), timeline)
.getExtraMetadata()
.get(TableSyncMetadata.XTABLE_METADATA);
- return
TableSyncMetadata.fromJson(syncMetadata).map(TableSyncMetadata::getSourceIdentifier);
+ return TableSyncMetadata.fromJson(syncMetadata)
+ .map(
+ metadata ->
+ new SyncedVersion(latestCompletionTime.get(),
metadata.getSourceIdentifier()));
} catch (IOException e) {
throw new UncheckedIOException(
"Failed to read the XTable sync metadata of " + lastCommit.get() + "
in " + dataPath, e);
}
}
+
+ /**
+ * Returns the source commit identifier in the sync metadata of the last
completed commit, or
+ * empty when there is no Hudi table or no completed commit yet.
+ */
+ static Optional<String> getLastSyncedSourceIdentifier(
+ String dataPath, Configuration configuration) {
+ return getLastSyncedVersion(dataPath,
configuration).map(SyncedVersion::getSourceIdentifier);
+ }
}
diff --git a/xtable-spark-extensions/pom.xml b/xtable-spark-extensions/pom.xml
new file mode 100644
index 00000000..2beda90b
--- /dev/null
+++ b/xtable-spark-extensions/pom.xml
@@ -0,0 +1,135 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ ~ 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.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0";
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance";
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0
http://maven.apache.org/xsd/maven-4.0.0.xsd";>
+ <modelVersion>4.0.0</modelVersion>
+ <parent>
+ <groupId>org.apache.xtable</groupId>
+ <artifactId>xtable</artifactId>
+ <version>0.5.0-SNAPSHOT</version>
+ </parent>
+
+ <artifactId>xtable-spark-extensions_${scala.binary.version}</artifactId>
+ <name>XTable Project Spark Extensions</name>
+ <description>Spark SQL extensions for XTable, such as SQL commands for the
Hudi backed Iceberg secondary index.</description>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.xtable</groupId>
+ <artifactId>xtable-core_${scala.binary.version}</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+
+ <!-- Scala dependencies -->
+ <dependency>
+ <groupId>org.scala-lang</groupId>
+ <artifactId>scala-library</artifactId>
+ <scope>provided</scope>
+ </dependency>
+
+ <!-- Spark, Iceberg and Hudi come from the Spark runtime the
extensions are loaded into -->
+ <dependency>
+ <groupId>org.apache.spark</groupId>
+ <artifactId>spark-sql_${scala.binary.version}</artifactId>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.spark</groupId>
+ <artifactId>spark-catalyst_${scala.binary.version}</artifactId>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.iceberg</groupId>
+
<artifactId>iceberg-spark-runtime-${spark.version.prefix}_${scala.binary.version}</artifactId>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hudi</groupId>
+
<artifactId>hudi-spark${spark.version.prefix}-bundle_${scala.binary.version}</artifactId>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-common</artifactId>
+ <scope>provided</scope>
+ </dependency>
+
+ <!-- Logging API -->
+ <dependency>
+ <groupId>org.apache.logging.log4j</groupId>
+ <artifactId>log4j-api</artifactId>
+ </dependency>
+
+ <!-- Junit -->
+ <dependency>
+ <groupId>org.junit.jupiter</groupId>
+ <artifactId>junit-jupiter-api</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.junit.jupiter</groupId>
+ <artifactId>junit-jupiter-params</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.junit.jupiter</groupId>
+ <artifactId>junit-jupiter-engine</artifactId>
+ <scope>test</scope>
+ </dependency>
+
+ <!-- Test logging -->
+ <dependency>
+ <groupId>org.apache.logging.log4j</groupId>
+ <artifactId>log4j-core</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.logging.log4j</groupId>
+ <artifactId>log4j-slf4j2-impl</artifactId>
+ <scope>test</scope>
+ </dependency>
+ </dependencies>
+
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>net.alchim31.maven</groupId>
+ <artifactId>scala-maven-plugin</artifactId>
+ <executions>
+ <!-- compile Scala before javac so Java sources and tests
can use the Scala classes -->
+ <execution>
+ <id>scala-compile-first</id>
+ <phase>process-resources</phase>
+ <goals>
+ <goal>add-source</goal>
+ <goal>compile</goal>
+ </goals>
+ </execution>
+ <execution>
+ <id>scala-test-compile-first</id>
+ <phase>process-test-resources</phase>
+ <goals>
+ <goal>testCompile</goal>
+ </goals>
+ </execution>
+ </executions>
+ </plugin>
+ </plugins>
+ </build>
+</project>
diff --git
a/xtable-spark-extensions/src/main/scala/org/apache/xtable/spark/extensions/XTableIndex.scala
b/xtable-spark-extensions/src/main/scala/org/apache/xtable/spark/extensions/XTableIndex.scala
new file mode 100644
index 00000000..bd32b30c
--- /dev/null
+++
b/xtable-spark-extensions/src/main/scala/org/apache/xtable/spark/extensions/XTableIndex.scala
@@ -0,0 +1,148 @@
+/*
+ * 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
+
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.connector.catalog.Identifier
+
+import org.apache.iceberg.{CatalogProperties, CatalogUtil, Table}
+import org.apache.iceberg.types.Type.TypeID
+
+import scala.collection.JavaConverters._
+
+import org.apache.xtable.catalog.TableFormatUtils
+import org.apache.xtable.conversion.SourceTable
+import org.apache.xtable.hudi.index.HudiBackedSecondaryIndexReader
+import org.apache.xtable.hudi.index.HudiBackedSecondaryIndexUtils.SyncedVersion
+import org.apache.xtable.iceberg.{IcebergCatalogConfig,
IcebergConversionSourceProvider}
+import org.apache.xtable.model.storage.TableFormat
+
+/**
+ * The XTable index of an Iceberg table, read by
[[HudiBackedSecondaryIndexReader]]. The index is
+ * built and kept current by an XTable sync of the table to a Hudi target with
a secondary index on
+ * a column, for example with {@code RunSync} or {@code XTableSparkSync}. The
Hudi target is stored
+ * at the data path of the Iceberg table, so every reader of the table finds
the index there.
+ */
+case class XTableIndex(spark: SparkSession, sourceTable: SourceTable) {
+
+ lazy val reader: HudiBackedSecondaryIndexReader =
+ new HudiBackedSecondaryIndexReader(sourceTable, new
IcebergConversionSourceProvider, spark)
+
+ /** Whether the index of the column is built. */
+ def isIndexed(column: String): Boolean = reader.doesIndexExist(column)
+
+ /**
+ * The version of the index, if it is built. Its source identifier is the id
of the Iceberg
+ * snapshot that the index was last synced with.
+ */
+ def syncedVersion: Option[SyncedVersion] = {
+ val version = reader.getLastSyncedVersion
+ if (version.isPresent) Some(version.get()) else None
+ }
+}
+
+object XTableIndex {
+
+ /** Column types the pruning rule converts filter literals of to Iceberg
values. */
+ val SupportedTypes: Set[TypeID] = Set(TypeID.STRING, TypeID.INTEGER,
TypeID.LONG)
+
+ /**
+ * The XTable index of a table.
+ *
+ * @param catalogName the Spark catalog the table was loaded from, if any
+ * @param ident the identifier of the table in that catalog, if any
+ */
+ def of(
+ spark: SparkSession,
+ table: Table,
+ catalogName: Option[String],
+ ident: Option[Identifier]): XTableIndex =
+ XTableIndex(
+ spark,
+ toSourceTable(
+ table.name(),
+ table.location(),
+ TableFormatUtils
+ .getTableDataLocation(TableFormat.ICEBERG, table.location(),
table.properties()),
+ catalogName,
+ ident,
+ spark.conf.getAll
+ )
+ )
+
+ /**
+ * Describes how XTable loads the table to read the type of the indexed
column. A table from a
+ * Spark catalog is loaded through an Iceberg catalog built from the same
+ * {@code spark.sql.catalog.<name>.*} options, and a table read by path is
loaded from its
+ * location.
+ *
+ * @param dataPath the location of the data files of the table, where the
Hudi metadata that holds
+ * the index is kept
+ * @param sparkConf the Spark session configuration, which holds the catalog
options
+ */
+ def toSourceTable(
+ tableName: String,
+ tableLocation: String,
+ dataPath: String,
+ catalogName: Option[String],
+ ident: Option[Identifier],
+ sparkConf: Map[String, String]): SourceTable = {
+ val builder = SourceTable
+ .builder()
+ .name(tableName)
+ .basePath(tableLocation)
+ .dataPath(dataPath)
+ .formatName(TableFormat.ICEBERG)
+ for (catalog <- catalogName; identifier <- ident) {
+ val prefix = s"spark.sql.catalog.$catalog."
+ val options = sparkConf.collect {
+ case (key, value) if key.startsWith(prefix) =>
key.substring(prefix.length) -> value
+ }
+ builder
+ .name(identifier.name())
+ .namespace(identifier.namespace())
+ .catalogConfig(
+ IcebergCatalogConfig
+ .builder()
+ .catalogName(catalog)
+ .catalogImpl(catalogImpl(options))
+ .catalogOptions(options.asJava)
+ .build())
+ }
+ builder.build()
+ }
+
+ /** Resolves the catalog implementation from the options the way Iceberg's
Spark catalog does. */
+ private def catalogImpl(options: Map[String, String]): String =
+ options.getOrElse(
+ CatalogProperties.CATALOG_IMPL,
+ options
+ .getOrElse(CatalogUtil.ICEBERG_CATALOG_TYPE,
CatalogUtil.ICEBERG_CATALOG_TYPE_HIVE)
+ .toLowerCase match {
+ case CatalogUtil.ICEBERG_CATALOG_TYPE_HIVE =>
CatalogUtil.ICEBERG_CATALOG_HIVE
+ case CatalogUtil.ICEBERG_CATALOG_TYPE_HADOOP =>
CatalogUtil.ICEBERG_CATALOG_HADOOP
+ case CatalogUtil.ICEBERG_CATALOG_TYPE_REST =>
CatalogUtil.ICEBERG_CATALOG_REST
+ case CatalogUtil.ICEBERG_CATALOG_TYPE_GLUE =>
CatalogUtil.ICEBERG_CATALOG_GLUE
+ case CatalogUtil.ICEBERG_CATALOG_TYPE_NESSIE =>
CatalogUtil.ICEBERG_CATALOG_NESSIE
+ case CatalogUtil.ICEBERG_CATALOG_TYPE_JDBC =>
CatalogUtil.ICEBERG_CATALOG_JDBC
+ case other =>
+ throw new IllegalArgumentException(s"Unknown Iceberg catalog type
$other")
+ }
+ )
+}
diff --git
a/xtable-spark-extensions/src/main/scala/org/apache/xtable/spark/extensions/XTableIndexConf.scala
b/xtable-spark-extensions/src/main/scala/org/apache/xtable/spark/extensions/XTableIndexConf.scala
new file mode 100644
index 00000000..025f8f0c
--- /dev/null
+++
b/xtable-spark-extensions/src/main/scala/org/apache/xtable/spark/extensions/XTableIndexConf.scala
@@ -0,0 +1,55 @@
+/*
+ * 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
+
+import org.apache.spark.sql.internal.SQLConf
+
+/** Session settings for the XTable index rule. */
+object XTableIndexConf {
+
+ /** Whether queries use XTable indexes. Set it to false to read tables
without their index. */
+ val PruningEnabled = "spark.xtable.index.pruning.enabled"
+
+ /**
+ * The rule uses the index only when the files left after Iceberg's
partition and min/max
+ * pruning reach this count, or their size reaches [[MinCandidateBytes]]. A
lookup runs a Spark
+ * job, so it costs more than it saves on a small scan.
+ */
+ val MinCandidateFiles = "spark.xtable.index.pruning.minCandidateFiles"
+ val MinCandidateBytes = "spark.xtable.index.pruning.minCandidateBytes"
+
+ /** The rule does not use the index for a filter with more keys than this. */
+ val MaxKeys = "spark.xtable.index.pruning.maxKeys"
+
+ private val DefaultMinCandidateFiles = 32L
+ private val DefaultMinCandidateBytes = 1024L * 1024 * 1024
+ private val DefaultMaxKeys = 10000L
+
+ def pruningEnabled(conf: SQLConf): Boolean =
+ conf.getConfString(PruningEnabled, "true").toBoolean
+
+ def minCandidateFiles(conf: SQLConf): Long =
+ conf.getConfString(MinCandidateFiles,
DefaultMinCandidateFiles.toString).toLong
+
+ def minCandidateBytes(conf: SQLConf): Long =
+ conf.getConfString(MinCandidateBytes,
DefaultMinCandidateBytes.toString).toLong
+
+ def maxKeys(conf: SQLConf): Long =
+ conf.getConfString(MaxKeys, DefaultMaxKeys.toString).toLong
+}
diff --git
a/xtable-spark-extensions/src/main/scala/org/apache/xtable/spark/extensions/XTableSparkSessionExtensions.scala
b/xtable-spark-extensions/src/main/scala/org/apache/xtable/spark/extensions/XTableSparkSessionExtensions.scala
new file mode 100644
index 00000000..33f123a4
--- /dev/null
+++
b/xtable-spark-extensions/src/main/scala/org/apache/xtable/spark/extensions/XTableSparkSessionExtensions.scala
@@ -0,0 +1,41 @@
+/*
+ * 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
+
+import org.apache.spark.sql.SparkSessionExtensions
+
+import org.apache.xtable.spark.extensions.optimizer.IndexPruningRule
+
+/**
+ * Spark SQL extensions for XTable. Add them after Iceberg's extensions:
+ *
+ * {{{
+ *
spark.sql.extensions=org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions,
+ * org.apache.xtable.spark.extensions.XTableSparkSessionExtensions
+ * }}}
+ *
+ * They add a rule that uses the XTable index of an Iceberg table to read only
the files that hold
+ * the keys of an equality or {@code IN} filter, in SQL queries and DataFrame
reads alike. An XTable
+ * sync to a Hudi target with a secondary index builds the index and keeps it
current.
+ */
+class XTableSparkSessionExtensions extends (SparkSessionExtensions => Unit) {
+ override def apply(extensions: SparkSessionExtensions): Unit = {
+ extensions.injectOptimizerRule(spark => IndexPruningRule(spark))
+ }
+}
diff --git
a/xtable-spark-extensions/src/main/scala/org/apache/xtable/spark/extensions/optimizer/IndexPruningRule.scala
b/xtable-spark-extensions/src/main/scala/org/apache/xtable/spark/extensions/optimizer/IndexPruningRule.scala
new file mode 100644
index 00000000..52b626ee
--- /dev/null
+++
b/xtable-spark-extensions/src/main/scala/org/apache/xtable/spark/extensions/optimizer/IndexPruningRule.scala
@@ -0,0 +1,391 @@
+/*
+ * 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.catalyst.trees.TreeNodeTag
+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.{BaseMetadataTable, FileScanTask, Schema, Snapshot,
Table}
+import org.apache.iceberg.expressions.{Expression => IcebergExpression,
Expressions}
+import org.apache.iceberg.spark.SparkReadOptions
+import org.apache.iceberg.spark.source.SparkTable
+import org.apache.iceberg.types.Type.TypeID
+import org.apache.iceberg.types.Types.NestedField
+
+import scala.collection.JavaConverters._
+import scala.util.control.NonFatal
+
+import org.apache.xtable.hudi.index.HudiBackedSecondaryIndexReader
+import org.apache.xtable.hudi.index.HudiBackedSecondaryIndexUtils.SyncedVersion
+import org.apache.xtable.spark.extensions.{XTableIndex, 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
+ * runs for SQL queries and DataFrame reads alike.
+ *
+ * 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, a
branch, a tag, the changes
+ * between snapshots or a metadata table, or when the scan is too small for a
lookup to pay off. The
+ * rule fails the query when the indexed column was renamed after the index
was synced.
+ */
+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 !IndexPruningRule.SnapshotReadOptions.exists(options.containsKey)
&&
+ table.snapshotId() == null && table.branch() == null &&
+ !table.table().isInstanceOf[BaseMetadataTable] =>
+ val keyFilters = IndexPruningRule.findKeyFilters(condition,
relation.output)
+ // The optimizer runs the rule in several batches until the plan stops
changing, so the
+ // relation remembers the filters it declined, and the lookup runs
once for them. The
+ // optimizer works on a copy of the analyzed plan, so the tag lives
for this query only.
+ val declined =
+
relation.getTagValue(IndexPruningRule.DeclinedKeyFilters).getOrElse(Set.empty)
+ if (keyFilters.isEmpty || declined.contains(keyFilters)) {
+ filter
+ } else {
+ val pruned =
+ try {
+ prune(filter, keyFilters, relation, table)
+ } catch {
+ case e: IndexedColumnRenamedException => throw e
+ case NonFatal(e) =>
+ logWarning(s"Not using the XTable index for ${table.name()}",
e)
+ None
+ }
+ if (pruned.isEmpty) {
+ relation.setTagValue(IndexPruningRule.DeclinedKeyFilters, declined
+ keyFilters)
+ }
+ pruned.getOrElse(filter)
+ }
+ }
+ }
+
+ private def prune(
+ filter: Filter,
+ keyFilters: Seq[(String, Seq[Any])],
+ relation: DataSourceV2Relation,
+ sparkTable: SparkTable): Option[LogicalPlan] = {
+ val table = sparkTable.table()
+ val index =
+ XTableIndex.of(spark, table, relation.catalog.map(_.name()),
relation.identifier)
+ for {
+ (column, keys) <- findIndexedFilter(table, index, keyFilters)
+ if withinKeyLimit(table, column, keys)
+ snapshot <- Option(table.currentSnapshot())
+ version = index.syncedVersion
+ if isCurrent(table, column, version, snapshot.snapshotId())
+ syncedSchema <- schemaOf(table, column, snapshot)
+ _ = checkNotRenamed(table, column, syncedSchema)
+ // the files a scan reads after Iceberg's partition and min/max pruning
+ candidates = planFiles(table, snapshot.snapshotId(), column, keys)
+ if isLargeEnough(table, column, candidates)
+ indexedFiles = lookup(index, column, keys)
+ // A sync, rollback or rebuild that completes between the check above
and the end of the
+ // lookup can point the index at other files than the candidates, even
at the same snapshot.
+ // Each of them completes a new action on the index, so an equal version
after the lookup
+ // shows that the lookup read the checked version.
+ if isUnchanged(table, column, version, index.syncedVersion)
+ } yield {
+ val tasks = candidates.filter(task =>
+
indexedFiles.contains(IndexPruningRule.normalizePath(task.file().location())))
+ if (tasks.isEmpty) {
+ logInfo(
+ s"Using the index of ${table.name()} on $column for ${keys.size}
keys: no file of " +
+ s"${candidates.size} candidate files holds them")
+ filter.copy(child = LocalRelation(relation.output, Nil, isStreaming =
false))
+ } else {
+ StagedScans.stage(table, tasks) match {
+ case Some(setId) =>
+ logInfo(
+ s"Using the index of ${table.name()} on $column for ${keys.size}
keys: reading " +
+ s"${tasks.size} of ${candidates.size} candidate files")
+ registerCleanupListener()
+ val stagedOptions = new java.util.HashMap[String,
String](options(relation))
+ stagedOptions.put(StagedScans.ScanTaskSetIdOption, setId)
+ filter.copy(child =
+ relation.copy(options = new
CaseInsensitiveStringMap(stagedOptions)))
+ case None =>
+ logWarning(
+ s"Not using the index of ${table.name()} on $column: too many
staged scans are " +
+ "live, which happens when plans are optimized but not run")
+ filter
+ }
+ }
+ }
+ }
+
+ /**
+ * Returns the first filter on an indexed column, with the name of the
column in the table
+ * schema. Hudi records the indexed column under that name.
+ */
+ private def findIndexedFilter(
+ table: Table,
+ index: XTableIndex,
+ keyFilters: Seq[(String, Seq[Any])]): Option[(String, Seq[Any])] = {
+ val schema = table.schema()
+ keyFilters.iterator
+ .flatMap { case (attribute, keys) =>
+ val field: NestedField =
+ if (conf.caseSensitiveAnalysis) schema.findField(attribute)
+ else schema.caseInsensitiveFindField(attribute)
+ Option(field)
+ .filter(field =>
XTableIndex.SupportedTypes.contains(field.`type`().typeId()))
+ .map(field => field.name() -> keys)
+ }
+ .find { case (column, _) => index.isIndexed(column) }
+ }
+
+ /**
+ * Returns the schema of the synced snapshot. Snapshots written before
Iceberg tracked the schema
+ * of each snapshot have none, and then a rename of the column cannot be
detected.
+ */
+ private def schemaOf(table: Table, column: String, snapshot: Snapshot):
Option[Schema] = {
+ val schema = Option(snapshot.schemaId()).flatMap(id =>
Option(table.schemas().get(id)))
+ if (schema.isEmpty) {
+ logInfo(
+ s"Not using the index of ${table.name()} on $column: snapshot
${snapshot.snapshotId()} " +
+ "has no schema id")
+ }
+ schema
+ }
+
+ /**
+ * Fails when the column names another field than in the schema of the
synced snapshot. A rename
+ * keeps the snapshot, so the index stays current, but it keys the column by
name and then holds
+ * the values of another field.
+ */
+ private def checkNotRenamed(table: Table, column: String, syncedSchema:
Schema): Unit = {
+ val syncedField = syncedSchema.findField(column)
+ val currentFieldId = table.schema().findField(column).fieldId()
+ if (syncedField == null || syncedField.fieldId() != currentFieldId) {
+ throw new IndexedColumnRenamedException(
+ s"The index of ${table.name()} on $column was synced when $column
named another field. " +
+ s"Rebuild the index after the next commit to the table, or set " +
+ s"${XTableIndexConf.PruningEnabled} to false.")
+ }
+ }
+
+ private def lookup(index: XTableIndex, column: String, keys: Seq[Any]):
Set[String] = {
+ IndexPruningRule.beforeLookup()
+ val keyColumn = spark
+ .createDataset(keys.map(_.toString).distinct.asJava)(Encoders.STRING)
+ .toDF(column)
+ // collect through the RDD, so the lookup runs as plain Spark jobs inside
the planning query
+ val files = index.reader
+ .lookup(index.sourceTable, keyColumn, column)
+ .select(HudiBackedSecondaryIndexReader.FILE_COLUMN)
+ .rdd
+ .map(_.getString(0))
+ .distinct()
+ .collect()
+ .map(IndexPruningRule.normalizePath)
+ .toSet
+ IndexPruningRule.afterLookup()
+ files
+ }
+
+ private def withinKeyLimit(table: Table, column: String, keys: Seq[Any]):
Boolean = {
+ val withinLimit = keys.size <= XTableIndexConf.maxKeys(conf)
+ if (!withinLimit) {
+ logInfo(s"Not using the index of ${table.name()} on $column:
${keys.size} keys")
+ }
+ withinLimit
+ }
+
+ private def isCurrent(
+ table: Table,
+ column: String,
+ version: Option[SyncedVersion],
+ currentSnapshot: Long): Boolean = {
+ val syncedSnapshot = version.map(_.getSourceIdentifier)
+ val current = syncedSnapshot.contains(currentSnapshot.toString)
+ if (!current) {
+ logWarning(
+ s"Not using the index of ${table.name()} on $column: it is synced to
snapshot " +
+ s"${syncedSnapshot.getOrElse("none")}, but the table is at
$currentSnapshot. Run the " +
+ "XTable sync of the table to update the index")
+ }
+ current
+ }
+
+ private def isUnchanged(
+ table: Table,
+ column: String,
+ checkedVersion: Option[SyncedVersion],
+ versionAfterLookup: Option[SyncedVersion]): Boolean = {
+ val unchanged = versionAfterLookup == checkedVersion
+ if (!unchanged) {
+ logInfo(
+ s"Not using the index of ${table.name()} on $column: it changed from
$checkedVersion " +
+ s"to ${versionAfterLookup.getOrElse("none")} during the lookup")
+ }
+ unchanged
+ }
+
+ private def isLargeEnough(
+ table: Table,
+ column: String,
+ candidates: Seq[FileScanTask]): Boolean = {
+ val candidateBytes = candidates.map(_.file().fileSizeInBytes()).sum
+ val largeEnough = candidates.size >=
XTableIndexConf.minCandidateFiles(conf) ||
+ candidateBytes >= XTableIndexConf.minCandidateBytes(conf)
+ if (!largeEnough) {
+ logInfo(
+ s"Not using the index of ${table.name()} on $column: the scan reads
only " +
+ s"${candidates.size} files of $candidateBytes bytes")
+ }
+ largeEnough
+ }
+
+ private def options(relation: DataSourceV2Relation): java.util.Map[String,
String] =
+ relation.options.asCaseSensitiveMap()
+
+ private def planFiles(
+ table: Table,
+ snapshotId: Long,
+ column: String,
+ keys: Seq[Any]): Seq[FileScanTask] = {
+ val fieldType = table.schema().findField(column).`type`().typeId()
+ val values = keys.map(IndexPruningRule.toIcebergValue(_, fieldType))
+ val predicate: IcebergExpression =
+ if (values.size == 1) Expressions.equal(column, values.head)
+ else Expressions.in(column, values: _*)
+ val files = table
+ .newScan()
+ .useSnapshot(snapshotId)
+ .caseSensitive(conf.caseSensitiveAnalysis)
+ .filter(predicate)
+ .planFiles()
+ try files.asScala.toList
+ finally files.close()
+ }
+
+ // registered on first use, because the session state does not exist yet
when the rule is built
+ private def registerCleanupListener(): Unit = {
+ if (listenerRegistered.compareAndSet(false, true)) {
+ spark.listenerManager.register(new StagedScanCleanupListener)
+ }
+ }
+}
+
+/** Thrown when the indexed column was renamed after the index was synced. */
+class IndexedColumnRenamedException(message: String) extends
IllegalStateException(message)
+
+object IndexPruningRule {
+
+ /**
+ * Read options that select other rows than the current snapshot of the
table, or that already
+ * select the tasks of the scan. The rule leaves scans with any of them
unchanged.
+ */
+ val SnapshotReadOptions: Seq[String] = Seq(
+ SparkReadOptions.SNAPSHOT_ID,
+ SparkReadOptions.AS_OF_TIMESTAMP,
+ SparkReadOptions.VERSION_AS_OF,
+ SparkReadOptions.TIMESTAMP_AS_OF,
+ SparkReadOptions.BRANCH,
+ SparkReadOptions.TAG,
+ SparkReadOptions.START_SNAPSHOT_ID,
+ SparkReadOptions.END_SNAPSHOT_ID,
+ SparkReadOptions.START_TIMESTAMP,
+ SparkReadOptions.END_TIMESTAMP,
+ StagedScans.ScanTaskSetIdOption
+ )
+
+ /** The key filters for which the rule left a relation unchanged during the
current query. */
+ private val DeclinedKeyFilters =
+ TreeNodeTag[Set[Seq[(String, Seq[Any])]]]("xtable.declinedKeyFilters")
+
+ /** Runs after the freshness check and before the lookup. Tests use it to
run a sync between. */
+ @volatile private[extensions] var beforeLookup: () => Unit = () => ()
+
+ /** Runs after the lookup and before the second check. Tests use it to
change the index there. */
+ @volatile private[extensions] var afterLookup: () => Unit = () => ()
+
+ /**
+ * Finds the conjuncts of the condition that compare a column of the
relation with literals, and
+ * returns the name of each column with its non-null literal values in
Spark's internal
+ * representation.
+ */
+ def findKeyFilters(condition: Expression, output: Seq[Attribute]):
Seq[(String, Seq[Any])] = {
+ def column(attribute: AttributeReference): Option[String] =
+ if (output.exists(_.exprId == attribute.exprId)) Some(attribute.name)
else None
+
+ splitConjuncts(condition)
+ .flatMap {
+ case EqualTo(attribute: AttributeReference, Literal(value, _)) if
value != null =>
+ column(attribute).map(_ -> Seq(value))
+ case EqualTo(Literal(value, _), attribute: AttributeReference) if
value != null =>
+ column(attribute).map(_ -> Seq(value))
+ case In(attribute: AttributeReference, values) if
values.forall(_.isInstanceOf[Literal]) =>
+ val keys = values.map(_.asInstanceOf[Literal].value).filter(_ !=
null)
+ column(attribute).map(_ -> keys)
+ case InSet(attribute: AttributeReference, values) =>
+ column(attribute).map(_ -> values.filter(_ != null).toSeq)
+ case _ => None
+ }
+ .filter(_._2.nonEmpty)
+ }
+
+ private def splitConjuncts(condition: Expression): Seq[Expression] =
condition match {
+ case And(left, right) => splitConjuncts(left) ++ splitConjuncts(right)
+ case other => Seq(other)
+ }
+
+ /** Converts a literal value from Spark's internal representation for an
Iceberg expression. */
+ def toIcebergValue(value: Any, typeId: TypeID): AnyRef = (value, typeId)
match {
+ case (v: Int, TypeID.INTEGER) => Int.box(v)
+ case (v: Int, TypeID.LONG) => Long.box(v.toLong)
+ case (v: Long, TypeID.LONG) => Long.box(v)
+ case (v, TypeID.STRING) => v.toString
+ case _ =>
+ throw new IllegalArgumentException(s"Unsupported key $value for a column
of type $typeId")
+ }
+
+ /** Paths from the index and from Iceberg can differ in scheme, so both
sides compare the path. */
+ def normalizePath(path: String): String = new Path(path).toUri.getPath
+}
diff --git
a/xtable-spark-extensions/src/main/scala/org/apache/xtable/spark/extensions/optimizer/StagedScans.scala
b/xtable-spark-extensions/src/main/scala/org/apache/xtable/spark/extensions/optimizer/StagedScans.scala
new file mode 100644
index 00000000..303ca1ce
--- /dev/null
+++
b/xtable-spark-extensions/src/main/scala/org/apache/xtable/spark/extensions/optimizer/StagedScans.scala
@@ -0,0 +1,107 @@
+/*
+ * 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.QueryPlanningTracker
+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._
+
+/**
+ * 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. Iceberg reads
+ * the tasks of a scan only when Spark plans its partitions, so a set is never
removed before its
+ * query ends. Plans that are only explained never end and leave their sets
staged, so this object
+ * stages no new set once too many are live, and the scans that would use them
read the table as
+ * usual.
+ */
+object StagedScans {
+ val ScanTaskSetIdOption = "scan-task-set-id"
+
+ val MaxLiveSets = 1000
+
+ private val liveSets = new java.util.HashMap[String, Table]()
+
+ /** Stages the tasks and returns the id of their set, or nothing when too
many sets are live. */
+ def stage(table: Table, tasks: Seq[FileScanTask]): Option[String] = {
+ val setId = UUID.randomUUID().toString
+ val reserved = liveSets.synchronized {
+ liveSets.size() < MaxLiveSets && liveSets.put(setId, table) == null
+ }
+ if (reserved) {
+ ScanTaskSetManager.get().stageTasks(table, setId, tasks.asJava)
+ Some(setId)
+ } else {
+ None
+ }
+ }
+
+ def release(setId: String): Unit = {
+ val table = liveSets.synchronized(liveSets.remove(setId))
+ if (table != null) {
+ removeTasks(table, setId)
+ }
+ }
+
+ def liveSetIds: Set[String] =
liveSets.synchronized(liveSets.keySet().asScala.toSet)
+
+ /** Releases the task sets that the scans of a plan use. */
+ def releaseAll(plan: LogicalPlan): Unit = {
+ def setIdOf(relation: DataSourceV2Relation): Option[String] =
+ Option(relation.options.get(ScanTaskSetIdOption))
+ plan
+ .collectWithSubqueries {
+ case scan: DataSourceV2ScanRelation => setIdOf(scan.relation)
+ case relation: DataSourceV2Relation => setIdOf(relation)
+ }
+ .flatten
+ .foreach(release)
+ }
+
+ private def removeTasks(table: Table, setId: String): Unit =
+ ScanTaskSetManager.get().removeTasks[FileScanTask](table, setId)
+}
+
+/** Releases the staged task sets of a query once it ends. */
+class StagedScanCleanupListener extends QueryExecutionListener {
+ override def onSuccess(funcName: String, qe: QueryExecution, durationNs:
Long): Unit =
+ release(qe)
+
+ override def onFailure(funcName: String, qe: QueryExecution, exception:
Exception): Unit =
+ release(qe)
+
+ // A query that failed during optimization has no optimized plan, and
reading it would run the
+ // optimizer and its index lookups again. Spark records the phase only when
it completes.
+ private def release(qe: QueryExecution): Unit =
+ if (qe.tracker.phases.contains(QueryPlanningTracker.OPTIMIZATION)) {
+ StagedScans.releaseAll(qe.optimizedPlan)
+ }
+}
diff --git
a/xtable-spark-extensions/src/test/java/org/apache/xtable/spark/extensions/ITXTableIndexExtensions.java
b/xtable-spark-extensions/src/test/java/org/apache/xtable/spark/extensions/ITXTableIndexExtensions.java
new file mode 100644
index 00000000..6e957bcf
--- /dev/null
+++
b/xtable-spark-extensions/src/test/java/org/apache/xtable/spark/extensions/ITXTableIndexExtensions.java
@@ -0,0 +1,742 @@
+/*
+ * 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;
+
+import static org.apache.spark.sql.functions.col;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Properties;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Supplier;
+import java.util.stream.Collectors;
+import java.util.stream.LongStream;
+
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.spark.SparkConf;
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Row;
+import org.apache.spark.sql.SparkSession;
+import org.apache.spark.sql.catalyst.plans.logical.LocalRelation;
+import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan;
+import org.apache.spark.sql.execution.datasources.v2.DataSourceV2ScanRelation;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import org.apache.hudi.client.HoodieReadClient;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+
+import org.apache.iceberg.FileScanTask;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.spark.ScanTaskSetManager;
+import org.apache.iceberg.spark.Spark3Util;
+
+import scala.Option;
+import scala.collection.JavaConverters;
+
+import org.apache.xtable.catalog.TableFormatUtils;
+import org.apache.xtable.conversion.ConversionConfig;
+import org.apache.xtable.conversion.ConversionController;
+import org.apache.xtable.conversion.SourceTable;
+import org.apache.xtable.conversion.TargetTable;
+import org.apache.xtable.hudi.HudiTargetConfig;
+import org.apache.xtable.iceberg.IcebergConversionSourceProvider;
+import org.apache.xtable.model.storage.TableFormat;
+import org.apache.xtable.model.sync.SyncResult;
+import org.apache.xtable.model.sync.SyncStatusCode;
+import org.apache.xtable.spark.extensions.optimizer.IndexPruningRule;
+import org.apache.xtable.spark.extensions.optimizer.IndexPruningRule$;
+import
org.apache.xtable.spark.extensions.optimizer.IndexedColumnRenamedException;
+import org.apache.xtable.spark.extensions.optimizer.StagedScans;
+
+/**
+ * Runs the index rule end to end against Iceberg tables in a Hadoop catalog,
with indexes that an
+ * XTable sync builds, and checks every query against the same query without
the rule.
+ */
+public class ITXTableIndexExtensions {
+ private static String TABLE;
+
+ @TempDir public static Path tempDir;
+
+ private static SparkSession spark;
+
+ @BeforeAll
+ public static void setupOnce() {
+ SparkConf sparkConf =
+ new SparkConf()
+ .setAppName("xtable-spark-extensions-testing")
+ .setMaster("local[2]")
+ .set(
+ "spark.sql.extensions",
+
"org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions,"
+ + XTableSparkSessionExtensions.class.getName())
+ .set("spark.sql.catalog.cat",
"org.apache.iceberg.spark.SparkCatalog")
+ .set("spark.sql.catalog.cat.type", "hadoop")
+ .set("spark.sql.catalog.cat.warehouse", tempDir.toUri().toString())
+ .set("spark.sql.shuffle.partitions", "2")
+ .set("spark.sql.adaptive.enabled", "false")
+ .set("spark.sql.session.timeZone", "UTC")
+ .set("spark.ui.enabled", "false")
+ // Hudi, which stores the index, needs Kryo
+ .set("spark.serializer",
"org.apache.spark.serializer.KryoSerializer")
+ // a lookup always pays off in these small tables
+ .set(XTableIndexConf.MinCandidateFiles(), "0")
+ .set(XTableIndexConf.MinCandidateBytes(), "0");
+ spark =
+
SparkSession.builder().config(HoodieReadClient.addHoodieSupport(sparkConf)).getOrCreate();
+ spark
+ .sparkContext()
+ .hadoopConfiguration()
+ .set("parquet.avro.write-old-list-structure", "false");
+ }
+
+ @AfterAll
+ public static void teardown() {
+ if (spark != null) {
+ spark.close();
+ }
+ }
+
+ @BeforeEach
+ public void newTableName() {
+ // a dropped table can leave its index behind, so every test uses a new
table
+ TABLE = "cat.db.events_" + UUID.randomUUID().toString().replace("-", "_");
+ }
+
+ @AfterEach
+ public void dropTable() {
+ spark.sql("DROP TABLE IF EXISTS " + TABLE + " PURGE");
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void indexLifecycle(boolean partitioned) throws Exception {
+ createTable(partitioned);
+ appendRows(0, 1600, 4);
+
+ List<String> emails = emails(7, 123, 1599);
+ // a table without an index is read as usual
+ assertDoesNotUseIndex("SELECT * FROM " + TABLE + " WHERE email = '" +
emails.get(0) + "'");
+
+ syncIndex("email");
+ assertUsesIndex("SELECT * FROM " + TABLE + " WHERE email = '" +
emails.get(0) + "'");
+ assertUsesIndex("SELECT * FROM " + TABLE + " WHERE '" + emails.get(0) + "'
= email");
+ assertUsesIndex("SELECT * FROM " + TABLE + " WHERE email IN " +
inList(emails));
+ // more than 10 values become an InSet
+ assertUsesIndex(
+ "SELECT * FROM " + TABLE + " WHERE email IN " +
inList(emails(LongStream.range(0, 20))));
+ assertUsesIndex(
+ "SELECT id FROM " + TABLE + " WHERE email IN " + inList(emails) + "
AND id > 100");
+ // the filter on the indexed column need not come first
+ assertUsesIndex(
+ "SELECT id FROM " + TABLE + " WHERE id = 7 AND email = '" +
emails.get(0) + "'");
+ assertReturnsNothingWithoutScan("SELECT * FROM " + TABLE + " WHERE email =
'missing'");
+ // an OR of other columns cannot use the index
+ assertDoesNotUseIndex(
+ "SELECT * FROM " + TABLE + " WHERE email = '" + emails.get(0) + "' OR
id = 5");
+ // a column without an index
+ assertDoesNotUseIndex("SELECT * FROM " + TABLE + " WHERE id = 7");
+
+ // queries of an older snapshot do not use the index
+ long snapshotId = icebergTable().currentSnapshot().snapshotId();
+ assertDoesNotUseIndex(
+ "SELECT * FROM "
+ + TABLE
+ + " VERSION AS OF "
+ + snapshotId
+ + " WHERE email = '"
+ + emails.get(0)
+ + "'");
+ // nor do queries with the rule disabled
+ withConf(
+ XTableIndexConf.PruningEnabled(),
+ "false",
+ () ->
+ assertDoesNotUseIndex("SELECT * FROM " + TABLE + " WHERE email IN
" + inList(emails)));
+ // nor do scans below the size thresholds
+ withConf(
+ XTableIndexConf.MinCandidateFiles(),
+ "100000",
+ () ->
+ withConf(
+ XTableIndexConf.MinCandidateBytes(),
+ String.valueOf(Long.MAX_VALUE),
+ () ->
+ assertDoesNotUseIndex(
+ "SELECT * FROM " + TABLE + " WHERE email IN " +
inList(emails))));
+
+ // new rows make the index stale until the next sync
+ appendRows(1600, 1700, 2);
+ List<String> newEmails = emails(7, 1650);
+ assertDoesNotUseIndex("SELECT * FROM " + TABLE + " WHERE email IN " +
inList(newEmails));
+ syncIndex("email");
+ assertUsesIndex("SELECT * FROM " + TABLE + " WHERE email IN " +
inList(newEmails));
+
+ // deleted rows are gone after a sync
+ spark.sql("DELETE FROM " + TABLE + " WHERE id < 10");
+ syncIndex("email");
+ assertReturnsNothingWithoutScan(
+ "SELECT * FROM " + TABLE + " WHERE email IN " + inList(emails(7, 8)));
+ assertUsesIndex("SELECT * FROM " + TABLE + " WHERE email IN " +
inList(emails(8, 123)));
+
+ // without the Hudi metadata the table has no index
+ deleteIndex();
+ assertDoesNotUseIndex("SELECT * FROM " + TABLE + " WHERE email = '" +
emails.get(1) + "'");
+
+ // a sync builds the index again, also on another column, without a new
snapshot
+ syncIndex("id");
+ assertUsesIndex("SELECT * FROM " + TABLE + " WHERE id IN (20, 900, 1650)");
+ assertDoesNotUseIndex("SELECT * FROM " + TABLE + " WHERE email = '" +
emails.get(1) + "'");
+
+ // staged scans are released once their queries end
+ spark.sparkContext().listenerBus().waitUntilEmpty();
+ assertTrue(StagedScans.liveSetIds().isEmpty());
+ }
+
+ @Test
+ void dataFrameReadsUseIndex() throws Exception {
+ createTable(false);
+ appendRows(0, 400, 4);
+ syncIndex("email");
+ List<String> emails = emails(7, 123);
+
+ assertUsesIndex(() ->
spark.table(TABLE).where(col("email").isin(emails.toArray())));
+ assertUsesIndex(
+ () ->
+
spark.read().format("iceberg").load(TABLE).where(col("email").equalTo(emails.get(0))));
+ // a read of an older snapshot by option does not use the index
+ long snapshotId = icebergTable().currentSnapshot().snapshotId();
+ assertDoesNotUseIndex(
+ () ->
+ spark
+ .read()
+ .format("iceberg")
+ .option("snapshot-id", snapshotId)
+ .load(TABLE)
+ .where(col("email").equalTo(emails.get(0))));
+ }
+
+ @Test
+ void incrementalReadsDoNotUseIndex() throws Exception {
+ createTable(false);
+ appendRows(0, 400, 4);
+ long firstSnapshot = icebergTable().currentSnapshot().snapshotId();
+ appendRows(400, 800, 4);
+ long secondSnapshot = icebergTable().currentSnapshot().snapshotId();
+ syncIndex("email");
+
+ // row 7 is in the first snapshot, so a read of the changes after it does
not return the row
+ String email = emails(7).get(0);
+ Supplier<Dataset<Row>> changes =
+ () ->
+ spark
+ .read()
+ .format("iceberg")
+ .option("start-snapshot-id", firstSnapshot)
+ .option("end-snapshot-id", secondSnapshot)
+ .load(TABLE)
+ .where(col("email").equalTo(email));
+ assertDoesNotUseIndex(changes);
+ assertTrue(changes.get().collectAsList().isEmpty());
+ String newEmail = emails(407).get(0);
+ assertDoesNotUseIndex(
+ () ->
+ spark
+ .read()
+ .format("iceberg")
+ .option("start-snapshot-id", firstSnapshot)
+ .load(TABLE)
+ .where(col("email").equalTo(newEmail)));
+ }
+
+ /**
+ * A sync that completes between the freshness check and the lookup points
the index at files that
+ * the candidates of the checked snapshot do not have. The rule must then
read the table as usual
+ * instead of reading no file.
+ */
+ @Test
+ void syncDuringLookupKeepsTheRows() throws Exception {
+ createTable(false);
+ appendRows(0, 400, 4);
+ syncIndex("email");
+ String email = emails(7).get(0);
+ String sql = "SELECT id FROM " + TABLE + " WHERE email = '" + email + "'";
+ // another row in the file of row 7, whose copy-on-write delete rewrites
the file and keeps row
+ // 7
+ long neighbor =
+ spark
+ .sql(
+ "SELECT id FROM "
+ + TABLE
+ + " WHERE id <> 7 AND _file = (SELECT _file FROM "
+ + TABLE
+ + " WHERE id = 7) LIMIT 1")
+ .collectAsList()
+ .get(0)
+ .getLong(0);
+
+ IndexPruningRule$.MODULE$.beforeLookup_$eq(
+ () -> {
+ IndexPruningRule$.MODULE$.beforeLookup_$eq(() -> null);
+ spark.sql("DELETE FROM " + TABLE + " WHERE id = " + neighbor);
+ syncIndex("email");
+ return null;
+ });
+ try {
+ Dataset<Row> query = spark.sql(sql);
+ // The rule must not intersect the candidates of the checked snapshot
with the files of the
+ // newer index, which share no file and would read nothing.
+ assertFalse(
+ leaves(query.queryExecution().optimizedPlan()).stream()
+ .allMatch(leaf -> leaf instanceof LocalRelation),
+ "expected a scan when the index changes during the lookup");
+ assertEquals(
+ Collections.singletonList(7L),
+ query.collectAsList().stream().map(row ->
row.getLong(0)).collect(Collectors.toList()));
+ } finally {
+ IndexPruningRule$.MODULE$.beforeLookup_$eq(() -> null);
+ }
+ // the next query sees the synced index
+ assertUsesIndex(sql);
+ }
+
+ /**
+ * Iceberg fetches the staged tasks of a scan only when Spark plans its
partitions, so planning
+ * other queries must not remove them. Once the registry is full, new scans
read the table as
+ * usual.
+ */
+ @Test
+ void stagedScansStayLiveUntilTheirQueryEnds() throws Exception {
+ createTable(false);
+ appendRows(0, 400, 4);
+ syncIndex("email");
+ String email = emails(7).get(0);
+ Dataset<Row> query = spark.sql("SELECT id FROM " + TABLE + " WHERE email =
'" + email + "'");
+ assertEquals(1,
stagedSetIds(query.queryExecution().optimizedPlan()).size());
+
+ List<FileScanTask> fillerTasks = new ArrayList<>();
+ try (CloseableIterable<FileScanTask> tasks =
icebergTable().newScan().planFiles()) {
+ fillerTasks.add(tasks.iterator().next());
+ }
+ List<String> fillers = new ArrayList<>();
+ while (StagedScans.liveSetIds().size() < StagedScans.MaxLiveSets()) {
+ fillers.add(
+ StagedScans.stage(icebergTable(),
JavaConverters.asScalaBuffer(fillerTasks).toSeq())
+ .get());
+ }
+ try {
+ assertDoesNotUseIndex("SELECT * FROM " + TABLE + " WHERE email = '" +
email + "'");
+ assertEquals(
+ Collections.singletonList(7L),
+ query.collectAsList().stream().map(row ->
row.getLong(0)).collect(Collectors.toList()));
+ } finally {
+ fillers.forEach(StagedScans::release);
+ }
+ }
+
+ /**
+ * A rollback and a rebuild of the index after the lookup bring the index
back to the checked
+ * snapshot, but the lookup read the files of a newer index. The rule must
compare the version of
+ * the index, not only its snapshot, and read the table as usual.
+ */
+ @Test
+ void rebuildDuringLookupKeepsTheRows() throws Exception {
+ createTable(false);
+ appendRows(0, 400, 4);
+ syncIndex("email");
+ long checkedSnapshot = icebergTable().currentSnapshot().snapshotId();
+ String email = emails(7).get(0);
+ String sql = "SELECT id FROM " + TABLE + " WHERE email = '" + email + "'";
+ long neighbor = neighborOf(7);
+
+ // before the lookup: a copy-on-write delete rewrites the file of row 7,
and a sync indexes it
+ IndexPruningRule$.MODULE$.beforeLookup_$eq(
+ () -> {
+ IndexPruningRule$.MODULE$.beforeLookup_$eq(() -> null);
+ spark.sql("DELETE FROM " + TABLE + " WHERE id = " + neighbor);
+ syncIndex("email");
+ return null;
+ });
+ // after the lookup: the table rolls back to the checked snapshot, and the
index is rebuilt
+ IndexPruningRule$.MODULE$.afterLookup_$eq(
+ () -> {
+ IndexPruningRule$.MODULE$.afterLookup_$eq(() -> null);
+
icebergTable().manageSnapshots().rollbackTo(checkedSnapshot).commit();
+ try {
+ deleteIndex();
+ } catch (Exception e) {
+ throw new IllegalStateException(e);
+ }
+ syncIndex("email");
+ return null;
+ });
+ try {
+ assertEquals(
+ Collections.singletonList(7L),
+ spark.sql(sql).collectAsList().stream()
+ .map(row -> row.getLong(0))
+ .collect(Collectors.toList()));
+ } finally {
+ IndexPruningRule$.MODULE$.beforeLookup_$eq(() -> null);
+ IndexPruningRule$.MODULE$.afterLookup_$eq(() -> null);
+ }
+ assertEquals(checkedSnapshot,
icebergTable().currentSnapshot().snapshotId());
+ // the next query sees the rebuilt index
+ assertUsesIndex(sql);
+ }
+
+ /**
+ * Renames keep the snapshot of the table, so the index stays current, but
after a swap of two
+ * column names the index of a column describes the values of another field.
The query fails
+ * instead of returning the rows of the other field.
+ */
+ @Test
+ void swappedColumnNamesFailQuery() throws Exception {
+ createTable(false);
+ appendRows(0, 400, 4);
+ syncIndex("email");
+ assertUsesIndex("SELECT id, email FROM " + TABLE + " WHERE email = '" +
emails(7).get(0) + "'");
+
+ spark.sql("ALTER TABLE " + TABLE + " RENAME COLUMN email TO old_email");
+ spark.sql("ALTER TABLE " + TABLE + " RENAME COLUMN level TO email");
+ String sql = "SELECT id, email FROM " + TABLE + " WHERE email = 'level_1'";
+ assertThrows(IndexedColumnRenamedException.class, () ->
spark.sql(sql).count());
+ // ids 1, 4, ..., 397
+ withConf(
+ XTableIndexConf.PruningEnabled(), "false", () -> assertEquals(133,
spark.sql(sql).count()));
+ }
+
+ /**
+ * Metadata tables report the snapshot of their table, but their scans plan
metadata rows instead
+ * of data files. A filter on a metadata column that has the name of the
indexed column must read
+ * the metadata table as usual, without a lookup.
+ */
+ @Test
+ void metadataTablesDoNotUseIndex() throws Exception {
+ // the files table of the table has a column named content too
+ spark.sql(
+ "CREATE TABLE "
+ + TABLE
+ + " (id BIGINT, content INT) USING iceberg TBLPROPERTIES
('format-version'='2')");
+ spark.range(0, 100).selectExpr("id", "0 AS
content").repartition(4).writeTo(TABLE).append();
+ syncIndex("content");
+ assertUsesIndex("SELECT * FROM " + TABLE + " WHERE content = 0");
+ AtomicInteger lookups = new AtomicInteger();
+ IndexPruningRule$.MODULE$.beforeLookup_$eq(
+ () -> {
+ lookups.incrementAndGet();
+ return null;
+ });
+ try {
+ assertDoesNotUseIndex("SELECT file_path FROM " + TABLE + ".files WHERE
content = 0");
+ assertEquals(0, lookups.get());
+ } finally {
+ IndexPruningRule$.MODULE$.beforeLookup_$eq(() -> null);
+ }
+ assertEquals(4, spark.sql("SELECT * FROM " + TABLE + ".files WHERE content
= 0").count());
+ }
+
+ /**
+ * The optimizer runs the rule in several batches until the plan stops
changing. Once the rule
+ * leaves a scan unchanged after a lookup, it must not look the same keys up
again in the query.
+ */
+ @Test
+ void declinedLookupRunsOnce() throws Exception {
+ createTable(false);
+ appendRows(0, 400, 4);
+ syncIndex("email");
+ String sql = "SELECT id FROM " + TABLE + " WHERE email = '" +
emails(7).get(0) + "'";
+ AtomicInteger lookups = new AtomicInteger();
+ IndexPruningRule$.MODULE$.afterLookup_$eq(
+ () -> {
+ lookups.incrementAndGet();
+ throw new IllegalStateException("lookup failed");
+ });
+ try {
+ assertDoesNotUseIndex(sql);
+ // assertDoesNotUseIndex plans the query once with the rule
+ assertEquals(1, lookups.get());
+ } finally {
+ IndexPruningRule$.MODULE$.afterLookup_$eq(() -> null);
+ }
+ }
+
+ /** Returns another row in the file of the row, which a copy-on-write delete
rewrites. */
+ private static long neighborOf(long id) {
+ return spark
+ .sql(
+ "SELECT id FROM "
+ + TABLE
+ + " WHERE id <> "
+ + id
+ + " AND _file = (SELECT _file FROM "
+ + TABLE
+ + " WHERE id = "
+ + id
+ + ") LIMIT 1")
+ .collectAsList()
+ .get(0)
+ .getLong(0);
+ }
+
+ /** Builds or updates the index of the column the way an XTable sync job
does. */
+ private static void syncIndex(String column) {
+ Table table = icebergTable();
+ String dataPath = dataPath(table);
+ SourceTable sourceTable =
+ XTableIndex.toSourceTable(
+ table.name(),
+ table.location(),
+ dataPath,
+ Option.empty(),
+ Option.empty(),
+ scala.collection.immutable.Map$.MODULE$.<String, String>empty());
+ Properties properties = new Properties();
+ properties.setProperty(HudiTargetConfig.HUDI_TABLE_VERSION, "9");
+ properties.setProperty(
+ HudiTargetConfig.EXECUTION_ENGINE,
HudiTargetConfig.ExecutionEngine.SPARK.getConfigValue());
+ properties.setProperty(HudiTargetConfig.SECONDARY_INDEX_COLUMN, column);
+ TargetTable targetTable =
+ TargetTable.builder()
+ .name(sourceTable.getName())
+ .basePath(dataPath)
+ .formatName(TableFormat.HUDI)
+ .additionalProperties(properties)
+ .build();
+ ConversionConfig conversionConfig =
+ ConversionConfig.builder()
+ .sourceTable(sourceTable)
+ .targetTables(Collections.singletonList(targetTable))
+ .build();
+ IcebergConversionSourceProvider sourceProvider = new
IcebergConversionSourceProvider();
+ sourceProvider.init(spark.sparkContext().hadoopConfiguration());
+ SyncResult syncResult =
+ new ConversionController(spark.sparkContext().hadoopConfiguration())
+ .sync(conversionConfig, sourceProvider)
+ .get(TableFormat.HUDI);
+ // no result means that the table has no commit the index has not seen
+ if (syncResult != null) {
+ assertEquals(
+ SyncStatusCode.SUCCESS,
+ syncResult.getTableFormatSyncStatus().getStatusCode(),
+ syncResult.getErrorMessage());
+ }
+ }
+
+ /** Deletes the Hudi metadata that holds the index, and keeps the data
files. */
+ private static void deleteIndex() throws Exception {
+ org.apache.hadoop.fs.Path metaPath =
+ new org.apache.hadoop.fs.Path(
+ dataPath(icebergTable()), HoodieTableMetaClient.METAFOLDER_NAME);
+ FileSystem fs =
metaPath.getFileSystem(spark.sparkContext().hadoopConfiguration());
+ assertTrue(fs.delete(metaPath, true));
+ }
+
+ private static String dataPath(Table table) {
+ return TableFormatUtils.getTableDataLocation(
+ TableFormat.ICEBERG, table.location(), table.properties());
+ }
+
+ private static void createTable(boolean partitioned) {
+ spark.sql(
+ "CREATE TABLE "
+ + TABLE
+ + " (id BIGINT, email STRING, level STRING, score DOUBLE) USING
iceberg"
+ + (partitioned ? " PARTITIONED BY (level)" : "")
+ + " TBLPROPERTIES ('format-version'='2',
'write.distribution-mode'='none',"
+ + " 'write.delete.mode'='copy-on-write')");
+ }
+
+ /** Appends rows with ids in [from, to) as several files whose email ranges
overlap. */
+ private static void appendRows(long from, long to, int files) throws
Exception {
+ spark
+ .range(from, to)
+ .selectExpr(
+ "id",
+ "md5(CAST(id AS STRING)) AS email",
+ "CONCAT('level_', CAST(id % 3 AS STRING)) AS level",
+ "CAST(id AS DOUBLE) / 7 AS score")
+ .repartition(files)
+ .sortWithinPartitions("level")
+ .writeTo(TABLE)
+ .append();
+ }
+
+ /**
+ * Checks that the query reads only the files that hold its keys, and
returns the same rows as
+ * without the index.
+ */
+ private static void assertUsesIndex(String sql) {
+ assertUsesIndex(
+ sql,
+ () -> spark.sql(sql),
+ () -> spark.sql(sql.replaceFirst("SELECT .*? FROM", "SELECT _file
FROM")),
+ // the index selects files by the key filter alone, so other filters
can drop rows of them
+ !sql.contains(" AND "));
+ }
+
+ /** Checks a DataFrame read that filters the table without a projection. */
+ private static void assertUsesIndex(Supplier<Dataset<Row>> newQuery) {
+ assertUsesIndex(
+ newQuery.get().queryExecution().analyzed().toString(),
+ newQuery,
+ () -> newQuery.get().select("_file"),
+ true);
+ }
+
+ private static void assertUsesIndex(
+ String description,
+ Supplier<Dataset<Row>> newQuery,
+ Supplier<Dataset<Row>> newFileQuery,
+ boolean readsOnlyFilesWithMatches) {
+ Dataset<Row> query = newQuery.get();
+ List<String> setIds = stagedSetIds(query.queryExecution().optimizedPlan());
+ assertEquals(1, setIds.size(), "expected the index to select the files of:
" + description);
+ Set<String> stagedFiles =
+ ScanTaskSetManager.get().<FileScanTask>fetchTasks(icebergTable(),
setIds.get(0)).stream()
+ .map(task ->
IndexPruningRule.normalizePath(task.file().location()))
+ .collect(Collectors.toSet());
+ List<Row> expected = withoutIndex(newQuery);
+ Set<String> filesWithMatches = new java.util.HashSet<>();
+ withoutIndex(newFileQuery)
+ .forEach(row ->
filesWithMatches.add(IndexPruningRule.normalizePath(row.getString(0))));
+ if (readsOnlyFilesWithMatches) {
+ assertEquals(filesWithMatches, stagedFiles, "files read for: " +
description);
+ } else {
+ assertTrue(stagedFiles.containsAll(filesWithMatches), "files read for: "
+ description);
+ assertTrue(stagedFiles.size() < dataFileCount(), "files read for: " +
description);
+ }
+ assertEquals(expected, sorted(query.collectAsList()), "rows of: " +
description);
+ }
+
+ private static void assertDoesNotUseIndex(String sql) {
+ assertDoesNotUseIndex(sql, () -> spark.sql(sql));
+ }
+
+ private static void assertDoesNotUseIndex(Supplier<Dataset<Row>> newQuery) {
+
assertDoesNotUseIndex(newQuery.get().queryExecution().analyzed().toString(),
newQuery);
+ }
+
+ private static void assertDoesNotUseIndex(String description,
Supplier<Dataset<Row>> newQuery) {
+ Dataset<Row> query = newQuery.get();
+ assertTrue(
+ stagedSetIds(query.queryExecution().optimizedPlan()).isEmpty(),
+ "expected a full scan for: " + description);
+ assertEquals(withoutIndex(newQuery), sorted(query.collectAsList()), "rows
of: " + description);
+ }
+
+ private static void assertReturnsNothingWithoutScan(String sql) {
+ Dataset<Row> query = spark.sql(sql);
+ List<LogicalPlan> leaves = leaves(query.queryExecution().optimizedPlan());
+ assertTrue(
+ leaves.stream().allMatch(leaf -> leaf instanceof LocalRelation),
+ "expected no scan for: " + sql);
+ assertTrue(query.collectAsList().isEmpty());
+ }
+
+ /** The rows of the query without the index, sorted. */
+ private static List<Row> withoutIndex(Supplier<Dataset<Row>> newQuery) {
+ List<Row> rows = new ArrayList<>();
+ withConf(
+ XTableIndexConf.PruningEnabled(),
+ "false",
+ () -> rows.addAll(sorted(newQuery.get().collectAsList())));
+ return rows;
+ }
+
+ private static long dataFileCount() {
+ return spark.sql("SELECT * FROM " + TABLE + ".files").count();
+ }
+
+ private static List<String> stagedSetIds(LogicalPlan plan) {
+ return leaves(plan).stream()
+ .filter(leaf -> leaf instanceof DataSourceV2ScanRelation)
+ .map(
+ leaf ->
+ ((DataSourceV2ScanRelation) leaf)
+ .relation()
+ .options()
+ .get(StagedScans.ScanTaskSetIdOption()))
+ .filter(setId -> setId != null)
+ .collect(Collectors.toList());
+ }
+
+ private static List<LogicalPlan> leaves(LogicalPlan plan) {
+ return JavaConverters.seqAsJavaList(plan.collectLeaves());
+ }
+
+ private static List<Row> sorted(List<Row> rows) {
+ List<Row> sorted = new ArrayList<>(rows);
+ sorted.sort((left, right) -> left.toString().compareTo(right.toString()));
+ return sorted;
+ }
+
+ private static List<String> emails(long... ids) {
+ return emails(LongStream.of(ids));
+ }
+
+ private static List<String> emails(LongStream ids) {
+ return ids.mapToObj(
+ id -> spark.sql("SELECT md5('" + id +
"')").collectAsList().get(0).getString(0))
+ .collect(Collectors.toList());
+ }
+
+ private static String inList(List<String> values) {
+ return values.stream()
+ .map(value -> "'" + value + "'")
+ .collect(Collectors.joining(", ", "(", ")"));
+ }
+
+ private static Table icebergTable() {
+ try {
+ return Spark3Util.loadIcebergTable(spark, TABLE);
+ } catch (Exception e) {
+ throw new IllegalStateException(e);
+ }
+ }
+
+ private static void withConf(String key, String value, Runnable action) {
+ String previous = spark.conf().getOption(key).getOrElse(() -> null);
+ spark.conf().set(key, value);
+ try {
+ action.run();
+ } finally {
+ if (previous == null) {
+ spark.conf().unset(key);
+ } else {
+ spark.conf().set(key, previous);
+ }
+ }
+ }
+}
diff --git
a/xtable-spark-extensions/src/test/java/org/apache/xtable/spark/extensions/TestXTableIndex.java
b/xtable-spark-extensions/src/test/java/org/apache/xtable/spark/extensions/TestXTableIndex.java
new file mode 100644
index 00000000..c99ab115
--- /dev/null
+++
b/xtable-spark-extensions/src/test/java/org/apache/xtable/spark/extensions/TestXTableIndex.java
@@ -0,0 +1,117 @@
+/*
+ * 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;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.spark.sql.connector.catalog.Identifier;
+import org.junit.jupiter.api.Test;
+
+import scala.Option;
+import scala.collection.JavaConverters;
+
+import org.apache.xtable.conversion.SourceTable;
+import org.apache.xtable.iceberg.IcebergCatalogConfig;
+
+public class TestXTableIndex {
+
+ @Test
+ void loadsTableFromTheSparkCatalogOptions() {
+ Map<String, String> sparkConf = new HashMap<>();
+ sparkConf.put("spark.sql.catalog.cat",
"org.apache.iceberg.spark.SparkCatalog");
+ sparkConf.put("spark.sql.catalog.cat.type", "hadoop");
+ sparkConf.put("spark.sql.catalog.cat.warehouse", "file:///warehouse");
+ sparkConf.put("spark.sql.catalog.other.type", "rest");
+
+ SourceTable sourceTable =
+ XTableIndex.toSourceTable(
+ "cat.db.orders",
+ "file:///warehouse/db/orders",
+ "file:///warehouse/db/orders/data",
+ Option.apply("cat"),
+ Option.apply(Identifier.of(new String[] {"db"}, "orders")),
+ toScala(sparkConf));
+
+ assertEquals("orders", sourceTable.getName());
+ assertEquals(Arrays.asList("db"),
Arrays.asList(sourceTable.getNamespace()));
+ assertEquals("file:///warehouse/db/orders", sourceTable.getBasePath());
+ assertEquals("file:///warehouse/db/orders/data",
sourceTable.getDataPath());
+ IcebergCatalogConfig catalogConfig = (IcebergCatalogConfig)
sourceTable.getCatalogConfig();
+ assertEquals("cat", catalogConfig.getCatalogName());
+ assertEquals("org.apache.iceberg.hadoop.HadoopCatalog",
catalogConfig.getCatalogImpl());
+ Map<String, String> expectedOptions = new HashMap<>();
+ expectedOptions.put("type", "hadoop");
+ expectedOptions.put("warehouse", "file:///warehouse");
+ assertEquals(expectedOptions, catalogConfig.getCatalogOptions());
+ }
+
+ @Test
+ void resolvesTheCatalogImplementationLikeIceberg() {
+ assertEquals("org.apache.iceberg.hive.HiveCatalog", catalogImpl(new
HashMap<>()));
+ Map<String, String> glue = new HashMap<>();
+ glue.put("spark.sql.catalog.cat.type", "glue");
+ assertEquals("org.apache.iceberg.aws.glue.GlueCatalog", catalogImpl(glue));
+ Map<String, String> custom = new HashMap<>();
+ custom.put("spark.sql.catalog.cat.type", "rest");
+ custom.put("spark.sql.catalog.cat.catalog-impl",
"com.example.CustomCatalog");
+ assertEquals("com.example.CustomCatalog", catalogImpl(custom));
+ Map<String, String> unknown = new HashMap<>();
+ unknown.put("spark.sql.catalog.cat.type", "unknown");
+ assertThrows(IllegalArgumentException.class, () -> catalogImpl(unknown));
+ }
+
+ @Test
+ void loadsTableReadByPathFromItsLocation() {
+ SourceTable sourceTable =
+ XTableIndex.toSourceTable(
+ "file:///tables/orders",
+ "file:///tables/orders",
+ "file:///tables/orders/data",
+ Option.empty(),
+ Option.empty(),
+ toScala(new HashMap<>()));
+ assertEquals("file:///tables/orders", sourceTable.getBasePath());
+ assertEquals("file:///tables/orders/data", sourceTable.getDataPath());
+ assertNull(sourceTable.getCatalogConfig());
+ }
+
+ private static String catalogImpl(Map<String, String> sparkConf) {
+ SourceTable sourceTable =
+ XTableIndex.toSourceTable(
+ "cat.db.orders",
+ "file:///warehouse/db/orders",
+ "file:///warehouse/db/orders/data",
+ Option.apply("cat"),
+ Option.apply(Identifier.of(new String[] {"db"}, "orders")),
+ toScala(sparkConf));
+ return ((IcebergCatalogConfig)
sourceTable.getCatalogConfig()).getCatalogImpl();
+ }
+
+ private static scala.collection.immutable.Map<String, String>
toScala(Map<String, String> map) {
+ return scala.collection.immutable.Map$.MODULE$
+ .<String, String>empty()
+ .$plus$plus(JavaConverters.mapAsScalaMap(map));
+ }
+}
diff --git a/xtable-spark-runtime/pom.xml b/xtable-spark-runtime/pom.xml
index 34a3de70..94e057a6 100644
--- a/xtable-spark-runtime/pom.xml
+++ b/xtable-spark-runtime/pom.xml
@@ -28,7 +28,12 @@
<artifactId>xtable-spark-runtime_${scala.binary.version}</artifactId>
<name>XTable Project Spark Runtime</name>
- <description>Thin, relocated Spark bundle that runs XTable metadata
sync.</description>
+ <description>Thin, relocated Spark bundle that runs XTable metadata sync
and the XTable Spark SQL extensions.</description>
+
+ <properties>
+ <!-- Spark version of the Hudi bundle that ITXTableSparkRuntimeBundle
passes to spark-submit -->
+
<engine.spark.version.prefix>${spark.version.prefix}</engine.spark.version.prefix>
+ </properties>
<dependencies>
<!-- Bundled -->
@@ -38,6 +43,12 @@
<version>${project.version}</version>
</dependency>
+ <dependency>
+ <groupId>org.apache.xtable</groupId>
+
<artifactId>xtable-spark-extensions_${scala.binary.version}</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+
<dependency>
<groupId>commons-cli</groupId>
<artifactId>commons-cli</artifactId>
@@ -221,9 +232,31 @@
<configuration>
<includeScope>provided</includeScope>
<excludeGroupIds>org.apache.spark,org.apache.hadoop,org.apache.xtable,org.projectlombok,org.slf4j,org.apache.logging.log4j,org.scala-lang,com.fasterxml.jackson.core,com.fasterxml.jackson.datatype,com.fasterxml.jackson.module</excludeGroupIds>
+ <!-- The Hudi bundle must match the Spark of
spark-submit; see engine-hudi-bundle -->
+
<excludeArtifactIds>hudi-spark${spark.version.prefix}-bundle_${scala.binary.version}</excludeArtifactIds>
<outputFile>${project.build.directory}/engine-classpath.txt</outputFile>
</configuration>
</execution>
+ <execution>
+ <id>engine-hudi-bundle</id>
+ <phase>pre-integration-test</phase>
+ <goals>
+ <goal>copy</goal>
+ </goals>
+ <configuration>
+ <artifactItems>
+ <artifactItem>
+ <groupId>org.apache.hudi</groupId>
+
<artifactId>hudi-spark${engine.spark.version.prefix}-bundle_${scala.binary.version}</artifactId>
+ <version>${hudi.version}</version>
+ <!-- one name, so a build for another
Spark version replaces the jar -->
+
<destFileName>hudi-spark-bundle.jar</destFileName>
+ </artifactItem>
+ </artifactItems>
+
<outputDirectory>${project.build.directory}/engine-jars</outputDirectory>
+ <overWriteReleases>true</overWriteReleases>
+ </configuration>
+ </execution>
</executions>
</plugin>
<!--
@@ -249,6 +282,7 @@
<include>org.apache.xtable:xtable-api</include>
<include>org.apache.xtable:xtable-core_${scala.binary.version}</include>
<include>org.apache.xtable:xtable-hudi-support-utils</include>
+
<include>org.apache.xtable:xtable-spark-extensions_${scala.binary.version}</include>
<include>com.google.guava:guava</include>
<include>com.google.guava:failureaccess</include>
<include>com.google.guava:listenablefuture</include>
diff --git
a/xtable-spark-runtime/src/main/java/org/apache/xtable/spark/TableSyncSpec.java
b/xtable-spark-runtime/src/main/java/org/apache/xtable/spark/TableSyncSpec.java
index 4aeb4daa..fc9e6c61 100644
---
a/xtable-spark-runtime/src/main/java/org/apache/xtable/spark/TableSyncSpec.java
+++
b/xtable-spark-runtime/src/main/java/org/apache/xtable/spark/TableSyncSpec.java
@@ -19,6 +19,7 @@
package org.apache.xtable.spark;
import java.util.List;
+import java.util.Map;
import lombok.Builder;
import lombok.NonNull;
@@ -55,4 +56,10 @@ public class TableSyncSpec {
/** Use the Spark-free Delta Kernel implementation for the Delta source
and/or target. */
boolean useDeltaKernel;
+
+ /**
+ * Optional properties of the target tables, keyed by the target format, for
example {@code
+ * xtable.hudi.target.secondary.index.column} for a {@code HUDI} target.
+ */
+ Map<String, Map<String, String>> targetProperties;
}
diff --git
a/xtable-spark-runtime/src/main/java/org/apache/xtable/spark/XTableSparkSync.java
b/xtable-spark-runtime/src/main/java/org/apache/xtable/spark/XTableSparkSync.java
index cf2f7877..ca34792e 100644
---
a/xtable-spark-runtime/src/main/java/org/apache/xtable/spark/XTableSparkSync.java
+++
b/xtable-spark-runtime/src/main/java/org/apache/xtable/spark/XTableSparkSync.java
@@ -22,6 +22,7 @@ import java.io.IOException;
import java.io.InputStream;
import java.util.ArrayList;
import java.util.Arrays;
+import java.util.HashMap;
import java.util.List;
import java.util.Locale;
import java.util.Map;
@@ -226,6 +227,7 @@ public final class XTableSparkSync {
.dataPath(table.tableDataPath)
.namespace(table.namespace == null ? null :
table.namespace.split("\\."))
.partitionSpec(table.partitionSpec)
+ .targetProperties(table.targetProperties)
.sourceFormat(sourceFormat)
.targets(targetFormats));
}
@@ -392,6 +394,14 @@ public final class XTableSparkSync {
throw new IllegalArgumentException("--datasetconfig 'datasets' entries
must be mappings");
}
requireField(table.tableBasePath, "tableBasePath");
+ if (table.targetProperties != null) {
+ // keyed like targetFormats, which are upper case
+ Map<String, Map<String, String>> targetProperties = new HashMap<>();
+ table.targetProperties.forEach(
+ (format, properties) ->
+ targetProperties.put(format.trim().toUpperCase(Locale.ROOT),
properties));
+ table.targetProperties = targetProperties;
+ }
}
return config;
}
@@ -421,6 +431,12 @@ public final class XTableSparkSync {
String tableName;
String partitionSpec;
String namespace;
+
+ /**
+ * Properties of the target tables, keyed by the target format, for
example {@code
+ * xtable.hudi.target.secondary.index.column} for a {@code HUDI} target.
+ */
+ Map<String, Map<String, String>> targetProperties;
}
}
diff --git
a/xtable-spark-runtime/src/main/java/org/apache/xtable/spark/XTableSyncService.java
b/xtable-spark-runtime/src/main/java/org/apache/xtable/spark/XTableSyncService.java
index e875dab7..ccc20b73 100644
---
a/xtable-spark-runtime/src/main/java/org/apache/xtable/spark/XTableSyncService.java
+++
b/xtable-spark-runtime/src/main/java/org/apache/xtable/spark/XTableSyncService.java
@@ -18,6 +18,7 @@
package org.apache.xtable.spark;
+import java.util.Collections;
import java.util.List;
import java.util.Locale;
import java.util.Map;
@@ -79,31 +80,7 @@ public class XTableSyncService {
.additionalProperties(sourceProperties)
.build();
- // Read by ConversionTargetFactory to pick the Delta Kernel target.
- List<TargetTable> targetTables =
- spec.getTargets().stream()
- .map(
- targetFormat -> {
- Properties targetProperties = new Properties();
- if (spec.isUseDeltaKernel()) {
- targetProperties.setProperty(
- DeltaConversionTargetConfig.USE_KERNEL,
Boolean.TRUE.toString());
- }
- // Hudi resolves partitions relative to its base path, so a
Hudi target sits
- // at the data path. Other targets sit at the table root.
- String targetPath =
- TableFormat.HUDI.equalsIgnoreCase(targetFormat)
- ? dataPath
- : spec.getBasePath();
- return TargetTable.builder()
- .name(spec.getKey())
- .basePath(targetPath)
- .namespace(spec.getNamespace())
- .formatName(targetFormat)
- .additionalProperties(targetProperties)
- .build();
- })
- .collect(Collectors.toList());
+ List<TargetTable> targetTables = targetTables(spec, dataPath);
ConversionConfig conversionConfig =
ConversionConfig.builder()
@@ -125,6 +102,39 @@ public class XTableSyncService {
return new ConversionController(hadoopConf).sync(conversionConfig,
sourceProvider);
}
+ /**
+ * The target tables of a spec. Each target gets the properties configured
for its format, and the
+ * Delta Kernel flag, which ConversionTargetFactory reads to pick the Delta
Kernel target.
+ */
+ // package-private for unit testing
+ static List<TargetTable> targetTables(TableSyncSpec spec, String dataPath) {
+ return spec.getTargets().stream()
+ .map(
+ targetFormat -> {
+ Properties targetProperties = new Properties();
+ if (spec.getTargetProperties() != null) {
+ targetProperties.putAll(
+ spec.getTargetProperties().getOrDefault(targetFormat,
Collections.emptyMap()));
+ }
+ if (spec.isUseDeltaKernel()) {
+ targetProperties.setProperty(
+ DeltaConversionTargetConfig.USE_KERNEL,
Boolean.TRUE.toString());
+ }
+ // Hudi resolves partitions relative to its base path, so a Hudi
target sits
+ // at the data path. Other targets sit at the table root.
+ String targetPath =
+ TableFormat.HUDI.equalsIgnoreCase(targetFormat) ? dataPath :
spec.getBasePath();
+ return TargetTable.builder()
+ .name(spec.getKey())
+ .basePath(targetPath)
+ .namespace(spec.getNamespace())
+ .formatName(targetFormat)
+ .additionalProperties(targetProperties)
+ .build();
+ })
+ .collect(Collectors.toList());
+ }
+
private static ConversionSourceProvider<?> sourceProviderFor(
String sourceFormat, boolean useDeltaKernel) {
switch (sourceFormat.toUpperCase(Locale.ROOT)) {
diff --git
a/xtable-spark-runtime/src/test/java/org/apache/xtable/spark/ITXTableSparkRuntimeBundle.java
b/xtable-spark-runtime/src/test/java/org/apache/xtable/spark/ITXTableSparkRuntimeBundle.java
index 061d6e35..6caac47a 100644
---
a/xtable-spark-runtime/src/test/java/org/apache/xtable/spark/ITXTableSparkRuntimeBundle.java
+++
b/xtable-spark-runtime/src/test/java/org/apache/xtable/spark/ITXTableSparkRuntimeBundle.java
@@ -38,11 +38,13 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.TimeUnit;
+import java.util.jar.JarFile;
import java.util.stream.Collectors;
import java.util.stream.Stream;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Encoders;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.types.ArrayType;
@@ -62,17 +64,25 @@ import org.apache.hudi.client.HoodieReadClient;
import org.apache.iceberg.PartitionSpec;
import org.apache.iceberg.Schema;
+import org.apache.iceberg.Table;
import org.apache.iceberg.TableProperties;
import org.apache.iceberg.hadoop.HadoopTables;
import org.apache.iceberg.spark.SparkSchemaUtil;
import org.apache.xtable.GenericTable;
+import org.apache.xtable.catalog.TableFormatUtils;
+import org.apache.xtable.conversion.SourceTable;
+import org.apache.xtable.hudi.HudiTargetConfig;
import org.apache.xtable.hudi.HudiTestUtil;
+import org.apache.xtable.hudi.index.HudiBackedSecondaryIndexReader;
+import org.apache.xtable.iceberg.IcebergConversionSourceProvider;
+import org.apache.xtable.model.storage.TableFormat;
/**
* Runs a sync with the shaded bundle through {@code
$SPARK_HOME/bin/spark-submit}, one case per
* direction, and checks the target data equals the source. The engines come
from {@code
- * target/engine-classpath.txt}. Skipped when {@code SPARK_HOME} is not set.
+ * target/engine-classpath.txt} and the Hudi bundle from {@code
target/engine-jars}. Skipped when
+ * {@code SPARK_HOME} is not set.
*/
class ITXTableSparkRuntimeBundle {
@@ -152,6 +162,95 @@ class ITXTableSparkRuntimeBundle {
sourceFormat, targetFormat, basePath, dataPath, columns,
orderByColumn, output);
}
+ /**
+ * Users name the Spark SQL extensions in {@code spark.sql.extensions}, so
the bundle must ship
+ * them under their own package, not relocated.
+ */
+ @Test
+ void bundleShipsSparkSqlExtensions() throws Exception {
+ try (JarFile bundle = new JarFile(findBundleJar())) {
+ assertTrue(
+
bundle.getEntry("org/apache/xtable/spark/extensions/XTableSparkSessionExtensions.class")
+ != null,
+ "XTableSparkSessionExtensions is missing from the bundle");
+ }
+ }
+
+ /**
+ * A sync job keeps the secondary index current: the dataset config passes
the index settings to
+ * the Hudi target, and the index is read back at the data path of the
Iceberg table.
+ */
+ @Test
+ void datasetConfigTargetPropertiesBuildTheSecondaryIndex() throws Exception {
+ String sparkHome = System.getenv("SPARK_HOME");
+ assumeTrue(sparkHome != null && !sparkHome.trim().isEmpty(), "SPARK_HOME
is not set");
+
+ String tableName = "iceberg_indexed_by_sync";
+ String basePath = tempDir.resolve(tableName).toUri().toString();
+ Dataset<Row> rows =
+ sparkSession
+ .range(100)
+ .selectExpr("id", "concat('user', cast(id % 40 AS string),
'@example.com') AS email");
+ Schema schema = SparkSchemaUtil.convert(rows.schema());
+ Table table =
+ new HadoopTables(jsc.hadoopConfiguration())
+ .create(schema, PartitionSpec.unpartitioned(), basePath);
+ rows.write().format("iceberg").mode("append").save(basePath);
+ table.refresh();
+ String dataPath =
+ TableFormatUtils.getTableDataLocation(
+ TableFormat.ICEBERG, table.location(), table.properties());
+
+ Path datasetConfig = tempDir.resolve(tableName + ".yaml");
+ Files.write(
+ datasetConfig,
+ Arrays.asList(
+ "sourceFormat: ICEBERG",
+ "targetFormats:",
+ " - HUDI",
+ "datasets:",
+ " - tableBasePath: " + basePath,
+ " tableDataPath: " + dataPath,
+ " tableName: " + tableName,
+ " targetProperties:",
+ " HUDI:",
+ " " + HudiTargetConfig.HUDI_TABLE_VERSION + ": '9'",
+ " "
+ + HudiTargetConfig.EXECUTION_ENGINE
+ + ": "
+ + HudiTargetConfig.ExecutionEngine.SPARK.getConfigValue(),
+ " " + HudiTargetConfig.SECONDARY_INDEX_COLUMN + ": email"),
+ StandardCharsets.UTF_8);
+ String output =
+ runBundleSync(
+ sparkHome,
+ Collections.singletonList(
+ "spark.serializer=org.apache.spark.serializer.KryoSerializer"),
+ Arrays.asList("--datasetconfig",
datasetConfig.toUri().toString()));
+
+ SourceTable sourceTable =
+ SourceTable.builder()
+ .name(tableName)
+ .basePath(basePath)
+ .dataPath(dataPath)
+ .formatName(TableFormat.ICEBERG)
+ .build();
+ HudiBackedSecondaryIndexReader reader =
+ new HudiBackedSecondaryIndexReader(
+ sourceTable, new IcebergConversionSourceProvider(), sparkSession);
+ assertTrue(reader.doesIndexExist("email"), "index not built. spark-submit
output:\n" + output);
+ assertEquals(
+ String.valueOf(table.currentSnapshot().snapshotId()),
+ reader.getLastSyncedSourceIdentifier().orElse(null));
+ Dataset<Row> keys =
+ sparkSession
+ .createDataset(
+ Arrays.asList("[email protected]", "[email protected]"),
Encoders.STRING())
+ .toDF("email");
+ // ids 7, 47 and 87 hold [email protected]
+ assertEquals(3, reader.lookup(sourceTable, keys, "email").count());
+ }
+
/**
* Iceberg keeps its metadata at the base path but writes data files to a
separate {@code
* write.data.path}. The Hudi target goes to the data path, so {@code
.hoodie} lands next to the
@@ -219,6 +318,11 @@ class ITXTableSparkRuntimeBundle {
/** Runs {@code XTableSparkSync} from the bundle through spark-submit and
returns its output. */
private String runBundleSync(String sparkHome, List<String> syncArgs) throws
Exception {
+ return runBundleSync(sparkHome, Collections.emptyList(), syncArgs);
+ }
+
+ private String runBundleSync(String sparkHome, List<String> sparkConfs,
List<String> syncArgs)
+ throws Exception {
File bundleJar = findBundleJar();
File sparkSubmit = new File(sparkHome, "bin/spark-submit");
assertTrue(sparkSubmit.canExecute(), "spark-submit not executable at " +
sparkSubmit);
@@ -234,6 +338,10 @@ class ITXTableSparkRuntimeBundle {
command.add("spark.driver.extraClassPath=" + engineClasspath);
command.add("--conf");
command.add("spark.executor.extraClassPath=" + engineClasspath);
+ for (String sparkConf : sparkConfs) {
+ command.add("--conf");
+ command.add(sparkConf);
+ }
command.add("--class");
command.add("org.apache.xtable.spark.XTableSparkSync");
command.add(bundleJar.getAbsolutePath());
@@ -347,7 +455,13 @@ class ITXTableSparkRuntimeBundle {
+ " (dependency:build-classpath)");
String cp = new String(Files.readAllBytes(cpFile.toPath()),
StandardCharsets.UTF_8).trim();
assertTrue(!cp.isEmpty(), "engine classpath file is empty");
- return cp;
+ // The Hudi bundle is built for the Spark of spark-submit
(-Dengine.spark.version.prefix), not
+ // for the Spark the module compiles against.
+ File hudiBundle = new File("target/engine-jars/hudi-spark-bundle.jar");
+ assertTrue(
+ hudiBundle.isFile(),
+ "Hudi Spark bundle not copied to " + hudiBundle.getAbsolutePath() + "
(dependency:copy)");
+ return cp + File.pathSeparator + hudiBundle.getAbsolutePath();
}
// The shaded jar is the main artifact; skip original-* and the sources,
javadoc and tests jars.
diff --git
a/xtable-spark-runtime/src/test/java/org/apache/xtable/spark/XTableSparkSyncTest.java
b/xtable-spark-runtime/src/test/java/org/apache/xtable/spark/XTableSparkSyncTest.java
index 733c27e9..d872cabb 100644
---
a/xtable-spark-runtime/src/test/java/org/apache/xtable/spark/XTableSparkSyncTest.java
+++
b/xtable-spark-runtime/src/test/java/org/apache/xtable/spark/XTableSparkSyncTest.java
@@ -29,6 +29,7 @@ import java.io.InputStream;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Arrays;
+import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -38,6 +39,8 @@ import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.CsvSource;
import org.junit.jupiter.params.provider.ValueSource;
+import org.apache.xtable.conversion.TargetTable;
+import org.apache.xtable.delta.DeltaConversionTargetConfig;
import org.apache.xtable.model.sync.ErrorDetails;
import org.apache.xtable.model.sync.SyncResult;
import org.apache.xtable.model.sync.SyncStatusCode;
@@ -206,6 +209,61 @@ class XTableSparkSyncTest {
assertNull(config.datasets.get(1).tableName);
}
+ @Test
+ void parseDatasetConfigReadsTargetPropertiesKeyedByUpperCaseFormat() {
+ XTableSparkSync.DatasetConfig config =
+ XTableSparkSync.parseDatasetConfig(
+ yaml(
+ "sourceFormat: ICEBERG\n"
+ + "targetFormats:\n"
+ + " - HUDI\n"
+ + "datasets:\n"
+ + " - tableBasePath: /data/customer\n"
+ + " targetProperties:\n"
+ + " hudi:\n"
+ + " xtable.hudi.target.secondary.index.column:
c_email_address\n"));
+
+ assertEquals(
+ Collections.singletonMap(
+ "HUDI",
+ Collections.singletonMap(
+ "xtable.hudi.target.secondary.index.column",
"c_email_address")),
+ config.datasets.get(0).targetProperties);
+ }
+
+ @Test
+ void targetTablesGetThePropertiesOfTheirFormat() {
+ Map<String, Map<String, String>> targetProperties = new HashMap<>();
+ targetProperties.put(
+ "HUDI",
+ Collections.singletonMap("xtable.hudi.target.secondary.index.column",
"c_email_address"));
+ TableSyncSpec spec =
+ TableSyncSpec.builder()
+ .key("customer")
+ .basePath("/warehouse/customer")
+ .sourceFormat("ICEBERG")
+ .targets(Arrays.asList("HUDI", "DELTA"))
+ .targetProperties(targetProperties)
+ .useDeltaKernel(true)
+ .build();
+
+ List<TargetTable> targetTables =
+ XTableSyncService.targetTables(spec, "/warehouse/customer/data");
+
+ TargetTable hudi = targetTables.get(0);
+ assertEquals("file:///warehouse/customer/data", hudi.getBasePath());
+ assertEquals(
+ "c_email_address",
+
hudi.getAdditionalProperties().getProperty("xtable.hudi.target.secondary.index.column"));
+ TargetTable delta = targetTables.get(1);
+ assertEquals("file:///warehouse/customer", delta.getBasePath());
+ assertNull(
+
delta.getAdditionalProperties().getProperty("xtable.hudi.target.secondary.index.column"));
+ assertEquals(
+ "true",
+
delta.getAdditionalProperties().getProperty(DeltaConversionTargetConfig.USE_KERNEL));
+ }
+
@Test
void parseDatasetConfigRejectsMissingSourceFormat() {
assertThrows(
