[C++][Substrait] RelCommon.emit indices are not bounds-checked before they index the schema

[alexandrefimov]

GetEmitInfo indexes the input schema with every value in RelCommon.emit.output_mapping — input_schema->field(map_id) — without checking the value is in range, and ProcessEmitProject does the same on the project path, where the value also indexes proj_options.expressions. An out-of-range or negative index therefore reads past the end of a FieldVector and the process dies. ProcessExtensionEmit, a few lines down in the same file, already returns Status::Invalid("Out of bounds emit index ", emit_idx).

read, emit [1, 0]      [DataType(double), DataType(int64)]
read, emit [5]            exit 139
read, emit [-1]           exit 139
project, emit [5]         exit 139
aggregate, emit [1, 0]    exit 139

The last variant is why this is more than a check on malformed input: that plan is valid Substrait. Arrow's vendored proto has no expression_references, so the aggregate's grouping keys are lost, its output schema comes out with no fields, and a correct [1, 0] is out of range. That loss is #50634; with the bounds check the plan would report "out of bounds emit index", naming the user's correct mapping rather than the dropped keys.

pyarrow 25.0.1 on macOS arm64 and on Linux x86-64; the source above is cpp/src/arrow/engine/substrait/relation_internal.cc on main at f251bc3. Happy to send the check as a PR if that is useful.

Reproducer — one file, pyarrow only

"""RelCommon.emit.output_mapping indices are used to index the input schema unchecked.

Only the first variant returns; run each in a process of its own:
    for v in 0 1 2 3 4; do python3 emit_repro.py $v || echo "   exit $?"; done
"""
import json, sys
import pyarrow as pa
import pyarrow.substrait as ps
from pyarrow._substrait import _parse_json_plan

SCHEMA = pa.schema([pa.field("c0", pa.int64(), nullable=False),
                    pa.field("c1", pa.float64(), nullable=False)])

def provider(names, schema=None):
    return pa.table({"c0": [1], "c1": [2.5]}, schema=schema or SCHEMA)

def ref(i):
    return {"selection": {"directReference": {"structField": {"field": i} if i else {}},
                          "rootReference": {}}}

def emit(m):
    return {"common": {"emit": {"outputMapping": m}}}

def read(m=None):
    r = {"baseSchema": {"names": ["c0", "c1"], "struct": {
             "types": [{"i64": {"nullability": "NULLABILITY_REQUIRED"}},
                       {"fp64": {"nullability": "NULLABILITY_REQUIRED"}}],
             "nullability": "NULLABILITY_REQUIRED"}},
         "namedTable": {"names": ["t"]}}
    return {"read": dict(r, **(emit(m) if m else {}))}

def run(rel, names):
    plan = {"version": {"minorNumber": 102, "producer": "repro"},
            "relations": [{"root": {"input": rel, "names": names}}]}
    return ps.run_query(_parse_json_plan(json.dumps(plan).encode()), table_provider=provider)

# A plan that is valid Substrait: the grouping keys are where current Substrait puts them.
# Arrow's vendored proto has no expression_references, so it drops them (#50634) and the
# aggregate's output schema has no fields at all - which puts a correct mapping out of range.
AGG = {"aggregate": dict({"input": read(),
                          "groupings": [{"expressionReferences": [0, 1]}],
                          "groupingExpressions": [ref(0), ref(1)]}, **emit([1, 0]))}

VARIANTS = [
    ("read, emit [1, 0]",       lambda: run(read([1, 0]), ["c1", "c0"])),
    ("read, emit [5]",          lambda: run(read([5]), ["x"])),
    ("read, emit [-1]",         lambda: run(read([-1]), ["x"])),
    ("project, emit [5]",       lambda: run({"project": dict(
                                    {"input": read(), "expressions": [ref(0)]},
                                    **emit([5]))}, ["x"])),
    ("aggregate, emit [1, 0]",  lambda: run(AGG, ["c1", "c0"])),
]

label, case = VARIANTS[int(sys.argv[1])]
print("%-22s" % label, end=" ", flush=True)
try:
    print(case().read_all().schema.types)
except Exception as e:
    print(type(e).__name__ + ":", str(e).replace("\n", " ")[:70])