Clarify docs and fix issue with input parameters not being broadcast properly

Author: PawelPeczek-Roboflow

docs/workflows/create_workflow_block.md

@@ -1528,7 +1528,7 @@ the method signatures.
         In this example, the block visualises crops predictions and creates tiles
         presenting all crops predictions in single output image.
 
-        ```{ .py linenums="1" hl_lines="29-31 48-49 59-60"}
+        ```{ .py linenums="1" hl_lines="30-32 34-36 53-55 65-66"}
         from typing import List, Literal, Type, Union
 
         import supervision as sv
@@ -1556,10 +1556,15 @@ the method signatures.
             crops_predictions: Selector(
                 kind=[OBJECT_DETECTION_PREDICTION_KIND]
             )
+            scalar_parameter: Union[float, Selector()]
         
             @classmethod
             def get_output_dimensionality_offset(cls) -> int:
                 return -1
+
+            @classmethod
+            def get_parameters_enforcing_auto_batch_casting(cls) -> List[str]:
+                return ["crops", "crops_predictions"]
         
             @classmethod
             def describe_outputs(cls) -> List[OutputDefinition]:
@@ -1578,6 +1583,7 @@ the method signatures.
                 self,
                 crops: Batch[WorkflowImageData],
                 crops_predictions: Batch[sv.Detections],
+                scalar_parameter: float,
             ) -> BlockResult:
                 annotator = sv.BoxAnnotator()
                 visualisations = []
@@ -1591,18 +1597,22 @@ the method signatures.
                 return {"visualisations": tile}
         ```
 
-        * in lines `29-31` manifest class declares output dimensionality 
+        * in lines `30-32` manifest class declares output dimensionality 
         offset - value `-1` should be understood as decreasing dimensionality level by `1`
 
-        * in lines `48-49` you can see the impact of output dimensionality decrease
-        on the method signature. Both inputs are artificially wrapped in `Batch[]` container.
-        This is done by Execution Engine automatically on output dimensionality decrease when 
-        all inputs have the same dimensionality to enable access to all elements occupying 
-        the last dimensionality level. Obviously, only elements related to the same element 
+        * in lines `34-36` manifest class declares `run(...)` method inputs that will be subject to auto-batch casting
+        ensuring that the signature is always stable. Auto-batch casting was introduced in Execution Engine `v0.1.6.0` 
+        - refer to [changelog](./execution_engine_changelog.md) for more details.
+
+        * in lines `53-55` you can see the impact of output dimensionality decrease
+        on the method signature. First two inputs (declared in line `36`) are artificially wrapped in `Batch[]`
+        container, whereas `scalar_parameter` remains primitive type. This is done by Execution Engine automatically 
+        on output dimensionality decrease when all inputs have the same dimensionality to enable access to 
+        all elements occupying the last dimensionality level. Obviously, only elements related to the same element 
         from top-level batch will be grouped. For instance, if you had two input images that you 
         cropped - crops from those two different images will be grouped separately.
 
-        * lines `59-60` illustrate how output is constructed - single value is returned and that value 
+        * lines `65-66` illustrate how output is constructed - single value is returned and that value 
         will be indexed by Execution Engine in output batch with reduced dimensionality
 
     === "different input dimensionalities"

docs/workflows/execution_engine_changelog.md

@@ -56,7 +56,7 @@ exception being output dimensionality changes introduced by the block itself. As
 
     * **collapse batches into scalars** (when the block decreases dimensionality).
 
-* The only potential friction point arises **when a block that does not accept batches** (and thus does not denote 
+* The two potential friction point arises - first **when a block that does not accept batches** (and thus does not denote 
 batch-accepting inputs) **decreases output dimensionality**. In previous versions, the Execution Engine handled this by 
 applying dimensionality wrapping: all batch-oriented inputs were wrapped with an additional `Batch[T]` dimension, 
 allowing the block’s `run(...)` method to perform reduce operations across the list dimension. With Auto Batch Casting, 
@@ -65,6 +65,22 @@ scalars or batches, making casting nondeterministic. To address this, a new mani
 `get_parameters_enforcing_auto_batch_casting(...)`. This method must return the list of parameters for which batch 
 casting should be enforced when dimensionality is decreased. It is not expected to be used in any other context.
 
+!!! warning "Impact of new method on existing blocks"
+
+    The requirement of defining `get_parameters_enforcing_auto_batch_casting(...)` method to fully use 
+    Auto Batch Casting feature in the case described above is non-strict. If the block will not be changed,
+    the only effect will be that workflows wchich were **previously failing** with compilation error may 
+    work or fail with **runtime error**, dependent on the details of block `run(...)` method implementation.
+
+* The second friction point arises when there is a block declaring input fields supporting batches and scalars using 
+`get_parameters_accepting_batches_and_scalars(...)` - by default, Execution Engine will skip auto-casting for such 
+parameters, as the method was historically **always a way to declare that block itself has ability to broadcast scalars 
+into batches** - see 
+[implementation of `roboflow_core/detections_transformation@v1`](/inference/core/workflows/core_steps/transformations/detections_transformation/v1.py) 
+block. In a way, Auto Batch Casting is *redundant* for those blocks - so we propose leaving them as is and 
+upgrade to use `get_parameters_enforcing_auto_batch_casting(...)` instead of 
+`get_parameters_accepting_batches_and_scalars(...)` in new versions of such blocks.
+
 * In earlier versions, a hard constraint existed: dimensionality collapse could only occur at levels ≥ 2 (i.e. only 
 on nested batches). This limitation is now removed. Dimensionality collapse blocks may also operate on scalars, with 
 the output dimensionality “bouncing off” the zero ground.
@@ -86,7 +102,88 @@ situation is known limitation of Workflows Compiler.
 Contact Roboflow team through github issues (https://github.com/roboflow/inference/issues) 
 providing full context of the problem - including workflow definition you use.
 ```
+### Migration guide
+
+??? Hint "Adding `get_parameters_enforcing_auto_batch_casting(...)` method"
+
+    Blocks which decrease output dimensionality and do not define batch-oriented inputs needs to 
+    declare all inputs which implementation expects to have wrapped in `Batch[T]` with the new class 
+    method of block manifest called `get_parameters_enforcing_auto_batch_casting(...)`
+
+    ```{ .py linenums="1" hl_lines="34-36 53-54"}
+    from typing import List, Literal, Type, Union
+
+    import supervision as sv
+    
+    from inference.core.workflows.execution_engine.entities.base import (
+        Batch,
+        OutputDefinition,
+        WorkflowImageData,
+    )
+    from inference.core.workflows.execution_engine.entities.types import (
+        IMAGE_KIND,
+        OBJECT_DETECTION_PREDICTION_KIND,
+        Selector,
+    )
+    from inference.core.workflows.prototypes.block import (
+        BlockResult,
+        WorkflowBlock,
+        WorkflowBlockManifest,
+    )
+    
+    
+    class BlockManifest(WorkflowBlockManifest):
+        type: Literal["my_plugin/tile_detections@v1"]
+        crops: Selector(kind=[IMAGE_KIND])
+        crops_predictions: Selector(
+            kind=[OBJECT_DETECTION_PREDICTION_KIND]
+        )
+        scalar_parameter: Union[float, Selector()]
+    
+        @classmethod
+        def get_output_dimensionality_offset(cls) -> int:
+            return -1
+    
+        @classmethod
+        def get_parameters_enforcing_auto_batch_casting(cls) -> List[str]:
+            return ["crops", "crops_predictions"]
+        
+        @classmethod
+        def describe_outputs(cls) -> List[OutputDefinition]:
+            return [
+                OutputDefinition(name="visualisations", kind=[IMAGE_KIND]),
+            ]
+    
+    
+    class TileDetectionsBlock(WorkflowBlock):
+    
+        @classmethod
+        def get_manifest(cls) -> Type[WorkflowBlockManifest]:
+            return BlockManifest
+    
+        def run(
+            self,
+            crops: Batch[WorkflowImageData],
+            crops_predictions: Batch[sv.Detections],
+            scalar_parameter: float,
+        ) -> BlockResult:
+            print("This is parameter which will not be auto-batch cast!", scalar_parameter)
+            annotator = sv.BoxAnnotator()
+            visualisations = []
+            for image, prediction in zip(crops, crops_predictions):
+                annotated_image = annotator.annotate(
+                    image.numpy_image.copy(),
+                    prediction,
+                )
+                visualisations.append(annotated_image)
+            tile = sv.create_tiles(visualisations)
+            return {"visualisations": tile}
+    ```
+
+    * in lines `34-36` one needs to add declaration of fields that will be subject to enforced auto-batch casting
 
+    * as a result of the above, input parameters of run method (lines `53-54`) will be wrapped into `Batch[T]` by 
+    Execution Engine.
 
 ## Execution Engine `v1.5.0` | inference `v0.38.0`
 

docs/workflows/workflow_execution.md

@@ -124,6 +124,14 @@ influencing the processing for all elements in the batch and this type of data w
     the reference images remain unchanged as you process each input. Thus, the reference images are considered 
     *scalar* data, while the list of input images is *batch-oriented*.
 
+    **Great news!**
+    
+    Since Execution Engine `v1.6.0`, the practical aspects of dealing with *scalars* and *batches* are offloaded to 
+    the Execution Engine (refer to [changelog](./execution_engine_changelog.md) for more details). As a block 
+    developer, it is still important to understand the difference, but when building blocks you are not forced to 
+    think about the nuances that much.
+
+
 To illustrate the distinction, Workflow definitions hold inputs of the two categories:
 
 - **Scalar inputs** - like `WorkflowParameter`

docs/workflows/workflows_execution_engine.md

@@ -86,6 +86,35 @@ batch-oriented input, it will be treated as a SIMD step.
 Non-SIMD steps, by contrast, are expected to deliver a single result for the input data. In the case of non-SIMD 
 flow-control steps, they affect all downstream steps as a whole, rather than individually for each element in a batch.
 
+Historically, Execution Engine could not handle well al scenarios when non-SIMD steps' outputs were fed into SIMD steps
+inputs - causing compilation error due to lack of ability to automatically cast such outputs into batches when feeding
+into SIMD seps. Starting with Execution Engine `v1.6.0`, the handling of SIMD and non-SIMD blocks has been improved 
+through the introduction of **Auto Batch Casting**:
+
+* When a SIMD input is detected but receives scalar data, the Execution Engine automatically casts it into a batch.
+
+* The dimensionality of the batch is determined at compile time, using *lineage* information from other 
+batch-oriented inputs when available. Missing dimensions are generated in a manner similar to `torch.unsqueeze(...)`.
+
+* Outputs are evaluated against the casting context - leaving them as scalars when block keeps or decreases output 
+dimensionality or **creating new batches** when increase of dimensionality is expected.
+
+!!! warning "We don't support multiple sources of batch-oriented data"
+
+    While Auto Batch Casting simplifies mixing SIMD and non-SIMD blocks, there is one major limitation to be aware of.
+
+    If multiple first-level batches are created from different origins (for instance inputs and steps taking scalars
+    and raising output dimensionality into batch at first level of depth), the Execution Engine cannot deterministically 
+    construct the output. In previous versions, the assumption was that **outputs were lists directly tied to inputs 
+    batch order**. With Auto Batch Casting, batches may also be generated dynamically, and no deterministic ordering 
+    can be guaranteed (imagine scenario when you feed batch of 4 images, and there is a block generating dynamic batch 
+    with 3 images - when results are to be returned, Execution Engine is unable to determine a single input batch which 
+    would dictate output order alignment, which is a hard requirement caused by falty design choices). 
+
+    To prevent unpredictable behaviour, the Execution Engine asserts in this scenario and raises an error instead of 
+    proceeding. Resolving this design flaw requires breaking changes and is therefore deferred to 
+    **Execution Engine v2.0.**
+
 
 ### Preparing step inputs
 

inference/core/version.py

@@ -1,4 +1,4 @@
-__version__ = "0.52.1"
+__version__ = "0.53.0"
 
 
 if __name__ == "__main__":

inference/core/workflows/core_steps/fusion/dimension_collapse/v1.py

@@ -59,6 +59,10 @@ def get_output_dimensionality_offset(
     ) -> int:
         return -1
 
+    @classmethod
+    def get_parameters_enforcing_auto_batch_casting(cls) -> List[str]:
+        return ["data"]
+
     @classmethod
     def describe_outputs(cls) -> List[OutputDefinition]:
         return [

inference/core/workflows/execution_engine/introspection/schema_parser.py

@@ -334,7 +334,10 @@ def retrieve_selectors_from_simple_property(
         )
         if declared_points_to_batch == "dynamic":
             if property_name in inputs_accepting_batches_and_scalars:
-                points_to_batch = {True, False}
+                if property_name in inputs_enforcing_auto_batch_casting:
+                    points_to_batch = {True}
+                else:
+                    points_to_batch = {True, False}
             else:
                 points_to_batch = {
                     property_name in inputs_accepting_batches

inference/core/workflows/execution_engine/v1/compiler/graph_constructor.py

@@ -1577,18 +1577,31 @@ def verify_declared_batch_compatibility_against_actual_inputs(
     batch_compatibility_of_properties: Dict[str, Set[bool]],
 ) -> Set[str]:
     scalar_parameters_to_be_batched = set()
+    parameters_accepting_batches_and_scalars = set(
+        step_node_data.step_manifest.get_parameters_accepting_batches_and_scalars()
+    )
+    hardcoded_inputs_to_be_batch_compatible = set(
+        step_node_data.step_manifest.get_parameters_enforcing_auto_batch_casting()
+        + step_node_data.step_manifest.get_parameters_accepting_batches()
+    )
     for property_name, input_definition in input_data.items():
         if property_name not in batch_compatibility_of_properties:
-            # only values plugged via selectors are to be validated
-            continue
-        if input_definition.is_compound_input():
+            actual_input_is_batch = {False}
+            if property_name in parameters_accepting_batches_and_scalars:
+                batch_compatibility = {True, False}
+            elif property_name in hardcoded_inputs_to_be_batch_compatible:
+                batch_compatibility = {True}
+            else:
+                continue
+        elif input_definition.is_compound_input():
             actual_input_is_batch = {
                 element.is_batch_oriented()
                 for element in input_definition.iterate_through_definitions()
             }
+            batch_compatibility = batch_compatibility_of_properties[property_name]
         else:
             actual_input_is_batch = {input_definition.is_batch_oriented()}
-        batch_compatibility = batch_compatibility_of_properties[property_name]
+            batch_compatibility = batch_compatibility_of_properties[property_name]
         step_accepts_batch_input = step_node_data.step_manifest.accepts_batch_input()
         if (
             step_accepts_batch_input

inference/core/workflows/execution_engine/v1/dynamic_blocks/block_assembler.py

@@ -381,7 +381,9 @@ def assembly_manifest_class_methods(
         "get_parameters_accepting_batches_and_scalars",
         classmethod(get_parameters_accepting_batches_and_scalars),
     )
-    get_parameters_enforcing_auto_batch_casting = lambda cls: list()
+    get_parameters_enforcing_auto_batch_casting = (
+        lambda cls: manifest_description.get_parameters_enforcing_auto_batch_casting
+    )
     setattr(
         manifest_class,
         "get_parameters_enforcing_auto_batch_casting",

inference/core/workflows/execution_engine/v1/dynamic_blocks/entities.py

@@ -116,6 +116,13 @@ class ManifestDescription(BaseModel):
         "Value will override `accepts_batch_input` if non-empty "
         "list is provided, `accepts_batch_input` is kept not to break backward compatibility.",
     )
+    get_parameters_enforcing_auto_batch_casting: List[str] = Field(
+        default_factory=list,
+        description="List of parameters, for which auto-batch casting should be enforced, making sure that the block "
+        "run(...) method will always receive the parameters as batches, not scalars. This property is important for "
+        "blocks decreasing output dimensionality which do not define neither `batch_oriented_parameters` nor "
+        "`parameters_with_scalars_and_batches`.",
+    )
 
 
 class PythonCode(BaseModel):