fix(#74): exclude CodegenFallback children from TimedWithCodegenExec stage

[minskya]

Fixes #74.

Summary

from_json (and any other CodegenFallback expression that reads ctx.INPUT_ROW) NPE'd in Block.code interpolation when DataFlint's TimedWithCodegenExec wrapped the operator that contained it under whole-stage codegen.

java.lang.NullPointerException: Cannot invoke "Object.getClass()" because "arg" is null
    at org.apache.spark.sql.errors.QueryExecutionErrors$.cannotInterpolateClassIntoCodeBlockError
    at org.apache.spark.sql.catalyst.expressions.codegen.Block$BlockHelper$.$anonfun$code$1
    ...
    at org.apache.spark.sql.catalyst.expressions.codegen.CodegenFallback.doGenCode
    at org.apache.spark.sql.catalyst.expressions.JsonToStructs.doGenCode

Root cause

Spark's CollapseCodegenStages already excludes plans whose expressions contain a CodegenFallback. But TimedWithCodegenExec exposes children = child.children (transparent wrapping for the SQL UI), which hides the wrapped child's expressions from that check — so an operator that would normally be broken out of whole-stage codegen ends up inside it, behind the wrapper, and the code\"…\" interpolator NPEs on the null INPUT_ROW.

The first attempt (the original commit on this branch) tried to plumb row.value through doConsume. That approach didn't actually prevent the NPE — Spark's consume() still resets ctx.INPUT_ROW = null before calling the parent's doConsume — and it also broke prepareRowVar's projection logic (corrupt counts, occasionally invalid Java in the generated method signature). It's been reverted on this branch.

The fix

Mirror the same CodegenFallback check that CollapseCodegenStages.supportCodegen already does, inside TimedWithCodegenExec.supportCodegen:

override def supportCodegen: Boolean = {
  val c = child.asInstanceOf[CodegenSupport]
  c.supportCodegen &&
    (TimedExec.isLegacySpark || child.children.length <= 1) &&
    !child.expressions.exists(_.find(_.isInstanceOf[CodegenFallback]).isDefined)
}

When the wrapped child has any CodegenFallback expression, we report supportCodegen = false. CollapseCodegenStages then inserts an InputAdapter between us and the next stage, breaking the whole-stage at the boundary — exactly the behavior end users get without the wrapper. The expression goes through the interpreted path, where INPUT_ROW doesn't matter.

TreeNode.find (not .exists) is used because Expression.exists was added in Spark 3.2; calling it at runtime on Spark 3.0/3.1 NoSuchMethodErrors. find has been on TreeNode since the earliest 3.x lines.

Test coverage

• DataFlintCodegenFallbackSpec — new regression test that runs the exact PySpark snippet from the issue (from_json → explode → filter) under whole-stage codegen and asserts the count matches. Without the fix this NPEs; with it, the wrapped ProjectExec is correctly excluded from the WholeStage and the query runs.
• Test runs against both Spark 3.5 and Spark 4.0.1 — pluginspark4 previously had no test setup at all; this PR adds one that shares version-portable specs from pluginspark3 so the same regression suite executes on the Spark 4 surface.
• DataFlintWriteMetricsSpec — guard for the executeCollect rebuild path: confirms DataWritingCommandExec metrics (numFiles, numOutputBytes, numOutputRows, etc.) survive the rebuild via the shared cmd instance.
• TimedExecMetricsSpec — covers the rddId / canEqual fixes below.

Drive-by hardening on TimedExec

While reviewing the wrapper for adjacent issues, fixed five small things that the original CI / runtime exposed:

1. postRddId accumulated: was +=, now set — re-executing the same TimedExec instance no longer sums every RDD id it ever wrapped.
2. canEqual collapsed variants: TimedExec(x).equals(TimedWithCodegenExec(x)) was true; they have different execution semantics so canonicalization could conflate them. Now matches by exact runtime class.
3. isLegacySpark parser: wrapped in Try — vendor versions with non-numeric major/minor parts no longer throw at class-load time.
4. executeCollect write path: child.children.head / innerChild.children.head replaced with headOption fall-through to super.executeCollect() if the assumed shape doesn't hold.
5. rddId metric type: was a "size" metric (rendered as bytes — \"12 B\" in the SparkUI), now a plain sum metric.

CI

CI was running on Java 8. Spark 4.0.1 needs Java 17+, so adding pluginspark4 tests caused the forked test JVM to hang at startup — every run timed out at 6h. Bumped to temurin@17. Spark 3.5 also supports Java 17, so pluginspark3 tests run on the same JVM unchanged. The published jars stay Java 8 compatible — verified via javap on the produced classes (major version 52 across all four module/Scala variants); Scala 2.12/2.13 default to Java 8 bytecode regardless of which JVM compiles them.

Test plan

• DataFlintCodegenFallbackSpec reproduces the original NPE without the fix and passes with it
• sbt +test green on Spark 3.5 (Scala 2.12 + 2.13) and Spark 4.0.1 (Scala 2.13)
• pluginspark4 runs DataFlintCodegenFallbackSpec against Spark 4.0.1 / Java 17
• PySpark cross-version harness (utils/.spark-versions/3.1.2) — parquet write no longer NoSuchMethodErrors

🤖 Generated with Claude Code

[CLAassistant]

Thank you for your submission! We really appreciate it. Like many open source projects, we ask that you all sign our Contributor License Agreement before we can accept your contribution.
1 out of 2 committers have signed the CLA.

✅ minskya
❌ claude
_{You have signed the CLA already but the status is still pending? Let us recheck it.}