This is an automated email from the ASF dual-hosted git repository.
zhouyuan pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/gluten.git
The following commit(s) were added to refs/heads/main by this push:
new 83711d98e5 [VL][Iceberg] Fallback Iceberg read if vended credentials
are present (#13008)
83711d98e5 is described below
commit 83711d98e56cbf597a831de1080634731040f9e6
Author: inf / Hazmi <[email protected]>
AuthorDate: Fri Oct 9 17:18:59 2026 +0300
[VL][Iceberg] Fallback Iceberg read if vended credentials are present
(#13008)
---
.../gluten/execution/VeloxIcebergSuite.scala | 140 ++++++++++++++++++++-
docs/get-started/VeloxIceberg.md | 5 +-
.../gluten/execution/IcebergScanTransformer.scala | 5 +
.../spark/source/GlutenIcebergSourceUtil.scala | 31 +++++
4 files changed, 178 insertions(+), 3 deletions(-)
diff --git
a/backends-velox/src-iceberg/test/scala/org/apache/gluten/execution/VeloxIcebergSuite.scala
b/backends-velox/src-iceberg/test/scala/org/apache/gluten/execution/VeloxIcebergSuite.scala
index 10617a6f07..5bec8136d7 100644
---
a/backends-velox/src-iceberg/test/scala/org/apache/gluten/execution/VeloxIcebergSuite.scala
+++
b/backends-velox/src-iceberg/test/scala/org/apache/gluten/execution/VeloxIcebergSuite.scala
@@ -17,16 +17,152 @@
package org.apache.gluten.execution
import org.apache.gluten.config.{GlutenConfig, VeloxConfig}
+import org.apache.gluten.extension.columnar.FallbackTags
import org.apache.spark.sql.Row
import org.apache.spark.sql.connector.catalog.{Identifier, TableCatalog}
+import org.apache.spark.sql.execution.datasources.v2.BatchScanExec
+import org.apache.spark.utils.GlutenSuiteUtils
-import org.apache.iceberg.UpdateSchema
+import org.apache.iceberg.{FileScanTask, UpdateSchema}
import org.apache.iceberg.expressions.Literal
-import org.apache.iceberg.spark.source.SparkTable
+import org.apache.iceberg.hadoop.HadoopCatalog
+import org.apache.iceberg.io.{ResolvingFileIO, StorageCredential}
+import org.apache.iceberg.spark.{ScanTaskSetManager, Spark3Util, SparkCatalog,
SparkReadOptions}
+import org.apache.iceberg.spark.source.{GlutenIcebergSourceUtil, SparkTable}
import org.apache.iceberg.types.{Type, Types}
+import scala.collection.JavaConverters._
+
class VeloxIcebergSuite extends IcebergSuite {
+ private def withCredentialTable(catalogName: String)(
+ f: (String, ResolvingFileIO) => Unit): Unit = {
+ withTempDir {
+ dir =>
+ withSQLConf(
+ s"spark.sql.catalog.$catalogName" -> classOf[SparkCatalog].getName,
+ s"spark.sql.catalog.$catalogName.type" -> "hadoop",
+ s"spark.sql.catalog.$catalogName.warehouse" -> dir.toURI.toString,
+ s"spark.sql.catalog.$catalogName.io-impl" ->
classOf[ResolvingFileIO].getName,
+ s"spark.sql.catalog.$catalogName.cache-enabled" -> "false"
+ ) {
+ val catalog = spark.sessionState.catalogManager.catalog(catalogName)
+
.asInstanceOf[SparkCatalog].icebergCatalog().asInstanceOf[HadoopCatalog]
+ val tableName = s"$catalogName.default.t"
+ try {
+ withTable(tableName) {
+ withSQLConf(GlutenConfig.GLUTEN_ENABLED.key -> "false") {
+ spark.sql(s"CREATE TABLE $tableName (id INT) USING iceberg")
+ spark.sql(s"INSERT INTO $tableName VALUES (1), (2)")
+ }
+ f(
+ tableName,
+ Spark3Util.loadIcebergTable(spark, tableName).io()
+ .asInstanceOf[ResolvingFileIO])
+ }
+ } finally {
+ try {
+ GlutenSuiteUtils.waitUntilEmpty(spark.sparkContext)
+ } finally {
+ catalog.close()
+ }
+ }
+ }
+ }
+ }
+
+ test("iceberg vended FileIO credentials fall back") {
+ withCredentialTable("vended_credentials") {
+ (tableName, io) =>
+ val originalProperties = io.properties().asScala.toMap
+ val keys = Map("s3.access-key-id" -> "test-key",
"s3.secret-access-key" -> "test-secret")
+ // These are the FileIO inputs supplied by REST catalogs. Exercise the
real scan path
+ // without duplicating Iceberg's REST client/server implementation.
+ val cases = Seq(
+ ("S3 key pair", keys, Seq.empty[StorageCredential]),
+ ("S3 session credentials", keys + ("s3.session-token" ->
"test-token"), Seq.empty),
+ (
+ "storage-credentials",
+ Map.empty[String, String],
+ Seq(StorageCredential.create("s3://bucket/table", keys.asJava))),
+ (
+ "credential refresh",
+ Map("client.refresh-credentials-endpoint" -> "credentials"),
+ Seq.empty)
+ )
+ cases.foreach {
+ case (name, properties, credentials) =>
+ withClue(name) {
+ io.initialize((originalProperties ++ properties).asJava)
+ io.setCredentials(credentials.asJava)
+ runQueryAndCompare(s"SELECT id FROM $tableName", noFallBack =
false) {
+ df =>
+ checkAnswer(df, Seq(Row(1), Row(2)))
+ checkGlutenPlanCount[IcebergScanTransformer](df, 0)
+ val scans = getExecutedPlan(df).collect { case scan:
BatchScanExec => scan }
+ assert(scans.size == 1)
+ val reason =
FallbackTags.get(scans.head.logicalLink.get).reason()
+ assert(reason.contains("catalog-vended credentials"))
+ assert(!reason.contains("test-secret") &&
!reason.contains("test-token"))
+ }
+ }
+ }
+ }
+ }
+
+ test("iceberg staged scans with vended FileIO credentials fall back") {
+ withCredentialTable("staged_vended_credentials") {
+ (tableName, io) =>
+ io.initialize((io.properties().asScala.toMap ++ Map(
+ "s3.access-key-id" -> "test-key",
+ "s3.secret-access-key" -> "test-secret")).asJava)
+ val table = Spark3Util.loadIcebergTable(spark, tableName)
+ val taskSetId = "vended-credentials"
+ val taskSetManager = ScanTaskSetManager.get()
+ val tasks = table.newScan().planFiles()
+ try {
+ taskSetManager.stageTasks(table, taskSetId,
tasks.asScala.toList.asJava)
+ compareDfResultsAgainstVanillaSpark(
+ () =>
+ spark.read
+ .format("iceberg")
+ .option(SparkReadOptions.SCAN_TASK_SET_ID, taskSetId)
+ .load(tableName),
+ noFallBack = false,
+ customCheck = df => {
+ checkAnswer(df, Seq(Row(1), Row(2)))
+ checkGlutenPlanCount[IcebergScanTransformer](df, 0)
+ val scans = getExecutedPlan(df).collect { case scan:
BatchScanExec => scan }
+ assert(scans.size == 1)
+
assert(GlutenIcebergSourceUtil.isSparkStagedScan(scans.head.scan))
+ val reason =
FallbackTags.get(scans.head.logicalLink.get).reason()
+ assert(reason.contains("catalog-vended credentials"))
+ assert(!reason.contains("test-secret"))
+ }
+ )
+ } finally {
+ taskSetManager.removeTasks[FileScanTask](table, taskSetId)
+ tasks.close()
+ }
+ }
+ }
+
+ test("iceberg catalog authentication and storage settings still offload") {
+ withCredentialTable("catalog_settings") {
+ (tableName, io) =>
+ io.initialize((io.properties().asScala.toMap ++ Map(
+ "credential" -> "catalog-secret",
+ "token" -> "catalog-token",
+ "s3.endpoint" -> "http://unused";,
+ "client.region" -> "us-east-1")).asJava)
+ runQueryAndCompare(s"SELECT id FROM $tableName") {
+ df =>
+ checkAnswer(df, Seq(Row(1), Row(2)))
+ checkGlutenPlanCount[IcebergScanTransformer](df, 1)
+ }
+ }
+ }
+
test("iceberg parquet split uses name mapping for projected columns") {
withTable("iceberg_parquet_name_mapping") {
withSQLConf(VeloxConfig.PARQUET_USE_COLUMN_NAMES.key -> "false") {
diff --git a/docs/get-started/VeloxIceberg.md b/docs/get-started/VeloxIceberg.md
index 6b81da3e77..b9f18256a4 100644
--- a/docs/get-started/VeloxIceberg.md
+++ b/docs/get-started/VeloxIceberg.md
@@ -110,7 +110,10 @@ the added column name is same to the deleted column, the
scan will fall back.
These options can be changed per session with `SET`.
### Catalogs
-All the catalog configurations are transparent to Gluten
+
+Catalog configuration is passed through to Iceberg. Scans, including staged
scans, fall back to Spark when table FileIO exposes storage credentials, an S3
access/secret key pair, or a credential refresh endpoint. Native readers do not
support these credentials.
+
+Catalog authentication (`credential`, `token`) and storage settings
(`s3.endpoint`, `client.region`) alone do not trigger this fallback.
### SQL Extensions
Fallback
diff --git
a/gluten-iceberg/src/main/scala/org/apache/gluten/execution/IcebergScanTransformer.scala
b/gluten-iceberg/src/main/scala/org/apache/gluten/execution/IcebergScanTransformer.scala
index 6ac22b956a..57ffcb68c6 100644
---
a/gluten-iceberg/src/main/scala/org/apache/gluten/execution/IcebergScanTransformer.scala
+++
b/gluten-iceberg/src/main/scala/org/apache/gluten/execution/IcebergScanTransformer.scala
@@ -86,6 +86,11 @@ case class IcebergScanTransformer(
}
override def doValidateInternal(): ValidationResult = {
+ if (GlutenIcebergSourceUtil.hasVendedCredentials(scan)) {
+ return ValidationResult.failed(
+ "Iceberg scan uses catalog-vended credentials, which native readers do
not support")
+ }
+
val validationResult = super.doValidateInternal()
if (!validationResult.ok()) {
return validationResult
diff --git
a/gluten-iceberg/src/main/scala/org/apache/iceberg/spark/source/GlutenIcebergSourceUtil.scala
b/gluten-iceberg/src/main/scala/org/apache/iceberg/spark/source/GlutenIcebergSourceUtil.scala
index 233c06c30c..7704fb54f6 100644
---
a/gluten-iceberg/src/main/scala/org/apache/iceberg/spark/source/GlutenIcebergSourceUtil.scala
+++
b/gluten-iceberg/src/main/scala/org/apache/iceberg/spark/source/GlutenIcebergSourceUtil.scala
@@ -29,7 +29,9 @@ import org.apache.spark.sql.connector.read.Scan
import org.apache.spark.sql.types.StructType
import org.apache.iceberg._
+import org.apache.iceberg.io.{FileIO, SupportsStorageCredentials}
import org.apache.iceberg.spark.SparkSchemaUtil
+import org.apache.iceberg.util.PropertyUtil
import java.lang.{Long => JLong}
import java.util.{ArrayList => JArrayList, HashMap => JHashMap, List => JList,
Map => JMap}
@@ -49,6 +51,35 @@ object GlutenIcebergSourceUtil {
def isSparkStagedScan(sparkScan: Scan): Boolean =
sparkScan.isInstanceOf[SparkStagedScan]
+ def hasVendedCredentials(sparkScan: Scan): Boolean =
+ isSupportedScan(sparkScan) &&
hasVendedCredentials(getTable(sparkScan).io())
+
+ private def hasVendedCredentials(io: FileIO): Boolean = {
+ io match {
+ case credentials: SupportsStorageCredentials if
!credentials.credentials().isEmpty =>
+ return true
+ case _ =>
+ }
+ val properties =
+ try {
+ io.properties()
+ } catch {
+ case _: UnsupportedOperationException => return false
+ }
+ (Option(properties.get("s3.access-key-id")).exists(_.nonEmpty) &&
+ Option(properties.get("s3.secret-access-key")).exists(_.nonEmpty)) ||
+ Option(properties.get("gcs.oauth2.token")).exists(_.nonEmpty) ||
+ properties.asScala.exists {
+ case (key, value) =>
+ key.startsWith("adls.sas-token.") && Option(value).exists(_.nonEmpty)
+ } ||
+ Seq("client", "gcs.oauth2", "adls").exists {
+ prefix =>
+
Option(properties.get(s"$prefix.refresh-credentials-endpoint")).exists(_.nonEmpty)
&&
+ PropertyUtil.propertyAsBoolean(properties,
s"$prefix.refresh-credentials-enabled", true)
+ }
+ }
+
def deleteExists(p: SparkDataSourceRDDPartition): Boolean = {
p.inputPartitions.exists {
case ip: SparkInputPartition =>
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
