Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
102 changes: 1 addition & 101 deletions datafusion/core/tests/sql/joins.rs
Original file line number Diff line number Diff line change
Expand Up @@ -521,33 +521,6 @@ async fn asof_join_all_match_directions_across_batches() -> Result<()> {
Ok(())
}

#[tokio::test]
async fn asof_join_coerces_equality_and_match_types() -> Result<()> {
let ctx = SessionContext::new();
let batches = ctx
.sql(
"SELECT t.id, p.price \
FROM (VALUES (CAST(1 AS INT), CAST(4 AS INT), 7)) t(k, ts, id) \
ASOF JOIN \
(VALUES (CAST(1 AS BIGINT), CAST(2 AS BIGINT), 20)) p(k, ts, price) \
MATCH_CONDITION (t.ts >= p.ts) ON t.k = p.k",
)
.await?
.collect()
.await?;
assert_batches_eq!(
[
"+----+-------+",
"| id | price |",
"+----+-------+",
"| 7 | 20 |",
"+----+-------+",
],
&batches
);
Ok(())
}

#[tokio::test]
async fn asof_join_broadcasts_multi_partition_right_input() -> Result<()> {
let config = SessionConfig::new().with_target_partitions(4);
Expand Down Expand Up @@ -639,28 +612,6 @@ async fn asof_join_projection_drops_output_ordering() -> Result<()> {
Ok(())
}

#[tokio::test]
async fn asof_join_explain_names_equality_and_match_conditions() -> Result<()> {
let ctx = SessionContext::new();
register_asof_test_tables(&ctx)?;
let batches = ctx
.sql(
"EXPLAIN SELECT t.trade_id, p.price FROM trades t \
ASOF JOIN prices p MATCH_CONDITION (t.ts >= p.ts) \
ON t.symbol = p.symbol",
)
.await?
.collect()
.await?;
let explain = arrow::util::pretty::pretty_format_batches(&batches)?.to_string();
assert_contains!(explain.as_str(), "AsOf Join: match=[t.ts >= p.ts]");
assert_contains!(explain.as_str(), "on=[t.symbol = p.symbol]");
assert_contains!(explain.as_str(), "AsOfJoinExec:");
assert_contains!(explain.as_str(), "on=[(symbol = symbol)]");
assert_contains!(explain.as_str(), "match=[ts >= ts]");
Ok(())
}

#[tokio::test]
async fn asof_join_rejects_unbounded_inputs_during_physical_planning() -> Result<()> {
let ctx = SessionContext::new();
Expand Down Expand Up @@ -698,7 +649,7 @@ async fn asof_join_rejects_unbounded_inputs_during_physical_planning() -> Result
}

#[tokio::test]
async fn asof_join_using_preserves_key_access_and_unparser_round_trips() -> Result<()> {
async fn asof_join_using_unparser_round_trips() -> Result<()> {
let ctx = SessionContext::new();
register_asof_test_tables(&ctx)?;
let df = ctx
Expand All @@ -707,45 +658,11 @@ async fn asof_join_using_preserves_key_access_and_unparser_round_trips() -> Resu
MATCH_CONDITION (t.ts >= p.ts) USING (symbol)",
)
.await?;
assert_eq!(
df.schema()
.fields()
.iter()
.map(|field| field.name())
.collect::<Vec<_>>(),
vec!["ts", "trade_id", "symbol", "ts", "price"]
);
let sql = plan_to_sql(df.logical_plan())?.to_string();
assert!(sql.contains("ASOF JOIN"));
assert!(sql.contains("MATCH_CONDITION"));
assert!(sql.contains("USING(symbol)"), "unexpected SQL: {sql}");
ctx.sql(&sql).await?;

let batches = ctx
.sql(
"SELECT t.trade_id, t.symbol AS left_symbol, p.symbol AS right_symbol \
FROM trades t ASOF JOIN prices p \
MATCH_CONDITION (t.ts >= p.ts) USING (symbol) \
ORDER BY t.trade_id",
)
.await?
.collect()
.await?;
assert_batches_eq!(
[
"+----------+-------------+--------------+",
"| trade_id | left_symbol | right_symbol |",
"+----------+-------------+--------------+",
"| 1 | A | |",
"| 2 | A | A |",
"| 3 | A | A |",
"| 4 | B | B |",
"| 5 | B | B |",
"| 6 | | |",
"+----------+-------------+--------------+",
],
&batches
);
Ok(())
}

Expand Down Expand Up @@ -784,20 +701,3 @@ async fn asof_join_unparser_preserves_right_preselection() -> Result<()> {
}
Ok(())
}

#[tokio::test]
async fn asof_join_rejects_invalid_contracts() -> Result<()> {
let ctx = SessionContext::new();
register_asof_test_tables(&ctx)?;
for sql in [
"SELECT * FROM trades t ASOF JOIN prices p MATCH_CONDITION (t.ts = p.ts) ON t.symbol = p.symbol",
"SELECT * FROM trades t ASOF JOIN prices p MATCH_CONDITION (p.ts >= t.ts) ON t.symbol = p.symbol",
"SELECT * FROM trades t ASOF JOIN prices p MATCH_CONDITION (t.ts >= p.ts) ON t.symbol > p.symbol",
"SELECT * FROM trades t ASOF JOIN prices p MATCH_CONDITION (1 >= p.ts) ON t.symbol = p.symbol",
"SELECT * FROM trades t ASOF JOIN prices p MATCH_CONDITION (t.ts >= p.ts) ON 1 = 1",
"SELECT * FROM trades t ASOF JOIN prices p MATCH_CONDITION (t.ts >= p.ts AND t.ts > p.ts) ON t.symbol = p.symbol",
] {
assert!(ctx.sql(sql).await.is_err(), "query should fail: {sql}");
}
Ok(())
}
Loading
Loading