Skip to content

Commit 785d02b

Browse files
authored
Add Join and Lookup IT on OpenSearch Table (#1025)
Signed-off-by: Peng Huo <[email protected]>
1 parent 98579e1 commit 785d02b

File tree

2 files changed

+56
-0
lines changed

2 files changed

+56
-0
lines changed

docs/opensearch-table.md

+6
Original file line numberDiff line numberDiff line change
@@ -48,6 +48,12 @@ val df = spark.sql("SELECT * FROM dev.default.`my_index*`")
4848
df.show()
4949
```
5050

51+
Join two indices
52+
```scala
53+
val df = spark.sql("SELECT * FROM dev.default.my_index as t1 JOIN dev.default.my_index as t2 ON t1.id == t2.id")
54+
df.show()
55+
```
56+
5157
## Limitation
5258
### catalog operation
5359
- List Tables: Not supported.
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,50 @@
1+
/*
2+
* Copyright OpenSearch Contributors
3+
* SPDX-License-Identifier: Apache-2.0
4+
*/
5+
6+
package org.apache.spark.opensearch.table
7+
8+
import org.opensearch.flint.spark.ppl.FlintPPLSuite
9+
10+
import org.apache.spark.sql.Row
11+
12+
/**
13+
* Test queries on OpenSearch Table
14+
*/
15+
class OpenSearchTableQueryITSuite extends OpenSearchCatalogSuite with FlintPPLSuite {
16+
test("SQL Join two indices") {
17+
val indexName1 = "t0001"
18+
val indexName2 = "t0002"
19+
withIndexName(indexName1) {
20+
withIndexName(indexName2) {
21+
simpleIndex(indexName1)
22+
simpleIndex(indexName2)
23+
val df = spark.sql(s"""
24+
SELECT t1.accountId, t2.eventName, t2.eventSource
25+
FROM ${catalogName}.default.$indexName1 as t1 JOIN ${catalogName}.default.$indexName2 as t2 ON
26+
t1.accountId == t2.accountId""")
27+
28+
checkAnswer(df, Row("123", "event", "source"))
29+
}
30+
}
31+
}
32+
33+
test("PPL Lookup") {
34+
val indexName1 = "t0001"
35+
val indexName2 = "t0002"
36+
val factTbl = s"${catalogName}.default.$indexName1"
37+
val lookupTbl = s"${catalogName}.default.$indexName2"
38+
withIndexName(indexName1) {
39+
withIndexName(indexName2) {
40+
simpleIndex(indexName1)
41+
simpleIndex(indexName2)
42+
43+
val df = spark.sql(
44+
s"source = $factTbl | stats count() by accountId " +
45+
s"| LOOKUP $lookupTbl accountId REPLACE eventSource")
46+
checkAnswer(df, Row(1, "123", "source"))
47+
}
48+
}
49+
}
50+
}

0 commit comments

Comments
 (0)