From 83beb7e36000c9fb7bda0963d6deaf63268fcdac Mon Sep 17 00:00:00 2001
From: Deepyaman Datta <deepyaman.datta@utexas.edu>
Date: Fri, 22 Sep 2023 13:50:04 -0600
Subject: [PATCH] feat(flink): support translating typed null values

---
 ibis/backends/flink/utils.py | 10 +++++++++-
 1 file changed, 9 insertions(+), 1 deletion(-)

diff --git a/ibis/backends/flink/utils.py b/ibis/backends/flink/utils.py
index 7a05f1561c10..d6e57986f71e 100644
--- a/ibis/backends/flink/utils.py
+++ b/ibis/backends/flink/utils.py
@@ -247,6 +247,9 @@ def _translate_interval(value, dtype):
 
 
 _to_pyflink_types = {
+    dt.String: DataTypes.STRING(),
+    dt.Boolean: DataTypes.BOOLEAN(),
+    dt.Binary: DataTypes.BYTES(),
     dt.Int8: DataTypes.TINYINT(),
     dt.Int16: DataTypes.SMALLINT(),
     dt.Int32: DataTypes.INT(),
@@ -258,6 +261,9 @@ def _translate_interval(value, dtype):
     dt.Float16: DataTypes.FLOAT(),
     dt.Float32: DataTypes.FLOAT(),
     dt.Float64: DataTypes.DOUBLE(),
+    dt.Date: DataTypes.DATE(),
+    dt.Time: DataTypes.TIME(),
+    dt.Timestamp: DataTypes.TIMESTAMP(),
 }
 
 
@@ -266,7 +272,9 @@ def translate_literal(op: ops.Literal) -> str:
     dtype = op.dtype
 
     if value is None:
-        return "NULL"
+        if dtype.is_null():
+            return "NULL"
+        return f"CAST(NULL AS {_to_pyflink_types[type(dtype)]!s})"
 
     if dtype.is_boolean():
         # TODO(chloeh13q): Flink supports a third boolean called "UNKNOWN"