fix(flink): correct the filtered count translation

ibis/backends/flink/registry.py

@@ -18,7 +18,13 @@
 
 
 def _count_star(translator: ExprTranslator, op: ops.Node) -> str:
-    return "count(*)"
+    # TODO(deepyaman): Use `FILTER` syntax; see note on `_filter` below.
+    if (where := op.where) is not None:
+        condition = f"CASE WHEN {translator.translate(where)} THEN 1 END"
+    else:
+        condition = "*"
+
+    return f"COUNT({condition})"
 
 
 def _date(translator: ExprTranslator, op: ops.Node) -> str:

ibis/backends/flink/tests/snapshots/test_translator/test_translate_complex_filtered_agg/out.sql

@@ -1,4 +1,4 @@
-SELECT t0.`b`, count(*) AS `total`, avg(t0.`a`) AS `avg_a`,
+SELECT t0.`b`, COUNT(*) AS `total`, avg(t0.`a`) AS `avg_a`,
        avg(CASE WHEN t0.`g` = 'A' THEN t0.`a` ELSE NULL END) AS `avg_a_A`,
        avg(CASE WHEN t0.`g` = 'B' THEN t0.`a` ELSE NULL END) AS `avg_a_B`
 FROM table t0

ibis/backends/flink/tests/snapshots/test_translator/test_translate_complex_groupby_aggregation/out.sql

@@ -1,5 +1,5 @@
 SELECT EXTRACT(year from t0.`i`) AS `year`,
-       EXTRACT(month from t0.`i`) AS `month`, count(*) AS `total`,
+       EXTRACT(month from t0.`i`) AS `month`, COUNT(*) AS `total`,
        count(DISTINCT t0.`b`) AS `b_unique`
 FROM table t0
 GROUP BY EXTRACT(year from t0.`i`), EXTRACT(month from t0.`i`)

ibis/backends/flink/tests/snapshots/test_translator/test_translate_count_star/out.sql

@@ -1,3 +1,3 @@
-SELECT t0.`i`, count(*) AS `CountStar(table)`
+SELECT t0.`i`, COUNT(*) AS `CountStar(table)`
 FROM table t0
 GROUP BY t0.`i`

ibis/backends/flink/tests/snapshots/test_translator/test_translate_having/out.sql

@@ -1,4 +1,4 @@
 SELECT t0.`g`, sum(t0.`b`) AS `b_sum`
 FROM table t0
 GROUP BY t0.`g`
-HAVING count(*) >= CAST(1000 AS SMALLINT)
+HAVING COUNT(*) >= CAST(1000 AS SMALLINT)

ibis/backends/flink/tests/snapshots/test_translator/test_translate_value_counts/out.sql

@@ -1,4 +1,4 @@
-SELECT t0.`ExtractYear(i)`, count(*) AS `ExtractYear(i)_count`
+SELECT t0.`ExtractYear(i)`, COUNT(*) AS `ExtractYear(i)_count`
 FROM (
   SELECT EXTRACT(year from t1.`i`) AS `ExtractYear(i)`
   FROM table t1