andygrove commented on code in PR #4720:
URL: https://github.com/apache/datafusion-comet/pull/4720#discussion_r3604073509
##########
spark/src/main/scala/org/apache/spark/sql/comet/operators.scala:
##########
@@ -1643,6 +1643,21 @@ trait CometBaseAggregate {
}
}
+ // CollectList/CollectSet declare their buffer as BinaryType in Spark but
produce a native
Review Comment:
Dropped. You are right, the general missingCometProducer +
aggsNotSupportingMixedExecution guard just above already returns None for the
same case, so this block was unreachable.
##########
spark/src/main/scala/org/apache/comet/rules/CometExecRule.scala:
##########
@@ -871,6 +871,24 @@ case class CometExecRule(session: SparkSession)
}
}
}
+
+ // CollectList/CollectSet round-trip an ArrayType buffer that Spark
declares as BinaryType.
Review Comment:
Added a comment on the block explaining why it is separate: the tagging
block above only fires when the Final cannot convert, but
canAggregateBeConverted skips the child-native check, so an all-native distinct
collect_list Final converts and slips past. This block catches that case.
##########
spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala:
##########
@@ -439,6 +440,22 @@ object QueryPlanSerde extends Logging with CometExprShim
with CometTypeShim {
}
}
+ /**
+ * Returns true if any aggregate function produces a native intermediate
buffer whose Arrow type
+ * (e.g. ArrayType for CollectList/CollectSet) differs from the BinaryType
that Spark declares
Review Comment:
Narrowed the doc comment to describe what the code actually matches
(CollectList/CollectSet) and noted that Percentile has the same buffer shape
but is intentionally not matched here since it already passes through the
general mixed-execution guard.
##########
spark/src/main/spark-4.2/org/apache/comet/shims/CometCollectShim.scala:
##########
@@ -0,0 +1,32 @@
+/*
+ * 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.shims
+
+import org.apache.spark.sql.catalyst.expressions.aggregate.{CollectList,
CollectSet}
+
+/**
+ * Shim for the `ignoreNulls` flag on `CollectList` / `CollectSet`. Spark 4.2
added the field
+ * (`ignoreNulls = false` via `RESPECT NULLS`), so this shim reports the
actual value. The serde
+ * falls back to Spark when it is `false`, since the native path always drops
nulls.
+ */
+object CometCollectShim {
Review Comment:
Moved the identical spark-3.4/spark-3.5 copies to spark-3.x. spark-4.x is
used by spark-4.0/4.1/4.2 too, and 4.2 needs the different shim, so I could not
consolidate 4.0/4.1 there — happy to add a fresh shared dir if you would prefer
that, but it seemed like more build machinery than the reduction was worth.
##########
spark/src/test/resources/sql-tests/expressions/aggregate/collect_list.sql:
##########
@@ -0,0 +1,405 @@
+-- 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.
+
+-- ConfigMatrix: parquet.enable.dictionary=false,true
+
+-- collect_list result order is non-deterministic across partitions, so
+-- every query wraps the result in sort_array to make comparisons stable.
+
+-- ============================================================
+-- Setup: tables
+-- ============================================================
+
+statement
+CREATE TABLE cl_src_int(i int, grp string) USING parquet
+
+statement
+INSERT INTO cl_src_int VALUES
+ (1, 'a'), (2, 'a'), (1, 'a'), (3, 'a'),
+ (4, 'b'), (4, 'b'), (NULL, 'b'), (5, 'b'),
+ (NULL, 'c'), (NULL, 'c')
+
+statement
+CREATE TABLE cl_src_nulls(val int, grp string) USING parquet
+
+statement
+INSERT INTO cl_src_nulls VALUES
Review Comment:
Added `collect_list FILTER (WHERE i > 1)` per-group and a map-input case
(`collect_list(m)` on a `map<string, int>` table with two entries and a null
row). The map assertion is a `spark_answer_only` size-only check since
sort_array cannot order a MapType.
--
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]
