Fix #906: execute viral propagation only at combination points (main, DuckDB-only)

src/vtlengine/Exceptions/messages.py

@@ -1063,8 +1063,10 @@
     "1-3-3-6": {
         "message": "Viral attribute {name} has no viral propagation rule; declare a "
         "'define viral propagation' rule for it.",
-        "description": "Raised when a viral attribute appears in a result without a "
-        "define viral propagation rule. Every viral attribute must declare one.",
+        "description": "Raised when a viral attribute is combined -- in an operation over "
+        "two or more datasets, an aggregation/analytic group-by, or a hierarchy roll-up -- "
+        "without a 'define viral propagation' rule. Viral attributes that are only copied "
+        "through row-preserving operators do not require a rule.",
     },
     # ---------- Interpreter ----------
     "1-3-5": {

src/vtlengine/Interpreter/__init__.py

@@ -167,13 +167,6 @@ def visit_Start(self, node: AST.Start) -> Any:
             if result is None:
                 continue
 
-            if isinstance(result, Dataset):
-                # Every viral attribute must declare a viral propagation rule (issue #877).
-                vp_registry = get_current_registry()
-                for viral_comp in result.get_viral_attributes():
-                    if vp_registry.rule_for(viral_comp) is None:
-                        raise SemanticError("1-3-3-6", name=viral_comp.name)
-
             vtlengine.Exceptions.dataset_output = None
             self.datasets[result.name] = copy(result)
             results[result.name] = result

src/vtlengine/Operators/Aggregation.py

@@ -22,6 +22,7 @@
 )
 from vtlengine.Exceptions import SemanticError
 from vtlengine.Model import Component, Dataset, Role
+from vtlengine.ViralPropagation import require_rules
 
 
 def extract_grouping_identifiers(
@@ -108,6 +109,9 @@ def validate(  # type: ignore[override]
             )
             result_components["int_var"] = new_comp
 
+        # Aggregation combines the data points of each group, so the surviving viral
+        # attributes are combined and require a propagation rule (issue #906).
+        require_rules(operand.get_viral_attributes())
         # VDS is handled in visit_Aggregation
         return Dataset(name="result", components=result_components, data=None)
 

src/vtlengine/Operators/Analytic.py

@@ -34,6 +34,7 @@
 from vtlengine.Exceptions import SemanticError
 from vtlengine.Model import Component, Dataset, Role
 from vtlengine.Utils.__Virtual_Assets import VirtualCounter
+from vtlengine.ViralPropagation import require_rules
 
 return_integer_operators = [MAX, MIN, SUM]
 
@@ -206,6 +207,9 @@ def validate(  # type: ignore[override]  # noqa: C901
                     nullable=nullable,
                 )
         dataset_name = VirtualCounter._new_ds_name()
+        # Analytic combines the data points within each partition, so the surviving viral
+        # attributes are combined and require a propagation rule (issue #906).
+        require_rules(operand.get_viral_attributes())
         return Dataset(name=dataset_name, components=result_components, data=None)
 
 

src/vtlengine/Operators/Conditional.py

@@ -11,6 +11,7 @@
 from vtlengine.Model import DataComponent, Dataset, Role, Scalar
 from vtlengine.Operators import Binary, Operator
 from vtlengine.Utils.__Virtual_Assets import VirtualCounter
+from vtlengine.ViralPropagation import combined_viral_components, require_rules
 
 
 class If(Operator):
@@ -103,6 +104,11 @@ def validate(  # noqa: C901
             if left.get_identifiers() != condition.get_identifiers():
                 raise SemanticError("1-1-9-6", op=cls.op)
         result_components = {comp_name: copy(comp) for comp_name, comp in left.components.items()}
+        # if-then-else over two datasets combines their data points per row, so viral
+        # attributes carried by both branches require a propagation rule (issue #906).
+        require_rules(
+            combined_viral_components([b for b in (left, right) if isinstance(b, Dataset)])
+        )
         return Dataset(name=dataset_name, components=result_components, data=None)
 
 
@@ -243,4 +249,7 @@ def validate(
             if isinstance(op, Dataset) and op.get_components_names() != comp_names:
                 raise SemanticError("2-1-9-7", op=cls.op)
 
+        # case over two or more datasets combines their data points, so viral attributes
+        # carried by two or more branches require a propagation rule (issue #906).
+        require_rules(combined_viral_components([op for op in ops if isinstance(op, Dataset)]))
         return Dataset(name=dataset_name, components=components, data=None)

src/vtlengine/Operators/HROperators.py

@@ -7,6 +7,7 @@
 from vtlengine.DataTypes import Boolean, Number
 from vtlengine.Model import Component, DataComponent, Dataset, Role
 from vtlengine.Utils.__Virtual_Assets import VirtualCounter
+from vtlengine.ViralPropagation import require_rules
 
 
 def get_measure_from_dataset(dataset: Dataset, code_item: str) -> DataComponent:
@@ -122,4 +123,7 @@ def validate(
         # Viral attributes propagate to the hierarchy result (issue #877).
         for viral_comp in viral_components or []:
             result_components[viral_comp.name] = copy(viral_comp)
+        # The roll-up combines child nodes into each computed node, so the combined viral
+        # attributes require a propagation rule (issue #906).
+        require_rules(viral_components or [])
         return Dataset(name=dataset_name, components=result_components, data=None)

src/vtlengine/Operators/Join.py

@@ -9,6 +9,7 @@
 from vtlengine.Model import Component, Dataset, Role
 from vtlengine.Operators import Operator
 from vtlengine.Utils.__Virtual_Assets import VirtualCounter
+from vtlengine.ViralPropagation import require_rules
 
 
 def merged_viral_attribute_names(
@@ -81,6 +82,15 @@ def merge_components(
         # (values combined via the viral propagation rule at execution time)
         # instead of being #-qualified like other shared components.
         viral_common = merged_viral_attribute_names([op.components for op in operands], set(using))
+        # A merged viral attribute has its data points combined across operands, so it
+        # requires a propagation rule (issue #906).
+        merged_viral_comps = {
+            name: op.components[name]
+            for op in operands
+            for name in viral_common
+            if name in op.components
+        }
+        require_rules(merged_viral_comps.values())
 
         for op in operands:
             for comp in op.components.values():

src/vtlengine/Operators/__init__.py

@@ -18,6 +18,7 @@
 from vtlengine.Exceptions import SemanticError
 from vtlengine.Model import Component, DataComponent, Dataset, Role, Scalar, ScalarSet
 from vtlengine.Utils.__Virtual_Assets import VirtualCounter
+from vtlengine.ViralPropagation import combined_viral_components, require_rules
 
 ALL_MODEL_DATA_TYPES = Union[Dataset, Scalar, DataComponent]
 
@@ -177,6 +178,11 @@ def dataset_validation(cls, left_operand: Dataset, right_operand: Dataset) -> Da
                 right_comp = right_operand.components[comp.name]
                 comp.nullable = left_comp.nullable or right_comp.nullable
 
+        # Viral attributes present in BOTH operands have their data points merged, so they
+        # are combined and require a propagation rule; a viral attribute in a single operand
+        # is copied through and needs none (issue #906).
+        require_rules(combined_viral_components([left_operand, right_operand]))
+
         result_dataset = Dataset(name=dataset_name, components=result_components, data=None)
         cls.apply_return_type_dataset(result_dataset, left_operand, right_operand)
         return result_dataset

src/vtlengine/ViralPropagation/__init__.py

@@ -8,7 +8,7 @@
 """
 
 from dataclasses import dataclass, field
-from typing import Any, Dict, List, Optional
+from typing import Any, Dict, Iterable, List, Optional
 
 
 @dataclass
@@ -87,3 +87,35 @@ def set_current_registry(registry: ViralPropagationRegistry) -> None:
     """Set the current viral propagation registry (called by Interpreter)."""
     global _current_registry  # noqa: PLW0603
     _current_registry = registry
+
+
+def require_rules(components: Iterable[Any]) -> None:
+    """Raise SemanticError 1-3-3-6 for any viral component lacking a propagation rule.
+
+    Call this at the combination points defined by the VTL 2.2 attribute propagation
+    rule (an operation over two or more datasets, an aggregation/analytic group-by, or
+    a hierarchy roll-up), where the default propagation algorithm must be executed. A
+    viral attribute that is only copied through (row-preserving operators) needs no rule.
+    """
+    from vtlengine.Exceptions import SemanticError  # local import avoids an import cycle
+
+    registry = get_current_registry()
+    for comp in components:
+        if registry.rule_for(comp) is None:
+            raise SemanticError("1-3-3-6", name=comp.name)
+
+
+def combined_viral_components(operands: Iterable[Any]) -> List[Any]:
+    """Return the viral components combined across ``operands``.
+
+    A viral attribute whose data points are combined appears (by name) as a viral
+    attribute in two or more operands; those require a propagation rule. A viral
+    attribute present in a single operand is copied through and needs no rule.
+    """
+    counts: Dict[str, int] = {}
+    comp_by_name: Dict[str, Any] = {}
+    for op in operands:
+        for comp in op.get_viral_attributes():
+            counts[comp.name] = counts.get(comp.name, 0) + 1
+            comp_by_name[comp.name] = comp
+    return [comp_by_name[name] for name, n in counts.items() if n >= 2]

src/vtlengine/ViralPropagation/sql.py

@@ -96,33 +96,37 @@ def vp_reduce_refs(rule: ViralPropagationRule, refs: List[str]) -> str:
     return acc
 
 
+def _enumerated_group_sql(rule: ViralPropagationRule, lst: str) -> str:
+    """Combine an enumerated rule over a group given as a DuckDB list expression ``lst``.
+
+    Mirrors ``resolve_group`` in the pandas path: an empty group is NULL, a single
+    value goes through the unary-clause mapping (``resolve_single``), and two or more
+    values are folded pairwise. ``list_reduce`` alone would skip the lambda for a
+    one-element list, leaving a lone value unmapped, so the single case is explicit.
+    """
+    single = _enumerated_single_case(rule, f"({lst})[1]")
+    pair = _enumerated_case(rule, "acc", "x")
+    return (
+        f"CASE WHEN len({lst}) = 0 THEN NULL "
+        f"WHEN len({lst}) = 1 THEN {single} "
+        f"ELSE list_reduce({lst}, (acc, x) -> {pair}) END"
+    )
+
+
 def vp_group_sql(rule: ViralPropagationRule, col_ref: str) -> str:
     """SQL aggregate expression combining a group of viral values."""
     if rule.aggregate_function is not None:
         return f"{_AGG_GROUP[rule.aggregate_function]}({col_ref})"
-    case = _enumerated_case(rule, "acc", "x")
-    return f"list_reduce(list({col_ref}), (acc, x) -> {case})"
+    return _enumerated_group_sql(rule, f"list({col_ref})")
 
 
 def vp_group_sql_windowed(rule: ViralPropagationRule, col_ref: str, over_clause: str) -> str:
     """Windowed form of vp_group_sql for analytic invocation (... OVER (window))."""
     if rule.aggregate_function is not None:
         return f"{_AGG_GROUP[rule.aggregate_function]}({col_ref}) OVER ({over_clause})"
-    case = _enumerated_case(rule, "acc", "x")
-    return f"list_reduce(list({col_ref}) OVER ({over_clause}), (acc, x) -> {case})"
+    return _enumerated_group_sql(rule, f"list({col_ref}) OVER ({over_clause})")
 
 
 def vp_no_rule_group_sql(col_ref: str) -> str:
     """Group no-rule keep: copy the value of a single-row group, else NULL."""
     return f"CASE WHEN COUNT(*) = 1 THEN MAX({col_ref}) ELSE NULL END"
-
-
-def vp_dataset_wide_sql(rule: ViralPropagationRule, col_ref: str) -> str:
-    """SQL executing the rule over a whole row-preserving operator result.
-
-    Aggregate rules collapse every viral value to one (``AGG(col) OVER ()``) applied
-    to every row; enumerated rules map each value per row.
-    """
-    if rule.aggregate_function is not None:
-        return f"{_AGG_GROUP[rule.aggregate_function]}({col_ref}) OVER ()"
-    return _enumerated_single_case(rule, col_ref)