feat(dmap): add distributed map operation

[nvasiu]

Summary

Adds the distributed map operation (ctx.distributed_map) to the Python SDK:
A map run processes a bounded dataset in parallel. A customer starts a map run from a durable function, naming a source to read items from, a processor function to invoke per batch, and concurrency, retry, and failure settings. The service reads items from the source, groups them into batches, invokes the processor for each batch, retries failures, tracks progress, routes successful results and failed items to destinations, and reports completion.

Changes

dmap/models.py

The result types a customer receives back from a map run.

• DistributedMapSummary: what ctx.distributed_map returns, describes the run's overall outcome.
• DistributedMapResult: returned when a DistributedMapResultConfig is passed. Contains individual map run item outcomes.
• DistributedMapResultItem and DistributedMapItemError: represent a single item's result / error.

dmap/__init__.py

Empty package marker for the dmap package.

config.py

The input types a customer constructs to describe a map run, and the distributed map enums.

• DistributedMapConfig: optional settings for distributed map.
• DistributedMapResultConfig: subclass that additionally collects item results inline, selecting the return type statically.
• InlineSource, S3Source, ReaderSource: describe where map run items come from.
• DistributedMapProcessor: describes the Lambda that processes items, how outcomes are reported back, and how failing items are retried via max_retry_attempts and max_retry_duration.
• DistributedMapCompletionConfig: defines item failure thresholds for marking the overall map run failed.
• S3Destination, DistributedMapOnSuccessConfig, DistributedMapOnFailureConfig, DistributedMapDestinationConfig: for routing successful and failed item records to S3.
• The enums live here rather than in lambda_service, so config.py has no runtime import from it.

context.py

The customer-facing entry point on the durable execution context.

• ctx.distributed_map: the method a customer calls to run a distributed map. Overloaded so passing a DistributedMapResultConfig types the return as DistributedMapResult and anything else as DistributedMapSummary.
• Validates max_concurrency against the documented ceiling of 10000.

dmap/handlers.py

Authoring decorators for the processor Lambda, so a customer writes a plain function rather than the item or batch protocol.

• distributed_map_item_handler, distributed_map_batch_handler, distributed_map_reader, and the durable variants durable_distributed_map_item_handler and durable_distributed_map_batch_handler.
• Named and shaped like durable_execution, so they work bare, with options, or called directly with a function.
• Responses are built from envelope dataclasses rather than hand-built dicts.

operation/dmap.py

The executor that drives the operation against the durable execution runtime.

• Suspends the caller's function while the run executes and resumes it with the finished outcome.
• Every terminal state resolves, and throw_if_error() opts into raising.
• All translation between the config types and the service shapes happens here.

lambda_service.py

Serialization for carrying the operation and its results to and from the backend service.

• One dataclass per API shape, named after the shape rather than with a Wire suffix.
• Items, Output, and the record body are opaque strings, matching the API model, so any serdes works rather than JSON only.

state.py

Durable execution state handling, so a run's outcome persists across suspend and resume.

• Records that a distributed map operation carries its outcome on the operation itself rather than as an operation-level result or error.

exceptions.py

The error type a customer catches.

• DistributedMapError: raised when a run or an item fails.

plugin.py

Operation type registration.

• Adds DISTRIBUTED_MAP to OperationType.

__init__.py

The package's public API surface.

• Exports the result types, their enums, and the five authoring decorators, matching how durable_execution and durable_step are exported.
• The input config and factory types are imported from config.

Tests

Tests are split by layer so each sits alongside its module.

tests/config_test.py

• Config and argument validation.

tests/lambda_service_test.py
tests/dmap/models_test.py

• Serialization round trips and the result types.

tests/operation/dmap_test.py
tests/context_test.py

• Executor behaviour and the ctx.distributed_map surface.

tests/dmap/handlers_test.py

• Authoring wrapper tests: checking that they process items, report failures, reject bad inputs.

tests/e2e/dmap_int_test.py
tests/e2e/dmap_helpers_int_test.py

• End to end tests mocking backend responses: suspend / resume, collect results, throw on failure.

Future Tasks

• Add distributed map to the local emulator (in the testing package).
  • When this is done, we can add full end to end tests using the emulator.
• Add distributed map examples to the examples package.

By submitting this pull request, I confirm that you can use, modify, copy, and redistribute this contribution, under the terms of your choice.

[nvasiu]

@yaythomas

I accidentally broke the last PR. So here's a new clean PR to add the distributed map operation with the restructure you suggested in your comment here (along with fixes to other issues you pointed out).

But I made the these modifications to your proposed restructure:

• Your suggestion: All enums should be defined in lambda_service, and customer visible enums should be imported into config.py.

  • My change: I moved customer facing enums into config, and kept the rest of the enums in lambda_service. So lambda_service imports what it needs from config, but config does not import from lambda_service.
  • Justifications:
    • Avoid having config importing from the serialization layer. Follows the contributing guide's rule that config should be the lowest level import.
    • Some enums (like DistributedMapStatus) aren't used in the serialization layer, so it doesn't make sense to keep them in lambda_service.

• Your suggestion: Export nothing new from the package root.

  • My change: I export the following from the package root: result types, customer facing enums, error type and the authoring helpers.
  • Justifications:
    • This change is consistent with other existing operations. Other operation's result types, customer facing enums and durable_* decorators are exported from the root.
    • Exporting these structures from the root lets customers import them without needing to reach into submodules.

I also used dmap for directory and file names, rather than distributed_map. You suggested dmap in an older comment to match naming for existing files / avoid underscores.

Thoughts on these changes?

I made this graph of the new operation structure to more easily see how the components relate:

---

Besides the restructure, I also made the following fixes based on your comments:

• Terminal failures resolve instead of raising.
• Items, Output and the record body pass through the serdes as strings, so non-JSON serdes work.
• Removed distributed_map_id. This just returned the last segment of the map run ARN, so it wasn't that useful. We decided to remove this from all SDKs.
• Factories renamed and flattened (batch, item_failures, item_results, InlineSource, S3Source, ReaderSource, S3Destination).
• Removed union return type. Now the return type is selected statically by config type.
• Translation has all been moved to the executor.
• Renamed serialization dataclasses after their API shapes, like existing ones.
• Envelope dataclasses to replace any hand built dicts.
• Removed the module global _CTX = SerDesContext(). Now context is built per call.
• Authoring helpers were turned into decorators and renamed to match existing decorators.
• Removed the unrelated execution.py invoke change.