YannByron commented on code in PR #9496:
URL: https://github.com/apache/paimon/pull/9496#discussion_r3895306927
##########
paimon-python/pypaimon/sample/robomind_agilex.py:
##########
@@ -453,8 +453,95 @@ def backfill_canonical_action(
)
+def backfill_canonical_action_ray(
+ warehouse,
+ *,
+ database=DEFAULT_DATABASE,
+ statistics_version=DEFAULT_STATISTICS_VERSION,
+ num_partitions=None):
+ """Run distributed action materialization, then refresh statistics."""
+ row_count, frames_snapshot_id = _materialize_canonical_action_ray(
+ warehouse,
+ database=database,
+ num_partitions=num_partitions,
+ )
+ statistics_snapshot_id = refresh_action_statistics(
+ warehouse,
+ database=database,
+ statistics_version=statistics_version,
+ )
+ return BackfillResult(
+ row_count=row_count,
+ frames_snapshot_id=frames_snapshot_id,
+ statistics_snapshot_id=statistics_snapshot_id,
+ statistics_version=statistics_version,
+ )
+
+
def materialize_canonical_action(warehouse, *, database=DEFAULT_DATABASE):
"""Stage one: add and populate canonical action, then commit it."""
+ connection, frames_table = _prepare_canonical_action_table(
+ warehouse, database)
+ row_count = _update_canonical_action_batches(frames_table.raw_table)
+ frames_table = connection.get_table(FRAMES_TABLE)
+ frames_snapshot_id = _snapshot_id(frames_table)
+ return row_count, frames_snapshot_id
+
+
+def _materialize_canonical_action_ray(
+ warehouse,
+ *,
+ database=DEFAULT_DATABASE,
+ num_partitions=None):
+ """Stage one: materialize canonical action with Ray self-merge."""
+ if num_partitions is not None:
+ num_partitions = _positive_int(num_partitions, "num_partitions")
+
+ try:
+ import ray
+ except ImportError:
+ raise ImportError(
+ "Ray backfill requires ray; install pypaimon[ray].")
+
+ initialized_here = not ray.is_initialized()
+ if initialized_here:
+ ray.init(
+ include_dashboard=False,
+ ignore_reinit_error=True,
+ num_cpus=2,
+ )
+ try:
+ from pypaimon.ray import WhenMatched, merge_into
+
+ connection, _ = _prepare_canonical_action_table(
Review Comment:
Accepted and fixed in 80cb1dd12. The new `run_ray_pipeline` validates Ray
2.50 or newer at the very start, before Ray initialization, ingestion, table
creation, or schema preparation. The documented end-to-end path therefore fails
without leaving an ingestion or schema-only snapshot.
##########
paimon-python/pypaimon/sample/robomind_agilex.py:
##########
@@ -453,8 +453,95 @@ def backfill_canonical_action(
)
+def backfill_canonical_action_ray(
+ warehouse,
+ *,
+ database=DEFAULT_DATABASE,
+ statistics_version=DEFAULT_STATISTICS_VERSION,
+ num_partitions=None):
+ """Run distributed action materialization, then refresh statistics."""
+ row_count, frames_snapshot_id = _materialize_canonical_action_ray(
+ warehouse,
+ database=database,
+ num_partitions=num_partitions,
+ )
+ statistics_snapshot_id = refresh_action_statistics(
+ warehouse,
+ database=database,
+ statistics_version=statistics_version,
+ )
+ return BackfillResult(
+ row_count=row_count,
+ frames_snapshot_id=frames_snapshot_id,
+ statistics_snapshot_id=statistics_snapshot_id,
+ statistics_version=statistics_version,
+ )
+
+
def materialize_canonical_action(warehouse, *, database=DEFAULT_DATABASE):
"""Stage one: add and populate canonical action, then commit it."""
+ connection, frames_table = _prepare_canonical_action_table(
+ warehouse, database)
+ row_count = _update_canonical_action_batches(frames_table.raw_table)
+ frames_table = connection.get_table(FRAMES_TABLE)
+ frames_snapshot_id = _snapshot_id(frames_table)
+ return row_count, frames_snapshot_id
+
+
+def _materialize_canonical_action_ray(
+ warehouse,
+ *,
+ database=DEFAULT_DATABASE,
+ num_partitions=None):
+ """Stage one: materialize canonical action with Ray self-merge."""
+ if num_partitions is not None:
+ num_partitions = _positive_int(num_partitions, "num_partitions")
+
+ try:
+ import ray
+ except ImportError:
+ raise ImportError(
+ "Ray backfill requires ray; install pypaimon[ray].")
+
+ initialized_here = not ray.is_initialized()
+ if initialized_here:
+ ray.init(
Review Comment:
Accepted and fixed in 80cb1dd12. `run_ray_pipeline` now accepts the cluster
address once, initializes Ray once, runs both ingestion and distributed
backfill on that runtime, and shuts it down once. `ray_address` and lifecycle
management were removed from both lower-level stage APIs; they assume Ray is
already initialized for independent retries.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
