feat(trino): table unnest

ibis/backends/trino/compiler.py

@@ -17,6 +17,7 @@
     exclude_nulls_from_array_collect,
     exclude_unsupported_window_frame_from_ops,
 )
+from ibis.util import gen_name
 
 
 class TrinoCompiler(SQLGlotCompiler):
@@ -511,3 +512,34 @@ def visit_ToJSONArray(self, op, *, arg):
             ),
             dt.Array(dt.json),
         )
+
+    def visit_TableUnnest(
+        self, op, *, parent, column, offset: str | None, keep_empty: bool
+    ):
+        quoted = self.quoted
+
+        tmpname = sg.to_identifier(gen_name("table_unnest"), quoted=quoted)
+        column_alias = sg.to_identifier(gen_name("table_unnest_column"), quoted=quoted)
+
+        selcols = [
+            sg.column(col, quoted=quoted)
+            if col != column.alias_or_name
+            else column_alias.as_(col)
+            for col in op.parent.schema.names
+        ]
+
+        if offset is not None:
+            offset_name = offset
+            offset = sg.to_identifier(offset_name, quoted=quoted)
+            selcols.append((offset - 1).as_(offset_name, quoted=quoted))
+
+        expr = sg.select(*selcols)
+
+        alias = sge.TableAlias(this=tmpname, columns=[column_alias])
+        unnest = sge.Unnest(expressions=[column], alias=alias, offset=offset)
+        res = expr.from_(parent).join(
+            unnest,
+            on=None if not keep_empty else sge.convert(True),
+            join_type="CROSS" if not keep_empty else "LEFT",
+        )
+        return res