feat(datafusion): support the $consumers system table

[jerry-024]

Purpose

Support querying Java-compatible Paimon consumer progress through the DataFusion $consumers system table.

Brief change log

• Add a branch-aware ConsumerManager for reading consumer/consumer-* progress files.
• Match Java's consumer JSON format and retry behavior for files being overwritten.
• Register $consumers with the columns consumer_id and next_snapshot_id.
• Push down consumer_id equality and IN filters to avoid reading unrelated consumer files.
• Document the new system table.

Tests

• cargo +1.94.0 test --locked -p paimon retries_a_consumer_being_overwritten --lib
• cargo +1.94.0 test --locked -p paimon-datafusion --test system_tables
• cargo +1.94.0 clippy --locked -p paimon-datafusion --tests -- -D warnings

API and Format

Adds ConsumerManager and Table::consumer_manager() to the Rust API. No storage format changes; the implementation reads the existing Java-compatible consumer-<id> files and nextSnapshot JSON field.

Documentation

Adds $consumers usage and column documentation to docs/src/sql.md.

[shyjsarah]

Thanks for adding $consumers. The filter pushdown and Java-compatible JSON handling look good overall. I found two minor compatibility/correctness issues:

1. The overwrite retry window is shorter than Java's (crates/paimon/src/table/consumer_manager.rs:27-28,66-84). Java release-1.3 retries mismatched consumer JSON 10 times with a 1-second delay, while this implementation uses 200 ms, exhausting the retries after about 1.8 seconds. I reproduced a case where the file becomes valid after 2.1 seconds: Rust returns DataInvalid, while Java would still be retrying. Could we align the interval/window with Java and add a delayed-overwrite test?

2. A POSIX consumer ID containing a backslash is silently omitted by list_all (crates/paimon/src/table/consumer_manager.rs:103-113). The listed filename preserves \\, but the local file:// path conversion normalizes it to / when get(id) reconstructs the path, so a file such as consumer-id\\part is listed and then read from a different path. Reading the listed FileStatus.path directly, or consistently validating such IDs, would avoid this.

Non-blocking: consumer files are currently read serially. Bounded concurrency may be worth considering for object-store tables with many consumers.

I also ran the PR's targeted tests and clippy commands locally; they all passed.