sunchao commented on code in PR #5365:
URL: https://github.com/apache/datafusion-comet/pull/5365#discussion_r3942085900


##########
contrib/delta-spark/src/main/scala/org/apache/comet/contrib/delta/DeltaScanSupport.scala:
##########
@@ -0,0 +1,1789 @@
+/*
+ * 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.contrib.delta
+
+import java.io.IOException
+import java.net.URI
+import java.util.Locale
+
+import scala.collection.mutable.{ListBuffer, Map => MutableMap}
+import scala.jdk.CollectionConverters._
+
+import org.apache.hadoop.conf.Configuration
+import org.apache.hadoop.fs.Path
+import org.apache.spark.sql.catalyst.expressions.{Alias, GenericInternalRow, 
InputFileBlockLength, InputFileBlockStart, InputFileName}
+import org.apache.spark.sql.catalyst.util.{ArrayBasedMapData, GenericArrayData}
+import 
org.apache.spark.sql.catalyst.util.ResolveDefaultColumns.getExistenceDefaultValues
+import org.apache.spark.sql.comet.CometScanExec
+import org.apache.spark.sql.delta.DeltaParquetFileFormat
+import org.apache.spark.sql.delta.actions.DeletionVectorDescriptor
+import org.apache.spark.sql.execution.{FileSourceScanExec, ProjectExec, 
SparkPlan}
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.types.{ArrayType, DataType, MapType, StructType}
+
+import org.apache.comet.CometConf
+import org.apache.comet.CometConf.COMET_LIBHDFS_SCHEMES
+import org.apache.comet.objectstore.NativeConfig
+import org.apache.comet.parquet.CometParquetUtils
+import org.apache.comet.rules.{CometScanRule, CometScanTypeChecker}
+import org.apache.comet.serde.operator.CometNativeScan
+import org.apache.comet.shims.ShimFileFormat
+
+/**
+ * Claim/decline gates for the native Delta scan. Correctness rule: when in 
doubt, decline,
+ * Spark's Delta reader handles the scan and results stay correct, just 
unaccelerated.
+ */
+object DeltaScanSupport {
+
+  /**
+   * Reader features the native path understands; anything else on the 
protocol declines the
+   * table. `deletionVectors`/`columnMapping` are declined separately below 
for specific reasons.
+   */
+  private val understoodReaderFeatures: Set[String] =
+    Set("columnMapping", "deletionVectors", "timestampNtz", "v2Checkpoint", 
"vacuumProtocolCheck")
+
+  /**
+   * Is this exactly Delta's DSv1 parquet format? Compared by class name, not 
`classOf`: a
+   * `classOf` reference would raise `NoClassDefFoundError` and break every 
parquet scan when
+   * delta-spark is absent from the classpath.
+   */
+  def isDeltaScan(scanExec: FileSourceScanExec): Boolean =
+    scanExec.relation.fileFormat.getClass.getName ==
+      "org.apache.spark.sql.delta.DeltaParquetFileFormat"
+
+  /**
+   * Claim-time artifacts [[declineReason]] already computes but 
[[CometDeltaNativeScan.convert]]
+   * also needs -- threaded through by reference (populated only on the 
claimable path, right
+   * before `declineReason` returns `None`) so a claimed scan does not pay to 
recompute either:
+   * the Hadoop conf 
([[org.apache.spark.sql.internal.SessionState#newHadoopConfWithOptions]] is
+   * not cheap) and the deletion-vector descriptors (base64-decoded, 
non-trivial only for DV-shape
+   * scans). One instance is created per claim attempt in `DeltaScanContrib` 
and passed to both
+   * `declineReason` and `convert`.
+   */
+  private[delta] final class DeltaClaimMemo {
+    var hadoopConf: Configuration = _
+    var dvDescriptors: Seq[DeletionVectorDescriptor] = Seq.empty
+  }
+
+  /**
+   * First reason this Delta scan cannot go native, or None when claimable (in 
which case `memo`
+   * is populated for [[CometDeltaNativeScan.convert]] to reuse). Only called 
when [[isDeltaScan]]
+   * is true. `scanHelper` is the [[CometScanExec]] built to drive `convert` 
on a claim, reused
+   * for the multi-store gate below.
+   */
+  def declineReason(
+      plan: SparkPlan,
+      scanExec: FileSourceScanExec,
+      scanHelper: CometScanExec,
+      memo: DeltaClaimMemo): Option[String] = {
+    val format = 
scanExec.relation.fileFormat.asInstanceOf[DeltaParquetFileFormat]
+    val protocol = format.protocol
+    val metadata = format.metadata
+    // Name mode is supported via physical-name schemas; id mode needs the 
field-id path and
+    // stays declined until validated. Hoisted here since several gates below 
reuse it.
+    val cmMode = metadata.columnMappingMode.name
+    // Descriptor deserialization is expensive, so hoist it into a `lazy val`, 
forced at most
+    // once in this method; on the claimable path the result is handed to 
`convert` through
+    // `memo` below, so a claimed scan deserializes the descriptors exactly 
once end to end.
+    val tableRoot = scanExec.relation.location.rootPaths.head.toString
+    lazy val dvDescriptors: Seq[DeletionVectorDescriptor] =
+      selectedDvDescriptors(scanHelper, tableRoot)
+
+    // Mirrors core's CometScanRule.isSchemaSupported so scan-time type gates 
(unsigned-small-int
+    // fallback, collation, shredded-variant-struct) apply identically here. 
Pure in-memory check,
+    // so it runs first, ahead of every I/O-bearing gate below.
+    val schemaFallbackReasons = new ListBuffer[String]()
+    val typeChecker = CometScanTypeChecker()
+    val requiredSchemaSupported =
+      typeChecker.isSchemaSupported(scanExec.requiredSchema, 
schemaFallbackReasons)
+    val partitionSchemaSupported =
+      typeChecker.isSchemaSupported(scanExec.relation.partitionSchema, 
schemaFallbackReasons)
+    if (!requiredSchemaSupported || !partitionSchemaSupported) {
+      return Some(
+        "Native Delta scan does not support the schema: " + 
schemaFallbackReasons.mkString(", "))
+    }
+
+    if (format.isCDCRead) {
+      return Some("Native Delta scan does not support Change Data Feed reads")
+    }
+
+    // Delta's DML machinery (findTouchedFiles) disables reader optimizations 
and needs real
+    // row indexes from Spark's reader; claiming here would feed NULL indexes 
into DV construction.
+    if (!format.optimizationsEnabled) {
+      return Some("Native Delta scan does not support reads with reader 
optimizations disabled")
+    }
+    if (scanExec.requiredSchema.exists(_.name == 
DeltaParquetFileFormat.ROW_INDEX_COLUMN_NAME) ||
+      scanExec.relation.dataSchema.exists(
+        _.name == DeltaParquetFileFormat.ROW_INDEX_COLUMN_NAME)) {
+      return Some("Native Delta scan does not support Delta's generated 
row-index column")
+    }
+
+    if (cmMode != "none" && cmMode != "name") {
+      return Some(s"Native Delta scan does not support column mapping mode 
$cmMode")
+    }
+    // createPhysicalSchema wholesale-replaces field metadata, silently 
dropping EXISTS_DEFAULT.
+    if (cmMode == "name" &&
+      getExistenceDefaultValues(scanExec.requiredSchema).exists(_ != null)) {
+      return Some(
+        "Native Delta scan does not support column defaults together with 
column mapping")
+    }
+    // createPhysicalSchema rewrites nested StructField names too, and the 
native builder emits the
+    // required schema verbatim as output, so name-sensitive expressions (e.g. 
to_json) would leak
+    // physical names. Decline until a rename adapter exists.
+    if (cmMode == "name" &&
+      scanExec.requiredSchema.exists(f => containsNestedStruct(f.dataType))) {
+      return Some("Native Delta scan does not support column mapping with 
nested struct fields")
+    }
+
+    val readerFeatures = protocol.readerFeatureNames
+    val unknownFeatures = readerFeatures -- understoodReaderFeatures
+    if (unknownFeatures.nonEmpty) {
+      return Some(
+        s"Native Delta scan does not support reader feature(s) 
${unknownFeatures.mkString(", ")}")
+    }
+
+    // Non-constant metadata columns are generated per-row by Spark's reader 
and unsupported,
+    // except Delta's DV bookkeeping columns, which the native path emits as 
constants.
+    val knownColNames =
+      scanExec.relation.dataSchema.map(_.name).toSet ++
+        scanExec.relation.partitionSchema.map(_.name).toSet ++
+        scanExec.fileConstantMetadataColumns.map(_.name).toSet ++
+        CometDeltaNativeScan.internalColumnNames
+    val unknownOutput = 
scanExec.output.map(_.name).filterNot(knownColNames.contains)
+    if (unknownOutput.nonEmpty) {
+      return Some(
+        s"Native Delta scan does not support generated column(s) 
${unknownOutput.mkString(", ")}")
+    }
+
+    // Deletion-vector shape invariants (see 
CometDeltaNativeScan.buildDvScanCommon).
+    if (CometDeltaNativeScan.isDvShape(scanExec)) {
+      // A row-index column WITHOUT is_row_deleted is Delta DML bookkeeping 
(real row indexes),
+      // not a DV read; claiming it with a constant would corrupt the DVs 
being written.
+      val hasIsRowDeleted =
+        scanExec.requiredSchema.exists(_.name == 
CometDeltaNativeScan.IsRowDeletedColumn)
+      val hasRowIndex =
+        scanExec.requiredSchema.exists(_.name == 
CometDeltaNativeScan.RowIndexColumn)
+      if (hasRowIndex && !hasIsRowDeleted) {
+        return Some(
+          "Native Delta scan does not support row-index reads outside a 
deletion-vector scan")
+      }
+      // Internal columns must form a suffix of the read schema so data-column 
positions agree
+      // between Spark's output and the stripped native schema.
+      val names = scanExec.requiredSchema.fields.map(_.name)
+      val firstInternal = 
names.indexWhere(CometDeltaNativeScan.internalColumnNames.contains)
+      if 
(!names.drop(firstInternal).forall(CometDeltaNativeScan.internalColumnNames.contains))
 {
+        return Some("Native Delta scan requires DV bookkeeping columns to 
trail the read schema")
+      }
+      // Native applies the DV itself and emits a dead constant for row-index, 
so the real value
+      // must be provably unused above the scan.
+      if (!rowIndexUnusedAbove(plan, scanExec)) {
+        return Some(
+          "Native Delta scan cannot supply _metadata.row_index values consumed 
by the query")
+      }
+      // The DV common builder does not serialize existence defaults yet.
+      if (getExistenceDefaultValues(scanExec.requiredSchema).exists(_ != 
null)) {
+        return Some(
+          "Native Delta scan does not support column defaults together with 
deletion vectors")
+      }
+      // Bounds native's memory for expanded DV row selectors (delta_dv.rs), 
pessimistically
+      // bounded by 2*cardinality + #row-groups; the conf below makes an 
over-pessimistic decline
+      // recoverable.
+      val maxDeletedRowsPerFile = 
DeltaScanConf.COMET_DELTA_MAX_DELETED_ROWS_PER_FILE.get()
+      val oversizedCardinalities = dvDescriptors
+        .map(_.cardinality)
+        .filter(_ > maxDeletedRowsPerFile)
+      if (oversizedCardinalities.nonEmpty) {
+        return Some(
+          "Native Delta scan does not support a deletion vector deleting " +
+            s"${oversizedCardinalities.max} rows in a single file, exceeding " 
+
+            
s"${DeltaScanConf.COMET_DELTA_MAX_DELETED_ROWS_PER_FILE.key}=$maxDeletedRowsPerFile")
+      }
+    }
+
+    // input_file_name & friends read from a thread-local Spark's FileScanRDD 
sets; the native scan
+    // does not populate it, and Delta's DML find-touched-files scans use it 
(mirrors core's check
+    // in CometScanRule.nativeScan).
+    if (plan.exists(node =>
+        node.expressions.exists(_.exists {
+          case _: InputFileName | _: InputFileBlockStart | _: 
InputFileBlockLength => true
+          case _ => false
+        }))) {
+      return Some(
+        "Native Delta scan is not compatible with input_file_name, " +
+          "input_file_block_start, or input_file_block_length")
+    }
+
+    // Row-index metadata columns are generated per-row by Spark's reader 
(mirrors core); the DV
+    // shape's trailing row-index column is exempt since the gates above 
already proved it dead.
+    if (!CometDeltaNativeScan.isDvShape(scanExec) &&
+      ShimFileFormat.findRowIndexColumnIndexInSchema(scanExec.requiredSchema) 
>= 0) {
+      return Some("Native Delta scan does not support row index generation")
+    }
+
+    // Mirror core's vectorized-reader compatibility gate.
+    if (!SQLConf.get.getConf(SQLConf.PARQUET_VECTORIZED_READER_ENABLED) &&
+      !CometConf.COMET_SCAN_ALLOW_DISABLED_PARQUET_VECTORIZED_READER.get()) {
+      return Some(
+        "Native Delta scan is incompatible with " +
+          s"${SQLConf.PARQUET_VECTORIZED_READER_ENABLED.key}=false")
+    }
+
+    // Decline ALL encrypted-parquet configurations (stricter than core): the 
exec node does not
+    // yet wire the decryption-key broadcast to executors.
+    val hadoopConf = scanExec.relation.sparkSession.sessionState
+      .newHadoopConfWithOptions(scanExec.relation.options)
+    // Populated now (rather than only at the very end) so it is available 
even though several
+    // early-return gates below still lie ahead: cheap to set, and every one 
of those gates
+    // declines the scan anyway, so `memo` is simply never read by `convert` 
in that case.
+    memo.hadoopConf = hadoopConf
+    if (CometParquetUtils.encryptionEnabled(hadoopConf)) {
+      return Some("Native Delta scan does not support encrypted parquet")
+    }
+
+    // Nested-type column defaults cannot be serialized; a dropped default 
would misalign the
+    // value/index lists consumed positionally on the native side. Mirrors 
core's
+    // transformV1Scan gate.
+    val possibleDefaultValues = 
getExistenceDefaultValues(scanExec.requiredSchema)
+    if (possibleDefaultValues.exists(d =>
+        d != null && (d.isInstanceOf[ArrayBasedMapData] || d
+          .isInstanceOf[GenericInternalRow] || 
d.isInstanceOf[GenericArrayData]))) {
+      return Some("Native Delta scan does not support default values for 
nested types")
+    }
+
+    // An opted-in S3-compliant alias scheme (fs.comet.s3Compliant.schemes) is 
declined before the
+    // generic scheme gate below so the reason says why: core's native scan 
reads it through the
+    // S3 client, but the S3 divergence gates further down model Hadoop's 
S3AFileSystem only.
+    val aliasReason =
+      s3CompliantAliasSchemeReason(hadoopConf, 
scanExec.relation.location.rootPaths.map(_.toUri))
+    if (aliasReason.isDefined) {
+      return aliasReason
+    }
+
+    // Only claim scans whose root paths object_store (or the configured 
libhdfs schemes) can
+    // actually read (mirrors core's unsupportedFsSchemes gate).
+    val libhdfs = libhdfsSchemes
+    val unsupportedRootSchemes =
+      unsupportedSchemes(scanExec.relation.location.rootPaths.map(_.toUri), 
libhdfs)
+    if (unsupportedRootSchemes.nonEmpty) {
+      return Some(
+        "Native Delta scan does not support filesystem scheme(s) " +
+          s"${unsupportedRootSchemes.mkString(", ")}")
+    }
+
+    // A shallow clone can span multiple object-store authorities, but the 
native builder resolves
+    // ObjectStoreUrl from only the FIRST selected file; force file listing 
and decline rather than
+    // risk reading a later file through the wrong handle.
+    val dataFileUris =
+      
scanHelper.selectedPartitions.iterator.flatMap(_.files).map(_.getPath.toUri).toSeq
+
+    // Both gates below need the DV absolute-path URIs; dvDescriptors is 
already memoized.
+    val dvUris = dvDescriptors
+      .filter(_.storageType != DeletionVectorDescriptor.INLINE_DV_MARKER)
+      .map(_.absolutePath(new Path(tableRoot)).toUri)
+
+    // The root-path gate above only inspects the table root(s); selected 
files can resolve
+    // through a different scheme (e.g. `viewfs:`). Checked before the 
authority gates below,
+    // which presume every URI is natively resolvable.
+    val unsupportedSelected = unsupportedSelectedSchemeReason(dataFileUris ++ 
dvUris, libhdfs)
+    if (unsupportedSelected.isDefined) {
+      return unsupportedSelected
+    }
+
+    // Checked before multiStoreReason, which presumes every URI resolves to a 
single store
+    // identity -- a userinfo-bearing authority provably does not (store 
keying drops userinfo).
+    val userInfoReason = userInfoBearingAuthorityReason(dataFileUris ++ dvUris)
+    if (userInfoReason.isDefined) {
+      return userInfoReason
+    }
+
+    val multiStore = multiStoreReason(dataFileUris)
+    if (multiStore.isDefined) {
+      return multiStore
+    }
+
+    // GCS's zero-I/O, conf-only credential-forwarding gate; ordered alongside 
the S3 credential
+    // gates below since all presume a single, well-formed store identity per 
URI.
+    val gcsAuthReason = gcsHadoopOnlyAuthReason(hadoopConf, dataFileUris ++ 
dvUris)
+    if (gcsAuthReason.isDefined) {
+      return gcsAuthReason
+    }
+
+    // Zero-I/O, conf-only, like the GCS gate above: decline any bucket 
configured for an
+    // encryption algorithm outside the allowlist (SSE-C, CSE-KMS, CSE-CUSTOM, 
or unknown) before
+    // the credential-divergence gates below, which do not otherwise notice 
this table is readable
+    // through Hadoop only because Hadoop's request factory (SSE-C) or 
SDK-level decryption layer
+    // (CSE-*) does something native never learns about.
+    val encryptionReason =
+      unsupportedEncryptionAlgorithmReason(hadoopConf, dataFileUris ++ dvUris)
+    if (encryptionReason.isDefined) {
+      return encryptionReason
+    }
+
+    // Shared across the two gates below: propagateBucketOptions is a full 
Configuration deep
+    // copy, and both gates would otherwise recompute it independently for the 
same bucket(s)
+    // (once here, then again per-key inside s3ConfigDivergenceReason). One 
cache, populated
+    // lazily per bucket on first use, makes it a single copy total per bucket 
across both gates.
+    val propagatedConfCache = MutableMap.empty[String, Configuration]
+
+    // Always zero-I/O (plain propagated-conf read, no keystore): native's S3 
client has no
+    // HTTP proxy support at all (no fs.s3a.proxy.* key is read anywhere in 
s3.rs), so a bucket
+    // requiring a proxy for S3 egress must decline here rather than claim and 
then connect
+    // directly, bypassing whatever network-segmentation/firewall policy 
required the proxy.
+    val proxyReason = proxyGateReason(hadoopConf, dataFileUris ++ dvUris, 
propagatedConfCache)
+    if (proxyReason.isDefined) {
+      return proxyReason
+    }
+
+    // Zero-I/O, conf-only, like the proxy gate above: Hadoop's 
AssumedRoleCredentialProvider
+    // sends fs.s3a.assumed.role.policy as the session policy of its STS 
AssumeRole request,
+    // while native's assumed-role provider never reads the key -- a claimed 
scan would assume
+    // the role WITHOUT the configured session restriction, silently widening 
permissions.
+    val rolePolicyReason =
+      assumedRolePolicyGateReason(hadoopConf, dataFileUris ++ dvUris, 
propagatedConfCache)
+    if (rolePolicyReason.isDefined) {
+      return rolePolicyReason
+    }
+
+    // Every fs.s3a.* option native's get_config (s3.rs) resolves must agree 
between what Hadoop
+    // itself would use and what native would read from the forwarded, 
substituted conf (covers
+    // long-form bucket credentials, JCEKS/credential-provider shadowing, and 
any other
+    // short-vs-effective divergence in one mechanism); reuses hadoopConf from 
the encryption gate
+    // above.
+    val s3Reason =
+      s3ConfigDivergenceReason(hadoopConf, dataFileUris ++ dvUris, 
propagatedConfCache)
+    if (s3Reason.isDefined) {
+      return s3Reason
+    }
+
+    // A credential-provider class native's 
build_aws_credential_provider_metadata (s3.rs) does
+    // not recognize errors at scan EXECUTION time, after the scan was already 
claimed; decline
+    // eagerly instead.
+    val providerReason = providerClassGateReason(hadoopConf, dataFileUris ++ 
dvUris)
+    if (providerReason.isDefined) {
+      return providerReason
+    }
+
+    // Reuse core's generic native-scan gates 
(ignoreCorruptFiles/ignoreMissingFiles, AQE DPP on
+    // Spark 3.4, exec enabled); tags its own fallback reasons.
+    if (!CometNativeScan.isSupported(scanExec)) {
+      return Some("Core native scan gates rejected the scan (see reasons 
above)")
+    }
+
+    // Claimable: hand the already-forced descriptors to `convert` via `memo` 
so it does not
+    // deserialize them a second time.
+    memo.dvDescriptors = dvDescriptors
+    None
+  }
+
+  /**
+   * Deletion-vector descriptors for every file this DV-shape scan selected, 
normalized to
+   * absolute on-disk paths. Returns `Seq.empty` for the plain shape. Shared 
by the DV cardinality
+   * gate and [[CometDeltaNativeScan.convert]]'s object-store option merge.
+   */
+  private[delta] def selectedDvDescriptors(
+      scanHelper: CometScanExec,
+      tableRoot: String): Seq[DeletionVectorDescriptor] = {
+    if (!CometDeltaNativeScan.isDvShape(scanHelper.wrapped)) {
+      return Seq.empty
+    }
+    val tableRootPath = new Path(tableRoot)
+    scanHelper.selectedPartitions.iterator
+      .flatMap(_.files)
+      .flatMap { file =>
+        file.metadata
+          .get(DeltaParquetFileFormat.FILE_ROW_INDEX_FILTER_ID_ENCODED)
+          .map(enc => 
DeletionVectorDescriptor.deserializeFromBase64(enc.asInstanceOf[String]))
+      }
+      .map(_.copyWithAbsolutePath(tableRootPath))
+      .toSeq
+  }
+
+  /**
+   * The libhdfs scheme exemption set from 
[[org.apache.comet.CometConf.COMET_LIBHDFS_SCHEMES]],
+   * parsed exactly like core's scan gate (`NativeConfig.parseSchemeSet`: 
split on commas,
+   * trimmed, lowercased) and defaulting to `Set("hdfs")` when unset.
+   */
+  private[delta] def libhdfsSchemes: Set[String] = COMET_LIBHDFS_SCHEMES.get() 
match {
+    case Some(s) => NativeConfig.parseSchemeSet(s)
+    case None => Set("hdfs")
+  }
+
+  /**
+   * Decline reason when any of `uris` uses a scheme opted in as an 
S3-compliant alias through
+   * `fs.comet.s3Compliant.schemes` (e.g. `blob`), or `None`. Core's native 
Parquet scan admits
+   * such a scheme and reads it through its S3 client, with `NativeConfig` 
translating the vendor
+   * `fs.<scheme>.<authority>.*` keys into `fs.s3a.bucket.*` options. Spark, 
however, reads the
+   * same table through the vendor's own Hadoop FileSystem, not 
`S3AFileSystem`, and every S3
+   * divergence gate in this object ([[s3ConfigDivergenceReason]] and its 
siblings) is verified
+   * against `S3AFileSystem`'s consumers only. With no model of how the vendor 
filesystem resolves
+   * its configuration, whether native and Spark would agree cannot be 
decided, so the scan is
+   * declined rather than claimed on a guess. Selected data-file and 
deletion-vector URIs under an
+   * alias scheme are declined by the generic scheme gates, which never admit 
an alias (see
+   * [[unsupportedSchemes]]).
+   */
+  private[delta] def s3CompliantAliasSchemeReason(
+      hadoopConf: Configuration,
+      uris: Seq[URI]): Option[String] = {
+    val aliases = NativeConfig.resolveS3CompliantSchemes(hadoopConf)
+    if (aliases.isEmpty) {
+      return None
+    }
+    val found = uris
+      .flatMap(uri => Option(uri.getScheme).map(_.toLowerCase(Locale.ROOT)))
+      .filter(aliases.contains)
+      .distinct
+    if (found.isEmpty) {
+      None
+    } else {
+      Some(
+        "Native Delta scan does not support S3-compliant alias filesystem 
scheme(s) " +
+          s"${found.sorted.mkString(", ")} 
(${CometConf.COMET_S3_COMPLIANT_SCHEMES_KEY}): " +
+          "Spark reads them through a vendor filesystem whose S3 configuration 
resolution the " +
+          "native scan's S3AFileSystem divergence model cannot verify")
+    }
+  }
+
+  /**
+   * The lowercased, deduplicated schemes among `uris` that neither `libhdfs` 
nor Comet's native
+   * object_store layer ([[CometScanRule.isNativelyReadableScheme]]) can read. 
A `null` scheme is
+   * tolerated, not flagged, since such a URI cannot come from a Hadoop-backed 
source. The alias
+   * set handed to core's gate is deliberately empty: an 
`fs.comet.s3Compliant.schemes` alias is
+   * never admitted here (see [[s3CompliantAliasSchemeReason]]), even though 
core admits it.
+   */
+  private[delta] def unsupportedSchemes(uris: Seq[URI], libhdfs: Set[String]): 
Set[String] = {
+    uris
+      .filter { uri =>
+        val sch = uri.getScheme
+        sch != null && {
+          val sl = sch.toLowerCase(Locale.ROOT)
+          !libhdfs.contains(sl) && 
!CometScanRule.isNativelyReadableScheme(uri, Set.empty)

Review Comment:
   ### Correctness
   
   [P2] Preserve the real-path fallback for Delta scans
   
   `isNativelyReadableScheme` now probes a synthetic URL such as `file:///`, so 
this call no longer checks whether object_store accepts the actual path. Core 
added a separate `objectStoreAcceptsPath` root gate, but Delta returns from the 
contrib hook before that gate, and `CometNativeScan.isSupported` does not apply 
it. With native Delta enabled, an otherwise supported local table under a 
newline-containing directory (URI such as `file:///tmp/dir%0A/data`) is claimed 
here; the encoded filename reaches native planning, where object_store 0.13.2 
decodes `%0A` and rejects the control character. Spark's filesystem reader 
would have remained usable.
   
   The newly introduced case is a fresh JVM with a cold file-scheme cache: the 
previous helper tested that actual URI and declined it. A previously warmed 
cache could already hide the old path rejection. Please mirror core's 
actual-root-path fallback, respecting libhdfs exemptions, and add a Delta 
regression asserting Spark fallback and the correct answer. This finding is 
verified from the exact source and locked parser; no full query reproduction 
was executed.
   



-- 
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]

Reply via email to