This is an automated email from the ASF dual-hosted git repository.
diqiu50 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/gravitino.git
The following commit(s) were added to refs/heads/main by this push:
new 7ff63ddbfd [#13614] improvement(spark-connector): Upgrade Kyuubi Spark
Hive connector to 1.12.0 to support dynamic partition pruning (#13615)
7ff63ddbfd is described below
commit 7ff63ddbfda3bfb0f35bd71568f0fdf1ecee73be
Author: Yuhui <[email protected]>
AuthorDate: Fri Oct 9 17:44:45 2026 +0800
[#13614] improvement(spark-connector): Upgrade Kyuubi Spark Hive connector
to 1.12.0 to support dynamic partition pruning (#13615)
### What changes were proposed in this pull request?
Upgrade Kyuubi Spark Hive connector from 1.11.0 to 1.12.0, add an
integration test for dynamic partition pruning, and document the
supported table formats.
### Why are the changes needed?
Kyuubi 1.12.0 supports DPP in HiveScan, so partitioned Hive tables can
be pruned at runtime.
Fix: #13614
### Does this PR introduce _any_ user-facing change?
DPP applies to Hive tables read by the Hive SerDe reader. Parquet and
ORC tables need
`spark.sql.kyuubi.hive.connector.read.convertMetastoreParquet/Orc=false`
to use DPP.
### How was this patch tested?
- New IT `SparkHiveCatalogIT#testDynamicPartitionPruning` verifies only
the matching partition is scanned, on Spark 3.5 and 4.0.
- Existing Spark 3.5/4.0 Hive ITs and unit tests pass.
---
docs/spark-connector/spark-catalog-hive.md | 18 ++++
gradle/libs.versions.toml | 2 +-
.../integration/test/hive/SparkHiveCatalogIT.java | 98 ++++++++++++++++++++++
3 files changed, 117 insertions(+), 1 deletion(-)
diff --git a/docs/spark-connector/spark-catalog-hive.md
b/docs/spark-connector/spark-catalog-hive.md
index 3a43421df6..59d7d0d21a 100644
--- a/docs/spark-connector/spark-catalog-hive.md
+++ b/docs/spark-connector/spark-catalog-hive.md
@@ -71,6 +71,24 @@ The current implementation materializes all view results on
the Spark driver usi
SELECT * FROM employee_view;
```
+## Dynamic Partition Pruning
+
+Dynamic partition pruning (DPP) is supported for partitioned Hive tables that
are read through the Hive SerDe reader.
+
+| Table storage format | Supports DPP
|
+|-------------------------------------------|--------------------------------------------------------------------------------------------|
+| Text, Avro, CSV, JSON, SequenceFile, etc. | Yes
|
+| Parquet | Only when
`spark.sql.kyuubi.hive.connector.read.convertMetastoreParquet` is set to
`false` |
+| ORC | Only when
`spark.sql.kyuubi.hive.connector.read.convertMetastoreOrc` is set to `false`
|
+
+Both `convertMetastore*` options default to `true`, which reads Parquet and
ORC tables with Spark's native readers. The native readers are faster but don't
support DPP. Setting an option to `false` enables DPP for that format, but
reads the data with the Hive SerDe reader instead.
+
+```sql
+SET spark.sql.kyuubi.hive.connector.read.convertMetastoreParquet=false;
+```
+
+DPP only prunes partitions, and the query must also meet Spark's own
conditions for dynamic partition pruning, see
`spark.sql.optimizer.dynamicPartitionPruning.enabled` in [Spark
configuration](https://spark.apache.org/docs/latest/configuration.html#runtime-sql-configuration).
+
## Catalog Properties
Gravitino spark connector will transform below property names which are
defined in catalog properties to Spark Hive connector configuration.
diff --git a/gradle/libs.versions.toml b/gradle/libs.versions.toml
index 179e9a3953..c491cc5134 100644
--- a/gradle/libs.versions.toml
+++ b/gradle/libs.versions.toml
@@ -79,7 +79,7 @@ spark33 = "3.3.4"
spark34 = "3.4.3"
spark35 = "3.5.3"
spark40 = "4.0.3"
-kyuubi4spark = "1.11.0"
+kyuubi4spark = "1.12.0"
kyuubi4authz = "1.10.2"
trino = '435'
# Trino version the iceberg-rest-trino-it module's in-process query runner is
tested against.
diff --git
a/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/hive/SparkHiveCatalogIT.java
b/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/hive/SparkHiveCatalogIT.java
index 1d70bdd708..bca934c64e 100644
---
a/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/hive/SparkHiveCatalogIT.java
+++
b/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/hive/SparkHiveCatalogIT.java
@@ -19,11 +19,14 @@
package org.apache.gravitino.spark.connector.integration.test.hive;
import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.ImmutableSet;
import com.google.common.collect.Maps;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Comparator;
+import java.util.HashSet;
import java.util.List;
+import java.util.Locale;
import java.util.Map;
import java.util.Set;
import org.apache.gravitino.NameIdentifier;
@@ -39,14 +42,24 @@ import
org.apache.gravitino.spark.connector.integration.test.util.SparkTableInfo
import
org.apache.gravitino.spark.connector.integration.test.util.SparkTableInfoChecker;
import org.apache.hadoop.fs.Path;
import org.apache.spark.sql.AnalysisException;
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Row;
import org.apache.spark.sql.catalyst.analysis.PartitionsAlreadyExistException;
import org.apache.spark.sql.connector.catalog.TableCatalog;
+import org.apache.spark.sql.connector.read.InputPartition;
+import org.apache.spark.sql.execution.SparkPlan;
+import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanExec;
+import org.apache.spark.sql.execution.adaptive.QueryStageExec;
+import org.apache.spark.sql.execution.datasources.FilePartition;
+import org.apache.spark.sql.execution.datasources.PartitionedFile;
+import org.apache.spark.sql.execution.datasources.v2.BatchScanExec;
import org.apache.spark.sql.types.DataTypes;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Tag;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.ValueSource;
+import scala.collection.JavaConverters;
@Tag("gravitino-docker-test")
public abstract class SparkHiveCatalogIT extends SparkCommonIT {
@@ -693,4 +706,89 @@ public abstract class SparkHiveCatalogIT extends
SparkCommonIT {
dropTableIfExists(tableName);
}
}
+
+ @ParameterizedTest
+ @ValueSource(strings = {"TEXTFILE", "PARQUET"})
+ void testDynamicPartitionPruning(String format) {
+ String factTable = "dpp_fact_" + format.toLowerCase(Locale.ROOT);
+ String dimTable = "dpp_dim_" + format.toLowerCase(Locale.ROOT);
+ // Parquet tables are read by Spark's native ParquetScan unless the
conversion is disabled,
+ // and only the Kyuubi HiveScan supports runtime filtering.
+ String convertParquetConf =
"spark.sql.kyuubi.hive.connector.read.convertMetastoreParquet";
+ getSparkSession().conf().set(convertParquetConf, "false");
+
+ dropTableIfExists(factTable);
+ dropTableIfExists(dimTable);
+ try {
+ sql(
+ String.format(
+ "CREATE TABLE %s (id INT, amount INT) PARTITIONED BY (dt STRING)
STORED AS %s",
+ factTable, format));
+ sql(String.format("CREATE TABLE %s (dt STRING, flag STRING) STORED AS
%s", dimTable, format));
+ sql(
+ String.format(
+ "INSERT INTO %s VALUES (1, 10, '2026-01-01'), (2, 20,
'2026-01-02'), "
+ + "(3, 30, '2026-01-03')",
+ factTable));
+ sql(
+ String.format(
+ "INSERT INTO %s VALUES ('2026-01-01', 'n'), ('2026-01-02', 'y'),
"
+ + "('2026-01-03', 'n')",
+ dimTable));
+
+ Dataset<Row> df =
+ getSparkSession()
+ .sql(
+ String.format(
+ "SELECT f.id, f.amount FROM %s f JOIN %s d ON f.dt =
d.dt WHERE d.flag = 'y'",
+ factTable, dimTable));
+ List<Row> rows = df.collectAsList();
+ Assertions.assertEquals(1, rows.size());
+ Assertions.assertEquals(2, rows.get(0).getInt(0));
+ Assertions.assertEquals(20, rows.get(0).getInt(1));
+
+ String plan = df.queryExecution().executedPlan().toString();
+ Assertions.assertTrue(
+ plan.toLowerCase(Locale.ROOT).contains("dynamicpruningexpression"),
+ "Expected dynamic partition pruning in plan:\n" + plan);
+
+ // Spark pushes the runtime filters into the scan before planning input
partitions, so
+ // re-planning the executed scan returns only the files of the pruned
partitions.
+ BatchScanExec factScan =
findBatchScan(df.queryExecution().executedPlan(), factTable);
+ Assertions.assertNotNull(factScan, "No BatchScanExec for " + factTable +
" in:\n" + plan);
+ Set<String> scannedPartitions = new HashSet<>();
+ for (InputPartition partition :
factScan.scan().toBatch().planInputPartitions()) {
+ for (PartitionedFile file : ((FilePartition) partition).files()) {
+
scannedPartitions.add(file.partitionValues().getUTF8String(0).toString());
+ }
+ }
+ Assertions.assertEquals(ImmutableSet.of("2026-01-02"),
scannedPartitions);
+ } finally {
+ getSparkSession().conf().unset(convertParquetConf);
+ dropTableIfExists(factTable);
+ dropTableIfExists(dimTable);
+ }
+ }
+
+ // Walks through adaptive wrappers and query stages, which are leaf nodes in
the final plan.
+ @SuppressWarnings("deprecation")
+ private BatchScanExec findBatchScan(SparkPlan plan, String tableName) {
+ if (plan instanceof AdaptiveSparkPlanExec) {
+ return findBatchScan(((AdaptiveSparkPlanExec) plan).executedPlan(),
tableName);
+ }
+ if (plan instanceof QueryStageExec) {
+ return findBatchScan(((QueryStageExec) plan).plan(), tableName);
+ }
+ if (plan instanceof BatchScanExec
+ && ((BatchScanExec) plan).table().name().contains(tableName)) {
+ return (BatchScanExec) plan;
+ }
+ for (SparkPlan child : JavaConverters.seqAsJavaList(plan.children())) {
+ BatchScanExec scan = findBatchScan(child, tableName);
+ if (scan != null) {
+ return scan;
+ }
+ }
+ return null;
+ }
}
