JingsongLi commented on code in PR #9496:
URL: https://github.com/apache/paimon/pull/9496#discussion_r3893551358
##########
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:
[P2] Preserve the selected Ray cluster across both pipeline stages.
`ingest_ray` accepts `ray_address` and shuts down a connection it created; when
the documented backfill call follows, Ray is no longer initialized and this
branch silently starts a local two-CPU cluster. The action backfill therefore
does not run on the remote cluster chosen for ingestion. Please mirror the
`ray_address` initialization contract here, or provide a shared Ray lifecycle
for the two stages.
--
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]
