[Subtask]: Prototype MO-streamed snapshot inputs for Sirius offload

## Parent Issue

Parent issue: #26154

## Detail of Subtask

Prototype an alternative Sirius input path in which MatrixOne executes the storage/MVCC scan and streams snapshot-correct Arrow batches to the sidecar, instead of requiring the sidecar to read TAE objects directly.

This is an experiment and must not replace the existing direct `TaeRead` path. Direct object reads remain the fast path for eligible flushed analytical tables. The purpose of this subtask is to determine whether an MO-produced input path is useful as a compatibility path for states that direct `TaeRead` cannot currently admit, including unflushed committed rows, visible tombstones, storage inaccessible to the sidecar, and eventually non-TAE or distributed inputs.

The core correctness invariant is:

> For every streamed Substrait read, Sirius consumes exactly the rows and schema visible at the MatrixOne transaction snapshot, with no loss, duplication, cross-query reuse, or resource lifetime beyond terminal execution.

The prototype must preserve these ownership boundaries:

- MatrixOne owns the transaction snapshot, native scan, MVCC application, projection, and production of input batches.
- The sidecar owns GPU execution and consumption of each query-local input stream.
- A read identity binds account, query, input relation, schema, and expiry; it cannot be reused by another query.
- Backpressure bounds memory on both sides. Cancellation, error, timeout, and disconnect propagate in both directions.
- Snapshot/scan resources are released only after all producers and the sidecar consumer have quiesced.
- Native fallback is allowed only before the offloaded execution becomes externally visible.

The first prototype should remain single-CN and narrowly prove the data path. It should support multiple named table inputs in one Substrait plan so a join query is a real protocol test, not a scan-only special case.

## Describe implementation you've considered

1. Add a versioned Substrait read extension for a query-local streamed input. It carries an opaque input identity and canonical schema, but no object paths or storage credentials.
2. At the logical-plan cut point, choose the experimental streamed input for selected reads while retaining the current direct `TaeRead` implementation unchanged.
3. Compile each selected read through MatrixOne's native snapshot scan and convert its projected output to bounded Arrow record batches.
4. Extend the authenticated Flight protocol so the sidecar can correlate each input stream with the corresponding Substrait `ReadRel`. The protocol must define prepare, producer-ready, running, EOF, error, cancel, and quiesced states.
5. Register each authenticated stream as a query-local sidecar relation and execute the remaining Substrait plan through the Sirius GPU physical planner, never ordinary DuckDB CPU execution.
6. Use bounded queues/windows rather than materializing a complete table. A slow sidecar must stop MO producers through backpressure rather than grow memory.
7. Make terminal cleanup idempotent across normal completion, MO cancellation, sidecar cancellation, partial input failure, Flight disconnect, timeout, and sidecar crash.

Do not introduce a general remote-execution framework in the prototype. The smallest acceptable protocol is one that supports multiple bounded Arrow inputs for one Sirius execution and closes the complete lifecycle above.

## Additional information

### Experiment matrix

Compare three paths using identical snapshots and typed results:

1. native MatrixOne;
2. current direct sidecar `TaeRead`;
3. experimental MO-streamed sidecar input.

Use TPC-H SF1 and SF10 with at least:

- Q1: large scan and aggregation;
- Q6: selective scan;
- Q9: multiple inputs and a substantial join.

Also include snapshot cases that the direct path currently rejects: an unflushed committed tail and visible tombstones.

### Required evidence

- Typed results equal native MatrixOne results.
- Execution evidence proves the Sirius GPU path consumed the streamed inputs.
- Peak MO and sidecar memory remain bounded under a deliberately slow consumer.
- Cancellation and injected producer/consumer failures terminate promptly without leaked scans, goroutines, Flight calls, or sidecar executions.
- Record storage bytes read, rows/bytes before Arrow serialization, transferred bytes, CN CPU/memory, sidecar GPU utilization, time-to-first-row, and total latency.
- Explain whether projection and predicate placement changes the break-even point.

The experiment is successful when it produces a correct, lifecycle-safe end-to-end implementation and enough measurements to decide where this path is useful. It does not need to outperform direct `TaeRead`; a measured conclusion that it should only serve compatibility/selective workloads is a valid outcome.

### Non-goals for the first prototype

- Replacing or slowing the current direct TAE fast path.
- Supporting all 22 TPC-H queries before the protocol is evaluated.
- Multi-CN producer fan-in or distributed scheduling.
- Current-transaction writes or mutable-query offload beyond what the native read-only snapshot scan can safely expose.