fix(flink): fix compilation of over aggregation query in flink backend (

#8359)

ibis/backends/flink/compiler.py

@@ -143,6 +143,13 @@ def _minimize_spec(start, end, spec):
             and end.following
         ):
             return None
+        elif (
+            isinstance(getattr(end, "value", None), ops.Cast)
+            and end.value.arg.value == 0
+            and end.following
+        ):
+            spec.args["end"] = "CURRENT ROW"
+            spec.args["end_side"] = None
         return spec
 
     def visit_TumbleWindowingTVF(self, op, *, table, time_col, window_size, offset):

ibis/backends/flink/tests/snapshots/test_compiler/test_window_aggregation/out.sql

@@ -6,7 +6,7 @@ SELECT
 FROM (
   SELECT
     `t0`.*
-  FROM TABLE(TUMBLE(TABLE `table`, DESCRIPTOR(`i`), INTERVAL '15' MINUTE)) AS `t0`
+  FROM TABLE(TUMBLE(TABLE `table`, DESCRIPTOR(`i`), INTERVAL '15' MINUTE(2))) AS `t0`
 ) AS `t1`
 GROUP BY
   `t1`.`window_start`,

ibis/backends/flink/tests/snapshots/test_compiler/test_window_topn/out.sql

@@ -29,7 +29,7 @@ FROM (
     FROM (
       SELECT
         `t0`.*
-      FROM TABLE(TUMBLE(TABLE `table`, DESCRIPTOR(`i`), INTERVAL '600' SECOND)) AS `t0`
+      FROM TABLE(TUMBLE(TABLE `table`, DESCRIPTOR(`i`), INTERVAL '600' SECOND(3))) AS `t0`
     ) AS `t1`
   ) AS `t2`
 ) AS `t3`

ibis/backends/flink/tests/snapshots/test_compiler/test_windowing_tvf/cumulate_window/out.sql

@@ -1,5 +1,5 @@
 SELECT
   `t0`.*
 FROM TABLE(
-  CUMULATE(TABLE `table`, DESCRIPTOR(`i`), INTERVAL '10' SECOND, INTERVAL '1' MINUTE)
+  CUMULATE(TABLE `table`, DESCRIPTOR(`i`), INTERVAL '10' SECOND(2), INTERVAL '1' MINUTE(2))
 ) AS `t0`

ibis/backends/flink/tests/snapshots/test_compiler/test_windowing_tvf/hop_window/out.sql

@@ -1,3 +1,5 @@
 SELECT
   `t0`.*
-FROM TABLE(HOP(TABLE `table`, DESCRIPTOR(`i`), INTERVAL '1' MINUTE, INTERVAL '15' MINUTE)) AS `t0`
+FROM TABLE(
+  HOP(TABLE `table`, DESCRIPTOR(`i`), INTERVAL '1' MINUTE(2), INTERVAL '15' MINUTE(2))
+) AS `t0`

ibis/backends/flink/tests/snapshots/test_compiler/test_windowing_tvf/tumble_window/out.sql

@@ -1,3 +1,3 @@
 SELECT
   `t0`.*
-FROM TABLE(TUMBLE(TABLE `table`, DESCRIPTOR(`i`), INTERVAL '15' MINUTE)) AS `t0`
+FROM TABLE(TUMBLE(TABLE `table`, DESCRIPTOR(`i`), INTERVAL '15' MINUTE(2))) AS `t0`

ibis/backends/flink/tests/snapshots/test_window/test_range_window/out.sql

@@ -1,3 +1,3 @@
 SELECT
-  SUM(`t0`.`f`) OVER (ORDER BY `t0`.`f` ASC NULLS LAST RANGE BETWEEN INTERVAL '500' MINUTE preceding AND CAST(0 AS INTERVAL MINUTE) following) AS `Sum(f)`
+  SUM(`t0`.`f`) OVER (ORDER BY `t0`.`f` ASC NULLS LAST RANGE BETWEEN INTERVAL '500' MINUTE(3) preceding AND CURRENT ROW) AS `Sum(f)`
 FROM `table` AS `t0`

ibis/backends/sql/dialects.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import contextlib
+import math
 
 import sqlglot.expressions as sge
 from sqlglot import transforms
@@ -68,6 +69,45 @@ class Generator(Postgres.Generator):
         }
 
 
+def _calculate_precision(interval_value: int) -> int:
+    """Calculate interval precision.
+
+    FlinkSQL interval data types use leading precision and fractional-
+    seconds precision. Because the leading precision defaults to 2, we need to
+    specify a different precision when the value exceeds 2 digits.
+
+    (see
+    https://learn.microsoft.com/en-us/sql/odbc/reference/appendixes/interval-literals)
+    """
+    # log10(interval_value) + 1 is equivalent to len(str(interval_value)), but is significantly
+    # faster and more memory-efficient
+    if interval_value == 0:
+        return 0
+    if interval_value < 0:
+        raise ValueError(
+            f"Expecting value to be a non-negative integer, got {interval_value}"
+        )
+    return int(math.log10(interval_value)) + 1
+
+
+def _interval_with_precision(self, e):
+    """Format interval with precision."""
+    arg = e.args["this"].this
+    formatted_arg = arg
+    with contextlib.suppress(AttributeError):
+        formatted_arg = arg.sql(self.dialect)
+
+    unit = e.args["unit"]
+    # when formatting interval scalars, need to quote arg and add precision
+    if isinstance(arg, str):
+        formatted_arg = f"'{formatted_arg}'"
+        prec = _calculate_precision(int(arg))
+        prec = max(prec, 2)
+        unit += f"({prec})"
+
+    return f"INTERVAL {formatted_arg} {unit}"
+
+
 class Flink(Hive):
     class Generator(Hive.Generator):
         TYPE_MAPPING = Hive.Generator.TYPE_MAPPING.copy() | {
@@ -91,6 +131,7 @@ class Generator(Hive.Generator):
             sge.DayOfYear: rename_func("dayofyear"),
             sge.DayOfWeek: rename_func("dayofweek"),
             sge.DayOfMonth: rename_func("dayofmonth"),
+            sge.Interval: _interval_with_precision,
         }
 
     class Tokenizer(Hive.Tokenizer):