Run SubgraphSampler on Dataproc 2.2 / Spark 3.5 before image 2.0 becomes uncreatable

[kmontemayor2-sc]

 ## Why now

 Google blocks cluster creation on Dataproc image 2.0 on **2026-08-25**:

 > "On August 25, 2026, 1.x and 2.0 image versions will become unavailable for use in creating clusters"

 `spark_job_manager.py:111` hardcodes `2.0.47-ubuntu18` for the `use_spark35=False` path, and
 `cora_nalp_test` / `cora_snc_test` / `cora_udl_test` all take it. On that date those three e2e tests
 stop being able to create a cluster. So does any V1 SubgraphSampler user without `graph_db_config`,
 so this is not CI-only.

 No newer 2.0.x helps — the whole minor is gated, and `2.0.160` was the final release. 2.1 is already
 out of support and goes uncreatable 2026-12-31. That leaves 2.2 (Spark 3.5.3, Scala 2.12.18), which
 matches the `scala_spark35` tree.

 ## Why it isn't a one-line image bump

 Image 2.2 requires the `scala_spark35` jar, and its `TaskRunner` refused the pureSpark path outright:

 ```scala
 throw new Exception("PureSpark SGS not supported in spark35 yet.")
 ```

 The pureSpark task classes were already ported and compiling in that tree, just never wired up — and
 never executed, so nothing had validated them.

 ## What changed

 - **`TaskRunner`** — replaced the `throw` with the dispatch ported from the spark31 tree, UDL
   detection included.
 - **`SGSPureSparkV1Task`** — restored `sampleWithReplacement`, which the spark35 copy had dropped.
   It gates 3 tests and cost ~25 lines to bring back.
 - **`subgraph_sampler.py`** — `use_spark35` now defaults to `True`, so the cora tests move off image
   2.0. The `use_spark35_runner` experimental flag remains as an escape hatch; the dead 2.0 branch can
   be deleted after 2026-08-25.
 - **Nested field names in the proto casts** — see below.
 - **Tests** — migrated the 4 pureSpark suites from `scala/` into `scala_spark35/`, plus 2 new tests
   covering the `castTo*` + `.as[proto]` boundary.

 `spark_job_manager.py` has no net change: a bump to `2.2.85` is reverted in this branch. On 2.2.85 the
 Spark job dies at startup with `NoClassDefFoundError: io/grpc/Context`, and `dblp_nalp` — already on
 the spark35 jar and 2.2.19 before this branch — fails too, so the sub-minor is the variable rather
 than the ported code. 2.2.19 is on a supported minor and escaping 2.0 does not require the newest
 sub-minor. Worth bumping separately once the classpath change is understood.

 ## The nested-field bug, and why Spark 3.1 never saw it

 The `castTo*ProtoSchema` SQL renames top-level and array *columns* to proto names but left the fields
 *inside* array-of-struct elements at their internal `_`-prefixed names:

 ```sql
 struct(_neighbor_nodes AS nodes, _neighbor_edges AS edges) AS neighborhood
 --     ^ column renamed          ^ but elements keep _node_id, _condensed_node_type, ...
 ```

 This was always wrong; Spark 3.1 just couldn't see it. The difference is the scalapb-spark artifact,
 not Spark itself:

 | Tree | Artifact | Repeated-field deserializer |
 | --- | --- | --- |
 | `scala/` | `sparksql31-scalapb0_11` 1.0.0 | resolved `MapObjects` → **positional** access; names irrelevant |
 | `scala_spark35/` | `sparksql35-scalapb0_11` 1.0.4 | `UnresolvedMapObjects` → analyzer resolves **by name** |

 Under 1.0.0 the field order happened to match the proto, so output was correct by accident. Under
 1.0.4 the same SQL raises:

 ```
 [FIELD_NOT_FOUND] No such struct field `node_id` in `_node_id`, `_condensed_node_type`, `_feature_values`
 ```

 Downgrading is not an option — only 1.0.4 and 1.0.5 exist for the sparksql35 artifact.

 The fix renames nested fields via SQL `transform`, in three shared helpers on the base class so the
 five call sites cannot drift apart. Two related traps handled at the same time:

 - Bare `ARRAY()` is `ARRAY<VOID>`, which fails the same deserializer with
   `INVALID_EXTRACT_BASE_FIELD_TYPE`. Empty arrays are now
   `CAST(ARRAY() AS ARRAY<STRUCT<...>>)`.
 - `F.coalesce(col, F.array())` is left alone: `coalesce` widens `array<void>` against its sibling
   operand, so it never reaches the encoder as VOID.

 ## Why the existing tests missed it

 No pureSpark test crossed the `castTo*` + `.as[proto]` boundary — the closest stopped at
 `createSubgraph`. The two new tests build inputs with the production loaders against the checked-in
 TFRecord assets, because the hand-written mocks give `_node_features` a scalar `Double` where real
 data has `array<float>`. Both tests fail on the unfixed code with the production error.

 ## Verification

 Local:

 - `sbt "subgraph_sampler/test"` — 27 tests, 0 failures (25 before, +2 new)
 - `make unit_test_scala` (both trees), `make check_format_scala`, `make type_check`,
   `make unit_test_py PY_TEST_FILES="subgraph_sampler_test.py"` — all pass

 Real Dataproc 2.2, run 31068555994 — all four V1 pipelines succeeded end to end:

 | Pipeline | SubgraphSampler | Path |
 | --- | --- | --- |
 | `cora_nalp` | 7m28s | ported pureSpark |
 | `cora_snc` | 12m04s | ported pureSpark |
 | `cora_udl` | 7m28s | ported pureSpark |
 | `dblp_nalp` | 13m04s | GraphDB (no-regression control) |

 SGS runs faster than the ~28-30 min Spark 3.1 baseline, so this does not add timeout pressure.

 That run's `integration-e2e-test` is red for an unrelated reason: `hom_cora_sup` (a GLT pipeline that
 touches neither Dataproc nor Scala SGS) failed in `data-preprocessor` on a `tensorflow_transform` GCS
 rename returning 404 for its own temp object. It failed in the previous run too, at a different stage.
 The e2e runner raises on the first failed pipeline, so one flake masks the four successes.

 ## Not in this PR

 - **`scala/` (spark31) is untouched.** It is correct-by-accident under 1.0.0's positional semantics,
   and the tree is retired.
 - **No heterogeneous support** in the ported `TaskRunner`. This is a port, not a feature.
 - **`appendIsolatedNodesToTrainingSamples`** has a 6-vs-7-column UNION arity bug, left alone —
   unreachable, `includeIsolatedNodesInTrainingSamples` is hardcoded `false`.

[kmontemayor2-sc]

/all_test

[github-actions]

GiGL Automation

@ 17:12:42UTC : 🔄 Python Unit Test started.

@ 18:29:45UTC : ✅ Workflow completed successfully.

[github-actions]

GiGL Automation

@ 17:12:42UTC : 🔄 E2E Test started.

[github-actions]

GiGL Automation

@ 17:12:42UTC : 🔄 C++ Unit Test started.

@ 17:14:38UTC : ✅ Workflow completed successfully.

[github-actions]

GiGL Automation

@ 17:12:43UTC : 🔄 Scala Unit Test started.

@ 17:23:46UTC : ✅ Workflow completed successfully.

[github-actions]

GiGL Automation

@ 17:12:44UTC : 🔄 Integration Test started.

@ 18:44:12UTC : ✅ Workflow completed successfully.

[github-actions]

GiGL Automation

@ 17:12:44UTC : 🔄 Lint Test started.

@ 17:22:21UTC : ✅ Workflow completed successfully.

[kmontemayor2-sc]

/all_test

[github-actions]

GiGL Automation

@ 03:29:10UTC : 🔄 Python Unit Test started.

@ 04:33:21UTC : ✅ Workflow completed successfully.

[github-actions]

GiGL Automation

@ 03:29:12UTC : 🔄 Scala Unit Test started.

@ 03:41:00UTC : ✅ Workflow completed successfully.

[github-actions]

GiGL Automation

@ 03:29:14UTC : 🔄 C++ Unit Test started.

@ 03:31:19UTC : ✅ Workflow completed successfully.

[github-actions]

GiGL Automation

@ 03:29:15UTC : 🔄 Integration Test started.

@ 05:04:55UTC : ✅ Workflow completed successfully.

[github-actions]

GiGL Automation

@ 03:29:19UTC : 🔄 Lint Test started.

@ 03:38:23UTC : ✅ Workflow completed successfully.

[github-actions]

GiGL Automation

@ 03:29:21UTC : 🔄 E2E Test started.

@ 04:44:37UTC : ❌ Workflow failed.
Please check the logs for more details.

[yliu2-sc]

Do we actually need all those tests? I think AI added a lot of tests which weren't there before.

[kmontemayor2-sc]

Do we actually need all those tests? I think AI added a lot of tests which weren't there before.

I think AI added the tests s.t. it could do dev here. (e.g. test locally).

I think they're probably useful as such