comphead commented on code in PR #7248:
URL: https://github.com/apache/datafusion-comet/pull/7248#discussion_r4391175690
##########
spark/src/test/resources/sql-tests/expressions/array/flatten.sql:
##########
@@ -34,3 +34,47 @@ SELECT flatten(arr) FROM test_flatten
-- literal arguments
query
SELECT flatten(array(array(1, 2), array(3, 4)))
+
+-- binary elements
+statement
+CREATE TABLE test_flatten_binary(arr array<array<binary>>) USING parquet
+
+statement
+INSERT INTO test_flatten_binary VALUES
+ (array(array(X'01', X'0203'), array(X'', X'04'))),
+ (array(array(), array(X'05'))),
+ (array()),
+ (NULL),
+ (array(array(X'06', NULL), array(NULL))),
+ (array(array(X'07'), CAST(NULL AS ARRAY<BINARY>))),
+ (array(CAST(NULL AS ARRAY<BINARY>), CAST(NULL AS ARRAY<BINARY>)))
+
+query
+SELECT flatten(arr) FROM test_flatten_binary
+
+-- flatten of binary elements as an aggregate input
+query
+SELECT count(flatten(arr)), count(DISTINCT flatten(arr)) FROM
test_flatten_binary
+
+-- nested flatten of binary elements, filtered on an element and used under a
distinct aggregate
+statement
+CREATE TABLE test_flatten_binary3(k string, id int, arr
array<array<array<binary>>>) USING parquet
+
+statement
+INSERT INTO test_flatten_binary3 VALUES
+ ('a', 1, array(array(array(X'30313233343536373839414243', X'01')),
array(array(X'02')))),
+ ('a', 2, array(array(array(X'30313233343536373839414243')),
array(array(X'03')))),
+ ('a', 2, array(array(array(X'3031')), array(array(X'03')))),
+ ('b', 3, array(array(array(X'30313233343536373839414243')))),
+ ('b', 4, array(array(array(CAST(NULL AS BINARY))))),
+ ('b', 5, NULL)
+
+query
+SELECT k, COUNT(DISTINCT id) FROM test_flatten_binary3
+WHERE length(flatten(flatten(arr))[0]) = 13 GROUP BY k
+
+query
+SELECT COUNT(1) FROM test_flatten_binary3 WHERE
length(flatten(flatten(arr))[0]) != 13
+
+query
+SELECT k, COUNT(1) FROM test_flatten_binary3 WHERE
length(flatten(flatten(arr))[0]) = 13 GROUP BY 1
Review Comment:
The gate now accepts `BinaryType` but still rejects struct and map elements,
and the second commit makes the reason name `flatten`, yet no fixture pins
either. I could not find an `expect_fallback` for `flatten` anywhere under
`spark/src/test`. Could a `query expect_fallback(flatten: ...)` over
`array<array<struct<a:int>>>` be added? It would pin that struct elements still
fall back and that the reason names the function. I have not checked that the
substring match works with the prefix, though other fixtures match on
substrings of reasons.
##########
spark/src/main/scala/org/apache/comet/CometSparkSessionExtensions.scala:
##########
@@ -334,24 +339,65 @@ object CometSparkSessionExtensions extends Logging {
* @return
* `node` with fallback reasons attached (as a side effect on its tag map).
*/
- def withFallbackReasons[T <: TreeNode[_]](node: T, info: Set[String]): T = {
+ def withFallbackReasons[T <: TreeNode[_]](node: T, info: Set[String]): T =
+ recordFallbackReasons(node, info, qualifyWithExpressionName = true)
+
+ /**
+ * Record reasons that were already recorded on another expression, so they
already name the
+ * expression that rejected the input. Qualifying them again would attribute
them to the
+ * expression they are lifted onto. See `QueryPlanSerde.liftFallbackReasons`.
+ */
+ private[comet] def withLiftedFallbackReasons[T <: TreeNode[_]](node: T,
info: Set[String]): T =
+ recordFallbackReasons(node, info, qualifyWithExpressionName = false)
+
+ private def recordFallbackReasons[T <: TreeNode[_]](
+ node: T,
+ info: Set[String],
+ qualifyWithExpressionName: Boolean): T = {
if (CometConf.COMET_EXPLAIN_FALLBACK_LOG_ENABLED.get()) {
for (reason <- info) {
logWarning(s"Comet cannot accelerate ${node.getClass.getSimpleName}
because: $reason")
}
}
+ val named = node match {
+ case expr: Expression if qualifyWithExpressionName =>
qualifyReasons(expr, info)
+ case _ => info
+ }
+ def previous =
+
node.getTagValue(CometExplainInfo.FALLBACK_REASONS).getOrElse(Set.empty[String])
val reasons = node match {
case expr: Expression =>
- CometExplainInfo.recordFallbackReasons(expr, info).getOrElse {
-
expr.getTagValue(CometExplainInfo.FALLBACK_REASONS).getOrElse(Set.empty[String])
++ info
- }
- case _ =>
-
node.getTagValue(CometExplainInfo.FALLBACK_REASONS).getOrElse(Set.empty[String])
++ info
+ CometExplainInfo.recordFallbackReasons(expr, named).getOrElse(previous
++ named)
+ case _ => previous ++ named
}
node.setTagValue(CometExplainInfo.FALLBACK_REASONS, reasons)
node
}
+ /**
+ * Name the expression that recorded each reason. Reasons are lifted onto
the enclosing operator
+ * as a flat set of strings, so a reason such as "data type not supported:
..." would not say
+ * which function rejected the type. Leaves, aliases and attributes are left
alone: their
+ * reasons describe a type or a config, and their names would add nothing.
So are reasons that
+ * already identify their subject, either by mentioning the expression's
name or by naming a
+ * `spark.comet.` config key (the key names the expression to enable, which
need not be the one
+ * that recorded the reason).
+ */
+ private def qualifyReasons(expr: Expression, reasons: Set[String]):
Set[String] = expr match {
+ case _: Attribute | _: Alias | _: Literal => reasons
+ case _ =>
+ val name = CometExplainInfo.exprDisplayName(expr match {
+ case agg: AggregateExpression => agg.aggregateFunction
+ case other => other
+ })
+ lazy val namePattern =
+ Pattern.compile(s"\\b${Pattern.quote(name)}\\b",
Pattern.CASE_INSENSITIVE)
+ reasons.map { reason =>
+ if (reason.contains("spark.comet.") ||
namePattern.matcher(reason).find()) reason
Review Comment:
This skips the prefix only when the reason contains `exprDisplayName`, but
several existing reasons name the expression by its Scala class name:
`unsupported arguments for ArrayInsert`, `CreateArray`, `ElementAt` and
`ArraysZip` in `arrays.scala`, and `CreateNamedStruct` and
`GetArrayStructFields` in `structs.scala`. They would now render as
`array_insert: unsupported arguments for ArrayInsert`, and `CreateArray` as
`array: ...`, since `\barray\b` does not match inside `CreateArray`.
`GetArrayItem` matches only by luck. Could the already-named check also accept
`expr.getClass.getSimpleName`, or could those messages drop to just the cause?
I read this from the code and Spark's `prettyName` values and did not run an
explain, and there may be more such reasons outside `arrays.scala` and
`structs.scala`.
--
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]
