Add ExternalSource to dynamic mode

[rostan-t]

Category:

New feature (non-breaking change which adds functionality)

Description:

Add ExternalSource to dynamic mode. It mimics fn.external_source to some extent but behaves differently, especially with batches, to integrate well with the imperative API.

This is not a very useful operator in and of itself but will enable sources other than a reader in transparent pipelining.

Additional information:

Affected modules and functionalities:

Dynamic mode.

Key points relevant for the review:

Tests:

• Existing tests apply
• New tests added
  • Python tests
  • GTests
  • Benchmark
  • Other
• N/A

Checklist

Documentation

• Existing documentation applies
• Documentation updated
  • Docstring
  • Doxygen
  • RST
  • Jupyter
  • Other
• N/A

ndd.ExternalSource doesn't appear in the documentation yet because it's only really useful with transparent pipelining.

DALI team only

Requirements

• Implements new requirements
• Affects existing requirements
• N/A

REQ IDs: N/A

JIRA TASK: DALI-4745

[github-advanced-security]

CodeQL found more than 20 potential problems in the proposed changes. Check the Files changed tab for more details.

[greptile-apps]

Greptile Summary

This PR adds ndd.ExternalSource to DALI's dynamic (imperative) mode, wrapping the existing get_callback_from_source machinery so that Python callables and iterables can feed data into the imperative API. It handles single- and multi-output sources, optional batch broadcasting via batch_size, per-output layout/dtype, and the full cycle vocabulary already supported by fn.external_source.

• New _external_source.py: implements ExternalSource with _broadcast_arg for per-output argument promotion, _get_outputs / _convert_output for dispatching between Tensor and Batch returns, and a _are_types_uniform type-guard that prevents mixed-type multi-output tuples.
• New test_external_source.py: covers callable/iterable/generator sources, cycle modes, batch broadcast, layout/dtype validation, multi-output, and CPU/GPU device placement.
• __init__.py: exports ExternalSource from the ndd namespace.

Confidence Score: 5/5

Self-contained new class with no modifications to existing code paths; the Batch/Tensor dispatch logic and cycle handling are correct.

No correctness bugs found. The type-uniformity guard, cycle handling, and broadcast logic all work as documented. Feedback is limited to style-level error message improvements and a missing validation test.

No files require special attention; the implementation and tests are clean.

Important Files Changed

Filename	Overview
dali/python/nvidia/dali/experimental/dynamic/_external_source.py	New ExternalSource class wrapping get_callback_from_source for dynamic mode; logic for num_outputs, cycle, layout/dtype broadcasting, and Batch vs Tensor dispatch is correct.
dali/test/python/experimental_mode/test_external_source.py	Good coverage of callable/iterable/generator sources, batch broadcast, multi-output, cycle modes, layout/dtype, and GPU; missing an assert_raises test for num_outputs <= 0 constructor validation.
dali/python/nvidia/dali/experimental/dynamic/init.py	One-line export addition for ExternalSource; no issues.

Flowchart

%%{init: {'theme': 'neutral'}}%%
flowchart TD
    A["ExternalSource.__call__(batch_size?)"] --> B["_callback() ← get_callback_from_source"]
    B --> C["_get_outputs(data)"]
    C -->|"num_outputs == 1"| D["wrap data in 1-tuple"]
    C -->|"num_outputs > 1"| E{data is Sequence of correct length?}
    E -->|No| F["ValueError: expected N outputs"]
    E -->|Yes| G["return data tuple"]
    D & G --> H["for each output: _convert_output(data, batch_size, idx)"]
    H --> I{_get_batch_size not None?}
    I -->|"Yes (Batch / TensorList)"| J["as_batch(...) + validate batch_size"]
    I -->|"No (scalar tensor)"| K["as_tensor(...)"]
    K -->|"batch_size given"| L["Batch.broadcast(tensor, batch_size)"]
    K -->|"no batch_size"| M["return Tensor"]
    J --> N["return Batch"]
    L --> N
    N & M --> O["_are_types_uniform(results)?"]
    O -->|No| P["TypeError: mixed Tensor/Batch"]
    O -->|Yes| Q{num_outputs == 1?}
    Q -->|Yes| R["return results[0]"]
    Q -->|No| S["return results tuple"]

Loading

_{Reviews (3): Last reviewed commit: "Add tests for ndd.ExternalSource" | Re-trigger Greptile}