feat(postgres): implement support for asof_join API via a lateral join (#11024)

Co-authored-by: Phillip Cloud <[email protected]>

Author: zblz

ibis/backends/sql/compilers/postgres.py

@@ -739,6 +739,37 @@ def visit_Hash(self, op, *, arg):
             f"{self.dialect} backend"
         )
 
+    def visit_JoinLink(self, op, *, how, table, predicates):
+        if how == "asof":
+            # Convert asof join to a lateral left join
+
+            # The asof match condition is always the first predicate
+            match_condition, *predicates = predicates
+            on = sg.and_(*predicates) if predicates else None
+
+            return sge.Join(
+                this=sge.Lateral(
+                    this=sge.Subquery(
+                        this=sg.select(sge.Star())
+                        .from_(table)
+                        .where(match_condition)
+                        # the ordering for the subquery depends on whether we
+                        # want to pick the one row with the largest or smallest
+                        # value that fulfills the match condition
+                        .order_by(
+                            match_condition.expression.asc()
+                            if match_condition.key in {"lte", "lt"}
+                            else match_condition.expression.desc()
+                        )
+                        .limit(1)
+                    )
+                ).as_(table.alias_or_name),
+                kind="left",
+                on=on,
+            )
+
+        return super().visit_JoinLink(op, how=how, table=table, predicates=predicates)
+
     def visit_TableUnnest(
         self,
         op,

ibis/backends/tests/test_asof_join.py

@@ -86,7 +86,6 @@ def time_keyed_right(time_keyed_df2):
     [
         "datafusion",
         "trino",
-        "postgres",
         "mysql",
         "pyspark",
         "druid",
@@ -96,7 +95,6 @@ def time_keyed_right(time_keyed_df2):
         "oracle",
         "mssql",
         "sqlite",
-        "risingwave",
         "flink",
         "databricks",
         "athena",
@@ -138,9 +136,7 @@ def test_asof_join(con, time_left, time_right, time_df1, time_df2, direction, op
         "mssql",
         "mysql",
         "oracle",
-        "postgres",
         "pyspark",
-        "risingwave",
         "sqlite",
         "trino",
         "athena",