feat(python): support streaming SQL results

[plusplusjiajia]

Purpose

Linked issue: N/A.

Python SQLContext.sql() and sql_with_options() collect all result batches before returning. Add sql_stream() and sql_stream_with_options() so callers can consume PyArrow batches incrementally and stop reading early.

Brief change log

• Return the existing RecordBatchReader from DataFusion's execution stream, preserving SQL error conversion and PyArrow string compatibility.
• Reuse the reader's cancellation support and release its stream on EOF or a read error.
• Finish write statements such as INSERT and COPY, including under EXPLAIN ANALYZE, before returning. DataFusion runs them only while the result is polled, so a reader closed unread would otherwise skip the write.
• Expose the result schema as RecordBatchReader.schema, so an empty result keeps its column names and types.
• Add regression coverage for multiple batches, Paimon reads, empty results and their schema, writes finished without reading, errors after a successful batch, early close, and SQLOptions restrictions.

Tests

• Python: test_sql_stream.py, test_datafusion.py, and test_read.py, using CPython 3.10 and locally generated filesystem Paimon fixtures.
• cargo test --locked -p pypaimon_rust --lib, including pending-read cancellation and stream cleanup, using an embedded CPython 3.11 interpreter.
• cargo clippy --locked -p pypaimon_rust --all-targets --no-deps -- -D warnings.
• cargo fmt --all -- --check and git diff --check.

API and Format

Adds SQLContext.sql_stream(sql) and SQLContext.sql_stream_with_options(sql, options), both returning RecordBatchReader. Adds RecordBatchReader.schema, which is the PyArrow result schema for SQL streams and None for table reads.

Existing readers shared by TableRead.read_arrow() and shard updates now release their stream on EOF or an underlying read error. The failing read raises the error; subsequent reads return None. Callers must create a new reader to retry.

The binding avoids collecting the entire output, while operators such as sorting and aggregation may still buffer data. No storage format changes.

Documentation

Adds Python type stubs and a README example using contextlib.closing to release the reader when iteration stops early. The README notes that SQLOptions restrict statements delegated to DataFusion, as in sql_with_options(), but not statements Paimon handles itself, such as DELETE, UPDATE, MERGE INTO or DROP TABLE on a Paimon table.

[JingsongLi]

+1