sunchao commented on code in PR #5526:
URL: https://github.com/apache/datafusion-comet/pull/5526#discussion_r3880413388
##########
spark/src/main/scala/org/apache/comet/codegen/CometBatchKernelCodegen.scala:
##########
@@ -117,7 +126,7 @@ object CometBatchKernelCodegen extends Logging with
CometExprTraitShim with Come
* nested-field count on `spark.sql.codegen.maxFields`.
*/
def canHandle(boundExpr: Expression): Option[String] = {
- if (!isSupportedDataType(boundExpr.dataType)) {
+ if (!isSupportedDataType(boundExpr.dataType, allowNullType = true)) {
Review Comment:
[P2] Handle NullType map values before enabling native broadcast
A required build-side projection such as `map(id, NULL)` over primitive
Parquet input now passes this gate and can enter a normal hinted Comet
broadcast hash join with AQE disabled. `CometBroadcastExchangeExec`
unconditionally calls `Utils.coalesceBroadcastBatches`, which creates an empty
target and appends the first batch. In the pinned Arrow Java 18.3.0, the
map-entries struct appender grows capacity before updating child counts, but
its `NullVector` value child reports capacity equal to its still-zero count and
`reAlloc()` does nothing. The struct therefore never reaches the positive entry
count while its key and validity buffers keep growing until allocation or size
failure. Finalizing the source batch does not fix this separate target. BASE
rejects this output and leaves the intermediate project/broadcast in Spark.
Please guard or fix coalescing before admitting this shape. This is
source-derived, not an executed reproduction.
##########
spark/src/main/scala/org/apache/comet/codegen/CometBatchKernelCodegen.scala:
##########
@@ -81,18 +81,27 @@ object CometBatchKernelCodegen extends Logging with
CometExprTraitShim with Come
/**
* Type surface the kernel covers on both input and output sides. Recursive:
complex types are
* supported when their children are.
+ *
+ * `NullType` is output-only: [[CometBatchKernelCodegenOutput]] can write an
all-null Arrow
+ * `NullVector`, but `CometScalaUDFCodegen.specFor` cannot build an
[[ArrowColumnSpec]] for one,
+ * so a `NullType` input (nested or not) has to keep falling back to Spark.
*/
- def isSupportedDataType(dt: DataType): Boolean = dt match {
+ def isSupportedDataType(dt: DataType): Boolean = isSupportedDataType(dt,
allowNullType = false)
+
+ private def isSupportedDataType(dt: DataType, allowNullType: Boolean):
Boolean = dt match {
+ case NullType => allowNullType
case BooleanType | ByteType | ShortType | IntegerType | LongType => true
case FloatType | DoubleType => true
case _: DecimalType => true
case _: StringType | _: BinaryType => true
case DateType | TimestampType | TimestampNTZType => true
case dt if isTimeType(dt) => true
case _: YearMonthIntervalType | _: DayTimeIntervalType |
CalendarIntervalType => true
- case ArrayType(inner, _) => isSupportedDataType(inner)
- case st: StructType => st.fields.forall(f =>
isSupportedDataType(f.dataType))
- case mt: MapType => isSupportedDataType(mt.keyType) &&
isSupportedDataType(mt.valueType)
+ case ArrayType(inner, _) => isSupportedDataType(inner, allowNullType)
+ case st: StructType => st.fields.forall(f =>
isSupportedDataType(f.dataType, allowNullType))
+ case mt: MapType =>
+ isSupportedDataType(mt.keyType, allowNullType) &&
Review Comment:
[P2] Preserve non-nullable NullType keys through broadcast IPC
Allowing NullType map keys also admits the valid empty result
`transform_values(map(), (k, v) -> id)` over a primitive column when
`spark.sql.legacy.createEmptyCollectionUsingStringType=false`. Retaining that
result on a normal hinted native broadcast build side with AQE disabled exposes
a separate failure. The first generated export preserves the non-nullable key
field, but ordinary native-to-Java reimport uses Arrow 18.3.0's `Types.NULL`
factory, which constructs `NullVector(name)` and changes its field to nullable.
Runtime `getField()` reconstruction and IPC preserve that flag; the broadcast
reader then throws `Map data key type should be a non-nullable` during schema
initialization, before reading any entries. Zero entries do not avoid this
check. The enclosing TransformValues remains nonfoldable on the inspected
maintained Spark 3.5/4.0 paths; BASE rejects its output and retains Spark
execution. Please preserve the key field across reimport/IPC or retain fallback
for this sha
pe. Source-derived, not an executed reproduction.
--
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]
