Skip to content

Implement ppl relation subquery command with Calcite #3378

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 1 commit into from
Mar 6, 2025
Merged

Implement ppl relation subquery command with Calcite #3378

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
6 changes: 0 additions & 6 deletions integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalciteSortCommandIT.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,10 +22,4 @@ public void init() throws IOException {
@Ignore
@Override
public void testSortIpField() throws IOException {}

// TODO: Fix incorrect results for NULL values, addressed by issue:
// https://github.com/opensearch-project/sql/issues/3375
@Ignore
@Override
public void testSortWithNullValue() throws IOException {}
}
204 changes: 186 additions & 18 deletions integ-test/src/test/java/org/opensearch/sql/calcite/standalone/CalcitePPLJoinIT.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@
import org.junit.Ignore;
import org.junit.Test;
import org.opensearch.client.Request;
import org.opensearch.sql.legacy.TestsConstants;

public class CalcitePPLJoinIT extends CalcitePPLIntegTestCase {

Expand All @@ -29,22 +30,22 @@ public void init() throws IOException {
loadIndex(Index.OCCUPATION);
loadIndex(Index.HOBBIES);
Request request1 =
new Request("PUT", "/opensearch-sql_test_index_state_country/_doc/5?refresh=true");
new Request("PUT", "/" + TestsConstants.TEST_INDEX_STATE_COUNTRY + "/_doc/5?refresh=true");
request1.setJsonEntity(
"{\"name\":\"Jim\",\"age\":27,\"state\":\"B.C\",\"country\":\"Canada\",\"year\":2023,\"month\":4}");
client().performRequest(request1);
Request request2 =
new Request("PUT", "/opensearch-sql_test_index_state_country/_doc/6?refresh=true");
new Request("PUT", "/" + TestsConstants.TEST_INDEX_STATE_COUNTRY + "/_doc/6?refresh=true");
request2.setJsonEntity(
"{\"name\":\"Peter\",\"age\":57,\"state\":\"B.C\",\"country\":\"Canada\",\"year\":2023,\"month\":4}");
client().performRequest(request2);
Request request3 =
new Request("PUT", "/opensearch-sql_test_index_state_country/_doc/7?refresh=true");
new Request("PUT", "/" + TestsConstants.TEST_INDEX_STATE_COUNTRY + "/_doc/7?refresh=true");
request3.setJsonEntity(
"{\"name\":\"Rick\",\"age\":70,\"state\":\"B.C\",\"country\":\"Canada\",\"year\":2023,\"month\":4}");
client().performRequest(request3);
Request request4 =
new Request("PUT", "/opensearch-sql_test_index_state_country/_doc/8?refresh=true");
new Request("PUT", "/" + TestsConstants.TEST_INDEX_STATE_COUNTRY + "/_doc/8?refresh=true");
request4.setJsonEntity(
"{\"name\":\"David\",\"age\":40,\"state\":\"Washington\",\"country\":\"USA\",\"year\":2023,\"month\":4}");
client().performRequest(request4);
Expand Down Expand Up @@ -213,9 +214,9 @@ public void testComplexLeftJoin() {
actual,
rows("Jane", 20, "Quebec", "Canada", "Scientist", "Canada", 90000),
rows("John", 25, "Ontario", "Canada", "Doctor", "Canada", 120000),
rows("Jim", 27, "B.C", "Canada", null, null, 0),
rows("Peter", 57, "B.C", "Canada", null, null, 0),
rows("Rick", 70, "B.C", "Canada", null, null, 0));
rows("Jim", 27, "B.C", "Canada", null, null, null),
rows("Peter", 57, "B.C", "Canada", null, null, null),
rows("Rick", 70, "B.C", "Canada", null, null, null));
}

@Test
Expand All @@ -240,10 +241,10 @@ public void testComplexRightJoin() {
actual,
rows("Jane", 20, "Quebec", "Canada", "Scientist", "Canada", 90000),
rows("John", 25, "Ontario", "Canada", "Doctor", "Canada", 120000),
rows(null, 0, null, null, "Engineer", "England", 100000),
rows(null, 0, null, null, "Artist", "USA", 70000),
rows(null, 0, null, null, "Doctor", "USA", 120000),
rows(null, 0, null, null, "Unemployed", "Canada", 0));
rows(null, null, null, null, "Engineer", "England", 100000),
rows(null, null, null, null, "Artist", "USA", 70000),
rows(null, null, null, null, "Doctor", "USA", 120000),
rows(null, null, null, null, "Unemployed", "Canada", 0));
}

@Test
Expand Down Expand Up @@ -392,6 +393,54 @@ public void testMultipleJoins() {
TEST_INDEX_STATE_COUNTRY));
}

@Ignore // TODO seems a calcite bug
public void testMultipleJoinsWithRelationSubquery() {
JSONObject actual =
executeQuery(
String.format(
"""
source = %s
| where country = 'Canada' OR country = 'England'
| inner join left=a, right=b
ON a.name = b.name AND a.year = 2023 AND a.month = 4 AND b.year = 2023 AND b.month = 4
[
source = %s
]
| eval a_name = a.name
| eval a_country = a.country
| eval b_country = b.country
| fields a_name, age, state, a_country, occupation, b_country, salary
| left join left=a, right=b
ON a.a_name = b.name
[
source = %s
]
| eval aa_country = a.a_country
| eval ab_country = a.b_country
| eval bb_country = b.country
| fields a_name, age, state, aa_country, occupation, ab_country, salary, bb_country, hobby, language
| cross join left=a, right=b
[
source = %s
]
| eval new_country = a.aa_country
| eval new_salary = b.salary
| stats avg(new_salary) as avg_salary by span(age, 5) as age_span, state
| left semi join left=a, right=b
ON a.state = b.state
[
source = %s
]
| eval new_avg_salary = floor(avg_salary)
| fields state, age_span, new_avg_salary
""",
TEST_INDEX_STATE_COUNTRY,
TEST_INDEX_OCCUPATION,
TEST_INDEX_HOBBIES,
TEST_INDEX_OCCUPATION,
TEST_INDEX_STATE_COUNTRY));
}

@Test
public void testMultipleJoinsWithoutTableAliases() {
JSONObject actual =
Expand Down Expand Up @@ -442,7 +491,7 @@ public void testMultipleJoinsWithPartTableAliases() {
}

@Test
public void testMultipleJoinsWithSelfJoin1() {
public void testMultipleJoinsWithSelfJoin() {
JSONObject actual =
executeQuery(
String.format(
Expand All @@ -469,8 +518,8 @@ public void testMultipleJoinsWithSelfJoin1() {
rows("John", "John", "John", "John"));
}

@Ignore // TODO table subquery not support
public void testMultipleJoinsWithSelfJoin2() {
@Test
public void testMultipleJoinsWithSubquerySelfJoin() {
JSONObject actual =
executeQuery(
String.format(
Expand All @@ -481,10 +530,24 @@ public void testMultipleJoinsWithSelfJoin2() {
TEST_INDEX_OCCUPATION,
TEST_INDEX_HOBBIES,
TEST_INDEX_STATE_COUNTRY));
verifySchema(
actual,
schema("name", "string"),
schema("name0", "string"),
schema("name1", "string"),
schema("name2", "string"));
verifyDataRows(
actual,
rows("David", "David", "David", "David"),
rows("David", "David", "David", "David"),
rows("Hello", "Hello", "Hello", "Hello"),
rows("Jake", "Jake", "Jake", "Jake"),
rows("Jane", "Jane", "Jane", "Jane"),
rows("John", "John", "John", "John"));
}

@Test
public void testCheckAccessTheReferenceByAliases1() {
public void testCheckAccessTheReferenceByAliases() {
String res1 =
execute(
String.format(
Expand Down Expand Up @@ -520,8 +583,8 @@ public void testCheckAccessTheReferenceByAliases1() {
assertEquals(res4, res5);
}

@Ignore // TODO table subquery not support
public void testCheckAccessTheReferenceByAliases2() {
@Test
public void testCheckAccessTheReferenceBySubqueryAliases() {
String res1 =
execute(
String.format(
Expand Down Expand Up @@ -559,7 +622,7 @@ public void testCheckAccessTheReferenceByAliases2() {
}

@Test
public void testCheckAccessTheReferenceByAliases3() {
public void testCheckAccessTheReferenceByOverrideAliases() {
String res1 =
execute(
String.format(
Expand All @@ -581,4 +644,109 @@ public void testCheckAccessTheReferenceByAliases3() {
assertEquals(res1, res2);
assertEquals(res1, res3);
}

@Test
public void testCheckAccessTheReferenceByOverrideSubqueryAliases() {
String res1 =
execute(
String.format(
"source = %s | JOIN left = t1 right = t2 ON t1.name = t2.name [ source = %s as tt ]"
+ " | fields tt.name",
TEST_INDEX_STATE_COUNTRY, TEST_INDEX_OCCUPATION));
String res2 =
execute(
String.format(
"source = %s | JOIN left = t1 right = t2 ON t1.name = t2.name [ source = %s as tt ]"
+ " as t2 | fields tt.name",
TEST_INDEX_STATE_COUNTRY, TEST_INDEX_OCCUPATION));
String res3 =
execute(
String.format(
"source = %s | JOIN left = t1 right = t2 ON t1.name = t2.name [ source = %s ] as tt"
+ " | fields tt.name",
TEST_INDEX_STATE_COUNTRY, TEST_INDEX_OCCUPATION));
assertEquals(res1, res2);
assertEquals(res1, res3);
}

@Test
public void testCheckAccessTheReferenceByOverrideSubqueryAliases2() {
String res1 =
execute(
String.format(
"source = %s | JOIN left = t1 right = t2 ON t1.name = t2.name [ source = %s as tt ]"
+ " | fields t2.name",
TEST_INDEX_STATE_COUNTRY, TEST_INDEX_OCCUPATION));
String res2 =
execute(
String.format(
"source = %s | JOIN left = t1 right = t2 ON t1.name = t2.name [ source = %s as tt ]"
+ " as t2 | fields t2.name",
TEST_INDEX_STATE_COUNTRY, TEST_INDEX_OCCUPATION));
String res3 =
execute(
String.format(
"source = %s | JOIN left = t1 right = t2 ON t1.name = t2.name [ source = %s ] as tt"
+ " | fields t2.name",
TEST_INDEX_STATE_COUNTRY, TEST_INDEX_OCCUPATION));
assertEquals(res1, res2);
assertEquals(res1, res3);
}

@Test
public void testInnerJoinWithRelationSubquery() {
JSONObject actual =
executeQuery(
String.format(
"""
source = %s
| where country = 'USA' OR country = 'England'
| inner join left=a, right=b
ON a.name = b.name
[
source = %s
| where salary > 0
| fields name, country, salary
| sort salary
| head 3
]
| stats avg(salary) by span(age, 10) as age_span, b.country
""",
TEST_INDEX_STATE_COUNTRY, TEST_INDEX_OCCUPATION));
verifySchema(
actual,
schema("b.country", "string"),
schema("age_span", "double"),
schema("avg(salary)", "double"));
verifyDataRows(actual, rows("USA", 30, 70000.0), rows("England", 70, 100000));
}

@Test
public void testLeftJoinWithRelationSubquery() {
JSONObject actual =
executeQuery(
String.format(
"""
source = %s
| where country = 'USA' OR country = 'England'
| left join left=a, right=b
ON a.name = b.name
[
source = %s
| where salary > 0
| fields name, country, salary
| sort salary
| head 3
]
| stats avg(salary) by span(age, 10) as age_span, b.country
""",
TEST_INDEX_STATE_COUNTRY, TEST_INDEX_OCCUPATION));
verifySchema(
actual,
schema("b.country", "string"),
schema("age_span", "double"),
schema("avg(salary)", "double"));
verifyDataRows(
actual, rows("USA", 30, 70000.0), rows("England", 70, 100000), rows(null, 40, 0));
}
}
21 changes: 21 additions & 0 deletions integ-test/src/test/java/org/opensearch/sql/calcite/standalone/CalcitePPLSortIT.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,9 +6,11 @@
package org.opensearch.sql.calcite.standalone;

import static org.opensearch.sql.legacy.TestsConstants.TEST_INDEX_BANK;
import static org.opensearch.sql.legacy.TestsConstants.TEST_INDEX_BANK_WITH_NULL_VALUES;
import static org.opensearch.sql.util.MatcherUtils.rows;
import static org.opensearch.sql.util.MatcherUtils.schema;
import static org.opensearch.sql.util.MatcherUtils.verifyDataRows;
import static org.opensearch.sql.util.MatcherUtils.verifyOrder;
import static org.opensearch.sql.util.MatcherUtils.verifySchema;

import java.io.IOException;
Expand All @@ -22,6 +24,7 @@ public void init() throws IOException {
super.init();

loadIndex(Index.BANK);
loadIndex(Index.BANK_WITH_NULL_VALUES);
}

@Test
Expand Down Expand Up @@ -191,4 +194,22 @@ public void testSortAgeNameAndFieldsNameAge() {
rows("Amber JOHnny", 32),
rows("Nanette", 28));
}

@Test
public void testSortWithNullValue() throws IOException {
JSONObject result =
executeQuery(
String.format(
"source=%s | sort balance | fields firstname, balance",
TEST_INDEX_BANK_WITH_NULL_VALUES));
verifyOrder(
result,
rows("Dale", 4180),
rows("Nanette", 32838),
rows("Amber JOHnny", 39225),
rows("Dillard", 48086),
rows("Hattie", null),
rows("Elinor", null),
rows("Virginia", null));
}
}
2 changes: 1 addition & 1 deletion .../src/main/java/org/opensearch/sql/opensearch/storage/scan/CalciteOpenSearchIndexScan.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -167,7 +167,7 @@ public CalciteOpenSearchIndexScan pushDownFilter(Filter filter) {
// TODO: handle the case where condition contains a score function
return newScan;
} catch (Exception e) {
LOG.warn("Cannot analyze the filter condition {}", filter.getCondition(), e);
LOG.warn("Cannot pushdown the filter condition {}, ", filter.getCondition());
}
return null;
}
Expand Down
Loading
Loading