JingsongLi commented on code in PR #9496:
URL: https://github.com/apache/paimon/pull/9496#discussion_r3893549970
##########
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:
[P2] Validate the required Ray version before altering the table. The
`pypaimon[ray]` extra still permits `ray>=2.10,<3`, while `merge_into` rejects
Ray versions below 2.50. With Ray 2.49, this call adds the `action` column and
only then `merge_into` raises, leaving a schema-only snapshot behind. Please
enforce or document the 2.50 minimum and fail before
`_prepare_canonical_action_table` mutates the target.
--
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]
