From d23b523c4c0b32860bc797aca41a0380ca4118ed Mon Sep 17 00:00:00 2001 From: Javier Hernandez Date: Thu, 16 Jul 2026 14:58:30 +0200 Subject: [PATCH 1/6] Port #906 to main: execute viral propagation only at combination points DuckDB-only adaptation of the 1.9.X change (cr-906 / PR #907). Viral propagation runs and is required only where input data points are combined (>=2 datasets, aggregation/analytic group-by, hierarchy roll-up); everywhere else viral attributes are copied unchanged. Adds require_rules/ combined_viral_components, moves the 1-3-3-6 check from the global per-statement site into the combining operators' validate methods, and makes the row-preserving DuckDB emissions copy (removing vp_dataset_wide_sql). nvl copies its primary operand. Supersedes #897's dataset-wide check_datapoint viral aggregate (a row-preserving validation now copies per datapoint). Fixes #906. --- src/vtlengine/Exceptions/messages.py | 6 +- src/vtlengine/Interpreter/__init__.py | 7 - src/vtlengine/Operators/Aggregation.py | 4 + src/vtlengine/Operators/Analytic.py | 4 + src/vtlengine/Operators/Conditional.py | 9 + src/vtlengine/Operators/HROperators.py | 4 + src/vtlengine/Operators/Join.py | 10 + src/vtlengine/Operators/__init__.py | 6 + src/vtlengine/ViralPropagation/__init__.py | 34 ++- src/vtlengine/ViralPropagation/sql.py | 11 - .../duckdb_transpiler/Transpiler/__init__.py | 157 +++++------ .../data/DataSet/output/13-1-DS_r.csv | 2 +- .../test_viral_rule_execution.py | 250 ++++++++++++------ 13 files changed, 313 insertions(+), 191 deletions(-) diff --git a/src/vtlengine/Exceptions/messages.py b/src/vtlengine/Exceptions/messages.py index 1b60de1df..45a6ec3ac 100644 --- a/src/vtlengine/Exceptions/messages.py +++ b/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": { diff --git a/src/vtlengine/Interpreter/__init__.py b/src/vtlengine/Interpreter/__init__.py index b70268087..001c780b1 100644 --- a/src/vtlengine/Interpreter/__init__.py +++ b/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 diff --git a/src/vtlengine/Operators/Aggregation.py b/src/vtlengine/Operators/Aggregation.py index 5b5fa9382..2b8c47ee3 100644 --- a/src/vtlengine/Operators/Aggregation.py +++ b/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) diff --git a/src/vtlengine/Operators/Analytic.py b/src/vtlengine/Operators/Analytic.py index 9206eea24..0c05c0392 100644 --- a/src/vtlengine/Operators/Analytic.py +++ b/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) diff --git a/src/vtlengine/Operators/Conditional.py b/src/vtlengine/Operators/Conditional.py index bc2a35939..75ce7fad9 100644 --- a/src/vtlengine/Operators/Conditional.py +++ b/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) diff --git a/src/vtlengine/Operators/HROperators.py b/src/vtlengine/Operators/HROperators.py index 5cf77c116..95c3c4a34 100644 --- a/src/vtlengine/Operators/HROperators.py +++ b/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) diff --git a/src/vtlengine/Operators/Join.py b/src/vtlengine/Operators/Join.py index 75d2d8053..57c06ab46 100644 --- a/src/vtlengine/Operators/Join.py +++ b/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(): diff --git a/src/vtlengine/Operators/__init__.py b/src/vtlengine/Operators/__init__.py index 24997a378..604d85d61 100644 --- a/src/vtlengine/Operators/__init__.py +++ b/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 diff --git a/src/vtlengine/ViralPropagation/__init__.py b/src/vtlengine/ViralPropagation/__init__.py index 7dc47e8e9..920c21ad3 100644 --- a/src/vtlengine/ViralPropagation/__init__.py +++ b/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] diff --git a/src/vtlengine/ViralPropagation/sql.py b/src/vtlengine/ViralPropagation/sql.py index f97352597..3e8eb8e04 100644 --- a/src/vtlengine/ViralPropagation/sql.py +++ b/src/vtlengine/ViralPropagation/sql.py @@ -115,14 +115,3 @@ def vp_group_sql_windowed(rule: ViralPropagationRule, col_ref: str, 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) diff --git a/src/vtlengine/duckdb_transpiler/Transpiler/__init__.py b/src/vtlengine/duckdb_transpiler/Transpiler/__init__.py index ad3146de9..ff71afafc 100644 --- a/src/vtlengine/duckdb_transpiler/Transpiler/__init__.py +++ b/src/vtlengine/duckdb_transpiler/Transpiler/__init__.py @@ -47,7 +47,6 @@ from vtlengine.Operators.Join import merged_viral_attribute_names from vtlengine.ViralPropagation import get_current_registry from vtlengine.ViralPropagation.sql import ( - vp_dataset_wide_sql, vp_group_sql, vp_group_sql_windowed, vp_no_rule_group_sql, @@ -584,16 +583,9 @@ def _apply_measures( if viral_expr_fn is not None: cols.append(f"{viral_expr_fn(name, comp)} AS {quote_name(name)}") elif output_ds is not None and name in output_ds.components: - # Row-preserving op: execute the viral propagation rule over the - # result (aggregate rules collapse dataset-wide, enumerated map per - # row); no rule = passthrough. Only when the output keeps it (issue #877). - rule = get_current_registry().rule_for(comp) - if rule is None: - cols.append(quote_name(name)) - else: - cols.append( - f"{vp_dataset_wide_sql(rule, quote_name(name))} AS {quote_name(name)}" - ) + # Row-preserving op: data points are not combined, so the viral + # attribute is copied through unchanged (issue #906). + cols.append(quote_name(name)) return SQLBuilder().select(*cols).from_table(table_src).build() @@ -737,6 +729,64 @@ def _make_binary_expr( # Typed or generic registry lookup, with function-call fallback return registry.sql(op, left_ref, right_ref, data_type=dt) + def _ds_ds_viral_cols( + self, + op: str, + left_ds: Dataset, + right_ds: Dataset, + output_ds: Optional[Dataset], + alias_a: str, + alias_b: str, + ) -> List[str]: + """Build the viral-attribute SELECT columns for a two-dataset binary op. + + Operators that combine data points run the propagation rule (``vp_pair_sql``); + ``nvl`` is single-operand and copies the primary operand (COALESCE-filled) (#906). + A viral attribute present in only one operand is copied from that operand. + """ + vp_registry = get_current_registry() + left_viral = {n for n, c in left_ds.components.items() if c.role == Role.VIRAL_ATTRIBUTE} + right_viral = {n for n, c in right_ds.components.items() if c.role == Role.VIRAL_ATTRIBUTE} + if output_ds is not None: + viral_names = [ + n for n, c in output_ds.components.items() if c.role == Role.VIRAL_ATTRIBUTE + ] + else: + viral_names = sorted(left_viral | right_viral) + result: List[str] = [] + for name in viral_names: + qn = quote_name(name) + in_left = name in left_viral + in_right = name in right_viral + if op == tokens.NVL: + # nvl copies the primary operand's viral value, filling nulls from the + # secondary operand; no rule is run (issue #906). + if in_left and in_right: + result.append(f"COALESCE({alias_a}.{qn}, {alias_b}.{qn}) AS {qn}") + elif in_left: + result.append(f"{alias_a}.{qn} AS {qn}") + elif in_right: + result.append(f"{alias_b}.{qn} AS {qn}") + continue + if in_left and in_right: + comp = ( + output_ds.components.get(name) if output_ds else None + ) or left_ds.components[name] + rule = vp_registry.rule_for(comp) + if rule is None: + result.append( + f"CAST(NULL AS {get_duckdb_type(comp.data_type.__name__)}) AS {qn}" + ) + else: + result.append( + f"{vp_pair_sql(rule, f'{alias_a}.{qn}', f'{alias_b}.{qn}')} AS {qn}" + ) + elif in_left: + result.append(f"{alias_a}.{qn} AS {qn}") + elif in_right: + result.append(f"{alias_b}.{qn} AS {qn}") + return result + def _build_ds_ds_binary( self, left_node: AST.AST, @@ -809,35 +859,8 @@ def _build_ds_ds_binary( out_name = output_measure_names[0] cols.append(f"{expr} AS {quote_name(out_name)}") - # Viral attribute propagation: combine values present in both operands. - vp_registry = get_current_registry() - left_viral = {n for n, c in left_ds.components.items() if c.role == Role.VIRAL_ATTRIBUTE} - right_viral = {n for n, c in right_ds.components.items() if c.role == Role.VIRAL_ATTRIBUTE} - if output_ds is not None: - viral_names = [ - n for n, c in output_ds.components.items() if c.role == Role.VIRAL_ATTRIBUTE - ] - else: - viral_names = sorted(left_viral | right_viral) - for name in viral_names: - qn = quote_name(name) - in_left = name in left_viral - in_right = name in right_viral - if in_left and in_right: - comp = ( - output_ds.components.get(name) if output_ds else None - ) or left_ds.components[name] - rule = vp_registry.rule_for(comp) - if rule is None: - cols.append(f"CAST(NULL AS {get_duckdb_type(comp.data_type.__name__)}) AS {qn}") - else: - cols.append( - f"{vp_pair_sql(rule, f'{alias_a}.{qn}', f'{alias_b}.{qn}')} AS {qn}" - ) - elif in_left: - cols.append(f"{alias_a}.{qn} AS {qn}") - elif in_right: - cols.append(f"{alias_b}.{qn} AS {qn}") + # Viral attribute propagation across the two operands (issue #906). + cols.extend(self._ds_ds_viral_cols(op, left_ds, right_ds, output_ds, alias_a, alias_b)) on_clause = self._join_on_clause(common_ids, alias_a, alias_b) @@ -1015,16 +1038,12 @@ def _visit_period_indicator(self, node: AST.UnaryOp) -> str: id_cols = [quote_name(c.name) for c in ds.get_identifiers()] # Execute the viral propagation rule on the (row-preserving) result (issue #877). - reg = get_current_registry() viral_cols: List[str] = [] for v_comp in ds.components.values(): if v_comp.role != Role.VIRAL_ATTRIBUTE: continue - v_qn = quote_name(v_comp.name) - v_rule = reg.rule_for(v_comp) - viral_cols.append( - v_qn if v_rule is None else f"{vp_dataset_wide_sql(v_rule, v_qn)} AS {v_qn}" - ) + # Row-preserving op: viral attribute copied through unchanged (issue #906). + viral_cols.append(quote_name(v_comp.name)) extract_expr = ( f'vtl_period_parse({quote_name(time_id)}).period_indicator AS "duration_var"' ) @@ -1983,25 +2002,8 @@ def visit_RegularAggregation_unpivot(self, node: AST.RegularAggregation) -> str: measure_names = ds.get_measures_names() viral_names = ds.get_viral_attributes_names() - # Execute the rule over the whole source before the melt so aggregate rules are not - # distorted by row replication; each UNION arm then reads the resolved value (issue #877). - reg = get_current_registry() - excl_list: List[str] = [] - expr_list: List[str] = [] - for comp in ds.components.values(): - if comp.role != Role.VIRAL_ATTRIBUTE: - continue - v_rule = reg.rule_for(comp) - if v_rule is None: - continue - qn = quote_name(comp.name) - excl_list.append(qn) - expr_list.append(f"{vp_dataset_wide_sql(v_rule, qn)} AS {qn}") - if expr_list: - table_src = ( - f"(SELECT * EXCLUDE ({', '.join(excl_list)}), {', '.join(expr_list)} " - f"FROM {table_src} AS _uv_in) AS _uv_src" - ) + # Viral attributes are copied (replicated) across the unpivoted rows below; no rule + # is executed on this row-preserving reshape (issue #906). if not measure_names: return f"SELECT * FROM {table_src}" @@ -2846,22 +2848,8 @@ def visit_DPValidation(self, node: AST.DPValidation) -> str: # type: ignore[ove ] union_sql = " UNION ALL ".join(rule_queries) - if viral_comps: - reg = get_current_registry() - excl_list: List[str] = [] - expr_list: List[str] = [] - for comp in viral_comps: - v_rule = reg.rule_for(comp) - if v_rule is None: - continue - qn = quote_name(comp.name) - excl_list.append(qn) - expr_list.append(f"{vp_dataset_wide_sql(v_rule, qn)} AS {qn}") - if expr_list: - union_sql = ( - f"SELECT * EXCLUDE ({', '.join(excl_list)}), {', '.join(expr_list)} " - f"FROM ({union_sql}) AS _dp_viral" - ) + # check_datapoint is row-preserving: viral attributes are copied per datapoint from + # union_sql; no rule is executed (issue #906, superseding #897's dataset-wide aggregate). if use_cte: cte = CTEBuilder() @@ -3121,17 +3109,16 @@ def _build_check_hierarchy_sql( # rule over the result (issue #877). viral_comps = [c for c in ds.components.values() if c.role == Role.VIRAL_ATTRIBUTE] if viral_comps: - reg = get_current_registry() + get_current_registry() keys = [*other_ids, rule_comp] keys_q = ", ".join(quote_name(k) for k in keys) raw = ", ".join(quote_name(c.name) for c in viral_comps) cte.cte("_chv", f"SELECT {keys_q}, {raw} FROM {table_src}") viral_sel = [] for c in viral_comps: - v_rule = reg.rule_for(c) + # check_hierarchy is row-preserving: viral attribute copied unchanged (#906). qn = quote_name(c.name) - expr = f"v.{qn}" if v_rule is None else vp_dataset_wide_sql(v_rule, f"v.{qn}") - viral_sel.append(f"{expr} AS {qn}") + viral_sel.append(f"v.{qn} AS {qn}") join = " AND ".join( f"r.{quote_name(k)} IS NOT DISTINCT FROM v.{quote_name(k)}" for k in keys ) diff --git a/tests/ViralAttributes/data/DataSet/output/13-1-DS_r.csv b/tests/ViralAttributes/data/DataSet/output/13-1-DS_r.csv index 09ce9c9dc..52ca8176b 100644 --- a/tests/ViralAttributes/data/DataSet/output/13-1-DS_r.csv +++ b/tests/ViralAttributes/data/DataSet/output/13-1-DS_r.csv @@ -1,3 +1,3 @@ Id_1,ruleid,Me_1,errorcode,errorlevel,VAt_1 -2,r1,20.0,e1,1,300.0 +2,r1,20.0,e1,1,100.0 1,r2,1.0,e2,1,300.0 diff --git a/tests/ViralAttributes/test_viral_rule_execution.py b/tests/ViralAttributes/test_viral_rule_execution.py index 06a978470..f9011a6b7 100644 --- a/tests/ViralAttributes/test_viral_rule_execution.py +++ b/tests/ViralAttributes/test_viral_rule_execution.py @@ -1,19 +1,27 @@ -"""Viral propagation rule execution across operators (issue #877, ported from 1.9.X). - -Covers the operator-level propagation and rule-execution behaviour introduced by -#878: unpivot, period_indicator, check_datapoint, hierarchy and check_hierarchy -propagate viral attributes and execute the propagation rule; row-preserving -operators execute aggregate rules dataset-wide and enumerated rules per row; and -the strict semantic checks 1-3-3-5 (aggregate sum/avg needs a numeric viral -attribute) and 1-3-3-6 (every viral attribute requires a rule). +"""Viral propagation rule execution across operators (issues #877, #906). + +The propagation rule is executed (and required) ONLY where input data points are +combined: an operation over two or more datasets, an aggregation/analytic group-by, +or a hierarchy roll-up. Row-preserving operators (unary, dataset-scalar, unpivot, +period_indicator, check_datapoint, check_hierarchy) COPY viral attributes unchanged. +Also covers the strict semantic checks 1-3-3-5 (aggregate sum/avg needs a numeric +viral attribute) and 1-3-3-6 (a combined viral attribute requires a rule). """ import pandas as pd import pytest from vtlengine import run, semantic_analysis +from vtlengine.DataTypes import Integer, String from vtlengine.Exceptions import SemanticError -from vtlengine.Model import Role +from vtlengine.Model import Component, Dataset, Role +from vtlengine.ViralPropagation import ( + ViralPropagationRegistry, + ViralPropagationRule, + combined_viral_components, + require_rules, + set_current_registry, +) # -- Shared propagation rules -- @@ -117,7 +125,9 @@ def test_unpivot_replicates_viral_attrs(self) -> None: for _, row in ds_r.data.iterrows(): assert row["VAt_1"] == expected[row["Id_1"]] - def test_unpivot_executes_aggregate_rule(self) -> None: + def test_unpivot_copies_viral_not_aggregate(self) -> None: + """Unpivot is row-preserving: an aggregate rule must NOT collapse the viral column; + each source row's value is copied (replicated) across the unpivoted rows (#906).""" df = pd.DataFrame( {"Id_1": [1, 2], "Me_1": [10.0, 20.0], "Me_2": [100.0, 200.0], "VAt_1": [1, 2]} ) @@ -126,8 +136,10 @@ def test_unpivot_executes_aggregate_rule(self) -> None: data_structures={"datasets": [self._ds("Number")]}, datapoints={"DS_1": df}, ) - # aggregate max over the source VAt_1 [1, 2] = 2, replicated to all 4 melted rows - assert list(result["DS_r"].data["VAt_1"]) == [2, 2, 2, 2] + # Copied per source row (NOT collapsed to the dataset-wide max 2). + expected = {1: 1, 2: 2} + for _, row in result["DS_r"].data.iterrows(): + assert row["VAt_1"] == expected[row["Id_1"]] # -- Period_indicator time operator -- @@ -163,15 +175,17 @@ def test_period_indicator_preserves_viral_attrs(self) -> None: for _, row in ds_r.data.iterrows(): assert row["VAt_1"] == expected[str(row["Id_1"])] - def test_period_indicator_executes_aggregate_rule(self) -> None: + def test_period_indicator_copies_viral_not_aggregate(self) -> None: + """period_indicator is row-preserving: an aggregate rule must NOT collapse the + viral column; each row's value is copied unchanged (issue #906).""" df = pd.DataFrame({"Id_1": ["2020-01", "2020-02"], "Me_1": [1.0, 2.0], "VAt_1": [100, 200]}) result = run( script=f"{AGGR_MAX_RULE}DS_r <- period_indicator(DS_1);", data_structures={"datasets": [self._ds("Number")]}, datapoints={"DS_1": df}, ) - # aggregate max over the whole dataset -> 200 on every result row - assert list(result["DS_r"].data["VAt_1"]) == [200, 200] + # Copied per row (NOT collapsed to the dataset-wide max 200). + assert sorted(result["DS_r"].data["VAt_1"].tolist()) == [100, 200] # -- check_datapoint validation operator -- @@ -205,27 +219,31 @@ def test_check_datapoint_reattaches_viral(self, output: str) -> None: for _, row in ds_r.data.iterrows(): assert row["VAt_1"] == expected[row["Id_1"]] - def test_check_datapoint_executes_aggregate_rule(self) -> None: + def test_check_datapoint_copies_viral_not_aggregate(self) -> None: + """check_datapoint is row-preserving: an aggregate rule must NOT collapse the viral + column; each source datapoint's value is copied unchanged (issue #906, superseding + #897's dataset-wide aggregate).""" df = pd.DataFrame({"Id_1": [1, 2, 3], "Me_1": [10.0, 20.0, 30.0], "VAt_1": [100, 200, 300]}) result = run( script=f"{AGGR_MAX_RULE}{_DPR}\nDS_r <- check_datapoint(DS_1, R all);", data_structures={"datasets": [_single_va_ds("Number")]}, datapoints={"DS_1": df}, ) - # aggregate max over all datapoints -> 300 on every validation row - assert list(result["DS_r"].data["VAt_1"]) == [300, 300, 300] + # Copied per datapoint (NOT collapsed to the dataset-wide max 300). + d = result["DS_r"].data.sort_values("Id_1") + assert list(d["VAt_1"]) == [100, 200, 300] # -- Rule execution on row-preserving dataset-level operators (issue #877) -- class TestViralRuleExecutionRowPreserving: - """Row-preserving dataset-level operators must EXECUTE the rule: aggregate rules - collapse over the whole dataset (one value on every row); enumerated rules map - per row (issue #877).""" + """Row-preserving dataset-level operators must COPY the viral attribute, NOT execute + the rule: an aggregate rule does not collapse it and an enumerated rule does not remap + it, because no data points are combined (issue #906).""" @pytest.mark.parametrize("expr", ["round(DS_1)", "abs(DS_1)", "DS_1 * 2"]) - def test_aggregate_rule_is_dataset_wide(self, expr: str) -> None: + def test_aggregate_rule_copies_per_row(self, expr: str) -> None: result = run( script=f"{AGGR_MAX_RULE}DS_r <- {expr};", data_structures={"datasets": [_single_va_ds("Number")]}, @@ -235,10 +253,11 @@ def test_aggregate_rule_is_dataset_wide(self, expr: str) -> None: ) }, ) - # aggregate max over the whole dataset -> 300 on every result row - assert list(result["DS_r"].data["VAt_1"]) == [300, 300, 300] + # Copied per row (NOT collapsed to the dataset-wide max 300). + d = result["DS_r"].data.sort_values("Id_1") + assert list(d["VAt_1"]) == [100, 200, 300] - def test_enumerated_rule_is_per_row(self) -> None: + def test_enumerated_rule_copies_per_row(self) -> None: result = run( script=f"{_ENUM_RULE}\nDS_r <- round(DS_1);", data_structures={"datasets": [_single_va_ds()]}, @@ -247,8 +266,8 @@ def test_enumerated_rule_is_per_row(self) -> None: }, ) df = result["DS_r"].data.sort_values("Id_1") - # "A" matches the unary clause -> "Z"; "B" is unmatched -> else "D" - assert list(df["VAt_1"]) == ["Z", "D"] + # Copied unchanged (NOT remapped "A"->"Z", "B"->"D"). + assert list(df["VAt_1"]) == ["A", "B"] # -- hierarchy aggregation operator -- @@ -386,82 +405,101 @@ def test_valuedomain_sum_avg_non_numeric_raises(self, fn: str) -> None: # -- Every viral attribute requires a rule (issue #877) -- -class TestEveryViralAttributeRequiresRule: - """Strict policy: a viral attribute reaching a result without a ``define viral - propagation`` rule is a SemanticError (1-3-3-6), even when no combination happens - (single-operand assignment, unary, keep, calc). A rule is not optional.""" +class TestViralRuleRequiredOnlyWhenCombined: + """Per the VTL 2.2 attribute propagation rule, a rule is required (and executed) only + where input data points are combined. A viral attribute merely copied through a + row-preserving / single-operand operator needs no rule (issue #906).""" - def test_identity_assignment_no_rule_raises(self) -> None: - """``DS_r <- DS_1`` with a viral attribute and no rule → error.""" - with pytest.raises(SemanticError) as exc: - run( - script="DS_r <- DS_1;", - data_structures={"datasets": [DS_1VA]}, - datapoints={"DS_1": pd.DataFrame({"Id_1": [1], "Me_1": [10.0], "VAt_1": ["A"]})}, - ) - assert "1-3-3-6" in str(exc.value) + # -- non-combining operators: viral attribute copied through, no rule required -- - def test_unary_no_rule_raises(self) -> None: - """A unary (row-preserving) operator over a viral attribute with no rule → error.""" - with pytest.raises(SemanticError) as exc: - run( - script="DS_r <- abs(DS_1);", - data_structures={"datasets": [DS_1VA]}, - datapoints={"DS_1": pd.DataFrame({"Id_1": [1], "Me_1": [-10.0], "VAt_1": ["A"]})}, - ) - assert "1-3-3-6" in str(exc.value) + def test_identity_assignment_no_rule_ok(self) -> None: + result = run( + script="DS_r <- DS_1;", + data_structures={"datasets": [DS_1VA]}, + datapoints={"DS_1": pd.DataFrame({"Id_1": [1], "Me_1": [10.0], "VAt_1": ["A"]})}, + ) + assert result["DS_r"].components["VAt_1"].role == Role.VIRAL_ATTRIBUTE + assert list(result["DS_r"].data["VAt_1"]) == ["A"] - def test_keep_no_rule_raises(self) -> None: - """A keep clause carries the viral attribute through; without a rule → error.""" - with pytest.raises(SemanticError) as exc: - run( - script="DS_r <- DS_1[keep Me_1];", - data_structures={"datasets": [DS_1VA]}, - datapoints={"DS_1": pd.DataFrame({"Id_1": [1], "Me_1": [10.0], "VAt_1": ["A"]})}, - ) - assert "1-3-3-6" in str(exc.value) + def test_unary_no_rule_ok(self) -> None: + result = run( + script="DS_r <- abs(DS_1);", + data_structures={"datasets": [DS_1VA]}, + datapoints={"DS_1": pd.DataFrame({"Id_1": [1], "Me_1": [-10.0], "VAt_1": ["A"]})}, + ) + assert list(result["DS_r"].data["VAt_1"]) == ["A"] + + def test_keep_no_rule_ok(self) -> None: + result = run( + script="DS_r <- DS_1[keep Me_1];", + data_structures={"datasets": [DS_1VA]}, + datapoints={"DS_1": pd.DataFrame({"Id_1": [1], "Me_1": [10.0], "VAt_1": ["A"]})}, + ) + assert result["DS_r"].components["VAt_1"].role == Role.VIRAL_ATTRIBUTE + assert list(result["DS_r"].data["VAt_1"]) == ["A"] + + def test_calc_creates_viral_no_rule_ok(self) -> None: + result = run( + script='DS_r <- DS_1[calc viral attribute VAt_1 := "X"];', + data_structures={"datasets": [DS_NO_VA]}, + datapoints={"DS_1": pd.DataFrame({"Id_1": [1], "Me_1": [10.0]})}, + ) + assert result["DS_r"].components["VAt_1"].role == Role.VIRAL_ATTRIBUTE + assert list(result["DS_r"].data["VAt_1"]) == ["X"] - def test_calc_creates_viral_no_rule_raises(self) -> None: - """A calc that creates a viral attribute must also declare a rule for it.""" + def test_two_viral_partial_rule_passthrough_ok(self) -> None: + # rule only for VAt_1; VAt_2 has none — pure passthrough combines neither. + result = run( + script=AGGR_MAX_RULE + "DS_r <- DS_1;", + data_structures={"datasets": [DS_2VA]}, + datapoints={ + "DS_1": pd.DataFrame({"Id_1": [1], "Me_1": [10.0], "VAt_1": ["A"], "VAt_2": [7]}) + }, + ) + assert set(result["DS_r"].get_viral_attributes_names()) == {"VAt_1", "VAt_2"} + + def test_semantic_analysis_no_rule_ok(self) -> None: + result = semantic_analysis( + script="DS_r <- DS_1;", + data_structures={"datasets": [DS_1VA]}, + ) + assert result["DS_r"].components["VAt_1"].role == Role.VIRAL_ATTRIBUTE + + # -- combination points: a rule IS required (SemanticError 1-3-3-6) -- + + def test_aggregation_no_rule_raises(self) -> None: with pytest.raises(SemanticError) as exc: - run( - script='DS_r <- DS_1[calc viral attribute VAt_1 := "X"];', - data_structures={"datasets": [DS_NO_VA]}, - datapoints={"DS_1": pd.DataFrame({"Id_1": [1], "Me_1": [10.0]})}, + semantic_analysis( + script="DS_r <- sum(DS_1 group by Id_1);", + data_structures={"datasets": [DS_1VA]}, ) assert "1-3-3-6" in str(exc.value) - def test_partial_rules_missing_one_raises(self) -> None: - """Two viral attributes but only one rule → the un-ruled one still errors.""" + def test_analytic_no_rule_raises(self) -> None: with pytest.raises(SemanticError) as exc: - run( - script=AGGR_MAX_RULE + "DS_r <- DS_1;", # rule only for VAt_1 - data_structures={"datasets": [DS_2VA]}, - datapoints={ - "DS_1": pd.DataFrame( - {"Id_1": [1], "Me_1": [10.0], "VAt_1": ["A"], "VAt_2": [7]} - ) - }, + semantic_analysis( + script="DS_r <- sum(DS_1 over (partition by Id_1));", + data_structures={"datasets": [DS_1VA]}, ) - # VAt_2 has no rule. assert "1-3-3-6" in str(exc.value) - assert "VAt_2" in str(exc.value) - def test_semantic_analysis_no_rule_raises(self) -> None: - """The rule requirement is enforced at semantic-analysis time (no execution).""" + def test_binary_two_datasets_no_rule_raises(self) -> None: with pytest.raises(SemanticError) as exc: semantic_analysis( - script="DS_r <- DS_1;", - data_structures={"datasets": [DS_1VA]}, + script="DS_r <- DS_1 + DS_2;", + data_structures={"datasets": [DS_1VA, {**DS_1VA, "name": "DS_2"}]}, ) assert "1-3-3-6" in str(exc.value) def test_rule_present_does_not_raise(self) -> None: - """Positive control: declaring the rule makes the same script valid.""" + """Positive control: declaring the rule makes a combining script valid.""" result = run( - script=CONF_RULE + "DS_r <- DS_1;", - data_structures={"datasets": [DS_1VA]}, - datapoints={"DS_1": pd.DataFrame({"Id_1": [1], "Me_1": [10.0], "VAt_1": ["C"]})}, + script=CONF_RULE + "DS_r <- DS_1 + DS_2;", + data_structures={"datasets": [DS_1VA, {**DS_1VA, "name": "DS_2"}]}, + datapoints={ + "DS_1": pd.DataFrame({"Id_1": [1], "Me_1": [10.0], "VAt_1": ["C"]}), + "DS_2": pd.DataFrame({"Id_1": [1], "Me_1": [5.0], "VAt_1": ["N"]}), + }, ) assert result["DS_r"].components["VAt_1"].role == Role.VIRAL_ATTRIBUTE @@ -488,3 +526,47 @@ def test_rule_survives_multi_statement_script(self) -> None: ) # The rule is registered and applied: max("A", "A") = "A", not NULL. assert list(result["DS_r"].data["VAt_1"]) == ["A"] + + +# -- Unit tests for the combination-point check helpers (issue #906) -- + + +def _viral_ds(name: str, viral_names: list) -> Dataset: + comps = {"Id_1": Component("Id_1", Integer, Role.IDENTIFIER, False)} + for v in viral_names: + comps[v] = Component(v, String, Role.VIRAL_ATTRIBUTE, True) + return Dataset(name=name, components=comps, data=None) + + +class TestViralCheckHelpers: + """``require_rules`` / ``combined_viral_components`` back the combination-point check.""" + + def test_require_rules_raises_when_missing(self) -> None: + set_current_registry(ViralPropagationRegistry()) + comp = Component("VAt_1", String, Role.VIRAL_ATTRIBUTE, True) + with pytest.raises(SemanticError) as exc: + require_rules([comp]) + assert "1-3-3-6" in str(exc.value) + + def test_require_rules_passes_when_present(self) -> None: + registry = ViralPropagationRegistry() + registry.register( + ViralPropagationRule( + name="VAt_1", + signature_type="variable", + target="VAt_1", + enumerated_clauses=[], + aggregate_function="max", + ) + ) + set_current_registry(registry) + require_rules([Component("VAt_1", String, Role.VIRAL_ATTRIBUTE, True)]) # must not raise + + def test_combined_viral_components_only_shared(self) -> None: + combined = combined_viral_components( + [_viral_ds("A", ["VAt_1", "VAt_2"]), _viral_ds("B", ["VAt_1"])] + ) + assert {c.name for c in combined} == {"VAt_1"} + + def test_combined_viral_components_empty_for_single_operand(self) -> None: + assert combined_viral_components([_viral_ds("A", ["VAt_1"])]) == [] From 4703c667d68cd6447b19fe049b0dcf2864707974 Mon Sep 17 00:00:00 2001 From: Javier Hernandez Date: Thu, 16 Jul 2026 15:10:53 +0200 Subject: [PATCH 2/6] test: assert dataset-scalar binary copies viral (no rule executed/required) (cr-906-main) --- .../test_viral_rule_execution.py | 19 +++++++++++++++++++ 1 file changed, 19 insertions(+) diff --git a/tests/ViralAttributes/test_viral_rule_execution.py b/tests/ViralAttributes/test_viral_rule_execution.py index f9011a6b7..6587727e2 100644 --- a/tests/ViralAttributes/test_viral_rule_execution.py +++ b/tests/ViralAttributes/test_viral_rule_execution.py @@ -269,6 +269,25 @@ def test_enumerated_rule_copies_per_row(self) -> None: # Copied unchanged (NOT remapped "A"->"Z", "B"->"D"). assert list(df["VAt_1"]) == ["A", "B"] + def test_dataset_scalar_binary_copies_and_needs_no_rule(self) -> None: + """A dataset-scalar binary (DS ⊕ scalar) is row-preserving: it copies the viral + attribute (an aggregate rule must NOT collapse it) and requires no rule (#906).""" + dp = pd.DataFrame({"Id_1": [1, 2, 3], "Me_1": [1.0, 2.0, 3.0], "VAt_1": [100, 200, 300]}) + # (a) rule present but not executed -> values copied per row (not the dataset-wide max) + result = run( + script=f"{AGGR_MAX_RULE}DS_r <- DS_1 + 5;", + data_structures={"datasets": [_single_va_ds("Number")]}, + datapoints={"DS_1": dp}, + ) + assert list(result["DS_r"].data.sort_values("Id_1")["VAt_1"]) == [100, 200, 300] + # (b) no rule at all -> succeeds (no 1-3-3-6), values copied + result = run( + script="DS_r <- DS_1 + 5;", + data_structures={"datasets": [_single_va_ds("Number")]}, + datapoints={"DS_1": dp}, + ) + assert list(result["DS_r"].data.sort_values("Id_1")["VAt_1"]) == [100, 200, 300] + # -- hierarchy aggregation operator -- From f62fa64bc8bbab2e92270bbcae5a9d14aa758603 Mon Sep 17 00:00:00 2001 From: Javier Hernandez Date: Thu, 16 Jul 2026 16:18:18 +0200 Subject: [PATCH 3/6] test: real operators in viral row-preserving tests + cover rule combine in group/partition (cr-906-main) --- .../test_viral_rule_execution.py | 129 +++++++++++++++++- 1 file changed, 123 insertions(+), 6 deletions(-) diff --git a/tests/ViralAttributes/test_viral_rule_execution.py b/tests/ViralAttributes/test_viral_rule_execution.py index 6587727e2..d8052962f 100644 --- a/tests/ViralAttributes/test_viral_rule_execution.py +++ b/tests/ViralAttributes/test_viral_rule_execution.py @@ -431,9 +431,10 @@ class TestViralRuleRequiredOnlyWhenCombined: # -- non-combining operators: viral attribute copied through, no rule required -- - def test_identity_assignment_no_rule_ok(self) -> None: + def test_calc_measure_no_rule_ok(self) -> None: + """A calc clause is row-preserving: it copies the viral attribute; no rule needed.""" result = run( - script="DS_r <- DS_1;", + script="DS_r <- DS_1[calc Me_2 := Me_1 * 2];", data_structures={"datasets": [DS_1VA]}, datapoints={"DS_1": pd.DataFrame({"Id_1": [1], "Me_1": [10.0], "VAt_1": ["A"]})}, ) @@ -466,10 +467,12 @@ def test_calc_creates_viral_no_rule_ok(self) -> None: assert result["DS_r"].components["VAt_1"].role == Role.VIRAL_ATTRIBUTE assert list(result["DS_r"].data["VAt_1"]) == ["X"] - def test_two_viral_partial_rule_passthrough_ok(self) -> None: - # rule only for VAt_1; VAt_2 has none — pure passthrough combines neither. + def test_two_viral_partial_rule_row_preserving_ok(self) -> None: + """Two viral attributes, a rule for only one, through a row-preserving operator → + no error (neither is combined, so even the un-ruled VAt_2 is fine).""" + # rule only for VAt_1; VAt_2 has none — a row-preserving op combines neither. result = run( - script=AGGR_MAX_RULE + "DS_r <- DS_1;", + script=AGGR_MAX_RULE + "DS_r <- abs(DS_1);", data_structures={"datasets": [DS_2VA]}, datapoints={ "DS_1": pd.DataFrame({"Id_1": [1], "Me_1": [10.0], "VAt_1": ["A"], "VAt_2": [7]}) @@ -478,8 +481,9 @@ def test_two_viral_partial_rule_passthrough_ok(self) -> None: assert set(result["DS_r"].get_viral_attributes_names()) == {"VAt_1", "VAt_2"} def test_semantic_analysis_no_rule_ok(self) -> None: + """A row-preserving op with no rule passes semantic analysis (no execution).""" result = semantic_analysis( - script="DS_r <- DS_1;", + script="DS_r <- DS_1 * 2;", data_structures={"datasets": [DS_1VA]}, ) assert result["DS_r"].components["VAt_1"].role == Role.VIRAL_ATTRIBUTE @@ -523,6 +527,119 @@ def test_rule_present_does_not_raise(self) -> None: assert result["DS_r"].components["VAt_1"].role == Role.VIRAL_ATTRIBUTE +# -- The rule COMBINES per group / partition at aggregation & analytic (issue #906) -- + +# Two identifiers; group/partition Id_1=1 -> {10, null, 30}; Id_1=2 -> {5}. +_GP_NUM_DS = { + "name": "DS_1", + "DataStructure": [ + {"name": "Id_1", "type": "Integer", "role": "Identifier", "nullable": False}, + {"name": "Id_2", "type": "Integer", "role": "Identifier", "nullable": False}, + {"name": "Me_1", "type": "Number", "role": "Measure", "nullable": True}, + {"name": "VAt_1", "type": "Number", "role": "Viral Attribute", "nullable": True}, + ], +} +_GP_STR_DS = { + **_GP_NUM_DS, + "DataStructure": [ + *_GP_NUM_DS["DataStructure"][:3], + {"name": "VAt_1", "type": "String", "role": "Viral Attribute", "nullable": True}, + ], +} + +_ENUM_PAIR_RULE = """ + define viral propagation R (variable VAt_1) is + when "A" and "B" then "AB"; + when "C" and "D" then "CD"; + else "F" + end viral propagation; +""" + + +def _gp_num_dp() -> pd.DataFrame: + return pd.DataFrame( + { + "Id_1": [1, 1, 1, 2], + "Id_2": [1, 2, 3, 1], + "Me_1": [10.0, 20.0, 30.0, 40.0], + "VAt_1": [10.0, None, 30.0, 5.0], + } + ) + + +def _gp_str_dp() -> pd.DataFrame: + return pd.DataFrame( + { + "Id_1": [1, 1, 2, 2], + "Id_2": [1, 2, 1, 2], + "Me_1": [10.0, 20.0, 30.0, 40.0], + "VAt_1": ["A", "B", "C", "D"], + } + ) + + +class TestViralRuleCombinesInGroupAndPartition: + """At the aggregation and analytic combination points the propagation rule is executed + within each group / partition (issue #906): an aggregate rule combines the group's + values (skipping nulls), analytic broadcasts the combined value to every row of the + partition, and an enumerated rule combines the group's values through its clauses.""" + + @pytest.mark.parametrize( + "agg_fn, group1", + # group Id_1=1 = {10, null, 30}; nulls are skipped: min 10, max 30, sum 40, avg 20. + [("min", 10.0), ("max", 30.0), ("sum", 40.0), ("avg", 20.0)], + ) + def test_aggregate_rule_combines_per_group_skipping_nulls( + self, agg_fn: str, group1: float + ) -> None: + rule = ( + f"define viral propagation S (variable VAt_1) is\n" + f" aggregate {agg_fn}\n" + f"end viral propagation;\n" + ) + result = run( + script=rule + "DS_r <- sum(DS_1 group by Id_1);", + data_structures={"datasets": [_GP_NUM_DS]}, + datapoints={"DS_1": _gp_num_dp()}, + ) + d = result["DS_r"].data.sort_values("Id_1").reset_index(drop=True) + # One combined value per group; group Id_1=2 keeps its lone value 5. + assert d["VAt_1"].iloc[0] == group1 + assert d["VAt_1"].iloc[1] == 5.0 + + def test_analytic_rule_combines_per_partition_and_broadcasts(self) -> None: + result = run( + script=AGGR_MAX_RULE + "DS_r <- sum(DS_1 over (partition by Id_1));", + data_structures={"datasets": [_GP_NUM_DS]}, + datapoints={"DS_1": _gp_num_dp()}, + ) + d = result["DS_r"].data.sort_values(["Id_1", "Id_2"]).reset_index(drop=True) + # max over partition Id_1=1 {10, null, 30} = 30, broadcast to all 3 rows; partition 2 = 5. + assert list(d[d["Id_1"] == 1]["VAt_1"]) == [30.0, 30.0, 30.0] + assert list(d[d["Id_1"] == 2]["VAt_1"]) == [5.0] + + def test_enumerated_rule_combines_per_group(self) -> None: + result = run( + script=_ENUM_PAIR_RULE + "DS_r <- sum(DS_1 group by Id_1);", + data_structures={"datasets": [_GP_STR_DS]}, + datapoints={"DS_1": _gp_str_dp()}, + ) + d = result["DS_r"].data.sort_values("Id_1").reset_index(drop=True) + # group {"A","B"} matches the binary clause -> "AB"; group {"C","D"} -> "CD". + assert list(d["VAt_1"]) == ["AB", "CD"] + + def test_enumerated_rule_combines_per_partition(self) -> None: + result = run( + script=_ENUM_PAIR_RULE + "DS_r <- sum(DS_1 over (partition by Id_1));", + data_structures={"datasets": [_GP_STR_DS]}, + datapoints={"DS_1": _gp_str_dp()}, + ) + d = result["DS_r"].data.sort_values(["Id_1", "Id_2"]).reset_index(drop=True) + # Combined per partition and broadcast to each row. + assert list(d[d["Id_1"] == 1]["VAt_1"]) == ["AB", "AB"] + assert list(d[d["Id_1"] == 2]["VAt_1"]) == ["CD", "CD"] + + # -- DAG statement sorting keeps the rule registered (issue #877) -- From 7ae070af20154f01b24005e8e5d800fbbe0f15f7 Mon Sep 17 00:00:00 2001 From: Javier Hernandez Date: Thu, 16 Jul 2026 17:01:35 +0200 Subject: [PATCH 4/6] fix: apply enumerated viral propagation rule to single-element groups in DuckDB (cr-906-main) --- src/vtlengine/ViralPropagation/sql.py | 23 ++++++++++--- .../test_viral_rule_execution.py | 32 +++++++++++++++++++ 2 files changed, 51 insertions(+), 4 deletions(-) diff --git a/src/vtlengine/ViralPropagation/sql.py b/src/vtlengine/ViralPropagation/sql.py index 3e8eb8e04..58a103e18 100644 --- a/src/vtlengine/ViralPropagation/sql.py +++ b/src/vtlengine/ViralPropagation/sql.py @@ -96,20 +96,35 @@ 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: diff --git a/tests/ViralAttributes/test_viral_rule_execution.py b/tests/ViralAttributes/test_viral_rule_execution.py index d8052962f..327723aca 100644 --- a/tests/ViralAttributes/test_viral_rule_execution.py +++ b/tests/ViralAttributes/test_viral_rule_execution.py @@ -639,6 +639,38 @@ def test_enumerated_rule_combines_per_partition(self) -> None: assert list(d[d["Id_1"] == 1]["VAt_1"]) == ["AB", "AB"] assert list(d[d["Id_1"] == 2]["VAt_1"]) == ["CD", "CD"] + @pytest.mark.parametrize( + "invocation", + ["sum(DS_1 group by Id_1)", "sum(DS_1 over (partition by Id_1))"], + ) + def test_enumerated_rule_applies_to_single_element_group(self, invocation: str) -> None: + """A group / partition with a single data point still has the enumerated rule + applied to it, exactly as a larger group does — the rule runs in every group + regardless of size (regression: DuckDB's ``list_reduce`` skipped the lambda for a + one-element list, leaving the lone value unmapped, issue #906).""" + rule = ( + "define viral propagation R (variable VAt_1) is\n" + ' when "A" then "A1";\n' + ' else "F"\n' + "end viral propagation;\n" + ) + # group/partition Id_1=1 -> two rows {"A","A"}; Id_1=2 -> a lone {"A"}. + dp = pd.DataFrame( + { + "Id_1": [1, 1, 2], + "Id_2": [1, 2, 1], + "Me_1": [10.0, 20.0, 30.0], + "VAt_1": ["A", "A", "A"], + } + ) + result = run( + script=rule + f"DS_r <- {invocation};", + data_structures={"datasets": [_GP_STR_DS]}, + datapoints={"DS_1": dp}, + ) + # Every output row maps "A" -> "A1"; the lone group is NOT copied through as "A". + assert set(result["DS_r"].data["VAt_1"]) == {"A1"} + # -- DAG statement sorting keeps the rule registered (issue #877) -- From 4628e055b5aaa1f78e18438b143799f21cae8267 Mon Sep 17 00:00:00 2001 From: Javier Hernandez Date: Thu, 16 Jul 2026 17:37:45 +0200 Subject: [PATCH 5/6] test: add join viral + row-preserving-copy coverage to align main with 1.9.X (cr-906-main) --- .../test_viral_rule_execution.py | 207 ++++++++++++++++++ 1 file changed, 207 insertions(+) diff --git a/tests/ViralAttributes/test_viral_rule_execution.py b/tests/ViralAttributes/test_viral_rule_execution.py index 327723aca..b27d63d6c 100644 --- a/tests/ViralAttributes/test_viral_rule_execution.py +++ b/tests/ViralAttributes/test_viral_rule_execution.py @@ -738,3 +738,210 @@ def test_combined_viral_components_only_shared(self) -> None: def test_combined_viral_components_empty_for_single_operand(self) -> None: assert combined_viral_components([_viral_ds("A", ["VAt_1"])]) == [] + + +# -- Row-preserving operators copy the viral attribute, they do NOT execute the rule (#906) -- + +NUM_VA_2ID = { + "name": "DS_1", + "DataStructure": [ + {"name": "Id_1", "type": "Integer", "role": "Identifier", "nullable": False}, + {"name": "Id_2", "type": "String", "role": "Identifier", "nullable": False}, + {"name": "Me_1", "type": "Number", "role": "Measure", "nullable": True}, + {"name": "VAt_1", "type": "Number", "role": "Viral Attribute", "nullable": True}, + ], +} + +ENUM_REMAP_RULE = """ + define viral propagation R (variable VAt_1) is + when "A" then "Z"; + else "F" + end viral propagation; +""" + + +class TestRowPreservingCopiesViral: + """A rule may be declared, but a row-preserving operator copies the viral attribute + unchanged: an aggregate rule must NOT collapse it and an enumerated rule must NOT + remap it, because no data points are combined (issue #906).""" + + def test_unary_aggregate_rule_copies_not_collapses(self) -> None: + result = run( + script=AGGR_MAX_RULE + "DS_r <- abs(DS_1);", + data_structures={"datasets": [NUM_VA_2ID]}, + datapoints={ + "DS_1": pd.DataFrame( + { + "Id_1": [1, 1, 2], + "Id_2": ["A", "B", "A"], + "Me_1": [-1.0, -2.0, -3.0], + "VAt_1": [10.0, None, 30.0], + } + ) + }, + ) + d = result["DS_r"].data.sort_values(["Id_1", "Id_2"]).reset_index(drop=True) + # Copied per row, NOT collapsed to the dataset-wide max (30). + assert d["VAt_1"].iloc[0] == 10.0 + assert pd.isna(d["VAt_1"].iloc[1]) + assert d["VAt_1"].iloc[2] == 30.0 + + def test_scalar_enumerated_rule_copies_not_remaps(self) -> None: + result = run( + script=ENUM_REMAP_RULE + "DS_r <- DS_1 + 5;", + data_structures={"datasets": [DS_1VA]}, + datapoints={ + "DS_1": pd.DataFrame({"Id_1": [1, 2], "Me_1": [10.0, 20.0], "VAt_1": ["A", "B"]}) + }, + ) + d = result["DS_r"].data.sort_values("Id_1").reset_index(drop=True) + # "A" copied (NOT remapped to "Z"); "B" copied (NOT defaulted to "F"). + assert list(d["VAt_1"]) == ["A", "B"] + + +# -- Propagation rules through join operators (#906) -- + +CONF_BINARY_RULE = """ + define viral propagation COMP_mix (variable VAt_1) is + when "C" and "M" then "N"; + when "M" then "M"; + else " " + end viral propagation; +""" + +_ID_1 = {"name": "Id_1", "type": "Integer", "role": "Identifier", "nullable": False} +_ID_2 = {"name": "Id_2", "type": "Integer", "role": "Identifier", "nullable": False} +_ME_1 = {"name": "Me_1", "type": "Number", "role": "Measure", "nullable": True} +_ME_2 = {"name": "Me_2", "type": "Number", "role": "Measure", "nullable": True} +_VA = {"name": "VAt_1", "type": "String", "role": "Viral Attribute", "nullable": True} + +DS_JOIN_1 = {"name": "DS_1", "DataStructure": [_ID_1, _ME_1, _VA]} +DS_JOIN_2 = {"name": "DS_2", "DataStructure": [_ID_1, _ME_2, _VA]} +DS_JOIN_2_NO_VA = {"name": "DS_2", "DataStructure": [_ID_1, _ME_2]} +DS_CROSS_2 = {"name": "DS_2", "DataStructure": [_ID_2, _ME_2, _VA]} +DS_CROSS_2_NO_VA = {"name": "DS_2", "DataStructure": [_ID_2, _ME_2]} + + +class TestViralPropagationJoins: + """A viral attribute shared by both join operands is combined with the + Attribute Propagation Rule, exactly like in binary operators; a viral + attribute coming from a single operand is carried over unchanged. This holds + for all four join operators, ``cross_join`` included.""" + + @pytest.mark.parametrize("join_op", ["inner_join", "left_join", "full_join"]) + def test_enumerated_propagation_join(self, join_op: str) -> None: + """Shared viral attribute is resolved by CONF_RULE inside the join.""" + result = run( + script=CONF_RULE + f"DS_r <- {join_op}(DS_1, DS_2);", + data_structures={"datasets": [DS_JOIN_1, DS_JOIN_2]}, + datapoints={ + "DS_1": pd.DataFrame( + {"Id_1": [1, 2, 3], "Me_1": [10.0, 20.0, 30.0], "VAt_1": ["C", "N", "F"]} + ), + "DS_2": pd.DataFrame( + {"Id_1": [1, 2, 3], "Me_2": [5.0, 15.0, 25.0], "VAt_1": ["N", "F", "F"]} + ), + }, + ) + ds_r = result["DS_r"] + # Single combined column, keeping the viral role (not #-qualified per operand). + assert ds_r.components["VAt_1"].role == Role.VIRAL_ATTRIBUTE + # C+N->C (unary "C"); N+F->N (unary "N"); F+F->F (else) + sorted_data = ds_r.data.sort_values("Id_1").reset_index(drop=True) + assert list(sorted_data["VAt_1"]) == ["C", "N", "F"] + + def test_binary_clause_propagation_join(self) -> None: + """Binary propagation clauses take precedence over unary ones in a join.""" + result = run( + script=CONF_BINARY_RULE + "DS_r <- left_join(DS_1, DS_2);", + data_structures={"datasets": [DS_JOIN_1, DS_JOIN_2]}, + datapoints={ + "DS_1": pd.DataFrame( + {"Id_1": [1, 2, 3], "Me_1": [10.0, 20.0, 30.0], "VAt_1": ["C", "M", "X"]} + ), + "DS_2": pd.DataFrame( + {"Id_1": [1, 2, 3], "Me_2": [5.0, 15.0, 25.0], "VAt_1": ["M", "F", "Y"]} + ), + }, + ) + # C+M->N (binary); M+F->M (unary "M"); X+Y->" " (else) + sorted_data = result["DS_r"].data.sort_values("Id_1").reset_index(drop=True) + assert list(sorted_data["VAt_1"]) == ["N", "M", " "] + + def test_enumerated_propagation_cross_join(self) -> None: + """A viral attribute shared by both cross_join operands is combined via the + propagation rule (cross_join pairs every row, so use one row per operand).""" + result = run( + script=CONF_RULE + "DS_r <- cross_join(DS_1, DS_2);", + data_structures={"datasets": [DS_JOIN_1, DS_CROSS_2]}, + datapoints={ + "DS_1": pd.DataFrame({"Id_1": [1], "Me_1": [10.0], "VAt_1": ["C"]}), + "DS_2": pd.DataFrame({"Id_2": [2], "Me_2": [5.0], "VAt_1": ["N"]}), + }, + ) + ds_r = result["DS_r"] + # Single combined viral column (not #-qualified per operand). + assert ds_r.components["VAt_1"].role == Role.VIRAL_ATTRIBUTE + assert "DS_1#VAt_1" not in ds_r.components + # C+N->C (unary "C") + assert list(ds_r.data["VAt_1"]) == ["C"] + + @pytest.mark.parametrize("join_op", ["inner_join", "left_join", "full_join"]) + def test_no_rule_combine_raises_join(self, join_op: str) -> None: + """Both operands viral but no rule defined -> SemanticError (issue #877).""" + with pytest.raises(SemanticError) as exc: + run( + script=f"DS_r <- {join_op}(DS_1, DS_2);", + data_structures={"datasets": [DS_JOIN_1, DS_JOIN_2]}, + datapoints={ + "DS_1": pd.DataFrame({"Id_1": [1], "Me_1": [10.0], "VAt_1": ["A"]}), + "DS_2": pd.DataFrame({"Id_1": [1], "Me_2": [5.0], "VAt_1": ["B"]}), + }, + ) + assert "1-3-3-6" in str(exc.value) + + def test_no_rule_combine_raises_cross_join(self) -> None: + """cross_join, both operands viral, no rule defined -> SemanticError (issue #877).""" + with pytest.raises(SemanticError) as exc: + run( + script="DS_r <- cross_join(DS_1, DS_2);", + data_structures={"datasets": [DS_JOIN_1, DS_CROSS_2]}, + datapoints={ + "DS_1": pd.DataFrame({"Id_1": [1], "Me_1": [10.0], "VAt_1": ["A"]}), + "DS_2": pd.DataFrame({"Id_2": [1], "Me_2": [5.0], "VAt_1": ["B"]}), + }, + ) + assert "1-3-3-6" in str(exc.value) + + @pytest.mark.parametrize("join_op", ["inner_join", "left_join", "full_join"]) + def test_viral_from_one_operand_kept(self, join_op: str) -> None: + """A viral attribute present in a single operand is carried over unchanged + (no propagation rule needed).""" + result = run( + script=CONF_RULE + f"DS_r <- {join_op}(DS_1, DS_2);", + data_structures={"datasets": [DS_JOIN_1, DS_JOIN_2_NO_VA]}, + datapoints={ + "DS_1": pd.DataFrame({"Id_1": [1, 2], "Me_1": [10.0, 20.0], "VAt_1": ["C", "N"]}), + "DS_2": pd.DataFrame({"Id_1": [1, 2], "Me_2": [5.0, 15.0]}), + }, + ) + ds_r = result["DS_r"] + assert ds_r.components["VAt_1"].role == Role.VIRAL_ATTRIBUTE + assert ds_r.data["VAt_1"].notna().all() + assert set(ds_r.data["VAt_1"]) == {"C", "N"} + + def test_viral_from_one_operand_kept_cross_join(self) -> None: + """cross_join: a viral attribute present in a single operand is kept + unchanged (repeated cartesian-wise).""" + result = run( + script=CONF_RULE + "DS_r <- cross_join(DS_1, DS_2);", + data_structures={"datasets": [DS_JOIN_1, DS_CROSS_2_NO_VA]}, + datapoints={ + "DS_1": pd.DataFrame({"Id_1": [1, 2], "Me_1": [10.0, 20.0], "VAt_1": ["C", "N"]}), + "DS_2": pd.DataFrame({"Id_2": [1, 2], "Me_2": [5.0, 15.0]}), + }, + ) + ds_r = result["DS_r"] + assert ds_r.components["VAt_1"].role == Role.VIRAL_ATTRIBUTE + assert ds_r.data["VAt_1"].notna().all() + assert set(ds_r.data["VAt_1"]) == {"C", "N"} From 0f43b30fb3be8cc985322526b38d54c62692b0e3 Mon Sep 17 00:00:00 2001 From: Javier Hernandez Date: Thu, 16 Jul 2026 18:00:55 +0200 Subject: [PATCH 6/6] test: port full viral coverage from 1.9.X to align main (operators, propagation, joins, keep, parsing, validation) (cr-906-main) --- .../ViralAttributes/test_viral_attributes.py | 58 ++ tests/ViralAttributes/test_viral_operators.py | 528 ++++++++++++++ .../ViralAttributes/test_viral_propagation.py | 662 ++++++++++++++++++ 3 files changed, 1248 insertions(+) create mode 100644 tests/ViralAttributes/test_viral_operators.py create mode 100644 tests/ViralAttributes/test_viral_propagation.py diff --git a/tests/ViralAttributes/test_viral_attributes.py b/tests/ViralAttributes/test_viral_attributes.py index 9c69adf15..64c4a04cb 100644 --- a/tests/ViralAttributes/test_viral_attributes.py +++ b/tests/ViralAttributes/test_viral_attributes.py @@ -4,6 +4,7 @@ import pytest from tests.Helper import TestHelper +from vtlengine import run from vtlengine.API import create_ast from vtlengine.DataTypes import Integer, Number, String from vtlengine.Exceptions import VTLSyntaxError @@ -196,3 +197,60 @@ def test_invalid_vp_body_raises_syntax_error(self, body: str) -> None: script = f"define viral propagation R (variable VAt_1) is\n{body}\nend viral propagation;" with pytest.raises(VTLSyntaxError): create_ast(script) + + def test_valid_enumerated_then_else_propagates(self) -> None: + """Valid body (enumerated clauses + trailing else): resolves on a binary op.""" + rule = ( + "define viral propagation R (variable VAt_1) is\n" + 'when "C" then "C";\nelse "F"\n' + "end viral propagation;" + ) + va = { + "name": "DS_1", + "DataStructure": [ + {"name": "Id_1", "type": "Integer", "role": "Identifier", "nullable": False}, + {"name": "Me_1", "type": "Number", "role": "Measure", "nullable": True}, + {"name": "VAt_1", "type": "String", "role": "Viral Attribute", "nullable": True}, + ], + } + result = run( + script=rule + "\nDS_r <- DS_1 + DS_2;", + data_structures={"datasets": [va, {**va, "name": "DS_2"}]}, + datapoints={ + "DS_1": pd.DataFrame({"Id_1": [1, 2], "Me_1": [10.0, 20.0], "VAt_1": ["C", "X"]}), + "DS_2": pd.DataFrame({"Id_1": [1, 2], "Me_1": [5.0, 15.0], "VAt_1": ["C", "Y"]}), + }, + ) + # C+C -> "C" (when "C"); X+Y -> "F" (else) + assert list(result["DS_r"].data["VAt_1"]) == ["C", "F"] + + def test_valid_single_aggregate_propagates(self) -> None: + """Valid body (single aggregate clause): max propagated through a group by.""" + rule = ( + "define viral propagation R (variable VAt_1) is\naggregate max\nend viral propagation;" + ) + ds = { + "name": "DS_1", + "DataStructure": [ + {"name": "Id_1", "type": "Integer", "role": "Identifier", "nullable": False}, + {"name": "Id_2", "type": "Integer", "role": "Identifier", "nullable": False}, + {"name": "Me_1", "type": "Number", "role": "Measure", "nullable": True}, + {"name": "VAt_1", "type": "Integer", "role": "Viral Attribute", "nullable": True}, + ], + } + result = run( + script=rule + "\nDS_r <- sum(DS_1 group by Id_1);", + data_structures={"datasets": [ds]}, + datapoints={ + "DS_1": pd.DataFrame( + { + "Id_1": [1, 1, 2], + "Id_2": [1, 2, 1], + "Me_1": [10.0, 20.0, 30.0], + "VAt_1": [3, 7, 5], + } + ) + }, + ) + sorted_data = result["DS_r"].data.sort_values("Id_1").reset_index(drop=True) + assert list(sorted_data["VAt_1"]) == [7, 5] diff --git a/tests/ViralAttributes/test_viral_operators.py b/tests/ViralAttributes/test_viral_operators.py new file mode 100644 index 000000000..5910bd5f5 --- /dev/null +++ b/tests/ViralAttributes/test_viral_operators.py @@ -0,0 +1,528 @@ +"""Tests for viral attribute propagation through all operator categories.""" + +import pandas as pd +import pytest + +from vtlengine import run +from vtlengine.Model import Role + +# -- Layered dataset builders -- + +BASE_COMPS = [ + {"name": "Id_1", "type": "Integer", "role": "Identifier", "nullable": False}, + {"name": "Me_1", "type": "Number", "role": "Measure", "nullable": True}, +] + +VA_1 = {"name": "VAt_1", "type": "String", "role": "Viral Attribute", "nullable": True} +VA_2 = {"name": "VAt_2", "type": "String", "role": "Viral Attribute", "nullable": True} +VA_3 = {"name": "VAt_3", "type": "String", "role": "Viral Attribute", "nullable": True} + +VA_COMPONENTS = [VA_1, VA_2, VA_3] +VA_NAMES = ["VAt_1", "VAt_2", "VAt_3"] +VA_VALUES = [["A", "B"], ["X", "Y"], ["P", "Q"]] + +# Every viral attribute requires a viral propagation rule (issue #877). Structural tests +# use `aggregate max`; value-asserting (per-row) tests use identity-enumerated rules so the +# source value is preserved on each row (an aggregate rule would collapse it dataset-wide). +VP_RULES = "".join( + f"define viral propagation VP_{n} (variable {n}) is aggregate max end viral propagation;\n" + for n in VA_NAMES +) +VP_IDENTITY = "".join( + "define viral propagation VP_{n} (variable {n}) is {clauses} end viral propagation;\n".format( + n=VA_NAMES[i], + clauses="; ".join(f'when "{v}" then "{v}"' for v in VA_VALUES[i]), + ) + for i in range(len(VA_NAMES)) +) + + +def _make_ds(name: str, num_viral: int) -> dict: + """Build a dataset definition with 0..3 viral attributes.""" + comps = BASE_COMPS + VA_COMPONENTS[:num_viral] + return {"name": name, "DataStructure": comps} + + +def _make_dp(num_viral: int) -> pd.DataFrame: + """Build datapoints matching a dataset with num_viral viral attributes.""" + data: dict = {"Id_1": [1, 2], "Me_1": [10.0, 20.0]} + for i in range(num_viral): + data[VA_NAMES[i]] = VA_VALUES[i] + return pd.DataFrame(data) + + +def _run_single(expr: str, num_viral: int, rules: str = "") -> dict: + """Run an expression with a single dataset (DS_1).""" + return run( + script=f"{rules}DS_r <- {expr};", + data_structures={"datasets": [_make_ds("DS_1", num_viral)]}, + datapoints={"DS_1": _make_dp(num_viral)}, + ) + + +def _run_pair(expr: str, num_viral: int, rules: str = "") -> dict: + """Run an expression with two datasets (DS_1, DS_2).""" + return run( + script=f"{rules}DS_r <- {expr};", + data_structures={"datasets": [_make_ds("DS_1", num_viral), _make_ds("DS_2", num_viral)]}, + datapoints={"DS_1": _make_dp(num_viral), "DS_2": _make_dp(num_viral)}, + ) + + +def _assert_viral_attrs(result: dict, num_viral: int) -> None: + """Assert that the expected viral attributes are present with correct role.""" + ds_r = result["DS_r"] + for va_name in VA_NAMES[:num_viral]: + assert va_name in ds_r.components, f"{va_name} missing from result components" + assert ds_r.components[va_name].role == Role.VIRAL_ATTRIBUTE + + +def _assert_component_data_parity(result: dict) -> None: + """Assert the result data columns match the declared components exactly.""" + ds_r = result["DS_r"] + assert set(ds_r.data.columns) == set(ds_r.components), ( + f"component/data mismatch: components={sorted(ds_r.components)}, " + f"data={sorted(ds_r.data.columns)}" + ) + + +# -- Unary operators -- + +unary_params = [ + "abs(DS_1)", + "ceil(DS_1)", + "floor(DS_1)", + "sqrt(DS_1)", + "ln(DS_1)", + "exp(DS_1)", + "isnull(DS_1)", +] + + +class TestViralAttributeUnaryOps: + @pytest.mark.parametrize("expr", unary_params) + @pytest.mark.parametrize("num_viral", [1, 2, 3]) + def test_unary_preserves_viral_attrs(self, expr: str, num_viral: int) -> None: + result = _run_single(expr, num_viral, rules=VP_RULES) + _assert_viral_attrs(result, num_viral) + + +# -- Binary operators (two datasets) -- + +binary_params = [ + "DS_1 + DS_2", + "DS_1 - DS_2", + "DS_1 * DS_2", + "DS_1 > DS_2", + "DS_1 = DS_2", +] + + +class TestViralAttributeBinaryOps: + @pytest.mark.parametrize("expr", binary_params) + @pytest.mark.parametrize("num_viral", [1, 2, 3]) + def test_binary_preserves_viral_attrs(self, expr: str, num_viral: int) -> None: + # The attribute is in both operands -> combined -> requires a rule (issue #877). + result = _run_pair(expr, num_viral, rules=VP_RULES) + _assert_viral_attrs(result, num_viral) + + +# -- Binary operators (dataset + scalar) -- + +binary_scalar_params = [ + "DS_1 + 5", + "DS_1 * 2", + "DS_1 - 1", +] + + +class TestViralAttributeBinaryScalarOps: + @pytest.mark.parametrize("expr", binary_scalar_params) + @pytest.mark.parametrize("num_viral", [1, 2, 3]) + def test_binary_scalar_preserves_viral_attrs(self, expr: str, num_viral: int) -> None: + result = _run_single(expr, num_viral, rules=VP_RULES) + _assert_viral_attrs(result, num_viral) + + +# -- Other operators -- + +other_single_params = [ + "between(DS_1, 5, 25)", +] + + +class TestViralAttributeOtherOps: + @pytest.mark.parametrize("expr", other_single_params) + @pytest.mark.parametrize("num_viral", [1, 2, 3]) + def test_other_single_preserves_viral_attrs(self, expr: str, num_viral: int) -> None: + result = _run_single(expr, num_viral, rules=VP_RULES) + _assert_viral_attrs(result, num_viral) + + @pytest.mark.parametrize("num_viral", [1, 2, 3]) + def test_intersect_preserves_viral_attrs(self, num_viral: int) -> None: + result = _run_pair("intersect(DS_1, DS_2)", num_viral, rules=VP_RULES) + _assert_viral_attrs(result, num_viral) + + @pytest.mark.parametrize("agg_op", ["sum", "avg", "count", "min", "max"]) + @pytest.mark.parametrize("num_viral", [1, 2, 3]) + def test_aggregation_preserves_viral_attrs(self, agg_op: str, num_viral: int) -> None: + comps = [ + {"name": "Id_1", "type": "Integer", "role": "Identifier", "nullable": False}, + {"name": "Id_2", "type": "Integer", "role": "Identifier", "nullable": False}, + {"name": "Me_1", "type": "Number", "role": "Measure", "nullable": True}, + ] + VA_COMPONENTS[:num_viral] + data: dict = { + "Id_1": [1, 1, 2], + "Id_2": [1, 2, 1], + "Me_1": [10.0, 20.0, 30.0], + } + for i in range(num_viral): + data[VA_NAMES[i]] = [VA_VALUES[i][0], VA_VALUES[i][0], VA_VALUES[i][1]] + result = run( + script=f"{VP_RULES}DS_r <- {agg_op}(DS_1 group by Id_1);", + data_structures={"datasets": [{"name": "DS_1", "DataStructure": comps}]}, + datapoints={"DS_1": pd.DataFrame(data)}, + ) + _assert_viral_attrs(result, num_viral) + for va_name in VA_NAMES[:num_viral]: + assert va_name in result["DS_r"].data.columns, f"{va_name} missing from result data" + + @pytest.mark.parametrize( + "expr", + [ + "DS_1[aggr Me_3 := count() group by Id_1]", + "DS_1[aggr Me_3 := sum(Me_1) group by Id_1]", + "DS_1[aggr Me_3 := avg(Me_1) group by Id_1]", + "DS_1[aggr Me_3 := min(Me_1) group by Id_1]", + "DS_1[aggr Me_3 := max(Me_1) group by Id_1]", + ], + ) + @pytest.mark.parametrize("num_viral", [1, 2, 3]) + def test_aggr_clause_preserves_viral_attrs(self, expr: str, num_viral: int) -> None: + comps = [ + {"name": "Id_1", "type": "Integer", "role": "Identifier", "nullable": False}, + {"name": "Id_2", "type": "Integer", "role": "Identifier", "nullable": False}, + {"name": "Me_1", "type": "Number", "role": "Measure", "nullable": True}, + ] + VA_COMPONENTS[:num_viral] + data: dict = { + "Id_1": [1, 1, 2], + "Id_2": [1, 2, 1], + "Me_1": [10.0, 20.0, 30.0], + } + for i in range(num_viral): + data[VA_NAMES[i]] = [VA_VALUES[i][0], VA_VALUES[i][0], VA_VALUES[i][1]] + result = run( + script=f"{VP_RULES}DS_r <- {expr};", + data_structures={"datasets": [{"name": "DS_1", "DataStructure": comps}]}, + datapoints={"DS_1": pd.DataFrame(data)}, + ) + _assert_viral_attrs(result, num_viral) + # The viral attribute column must survive the aggr clause (component/data parity). + for va_name in VA_NAMES[:num_viral]: + assert va_name in result["DS_r"].data.columns, f"{va_name} missing from result data" + + +# -- Conditional operators (if-then-else) -- + + +class TestViralAttributeConditionalOps: + """Viral attributes must survive an if-then-else whose condition is a dataset. + + The condition dataset (e.g. ``DS_1#Id_2 = "A"``) also carries the viral + attribute, which previously collided on merge with the branch operands and + corrupted the result (component/data mismatch -> downstream crash). The + viral attribute itself combines across branches like a binary operator; with + an identity rule (each value maps to itself) the per-row value is preserved.""" + + @pytest.mark.parametrize("num_viral", [1, 2, 3]) + def test_if_dataset_condition_preserves_viral_attrs(self, num_viral: int) -> None: + comps = [ + {"name": "Id_1", "type": "Integer", "role": "Identifier", "nullable": False}, + {"name": "Id_2", "type": "String", "role": "Identifier", "nullable": False}, + {"name": "Me_1", "type": "Number", "role": "Measure", "nullable": True}, + ] + VA_COMPONENTS[:num_viral] + data: dict = { + "Id_1": [1, 1, 2], + "Id_2": ["A", "B", "A"], + "Me_1": [10.0, 20.0, 30.0], + } + for i in range(num_viral): + data[VA_NAMES[i]] = [VA_VALUES[i][0], VA_VALUES[i][0], VA_VALUES[i][1]] + result = run( + script=f'{VP_IDENTITY}DS_r <- if DS_1#Id_2 = "A" then DS_1 else DS_1;', + data_structures={"datasets": [{"name": "DS_1", "DataStructure": comps}]}, + datapoints={"DS_1": pd.DataFrame(data)}, + ) + _assert_viral_attrs(result, num_viral) + _assert_component_data_parity(result) + # Both branches are DS_1 and the rule is identity -> per-row value preserved. + for i in range(num_viral): + va = VA_NAMES[i] + expected = sorted([VA_VALUES[i][0], VA_VALUES[i][0], VA_VALUES[i][1]]) + assert sorted(result["DS_r"].data[va].tolist()) == expected + + @pytest.mark.parametrize("num_viral", [1, 2, 3]) + def test_count_over_if_dataset_condition(self, num_viral: int) -> None: + """count() over an if-then-else result must not crash on viral attrs.""" + comps = [ + {"name": "Id_1", "type": "Integer", "role": "Identifier", "nullable": False}, + {"name": "Id_2", "type": "String", "role": "Identifier", "nullable": False}, + {"name": "Me_1", "type": "Number", "role": "Measure", "nullable": True}, + ] + VA_COMPONENTS[:num_viral] + data: dict = { + "Id_1": [1, 1, 2], + "Id_2": ["A", "B", "A"], + "Me_1": [10.0, 20.0, 30.0], + } + for i in range(num_viral): + data[VA_NAMES[i]] = [VA_VALUES[i][0], VA_VALUES[i][0], VA_VALUES[i][1]] + result = run( + script=f'{VP_RULES}DS_r <- count(if DS_1#Id_2 = "A" then DS_1 else DS_1 group by Id_1);', + data_structures={"datasets": [{"name": "DS_1", "DataStructure": comps}]}, + datapoints={"DS_1": pd.DataFrame(data)}, + ) + _assert_viral_attrs(result, num_viral) + for va_name in VA_NAMES[:num_viral]: + assert va_name in result["DS_r"].data.columns, f"{va_name} missing from result data" + + @pytest.mark.parametrize("num_viral", [1, 2, 3]) + def test_case_dataset_condition_preserves_viral_attrs(self, num_viral: int) -> None: + """A dataset-level ``case`` keeps viral attrs (1:1) with no phantom columns.""" + comps = [ + {"name": "Id_1", "type": "Integer", "role": "Identifier", "nullable": False}, + {"name": "Id_2", "type": "String", "role": "Identifier", "nullable": False}, + {"name": "Me_1", "type": "Number", "role": "Measure", "nullable": True}, + ] + VA_COMPONENTS[:num_viral] + data: dict = { + "Id_1": [1, 1, 2], + "Id_2": ["A", "B", "A"], + "Me_1": [10.0, 20.0, 30.0], + } + for i in range(num_viral): + data[VA_NAMES[i]] = [VA_VALUES[i][0], VA_VALUES[i][0], VA_VALUES[i][1]] + result = run( + script=f'{VP_IDENTITY}DS_r <- case when DS_1#Id_2 = "A" then DS_1 else DS_1;', + data_structures={"datasets": [{"name": "DS_1", "DataStructure": comps}]}, + datapoints={"DS_1": pd.DataFrame(data)}, + ) + _assert_viral_attrs(result, num_viral) + _assert_component_data_parity(result) + + @pytest.mark.parametrize( + "expr", + ["nvl(DS_1, DS_2)", "nvl(DS_1, 0.0)"], + ) + @pytest.mark.parametrize("num_viral", [1, 2, 3]) + def test_nvl_preserves_viral_attrs(self, expr: str, num_viral: int) -> None: + comps = BASE_COMPS + VA_COMPONENTS[:num_viral] + structures = [{"name": "DS_1", "DataStructure": comps}] + datapoints = {"DS_1": _make_dp(num_viral)} + if "DS_2" in expr: + structures.append({"name": "DS_2", "DataStructure": comps}) + datapoints["DS_2"] = _make_dp(num_viral) + result = run( + script=f"{VP_IDENTITY}DS_r <- {expr};", + data_structures={"datasets": structures}, + datapoints=datapoints, + ) + _assert_viral_attrs(result, num_viral) + _assert_component_data_parity(result) + + +# -- String parameterized operators (substr, replace, instr) -- + + +class TestViralAttributeStringParameterizedOps: + """Viral attributes must keep BOTH their component role and their data values + through parameterized string operators (substr, replace, instr).""" + + @staticmethod + def _string_ds(num_viral: int) -> dict: + comps = [ + {"name": "Id_1", "type": "Integer", "role": "Identifier", "nullable": False}, + {"name": "Me_1", "type": "String", "role": "Measure", "nullable": True}, + ] + VA_COMPONENTS[:num_viral] + return {"name": "DS_1", "DataStructure": comps} + + @staticmethod + def _string_dp(num_viral: int) -> pd.DataFrame: + data: dict = {"Id_1": [1, 2], "Me_1": ["hello", "world"]} + for i in range(num_viral): + data[VA_NAMES[i]] = VA_VALUES[i] + return pd.DataFrame(data) + + @pytest.mark.parametrize( + "expr", + [ + "substr(DS_1, 2)", + "substr(DS_1, 2, 3)", + 'replace(DS_1, "l", "L")', + 'instr(DS_1, "o")', + ], + ) + @pytest.mark.parametrize("num_viral", [1, 2, 3]) + def test_string_param_preserves_viral_attrs(self, expr: str, num_viral: int) -> None: + result = run( + script=f"{VP_IDENTITY}DS_r <- {expr};", + data_structures={"datasets": [self._string_ds(num_viral)]}, + datapoints={"DS_1": self._string_dp(num_viral)}, + ) + _assert_viral_attrs(result, num_viral) + # The viral attribute data values must be carried over unchanged (issue #782). + for i in range(num_viral): + va_name = VA_NAMES[i] + assert va_name in result["DS_r"].data.columns, f"{va_name} missing from result data" + assert list(result["DS_r"].data[va_name]) == VA_VALUES[i] + + +# -- Numeric parameterized operators (round, trunc) -- + + +class TestViralAttributeNumericParameterizedOps: + """Viral attributes must keep BOTH their component role and their data values + through parameterized numeric operators (round, trunc). + + Regression (issue #833): ``Parameterized.dataset_evaluation`` rebuilt the + result data with only identifiers and measures, so the viral attribute was + kept in ``result.components`` but dropped from ``result.data`` (component/data + mismatch).""" + + @pytest.mark.parametrize( + "expr", + [ + "round(DS_1, 2)", + "round(DS_1)", + "trunc(DS_1, 1)", + "trunc(DS_1)", + ], + ) + @pytest.mark.parametrize("num_viral", [1, 2, 3]) + def test_numeric_param_preserves_viral_attrs(self, expr: str, num_viral: int) -> None: + result = _run_single(expr, num_viral, rules=VP_IDENTITY) + _assert_viral_attrs(result, num_viral) + _assert_component_data_parity(result) + # The viral attribute data values must be carried over unchanged. + for i in range(num_viral): + va_name = VA_NAMES[i] + assert va_name in result["DS_r"].data.columns, f"{va_name} missing from result data" + assert list(result["DS_r"].data[va_name]) == VA_VALUES[i] + + +# -- Set comparison operators (in / not_in) -- + + +class TestViralAttributeCompOps: + """Viral attributes must keep BOTH their component role and their data values + through the operators ``in`` and ``not_in``.""" + + @pytest.mark.parametrize("expr", ["DS_1 in {10, 30}", "DS_1 not_in {10, 30}"]) + @pytest.mark.parametrize("num_viral", [1, 2, 3]) + def test_set_membership_preserves_viral_attrs(self, expr: str, num_viral: int) -> None: + result = _run_single(expr, num_viral, rules=VP_IDENTITY) + _assert_viral_attrs(result, num_viral) + # The viral attribute data values must be carried over unchanged. + for i in range(num_viral): + va_name = VA_NAMES[i] + assert va_name in result["DS_r"].data.columns, f"{va_name} missing from result data" + assert list(result["DS_r"].data[va_name]) == VA_VALUES[i] + + +# -- Special cases -- + + +class TestViralAttributeSpecialCases: + def test_non_viral_attribute_still_dropped(self) -> None: + ds = { + "name": "DS_1", + "DataStructure": [ + *BASE_COMPS, + {"name": "VAt_1", "type": "String", "role": "Attribute", "nullable": True}, + ], + } + result = run( + script="DS_r <- DS_1 + DS_2;", + data_structures={"datasets": [ds, {"name": "DS_2", "DataStructure": BASE_COMPS}]}, + datapoints={ + "DS_1": pd.DataFrame({"Id_1": [1, 2], "Me_1": [10.0, 20.0], "VAt_1": ["A", "B"]}), + "DS_2": pd.DataFrame({"Id_1": [1, 2], "Me_1": [5.0, 15.0]}), + }, + ) + assert "VAt_1" not in result["DS_r"].components + + def test_calc_viral_attribute(self) -> None: + result = run( + script=f'{VP_RULES}DS_r <- DS_1 [calc viral attribute VAt_1 := "X"];', + data_structures={"datasets": [{"name": "DS_1", "DataStructure": BASE_COMPS}]}, + datapoints={"DS_1": pd.DataFrame({"Id_1": [1, 2], "Me_1": [10.0, 20.0]})}, + ) + assert result["DS_r"].components["VAt_1"].role == Role.VIRAL_ATTRIBUTE + + def test_input_viral_attribute_legacy_format(self) -> None: + ds = { + "name": "DS_1", + "DataStructure": [ + *BASE_COMPS, + {"name": "VAt_1", "type": "String", "role": "ViralAttribute", "nullable": True}, + ], + } + result = run( + script=f"{VP_RULES}DS_r <- DS_1;", + data_structures={"datasets": [ds]}, + datapoints={"DS_1": pd.DataFrame({"Id_1": [1], "Me_1": [10.0], "VAt_1": ["A"]})}, + ) + assert result["DS_r"].components["VAt_1"].role == Role.VIRAL_ATTRIBUTE + + def test_binary_one_operand_viral(self) -> None: + """Only DS_1 has viral attr, DS_2 doesn't — viral attr propagated from DS_1. + + With no combination happening (DS_2 has no viral attr) an identity rule keeps + each source value on its row.""" + result = run( + script=f"{VP_IDENTITY}DS_r <- DS_1 + DS_2;", + data_structures={"datasets": [_make_ds("DS_1", 2), _make_ds("DS_2", 0)]}, + datapoints={ + "DS_1": _make_dp(2), + "DS_2": pd.DataFrame({"Id_1": [1, 2], "Me_1": [5.0, 15.0]}), + }, + ) + _assert_viral_attrs(result, 2) + assert list(result["DS_r"].data["VAt_1"]) == ["A", "B"] + assert list(result["DS_r"].data["VAt_2"]) == ["X", "Y"] + + +# -- Backends exercised by the operator-specific suites below (issue #877) -- + +BACKENDS = [False, True] + + +# -- Unpivot clause -- + + +# -- Period_indicator time operator -- + + +# -- check_datapoint validation operator -- + +_DPR = """ +define datapoint ruleset R (variable Me_1) is + r1: when Me_1 > 0 then Me_1 < 15 errorcode "e1" errorlevel 1 +end datapoint ruleset; +""" + + +# -- Rule execution on row-preserving dataset-level operators (issue #877) -- + +_AGG_RULE = "define viral propagation VP (variable VAt_1) is aggregate max end viral propagation;" +_ENUM_RULE = ( + 'define viral propagation VP (variable VAt_1) is when "A" then "Z"; else "D" ' + "end viral propagation;" +) + + +# -- hierarchy aggregation operator -- + + +# -- check_hierarchy validation operator -- + + +# -- Aggregate viral-rule type validation (issue #877) -- diff --git a/tests/ViralAttributes/test_viral_propagation.py b/tests/ViralAttributes/test_viral_propagation.py new file mode 100644 index 000000000..67ee7fd64 --- /dev/null +++ b/tests/ViralAttributes/test_viral_propagation.py @@ -0,0 +1,662 @@ +"""Tests for define viral propagation: parsing, end-to-end execution, and semantic validation.""" + +from typing import Optional + +import pandas as pd +import pytest + +from vtlengine import run +from vtlengine.DataTypes import Integer, String +from vtlengine.Exceptions import SemanticError +from vtlengine.Model import Component, Dataset, Role + +# -- Shared propagation rules -- + +CONF_RULE = """ + define viral propagation CONF (variable VAt_1) is + when "C" then "C"; + when "N" then "N"; + else "F" + end viral propagation; +""" + +CONF_BINARY_RULE = """ + define viral propagation COMP_mix (variable VAt_1) is + when "C" and "M" then "N"; + when "M" then "M"; + else " " + end viral propagation; +""" + +TWO_RULES = """ + define viral propagation R1 (variable VAt_1) is + when "C" then "C"; + when "N" then "N"; + else "F" + end viral propagation; + define viral propagation R2 (variable VAt_2) is + aggregate max + end viral propagation; +""" + +AGGR_MAX_RULE = """ + define viral propagation S (variable VAt_1) is + aggregate max + end viral propagation; +""" + +# -- Shared datasets -- + +DS_1VA = { + "name": "DS_1", + "DataStructure": [ + {"name": "Id_1", "type": "Integer", "role": "Identifier", "nullable": False}, + {"name": "Me_1", "type": "Number", "role": "Measure", "nullable": True}, + {"name": "VAt_1", "type": "String", "role": "Viral Attribute", "nullable": True}, + ], +} + +DS_2VA = { + "name": "DS_1", + "DataStructure": [ + {"name": "Id_1", "type": "Integer", "role": "Identifier", "nullable": False}, + {"name": "Me_1", "type": "Number", "role": "Measure", "nullable": True}, + {"name": "VAt_1", "type": "String", "role": "Viral Attribute", "nullable": True}, + {"name": "VAt_2", "type": "Integer", "role": "Viral Attribute", "nullable": True}, + ], +} + +DS_NO_VA = { + "name": "DS_1", + "DataStructure": [ + {"name": "Id_1", "type": "Integer", "role": "Identifier", "nullable": False}, + {"name": "Me_1", "type": "Number", "role": "Measure", "nullable": True}, + ], +} + +SIMPLE_DS = {"datasets": [DS_NO_VA]} +SIMPLE_DP = {"DS_1": pd.DataFrame({"Id_1": [1], "Me_1": [10.0]})} + + +def _ds_pair(ds_def: dict) -> dict: + """Create a two-dataset structure from a single definition (DS_1, DS_2).""" + ds2 = {**ds_def, "name": "DS_2"} + return {"datasets": [ds_def, ds2]} + + +# -- Parsing tests -- + + +class TestViralPropagationParsing: + def test_parse_enumerated(self) -> None: + result = run( + script=CONF_RULE + "DS_r <- DS_1;", + data_structures={"datasets": [DS_1VA]}, + datapoints={"DS_1": pd.DataFrame({"Id_1": [1], "Me_1": [10.0], "VAt_1": ["A"]})}, + ) + assert result["DS_r"].components["VAt_1"].role == Role.VIRAL_ATTRIBUTE + + def test_parse_aggregate(self) -> None: + ds = { + **DS_NO_VA, + "DataStructure": [ + *DS_NO_VA["DataStructure"], + {"name": "VAt_1", "type": "Integer", "role": "Viral Attribute", "nullable": True}, + ], + } + result = run( + script=AGGR_MAX_RULE + "DS_r <- DS_1;", + data_structures={"datasets": [ds]}, + datapoints={"DS_1": pd.DataFrame({"Id_1": [1], "Me_1": [10.0], "VAt_1": [5]})}, + ) + assert "VAt_1" in result["DS_r"].components + + def test_parse_binary_clause(self) -> None: + result = run( + script=CONF_BINARY_RULE + "DS_r <- DS_1;", + data_structures={"datasets": [DS_1VA]}, + datapoints={"DS_1": pd.DataFrame({"Id_1": [1], "Me_1": [10.0], "VAt_1": ["C"]})}, + ) + assert "DS_r" in result + + def test_parse_valuedomain(self) -> None: + script = """ + define viral propagation OBS (valuedomain CL_OBS) is + when "M" then "M"; + else "A" + end viral propagation; + DS_r <- DS_1; + """ + result = run(script=script, data_structures=SIMPLE_DS, datapoints=SIMPLE_DP) + assert "DS_r" in result + + +# -- End-to-end propagation (single viral attribute) -- + +propagation_binary_params = [ + "DS_1 + DS_2", + "DS_1 - DS_2", + "DS_1 * DS_2", +] + + +class TestViralPropagationEndToEnd: + @pytest.mark.parametrize("expr", propagation_binary_params) + def test_enumerated_propagation_binary(self, expr: str) -> None: + """Same CONF_RULE resolution regardless of binary operator.""" + result = run( + script=CONF_RULE + f"DS_r <- {expr};", + data_structures=_ds_pair(DS_1VA), + datapoints={ + "DS_1": pd.DataFrame( + {"Id_1": [1, 2, 3], "Me_1": [10.0, 20.0, 30.0], "VAt_1": ["C", "N", "F"]} + ), + "DS_2": pd.DataFrame( + {"Id_1": [1, 2, 3], "Me_1": [5.0, 15.0, 25.0], "VAt_1": ["N", "F", "F"]} + ), + }, + ) + # C+N→C (unary "C"); N+F→N (unary "N"); F+F→F (else) + assert list(result["DS_r"].data["VAt_1"]) == ["C", "N", "F"] + + def test_binary_clause_precedence(self) -> None: + """Binary clauses take precedence over unary clauses.""" + result = run( + script=CONF_BINARY_RULE + "DS_r <- DS_1 + DS_2;", + data_structures=_ds_pair(DS_1VA), + datapoints={ + "DS_1": pd.DataFrame( + {"Id_1": [1, 2, 3], "Me_1": [10.0, 20.0, 30.0], "VAt_1": ["C", "M", "X"]} + ), + "DS_2": pd.DataFrame( + {"Id_1": [1, 2, 3], "Me_1": [5.0, 15.0, 25.0], "VAt_1": ["M", "F", "Y"]} + ), + }, + ) + # C+M→N (binary); M+F→M (unary "M"); X+Y→" " (else) + assert list(result["DS_r"].data["VAt_1"]) == ["N", "M", " "] + + def test_null_condition_matches_null_value(self) -> None: + """A `when null` clause must match a null (pd.NA/None) viral value, not fall to else.""" + result = run( + script=( + "define viral propagation ee (variable VAt_1) is\n" + ' when null then "Nullable";\n' + ' else "NO_COINCIDENCE"\n' + "end viral propagation;\n" + "DS_r <- DS_1 + DS_2;" + ), + data_structures=_ds_pair(DS_1VA), + datapoints={ + "DS_1": pd.DataFrame({"Id_1": [1, 2], "Me_1": [10.0, 20.0], "VAt_1": ["X", None]}), + "DS_2": pd.DataFrame({"Id_1": [1, 2], "Me_1": [5.0, 15.0], "VAt_1": ["X", None]}), + }, + ) + # X+X→else "NO_COINCIDENCE"; null+null→"Nullable" (when null) on both engines. + assert list(result["DS_r"].data["VAt_1"]) == ["NO_COINCIDENCE", "Nullable"] + + def test_no_rule_combine_raises(self) -> None: + """Both operands viral but no rule defined → SemanticError (issue #877).""" + with pytest.raises(SemanticError) as exc: + run( + script="DS_r <- DS_1 + DS_2;", + data_structures=_ds_pair(DS_1VA), + datapoints={ + "DS_1": pd.DataFrame({"Id_1": [1], "Me_1": [10.0], "VAt_1": ["A"]}), + "DS_2": pd.DataFrame({"Id_1": [1], "Me_1": [5.0], "VAt_1": ["B"]}), + }, + ) + assert "1-3-3-6" in str(exc.value) + + def test_aggregate_max_in_aggregation(self) -> None: + """Aggregate max propagation in group by.""" + ds = { + "name": "DS_1", + "DataStructure": [ + {"name": "Id_1", "type": "Integer", "role": "Identifier", "nullable": False}, + {"name": "Id_2", "type": "Integer", "role": "Identifier", "nullable": False}, + {"name": "Me_1", "type": "Number", "role": "Measure", "nullable": True}, + {"name": "VAt_1", "type": "Integer", "role": "Viral Attribute", "nullable": True}, + ], + } + result = run( + script=AGGR_MAX_RULE + "DS_r <- sum(DS_1 group by Id_1);", + data_structures={"datasets": [ds]}, + datapoints={ + "DS_1": pd.DataFrame( + { + "Id_1": [1, 1, 2], + "Id_2": [1, 2, 1], + "Me_1": [10.0, 20.0, 30.0], + "VAt_1": [3, 7, 5], + } + ) + }, + ) + sorted_data = result["DS_r"].data.sort_values("Id_1").reset_index(drop=True) + assert list(sorted_data["VAt_1"]) == [7, 5] + + @pytest.mark.parametrize( + "agg, both_present, one_null", + # min/max skip nulls (LEAST/GREATEST); sum/avg propagate nulls (a + b). + [("min", 10, 5), ("max", 20, 5), ("sum", 30, None), ("avg", 15, None)], + ) + def test_aggregate_binary_with_nulls( + self, agg: str, both_present: int, one_null: Optional[int] + ) -> None: + """Aggregate viral propagation on a binary op handles nulls identically on both engines. + + min/max ignore a null operand (SQL LEAST/GREATEST); sum/avg propagate it (a + b). + Two nulls always yield null. Pandas and DuckDB must agree. + """ + num_va_ds = { + "name": "DS_1", + "DataStructure": [ + {"name": "Id_1", "type": "Integer", "role": "Identifier", "nullable": False}, + {"name": "Me_1", "type": "Number", "role": "Measure", "nullable": True}, + {"name": "VAt_1", "type": "Number", "role": "Viral Attribute", "nullable": True}, + ], + } + rule = ( + f"define viral propagation ee (variable VAt_1) is\n" + f" aggregate {agg}\n" + f"end viral propagation;\n" + ) + result = run( + script=rule + "DS_r <- DS_1 + DS_2;", + data_structures=_ds_pair(num_va_ds), + datapoints={ + "DS_1": pd.DataFrame( + {"Id_1": [1, 2, 3], "Me_1": [10.0, 20.0, 30.0], "VAt_1": [10.0, None, None]} + ), + "DS_2": pd.DataFrame( + {"Id_1": [1, 2, 3], "Me_1": [5.0, 15.0, 25.0], "VAt_1": [20.0, 5.0, None]} + ), + }, + ) + va = result["DS_r"].data.sort_values("Id_1").reset_index(drop=True)["VAt_1"] + # row1: both present; row2: one operand null; row3: both null -> always null + assert int(va.iloc[0]) == both_present + if one_null is None: + assert pd.isna(va.iloc[1]) + else: + assert int(va.iloc[1]) == one_null + assert pd.isna(va.iloc[2]) + + +# -- A viral attribute requires a rule ONLY when it is combined (issue #906) -- + + +# -- The rule COMBINES per group / partition at aggregation & analytic (issue #906) -- + +# Two identifiers; group/partition Id_1=1 -> {10, null, 30}; Id_1=2 -> {5}. +_GP_NUM_DS = { + "name": "DS_1", + "DataStructure": [ + {"name": "Id_1", "type": "Integer", "role": "Identifier", "nullable": False}, + {"name": "Id_2", "type": "Integer", "role": "Identifier", "nullable": False}, + {"name": "Me_1", "type": "Number", "role": "Measure", "nullable": True}, + {"name": "VAt_1", "type": "Number", "role": "Viral Attribute", "nullable": True}, + ], +} +_GP_STR_DS = { + **_GP_NUM_DS, + "DataStructure": [ + *_GP_NUM_DS["DataStructure"][:3], + {"name": "VAt_1", "type": "String", "role": "Viral Attribute", "nullable": True}, + ], +} + +_ENUM_PAIR_RULE = """ + define viral propagation R (variable VAt_1) is + when "A" and "B" then "AB"; + when "C" and "D" then "CD"; + else "F" + end viral propagation; +""" + + +def _gp_num_dp() -> pd.DataFrame: + return pd.DataFrame( + { + "Id_1": [1, 1, 1, 2], + "Id_2": [1, 2, 3, 1], + "Me_1": [10.0, 20.0, 30.0, 40.0], + "VAt_1": [10.0, None, 30.0, 5.0], + } + ) + + +def _gp_str_dp() -> pd.DataFrame: + return pd.DataFrame( + { + "Id_1": [1, 1, 2, 2], + "Id_2": [1, 2, 1, 2], + "Me_1": [10.0, 20.0, 30.0, 40.0], + "VAt_1": ["A", "B", "C", "D"], + } + ) + + +# -- Multi-attribute propagation (enumerated + aggregate in one script) -- + + +class TestViralPropagationMultiAttribute: + @pytest.mark.parametrize("expr", propagation_binary_params) + def test_two_rules_two_attrs_binary(self, expr: str) -> None: + """TWO_RULES: VAt_1 enumerated + VAt_2 aggr max, applied to binary ops.""" + result = run( + script=TWO_RULES + f"DS_r <- {expr};", + data_structures=_ds_pair(DS_2VA), + datapoints={ + "DS_1": pd.DataFrame( + { + "Id_1": [1, 2, 3], + "Me_1": [10.0, 20.0, 30.0], + "VAt_1": ["C", "N", "F"], + "VAt_2": [3, 5, 1], + } + ), + "DS_2": pd.DataFrame( + { + "Id_1": [1, 2, 3], + "Me_1": [5.0, 15.0, 25.0], + "VAt_1": ["N", "F", "F"], + "VAt_2": [7, 2, 4], + } + ), + }, + ) + ds_r = result["DS_r"] + # VAt_1 enumerated: C+N→C, N+F→N, F+F→F + assert list(ds_r.data["VAt_1"]) == ["C", "N", "F"] + # VAt_2 aggr max: max(3,7)=7, max(5,2)=5, max(1,4)=4 + assert list(ds_r.data["VAt_2"]) == [7, 5, 4] + + +# -- Semantic validation -- + + +class TestViralPropagationValidation: + def test_duplicate_variable_rule_raises_error(self) -> None: + script = """ + define viral propagation r1 (variable VAt_1) is + when "C" then "C" + end viral propagation; + define viral propagation r2 (variable VAt_1) is + when "N" then "N" + end viral propagation; + DS_r <- DS_1; + """ + with pytest.raises(SemanticError, match="1-3-3-1"): + run(script=script, data_structures=SIMPLE_DS, datapoints=SIMPLE_DP) + + def test_duplicate_enumeration_raises_error(self) -> None: + script = """ + define viral propagation dup (variable VAt_1) is + when "C" then "C"; + when "C" then "N" + end viral propagation; + DS_r <- DS_1; + """ + with pytest.raises(SemanticError, match="1-3-3-4"): + run(script=script, data_structures=SIMPLE_DS, datapoints=SIMPLE_DP) + + +# -- vpBody grammar restriction (VTL 2.2 reference manual) -- + + +def _vp(body: str) -> str: + return f"define viral propagation R (variable VAt_1) is\n{body}\nend viral propagation;" + + +# -- Propagation rules through join operators -- + +_ID_1 = {"name": "Id_1", "type": "Integer", "role": "Identifier", "nullable": False} +_ID_2 = {"name": "Id_2", "type": "Integer", "role": "Identifier", "nullable": False} +_ME_1 = {"name": "Me_1", "type": "Number", "role": "Measure", "nullable": True} +_ME_2 = {"name": "Me_2", "type": "Number", "role": "Measure", "nullable": True} +_VA = {"name": "VAt_1", "type": "String", "role": "Viral Attribute", "nullable": True} + +DS_JOIN_1 = {"name": "DS_1", "DataStructure": [_ID_1, _ME_1, _VA]} +DS_JOIN_2 = {"name": "DS_2", "DataStructure": [_ID_1, _ME_2, _VA]} +DS_JOIN_2_NO_VA = {"name": "DS_2", "DataStructure": [_ID_1, _ME_2]} +DS_CROSS_2 = {"name": "DS_2", "DataStructure": [_ID_2, _ME_2, _VA]} +DS_CROSS_2_NO_VA = {"name": "DS_2", "DataStructure": [_ID_2, _ME_2]} + + +# -- Null operands in the join viral combination -- +# +# Regression: the join combined two shared viral columns by dropping nulls +# *before* resolving, so a ``(null, X)`` pair collapsed to ``[X]`` and leaked X +# unchanged instead of going through the propagation rule (which a binary +# operator like ``DS_1 + DS_2`` applied correctly via resolve_pair). + +DS_PLUS_2 = {"name": "DS_2", "DataStructure": [_ID_1, _ME_1, _VA]} # Me_1 in both, for ``+`` + + +class TestViralPropagationJoinNulls: + """A shared viral attribute where one operand's value is null must still go + through the propagation rule in a join, exactly like in a binary operator.""" + + @pytest.mark.parametrize("join_op", ["inner_join", "left_join", "full_join"]) + def test_null_pair_applies_rule(self, join_op: str) -> None: + """(null, X) and (X, null) resolve through the rule, not leaking X.""" + result = run( + script=CONF_RULE + f"DS_r <- {join_op}(DS_1, DS_2);", + data_structures={"datasets": [DS_JOIN_1, DS_JOIN_2]}, + datapoints={ + "DS_1": pd.DataFrame( + { + "Id_1": [1, 2, 3, 4], + "Me_1": [1.0, 2.0, 3.0, 4.0], + "VAt_1": ["C", "Z", None, None], + } + ), + "DS_2": pd.DataFrame( + { + "Id_1": [1, 2, 3, 4], + "Me_2": [1.0, 2.0, 3.0, 4.0], + "VAt_1": ["N", None, "Z", None], + } + ), + }, + ) + d = result["DS_r"].data.sort_values("Id_1").reset_index(drop=True) + # (C,N)->C (unary); (Z,null)->F (else); (null,Z)->F (else); (null,null)->F (else) + assert list(d["VAt_1"]) == ["C", "F", "F", "F"] + # The lone non-null value must NOT survive unchanged. + assert "Z" not in list(d["VAt_1"]) + + def test_join_viral_matches_binary_plus(self) -> None: + """The join's viral combination is identical to the binary ``+`` one for + the same viral data (including null operands). The ``+`` uses a matching + measure; the join uses distinct measures so the non-key columns do not + collide on the final un-prefix step.""" + vat_1 = ["C", "Z", None, None, "N"] + vat_2 = ["N", None, "Z", None, "N"] + ids = [1, 2, 3, 4, 5] + nums = [1.0, 2.0, 3.0, 4.0, 5.0] + plus = run( + script=CONF_RULE + "DS_r <- DS_1 + DS_2;", + data_structures={"datasets": [DS_JOIN_1, DS_PLUS_2]}, + datapoints={ + "DS_1": pd.DataFrame({"Id_1": ids, "Me_1": nums, "VAt_1": vat_1}), + "DS_2": pd.DataFrame({"Id_1": ids, "Me_1": nums, "VAt_1": vat_2}), + }, + ) + join = run( + script=CONF_RULE + "DS_r <- inner_join(DS_1, DS_2);", + data_structures={"datasets": [DS_JOIN_1, DS_JOIN_2]}, + datapoints={ + "DS_1": pd.DataFrame({"Id_1": ids, "Me_1": nums, "VAt_1": vat_1}), + "DS_2": pd.DataFrame({"Id_1": ids, "Me_2": nums, "VAt_1": vat_2}), + }, + ) + p = plus["DS_r"].data.sort_values("Id_1").reset_index(drop=True) + j = join["DS_r"].data.sort_values("Id_1").reset_index(drop=True) + assert list(p["VAt_1"]) == list(j["VAt_1"]) + + def test_aggregate_rule_ignores_nulls_in_join(self) -> None: + """An aggregate (max) viral rule ignores nulls in a join: (null, X)->X, + (null, null)->null.""" + result = run( + script=AGGR_MAX_RULE + "DS_r <- inner_join(DS_1, DS_2);", + data_structures={"datasets": [DS_JOIN_1, DS_JOIN_2]}, + datapoints={ + "DS_1": pd.DataFrame( + {"Id_1": [1, 2, 3], "Me_1": [1.0, 2.0, 3.0], "VAt_1": ["B", None, None]} + ), + "DS_2": pd.DataFrame( + {"Id_1": [1, 2, 3], "Me_2": [1.0, 2.0, 3.0], "VAt_1": [None, "A", None]} + ), + }, + ) + d = result["DS_r"].data.sort_values("Id_1").reset_index(drop=True) + assert d["VAt_1"].iloc[0] == "B" # (B, null) -> B + assert d["VAt_1"].iloc[1] == "A" # (null, A) -> A + assert pd.isna(d["VAt_1"].iloc[2]) # (null, null) -> null + + +# -- keep clause must preserve viral attributes (they always propagate) -- + +_AT = {"name": "At_1", "type": "String", "role": "Attribute", "nullable": True} +_VA_2 = {"name": "VAt_2", "type": "Integer", "role": "Viral Attribute", "nullable": True} +DS_KEEP = {"name": "DS_1", "DataStructure": [_ID_1, _ME_1, _ME_2, _AT, _VA]} +DS_KEEP_2VA = {"name": "DS_1", "DataStructure": [_ID_1, _ME_1, _VA, _VA_2]} +DS_NA_VA_1 = {"name": "DS_1", "DataStructure": [_ID_1, _ME_1, _AT, _VA]} +DS_NA_VA_2 = {"name": "DS_2", "DataStructure": [_ID_1, _ME_2, _AT, _VA]} + + +class TestKeepPreservesViralAttributes: + """A keep clause restricts identifiers/measures/non-viral attributes, but viral + attributes always propagate and survive implicitly (without being listed).""" + + def test_standalone_keep_preserves_viral_drops_rest(self) -> None: + """Keep Me_1 keeps the viral attr but drops the other measure and the + non-viral attribute.""" + result = run( + script=CONF_RULE + "DS_r <- DS_1[keep Me_1];", + data_structures={"datasets": [DS_KEEP]}, + datapoints={ + "DS_1": pd.DataFrame( + { + "Id_1": [1, 2], + "Me_1": [10.0, 20.0], + "Me_2": [1.0, 2.0], + "At_1": ["a", "b"], + "VAt_1": ["C", "N"], + } + ) + }, + ) + ds = result["DS_r"] + assert set(ds.components) == {"Id_1", "Me_1", "VAt_1"} + assert ds.components["VAt_1"].role == Role.VIRAL_ATTRIBUTE + # Single operand: viral value passes through unchanged (no combination). + d = ds.data.sort_values("Id_1").reset_index(drop=True) + assert "VAt_1" in d.columns + assert list(d["VAt_1"]) == ["C", "N"] + + def test_keep_listing_viral_no_duplicate(self) -> None: + """Listing the viral attribute explicitly in keep does not duplicate it.""" + result = run( + script=CONF_RULE + "DS_r <- DS_1[keep Me_1, VAt_1];", + data_structures={"datasets": [DS_KEEP]}, + datapoints={ + "DS_1": pd.DataFrame( + { + "Id_1": [1], + "Me_1": [10.0], + "Me_2": [1.0], + "At_1": ["a"], + "VAt_1": ["C"], + } + ) + }, + ) + ds = result["DS_r"] + assert set(ds.components) == {"Id_1", "Me_1", "VAt_1"} + assert list(ds.data.columns).count("VAt_1") == 1 + + def test_keep_preserves_multiple_viral(self) -> None: + """All viral attributes survive a keep (each has a declared rule).""" + result = run( + script=TWO_RULES + "DS_r <- DS_1[keep Me_1];", + data_structures={"datasets": [DS_KEEP_2VA]}, + datapoints={ + "DS_1": pd.DataFrame({"Id_1": [1], "Me_1": [10.0], "VAt_1": ["C"], "VAt_2": [7]}) + }, + ) + ds = result["DS_r"] + assert set(ds.components) == {"Id_1", "Me_1", "VAt_1", "VAt_2"} + assert ds.components["VAt_2"].role == Role.VIRAL_ATTRIBUTE + + @pytest.mark.parametrize("join_op", ["inner_join", "left_join", "full_join"]) + def test_keep_in_join_preserves_combined_viral(self, join_op: str) -> None: + """A keep inside a join keeps the merged viral attribute, combined by the + rule, even though only a measure is listed.""" + result = run( + script=CONF_RULE + f"DS_r <- {join_op}(DS_1, DS_2 keep Me_1);", + data_structures={"datasets": [DS_JOIN_1, DS_JOIN_2]}, + datapoints={ + "DS_1": pd.DataFrame({"Id_1": [1, 2], "Me_1": [10.0, 20.0], "VAt_1": ["C", "N"]}), + "DS_2": pd.DataFrame({"Id_1": [1, 2], "Me_2": [5.0, 15.0], "VAt_1": ["N", "F"]}), + }, + ) + ds = result["DS_r"] + assert set(ds.components) == {"Id_1", "Me_1", "VAt_1"} + assert ds.components["VAt_1"].role == Role.VIRAL_ATTRIBUTE + d = ds.data.sort_values("Id_1").reset_index(drop=True) + # C+N->C (unary "C"); N+F->N (unary "N") + assert list(d["VAt_1"]) == ["C", "N"] + + def test_keep_nonviral_attr_in_join_keeps_viral_with_null(self) -> None: + """The reported scenario: keeping a non-viral attribute in a join still + propagates the viral attribute, and a (null, X) pair resolves via the rule.""" + result = run( + script=CONF_RULE + "DS_r <- inner_join(DS_1, DS_2 keep DS_2#At_1);", + data_structures={"datasets": [DS_NA_VA_1, DS_NA_VA_2]}, + datapoints={ + "DS_1": pd.DataFrame({"Id_1": [1], "Me_1": [10.0], "At_1": ["x"], "VAt_1": [None]}), + "DS_2": pd.DataFrame({"Id_1": [1], "Me_2": [5.0], "At_1": ["y"], "VAt_1": ["Z"]}), + }, + ) + ds = result["DS_r"] + # The explicitly kept non-viral attribute and the viral attribute survive. + assert set(ds.components) == {"Id_1", "At_1", "VAt_1"} + assert ds.components["VAt_1"].role == Role.VIRAL_ATTRIBUTE + assert ds.components["At_1"].role == Role.ATTRIBUTE + # (null, "Z") -> "F" (else), not the leaked "Z". + assert ds.data["VAt_1"].iloc[0] == "F" + assert ds.data["At_1"].iloc[0] == "y" + + +# -- Unit tests for the combination-point check helpers -- + + +def _viral_ds(name: str, viral_names: list) -> Dataset: + comps = {"Id_1": Component("Id_1", Integer, Role.IDENTIFIER, False)} + for v in viral_names: + comps[v] = Component(v, String, Role.VIRAL_ATTRIBUTE, True) + return Dataset(name=name, components=comps, data=None) + + +# -- Row-preserving operators copy viral attributes, they do NOT execute the rule (#906) -- + +NUM_VA_2ID = { + "name": "DS_1", + "DataStructure": [ + {"name": "Id_1", "type": "Integer", "role": "Identifier", "nullable": False}, + {"name": "Id_2", "type": "String", "role": "Identifier", "nullable": False}, + {"name": "Me_1", "type": "Number", "role": "Measure", "nullable": True}, + {"name": "VAt_1", "type": "Number", "role": "Viral Attribute", "nullable": True}, + ], +} + +ENUM_REMAP_RULE = """ + define viral propagation R (variable VAt_1) is + when "A" then "Z"; + else "F" + end viral propagation; +"""