From 84fd059df7aac53a25db52641c4ecd84a3497db8 Mon Sep 17 00:00:00 2001
From: Chloe He <chloehe47@gmail.com>
Date: Fri, 1 Mar 2024 13:10:23 -0800
Subject: [PATCH] ci(flink): update tests that involve unnest in Flink

---
 ibis/backends/tests/test_array.py   | 16 +++++++++++-----
 ibis/backends/tests/test_generic.py |  8 +++++++-
 ibis/backends/tests/test_map.py     |  1 -
 3 files changed, 18 insertions(+), 7 deletions(-)

diff --git a/ibis/backends/tests/test_array.py b/ibis/backends/tests/test_array.py
index 250db248f6711..e30f790b1fc08 100644
--- a/ibis/backends/tests/test_array.py
+++ b/ibis/backends/tests/test_array.py
@@ -261,7 +261,7 @@ def test_unnest_simple(backend):
 
 
 @builtin_array
-@pytest.mark.notimpl(["datafusion", "flink"], raises=com.OperationNotDefinedError)
+@pytest.mark.notimpl(["datafusion"], raises=com.OperationNotDefinedError)
 def test_unnest_complex(backend):
     array_types = backend.array_types
     df = array_types.execute()
@@ -356,7 +356,7 @@ def test_unnest_no_nulls(backend):
     raises=ValueError,
     reason="all the input arrays must have same number of dimensions",
 )
-@pytest.mark.notimpl(["datafusion", "flink"], raises=com.OperationNotDefinedError)
+@pytest.mark.notimpl(["datafusion"], raises=com.OperationNotDefinedError)
 def test_unnest_default_name(backend):
     array_types = backend.array_types
     df = array_types.execute()
@@ -803,10 +803,13 @@ def test_array_intersect(con, data):
 )
 @pytest.mark.notimpl(["postgres"], raises=PsycoPg2SyntaxError)
 @pytest.mark.notimpl(["risingwave"], raises=PsycoPg2InternalError)
-@pytest.mark.notimpl(["datafusion", "flink"], raises=com.OperationNotDefinedError)
+@pytest.mark.notimpl(["datafusion"], raises=com.OperationNotDefinedError)
 @pytest.mark.broken(
     ["trino"], reason="inserting maps into structs doesn't work", raises=TrinoUserError
 )
+@pytest.mark.broken(
+    ["flink"], raises=Py4JJavaError, reason="error on memtable compilation"
+)
 def test_unnest_struct(con):
     data = {"value": [[{"a": 1}, {"a": 2}], [{"a": 3}, {"a": 4}]]}
     t = ibis.memtable(data, schema=ibis.schema({"value": "!array<!struct<a: !int>>"}))
@@ -904,7 +907,7 @@ def test_zip_null(con, fn):
     ["trino"], reason="inserting maps into structs doesn't work", raises=TrinoUserError
 )
 @pytest.mark.broken(
-    ["flink"], raises=Py4JJavaError, reason="cannot insert schema type ROW<MAP<k,v>>"
+    ["flink"], raises=Py4JJavaError, reason="error on memtable compilation"
 )
 def test_array_of_struct_unnest(con):
     jobs = ibis.memtable(
@@ -1082,10 +1085,13 @@ def test_range_start_stop_step_zero(con, start, stop):
     reason="ibis hasn't implemented this behavior yet",
 )
 @pytest.mark.notyet(
-    ["datafusion", "flink"],
+    ["datafusion"],
     raises=com.OperationNotDefinedError,
     reason="backend doesn't support unnest",
 )
+@pytest.mark.broken(
+    ["flink"], raises=Py4JJavaError, reason="error on memtable compilation"
+)
 def test_unnest_empty_array(con):
     t = ibis.memtable({"arr": [[], ["a"], ["a", "b"]]})
     expr = t.arr.unnest()
diff --git a/ibis/backends/tests/test_generic.py b/ibis/backends/tests/test_generic.py
index fceababc49daa..10080d10280c4 100644
--- a/ibis/backends/tests/test_generic.py
+++ b/ibis/backends/tests/test_generic.py
@@ -26,6 +26,7 @@
     MySQLProgrammingError,
     OracleDatabaseError,
     PsycoPg2InternalError,
+    Py4JJavaError,
     PyDruidProgrammingError,
     PyODBCDataError,
     PyODBCProgrammingError,
@@ -1027,7 +1028,7 @@ def query(t, group_cols):
 
 
 @pytest.mark.notimpl(
-    ["dask", "pandas", "oracle", "flink", "exasol"], raises=com.OperationNotDefinedError
+    ["dask", "pandas", "oracle", "exasol"], raises=com.OperationNotDefinedError
 )
 @pytest.mark.notimpl(["druid"], raises=AssertionError)
 @pytest.mark.notyet(
@@ -1040,6 +1041,11 @@ def query(t, group_cols):
     reason="invalid code generated for unnesting a struct",
     raises=TrinoUserError,
 )
+@pytest.mark.broke(
+    ["flink"],
+    reason="invalid code generated for unnesting a struct",
+    raises=Py4JJavaError,
+)
 def test_pivot_longer(backend):
     diamonds = backend.diamonds
     df = diamonds.execute()
diff --git a/ibis/backends/tests/test_map.py b/ibis/backends/tests/test_map.py
index a80945653325f..76ed064a74271 100644
--- a/ibis/backends/tests/test_map.py
+++ b/ibis/backends/tests/test_map.py
@@ -345,7 +345,6 @@ def test_map_length(con):
     assert con.execute(expr) == 2
 
 
-@pytest.mark.notimpl(["flink"], raises=exc.OperationNotDefinedError)
 def test_map_keys_unnest(backend):
     expr = backend.map.kv.keys().unnest()
     result = expr.to_pandas()