Skip to content

Commit df9f5dd

Browse files
yuancuyuancu
authored
Support Limit pushdown (#3615)
* Support Limit pushdown Signed-off-by: Yuanchun Shen <yuanchu@amazon.com> * Minor: add copyright header Signed-off-by: Yuanchun Shen <yuanchu@amazon.com> * Refactor: create a isLogicalSortLimit method to enhance readability Signed-off-by: Yuanchun Shen <yuanchu@amazon.com> * Disable limit-then-filter pushdown with Calcite Signed-off-by: Yuanchun Shen <yuanchu@amazon.com> * Disable limit-then-agg pushdown & add ITs for multiple head clauses Signed-off-by: Yuanchun Shen <yuanchu@amazon.com> * Add limit-filter-limit pushdown IT Signed-off-by: Yuanchun Shen <yuanchu@amazon.com> * Disable limit-then-filter pushdown without Calcite Signed-off-by: Yuanchun Shen <yuanchu@amazon.com> * Move predicates of limit pushdown ahead Signed-off-by: Yuanchun Shen <yuanchu@amazon.com> * Accumulate offsets when pushing down limit Signed-off-by: Yuanchun Shen <yuanchu@amazon.com> * Improve error message when failed to pushdown limit Signed-off-by: Yuanchun Shen <yuanchu@amazon.com> * Update limit pushdown error message Signed-off-by: Yuanchun Shen <yuanchu@amazon.com> * Fix multiple limit pushdown with shirinking caused by offset Signed-off-by: Yuanchun Shen <yuanchu@amazon.com> --------- Signed-off-by: Yuanchun Shen <yuanchu@amazon.com>
1 parent 3a9475f commit df9f5dd

26 files changed

Lines changed: 472 additions & 16 deletions

‎integ-test/src/test/java/org/opensearch/sql/ppl/ExplainIT.java‎

Lines changed: 98 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -95,7 +95,6 @@ public void testSortPushDownExplain() throws Exception {
9595

9696
@Test
9797
public void testLimitPushDownExplain() throws Exception {
98-
// TODO fix after https://github.com/opensearch-project/sql/issues/3381
9998
String expected =
10099
isCalciteEnabled()
101100
? loadFromFile("expectedOutput/calcite/explain_limit_push.json")
@@ -110,6 +109,104 @@ public void testLimitPushDownExplain() throws Exception {
110109
+ "| fields ageMinus"));
111110
}
112111

112+
@Test
113+
public void testLimitWithFilterPushdownExplain() throws Exception {
114+
String expectedFilterThenLimit =
115+
isCalciteEnabled()
116+
? loadFromFile("expectedOutput/calcite/explain_filter_then_limit_push.json")
117+
: loadFromFile("expectedOutput/ppl/explain_filter_then_limit_push.json");
118+
assertJsonEqualsIgnoreId(
119+
expectedFilterThenLimit,
120+
explainQueryToString(
121+
"source=opensearch-sql_test_index_account"
122+
+ "| where age > 30 "
123+
+ "| head 5 "
124+
+ "| fields age"));
125+
126+
// The filter in limit-then-filter queries should not be pushed since the current DSL will
127+
// execute it as filter-then-limit
128+
String expectedLimitThenFilter =
129+
isCalciteEnabled()
130+
? loadFromFile("expectedOutput/calcite/explain_limit_then_filter_push.json")
131+
: loadFromFile("expectedOutput/ppl/explain_limit_then_filter_push.json");
132+
assertJsonEqualsIgnoreId(
133+
expectedLimitThenFilter,
134+
explainQueryToString(
135+
"source=opensearch-sql_test_index_account"
136+
+ "| head 5 "
137+
+ "| where age > 30 "
138+
+ "| fields age"));
139+
}
140+
141+
@Test
142+
public void testMultipleLimitExplain() throws Exception {
143+
String expected5Then10 =
144+
isCalciteEnabled()
145+
? loadFromFile("expectedOutput/calcite/explain_limit_5_10_push.json")
146+
: loadFromFile("expectedOutput/ppl/explain_limit_5_10_push.json");
147+
assertJsonEqualsIgnoreId(
148+
expected5Then10,
149+
explainQueryToString(
150+
"source=opensearch-sql_test_index_account"
151+
+ "| head 5 "
152+
+ "| head 10 "
153+
+ "| fields age"));
154+
155+
String expected10Then5 =
156+
isCalciteEnabled()
157+
? loadFromFile("expectedOutput/calcite/explain_limit_10_5_push.json")
158+
: loadFromFile("expectedOutput/ppl/explain_limit_10_5_push.json");
159+
assertJsonEqualsIgnoreId(
160+
expected10Then5,
161+
explainQueryToString(
162+
"source=opensearch-sql_test_index_account"
163+
+ "| head 10 "
164+
+ "| head 5 "
165+
+ "| fields age"));
166+
167+
String expected10from1then10from2 =
168+
isCalciteEnabled()
169+
? loadFromFile("expectedOutput/calcite/explain_limit_10from1_10from2_push.json")
170+
: loadFromFile("expectedOutput/ppl/explain_limit_10from1_10from2_push.json");
171+
assertJsonEqualsIgnoreId(
172+
expected10from1then10from2,
173+
explainQueryToString(
174+
"source=opensearch-sql_test_index_account"
175+
+ "| head 10 from 1 "
176+
+ "| head 10 from 2 "
177+
+ "| fields age"));
178+
179+
// The second limit should not be pushed down for limit-filter-limit queries
180+
String expected10ThenFilterThen5 =
181+
isCalciteEnabled()
182+
? loadFromFile("expectedOutput/calcite/explain_limit_10_filter_5_push.json")
183+
: loadFromFile("expectedOutput/ppl/explain_limit_10_filter_5_push.json");
184+
assertJsonEqualsIgnoreId(
185+
expected10ThenFilterThen5,
186+
explainQueryToString(
187+
"source=opensearch-sql_test_index_account"
188+
+ "| head 10 "
189+
+ "| where age > 30 "
190+
+ "| head 5 "
191+
+ "| fields age"));
192+
}
193+
194+
@Test
195+
public void testLimitWithMultipleOffsetPushdownExplain() throws Exception {
196+
String expected =
197+
isCalciteEnabled()
198+
? loadFromFile("expectedOutput/calcite/explain_limit_offsets_push.json")
199+
: loadFromFile("expectedOutput/ppl/explain_limit_offsets_push.json");
200+
201+
assertJsonEqualsIgnoreId(
202+
expected,
203+
explainQueryToString(
204+
"source=opensearch-sql_test_index_account"
205+
+ "| head 10 from 1 "
206+
+ "| head 5 from 2 "
207+
+ "| fields age"));
208+
}
209+
113210
@Test
114211
public void testFillNullPushDownExplain() throws Exception {
115212
String expected =

‎integ-test/src/test/resources/expectedOutput/calcite/explain_filter_then_limit_push.json‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,6 @@
1+
{
2+
"calcite": {
3+
"logical": "LogicalProject(age=[$8])\n LogicalSort(fetch=[5])\n LogicalFilter(condition=[>($8, 30)])\n CalciteLogicalIndexScan(table=[[OpenSearch, opensearch-sql_test_index_account]])\n",
4+
"physical": "CalciteEnumerableIndexScan(table=[[OpenSearch, opensearch-sql_test_index_account]], PushDownContext=[[FILTER->>($8, 30), LIMIT->5, PROJECT->[age]], OpenSearchRequestBuilder(sourceBuilder={\"from\":0,\"size\":5,\"timeout\":\"1m\",\"query\":{\"range\":{\"age\":{\"from\":30,\"to\":null,\"include_lower\":false,\"include_upper\":true,\"boost\":1.0}}},\"_source\":{\"includes\":[\"age\"],\"excludes\":[]},\"sort\":[{\"_doc\":{\"order\":\"asc\"}}]}, requestedTotalSize=5, pageSize=null, startFrom=0)])\n"
5+
}
6+
}

‎integ-test/src/test/resources/expectedOutput/calcite/explain_limit_10_5_push.json‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,6 @@
1+
{
2+
"calcite": {
3+
"logical": "LogicalProject(age=[$8])\n LogicalSort(fetch=[5])\n LogicalSort(fetch=[10])\n CalciteLogicalIndexScan(table=[[OpenSearch, opensearch-sql_test_index_account]])\n",
4+
"physical": "CalciteEnumerableIndexScan(table=[[OpenSearch, opensearch-sql_test_index_account]], PushDownContext=[[LIMIT->10, LIMIT->5, PROJECT->[age]], OpenSearchRequestBuilder(sourceBuilder={\"from\":0,\"size\":5,\"timeout\":\"1m\",\"_source\":{\"includes\":[\"age\"],\"excludes\":[]}}, requestedTotalSize=5, pageSize=null, startFrom=0)])\n"
5+
}
6+
}

‎integ-test/src/test/resources/expectedOutput/calcite/explain_limit_10_filter_5_push.json‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,6 @@
1+
{
2+
"calcite": {
3+
"logical": "LogicalProject(age=[$8])\n LogicalSort(fetch=[5])\n LogicalFilter(condition=[>($8, 30)])\n LogicalSort(fetch=[10])\n CalciteLogicalIndexScan(table=[[OpenSearch, opensearch-sql_test_index_account]])\n",
4+
"physical": "EnumerableLimit(fetch=[5])\n EnumerableCalc(expr#0=[{inputs}], expr#1=[30], expr#2=[>($t0, $t1)], age=[$t0], $condition=[$t2])\n CalciteEnumerableIndexScan(table=[[OpenSearch, opensearch-sql_test_index_account]], PushDownContext=[[LIMIT->10, PROJECT->[age]], OpenSearchRequestBuilder(sourceBuilder={\"from\":0,\"size\":10,\"timeout\":\"1m\",\"_source\":{\"includes\":[\"age\"],\"excludes\":[]}}, requestedTotalSize=10, pageSize=null, startFrom=0)])\n"
5+
}
6+
}

‎integ-test/src/test/resources/expectedOutput/calcite/explain_limit_10from1_10from2_push.json‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,6 @@
1+
{
2+
"calcite": {
3+
"logical": "LogicalProject(age=[$8])\n LogicalSort(offset=[2], fetch=[10])\n LogicalSort(offset=[1], fetch=[10])\n CalciteLogicalIndexScan(table=[[OpenSearch, opensearch-sql_test_index_account]])\n",
4+
"physical": "CalciteEnumerableIndexScan(table=[[OpenSearch, opensearch-sql_test_index_account]], PushDownContext=[[LIMIT->10, LIMIT->10, PROJECT->[age]], OpenSearchRequestBuilder(sourceBuilder={\"from\":3,\"size\":8,\"timeout\":\"1m\",\"_source\":{\"includes\":[\"age\"],\"excludes\":[]}}, requestedTotalSize=8, pageSize=null, startFrom=3)])\n"
5+
}
6+
}

‎integ-test/src/test/resources/expectedOutput/calcite/explain_limit_5_10_push.json‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,6 @@
1+
{
2+
"calcite": {
3+
"logical": "LogicalProject(age=[$8])\n LogicalSort(fetch=[10])\n LogicalSort(fetch=[5])\n CalciteLogicalIndexScan(table=[[OpenSearch, opensearch-sql_test_index_account]])\n",
4+
"physical": "EnumerableLimit(fetch=[10])\n CalciteEnumerableIndexScan(table=[[OpenSearch, opensearch-sql_test_index_account]], PushDownContext=[[LIMIT->5, PROJECT->[age]], OpenSearchRequestBuilder(sourceBuilder={\"from\":0,\"size\":5,\"timeout\":\"1m\",\"_source\":{\"includes\":[\"age\"],\"excludes\":[]}}, requestedTotalSize=5, pageSize=null, startFrom=0)])\n"
5+
}
6+
}

‎integ-test/src/test/resources/expectedOutput/calcite/explain_limit_offsets_push.json‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,6 @@
1+
{
2+
"calcite": {
3+
"logical": "LogicalProject(age=[$8])\n LogicalSort(offset=[2], fetch=[5])\n LogicalSort(offset=[1], fetch=[10])\n CalciteLogicalIndexScan(table=[[OpenSearch, opensearch-sql_test_index_account]])\n",
4+
"physical": "CalciteEnumerableIndexScan(table=[[OpenSearch, opensearch-sql_test_index_account]], PushDownContext=[[LIMIT->10, LIMIT->5, PROJECT->[age]], OpenSearchRequestBuilder(sourceBuilder={\"from\":3,\"size\":5,\"timeout\":\"1m\",\"_source\":{\"includes\":[\"age\"],\"excludes\":[]}}, requestedTotalSize=5, pageSize=null, startFrom=3)])\n"
5+
}
6+
}

‎integ-test/src/test/resources/expectedOutput/calcite/explain_limit_push.json‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,6 @@
11
{
22
"calcite": {
33
"logical": "LogicalProject(ageMinus=[$17])\n LogicalSort(fetch=[5])\n LogicalProject(account_number=[$0], firstname=[$1], address=[$2], balance=[$3], gender=[$4], city=[$5], employer=[$6], state=[$7], age=[$8], email=[$9], lastname=[$10], _id=[$11], _index=[$12], _score=[$13], _maxscore=[$14], _sort=[$15], _routing=[$16], ageMinus=[-($8, 30)])\n CalciteLogicalIndexScan(table=[[OpenSearch, opensearch-sql_test_index_account]])\n",
4-
"physical": "EnumerableCalc(expr#0=[{inputs}], expr#1=[30], expr#2=[-($t0, $t1)], $f0=[$t2])\n EnumerableLimit(fetch=[5])\n CalciteEnumerableIndexScan(table=[[OpenSearch, opensearch-sql_test_index_account]], PushDownContext=[[PROJECT->[age]], OpenSearchRequestBuilder(sourceBuilder={\"from\":0,\"timeout\":\"1m\",\"_source\":{\"includes\":[\"age\"],\"excludes\":[]}}, requestedTotalSize=2147483647, pageSize=null, startFrom=0)])\n"
4+
"physical": "EnumerableCalc(expr#0=[{inputs}], expr#1=[30], expr#2=[-($t0, $t1)], $f0=[$t2])\n CalciteEnumerableIndexScan(table=[[OpenSearch, opensearch-sql_test_index_account]], PushDownContext=[[LIMIT->5, PROJECT->[age]], OpenSearchRequestBuilder(sourceBuilder={\"from\":0,\"size\":5,\"timeout\":\"1m\",\"_source\":{\"includes\":[\"age\"],\"excludes\":[]}}, requestedTotalSize=5, pageSize=null, startFrom=0)])\n"
55
}
66
}

‎integ-test/src/test/resources/expectedOutput/calcite/explain_limit_then_filter_push.json‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,6 @@
1+
{
2+
"calcite": {
3+
"logical": "LogicalProject(age=[$8])\n LogicalFilter(condition=[>($8, 30)])\n LogicalSort(fetch=[5])\n CalciteLogicalIndexScan(table=[[OpenSearch, opensearch-sql_test_index_account]])\n",
4+
"physical": "EnumerableCalc(expr#0=[{inputs}], expr#1=[30], expr#2=[>($t0, $t1)], age=[$t0], $condition=[$t2])\n CalciteEnumerableIndexScan(table=[[OpenSearch, opensearch-sql_test_index_account]], PushDownContext=[[LIMIT->5, PROJECT->[age]], OpenSearchRequestBuilder(sourceBuilder={\"from\":0,\"size\":5,\"timeout\":\"1m\",\"_source\":{\"includes\":[\"age\"],\"excludes\":[]}}, requestedTotalSize=5, pageSize=null, startFrom=0)])\n"
5+
}
6+
}

‎integ-test/src/test/resources/expectedOutput/ppl/explain_filter_then_limit_push.json‎

Lines changed: 17 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,17 @@
1+
{
2+
"root": {
3+
"name": "ProjectOperator",
4+
"description": {
5+
"fields": "[age]"
6+
},
7+
"children": [
8+
{
9+
"name": "OpenSearchIndexScan",
10+
"description": {
11+
"request": "OpenSearchQueryRequest(indexName=opensearch-sql_test_index_account, sourceBuilder={\"from\":0,\"size\":5,\"timeout\":\"1m\",\"query\":{\"range\":{\"age\":{\"from\":30,\"to\":null,\"include_lower\":false,\"include_upper\":true,\"boost\":1.0}}},\"_source\":{\"includes\":[\"age\"],\"excludes\":[]},\"sort\":[{\"_doc\":{\"order\":\"asc\"}}]}, needClean=true, searchDone=false, pitId=null, cursorKeepAlive=null, searchAfter=null, searchResponse=null)"
12+
},
13+
"children": []
14+
}
15+
]
16+
}
17+
}

0 commit comments

Comments
 (0)