From bde4fc368a8cecb711641fc6941110e786a69b25 Mon Sep 17 00:00:00 2001 From: DongjunHwang Date: Mon, 24 Aug 2026 10:47:34 +0900 Subject: [PATCH] [GLUTEN-12712][CORE] Derive Iceberg getRootPathsInternal from actual scanned file paths IcebergScanTransformer.getRootPathsInternal was hardcoded to return Seq.empty behind a "TODO: get root paths from table." comment. BasicScanExecTransformer.doValidateInternal passes this value as rootPaths into VeloxBackend.validateScanExec, whose validateScheme() only performs native filesystem scheme validation (VeloxFileSystemValidationJniWrapper.allSupportedByRegisteredFileSystems) "if (filteredRootPaths.nonEmpty && ...)". An empty Seq is therefore treated as "nothing to check", silently skipping scheme validation for every Iceberg scan regardless of the table's actual filesystem. As a result, Gluten does not correctly fall back to vanilla Spark for Iceberg tables backed by a filesystem scheme unsupported by the native build; any resulting failure instead surfaces later inside native code, in a much less clear form. Fix: add GlutenIcebergSourceUtil.getRootPaths, which returns one representative scanned data file path per planned Spark input partition, taken from the actual FileScanTask (task.file().path()), and wire it into IcebergScanTransformer.getRootPathsInternal via the already-computed finalPartitions. This intentionally reads the real per-file scan path rather than the Iceberg table's declared base location (Table.location()): - Iceberg lets a table relocate its actual data files independently of the table location via the write.data.path property or a custom LocationProvider, so Table.location() is not guaranteed to reflect the filesystem the data is actually read from. - It works uniformly across all supported Spark versions. An earlier version of this patch read the location via SparkShims.getBatchScanExecTable(batchScan), which always returns null on Spark 3.3 (BatchScanExec has no `table` field until Spark 3.4), silently no-oping there; deriving paths from the scan's own planned partitions has no such gap. Trade-off: since scanFileSchemeValidationEnabled defaults to true, computing root paths now forces Iceberg partition/split planning (finalPartitions) during plan validation rather than only at execution time. This is the same lazily-memoized value already forced during validation for other checks (e.g. the format-version >= 3 delete-file check), just no longer conditional on format version. The reused task-decoding helper (asFileScanTask) can throw UnsupportedOperationException for an unexpected ScanTask type; that is already safely converted to a validation failure (fallback, not a crash) by ValidatablePlan.failValidationWithException. Adds a regression test asserting getRootPathsInternal returns the actual per-file scanned path (not just a non-empty placeholder), plus a Velox-specific test confirming that (a) those real `file`-scheme paths pass native filesystem validation as expected, and (b) a genuinely unsupported scheme -- a clean synthetic URI, not derived by string-concatenating a fake scheme onto an already-scheme-prefixed real path -- is correctly rejected by the same native filesystem check VeloxBackend.validateScanExec relies on. Reflow the getRootPaths doc comment to satisfy spotless/scalafmt's line-wrapping, which our local formatting pass (JDK 25, incompatible with the pinned spotless-maven-plugin version) couldn't verify and was caught by CI instead. Generated-by: OpenCode claude-sonnet-5 --- .../gluten/execution/VeloxIcebergSuite.scala | 39 +++++++++++++++++++ .../execution/IcebergScanTransformer.scala | 4 +- .../source/GlutenIcebergSourceUtil.scala | 26 +++++++++++++ .../gluten/execution/IcebergSuite.scala | 26 +++++++++++++ 4 files changed, 93 insertions(+), 2 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 4e540c15af5..6eb262a6fd6 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,6 +17,7 @@ package org.apache.gluten.execution import org.apache.gluten.config.GlutenConfig +import org.apache.gluten.utils.VeloxFileSystemValidationJniWrapper import org.apache.spark.sql.Row import org.apache.spark.sql.connector.catalog.{Identifier, TableCatalog} @@ -64,4 +65,42 @@ 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 per-file scan 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/src/main/scala/org/apache/gluten/execution/IcebergScanTransformer.scala b/gluten-iceberg/src/main/scala/org/apache/gluten/execution/IcebergScanTransformer.scala index 16018e086df..f5a19713aff 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 @@ -192,8 +192,8 @@ case class IcebergScanTransformer( override def getDataSchema: StructType = new StructType() - // TODO: get root paths from table. - override def getRootPathsInternal: Seq[String] = Seq.empty + override def getRootPathsInternal: Seq[String] = + GlutenIcebergSourceUtil.getRootPaths(finalPartitions) private lazy val readSchemaFields = scan.readSchema().fieldNames.map(_.toLowerCase(Locale.ROOT)).toSet 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 db1bb024afa..053f478319f 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 @@ -23,6 +23,7 @@ import org.apache.gluten.execution.SparkDataSourceRDDPartition import org.apache.gluten.substrait.rel.{IcebergLocalFilesBuilder, SplitInfo} import org.apache.gluten.substrait.rel.LocalFilesNode.ReadFileFormat +import org.apache.spark.Partition import org.apache.spark.softaffinity.SoftAffinity import org.apache.spark.sql.catalyst.catalog.ExternalCatalogUtils import org.apache.spark.sql.connector.read.Scan @@ -112,6 +113,31 @@ object GlutenIcebergSourceUtil { ) } + /** + * Returns one representative data file path per planned Spark input partition, so callers can + * validate the underlying filesystem scheme(s) without enumerating every data file. + * + * We deliberately read the actual scanned file paths (task.file().path()) rather than the Iceberg + * table's declared base location (Table.location()): Iceberg lets a table relocate its actual + * data files independently of the table location via the write.data.path property or a custom + * LocationProvider, so the table location is not guaranteed to reflect the filesystem the data is + * actually read from. This also works uniformly across all supported Spark versions, since it + * does not depend on BatchScanExec exposing its Table (only added in Spark 3.4; see + * SparkShims.getBatchScanExecTable). + */ + def getRootPaths(partitions: Seq[Partition]): Seq[String] = { + partitions.flatMap { + case p: SparkDataSourceRDDPartition => + p.inputPartitions.flatMap { + case ip: SparkInputPartition => + val tasks = ip.taskGroup[ScanTask]().tasks().asScala + asFileScanTask(tasks.toList).headOption.map(_.file().path().toString) + case _ => None + } + case _ => Seq.empty + } + } + def getFieldIds(sparkScan: Scan): JHashMap[String, Integer] = { val fieldIds = new JHashMap[String, Integer]() sparkScan match { 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 56f3fbdace7..c2cafa2f4a9 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 @@ -60,6 +60,32 @@ abstract class IcebergSuite extends WholeStageTransformerSuite { } } + test("iceberg getRootPathsInternal returns actual scanned file paths") { + // 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)") + + 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 the paths point at the actual scanned data files, not just some non-empty + // placeholder: Iceberg tables can relocate real data files independently of the + // table's declared base location (e.g. via write.data.path / a custom + // LocationProvider), so only the real per-file scan path reliably reflects the + // filesystem the data is actually read from. + assert(rootPaths.forall(p => p.contains("/data/") && p.endsWith(".parquet"))) + } + } + } + test("iceberg input_file_name") { withTable("iceberg_input_file_tb") { spark.sql("""