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 60200390f4 [GLUTEN-12712][CORE] Derive Iceberg getRootPathsInternal
from actual scanned file paths (#12833)
60200390f4 is described below
commit 60200390f473e93203cae1781521705b7cc251c2
Author: DongjunHwang <[email protected]>
AuthorDate: Fri Oct 16 17:23:17 2026 +0900
[GLUTEN-12712][CORE] Derive Iceberg getRootPathsInternal from actual
scanned file paths (#12833)
---
.../gluten/execution/VeloxIcebergSuite.scala | 40 +++++++++++++
gluten-iceberg/pom.xml | 12 ++--
.../gluten/execution/IcebergScanTransformer.scala | 11 +++-
.../spark/source/GlutenIcebergSourceUtil.scala | 67 ++++++++++++++++++++--
.../org/apache/gluten/execution/IcebergSuite.scala | 60 +++++++++++++++++++
5 files changed, 175 insertions(+), 15 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 5bec8136d7..4b71e6d4b6 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
@@ -18,6 +18,7 @@ package org.apache.gluten.execution
import org.apache.gluten.config.{GlutenConfig, VeloxConfig}
import org.apache.gluten.extension.columnar.FallbackTags
+import org.apache.gluten.utils.VeloxFileSystemValidationJniWrapper
import org.apache.spark.sql.Row
import org.apache.spark.sql.connector.catalog.{Identifier, TableCatalog}
@@ -226,4 +227,43 @@ class VeloxIcebergSuite extends IcebergSuite {
}
}
}
+
+ test("iceberg root paths on an unsupported scheme fail native scheme
validation") {
+ // See https://github.com/apache/gluten/issues/12712:
IcebergScanTransformer used to always
+ // report Seq.empty root paths, so VeloxBackend.validateScanExec's scheme
check was silently
+ // skipped for every Iceberg scan, regardless of the scan's actual
filesystem.
+ withTable("iceberg_root_paths_scheme_tb") {
+ spark.sql("""
+ |CREATE TABLE iceberg_root_paths_scheme_tb (id INT)
+ |USING iceberg
+ |""".stripMargin)
+ spark.sql("INSERT INTO iceberg_root_paths_scheme_tb VALUES (1), (2)")
+
+ runQueryAndCompare("SELECT * FROM iceberg_root_paths_scheme_tb") {
+ df =>
+ val scans = getExecutedPlan(df).collect { case i:
IcebergScanTransformer => i }
+ assert(scans.size == 1)
+ val rootPaths = scans.head.getRootPathsInternal
+ assert(rootPaths.nonEmpty)
+ // Confirm these are real table root paths (registered as `file`
scheme), which
+ // VeloxBackendSettings.distinctRootPaths always excludes from
scheme validation (the
+ // local filesystem is always registered) -- so a real root path
from this suite
+ // passes validation, as expected for a supported local table.
+ assert(
+
VeloxFileSystemValidationJniWrapper.allSupportedByRegisteredFileSystems(
+ rootPaths.toArray))
+
+ // Since this suite only exercises the local `file` scheme, which is
always
+ // considered supported, separately confirm -- with a clean,
synthetic URI, not
+ // derived from the real paths above -- that Velox's own native
filesystem check (the
+ // same one VeloxBackend.validateScanExec relies on to decide
whether to fall back)
+ // correctly rejects a genuinely unsupported scheme.
+ assert(
+
!VeloxFileSystemValidationJniWrapper.allSupportedByRegisteredFileSystems(
+ Array("unsupported-test-scheme://bucket/path/file.parquet")),
+ "expected an unsupported scheme to fail native filesystem
validation"
+ )
+ }
+ }
+ }
}
diff --git a/gluten-iceberg/pom.xml b/gluten-iceberg/pom.xml
index 2cb694f8e9..95230f5eea 100644
--- a/gluten-iceberg/pom.xml
+++ b/gluten-iceberg/pom.xml
@@ -56,6 +56,12 @@
<version>${iceberg.version}</version>
<scope>provided</scope>
</dependency>
+ <dependency>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-client</artifactId>
+ <version>${hadoop.version}</version>
+ <scope>provided</scope>
+ </dependency>
<!-- For test -->
<dependency>
@@ -90,12 +96,6 @@
<version>${spark.version}</version>
<scope>test</scope>
</dependency>
- <dependency>
- <groupId>org.apache.hadoop</groupId>
- <artifactId>hadoop-client</artifactId>
- <version>${hadoop.version}</version>
- <scope>test</scope>
- </dependency>
<dependency>
<groupId>org.scalatest</groupId>
<artifactId>scalatest_${scala.binary.version}</artifactId>
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 57ffcb68c6..535c97a1c7 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
@@ -17,6 +17,7 @@
package org.apache.gluten.execution
import org.apache.gluten.backendsapi.BackendsApiManager
+import org.apache.gluten.config.GlutenConfig
import org.apache.gluten.exception.GlutenNotSupportException
import
org.apache.gluten.execution.IcebergScanTransformer.{containsMetadataColumn,
containsUuidOrFixedType}
import org.apache.gluten.sql.shims.SparkShimLoader
@@ -198,8 +199,12 @@ case class IcebergScanTransformer(
override def getDataSchema: StructType = new StructType()
- // TODO: get root paths from table.
- override def getRootPathsInternal: Seq[String] = Seq.empty
+ private lazy val fileFormatAndRootPaths: (ReadFileFormat, Seq[String]) =
+ GlutenIcebergSourceUtil.getFileFormatAndRootPaths(
+ scan,
+ collectRootPaths = GlutenConfig.get.scanFileSchemeValidationEnabled)
+
+ override def getRootPathsInternal: Seq[String] = fileFormatAndRootPaths._2
private lazy val readSchemaFields =
scan.readSchema().fieldNames.map(_.toLowerCase(Locale.ROOT)).toSet
@@ -218,7 +223,7 @@ case class IcebergScanTransformer(
metadataColumns ++ extraMetadataColumns
}
- override lazy val fileFormat: ReadFileFormat =
GlutenIcebergSourceUtil.getFileFormat(scan)
+ override lazy val fileFormat: ReadFileFormat = fileFormatAndRootPaths._1
override def getSplitInfosFromPartitions(
partitions: Seq[(Partition, ReadFileFormat)]): Seq[SplitInfo] = {
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 7704fb54f6..20364bc85c 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
@@ -28,6 +28,7 @@ import
org.apache.spark.sql.catalyst.catalog.ExternalCatalogUtils
import org.apache.spark.sql.connector.read.Scan
import org.apache.spark.sql.types.StructType
+import org.apache.hadoop.fs.Path
import org.apache.iceberg._
import org.apache.iceberg.io.{FileIO, SupportsStorageCredentials}
import org.apache.iceberg.spark.SparkSchemaUtil
@@ -186,16 +187,70 @@ object GlutenIcebergSourceUtil {
metadataColumns
}
- def getFileFormat(sparkScan: Scan): ReadFileFormat = {
+ /**
+ * Derives both the file format and, optionally, the root path(s) actually
scanned from the
+ * planned Iceberg scan tasks in a single pass over
`getScanTasks(sparkScan)`.
+ *
+ * Root paths are derived from the real per-file (and, when present,
per-delete-file) paths rather
+ * than table metadata (Table.location() / write.data.path), since data
files can live at a
+ * location the static metadata does not reflect (e.g. after write.data.path
is changed without
+ * moving previously-written files). One representative path per distinct
scheme prefix is kept,
+ * since callers only need this to validate that the native filesystem
implementation supports
+ * every scheme in play, not to enumerate every file.
+ *
+ * NOTE: this deliberately reuses the same `getScanTasks(sparkScan)`
(Iceberg's own
+ * SparkPartitioningAwareScan#tasks()) that `getReadPartitionSchema` already
calls unconditionally
+ * from `doValidateInternal()`, not Spark's
BatchScanExecShim#filteredPartitions (which
+ * additionally calls SupportsRuntimeV2Filtering#filter() and
+ * Scan#toBatch().planInputPartitions(), and was the root cause of the DPP
regression from
+ * https://github.com/apache/gluten/issues/12712 when a prior revision of
this fix read paths from
+ * BatchScanExecTransformerBase#finalPartitions).
+ */
+ def getFileFormatAndRootPaths(
+ sparkScan: Scan,
+ collectRootPaths: Boolean): (ReadFileFormat, Seq[String]) = {
+ var fileFormat: ReadFileFormat = ReadFileFormat.UnknownFormat
+ // Built with plain mutable local collections that are never returned or
captured, then
+ // copied into an immutable Seq at the end: the result is cached in a
`lazy val` on
+ // IcebergScanTransformer (a case class captured by Spark task closures),
so it must not
+ // retain any reference to a non-serializable mutable collection.
+ val seenSchemes = scala.collection.mutable.HashSet.empty[String]
+ val rootPathsBuilder = Seq.newBuilder[String]
+
+ def recordPath(path: String): Unit = {
+ if (collectRootPaths) {
+ // Use Hadoop's Path to extract the scheme rather than java.net.URI
(whose strict RFC
+ // 3986 parsing throws URISyntaxException on characters like spaces or
'[' / ']' that
+ // Hadoop-style paths otherwise tolerate, e.g. from a partition value
baked into the file
+ // name) or plain "://"-based string slicing (which mis-detects
single-slash Hadoop paths
+ // like "file:/tmp/x.parquet" or "hdfs:/warehouse/x.parquet" as having
no scheme at all,
+ // causing them to collide with each other and silently drop a real
scheme).
+ val scheme = Option(new Path(path).toUri.getScheme).getOrElse("")
+ if (seenSchemes.add(scheme)) {
+ rootPathsBuilder += path
+ }
+ }
+ }
+
asFileScanTask(getScanTasks(sparkScan)).foreach {
task =>
- task.file().format() match {
- case FileFormat.PARQUET => return ReadFileFormat.ParquetReadFormat
- case FileFormat.ORC => return ReadFileFormat.OrcReadFormat
- case _ =>
+ if (fileFormat == ReadFileFormat.UnknownFormat) {
+ task.file().format() match {
+ case FileFormat.PARQUET => fileFormat =
ReadFileFormat.ParquetReadFormat
+ case FileFormat.ORC => fileFormat = ReadFileFormat.OrcReadFormat
+ case _ =>
+ }
+ }
+ if (collectRootPaths) {
+ recordPath(task.file().path().toString)
+ task.deletes().asScala.foreach(deleteFile =>
recordPath(deleteFile.path().toString))
}
}
- throw new GlutenNotSupportException("Iceberg Only support parquet and orc
file format.")
+
+ if (fileFormat == ReadFileFormat.UnknownFormat) {
+ throw new GlutenNotSupportException("Iceberg Only support parquet and
orc file format.")
+ }
+ (fileFormat, rootPathsBuilder.result())
}
def getReadPartitionSchema(sparkScan: Scan): StructType = {
diff --git
a/gluten-iceberg/src/test/scala/org/apache/gluten/execution/IcebergSuite.scala
b/gluten-iceberg/src/test/scala/org/apache/gluten/execution/IcebergSuite.scala
index 4b97852a5a..d69e2229a0 100644
---
a/gluten-iceberg/src/test/scala/org/apache/gluten/execution/IcebergSuite.scala
+++
b/gluten-iceberg/src/test/scala/org/apache/gluten/execution/IcebergSuite.scala
@@ -66,6 +66,66 @@ abstract class IcebergSuite extends
WholeStageTransformerSuite {
}
}
+ test("iceberg getRootPathsInternal covers the actual scanned data files") {
+ // See https://github.com/apache/gluten/issues/12712: getRootPathsInternal
used to always
+ // return Seq.empty for Iceberg scans, silently skipping native filesystem
scheme validation.
+ withTable("iceberg_root_paths_tb") {
+ spark.sql("""
+ |CREATE TABLE iceberg_root_paths_tb (id INT)
+ |USING iceberg
+ |""".stripMargin)
+ spark.sql("INSERT INTO iceberg_root_paths_tb VALUES (1), (2)")
+
+ val actualFilePath = spark
+ .sql("SELECT input_file_name() FROM iceberg_root_paths_tb LIMIT 1")
+ .collect()(0)
+ .getString(0)
+
+ runQueryAndCompare("SELECT * FROM iceberg_root_paths_tb") {
+ df =>
+ val scans = getExecutedPlan(df).collect { case i:
IcebergScanTransformer => i }
+ assert(scans.size == 1)
+ val rootPaths = scans.head.getRootPathsInternal
+ assert(rootPaths.nonEmpty, "getRootPathsInternal should not be empty
for Iceberg tables")
+ assert(
+ rootPaths.exists(actualFilePath.startsWith),
+ s"expected one of root paths ($rootPaths) to be a prefix of the
actual scanned " +
+ s"file ($actualFilePath)"
+ )
+ }
+ }
+ }
+
+ test("iceberg getRootPathsInternal reflects a custom write.data.path") {
+ // See https://github.com/apache/gluten/issues/12712 (review discussion):
the table's
+ // declared location can differ from where data files actually live, e.g.
via the
+ // write.data.path property. getRootPathsInternal must reflect that, not
just the table's
+ // default location.
+ withTempDir {
+ dataDir =>
+ withTable("iceberg_custom_data_path_tb") {
+ spark.sql(s"""
+ |CREATE TABLE iceberg_custom_data_path_tb (id INT)
+ |USING iceberg
+ |TBLPROPERTIES ('write.data.path' =
'${dataDir.getCanonicalPath}')
+ |""".stripMargin)
+ spark.sql("INSERT INTO iceberg_custom_data_path_tb VALUES (1), (2)")
+
+ runQueryAndCompare("SELECT * FROM iceberg_custom_data_path_tb") {
+ df =>
+ val scans = getExecutedPlan(df).collect { case i:
IcebergScanTransformer => i }
+ assert(scans.size == 1)
+ val rootPaths = scans.head.getRootPathsInternal
+ assert(
+ rootPaths.exists(_.contains(dataDir.getCanonicalPath)),
+ s"expected root paths ($rootPaths) to include the configured "
+
+ s"write.data.path (${dataDir.getCanonicalPath})"
+ )
+ }
+ }
+ }
+ }
+
test("rewrite_data_files uses an iceberg staged scan transformer") {
val tableName = "iceberg_rewrite_tb"
withTable(tableName) {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
