spark: ReadRel.projection is ignored when converting a named read

[alexandrefimov]

The Spark consumer ignores ReadRel.projection on a named read. A mask selecting columns 0 and 2 from (i64, string, boolean) converts to (i64, string), so the second output names the wrong column and has the wrong type.

Reproduced at fff639064df794840db36fffcd881c09100e23df with the Spark 3.5.4 / Scala 2.12 variant and JDK 17.

Reproducer

Register t_mix with required fields c0: bigint, c1: string, c2: boolean, then convert this plan through ProtoPlanConverter and ToLogicalPlan.convert:

{
  "version": {"minorNumber": 102},
  "relations": [{"root": {
    "names": ["selected_i64", "selected_bool"],
    "input": {"read": {
      "common": {"direct": {}},
      "namedTable": {"names": ["t_mix"]},
      "baseSchema": {
        "names": ["c0", "c1", "c2"],
        "struct": {
          "nullability": "NULLABILITY_REQUIRED",
          "types": [
            {"i64": {"nullability": "NULLABILITY_REQUIRED"}},
            {"string": {"nullability": "NULLABILITY_REQUIRED"}},
            {"bool": {"nullability": "NULLABILITY_REQUIRED"}}
          ]
        }
      },
      "projection": {"select": {"structItems": [{"field": 0}, {"field": 2}]}}
    }}
  }}]
}

Expected: [selected_i64:bigint, selected_bool:boolean].

Actual: [selected_i64:bigint, selected_bool:string].

The mask preserves field order and selects two fields, so this does not depend on reordering or single-field struct unwrapping.

The published Spark probe registers the named table from the plan's exact input schema. Use a substrait-java checkout at the revision above and an installed JDK 17. Set these two absolute paths before running the commands:

export SUBSTRAIT_JAVA_DIR=/absolute/path/to/substrait-java
export JAVA17_HOME=/absolute/path/to/jdk-17
export JAVA_HOME="$JAVA17_HOME"
git clone https://github.com/alexandrefimov/substrait-conformance-cases conformance-cases
git -C conformance-cases checkout f12ba6103217fbd50b95c407b5dc3e8fae1e05a0
cd conformance-cases
mkdir -p cases

JAVA17_HOME is required by the probe on Linux; it only discovers JDK 17 automatically on macOS. JAVA_HOME selects the JDK used by Gradle.

Save the JSON above as cases/read_projection.json, then run from the same corpus directory:

bash probe/spark_all.sh cases

Where the columns are lost

ToLogicalPlan.visit(NamedScan) resolves the table and applies getRemap(), but does not apply getProjection(). Converting the read as a Rel, before root naming, returns all three input fields. Converting the full plan then pairs the leading fields with the two target fields during root naming, leaving the first two source columns in the result.

The mask semantics require the selected subset. ReadRel.projection needs to be applied before RelCommon.emit. A control with only an emit mapping selects the expected columns, so emit support alone does not cover this field.

This is the Spark counterpart of the Isthmus issue #1204. The reproducer covers NamedScan; filters and other read kinds are outside this report.