JingsongLi commented on code in PR #9496:
URL: https://github.com/apache/paimon/pull/9496#discussion_r3896826473
##########
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:
[P2] Keep the Ray version guard on the independently callable backfill
stage. The documented retry path can call `backfill_canonical_action_ray`
directly on an already initialized Ray 2.49 runtime, which satisfies this stage
API contract but bypasses `run_ray_pipeline`. This line then adds the `action`
column before `merge_into` rejects Ray below 2.50, so the original schema-only
failure still remains for the lower-level API. Please share the version check
and run it before `_prepare_canonical_action_table`, with a direct-stage
regression test.
##########
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:
[P2] Validate backfill arguments before starting append-only ingestion. For
example, `num_partitions=0` reaches this call only after `ingest_ray` has
committed the episode and frame rows; `_materialize_canonical_action_ray` then
rejects the value. Since `run_ray_pipeline` returns no ingestion result on that
failure, rerunning it after correcting the argument appends duplicate rows.
Please validate `num_partitions` before invoking `ingest_ray` and add a test
asserting ingestion is not called for an invalid value.
--
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]
