[python] Add AsyncStreamingTableScan and StreamReadBuilder for continuous streaming reads

[tub]

What does this PR do?

Adds continuous streaming read support to paimon-python via two new classes:

• AsyncStreamingTableScan — async generator that continuously polls for new snapshots and yields Plan objects as new data arrives. Python equivalent of Java's DataTableStreamScan.
• StreamReadBuilder — fluent builder (analogous to ReadBuilder) for configuring and constructing a streaming scan.
• Table.new_stream_read_builder() — entry point on the table interface.

Key features

• Initial scan: on first call, performs a full scan of the latest snapshot and yields a Plan
• Delta / changelog follow-up: subsequent iterations yield only new data per snapshot, using DeltaFollowUpScanner (changelog-producer=none) or ChangelogFollowUpScanner (input/lookup/full-compaction)
• Diff catch-up: when starting from a historical snapshot with a large gap to the latest, uses `IncrementalDiffScanner` to collapse the catch-up into a single efficient plan instead of replaying every intermediate snapshot
• Batch lookahead: `SnapshotManager.find_next_scannable()` fetches a batch of snapshot IDs in parallel to skip non-scannable snapshots (e.g. compaction-only) with minimal S3 round-trips
• Prefetching: optional background thread pre-fetches the next scannable snapshot while the caller processes the current plan
• Bucket/shard filtering: `with_buckets([0,1])` / `with_bucket_filter(fn)` for parallel consumer sharding
• Synchronous wrapper: `stream_sync()` for non-async usage

Builder API

scan = (
    table.new_stream_read_builder()
    .with_filter(predicate)
    .with_projection(["col1", "col2"])
    .with_poll_interval_ms(500)
    .new_streaming_scan()
)
reader = table.new_stream_read_builder().new_read()

async for plan in scan.stream():
    arrow_table = reader.to_arrow(plan.splits())
    process(arrow_table)

Part of streaming read PR stack

This is PR 3 of the streaming read stack. Prerequisites already merged:

• [python] Backtick quoting for identifiers, exists_batch optimization #7347 LRU caching, `ChangelogProducer` enum
• [python] Add FollowUpScanner, IncrementalDiffScanner, sharding #7348 `FollowUpScanner` hierarchy, `IncrementalDiffScanner`
• [python] Add row kind support for TableRead #7394 `TableRead` row kind support
• [python] Add consumer management for streaming progress #7349 Consumer registration
• [python] Improve the consumer_manager API and support in file_store_table #7415 `ConsumerManager` API
• [python] Add ChangelogProducer enum and changelog-producer config option #7417 `ChangelogProducer` config option
• [python] Add SnapshotManager batch lookahead with get_snapshots_batch and find_next_scannable #7418 `SnapshotManager` batch lookahead

Tracking issue: #7152

Test plan

• `pypaimon/tests/streaming_table_scan_test.py` — 12 unit tests covering initial scan, delta/changelog follow-up, diff catch-up, prefetch, bucket filtering, `stream_sync()`
• `pypaimon/tests/stream_read_builder_test.py` — 8 unit tests covering builder configuration and scan/read construction

python -m pytest pypaimon/tests/streaming_table_scan_test.py pypaimon/tests/stream_read_builder_test.py -v

🤖 Generated with Claude Code

[JingsongLi]

+1