Merge pull request #1504 from roboflow/feature/try-to-beat-the-limitation-of-ee-in-terms-of-singular-elements-pushed-into-batch-inputs

Beat the limitations of EE in terms of singular elements pushed into batch inputs

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

@@ -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`
@@ -356,6 +364,16 @@ execution excludes steps at higher `dimensionality levels` from producing output
 output field selecting that values will be presented as nested list of empty lists, with depth matching  
 `dimensionality level - 1` of referred output.
 
+Since Execution Engine `v1.6.0`, blocks within a workflow may collapse batches into scalars, as well as create new 
+batches from scalar inputs. The first scenario is pretty easy to understand - each dictionary in the output list will 
+simply be populated with the same scalar value. The case of *emergent* batch is slightly more complicated. 
+In such case we can find batch at dimensionality level 1, which has shape or elements order not compliant 
+with input batches. To prevent semantic ambiguity, we treat such batch as if it's dimensionality is one level higher
+(as if **there is additional batch-oriented input of size one attached to the input of the block creating batch 
+dynamically**). Such virtually nested outputs are broadcast, such that each dictionary in the output list will be given 
+new key with the same nested output. This nesting property is preserved even if there is no input-derived outputs 
+for given workflow - in such case, output is a list of size 1 which contains dictionary with nested output.
+
 Some outputs would require serialisation when Workflows Execution Engine runs behind HTTP API. We use the following
 serialisation strategies:
 

docs/workflows/workflow_execution.md

@@ -86,6 +86,19 @@ 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 all 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.
+
 
 ### Preparing step inputs
 

docs/workflows/workflows_execution_engine.md

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

inference/core/version.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/core_steps/fusion/dimension_collapse/v1.py

@@ -4,7 +4,7 @@
 
 
 class WorkflowBlockError(BaseModel):
-    block_id: str
+    block_id: Optional[str] = None
     block_type: Optional[str] = None
     block_details: Optional[str] = None
     property_name: Optional[str] = None

inference/core/workflows/errors.py

@@ -2,6 +2,7 @@
 PARSED_NODE_INPUT_SELECTORS_PROPERTY = "parsed_node_input_selectors"
 STEP_DEFINITION_PROPERTY = "definition"
 WORKFLOW_INPUT_BATCH_LINEAGE_ID = "<workflow_input>"
+TOP_LEVEL_LINEAGES_KEY = "top_level_lineages"
 IMAGE_TYPE_KEY = "type"
 IMAGE_VALUE_KEY = "value"
 ROOT_PARENT_ID_KEY = "root_parent_id"

inference/core/workflows/execution_engine/constants.py

@@ -64,12 +64,16 @@ def parse_block_manifest(
     inputs_accepting_batches_and_scalars = set(
         manifest_type.get_parameters_accepting_batches_and_scalars()
     )
+    inputs_enforcing_auto_batch_casting = set(
+        manifest_type.get_parameters_enforcing_auto_batch_casting()
+    )
     return parse_block_manifest_schema(
         schema=schema,
         inputs_dimensionality_offsets=inputs_dimensionality_offsets,
         dimensionality_reference_property=dimensionality_reference_property,
         inputs_accepting_batches=inputs_accepting_batches,
         inputs_accepting_batches_and_scalars=inputs_accepting_batches_and_scalars,
+        inputs_enforcing_auto_batch_casting=inputs_enforcing_auto_batch_casting,
     )
 
 
@@ -79,6 +83,7 @@ def parse_block_manifest_schema(
     dimensionality_reference_property: Optional[str],
     inputs_accepting_batches: Set[str],
     inputs_accepting_batches_and_scalars: Set[str],
+    inputs_enforcing_auto_batch_casting: Set[str],
 ) -> BlockManifestMetadata:
     primitive_types = retrieve_primitives_from_schema(
         schema=schema,
@@ -89,6 +94,7 @@ def parse_block_manifest_schema(
         dimensionality_reference_property=dimensionality_reference_property,
         inputs_accepting_batches=inputs_accepting_batches,
         inputs_accepting_batches_and_scalars=inputs_accepting_batches_and_scalars,
+        inputs_enforcing_auto_batch_casting=inputs_enforcing_auto_batch_casting,
     )
     return BlockManifestMetadata(
         primitive_types=primitive_types,
@@ -255,6 +261,7 @@ def retrieve_selectors_from_schema(
     dimensionality_reference_property: Optional[str],
     inputs_accepting_batches: Set[str],
     inputs_accepting_batches_and_scalars: Set[str],
+    inputs_enforcing_auto_batch_casting: Set[str],
 ) -> Dict[str, SelectorDefinition]:
     result = []
     for property_name, property_definition in schema[PROPERTIES_KEY].items():
@@ -277,6 +284,7 @@ def retrieve_selectors_from_schema(
                 is_list_element=True,
                 inputs_accepting_batches=inputs_accepting_batches,
                 inputs_accepting_batches_and_scalars=inputs_accepting_batches_and_scalars,
+                inputs_enforcing_auto_batch_casting=inputs_enforcing_auto_batch_casting,
             )
         elif property_definition.get(TYPE_KEY) == OBJECT_TYPE and isinstance(
             property_definition.get(ADDITIONAL_PROPERTIES_KEY), dict
@@ -290,6 +298,7 @@ def retrieve_selectors_from_schema(
                 is_dict_element=True,
                 inputs_accepting_batches=inputs_accepting_batches,
                 inputs_accepting_batches_and_scalars=inputs_accepting_batches_and_scalars,
+                inputs_enforcing_auto_batch_casting=inputs_enforcing_auto_batch_casting,
             )
         else:
             selector = retrieve_selectors_from_simple_property(
@@ -300,6 +309,7 @@ def retrieve_selectors_from_schema(
                 is_dimensionality_reference_property=is_dimensionality_reference_property,
                 inputs_accepting_batches=inputs_accepting_batches,
                 inputs_accepting_batches_and_scalars=inputs_accepting_batches_and_scalars,
+                inputs_enforcing_auto_batch_casting=inputs_enforcing_auto_batch_casting,
             )
         if selector is not None:
             result.append(selector)
@@ -314,6 +324,7 @@ def retrieve_selectors_from_simple_property(
     is_dimensionality_reference_property: bool,
     inputs_accepting_batches: Set[str],
     inputs_accepting_batches_and_scalars: Set[str],
+    inputs_enforcing_auto_batch_casting: Set[str],
     is_list_element: bool = False,
     is_dict_element: bool = False,
 ) -> Optional[SelectorDefinition]:
@@ -323,9 +334,15 @@ 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}
+                points_to_batch = {
+                    property_name in inputs_accepting_batches
+                    or property_name in inputs_enforcing_auto_batch_casting
+                }
         else:
             points_to_batch = {declared_points_to_batch}
         allowed_references = [
@@ -359,6 +376,7 @@ def retrieve_selectors_from_simple_property(
             is_dimensionality_reference_property=is_dimensionality_reference_property,
             inputs_accepting_batches=inputs_accepting_batches,
             inputs_accepting_batches_and_scalars=inputs_accepting_batches_and_scalars,
+            inputs_enforcing_auto_batch_casting=inputs_enforcing_auto_batch_casting,
             is_list_element=True,
         )
     if property_defines_union(property_definition=property_definition):
@@ -372,6 +390,7 @@ def retrieve_selectors_from_simple_property(
             is_dimensionality_reference_property=is_dimensionality_reference_property,
             inputs_accepting_batches=inputs_accepting_batches,
             inputs_accepting_batches_and_scalars=inputs_accepting_batches_and_scalars,
+            inputs_enforcing_auto_batch_casting=inputs_enforcing_auto_batch_casting,
         )
     return None
 
@@ -394,6 +413,7 @@ def retrieve_selectors_from_union_definition(
     is_dimensionality_reference_property: bool,
     inputs_accepting_batches: Set[str],
     inputs_accepting_batches_and_scalars: Set[str],
+    inputs_enforcing_auto_batch_casting: Set[str],
 ) -> Optional[SelectorDefinition]:
     union_types = (
         union_definition.get(ANY_OF_KEY, [])
@@ -410,6 +430,7 @@ def retrieve_selectors_from_union_definition(
             is_dimensionality_reference_property=is_dimensionality_reference_property,
             inputs_accepting_batches=inputs_accepting_batches,
             inputs_accepting_batches_and_scalars=inputs_accepting_batches_and_scalars,
+            inputs_enforcing_auto_batch_casting=inputs_enforcing_auto_batch_casting,
             is_list_element=is_list_element,
         )
         if result is None:

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

@@ -216,6 +216,12 @@ def iterate_through_definitions(self) -> Generator[StepInputDefinition, None, No
 StepInputData = Dict[str, Union[StepInputDefinition, CompoundStepInputDefinition]]
 
 
+@dataclass
+class AutoBatchCastingConfig:
+    casted_dimensionality: int
+    lineage_support: Optional[List[str]]
+
+
 @dataclass
 class StepNode(ExecutionGraphNode):
     step_manifest: WorkflowBlockManifest
@@ -224,6 +230,9 @@ class StepNode(ExecutionGraphNode):
     child_execution_branches: Dict[str, str] = field(default_factory=dict)
     execution_branches_impacting_inputs: Set[str] = field(default_factory=set)
     batch_oriented_parameters: Set[str] = field(default_factory=set)
+    auto_batch_casting_lineage_supports: Dict[str, AutoBatchCastingConfig] = field(
+        default_factory=dict
+    )
     step_execution_dimensionality: int = 0
 
     def controls_flow(self) -> bool: