alexandrefimov opened a new issue, #51243:
URL: https://github.com/apache/arrow/issues/51243
`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 f251bc328.
Happy to send
the check as a PR if that is useful.
<details>
<summary>Reproducer — one file, pyarrow only</summary>
```python
"""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])
```
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
