[SPARK-1225, 1241] [MLLIB] Add AreaUnderCurve and BinaryClassificationMetrics - #364

…nMetrics

This PR implements a generic version of `AreaUnderCurve` using the `RDD.sliding` implementation from #136 . It also contains refactoring of #160 for binary classification evaluation.

Author: Xiangrui Meng <[email protected]>

Closes #364 from mengxr/auc and squashes the following commits:

a05941d [Xiangrui Meng] replace TP/FP/TN/FN by their full names
3f42e98 [Xiangrui Meng] add (0, 0), (1, 1) to roc, and (0, 1) to pr
fb4b6d2 [Xiangrui Meng] rename Evaluator to Metrics and add more metrics
b1b7dab [Xiangrui Meng] fix code styles
9dc3518 [Xiangrui Meng] add tests for BinaryClassificationEvaluator
ca31da5 [Xiangrui Meng] remove PredictionAndResponse
3d71525 [Xiangrui Meng] move binary evalution classes to evaluation.binary
8f78958 [Xiangrui Meng] add PredictionAndResponse
dda82d5 [Xiangrui Meng] add confusion matrix
aa7e278 [Xiangrui Meng] add initial version of binary classification evaluator
221ebce [Xiangrui Meng] add a new test to sliding
a920865 [Xiangrui Meng] Merge branch 'sliding' into auc
a9b250a [Xiangrui Meng] move sliding to mllib
cab9a52 [Xiangrui Meng] use last for the last element
db6cb30 [Xiangrui Meng] remove unnecessary toSeq
9916202 [Xiangrui Meng] change RDD.sliding return type to RDD[Seq[T]]
284d991 [Xiangrui Meng] change SlidedRDD to SlidingRDD
c1c6c22 [Xiangrui Meng] add AreaUnderCurve
65461b2 [Xiangrui Meng] Merge branch 'sliding' into auc
5ee6001 [Xiangrui Meng] add TODO
d2a600d [Xiangrui Meng] add sliding to rdd

(cherry picked from commit f5ace8d)
Signed-off-by: Matei Zaharia <[email protected]>

…nMetrics

This PR implements a generic version of `AreaUnderCurve` using the `RDD.sliding` implementation from apache#136 . It also contains refactoring of apache#160 for binary classification evaluation.

Author: Xiangrui Meng <[email protected]>

Closes apache#364 from mengxr/auc and squashes the following commits:

a05941d [Xiangrui Meng] replace TP/FP/TN/FN by their full names
3f42e98 [Xiangrui Meng] add (0, 0), (1, 1) to roc, and (0, 1) to pr
fb4b6d2 [Xiangrui Meng] rename Evaluator to Metrics and add more metrics
b1b7dab [Xiangrui Meng] fix code styles
9dc3518 [Xiangrui Meng] add tests for BinaryClassificationEvaluator
ca31da5 [Xiangrui Meng] remove PredictionAndResponse
3d71525 [Xiangrui Meng] move binary evalution classes to evaluation.binary
8f78958 [Xiangrui Meng] add PredictionAndResponse
dda82d5 [Xiangrui Meng] add confusion matrix
aa7e278 [Xiangrui Meng] add initial version of binary classification evaluator
221ebce [Xiangrui Meng] add a new test to sliding
a920865 [Xiangrui Meng] Merge branch 'sliding' into auc
a9b250a [Xiangrui Meng] move sliding to mllib
cab9a52 [Xiangrui Meng] use last for the last element
db6cb30 [Xiangrui Meng] remove unnecessary toSeq
9916202 [Xiangrui Meng] change RDD.sliding return type to RDD[Seq[T]]
284d991 [Xiangrui Meng] change SlidedRDD to SlidingRDD
c1c6c22 [Xiangrui Meng] add AreaUnderCurve
65461b2 [Xiangrui Meng] Merge branch 'sliding' into auc
5ee6001 [Xiangrui Meng] add TODO
d2a600d [Xiangrui Meng] add sliding to rdd

* Adding PySpark Submit functionality. Launching Python from JVM

* Addressing scala idioms related to PR351

* Removing extends Logging which was necessary for LogInfo

* Refactored code to leverage the ContainerLocalizedFileResolver

* Modified Unit tests so that they would pass

* Modified Unit Test input to pass Unit Tests

* Setup working environent for integration tests for PySpark

* Comment out Python thread logic until Jenkins has python in Python

* Modifying PythonExec to pass on Jenkins

* Modifying python exec

* Added unit tests to ClientV2 and refactored to include pyspark submission resources

* Modified unit test check

* Scalastyle

* PR 348 file conflicts

* Refactored unit tests and styles

* further scala stylzing and logic

* Modified unit tests to be more specific towards Class in question

* Removed space delimiting for methods

* Submission client redesign to use a step-based builder pattern.

This change overhauls the underlying architecture of the submission
client, but it is intended to entirely preserve existing behavior of
Spark applications. Therefore users will find this to be an invisible
change.

The philosophy behind this design is to reconsider the breakdown of the
submission process. It operates off the abstraction of "submission
steps", which are transformation functions that take the previous state
of the driver and return the new state of the driver. The driver's state
includes its Spark configurations and the Kubernetes resources that will
be used to deploy it.

Such a refactor moves away from a features-first API design, which
considers different containers to serve a set of features. The previous
design, for example, had a container files resolver API object that
returned different resolutions of the dependencies added by the user.
However, it was up to the main Client to know how to intelligently
invoke all of those APIs. Therefore the API surface area of the file
resolver became untenably large and it was not intuitive of how it was
to be used or extended.

This design changes the encapsulation layout; every module is now
responsible for changing the driver specification directly. An
orchestrator builds the correct chain of steps and hands it to the
client, which then calls it verbatim. The main client then makes any
final modifications that put the different pieces of the driver
together, particularly to attach the driver container itself to the pod
and to apply the Spark configuration as command-line arguments.

* Don't add the init-container step if all URIs are local.

* Python arguments patch + tests + docs

* Revert "Python arguments patch + tests + docs"

This reverts commit 4533df2.

* Revert "Don't add the init-container step if all URIs are local."

This reverts commit e103225.

* Revert "Submission client redesign to use a step-based builder pattern."

This reverts commit 5499f6d.

* style changes

* space for styling

* Adding PySpark Submit functionality. Launching Python from JVM

* Addressing scala idioms related to PR351

* Removing extends Logging which was necessary for LogInfo

* Refactored code to leverage the ContainerLocalizedFileResolver

* Modified Unit tests so that they would pass

* Modified Unit Test input to pass Unit Tests

* Setup working environent for integration tests for PySpark

* Comment out Python thread logic until Jenkins has python in Python

* Modifying PythonExec to pass on Jenkins

* Modifying python exec

* Added unit tests to ClientV2 and refactored to include pyspark submission resources

* Modified unit test check

* Scalastyle

* PR 348 file conflicts

* Refactored unit tests and styles

* further scala stylzing and logic

* Modified unit tests to be more specific towards Class in question

* Removed space delimiting for methods

* Submission client redesign to use a step-based builder pattern.

This change overhauls the underlying architecture of the submission
client, but it is intended to entirely preserve existing behavior of
Spark applications. Therefore users will find this to be an invisible
change.

The philosophy behind this design is to reconsider the breakdown of the
submission process. It operates off the abstraction of "submission
steps", which are transformation functions that take the previous state
of the driver and return the new state of the driver. The driver's state
includes its Spark configurations and the Kubernetes resources that will
be used to deploy it.

Such a refactor moves away from a features-first API design, which
considers different containers to serve a set of features. The previous
design, for example, had a container files resolver API object that
returned different resolutions of the dependencies added by the user.
However, it was up to the main Client to know how to intelligently
invoke all of those APIs. Therefore the API surface area of the file
resolver became untenably large and it was not intuitive of how it was
to be used or extended.

This design changes the encapsulation layout; every module is now
responsible for changing the driver specification directly. An
orchestrator builds the correct chain of steps and hands it to the
client, which then calls it verbatim. The main client then makes any
final modifications that put the different pieces of the driver
together, particularly to attach the driver container itself to the pod
and to apply the Spark configuration as command-line arguments.

* Don't add the init-container step if all URIs are local.

* Python arguments patch + tests + docs

* Revert "Python arguments patch + tests + docs"

This reverts commit 4533df2.

* Revert "Don't add the init-container step if all URIs are local."

This reverts commit e103225.

* Revert "Submission client redesign to use a step-based builder pattern."

This reverts commit 5499f6d.

* style changes

* space for styling

Add document links into README

…d children

The previous child.resolved guard was insufficient: when building SQL table
function plans (makeSQLTableFunctionPlan), Cast nodes are constructed over an
OuterReference wrapping an unresolved attribute. OuterReference is a leaf and
reports resolved == true, yet OuterReference.dataType delegates to the
unresolved attribute and throws UnresolvedException, breaking CREATE FUNCTION
... RETURNS TABLE (e.g. sql-udf.sql query apache#364).

Key the CURRENT_LIKE pattern off the always-available target dataType (tag every
cast to the NTZ family) instead of inspecting child.dataType. ComputeCurrentTime
still applies the precise TIME-source check on the resolved plan, so only
TIME -> TIMESTAMP_NTZ casts are rewritten; other NTZ-target casts are left
untouched.

The merge of master re-added VarkaRangeFilterBoundarySuite, which apache#364 introduced and design B replaces with VarkaRangeFilterFusionSuite; its 48-and-49 boundary no longer exists.

Generated-by: Claude Code (Claude Opus 5.5)

### What changes were proposed in this pull request?

Task 172, design B: Varka takes TPC-DS `modified-q3`'s filter, 200 date ranges joined by `or`, which today it declines past 48 ranges and which vanilla runs with its scan loop in the interpreter.

*This PR is stacked on apache#364 (task 172's step 1), whose commits it includes until apache#364 merges. This description covers only the two commits after it.*

**The mechanism.** Written as the tree of comparisons the query spells, a disjunction of ranges is one condition, and the emitter puts one condition into one method. Every range adds about 165 bytes to it, and at 49 ranges it passes the 8000-byte budget, so Varka declines it at plan time. Design B gives the shape its own node, and code that doesn't grow with the ranges:
- **`InRanges(child, bounds)`, a new condition node.** It's true where some range `[lo, hi]` contains the value, false where none does, and unknown where the value is null, the same three-valued answer the `or` chain gives. The bounds are part of the node, like `GuardedRange`'s, so two range sets never share a kernel. The constructor requires them sorted, disjoint and non-adjacent, and the node is int-lane only.
- **The compiler recognises the shape.** A disjunction of two or more ranges over one int or date column with literal bounds becomes one range set. The ranges can be written as `>=`/`<=` in either operand order, as strict bounds (which move by one), or as equalities. The compiler sorts them, merges those that overlap or touch, and drops empty ones, so one set of rows is one shape however the query ordered it. Anything else compiles as the comparisons it's written as.
- **The kernel loops over a table.** Each range set's bounds live in a `private static final int[]` that the class initialiser fills, the first emitted class with a field and a `<clinit>`. The node emits one loop that ORs, range by range, the lanes with `lo <= v && v <= hi` into a mask, comparing against each bound as a scalar. The loop's code is the same size whatever the number of ranges; its time grows with them.

**Why a loop and not the plan's binary search.** A vector binary search needs a gather per step, whose indices must be spilled from the vector into an `int[]`. The loop needs neither. The bounds are not literal slots either: the literal table is keyed by value, so it can't hold a contiguous block of bounds, and every slot is copied into a local at each method's entry, about 4 KB a method at 400 bounds. The binary search stays possible as a variant to measure against the loop. There is no emit option: step 1's committed baseline is the "off" arm. `PLAN_TASK_172.md` 9.2 records both departures.

**What checks it:**
- `VarkaRangeSetCompilerSuite` pins the matching: sorting, merging, strict bounds, equalities, operand order, empty ranges, the int extremes, date literals, and five shapes that are not a set.
- `VarkaRangeSetSuite` runs such filters on the row engine and on Varka over the same Arrow-cached rows, nulls and the int extremes included. It requires the same answer, with the kernel having run, for `modified-q3`'s 200 ranges among others.
- `VarkaRangeFilterFusionSuite` replaces step 1's boundary suite. It pins that every number of ranges fuses, and that the 200-range filter is one range set whose kernel methods are all under 2000 bytes.
- The IR fuzzer draws range sets against a reference evaluator written from the node's meaning. The reach test, the lane-type table and the range-analysis walks all know the node.
- The bytes oracle moved in exactly the twelve keys that hash every fuzz shape, and no coverage key.

**Measured** (`PLAN_TASK_172.md` 9.3; laptop, both widths, `VarkaRangeFilterBenchmark` regenerated with the vanilla arm beside it), in ns a row at the wide width:

| ranges | vanilla | Varka, range set | vanilla / Varka |
|---:|---:|---:|---:|
| 10 | 19.3 | 8.9 | 2.2 |
| 49 | 26.9 | 12.6 | 2.1 |
| 100 | 3736.6 | 19.5 | 192 |
| 200 | 6781.7 | 36.6 | 185 |

- **Varka now runs its kernel at every rung.** It's about twice as fast as vanilla where both compile, and **185 times faster at the query's 200 ranges**, where vanilla's scan loop runs in the interpreter.
- At 128 bits, the 200-range ratio is 102.
- **The loop's time grows with the ranges,** about 0.16 ns a row per range, so the plan's prediction 3 ("barely moves with the ranges") failed, as expected. Prediction 4 held for design B.

### Why are the changes needed?

The milestone's realistic query is the one TPC-DS stage past the method-size cliff, where vanilla runs about 135 times slower than just below it. Varka could only show the cliff there, not take the query.

### Does this PR introduce _any_ user-facing change?

No.

### How was this patch tested?

Every Varka suite in catalyst and sql passes: 409 and 431 tests, none failed. So do the new suites above. `dev/scalastyle`, `dev/lint-java`, `dev/varka_quote_check.py` and the 100-column and non-ASCII scans pass.

### Was this patch authored or co-authored using generative AI tooling?

Generated-by: Claude Code (Claude Opus 5.5)

### What changes were proposed in this pull request?

Runner figures for tasks 192 and 172, all taken on GitHub-hosted runners from master.

**Task 192: the runner files re-measured on the Arrow cache** (`PLAN_TASK_192.md` 9.5). The runner files carried the fault apache#364 fixed: they measured vanilla over Spark's default cache, not the Arrow cache their tables name. Both runs are repeated from master `0ea8414fef5`:
- **The defaults run drew the EPYC 9V45**, the same machine as task 171's runner ladder. Its defaults arm reads 892.1 ns a row at 52 entries, against the ladder's 892.8, so the two files now measure the same vanilla.
- **On that one machine and cache, the best tuned vanilla (`hugeMethodLimit=8000`) is 16 times slower than Varka at 54 entries and 19 at a hundred.** That replaces the withdrawn 16 and 22.
- **The flag run drew an EPYC 7763.** The flag's second cliff reproduces there: compiled through 54 entries, 17522.9 ns a row at a hundred. That file is read only against itself.

**Task 172: design B on a runner** (`PLAN_TASK_172.md` 9.4). `VarkaRangeFilterBenchmark` from master `efe3017de8a` drew an EPYC 9V74, Zen 4 without the full-width datapath, so the file is named for it.
- The laptop's shape holds, steeper: vanilla steps about 180 times between 49 and 100 ranges.
- **At the query's 200 ranges Varka is 250 times faster** (17520.9 against 70.2 ns a row), and about 2.2 to 2.6 times faster below vanilla's crossing.

The published figure is to come from the EPYC 9V45, and two more dispatches are out for it. Rows 192 and 172 are updated.

### Why are the changes needed?

Headline figures come from runners a reader can dispatch, and task 192's runner files had been measured on the wrong input path.

### Does this PR introduce _any_ user-facing change?

No.

### How was this patch tested?

The files are the workflow's artifacts, unchanged, with provenance naming each run and CPU. `dev/varka_quote_check.py` passes.

### Was this patch authored or co-authored using generative AI tooling?

Generated-by: Claude Code (Claude Opus 5.5)