fix: load the Iceberg storage scheme lists from the native factory - #6065
dwsmith1983 wants to merge 32 commits into
Conversation
The native storage factory now exposes the schemes it can open per access mode, and a JNI entry returns them, so the JVM read and write gates can load the list instead of mirroring it by hand. Tests keep the list and the factory's match arms in step in both directions.
The JVM Iceberg scan and write gates hardcoded the schemes the native storage factory can open and had already drifted from it. Load both lists over JNI through a new IcebergStorageSchemes object, keeping fallback constants only for when the library cannot load, and pin those constants to what native publishes in CometScanSchemeFallbackSuite. A wasb-backed Iceberg table now falls back to Spark with a reason naming the scheme instead of failing natively at execution.
…es verbatim Make builtin_storage_schemes the single point of change by rejecting any scheme it does not list before the factory match, drop memory from the read list since an OpenDAL memory backend is a fresh empty store, and load the JVM sets lazily behind the library-loaded check so a disabled Comet never touches the native library. The JVM gates now match the built-in set verbatim, as OpenDAL strips the scheme prefix case-sensitively at open time. Tests round-trip the JNI lists directly and cover the wasb write gate.
github-actions
Bot
added
bug
sunchao
left a comment
sunchao
left a comment
There was a problem hiding this comment.
Correctness
Reviewed ab9288bbdaf5 against d1bf687ebb13. I found no new P1/P2 issue in the changed storage admission and JNI paths.
The earlier scheme alignment already made the JVM reject unsupported Iceberg locations. This PR addresses the remaining maintenance problem: read admission, write admission and native factory selection were separate lists. The native factory now uses the mode-specific lists itself, and both JVM gates load those lists through the new probe.
The read/write distinction is preserved: oss is admitted for reads, while memory remains available for the writer's temporary manifest store. That manifest is read back through the same FileIO created with write access, so rejecting a new memory-backed read factory does not break manifest serialization. Built-in schemes now match the native factory's case-sensitive admission. Custom S3 aliases remain additive and case-insensitive for scans. The write planner still declines aliases.
I compared the fallback boundary with the maintained Spark 3.5 and 4.0 sources. Their V2 scan delegates to the data source's reader factory, and their V2 write delegates to its writer factory and commit protocol. Declining native admission retains that JVM reader or the existing Iceberg JVM writer path. This change does not alter expression types, null handling, ANSI arithmetic or overflow behavior. I could not check the absent maintained 3.4 and 4.1 branches and am not claiming runtime compatibility across those versions.
The lazy JVM helper is reached after the production rules' native-load checks. The new JNI entry point uses try_unwrap_or_throw, returns a Java string and introduces no Arrow pointer or batch-lifetime change. The helper does not catch a failed native probe and silently substitute an optimistic list.
Validation
The added tests exercise native mode admission, direct JNI list retrieval, the fallback constants and mixed-case rejection. The local-disk wasb test checks both the write fallback reason and the scan result against Spark. It is regression coverage for an already-rejected scheme. The two changed Scala suites are registered in both Linux and macOS CI.
My validation was source review and exact-revision consistency checks, including git diff --check. I did not execute the Rust, Spark or JNI suites. The author reports 9 Rust tests and Spark 3.5 / Java 17 results of 12 scheme tests, 53 write-detection tests and 99 native Iceberg tests, with 8 version-gated cancellations.
Current Comet CI, CodeQL and title validation require approval and have zero jobs. Only the label check passed. The merge commit has the reviewed head's tree, but there is no product CI execution to credit yet.
Performance
Each mode's scheme set is loaded lazily and cached. The file-classification loop performs set membership checks without calling JNI per file, row or batch. The native admission check scans a five-element static list when selecting a factory. I found no material new allocation or repeated-work concern in these paths. I did not measure query time or planning time, so this review makes no performance claim.
Design
Having the native factory consume the same lists it publishes makes the capability probe useful: adding a scheme to a JVM constant alone can no longer expand normal native admission. Keeping separate read and write lists reflects the existing backend restrictions. Keeping catalog aliases outside the built-in lists also preserves their explicit opt-in behavior.
The existing FileIO, credential and GCS-specific eligibility checks remain necessary and remain in place. Scheme support is only one admission condition. I found no additional design change worth requesting before merge.
Abstraction & complexity
The small IcebergStorageSchemes helper centralizes two cached sets, parsing and the unloaded-library fallback. It fits the existing native capability-probe pattern without introducing a registry or changing operator serialization. The duplicated fallback constants are checked directly against JNI output, which makes their remaining maintenance obligation explicit. I found no actionable simplification in the changed scope.
sunchao
left a comment
sunchao
left a comment
There was a problem hiding this comment.
Re-reviewed dc533139807e against 5ca149928f77, including the changes since ab9288bbdaf5. The only intervening commit merges current main. All ten files in this PR's contribution are unchanged, with no merge-resolution edits. I found no new or remaining P1/P2 issue and am preserving the existing approval.
I rechecked the native scheme lists, lazy JVM loading, reflection fallback and registry integration. Scheme discovery itself has no Iceberg Java dependency. The file-task accessors used by its scan caller remain present in the Iceberg 1.8.1, 1.9.2, 1.10.0 and 1.11.0 release sources. The probe still runs once per access mode, and the merge introduces no additional per-file JNI work or new abstraction in this path.
Validation was source and exact-revision checking, including the maintained Spark 3.5/4.0 fallback boundaries and git diff --check. I did not execute Spark, JNI, Rust tests or benchmarks, and am not claiming an Iceberg runtime matrix. Maintained Spark 3.4/4.1 source remains unavailable. The merge ref has parents equal to the reviewed base and head and the same tree as the head. Comet CI and CodeQL still require approval with zero jobs. Only the label check passed, so there is no product CI execution or actual product-test checkout to credit.
| private[comet] val FallbackRead: Set[String] = Set("file", "s3", "s3a", "gs", "oss") | ||
| private[comet] val FallbackWrite: Set[String] = Set("file", "memory", "s3", "s3a", "gs") |
There was a problem hiding this comment.
Since every caller sits behind isCometLoaded, FallbackRead and FallbackWrite are only consulted in a JVM where nothing runs natively. Could load return Set.empty when NativeBase.isLoaded is false instead? That would remove the last hand-maintained copy of the list, which is the goal of the PR. It would also let us drop the two pinning tests in CometScanSchemeFallbackSuite, which mostly check the same thing. Is there a path I'm missing where the fallback does real work?
There was a problem hiding this comment.
Is there a path I'm missing where the fallback does real work?
No. Both lazy sets are first touched inside CometScanRule and CometExecRule, whose _apply returns the plan untouched unless isCometLoaded is true, and isCometLoaded ends by evaluating NativeBase.isLoaded. So by the time either set is read the library is loaded. In 1d3f95d load returns the empty set with a warning when it is not, a blank list from a loaded library throws since native publishes a fixed non-empty constant and a blank one can only be a build bug, and the two fallback constants and the two pinning tests are gone. load takes the loaded flag as a parameter so the unloaded answer is tested without unloading the library, and one test checks the lazy sets against a direct parse of the probe.
| private def isAdmittedIcebergScheme(scheme: String, s3CompliantSchemes: Set[String]): Boolean = | ||
| icebergReadableSchemes.contains(scheme) || | ||
| s3CompliantSchemes.contains(scheme.toLowerCase(Locale.ROOT)) |
There was a problem hiding this comment.
The reason for matching built-in schemes verbatim is that OpenDAL compares the prefix case-sensitively. In the pinned iceberg-rust, the S3 arm builds that prefix from Url::parse(path).scheme(), which is always lowercase, and then does path.starts_with(&prefix) on the raw path. So I think BLOB://bucket/key with blob opted in would pass both gates (this one and is_s3_compliant_alias_scheme natively, which are both case-insensitive) and then fail with Invalid s3 url ... should start with blob://bucket/. BlobHostPromotingS3Storage only rewrites hostless forms, so it doesn't help here. Does that match what you see? If so, should aliases be matched verbatim as well, with a test for a mixed-case alias location? Or is there a reason aliases need to stay case-insensitive?
There was a problem hiding this comment.
Does that match what you see? If so, should aliases be matched verbatim as well, with a test for a mixed-case alias location?
It matches. The S3 arm builds {scheme}://{bucket}/ from Url::parse, whose scheme is lowercase, and calls starts_with on the raw path, so BLOB://bucket/key with blob opted in passed the gate and the native check and then failed at open. There is no reason for aliases to stay case-insensitive on this path; the Parquet path keeps it because it rewrites alias locations to s3:// before anything opens them.
1d3f95d matches aliases verbatim on the Iceberg path. The gate looks up the scheme as written against the lowercased list, and native admits an alias only when the location's scheme has no uppercase, on the scan, write and region-default paths. One consequence: a hostless mixed-case location such as BLOB:///bucket/key, which the host promotion used to rewrite and open, is declined too, so aliases follow the same rule as the built-ins everywhere. Tests cover the mixed-case location on both sides, and the data sources guide says the location must spell the alias in lowercase.
The JVM scheme sets are only read after the native library has loaded, so the hand-maintained fallback lists never did work. `load` now returns the empty set when the library is not loaded, and a blank native list parses to the empty set, each with a warning. iceberg-rust builds the S3 prefix from the parsed scheme, which is lowercase, and compares it to the raw path, so a mixed-case alias location passed the gates and failed at open. Aliases are now matched as written on the Iceberg path, on the JVM gate and in the native factory, like the built-in schemes. The data sources guide says an Iceberg location must spell the alias in lowercase.
andygrove
left a comment
andygrove
left a comment
There was a problem hiding this comment.
Both of my earlier points are addressed in 1d3f95d, thanks. I merged this into current main, and CometScanSchemeFallbackSuite, CometIcebergWriteDetectionSuite and the iceberg_common Rust tests all pass. Putting the lowercasing back in storageScheme makes the new S3:// test fail with got Compatible(None,None), so it pins the change. Could you update the PR description before this merges? It still says IcebergStorageSchemes falls back to the old constants when the library isn't loaded, and that aliases are matched case-insensitively, and neither is true after that commit.
Updated. It now says an unloaded library yields an empty set and logs a warning, that the opt-in aliases are matched verbatim like the built-in schemes, and the test paragraph describes what |
|
@andygrove could you approve a CI run on |
|
@andygrove could you approve a CI run on |
Head branch was pushed to by a user without write access
3 participants
Which issue does this PR close?
Closes #5541.
Rationale for this change
The JVM validator for native Iceberg scans checked file schemes against a hand-written set that disagreed with the native storage factory, so a table under
gcs,abfs,abfss,wasborwasbswas claimed and every task died with "Unsupported storage scheme", whileossworked natively and was rejected. #5314 aligned the read list, but three lists still exist by hand: the read gate, the write gate and the native factory, with a comment telling contributors to update the JVM when the native arms change.What changes are included in this PR?
builtin_storage_schemes(access_mode)becomes the single point of change:storage_factory_forrejects any scheme that is neither listed for the access mode nor an opted-in S3-compliant alias before matching an arm, so theossarm no longer carries its own write rejection andmemoryno longer needs one. The list is exposed over JNI asNativeBase.icebergStorageSchemes(forWrite), the same pattern as the existingisObjectStoreSchemeSupportedprobe. Opt-in aliases stay JVM-side and additive, as before.IcebergStorageSchemesobject loads the read and write sets lazily, once per JVM, from that probe. It consultsNativeBase.isLoadedfirst. When the library is not loaded it logs a warning and returns an empty set, so nothing is admitted. Every caller sits behindisCometLoaded, so that case only arises in a JVM where nothing runs natively. A native fault while answering the probe propagates instead of being swallowed.CometScanRule.icebergReadableSchemesandCometIcebergNativeWrite.SupportedStorageSchemesare lazy views of it, so constructing the extension no longer touches the native library.memoryis not admitted for reads, on either side. The nativememoryarm builds a fresh, empty in-process OpenDAL store perFileIO; it exists for the write path, which assembles manifest bytes in it, and amemory:read can never find a table. Main declined it at plan time; this PR keeps that, and the write gate keeps admitting it.S3://bucket/keyfails itss3://bucket/prefix check before any request is made), so aS3://location the JVM admitted was failing natively. Both gates decline it now. The opt-in alias list is matched the same way: native rejects an alias scheme with any uppercase letter, and the JVM matches the configured aliases verbatim.How are these changes tested?
Rust tests in
iceberg_common.rscheck that every listed scheme builds a factory for its mode, that the pre-check is load-bearing (ossreads but is rejected for writes purely because it is absent from the write list, andmemorythe other way round), thathdfs,abfs,abfss,wasb,wasbs,gcs,http,httpsandazureare rejected and unlisted, and that mixed-case schemes are rejected for both modes. They do not prove that the JVM declines everything native rejects; the pre-check makes the list the only thing native consults, which is what closes that direction.CometScanSchemeFallbackSuiteunit-tests the JNI list parser, checks that an unloaded library yields no schemes, checks that the sets come from native and carry each mode's own entries, checks that an opt-in alias is matched verbatim, and asserts the scan gate admitsfile,s3,s3a,gsandosswhile decliningmemory,S3://,File://,http(s),abfs(s)andwasb(s). Thewasbtest is a regression guard rather than proof of the fix (main already declinedwasb): it backs the scheme with a local filesystem, runs the INSERT under plan capture and asserts noCometIcebergWriteExecwith the reasonunsupported storage scheme: wasb, which gives the write gate its only end-to-end coverage, then asserts the scan is declined with a reason naming the scheme, that nothing but the scheme caused the fallback, and that the answer matches Spark.CometIcebergWriteDetectionSuitegains a case for aS3://data location, which must be declined.Run locally with Spark 3.5 / Java 17:
cargo test -p datafusion-comet --lib iceberg_common: 9 passed../mvnw -Pspark-3.5 test -DwildcardSuites=org.apache.comet.rules.CometScanSchemeFallbackSuite,org.apache.comet.CometIcebergNativeSuite,org.apache.comet.CometIcebergWriteDetectionSuite:CometScanSchemeFallbackSuite12 succeeded,CometIcebergWriteDetectionSuite53 succeeded,CometIcebergNativeSuite99 succeeded and 8 canceled (the pre-existing Iceberg 1.11+ and Spark 4.0+ gates); 0 failed.cargo clippy --all-targets -- -D warnings,cargo fmt,spotless:check,scalafixCHECK,dev/ci/check-suites.py: clean.