[Issue #1277] Integrate S3QS storage backend with coordinator-managed partitioned shuffle execution

[vladaliii]

Overview

Related to #1277.

This PR integrates the S3QS storage backend into the partitioned shuffle execution path and introduces coordinator-managed task and worker execution for S3QS shuffle stages.

The primary goal of this PR is to establish the infrastructure required for decoupled producers and consumers. Shuffle data is persisted in S3, while SQS carries data availability and producer completion messages.

This is submitted as a draft because the basic execution path is implemented and tested, while automatic scaling policy, failure recovery, and remote platform validation remain follow-up work.

Architecture

The S3QS execution path is organized as follows:

PixelsPlanner
    |
    v
PartitionedJoinS3QSOperator
    |
    v
CoordinatedPlanExecution
    |
    v
PlanCoordinator
    |
    +-- Producer StageCoordinator
    |       |
    |       +-- logical partition tasks
    |       +-- runtime producer workers
    |
    +-- Consumer StageCoordinator
            |
            +-- logical join tasks
            +-- runtime consumer workers

Producer workers -- write objects --> S3
Producer workers -- publish DATA / PRODUCER_END --> SQS
Consumer workers -- poll and acknowledge messages --> SQS
Consumer workers -- read persisted objects --> S3

Operators describe and activate stages. The coordinator owns logical tasks, worker registration, task assignment, completion tracking, and worker-capacity control. Runtime workers obtain tasks through the coordinator instead of each operator permanently binding one invocation to one task.

Main Changes

• Added s3qs as a dedicated shuffle exchange method.
• Added ShuffleInfo and ShuffleQueueInfo protocol metadata.
• Added producer-side S3QS partition output.
• Added S3QS consumers for partitioned join and partitioned chain join.
• Added structured DATA and PRODUCER_END queue messages.
• Added coordinator-managed S3QS stage execution.
• Added runtime stage worker launchers and worker runners.
• Added worker-capacity control APIs as the foundation for dynamic scaling.
• Added S3QS worker and invoker implementations for Lambda, Spike, and vHive.
• Added shuffle queue resource creation and cleanup lifecycle management.
• Changed queue caching to use (shuffleId, partitionId) as the identity, preventing queues from different shuffle edges or queries from colliding.
• Preserved the existing batch and HTTP stream execution paths.

Shuffle Protocol

Each producer task writes partitioned Pixels objects to S3 and publishes a DATA message to the corresponding SQS partition queue. After completing all output for a partition, the producer publishes a PRODUCER_END message.

Consumers may start before all producers finish. They continuously poll their assigned queues, process available objects, acknowledge successfully processed messages, and exit after observing completion from all expected producers.

The current first version fails the stage directly on worker or protocol errors. Transparent retry and exactly-once processing are intentionally left for follow-up work.

Configuration

The new execution path is enabled by the S3QS exchange method together with the S3QS shuffle storage configuration.

Existing input and intermediate storage settings retain their previous semantics. The S3QS-specific configuration currently only affects the new shuffle path.

Testing

The following coverage was added and executed:

• S3QS storage and protocol unit tests.
• Producer-side shuffle wiring tests.
• Partitioned join consumer wiring tests.
• Partitioned chain join consumer wiring tests.
• Stage worker runner tests.
• Stage coordinator and runtime controller tests.
• Coordinated plan execution tests.
• S3QS operator scheduling tests.
• Real AWS S3/SQS offer, poll, read, acknowledge, and cleanup integration test.
• Coordinated in-process Spike end-to-end test using real AWS S3 and SQS.

Final observed results:

• S3QS storage: 13 tests, 12 passed, 1 environment-dependent test skipped.
• Planner/coordinator: 21 tests passed.
• Worker-common: 10 tests, 9 passed, 1 environment-dependent test skipped.
• Real AWS storage integration: passed.
• Real AWS coordinated Spike end-to-end integration: passed.

The AWS integration tests require:

• PIXELS_S3QS_IT_BUCKET
• PIXELS_S3QS_IT_QUEUE_PREFIX
• PIXELS_S3QS_IT_PREFIX (optional)

Current Limitations

The following items are not completed by this PR:

• Automatic scale-out and scale-in decision policies.
• Splitting or migrating an already-running logical task.
• Dynamic redistribution of consumer partition ownership.
• Worker retry and transparent failure recovery.
• Exactly-once processing under worker failures.
• Bounded coordinator abort and cleanup behavior for every failure path.
• End-to-end validation against remotely deployed Spike, Lambda, and vHive environments.
• Adoption of coordinated S3QS execution by every production query entry point.

The coordinator now provides the ownership and control points needed for future scaling work, but this PR should not yet be considered a complete production-grade elastic shuffle implementation.

Review Focus

Reviewers are especially invited to examine:

• Planner, operator, and coordinator responsibility boundaries.
• Logical task identity and worker-attempt separation.
• Producer and consumer termination protocol.
• Queue and object lifecycle ownership.
• Compatibility with existing batch and HTTP stream execution.
• Dependency boundaries between the planner and AWS-specific resource management.