Fix #918: support Time (TimeInterval) identifiers in DuckDB time operators

src/vtlengine/Operators/Time.py

@@ -92,6 +92,28 @@ def _get_period(cls, value: str) -> str:
         tp_value = TimePeriodHandler(value)
         return tp_value.period_indicator
 
+    @classmethod
+    def _iter_groups(cls, data: pd.DataFrame) -> Any:
+        """Iterate over the series the operand holds, one per combination of the
+        non-time identifiers.
+
+        A Data Set may have the time identifier as its only identifier, in which
+        case it holds a single series and Pandas cannot group it.
+        """
+        if not cls.other_ids:
+            return [((), data)]
+        return data.groupby(cls.other_ids)
+
+    @staticmethod
+    def _group_key(values: Any) -> Any:
+        """The dictionary key a group of non-time identifier values maps to.
+
+        Pandas hands over a one-element tuple when grouping by a single column, so
+        the values have to be unwrapped the same way on both sides of a lookup.
+        """
+        values = tuple(values)
+        return values[0] if len(values) == 1 else values
+
     @classmethod
     def parse_date(cls, date_str: str) -> date:
         return parse_date_value(date_str)
@@ -142,16 +164,24 @@ def find_min_frequency(cls, dates: Any) -> str:
         (0, 0, 1): "D",
     }
 
+    @classmethod
+    def _interval_endpoints(cls, interval: str) -> Any:
+        """The two endpoint dates of a TimeInterval.
+
+        The endpoints may carry a time component, which the input format allows,
+        so only the date part is read.
+        """
+        start_str, end_str = interval.split("/")
+        return date.fromisoformat(start_str[:10]), date.fromisoformat(end_str[:10])
+
     @classmethod
     def get_frequency_from_time(cls, interval: str) -> Any:
-        start_date, end_date = interval.split("/")
-        return date.fromisoformat(end_date) - date.fromisoformat(start_date)
+        start, end = cls._interval_endpoints(interval)
+        return end - start
 
     @classmethod
     def _classify_interval_period(cls, interval: str) -> str:
-        start_str, end_str = interval.split("/")
-        start = date.fromisoformat(start_str)
-        end = date.fromisoformat(end_str)
+        start, end = cls._interval_endpoints(interval)
         candidates = [relativedelta(endpoint, start) for endpoint in (end, end + timedelta(days=1))]
         candidates = [
             rd
@@ -264,11 +294,15 @@ def evaluate(cls, operand: Any) -> Any:
         if data_type == TimePeriod:
             result.data = cls._period_accumulation(result.data, measure_names)
         elif data_type in (Date, TimeInterval):
-            result.data[measure_names] = (
-                result.data.groupby(cls.other_ids)[measure_names]
-                .apply(cls.py_op)
-                .reset_index(drop=True)
-            )
+            if cls.other_ids:
+                accumulated = (
+                    result.data.groupby(cls.other_ids)[measure_names]
+                    .apply(cls.py_op)
+                    .reset_index(drop=True)
+                )
+            else:
+                accumulated = cls.py_op(result.data[measure_names].copy()).reset_index(drop=True)
+            result.data[measure_names] = accumulated
         else:
             raise SemanticError("1-1-19-8", op=cls.op, comp_type="dataset", param="date type")
         return result
@@ -401,6 +435,9 @@ def evaluate(cls, operand: Dataset, fill_type: str) -> Dataset:
         result.data[cls.time_id] = result.data[cls.time_id].astype("string[pyarrow]")
         if len(result.data) < 2:
             return result
+        if not cls.other_ids:
+            # The operand holds a single series, so both limits span the same range.
+            fill_type = "all"
         data_type = result.components[cls.time_id].data_type
         if data_type == TimePeriod:
             result.data = cls.fill_periods(result.data, fill_type)
@@ -516,12 +553,10 @@ def compute_min_max(group: Any) -> Dict[str, Any]:
         if fill_type == "all":
             return compute_min_max(data[cls.time_id])
 
-        grouped = data.groupby(cls.other_ids)
-        result_dict = {
-            name if len(name) > 1 else name[0]: compute_min_max(group[cls.time_id])
-            for name, group in grouped
+        return {
+            cls._group_key(name): compute_min_max(group[cls.time_id])
+            for name, group in cls._iter_groups(data)
         }
-        return result_dict
 
     @classmethod
     def fill_dates(cls, data: pd.DataFrame, fill_type: str, min_frequency: str) -> pd.DataFrame:
@@ -539,12 +574,13 @@ def date_filler(cls, data: pd.DataFrame, fill_type: str, min_frequency: str) ->
         def create_filled_dates(group: Any, min_max: Dict[str, Any]) -> (pd.DataFrame, str):  # type: ignore[syntax]
             date_range = cls._period_range(min_max["min"], min_max["max"], min_frequency)
             date_df = pd.DataFrame(date_range, columns=[cls.time_id])
-            date_df[cls.other_ids] = group.iloc[0][cls.other_ids]
+            if cls.other_ids:
+                date_df[cls.other_ids] = group.iloc[0][cls.other_ids]
             date_df[cls.measures] = None
             return date_df, min_max["date_format"]
 
-        for name, group in data.groupby(cls.other_ids):
-            min_max = MAX_MIN if fill_type == "all" else MAX_MIN[name if len(name) > 1 else name[0]]
+        for name, group in cls._iter_groups(data):
+            min_max = MAX_MIN if fill_type == "all" else MAX_MIN[cls._group_key(name)]
             filled_dates, date_format = create_filled_dates(group, min_max)
             filled_data.append(filled_dates)
 
@@ -556,72 +592,61 @@ def create_filled_dates(group: Any, min_max: Dict[str, Any]) -> (pd.DataFrame, s
 
     @classmethod
     def max_min_from_time(cls, data: pd.DataFrame, fill_type: str = "all") -> Dict[str, Any]:
-        data = data.applymap(str).sort_values(  # type: ignore[operator]
-            by=cls.other_ids + [cls.time_id]
-        )
+        data = data.sort_values(by=cls.other_ids + [cls.time_id])
 
         def extract_max_min(group: Any) -> Dict[str, Any]:
-            start_dates = group.str.split("/").str[0]
-            end_dates = group.str.split("/").str[1]
+            intervals = group.astype(str)
+            start_dates = intervals.str.split("/").str[0]
+            end_dates = intervals.str.split("/").str[1]
             return {
                 "start": {"min": start_dates.min(), "max": start_dates.max()},
                 "end": {"min": end_dates.min(), "max": end_dates.max()},
             }
 
         if fill_type == "all":
             return extract_max_min(data[cls.time_id])
-        else:
-            return {
-                (name if len(name) > 1 else name[0]): extract_max_min(group[cls.time_id])
-                for name, group in data.groupby(cls.other_ids)
-            }
+        return {
+            cls._group_key(name): extract_max_min(group[cls.time_id])
+            for name, group in cls._iter_groups(data)
+        }
 
     @classmethod
     def fill_time_intervals(
         cls, data: pd.DataFrame, fill_type: str, frequency: str
     ) -> pd.DataFrame:
-        result_data = cls.time_filler(data, fill_type, frequency)
-        not_na = result_data[cls.measures].notna().any(axis=1)
-        duplicated = result_data.duplicated(subset=(cls.other_ids + [cls.time_id]), keep=False)
-        return result_data[~duplicated | not_na]
+        """Add the Data Points the interval grid expects and keep the operand's own.
 
-    @classmethod
-    def time_filler(cls, data: pd.DataFrame, fill_type: str, frequency: str) -> pd.DataFrame:
+        Both endpoints are stepped by one frequency from their own lower limit, and
+        the k-th start is then paired with the k-th end. Only Data Points whose key
+        is missing are added, so the operand's own intervals always survive, even
+        when they overlap and the two endpoint grids come out different lengths.
+        """
         MAX_MIN = cls.max_min_from_time(data, fill_type)
+        non_ids = [c for c in data.columns if c not in cls.other_ids and c != cls.time_id]
 
-        def fill_group(group_df: pd.DataFrame) -> pd.DataFrame:
-            group_key = group_df.iloc[0][cls.other_ids].values
-            if fill_type != "all":
-                group_key = group_key[0] if len(group_key) == 1 else tuple(group_key)
-            group_dict = MAX_MIN if fill_type == "all" else MAX_MIN[group_key]
+        def grid_intervals(limits: Dict[str, Any], sample: str) -> List[str]:
+            # The limits stay ISO strings so that a time of day survives the range.
+            starts = cls._period_range(limits["start"]["min"], limits["start"]["max"], frequency)
+            ends = cls._period_range(limits["end"]["min"], limits["end"]["max"], frequency)
+            fmt = "%Y-%m-%dT%H:%M:%S" if _has_time_component(sample.split("/")[0]) else "%Y-%m-%d"
+            return [f"{s.strftime(fmt)}/{e.strftime(fmt)}" for s, e in zip(starts, ends)]
 
-            intervals = [
-                f"{group_dict['start']['min']}/{group_dict['end']['min']}",
-                f"{group_dict['start']['max']}/{group_dict['end']['max']}",
-            ]
-            for interval in intervals:
-                if interval not in group_df[cls.time_id].values:
-                    empty_row = group_df.iloc[0].copy()
-                    empty_row[cls.time_id] = interval
-                    empty_row[cls.measures] = None
-                    group_df = pd.concat([group_df, pd.DataFrame([empty_row])], ignore_index=True)
-            start_group_df = group_df.copy()
-            start_group_df[cls.time_id] = start_group_df[cls.time_id].str.split("/").str[0]
-            end_group_df = group_df.copy()
-            end_group_df[cls.time_id] = end_group_df[cls.time_id].str.split("/").str[1]
-            start_filled = cls.date_filler(start_group_df, fill_type, frequency)
-            end_filled = cls.date_filler(end_group_df, fill_type, frequency)
-            start_filled[cls.time_id] = start_filled[cls.time_id].str.cat(
-                end_filled[cls.time_id], sep="/"
-            )
-            return start_filled
+        filled_rows: List[Dict[str, Any]] = []
+        for name, group_df in cls._iter_groups(data):
+            limits = MAX_MIN if fill_type == "all" else MAX_MIN[cls._group_key(name)]
+            existing = set(group_df[cls.time_id].astype(str))
+            other_vals = dict(zip(cls.other_ids, tuple(name)))
+            for interval in grid_intervals(limits, str(group_df[cls.time_id].iloc[0])):
+                if interval not in existing:
+                    filled_rows.append(
+                        {**other_vals, cls.time_id: interval, **dict.fromkeys(non_ids)}
+                    )
 
-        filled_data = [fill_group(group_df) for _, group_df in data.groupby(cls.other_ids)]
-        return (
-            pd.concat(filled_data, ignore_index=True)
-            .sort_values(by=cls.other_ids + [cls.time_id])
-            .drop_duplicates()
-        )
+        result = data
+        if filled_rows:
+            result = pd.concat([data, pd.DataFrame(filled_rows)], ignore_index=True)
+        result[cls.time_id] = result[cls.time_id].astype("string[pyarrow]")
+        return result.sort_values(by=cls.other_ids + [cls.time_id]).reset_index(drop=True)
 
 
 class Time_Shift(Binary):

src/vtlengine/duckdb_transpiler/Transpiler/__init__.py

@@ -1264,7 +1264,7 @@ def _visit_params(self, params: List[Any]) -> List[Optional[str]]:
     def _resolve_time_identifier(self, ds: Dataset, op_name: str) -> Any:
         """Return the time identifier name and type for time-based operators."""
         for comp in ds.components.values():
-            if comp.data_type in (TimePeriod, Date) and comp.role == Role.IDENTIFIER:
+            if comp.data_type in (TimePeriod, Date, TimeInterval) and comp.role == Role.IDENTIFIER:
                 return comp.name, comp.data_type
 
     def _build_time_grid_parts(
@@ -1321,6 +1321,8 @@ def visit_ParamOp_fill_time_series(self, node: AST.ParamOp) -> str:
 
         if time_type == Date:
             return self._fill_time_series_date(ds, src, time_id, fill_mode)
+        if time_type == TimeInterval:
+            return self._fill_time_series_interval(ds, src, time_id, fill_mode)
         return self._fill_time_series_period(ds, src, time_id, fill_mode)
 
     def _fill_time_series_period(self, ds: Dataset, src: str, time_id: str, fill_mode: str) -> str:
@@ -1461,6 +1463,88 @@ def _fill_time_series_date(self, ds: Dataset, src: str, time_id: str, fill_mode:
         )
         return cte.select(final)
 
+    def _fill_time_series_interval(
+        self, ds: Dataset, src: str, time_id: str, fill_mode: str
+    ) -> str:
+        """Fill time series for TimeInterval identifiers.
+
+        The start dates and the end dates are each stepped by one frequency from
+        their own lower bound, and then the k-th start is paired with the k-th
+        end. That mirrors Fill_time_series.fill_time_intervals, and like it only
+        adds the Data Points whose key the operand is missing.
+        """
+        time_col, other_id_cols, _, join_on, final_select, order_by = self._build_time_grid_parts(
+            ds, time_id
+        )
+        per_group = fill_mode == "single" and bool(other_id_cols)
+        freq_step = "(SELECT step FROM freq)"
+
+        cte = CTEBuilder()
+        cte.cte("source", f"SELECT * FROM {src}")
+        # One frequency for the whole operand, as Fill_time_series.evaluate requires.
+        # sample carries the operand's own representation over to the added intervals.
+        cte.cte(
+            "freq",
+            "SELECT CASE WHEN COUNT(DISTINCT f) > 1 THEN error("
+            "'VTL 1-1-19-9: fill_time_series needs a single time interval frequency') "
+            "ELSE vtl_interval_freq_to_step(MIN(f)) END AS step, MIN(iv) AS sample "
+            f"FROM (SELECT vtl_interval_freq({time_col}) AS f, {time_col} AS iv "
+            f"FROM source WHERE {time_col} IS NOT NULL)",
+        )
+
+        bounds_cols = (
+            f"MIN(vtl_interval_start_ts({time_col})) AS min_s, "
+            f"MAX(vtl_interval_start_ts({time_col})) AS max_s, "
+            f"MIN(vtl_interval_end_ts({time_col})) AS min_e, "
+            f"MAX(vtl_interval_end_ts({time_col})) AS max_e"
+        )
+        if per_group:
+            oid_csv = ", ".join(other_id_cols)
+            cte.cte("bounds", f"SELECT {oid_csv}, {bounds_cols} FROM source GROUP BY {oid_csv}")
+            b_cols = ", ".join(f"b.{oc}" for oc in other_id_cols) + ", "
+            partition = "PARTITION BY " + ", ".join(f"b.{oc}" for oc in other_id_cols) + " "
+        else:
+            cte.cte("bounds", f"SELECT {bounds_cols} FROM source")
+            b_cols = partition = ""
+
+        for name, lo, hi, alias in (
+            ("starts", "min_s", "max_s", "d1"),
+            ("ends", "min_e", "max_e", "d2"),
+        ):
+            cte.cte(
+                name,
+                f"SELECT {b_cols}ROW_NUMBER() OVER ({partition}ORDER BY d) AS k, "
+                f"CAST(d AS TIMESTAMP) AS {alias} "
+                f"FROM bounds b, generate_series(b.{lo}, b.{hi}, {freq_step}) AS t(d)",
+            )
+
+        grid = f"vtl_interval_build(st.d1, en.d2, (SELECT sample FROM freq)) AS {time_col}"
+        if per_group:
+            on_ids = "".join(f" AND st.{oc} = en.{oc}" for oc in other_id_cols)
+            st_cols = ", ".join(f"st.{oc}" for oc in other_id_cols)
+            cte.cte(
+                "full_grid",
+                f"SELECT {st_cols}, {grid} FROM starts st JOIN ends en ON st.k = en.k{on_ids}",
+            )
+        elif other_id_cols:
+            oid_csv = ", ".join(other_id_cols)
+            cte.cte("interval_grid", f"SELECT {grid} FROM starts st JOIN ends en ON st.k = en.k")
+            cte.cte("group_keys", f"SELECT DISTINCT {oid_csv} FROM source")
+            gk_cols = ", ".join(f"gk.{oc}" for oc in other_id_cols)
+            cte.cte(
+                "full_grid",
+                f"SELECT {gk_cols}, ig.{time_col} FROM group_keys gk, interval_grid ig",
+            )
+        else:
+            cte.cte("full_grid", f"SELECT {grid} FROM starts st JOIN ends en ON st.k = en.k")
+
+        keys = self._grid_with_source_keys(other_id_cols, time_col)
+        final = (
+            f"SELECT {final_select} FROM ({keys}) g "
+            f"LEFT JOIN source s ON {join_on} ORDER BY {order_by}"
+        )
+        return cte.select(final)
+
     def _visit_flow_stock(self, node: AST.UnaryOp, op: str) -> str:
         """Visit FLOW_TO_STOCK or STOCK_TO_FLOW: window functions over time series."""
         ds = self._get_dataset_structure(node.operand)
@@ -1482,7 +1566,9 @@ def _visit_flow_stock(self, node: AST.UnaryOp, op: str) -> str:
             col = quote_name(comp.name)
             if comp.role == Role.IDENTIFIER:
                 cols.append(col)
-            elif comp.data_type in (Integer, Number):
+            # Only number measures accumulate: the reference manual types the operand
+            # as measure<number>, so attributes pass through untouched (issue #931).
+            elif comp.role == Role.MEASURE and comp.data_type in (Integer, Number):
                 if op == tokens.FLOW_TO_STOCK:
                     cols.append(
                         f"CASE WHEN {col} IS NULL THEN NULL ELSE "
@@ -1513,19 +1599,34 @@ def visit_BinOp_timeshift(self, node: AST.BinOp) -> str:
                 col = quote_name(comp.name)
                 cols.append(shifted if comp.name == time_id else col)
             return SQLBuilder().select(*cols).from_table(src).build()
+        elif time_type == TimeInterval:
+            # A TimeInterval series has a single frequency, taken from the duration of
+            # one interval. Time_Shift.evaluate reads it off row 0 of the unsorted
+            # operand; MIN() is the deterministic equivalent in SQL, and it agrees
+            # with row 0 for any operand in chronological order.
+            cols = [
+                (
+                    f"vtl_interval_shift({quote_name(comp.name)}, {shift_sql}, freq.step) "
+                    f"AS {quote_name(comp.name)}"
+                )
+                if comp.name == time_id
+                else quote_name(comp.name)
+                for comp in ds.components.values()
+            ]
+            freq_sql = (
+                f"SELECT vtl_interval_step(MIN({time_col})) AS step "
+                f"FROM {src} WHERE {time_col} IS NOT NULL"
+            )
+            return f"""SELECT {", ".join(cols)}
+FROM {src}, (
+    {freq_sql}
+) AS freq"""
         else:
             cols = []
             for comp in ds.components.values():
                 col = quote_name(comp.name)
                 if comp.name == time_id:
-                    shifted_expr = (
-                        f"CASE WHEN freq.period_ind IN ('M','Q','S','A') "
-                        f"AND CAST({col} AS DATE) = LAST_DAY(CAST({col} AS DATE)) "
-                        f"THEN LAST_DAY(CAST("
-                        f"vtl_dateadd({col}, {shift_sql}, freq.period_ind) AS DATE)) "
-                        f"ELSE vtl_dateadd({col}, {shift_sql}, freq.period_ind) END"
-                    )
-                    cols.append(f"{shifted_expr} AS {col}")
+                    cols.append(f"vtl_dateadd({col}, {shift_sql}, freq.period_ind) AS {col}")
                 else:
                     cols.append(col)