sunchao commented on code in PR #5365: URL: https://github.com/apache/datafusion-comet/pull/5365#discussion_r3836543179
########## contrib/delta-spark/src/main/scala/org/apache/comet/contrib/delta/CometDeltaNativeScan.scala: ########## @@ -0,0 +1,533 @@ +/* + * 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 scala.jdk.CollectionConverters._ + +import org.apache.hadoop.fs.Path +import org.apache.spark.internal.Logging +import org.apache.spark.sql.catalyst.expressions.Literal +import org.apache.spark.sql.comet.{CometScanExec, DeltaPlanDataInjector} +import org.apache.spark.sql.delta.DeltaParquetFileFormat +import org.apache.spark.sql.delta.RowIndexFilterType +import org.apache.spark.sql.delta.actions.DeletionVectorDescriptor +import org.apache.spark.sql.execution.{FileSourceScanExec, ScalarSubquery => ExecScalarSubquery} +import org.apache.spark.sql.execution.datasources.{FilePartition, PartitionedFile} +import org.apache.spark.sql.types.{ByteType, LongType, MetadataBuilder, StructField, StructType} + +import org.apache.comet.objectstore.NativeConfig +import org.apache.comet.serde.OperatorOuterClass +import org.apache.comet.serde.OperatorOuterClass.Operator +import org.apache.comet.serde.QueryPlanSerde.{exprToProto, serializeDataType} +import org.apache.comet.serde.operator.{literalToProto, partition2Proto, schema2Proto, CometNativeScan} +import org.apache.comet.shims.ShimFileFormat + +/** + * Serde for the native Delta scan. Two shapes: + * - Plain reads reuse core's `NativeScanCommon` builder wholesale. + * - Deletion-vector reads: Delta's planner appends `__delta_internal_is_row_deleted` (tinyint) + * and Spark's row-index temp column (bigint) to the read schema and filters on is_row_deleted + * above the scan. The native reader applies the DV as a row selection, so surviving rows are + * by construction not deleted: both internal columns are emitted as per-file constants (0), + * the parquet read schema is stripped to the real data columns, and the DV descriptor ships + * per file for the native side to fetch and decode. + */ +object CometDeltaNativeScan + extends Logging + with org.apache.spark.sql.catalyst.expressions.PredicateHelper { + + val IsRowDeletedColumn: String = DeltaParquetFileFormat.IS_ROW_DELETED_COLUMN_NAME + val RowIndexColumn: String = ShimFileFormat.ROW_INDEX_TEMPORARY_COLUMN_NAME + + private[delta] val internalColumnNames: Set[String] = Set(IsRowDeletedColumn, RowIndexColumn) + + // Prefix for the internal columns' slots in the partition schema, mirroring core's + // _comet_metadata_ prefix rationale: DataFusion matches partition columns by name. + private val deltaConstFieldPrefix = "_comet_delta_" + + def isDvShape(scanExec: FileSourceScanExec): Boolean = + scanExec.requiredSchema.exists(f => internalColumnNames.contains(f.name)) + + private def deltaFormat(scanExec: FileSourceScanExec): DeltaParquetFileFormat = + scanExec.relation.fileFormat.asInstanceOf[DeltaParquetFileFormat] + + private def columnMappingMode(scanExec: FileSourceScanExec): String = + deltaFormat(scanExec).metadata.columnMappingMode.name + + /** + * Under column mapping, parquet files store physical column names (stable UUIDs / ids), so the + * schemas passed to the native parquet reader must be physical. Positions and structure are + * preserved, so all positional output binding and projection are unaffected. The scan's + * internal DV columns are not part of the table schema and must be stripped before calling + * this. + */ + private def toPhysical(scanExec: FileSourceScanExec, schema: StructType): StructType = { + val format = deltaFormat(scanExec) + if (format.metadata.columnMappingMode.name == "none") { + schema + } else { + // Name mode matches file columns by physical NAME. createPhysicalSchema also stamps + // parquet.field.id metadata, but files written before the column-mapping upgrade have + // no field ids and would fail the reader's id expectations, strip the ids so the + // reader stays purely name-based (id mode, when enabled, will keep them). + stripFieldIds(org.apache.spark.sql.delta.DeltaColumnMapping + .createPhysicalSchema(schema, format.metadata.schema, format.metadata.columnMappingMode)) + } + } + + private def stripFieldIds(schema: StructType): StructType = { + import org.apache.spark.sql.types._ + def stripType(dt: DataType): DataType = dt match { + case s: StructType => stripFieldIds(s) + case a: ArrayType => a.copy(elementType = stripType(a.elementType)) + case m: MapType => + m.copy(keyType = stripType(m.keyType), valueType = stripType(m.valueType)) + case other => other + } + StructType(schema.fields.map { f => + val metadata = new MetadataBuilder() + .withMetadata(f.metadata) + .remove("parquet.field.id") + // Sibling key Delta stamps on array/map fields under IcebergCompat/Uniform. + .remove("parquet.field.nested.ids") + .build() + f.copy(dataType = stripType(f.dataType), metadata = metadata) + }) + } + + /** + * Build the planning-time `DeltaScan` operator (common data only; file partitions are injected + * lazily at execution). Returns None when an output data type cannot be serialized or the plan + * shape is not one we can translate faithfully. + */ + def convert(scanExec: FileSourceScanExec, scanHelper: CometScanExec): Option[Operator] = { + val relation = scanExec.relation + + val firstFileUri = scanHelper.selectedPartitions + .flatMap(_.files.headOption) + .headOption + .map(_.getPath.toUri) + + val hadoopConf = relation.sparkSession.sessionState + .newHadoopConfWithOptions(relation.options) + + val tableRootPath = relation.location.rootPaths.head + val tableRoot = tableRootPath.toString + + val commonOpt = if (!isDvShape(scanExec)) { + // Under column mapping (name mode) the parquet reader must see physical names; + // positions are preserved so output binding and projection stay untouched. + CometNativeScan.buildNativeScanCommon( + source = scanExec.simpleStringWithNodeId(), + output = scanExec.output, + requiredSchema = toPhysical(scanExec, scanExec.requiredSchema), + dataSchema = toPhysical(scanExec, relation.dataSchema), + partitionSchema = relation.partitionSchema, + fileConstantMetadataColumns = scanExec.fileConstantMetadataColumns, + dataFilters = scanHelper.supportedDataFilters, + firstFileUri = firstFileUri, + hadoopConf = hadoopConf, + conf = scanExec.conf) + } else { + buildDvScanCommon(scanExec, scanHelper, firstFileUri, hadoopConf) + } + + commonOpt.map { commonBuilder => + // Union object-store options over every authority a partition of this scan may need a + // store for, not just the first data file's scheme (finding 8). + val dvDescriptors = DeltaScanSupport.selectedDvDescriptors(scanHelper, tableRoot) + commonBuilder.putAllObjectStoreOptions( + mergedObjectStoreOptions( + hadoopConf, + storeUris(dvDescriptors, tableRootPath, firstFileUri)).asJava) Review Comment: **[P2] Also check Hadoop's long bucket credential aliases** The new fallback catches the original global and short bucket aliases, but one case remains at `92ae71b8`. Hadoop's `S3AUtils.lookupPassword` also accepts these long bucket aliases, which are absent from [`s3aCredentialAliases`](https://github.com/apache/datafusion-comet/blob/92ae71b8f7bb9631dd151c0e3df522a36754b9ae/contrib/delta-spark/src/main/scala/org/apache/comet/contrib/delta/DeltaScanSupport.scala#L567-L574): ```text fs.s3a.bucket.mybucket.fs.s3a.access.key fs.s3a.bucket.mybucket.fs.s3a.secret.key ``` With only those aliases in a globally configured JCEKS and `SimpleAWSCredentialsProvider` explicitly selected, Hadoop resolves the credentials while the current `credentialAliasReason` returns `None`. `NativeConfig` forwards none of the keystore entries, and the native static provider only checks short bucket aliases and globals. A fresh native scan therefore still has no credentials. If different global plain credentials are also configured, Hadoop uses the long aliases but native uses those different globals instead. I reproduced both cases with the unchanged current admission/extraction methods, Hadoop 3.4.1, and a real local JCEKS. The short bucket alias control correctly declines. The native consequence was checked in source, not through a full JNI scan or live S3 request. Could we account for the long aliases when deciding whether native credentials preserve Hadoop's resolution, and add long-only and global-shadowing regressions? A conservative fallback is sufficient here. This is a remaining case of the existing P2, not a separate finding. -- 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]
