[FEA]: Add scatter / gather stages for Job partitioning and recombination

Is this a new feature, an improvement, or a change to existing functionality?

New Feature

How would you describe the priority of this feature request

Significant improvement

Please provide a clear description of problem this feature solves

Introduce two new pipeline stages:

ScatterStage: Partitions incoming DataFrames using a user-supplied Callable, tagging each fragment with a common job UUID and a fragment ID.

GatherStage: Collects incoming DataFrame fragments by job UUID. When all expected fragments are received, it concatenates them and forwards the full result downstream.

This enables fan-out/fan-in logic for tasks like per-shard processing or distributed transformations.

Describe the feature, and optionally a solution or implementation and any alternatives

class ScatterStage:
    def __init__(self, partition_fn: Callable[[pd.DataFrame], List[pd.DataFrame]]):
        self.partition_fn = partition_fn

    def on_data(self, control_message) -> List:
        df = control_message.payload["data"]
        job_uuid = str(uuid.uuid4())
        partitions = self.partition_fn(df)
        num_parts = len(partitions)

        return [
            control_message.copy_with_updates(
                payload={"data": part},
                metadata={
                    **control_message.metadata,
                    "job_id": job_uuid,
                    "fragment_index": idx,
                    "fragment_count": num_parts,
                }
            )
            for idx, part in enumerate(partitions)
        ]

class GatherStage:
    def __init__(self):
        self.fragments = defaultdict(dict)  # job_id -> {idx: df}
        self.expected_counts = {}  # job_id -> total_fragments

    def on_data(self, control_message):
        job_id = control_message.metadata["job_id"]
        idx = control_message.metadata["fragment_index"]
        count = control_message.metadata["fragment_count"]

        if job_id not in self.expected_counts:
            self.expected_counts[job_id] = count

        self.fragments[job_id][idx] = control_message.payload["data"]

        if len(self.fragments[job_id]) == self.expected_counts[job_id]:
            # All fragments received; concatenate and forward
            fragments_list = [
                self.fragments[job_id][i] for i in range(self.expected_counts[job_id])
            ]
            final_df = pd.concat(fragments_list, ignore_index=True)
            del self.fragments[job_id]
            del self.expected_counts[job_id]

            return control_message.copy_with_updates(
                payload={"data": final_df},
                metadata={k: v for k, v in control_message.metadata.items() if not k.startswith("fragment")}
            )
        return None

Additional context

flowchart TD
    A[Input ControlMessage] --> B[Scatter Stage]

    B --> C1[Fragment 1<br/>job_id=X, index=0]
    B --> C2[Fragment 2<br/>job_id=X, index=1]
    B --> C3[Fragment 3<br/>job_id=X, index=2]

    C1 --> D[Gather Stage]
    C2 --> D
    C3 --> D

    D --> E[Recombined ControlMessage]

Start by locating the existing Python pipeline stages and ControlMessage implementation, then compare their contracts with the proposed ScatterStage and GatherStage behavior. Done means fragments carry the specified job metadata, GatherStage emits only after all indexed fragments arrive, and the combined DataFrame is forwarded with fragment metadata removed; the issue names no tests, so identify the relevant test location first.