andygrove commented on code in PR #5692:
URL: https://github.com/apache/datafusion-comet/pull/5692#discussion_r3938192745
##########
spark/src/main/scala/org/apache/comet/serde/statics.scala:
##########
@@ -78,16 +95,38 @@ object CometStaticInvoke extends
CometExpressionSerde[StaticInvoke] {
case Some(handler) =>
handler.convert(expr, inputs, binding)
case None =>
- // Every Iceberg system function is named `invoke`, so name the
declaring class too.
- withFallbackReason(
- expr,
- s"Static invoke expression: ${expr.functionName} is not supported " +
- s"(declared on ${expr.staticObject.getName})")
- None
+ // Nothing in the allowlist covers this lowering, so run Spark's own
implementation inside
+ // the Comet pipeline rather than failing the whole operator back to
Spark.
+ // `StaticInvoke.doGenCode` emits a static method call, so the kernel
matches Spark by
+ // construction. Spark 4.x keeps lowering more `RuntimeReplaceable`
functions this way
+ // (`encode`, `is_valid_utf8`, the `TIME` family, ...) and `lpad` /
`rpad` on binary has
+ // lowered to `StaticInvoke(ByteArray, ...)` since Spark 3.4.
+ //
+ // The encoder and deserializer trees that make up most `StaticInvoke`
usage in typed
+ // Dataset operations are unaffected: their arguments are
`ObjectType`, which
+ // `CometBatchKernelCodegen.isSupportedDataType` rejects, so the
dispatcher declines them
+ // and they fall back exactly as before.
+ CometStaticInvokeCodegenDispatch.convert(expr, inputs, binding).orElse
{
Review Comment:
Added `CometCodegenDispatchBenchmark` in 9f8fade and ran it.
Three arms per case: codegen dispatch,
`spark.comet.exec.scalaUDF.codegen.enabled=false` (which is exactly the
operator fallback that shipped before this PR), and Comet off as a reference.
Same corpus for all three, at 1024 rows (sub-batch) and 1,048,576 rows (~128
batches). All three arms are run and their rows compared before anything is
timed, and every table repeats its baseline at the end so the noise floor sits
next to the arms rather than in a footnote.
Spark 4.1.3, OpenJDK 17.0.10, Scala 2.13.17, `local[1]`, Apple M3 Max on
macOS 26.6.2, `spark.comet.batchSize=8192`, whole-stage codegen on. Warmup is
Spark's `Benchmark` default: 2s untimed per case, then at least two iterations
and at least 2s of timed iterations. Best times in ms at 1,048,576 rows:
| case | dispatch off (pre-PR) | codegen dispatch | Comet off | baseline
repeated |
| --- | --- | --- | --- | --- |
| `lpad(binary)` | 51 | 61 | 71 | 53 |
| `rpad(binary)` | 64 | 61 | 79 | 67 |
| `encode(utf-8)` | 79 | 82 | 99 | 79 |
| `to_binary(utf-8)` | 76 | 79 | 97 | 76 |
| `to_time(fmt)` | 261 | 261 | 274 | 260 |
| mixed projection | 137 | 101 | 226 | 141 |
| group by dispatch | 73 | 65 | 61 | 73 |
The last column is the same work as the first, so ±4ms is noise here.
Where the projection is nothing but the one dispatched call, dispatch runs
between parity and about 20% slower than letting the projection fall back.
`lpad` is the worst of them at 51 → 61ms; `rpad` and `to_time` are parity;
`encode` and `to_binary` are ~4% slower. That is the bridge cost you were
asking about and it is real — for a bare `SELECT f(col) FROM t` there is
nothing else in the projection for staying in Comet to save, so the transport
and the Arrow output are pure addition.
It goes the other way as soon as the plan has anything else in it. The mixed
projection is `length(c_str), c_long + 1, substring(c_str, 1, 4), lpad(c_bin,
24, c_pad)`, and dispatch is 1.4x there (137 → 101ms) because the three native
expressions no longer follow the fourth out to Spark. The grouped case is
`count(*)` over `lpad(c_pad, 8, c_pad)`, 1.1x (73 → 65ms), because the partial
aggregate, the exchange and the final aggregate all stay native instead of
following the projection. Comet-off is nominally best on that last row, but its
average is 84ms against dispatch's 69ms with a 44ms stdev, so I would not read
the 61 as a real win.
At 1024 rows nothing is resolvable. Every Comet arm lands in 4–11ms, fixed
job cost dominates, and the baseline's own repeat moves by as much as the arms
differ from each other. I would not read a direction into that table in either
direction.
First use is separated out into its own table (1024 rows, taken before
anything else has executed these queries, and after warming the dispatcher on
an unrelated expression so that what is left is mostly the one kernel). 1st run
/ 2nd run / difference, in ms: `lpad` 85 / 29 / 56, `rpad` 44 / 27 / 17,
`encode` 55 / 29 / 26, `to_binary` 33 / 23 / 10, `to_time` 69 / 25 / 44, mixed
103 / 39 / 64, group by 214 / 38 / 175. The dispatcher reported exactly one
compile in each first run. These are upper bounds rather than compile times:
the grouped case's 175ms is mostly the first shuffle in that JVM, and
`to_binary` looks cheap only because `encode` ran before it and the two compile
to the same source, which `CodeGenerator` caches JVM-wide. Net of that it is
tens of ms per distinct kernel source per JVM, paid once and amortised by the
per-task cache after.
Two things about how the benchmark is built, because a case that quietly
measures the wrong thing is worse than no case. Each table asserts before
timing that the dispatch arm is fully Comet native, that the dispatch-off arm
is not, and that the dispatcher actually ran, and writes a warning into the
results file if any of that fails. The first two shapes I tried for "the
operator above stays native" — a filter, and an ungrouped `sum` — both tripped
it, because I had written `length()` over binary and that has no native Comet
path, so the operator was falling back for a reason that has nothing to do with
this PR. Both would have printed a clean 1.0x.
Where that leaves me: the shapes that lose are the ones where falling back
was cheapest to begin with, and they lose single digits to 20% on a query that
is mostly scan; the shapes that win are the ones where falling back was
expensive. I think that is the right trade, and I would rather not gate the
catch-all on something like "only dispatch when the projection has other
convertible expressions" until a real query regresses, since that gate would
also have to be right about what is downstream of the projection and getting it
wrong costs more than the 10ms.
--
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]
