YannByron commented on code in PR #9496:
URL: https://github.com/apache/paimon/pull/9496#discussion_r3900143875
##########
paimon-python/pypaimon/sample/robomind_agilex.py:
##########
@@ -453,8 +492,78 @@ def backfill_canonical_action(
)
+def backfill_canonical_action_ray(
+ warehouse,
+ *,
+ database=DEFAULT_DATABASE,
+ statistics_version=DEFAULT_STATISTICS_VERSION,
+ num_partitions=None):
+ """Run distributed backfill on an already initialized Ray cluster."""
+ 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")
+
+ from pypaimon.ray import WhenMatched, merge_into
+
+ connection, _ = _prepare_canonical_action_table(
Review Comment:
Accepted and fixed in eafcdba02. The Ray >= 2.50 check is now shared through
`_require_ray_250` and is invoked by both `run_ray_pipeline` and direct
`backfill_canonical_action_ray` calls before `_prepare_canonical_action_table`.
A direct-stage regression test verifies that an older Ray runtime cannot mutate
the table schema.
##########
paimon-python/pypaimon/sample/robomind_agilex.py:
##########
@@ -377,33 +387,62 @@ def ingest_ray(
"ray_address cannot be set after Ray has already been
initialized.")
try:
- from pypaimon.ray import load_from_hdf5
- catalog_options = {
- "warehouse": str(Path(warehouse).expanduser().resolve())}
- paths = [episode.path for episode in episodes]
- episode_result = load_from_hdf5(
- "%s.%s" % (database, EPISODES_TABLE), paths, catalog_options,
- transform=RoboMindAgileXEpisodeTransform(episodes),
+ ingest = ingest_ray(
+ input_root,
+ warehouse,
+ database=database,
+ batch_size=batch_size,
concurrency=concurrency,
)
- frame_result = load_from_hdf5(
- "%s.%s" % (database, FRAMES_TABLE), paths, catalog_options,
- transform=RoboMindAgileXFrameTransform(
- episodes, batch_size=batch_size),
- concurrency=concurrency,
- )
- return IngestResult(
- mode="ray",
- episode_count=episode_result.row_count,
- frame_count=frame_result.row_count,
- episodes_snapshot_id=episode_result.snapshot_id,
- frames_snapshot_id=frame_result.snapshot_id,
+ backfill = backfill_canonical_action_ray(
Review Comment:
Accepted and fixed in eafcdba02. `run_ray_pipeline` now validates
`num_partitions` before Ray initialization and before `ingest_ray`, and the
regression test asserts ingestion is not called for an invalid value. I also
moved the existing `statistics_version` validation to the same preflight point
because it had the same partial-commit risk.
--
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]
