Merge pull request duckdb/duckdb#7987 from Tmonster/add_asof_join_to_relational

Add JoinReftype to Relational Joins (to add asof, positional, dependent joins)

Author: Mytherin

R/cpp11.R

@@ -76,8 +76,8 @@ rapi_expr_window <- function(window_function, partitions, order_bys, window_boun
   .Call(`_duckdb_rapi_expr_window`, window_function, partitions, order_bys, window_boundary_start, window_boundary_end, start_expr, end_expr, offset_expr, default_expr)
 }
 
-rapi_rel_join <- function(left, right, conds, join) {
-  .Call(`_duckdb_rapi_rel_join`, left, right, conds, join)
+rapi_rel_join <- function(left, right, conds, join, join_ref_type) {
+  .Call(`_duckdb_rapi_rel_join`, left, right, conds, join, join_ref_type)
 }
 
 rapi_rel_union_all <- function(rel_a, rel_b) {

R/relational.R

@@ -212,13 +212,21 @@ expr_window_ <- function (window_function, partitions=list(), order_bys=list(),
 #' rel2 <- rel_join(left, right, cond, "right")
 #' rel2 <- rel_join(left, right, cond, "left")
 #' rel2 <- rel_join(left, right, cond, "outer")
+
 rel_inner_join <- function(left, right, conds) {
-  rel_join(left, right, conds, "inner")
+  rel_join_(left, right, conds, "inner", "regular")
+}
+
+rel_join <- function(left, right, conds, join = "inner", ref_type = "regular") {
+    rel_join_(left, right, conds, join, ref_type)
 }
 
-rel_join <- function(left, right, conds, join = c("inner", "left", "right", "outer", "cross", "semi", "anti")) {
+rel_join_ <- function(left, right, conds,
+                      join = c("inner", "left", "right", "outer", "cross", "semi", "anti"),
+                      join_ref_type = c("regular", "natural", "cross", "positional", "asof")) {
   join <- match.arg(join)
-  rapi_rel_join(left, right, conds, join)
+  join_ref_type <- match.arg(join_ref_type)
+  rapi_rel_join(left, right, conds, join, join_ref_type)
 }
 
 #' UNION ALL on two DuckDB relation objects

src/cpp11.cpp

@@ -146,10 +146,10 @@ extern "C" SEXP _duckdb_rapi_expr_window(SEXP window_function, SEXP partitions,
   END_CPP11
 }
 // relational.cpp
-SEXP rapi_rel_join(duckdb::rel_extptr_t left, duckdb::rel_extptr_t right, list conds, std::string join);
-extern "C" SEXP _duckdb_rapi_rel_join(SEXP left, SEXP right, SEXP conds, SEXP join) {
+SEXP rapi_rel_join(duckdb::rel_extptr_t left, duckdb::rel_extptr_t right, list conds, std::string join, std::string join_ref_type);
+extern "C" SEXP _duckdb_rapi_rel_join(SEXP left, SEXP right, SEXP conds, SEXP join, SEXP join_ref_type) {
   BEGIN_CPP11
-    return cpp11::as_sexp(rapi_rel_join(cpp11::as_cpp<cpp11::decay_t<duckdb::rel_extptr_t>>(left), cpp11::as_cpp<cpp11::decay_t<duckdb::rel_extptr_t>>(right), cpp11::as_cpp<cpp11::decay_t<list>>(conds), cpp11::as_cpp<cpp11::decay_t<std::string>>(join)));
+    return cpp11::as_sexp(rapi_rel_join(cpp11::as_cpp<cpp11::decay_t<duckdb::rel_extptr_t>>(left), cpp11::as_cpp<cpp11::decay_t<duckdb::rel_extptr_t>>(right), cpp11::as_cpp<cpp11::decay_t<list>>(conds), cpp11::as_cpp<cpp11::decay_t<std::string>>(join), cpp11::as_cpp<cpp11::decay_t<std::string>>(join_ref_type)));
   END_CPP11
 }
 // relational.cpp
@@ -396,7 +396,7 @@ static const R_CallMethodDef CallEntries[] = {
     {"_duckdb_rapi_rel_from_df",             (DL_FUNC) &_duckdb_rapi_rel_from_df,             3},
     {"_duckdb_rapi_rel_from_table",          (DL_FUNC) &_duckdb_rapi_rel_from_table,          3},
     {"_duckdb_rapi_rel_from_table_function", (DL_FUNC) &_duckdb_rapi_rel_from_table_function, 4},
-    {"_duckdb_rapi_rel_join",                (DL_FUNC) &_duckdb_rapi_rel_join,                4},
+    {"_duckdb_rapi_rel_join",                (DL_FUNC) &_duckdb_rapi_rel_join,                5},
     {"_duckdb_rapi_rel_limit",               (DL_FUNC) &_duckdb_rapi_rel_limit,               2},
     {"_duckdb_rapi_rel_names",               (DL_FUNC) &_duckdb_rapi_rel_names,               1},
     {"_duckdb_rapi_rel_order",               (DL_FUNC) &_duckdb_rapi_rel_order,               2},

src/relational.cpp

@@ -26,6 +26,8 @@
 #include "duckdb/main/relation/distinct_relation.hpp"
 #include "duckdb/main/relation/table_function_relation.hpp"
 
+#include "duckdb/common/enums/joinref_type.hpp"
+
 using namespace duckdb;
 using namespace cpp11;
 
@@ -300,10 +302,23 @@ bool constant_expression_is_not_null(duckdb::expr_extptr_t expr) {
 }
 
 [[cpp11::register]] SEXP rapi_rel_join(duckdb::rel_extptr_t left, duckdb::rel_extptr_t right, list conds,
-                                       std::string join) {
+                                       std::string join, std::string join_ref_type) {
 	auto join_type = JoinType::INNER;
+	auto ref_type = JoinRefType::REGULAR;
 	unique_ptr<ParsedExpression> cond;
 
+	if (join_ref_type == "regular") {
+		ref_type = JoinRefType::REGULAR;
+	} else if (join_ref_type == "cross") {
+		ref_type = JoinRefType::CROSS;
+	} else if (join_ref_type == "positional") {
+		ref_type = JoinRefType::POSITIONAL;
+	} else if (join_ref_type == "asof") {
+		ref_type = JoinRefType::ASOF;
+	}
+
+	cpp11::writable::list prot = {left, right};
+
 	if (join == "left") {
 		join_type = JoinType::LEFT;
 	} else if (join == "right") {
@@ -314,9 +329,19 @@ bool constant_expression_is_not_null(duckdb::expr_extptr_t expr) {
 		join_type = JoinType::SEMI;
 	} else if (join == "anti") {
 		join_type = JoinType::ANTI;
-	} else if (join == "cross") {
-		auto res = std::make_shared<CrossProductRelation>(left->rel, right->rel);
-		return make_external<RelationWrapper>("duckdb_relation", res);
+	} else if (join == "cross" || ref_type == JoinRefType::POSITIONAL) {
+		if (ref_type != JoinRefType::POSITIONAL) {
+			// users can only supply positional cross join, or cross join.
+			warning("Automatically converting join to cross join");
+			ref_type = JoinRefType::CROSS;
+		}
+		auto res = std::make_shared<CrossProductRelation>(left->rel, right->rel, ref_type);
+		auto rel = make_external_prot<RelationWrapper>("duckdb_relation", prot, res);
+		// if the user described filters, apply them on top of the cross product relation
+		if (conds.size() > 0) {
+			return rapi_rel_filter(rel, conds);
+		}
+		return rel;
 	}
 
 	if (conds.size() == 1) {
@@ -329,10 +354,7 @@ bool constant_expression_is_not_null(duckdb::expr_extptr_t expr) {
 		cond = make_uniq<ConjunctionExpression>(ExpressionType::CONJUNCTION_AND, std::move(cond_args));
 	}
 
-	auto res = std::make_shared<JoinRelation>(left->rel, right->rel, std::move(cond), join_type);
-
-	cpp11::writable::list prot = {left, right};
-
+	auto res = std::make_shared<JoinRelation>(left->rel, right->rel, std::move(cond), join_type, ref_type);
 	return make_external_prot<RelationWrapper>("duckdb_relation", prot, res);
 }
 

tests/testthat/test_relational.R

@@ -268,6 +268,117 @@ test_that("Union all keeps duplicates", {
     expect_equal(rel_df, expected_result)
 })
 
+test_that("Inner join returns all inner relations", {
+    dbExecute(con, "CREATE OR REPLACE MACRO eq(a, b) AS a = b")
+    left <- rel_from_df(con, data.frame(left_a=c(1, 2, 3), left_b=c(1, 1, 2)))
+    right <- rel_from_df(con, data.frame(right_b=c(1, 3), right_c=c(4, 5)))
+    cond <- list(expr_function("eq", list(expr_reference("left_b"), expr_reference("right_b"))))
+    rel2 <- rel_join(left, right, cond, "inner")
+    rel_df <- rel_to_altrep(rel2)
+    dim(rel_df)
+    expected_result <- data.frame(left_a=c(1, 2), left_b=c(1, 1), right_b=c(1, 1), right_c=c(4, 4))
+    expect_equal(rel_df, expected_result)
+})
+
+test_that("ASOF join works", {
+    dbExecute(con, "CREATE OR REPLACE MACRO gte(a, b) AS a >= b")
+    test_df1 <- duckdb:::rel_from_df(con, data.frame(ts=c(1, 2, 3, 4, 5, 6, 7, 8, 9)))
+    test_df2 <- duckdb:::rel_from_df(con, data.frame(event_ts=c(1, 3, 6, 8), event_id=c(0, 1, 2, 3)))
+    cond <- list(duckdb:::expr_function("gte", list(duckdb:::expr_reference("ts"), duckdb:::expr_reference("event_ts"))))
+    rel <- duckdb:::rel_join(test_df1, test_df2, cond, ref_type="asof")
+    rel_proj <- duckdb:::rel_project(rel, list(duckdb:::expr_reference("ts"), duckdb:::expr_reference("event_id")))
+    rel_df <- duckdb:::rel_to_altrep(rel_proj)
+    expected_result <- data.frame(ts=c(1, 2, 3, 4, 5, 6, 7, 8, 9), event_id=c(0, 0, 1, 1, 1, 2, 2, 3, 3))
+    expect_equal(expected_result, rel_df)
+})
+
+test_that("LEFT ASOF join works", {
+    dbExecute(con, "CREATE OR REPLACE MACRO gte(a, b) AS a >= b")
+    test_df1 <- duckdb:::rel_from_df(con, data.frame(ts=c(0, 1, 2, 3, 4, 5, 6, 7, 8, 9)))
+    test_df2 <- duckdb:::rel_from_df(con, data.frame(event_ts=c(2, 4, 6, 8), event_id=c(0, 1, 2, 3)))
+    cond <- list(duckdb:::expr_function("gte", list(duckdb:::expr_reference("ts"), duckdb:::expr_reference("event_ts"))))
+    rel <- duckdb:::rel_join(test_df1, test_df2, cond, join="left", ref_type="asof")
+    rel_proj <- duckdb:::rel_project(rel, list(duckdb:::expr_reference("ts"), duckdb:::expr_reference("event_ts"), duckdb:::expr_reference("event_id")))
+    order <- duckdb:::rel_order(rel_proj, list(duckdb:::expr_reference("ts")))
+    rel_df <- duckdb:::rel_to_altrep(order)
+    expected_result <- data.frame(ts=c(0, 1, 2, 3, 4, 5, 6, 7, 8, 9), event_ts=c(NA, NA,2, 2, 4, 4, 6, 6, 8, 8), event_id=c(NA, NA, 0, 0, 1, 1, 2, 2, 3, 3))
+    expect_equal(expected_result, rel_df)
+})
+
+test_that("Positional cross join works", {
+    test_df1 <- duckdb:::rel_from_df(con, data.frame(a=c(11, 12, 13), b=c(1, 2, 3)))
+    test_df2 <- duckdb:::rel_from_df(con, data.frame(c=c(11, 12), d=c(1, 2)))
+    rel <- duckdb:::rel_join(test_df1, test_df2, list(), join="cross", ref_type="positional")
+    rel_df <- duckdb:::rel_to_altrep(rel)
+    expected_result <- data.frame(a=c(11, 12, 13), b=c(1, 2, 3), c=c(11, 12, NA), d=c(1, 2, NA))
+    expect_equal(expected_result, rel_df)
+})
+
+test_that("regular positional join works", {
+    dbExecute(con, "CREATE OR REPLACE MACRO eq(a, b) AS a = b")
+    test_df1 <- duckdb:::rel_from_df(con, data.frame(a=c(11, 12, 13), b=c(1, 2, 3)))
+    test_df2 <- duckdb:::rel_from_df(con, data.frame(c=c(11, 12, 14, 11), d=c(4, 5, 6, 8)))
+    cond <- duckdb:::expr_function("eq", list(duckdb:::expr_reference("a"), duckdb:::expr_reference("c")))
+    rel <- duckdb:::rel_join(test_df1, test_df2, list(cond), ref_type="positional")
+    rel_df <- duckdb:::rel_to_altrep(rel)
+    expected_result <- data.frame(a=c(11, 12), b=c(1, 2), c=c(11, 12), d=c(4, 5))
+    expect_equal(expected_result, rel_df)
+})
+
+test_that("Invalid asof join condition throws error", {
+    dbExecute(con, "CREATE OR REPLACE MACRO neq(a, b) AS a <> b")
+    test_df1 <- rel_from_df(con, data.frame(ts=c(1, 2, 3, 4, 5, 6, 7, 8, 9)))
+    test_df2 <- rel_from_df(con, data.frame(begin=c(1, 3, 6, 8), value=c(0, 1, 2, 3)))
+    cond <- list(expr_function("neq", list(expr_reference("ts"), expr_reference("begin"))))
+    expect_error(rel_join(test_df1, test_df2, cond, ref_type="asof"), "Binder Error")
+})
+
+test_that("multiple inequality conditions for asof join throws error", {
+    dbExecute(con, "CREATE OR REPLACE MACRO gte(a, b) AS a >= b")
+    test_df1 <- duckdb:::rel_from_df(con, data.frame(ts=c(1, 2, 3, 4, 5, 6, 7, 8, 9)))
+    test_df2 <- duckdb:::rel_from_df(con, data.frame(begin=c(1, 3, 6, 8), value=c(0, 1, 2, 3)))
+    cond1 <- duckdb:::expr_function("gte", list(duckdb:::expr_reference("ts"), duckdb:::expr_reference("begin")))
+    cond2 <- duckdb:::expr_function("gte", list(duckdb:::expr_reference("ts"), duckdb:::expr_reference("value")))
+    conds <- list(cond1, cond2)
+    expect_error(rel_join(test_df1, test_df2, conds, ref_type="asof"), "Binder Error")
+})
+
+
+test_that("Inequality joins work", {
+    dbExecute(con, "CREATE OR REPLACE MACRO gte(a, b) AS a >= b")
+    timing_df <- rel_from_df(con, data.frame(ts=c(1, 2, 3, 4, 5, 6)))
+    events_df <- rel_from_df(con, data.frame(event_ts=c(1, 3, 6, 8), event_id=c(0, 1, 2, 3)))
+    cond <- list(expr_function("gte", list(expr_reference("ts"), expr_reference("event_ts"))))
+    rel <- rel_inner_join(timing_df, events_df, cond)
+    rel_proj <- rel_project(rel, list(expr_reference("ts"), expr_reference("event_ts")))
+    rel_order <- rel_order(rel_proj, list(expr_reference("ts"), expr_reference("event_ts")))
+    rel_df <- rel_to_altrep(rel_order)
+    expected_result <- data.frame(ts=c(1, 2, 3, 3, 4, 4, 5, 5, 6, 6, 6), event_ts=c(1, 1, 1, 3, 1, 3, 1, 3, 1, 3, 6))
+    expect_equal(expected_result, rel_df)
+})
+
+
+test_that("Inequality join works to perform between operation", {
+    dbExecute(con, "CREATE OR REPLACE MACRO gt(a, b) AS a > b")
+    dbExecute(con, "CREATE OR REPLACE MACRO lt(a, b) AS a < b")
+    timing_df <- rel_from_df(con, data.frame(ts=c(1, 2, 3, 4, 5, 6, 7, 8, 9)))
+    events_df <- rel_from_df(con, data.frame(event_ts=c(1, 3, 6, 8), event_id=c(0, 1, 2, 3)))
+    lead <- expr_function("lead", list(expr_reference("event_ts")))
+    window_lead <- expr_window(lead, offset_expr=expr_constant(1))
+    expr_set_alias(window_lead, "lead")
+    proj_window <- rel_project(events_df, list(expr_reference("event_ts"), window_lead, expr_reference("event_id")))
+    cond1 <- expr_function("gt", list(expr_reference("ts"), expr_reference("event_ts")))
+    cond2 <- expr_function("lt", list(expr_reference("ts"), expr_reference("lead")))
+    conds <- list(cond1, cond2)
+    rel <- rel_inner_join(timing_df, proj_window, conds)
+    rel_proj <- rel_project(rel, list(expr_reference("ts")))
+    rel_order <- rel_order(rel_proj, list(expr_reference("ts")))
+    rel_df <- rel_to_altrep(rel_order)
+    expected_result <- data.frame(ts=c(2, 4, 5, 7))
+    expect_equal(expected_result, rel_df)
+})
+
+
 # nobody should do this in reality. It's a pretty dumb idea
 test_that("we can union the same relation to itself", {
      test_df_a2 <- rel_from_df(con, data.frame(a=c('1', '2'), b=c('3', '4')))
@@ -285,10 +396,10 @@ test_that("we throw an error when attempting to union all relations that are not
 })
 
 test_that("A union with different column types casts to the richer type", {
-     test_df_a1 <- duckdb:::rel_from_df(con, data.frame(a=c(1)))
-     test_df_a2 <- duckdb:::rel_from_df(con, data.frame(a=c('1')))
-     rel <- duckdb:::rel_union_all(test_df_a1, test_df_a2)
-     res <- duckdb:::rapi_rel_to_df(rel)
+     test_df_a1 <- rel_from_df(con, data.frame(a=c(1)))
+     test_df_a2 <- rel_from_df(con, data.frame(a=c('1')))
+     rel <- rel_union_all(test_df_a1, test_df_a2)
+     res <- rapi_rel_to_df(rel)
      expected <- data.frame(a=c('1.0', '1'))
      expect_equal(class(res$a), class(expected$a))
      expect_equal(res$a[1], expected$a[1])
@@ -644,9 +755,8 @@ test_that("rel_to_sql works for row_number", {
     expect_equal(sub_str_sql, "SELECT row_number() OVER () AS ___row_number")
 })
 
-
 test_that("rel_from_table_function works", {
-    rel <- duckdb:::rel_from_table_function(duckdb:::default_connection(), 'generate_series', list(1L, 10L, 2L))
+    rel <- rel_from_table_function(default_connection(), 'generate_series', list(1L, 10L, 2L))
     df <- as.data.frame(rel)
     expect_equal(df$generate_series, c(1,3,5,7,9))
 })