[GLUTEN-12712][CORE] Derive Iceberg getRootPathsInternal from actual scanned file paths - #12833
[GLUTEN-12712][CORE] Derive Iceberg getRootPathsInternal from actual scanned file paths#12833HwangDongJun wants to merge 1 commit into
Conversation
|
Run Gluten Clickhouse CI on x86 |
8784a8c to
ee68b26
Compare
|
Run Gluten Clickhouse CI on x86 |
| // SparkShims.getBatchScanExecTable returns null on Spark 3.3 (BatchScanExec has no `table` | ||
| // field until Spark 3.4), so IcebergScanTransformer.table is always null there and this | ||
| // improvement does not take effect; it does on Spark 3.4+. | ||
| testWithMinSparkVersion("iceberg getRootPathsInternal returns table location", "3.4") { | ||
| // 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(rootPaths.forall(_.nonEmpty)) | ||
| } | ||
| } | ||
| } | ||
|
|
There was a problem hiding this comment.
This test only verifies that getRootPathsInternal is non-empty, but the test warehouse uses file:// already:
Velox explicitly filters file:// paths out of filesystem scheme validation:
So this does not test any regression. Can we change the test to a case with an unsupported scheme and verify that the Iceberg scan falls back instead?
There was a problem hiding this comment.
Thanks for pointing that out — you're right that file:// gets excluded from distinctRootPaths, so that test wasn't really proving anything. I wasn't able to get a real table working on an actually-unsupported scheme (Iceberg's FileIO fails the write itself in that case), so I took a different approach instead.
The shared test now checks that the returned path is a real per-file path (ends in .parquet, under .../data/...), and I added a Velox-side test that feeds a fake unsupported scheme directly into VeloxFileSystemValidationJniWrapper.allSupportedByRegisteredFileSystems (the same function validateScanExec relies on) and checks that it's correctly rejected.
Happy to adjust further if you think there's a better way to test this.
| case t: SparkTable => Seq(t.table().location()) | ||
| case _ => Seq.empty | ||
| } | ||
| } |
There was a problem hiding this comment.
Could we derive the paths from the Iceberg scan instead of BatchScanExec.table? GlutenIcebergSourceUtil already accesses SparkBatchQueryScan.table(), so this also works on Spark3.3 and we wouldn't need any version limitations:
Also, table.location() is not necessarily the filesystem containing the files being read. Iceberg supports a separate write.data.path and write.location-provider.impl:
https://github.com/apache/iceberg/blob/apache-iceberg-1.10.0/docs/docs/configuration.md#write-properties
Gluten already gets the actual data path from task.file().path() when building the native scan:
Could we add a helper in GlutenIcebergSourceUtil that extracts the actual scan file path and use that instead?
There was a problem hiding this comment.
Thank you, that's a really good catch — I hadn't considered write.data.path at all. I've updated the fix to derive the path from the actual scan tasks instead (GlutenIcebergSourceUtil.getRootPaths(finalPartitions), one task.file().path() per partition, following the same pattern already used in genSplitInfo/deleteExists). As a bonus, this also resolves the Spark 3.3 gap, since it no longer relies on BatchScanExec.table.
One thing I'd like to flag for visibility: this means finalPartitions now gets computed during validation for every scan, rather than only for format-v3 tables as before. It's the same cached value, just triggered more often. Also, asFileScanTask can throw for an unexpected task type, but that's already caught upstream and converted into a fallback rather than a crash, so it should be safe.
Please let me know if you'd like me to handle this differently.
ee68b26 to
d00725b
Compare
|
Run Gluten Clickhouse CI on x86 |
d00725b to
6862b36
Compare
|
Run Gluten Clickhouse CI on x86 |
HwangDongJun
changed the title
…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. Generated-by: OpenCode claude-sonnet-5
6862b36 to
7879935
Compare
|
Run Gluten Clickhouse CI on x86 |
2 participants
What changes are proposed in this pull request?
IcebergScanTransformer.getRootPathsInternalwas hardcoded to returnSeq.empty, behind a// TODO: get root paths from table.comment (gluten-iceberg/src/main/scala/org/apache/gluten/execution/IcebergScanTransformer.scala:196).BasicScanExecTransformer.doValidateInternalpasses this value asrootPathsintoVeloxBackend.validateScanExec, whosevalidateScheme()only performs native filesystem scheme validation (VeloxFileSystemValidationJniWrapper.allSupportedByRegisteredFileSystems) whenfilteredRootPaths.nonEmpty. An emptySeqis therefore treated as "nothing to check", so scheme validation is silently skipped 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. See #12712.
This PR returns the Iceberg table's base location, mirroring how DSv1/DSv2 scans already expose their root paths via
FileIndex.rootPaths(BatchScanExecTransformer.getRootPathsInternal).table.location()is the standard Iceberg API for a table's base location and is already used elsewhere in this codebase (e.g.ContentFileUtil.java).Fixes #12712.
How was this patch tested?
Added a regression test (
gluten-iceberg/src/test/scala/org/apache/gluten/execution/IcebergSuite.scala) assertinggetRootPathsInternalreturns a non-empty root path for a plain Iceberg table, instead of the previousSeq.empty.Verified locally that
gluten-iceberg(main + test sources) compiles cleanly against the change with./build/mvn compile/test-compile -pl gluten-iceberg -am -Piceberg, and that scalastyle reports 0 errors/warnings.Was this patch authored or co-authored using generative AI tooling?
Generated-by: OpenCode claude-sonnet-5