dwsmith1983 commented on code in PR #5365: URL: https://github.com/apache/datafusion-comet/pull/5365#discussion_r3826283650
########## contrib/delta-spark/src/main/scala/org/apache/comet/contrib/delta/CometDeltaNativeScan.scala: ########## @@ -0,0 +1,450 @@ +/* + * 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.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 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 => + val common = commonBuilder.build() + val tableRoot = relation.location.rootPaths.head.toString + val deltaCommon = OperatorOuterClass.DeltaSparkScanCommon + .newBuilder() + .setTableRoot(tableRoot) + .setColumnMappingMode(columnMappingMode(scanExec)) + .setSourceKey(DeltaPlanDataInjector.sourceKey(tableRoot, common)) + .build() + val deltaScan = OperatorOuterClass.DeltaSparkScan + .newBuilder() + .setCommon(common) + .setDeltaCommon(deltaCommon) + Operator + .newBuilder() + .setPlanId(scanExec.id) + .setContribScan(DeltaSparkScanEnvelope.pack(deltaScan.build())) + .build() + } + } + + /** + * Harvest subquery-bearing predicates for this scan from its covering FilterExec. Spark 3.x + * strips them from a scan's `dataFilters` at planning (`FileSourceStrategy` routes them to the + * post-scan filter only), while Spark 4.x keeps them in `dataFilters`. Collecting them here at + * claim time gives the execution-time resolve-and-push path the same inputs on every Spark + * version; the dedup keeps Spark 4.x from carrying duplicates. + */ + def subqueryFiltersFromParent( + plan: org.apache.spark.sql.execution.SparkPlan, + scanExec: FileSourceScanExec): Seq[org.apache.spark.sql.catalyst.expressions.Expression] = { + import org.apache.spark.sql.catalyst.expressions.{PlanExpression, SubqueryExpression} + // Nearest FilterExec above the scan (the DV shape interposes nodes between them, so do + // not require a direct parent-child edge). Safety comes from the reference guard, not the + // plan walk: only conjuncts expressed directly over the scan's own output attributes + // survive, so filters from other branches or over aliased projections contribute nothing. + val filtersAboveScan = plan.collect { + case f: org.apache.spark.sql.execution.FilterExec if f.find(_ eq scanExec).isDefined => f + } + filtersAboveScan.lastOption + .map { f => + splitConjunctivePredicates(f.condition) + .filter(_.references.subsetOf(scanExec.outputSet)) Review Comment: Fixed. The harvest now walks the plan from the covering filter down to the scan and only accepts Project and Filter nodes in between, anything else stops it, plus a deterministic check on the conjuncts. Your exact query shape is a regression test now: empty result, scan still native, and an assertion that nothing was pushed. The covering filter is never removed, so this only prevents evaluating the predicate too early. -- 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]
