dwsmith1983 commented on code in PR #5365:
URL: https://github.com/apache/datafusion-comet/pull/5365#discussion_r3920753757
##########
spark/src/main/scala/org/apache/comet/serde/operator/CometNativeScan.scala:
##########
@@ -125,166 +150,233 @@ object CometNativeScan extends
CometOperatorSerde[CometScanExec] with CometTypeS
scan: CometScanExec,
builder: Operator.Builder,
childOp: OperatorOuterClass.Operator*):
Option[OperatorOuterClass.Operator] = {
- val nativeScanBuilder = OperatorOuterClass.NativeScan.newBuilder()
+ // Extract object store options from first file (S3 configs apply to all
files in scan).
+ // Use selectedPartitions (static) instead of getFilePartitions() because
at planning time
+ // DPP subqueries haven't been resolved yet. Object store options don't
depend on DPP.
+ val firstFileUri = scan.selectedPartitions
+ .flatMap(_.files.headOption)
+ .headOption
+ .map(_.getPath.toUri)
+
+ // Collect S3/cloud storage configurations
+ val hadoopConf = scan.relation.sparkSession.sessionState
+ .newHadoopConfWithOptions(scan.relation.options)
+
+ buildNativeScanCommon(
+ source = scan.simpleStringWithNodeId(),
+ output = scan.output,
+ requiredSchema = scan.requiredSchema,
+ dataSchema = scan.relation.dataSchema,
+ partitionSchema = scan.relation.partitionSchema,
+ fileConstantMetadataColumns = scan.wrapped.fileConstantMetadataColumns,
+ dataFilters = scan.supportedDataFilters,
+ firstFileUri = firstFileUri,
+ hadoopConf = hadoopConf,
+ conf = scan.conf) match {
+ case Some(commonBuilder) =>
+ // Sink operators don't have children
+ builder.clearChildren()
+ val nativeScanBuilder = OperatorOuterClass.NativeScan.newBuilder()
+ // Set common data in NativeScan (file_partition will be populated at
execution time)
+ nativeScanBuilder.setCommon(commonBuilder.build())
+ Some(builder.setNativeScan(nativeScanBuilder).build())
+ case None =>
+ // There are unsupported scan type
+ withFallbackReason(
+ scan,
+ s"unsupported Comet operator: ${scan.nodeName}, due to unsupported
data types above")
+ None
+ }
+ }
+
+ /**
+ * Build the `NativeScanCommon` proto shared by the core parquet scan and
contrib scans that
+ * delegate to the same native parquet machinery (e.g. a Delta scan contrib,
which passes
+ * physical-name schemas under column mapping). Returns `None` when an
output data type cannot
+ * be serialized; the caller is responsible for tagging a fallback reason.
+ *
+ * Visibility note: `private[comet]` means a contrib caller must live under
an
+ * `org.apache.comet.*` package (the same constraint `PlanDataInjector`
implementers have).
+ */
+ private[comet] def buildNativeScanCommon(
+ source: String,
+ output: Seq[Attribute],
+ requiredSchema: StructType,
+ dataSchema: StructType,
+ partitionSchema: StructType,
+ fileConstantMetadataColumns: Seq[AttributeReference],
+ dataFilters: Seq[Expression],
+ firstFileUri: Option[URI],
+ hadoopConf: Configuration,
+ conf: SQLConf): Option[OperatorOuterClass.NativeScanCommon.Builder] = {
val commonBuilder = OperatorOuterClass.NativeScanCommon.newBuilder()
// Set source in common (used as part of injection key)
- commonBuilder.setSource(scan.simpleStringWithNodeId())
+ commonBuilder.setSource(source)
- val scanTypes = scan.output.flatten { attr =>
+ val scanTypes = output.flatten { attr =>
serializeDataType(attr.dataType)
}
- if (scanTypes.length == scan.output.length) {
- commonBuilder.addAllFields(scanTypes.asJava)
-
- // Sink operators don't have children
- builder.clearChildren()
-
- if (scan.conf.getConf(SQLConf.PARQUET_FILTER_PUSHDOWN_ENABLED)) {
- val dataFilters = new ListBuffer[Expr]()
- for (filter <- scan.supportedDataFilters) {
- exprToProto(filter, scan.output) match {
- case Some(proto) => dataFilters += proto
- case _ =>
- logWarning(s"Unsupported data filter $filter")
- }
+ if (scanTypes.length != output.length) {
+ // There are unsupported scan types
+ return None
+ }
+ commonBuilder.addAllFields(scanTypes.asJava)
+
+ if (conf.getConf(SQLConf.PARQUET_FILTER_PUSHDOWN_ENABLED)) {
+ val filterProtos = new ListBuffer[Expr]()
+ for (filter <- dataFilters) {
+ exprToProto(filter, output) match {
+ case Some(proto) => filterProtos += proto
+ case _ =>
+ logWarning(s"Unsupported data filter $filter")
}
- commonBuilder.addAllDataFilters(dataFilters.asJava)
- }
-
- val possibleDefaultValues =
getExistenceDefaultValues(scan.requiredSchema)
- if (possibleDefaultValues.exists(_ != null)) {
- // Our schema has default values. Serialize two lists, one with the
default values
- // and another with the indexes in the schema so the native side can
map missing
- // columns to these default values.
- val (defaultValues, indexes) =
possibleDefaultValues.iterator.zipWithIndex
- .filter { case (expr, _) => expr != null }
- .map { case (expr, index) =>
- // ResolveDefaultColumnsUtil.getExistenceDefaultValues has
evaluated these
- // expressions and they should now just be literals.
- (Literal(expr), index.toLong.asInstanceOf[java.lang.Long])
- }
- .toList
- .unzip
- commonBuilder.addAllDefaultValues(
- defaultValues.flatMap(exprToProto(_, scan.output)).asJava)
- commonBuilder.addAllDefaultValuesIndexes(indexes.asJava)
}
+ commonBuilder.addAllDataFilters(filterProtos.asJava)
+ }
- // Extract object store options from first file (S3 configs apply to all
files in scan).
- // Use selectedPartitions (static) instead of getFilePartitions()
because at planning time
- // DPP subqueries haven't been resolved yet. Object store options don't
depend on DPP.
- val firstFileUri = scan.selectedPartitions
- .flatMap(_.files.headOption)
- .headOption
- .map(_.getPath.toUri)
-
- // Constant metadata columns (file_path, file_name, file_size,
file_block_start,
- // file_block_length, file_modification_time) are known before opening
the file and
- // constant for every row read from it, exactly like partition columns.
Spark places
- // them immediately after partition columns in `scan.output`
- // (FileSourceStrategy.scala: readDataColumns ++
generatedMetadataColumns ++
- // partitionColumns ++ constantMetadataColumns), so appending them after
the real
- // partition schema here keeps the two in lockstep.
- val constantMetadataFields =
scan.wrapped.fileConstantMetadataColumns.map(attr =>
- StructField(s"$constantMetadataFieldPrefix${attr.name}",
attr.dataType, attr.nullable))
- val partitionSchemaFields = scan.relation.partitionSchema.fields.toSeq ++
- constantMetadataFields
- val partitionSchema = schema2Proto(partitionSchemaFields)
- val requiredSchema = schema2Proto(scan.requiredSchema)
-
- // Spark's required schema can prune a Variant column, including a
Variant nested under an
- // unrequested struct. The complete relation schema still contains that
unsupported type,
- // and serializing it would throw even though the native reader never
needs those bytes.
- // Keep ordinary fields unchanged and replace a requested
Variant-bearing root with its
- // already-validated, pruned required field. A requested actual Variant
never reaches this
- // point because CometScanRule keeps those scans on Spark.
- val nativeDataSchema =
StructType(scan.relation.dataSchema.fields.flatMap { field =>
- if (containsVariantType(field.dataType)) {
- scan.requiredSchema.fields.find(requiredField =>
- scan.conf.resolver(requiredField.name, field.name))
- } else {
- Some(field)
- }
- })
- val dataSchema = schema2Proto(nativeDataSchema)
-
- val dataSchemaIndexes = scan.requiredSchema.map(field => {
- nativeDataSchema.fieldIndex(field.name)
- })
- val partitionSchemaIndexes = nativeDataSchema.fields.length until
- (nativeDataSchema.length + partitionSchemaFields.length)
-
- val projectionVector = (dataSchemaIndexes ++
partitionSchemaIndexes).map(idx =>
- idx.toLong.asInstanceOf[java.lang.Long])
-
- commonBuilder.addAllProjectionVector(projectionVector.asJava)
-
- // In `CometScanRule`, we ensure partitionSchema (including constant
metadata columns)
- // is supported.
- assert(partitionSchema.length == partitionSchemaFields.length)
-
- commonBuilder.addAllDataSchema(dataSchema.asJava)
- commonBuilder.addAllRequiredSchema(requiredSchema.asJava)
- commonBuilder.addAllPartitionSchema(partitionSchema.asJava)
-
commonBuilder.setSessionTimezone(scan.conf.getConfString("spark.sql.session.timeZone"))
-
commonBuilder.setCaseSensitive(scan.conf.getConf[Boolean](SQLConf.CASE_SENSITIVE))
-
- // SPARK-53535 (Spark 4.1+): when reading a struct whose requested
fields are all
- // missing in the Parquet file, the new default preserves the parent
struct's
- // nullness from the file (so non-null parents materialize as a struct
of all-null
- // fields). Pre-4.1 Spark hardcodes the legacy behavior (whole struct
null), which
- // matches the Comet default we use as fallback.
- val returnNullStructConfKey =
- "spark.sql.legacy.parquet.returnNullStructIfAllFieldsMissing"
- val returnNullStructDefault = if (isSpark41Plus) "false" else "true"
- commonBuilder.setReturnNullStructIfAllFieldsMissing(
- scan.conf.getConfString(returnNullStructConfKey,
returnNullStructDefault).toBoolean)
-
- // Field-ID matching: only ask the native side to do extra work when the
conf is on AND
- // the requested schema actually carries IDs. Spark's ParquetReadSupport
applies the same
- // gate before invoking matchIdField.
- val useFieldId =
- scan.conf.getConf(SQLConf.PARQUET_FIELD_ID_READ_ENABLED) &&
- ParquetUtils.hasFieldIds(scan.requiredSchema)
- commonBuilder.setUseFieldId(useFieldId)
- commonBuilder.setIgnoreMissingFieldId(
- scan.conf.getConf(SQLConf.IGNORE_MISSING_PARQUET_FIELD_ID))
-
-
commonBuilder.setAllowTypePromotion(CometConf.COMET_SCHEMA_EVOLUTION_ENABLED)
-
commonBuilder.setAllowTimestampLtzToNtz(CometConf.COMET_ALLOW_TIMESTAMP_LTZ_AS_NTZ)
-
- // Collect S3/cloud storage configurations
- val hadoopConf = scan.relation.sparkSession.sessionState
- .newHadoopConfWithOptions(scan.relation.options)
-
-
commonBuilder.setEncryptionEnabled(CometParquetUtils.encryptionEnabled(hadoopConf))
-
- firstFileUri.foreach { uri =>
- val objectStoreOptions =
- NativeConfig.extractObjectStoreOptions(hadoopConf, uri)
- objectStoreOptions.foreach { case (key, value) =>
- commonBuilder.putObjectStoreOptions(key, value)
+ val possibleDefaultValues = getExistenceDefaultValues(requiredSchema)
+ if (possibleDefaultValues.exists(_ != null)) {
+ // Our schema has default values. Serialize two lists, one with the
default values
+ // and another with the indexes in the schema so the native side can map
missing
+ // columns to these default values.
+ val (defaultValues, indexes) =
possibleDefaultValues.iterator.zipWithIndex
+ .filter { case (expr, _) => expr != null }
+ .map { case (expr, index) =>
+ // ResolveDefaultColumnsUtil.getExistenceDefaultValues has evaluated
these
+ // expressions and they should now just be literals.
+ (Literal(expr), index.toLong.asInstanceOf[java.lang.Long])
}
+ .toList
+ .unzip
+ commonBuilder.addAllDefaultValues(defaultValues.flatMap(exprToProto(_,
output)).asJava)
+ commonBuilder.addAllDefaultValuesIndexes(indexes.asJava)
+ }
+
+ // Constant metadata columns (file_path, file_name, file_size,
file_block_start,
+ // file_block_length, file_modification_time) are known before opening the
file and
+ // constant for every row read from it, exactly like partition columns.
Spark places
+ // them immediately after partition columns in the scan output
+ // (FileSourceStrategy.scala: readDataColumns ++ generatedMetadataColumns
++
+ // partitionColumns ++ constantMetadataColumns), so appending them after
the real
+ // partition schema here keeps the two in lockstep.
+ val constantMetadataFields = uniqueConstantMetadataFields(
+ fileConstantMetadataColumns,
+ dataSchema.fields.map(_.name).toSet ++
partitionSchema.fields.map(_.name).toSet)
+ val partitionSchemaFields = partitionSchema.fields.toSeq ++
constantMetadataFields
+ val partitionSchemaProto = schema2Proto(partitionSchemaFields)
+ val requiredSchemaProto = schema2Proto(requiredSchema)
+
+ // Spark's required schema can prune a Variant column, including a Variant
nested under an
+ // unrequested struct. The complete relation schema still contains that
unsupported type,
+ // and serializing it would throw even though the native reader never
needs those bytes.
+ // Keep ordinary fields unchanged and replace a requested Variant-bearing
root with its
+ // already-validated, pruned required field. A requested actual Variant
never reaches this
+ // point because CometScanRule keeps those scans on Spark.
+ val nativeDataSchema = StructType(dataSchema.fields.flatMap { field =>
+ if (containsVariantType(field.dataType)) {
+ requiredSchema.fields.find(requiredField =>
conf.resolver(requiredField.name, field.name))
+ } else {
+ Some(field)
}
+ })
+ val dataSchemaProto = schema2Proto(nativeDataSchema)
- // Set common data in NativeScan (file_partition will be populated at
execution time)
- nativeScanBuilder.setCommon(commonBuilder.build())
+ val dataSchemaIndexes = requiredSchema.map(field => {
+ nativeDataSchema.fieldIndex(field.name)
+ })
+ val partitionSchemaIndexes = nativeDataSchema.fields.length until
+ (nativeDataSchema.length + partitionSchemaFields.length)
- Some(builder.setNativeScan(nativeScanBuilder).build())
+ val projectionVector = (dataSchemaIndexes ++
partitionSchemaIndexes).map(idx =>
+ idx.toLong.asInstanceOf[java.lang.Long])
- } else {
- // There are unsupported scan type
- withFallbackReason(
- scan,
- s"unsupported Comet operator: ${scan.nodeName}, due to unsupported
data types above")
- None
+ commonBuilder.addAllProjectionVector(projectionVector.asJava)
+
+ // In `CometScanRule`, we ensure partitionSchema (including constant
metadata columns)
+ // is supported.
+ assert(partitionSchemaProto.length == partitionSchemaFields.length)
+
+ commonBuilder.addAllDataSchema(dataSchemaProto.asJava)
+ commonBuilder.addAllRequiredSchema(requiredSchemaProto.asJava)
+ commonBuilder.addAllPartitionSchema(partitionSchemaProto.asJava)
+
+ populateScanConfFlags(commonBuilder, requiredSchema, firstFileUri,
hadoopConf, conf)
+
+ Some(commonBuilder)
+ }
+
+ /**
+ * Populate the configuration-derived flags of a `NativeScanCommon`: session
timezone, case
+ * sensitivity, struct-nullness legacy flag, field-ID matching, type
promotion, encryption, and
+ * object-store options. Shared with contrib scans that assemble their own
schemas/projection
+ * (e.g. the Delta contrib's deletion-vector shape) so new flags added here
reach them without
+ * drift.
+ */
+ private[comet] def populateScanConfFlags(
+ commonBuilder: OperatorOuterClass.NativeScanCommon.Builder,
+ requiredSchema: StructType,
+ firstFileUri: Option[URI],
+ hadoopConf: Configuration,
+ conf: SQLConf): Unit = {
+
commonBuilder.setSessionTimezone(conf.getConfString("spark.sql.session.timeZone"))
+ val caseSensitive = conf.getConf[Boolean](SQLConf.CASE_SENSITIVE)
+ commonBuilder.setCaseSensitive(caseSensitive)
+ if (!caseSensitive) {
+ // Ship THIS JVM's case data so native's case-insensitive footer
matching reproduces
+ // this JVM's `toLowerCase(Locale.ROOT)` exactly, whatever Unicode
version it bundles.
+ JvmCaseTables.populate(commonBuilder)
Review Comment:
Gone with the case tables: nothing rides in NativeScanCommon now, so plans
are back to their interned size and QueryContextInternerSuite passes on the
default configuration again with no pin. Good catch on the pin masking that,
the suite is untouched from main on this branch.
--
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]
