[python] Add self-contained Ray datasource and top-level read_paimon/write_paimon API

[TheR1sing3un]

Purpose

Today, reading a Paimon table into a Ray Dataset requires the caller to first build a TableRead by hand:

catalog = CatalogFactory.create({"warehouse": ...})
table = catalog.get_table("db.table")
rb = table.new_read_builder()
read = rb.new_read()
splits = rb.new_scan().plan().splits()
ds = read.to_ray(splits)

That works for the existing TableRead.to_ray() helper, but it forces every user to repeat the same catalog → table → builder boilerplate. The Iceberg integration has long had a single-line IcebergDatasource(table_identifier, catalog_options, ...), and that's the missing surface here.

This PR makes RayDatasource self-contained and adds a top-level facade so reading and writing a Paimon table from Ray is one call.

Datasource refactor

• RayDatasource(table_identifier, catalog_options, predicate=None, projection=None, limit=None) is the new constructor. The catalog, table, splits and read_type are loaded lazily via @property (table, splits, read_type), so the object is cheap to instantiate and easy to ship across Ray workers.
• The legacy entry point — TableRead.to_ray() — keeps working through a new RayDatasource._from_table_read(table_read, splits) classmethod that wraps an already-resolved (table_read, splits) pair without a second catalog round-trip.
• The read closure is hoisted to a module-level _paimon_read_task generator that yields one Arrow table per batch. This:
  • avoids the closure-over-self serialization overhead documented in ray-project/ray#49107;
  • keeps memory proportional to one batch rather than the whole split chunk.

New module: pypaimon.ray

from pypaimon.ray import read_paimon, write_paimon

ds = read_paimon("db.table", catalog_options={"warehouse": "/path"})
write_paimon(ds, "db.table", catalog_options={"warehouse": "/path"})

read_paimon(table_identifier, catalog_options, *, filter=None, projection=None, limit=None, ray_remote_args=None, concurrency=None, override_num_blocks=None, **read_args) -> ray.data.Dataset

write_paimon(dataset, table_identifier, catalog_options, *, overwrite=False, concurrency=None, ray_remote_args=None) -> None

Linked issue

N/A — surfaced when wiring pypaimon into a Ray-based ingestion pipeline alongside Iceberg, where users expected a read_paimon/write_paimon facade analogous to read_iceberg/write_iceberg.

Tests

New pypaimon/tests/ray_integration_test.py (8 cases):

• test_read_paimon_basic — round-trip of three rows.
• test_read_paimon_with_projection — projection narrowing.
• test_read_paimon_with_filter — predicate pushdown via PredicateBuilder.
• test_read_paimon_empty_table — empty datasets work.
• test_write_paimon_basic — write-then-read round-trip via the facade.
• test_write_paimon_overwrite — overwrite=True replaces data.
• test_read_paimon_primary_key — PK upsert + merge-on-read row count and values.
• test_read_paimon_invalid_override_num_blocks — guards override_num_blocks < 1.

Existing ray_data_test / ray_sink_test (which exercise TableRead.to_ray() and PaimonDatasink directly) continue to pass via the _from_table_read bridge — no behavioural change for those paths.

Local: pytest pypaimon/tests/{ray_data_test,ray_sink_test,ray_integration_test}.py → 26 passed; flake8 --config=dev/cfg.ini clean.

API and format

Public Python API additions (pypaimon.ray.read_paimon, pypaimon.ray.write_paimon). The existing RayDatasource constructor signature changes shape (now takes table_identifier + catalog_options instead of table_read + splits). The only in-tree caller of the old form, TableRead.to_ray(), is migrated to RayDatasource._from_table_read() in the same change. No external project should be importing RayDatasource directly today, but the bridge classmethod is kept available for callers that do.

No file format change.

Documentation

Public docstrings on RayDatasource, read_paimon, and write_paimon describe the new contract and parameters.

Generative AI disclosure

Drafted with assistance from an AI coding tool; the design follows Iceberg's IcebergDatasource pattern, and every behavioural guarantee made by the new facade is exercised by a test in ray_integration_test.py.

[TheR1sing3un]

Thanks for the review! Addressed all three comments in c4e86cd:

1. _ensure_planned() helper — splits and read_type properties now share a single entry point that runs the ReadBuilder plan once and populates both fields together, instead of each property doing its own if x is None: self._plan() check.

2. _from_table_read no longer bypasses __init__ — added a private _resolved=(table, splits, read_type) sentinel parameter to __init__. When supplied, the catalog/identifier path is skipped and the pre-resolved values are used directly. _from_table_read now forwards through __init__, so any future field added to __init__ is automatically initialized for both construction paths. Also added validation: table_identifier and catalog_options are required when _resolved is None.

3. limit test case — added test_read_paimon_with_limit to ray_integration_test.py. Writes 10 rows across two partitions (forces two raw-convertible splits) and asserts limit=3 causes the scan to drop the second split (Ray Dataset row count < 10), with the full unbounded read as sanity baseline. The assertion uses < 10 rather than exact == N because Paimon's scan-time limit is per-split (whole-split granularity at this layer); row-exact short-circuiting in the reader is a separate follow-up.

Tests: pypaimon/tests/ray_integration_test.py 9/9 pass, flake8 clean. Ready for re-review.

[JingsongLi]

Perhaps we should do some abstraction, such as providing a SplitProvider to obtain splits.

[TheR1sing3un]

Perhaps we should do some abstraction, such as providing a SplitProvider to obtain splits.

Do you mean to extract the logic of how to find the corresponding read splits through identifiers, predicates, limits, and projections into a separate method? To achieve possible reuse in the future? For example, the integration of the high level into the daft engine

[JingsongLi]

I just mean you should do some abstraction, introduce SplitProvider ABC, and implement two implementations.

[TheR1sing3un]

I just mean you should do some abstraction, introduce SplitProvider ABC, and implement two implementations.

Thank you for your reminder. I'll make it happen

[TheR1sing3un]

I just mean you should do some abstraction, introduce SplitProvider ABC, and implement two implementations.

done~