Fix #892: materialize viral-propagation operand and unpivot window once

src/vtlengine/duckdb_transpiler/Transpiler/__init__.py

@@ -1997,14 +1997,20 @@ def visit_RegularAggregation_unpivot(self, node: AST.RegularAggregation) -> str:
             qn = quote_name(comp.name)
             excl_list.append(qn)
             expr_list.append(f"{vp_dataset_wide_sql(v_rule, qn)} AS {qn}")
+        cte: Optional[CTEBuilder] = None
         if expr_list:
-            table_src = (
-                f"(SELECT * EXCLUDE ({', '.join(excl_list)}), {', '.join(expr_list)} "
-                f"FROM {table_src} AS _uv_in) AS _uv_src"
+            cte = CTEBuilder()
+            cte.cte(
+                "_uv_src",
+                f"SELECT * EXCLUDE ({', '.join(excl_list)}), {', '.join(expr_list)} "
+                f"FROM {table_src} AS _uv_in",
+                materialized=True,
             )
+            table_src = "_uv_src"
 
         if not measure_names:
-            return f"SELECT * FROM {table_src}"
+            sql = f"SELECT * FROM {table_src}"
+            return cte.select(sql) if cte is not None else sql
 
         parts: List[str] = []
         for measure in measure_names:
@@ -2019,7 +2025,8 @@ def visit_RegularAggregation_unpivot(self, node: AST.RegularAggregation) -> str:
             )
             parts.append(part)
 
-        return " UNION ALL ".join(parts)
+        union_sql = " UNION ALL ".join(parts)
+        return cte.select(union_sql) if cte is not None else union_sql
 
     # Aggregation visitor
 

tests/duckdb_transpiler/test_run.py

@@ -3266,3 +3266,45 @@ def test_rule_priority_mode(self, hierarchy_structures):
         )
         expected = expected.sort_values(["Id_1", "Id_2"]).reset_index(drop=True)
         pd.testing.assert_frame_equal(result, expected, check_dtype=False, check_like=True)
+
+
+class TestViralPropagationSQLDeduplication:
+    def test_unpivot_viral_window_computed_once(self):
+        """The dataset-wide viral window appears once, referenced by every measure arm."""
+        script = (
+            "define viral propagation VP (variable VAt_1) is aggregate max "
+            "end viral propagation;\n"
+            "DS_u <- DS_2[unpivot Id_2, Val];"
+        )
+        data_structures = {
+            "datasets": [
+                {
+                    "name": "DS_2",
+                    "DataStructure": [
+                        {
+                            "name": "Id_1",
+                            "type": "Integer",
+                            "role": "Identifier",
+                            "nullable": False,
+                        },
+                        {"name": "Me_1", "type": "Number", "role": "Measure", "nullable": True},
+                        {"name": "Me_2", "type": "Number", "role": "Measure", "nullable": True},
+                        {"name": "Me_3", "type": "Number", "role": "Measure", "nullable": True},
+                        {
+                            "name": "VAt_1",
+                            "type": "Number",
+                            "role": "Viral Attribute",
+                            "nullable": True,
+                        },
+                    ],
+                }
+            ]
+        }
+
+        queries = {name: sql for name, sql, _ in transpile(script, data_structures)}
+
+        window = 'MAX("VAt_1") OVER ()'
+        assert queries["DS_u"].count(window) == 1, (
+            f"Dataset-wide viral window should be computed once, found "
+            f"{queries['DS_u'].count(window)} copies:\n{queries['DS_u']}"
+        )