andygrove commented on code in PR #6065: URL: https://github.com/apache/datafusion-comet/pull/6065#discussion_r4073336209
########## spark/src/main/scala/org/apache/comet/iceberg/IcebergStorageSchemes.scala: ########## @@ -0,0 +1,73 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.comet.iceberg + +import java.util.Locale + +import org.apache.spark.internal.Logging + +import org.apache.comet.NativeBase + +/** + * The storage schemes the native Iceberg storage factory publishes over JNI, so the JVM scan and + * write gates decline what native cannot open instead of failing at execution. Every caller sits + * behind `isCometLoaded`, so the fallback constants are only consulted in a JVM where nothing + * runs natively; the pinning test in `CometScanSchemeFallbackSuite` keeps them equal to the + * native lists. + */ +private[comet] object IcebergStorageSchemes extends Logging { + + private[comet] val FallbackRead: Set[String] = Set("file", "s3", "s3a", "gs", "oss") + private[comet] val FallbackWrite: Set[String] = Set("file", "memory", "s3", "s3a", "gs") Review 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? ########## spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala: ########## @@ -1192,20 +1192,22 @@ object CometScanRule extends Logging { org.apache.spark.sql.catalyst.trees.TreeNodeTag[Unit]("comet.skipCometScan") /** - * Schemes Comet's native Iceberg scan can actually open, mirroring the match arms in - * `native/core/src/execution/operators/iceberg_common.rs::storage_factory_for`. Deliberately - * NOT delegated to `isNativelyReadableScheme`: object_store recognizes schemes (http/https, - * azure, memory) that iceberg-rust's OpenDAL storage factory cannot build, and admitting them - * here turns a clean JVM fallback into a native runtime "Unsupported storage scheme" error. Add - * here what you add to `storage_factory_for` (currently Aliyun `oss` and GCS `gs`). - * S3-compliant aliases like `blob` are opt-in via `fs.comet.s3Compliant.schemes` (see - * `isIcebergReadableScheme`), not hardcoded, since the native planner opens them via S3. The - * write path keeps its own list (`CometIcebergNativeWrite.SupportedStorageSchemes`), which - * differs deliberately: it excludes `oss` (fails closed, see `storage_factory_for`) and - * includes `memory`. + * Schemes Comet's native Iceberg scan can open, loaded from the native storage factory over JNI + * so this gate cannot drift from `storage_factory_for`. Lazy so that constructing the rule does + * not touch the native library before `isCometLoaded` has been consulted. Opt-in aliases from + * `fs.comet.s3Compliant.schemes` are additive (see `isIcebergReadableScheme`); the write path + * loads its own set (`CometIcebergNativeWrite.SupportedStorageSchemes`). */ - private val icebergReadableSchemes: Set[String] = - Set("file", "s3", "s3a", "gs", "oss") + private lazy val icebergReadableSchemes: Set[String] = IcebergStorageSchemes.read + + /** + * True when the Iceberg scan gate admits `scheme`. The built-in set matches verbatim because + * native opens a location by its raw scheme and OpenDAL strips that prefix case-sensitively, so + * `S3://` is not `s3://`; the opt-in alias list is matched case-insensitively on both sides. + */ + private def isAdmittedIcebergScheme(scheme: String, s3CompliantSchemes: Set[String]): Boolean = + icebergReadableSchemes.contains(scheme) || + s3CompliantSchemes.contains(scheme.toLowerCase(Locale.ROOT)) Review 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? -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
