Skip to content

Commit 691a93c

Browse files
LantaoJinLantaoJin
committed
Use fetchSource API instead of fetchFields for LAST/FIRST/TAKE aggregation
Signed-off-by: Lantao Jin <ltjin@amazon.com>
1 parent 92e73ea commit 691a93c

18 files changed

Lines changed: 185 additions & 45 deletions

File tree

‎integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalciteExplainIT.java‎

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -835,6 +835,18 @@ public void testExplainOnFirstLast() throws IOException {
835835
TEST_INDEX_BANK)));
836836
}
837837

838+
@Test
839+
public void testExplainOnTextFirstLast() throws IOException {
840+
String expected = loadExpectedPlan("explain_first_last_text.yaml");
841+
assertYamlEqualsIgnoreId(
842+
expected,
843+
explainQueryYaml(
844+
String.format(
845+
"source=%s | stats first(employer) as first_employer, last(employer) as"
846+
+ " last_employer by gender",
847+
TEST_INDEX_BANK)));
848+
}
849+
838850
// Only for Calcite
839851
@Test
840852
public void testExplainOnEventstatsEarliestLatest() throws IOException {

‎integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalcitePPLAggregationIT.java‎

Lines changed: 69 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -311,6 +311,75 @@ public void testFirstLastDifferentFields() throws IOException {
311311
verifyDataRows(actual, rows(1L, 48086L, 32L));
312312
}
313313

314+
@Test
315+
public void testFirstAggregationOnTextField() throws IOException {
316+
JSONObject actual =
317+
executeQuery(
318+
String.format("source=%s | stats first(employer), first(email)", TEST_INDEX_BANK));
319+
verifySchema(actual, schema("first(employer)", "string"), schema("first(email)", "string"));
320+
verifyDataRows(actual, rows("Pyrami", "amberduke@pyrami.com"));
321+
}
322+
323+
@Test
324+
public void testLastAggregationOnTextField() throws IOException {
325+
JSONObject actual =
326+
executeQuery(
327+
String.format("source=%s | stats last(employer), first(email)", TEST_INDEX_BANK));
328+
verifySchema(actual, schema("last(employer)", "string"), schema("first(email)", "string"));
329+
verifyDataRows(actual, rows("Quailcom", "amberduke@pyrami.com"));
330+
}
331+
332+
@Test
333+
public void testFirstLastByGroupOnTextField() throws IOException {
334+
JSONObject actual =
335+
executeQuery(
336+
String.format(
337+
"source=%s | stats first(employer), last(email) by gender", TEST_INDEX_BANK));
338+
verifySchema(
339+
actual,
340+
schema("first(employer)", "string"),
341+
schema("last(email)", "string"),
342+
schema("gender", "string"));
343+
verifyDataRows(
344+
actual,
345+
rows("Quility", "dillardmcpherson@quailcom.com", "F"),
346+
rows("Pyrami", "elinorratliff@scentric.com", "M"));
347+
}
348+
349+
@Test
350+
public void testFirstLastWithOtherAggregationsOnTextField() throws IOException {
351+
JSONObject actual =
352+
executeQuery(
353+
String.format(
354+
"source=%s | stats first(employer), last(email), count(), avg(age) by gender",
355+
TEST_INDEX_BANK));
356+
verifySchema(
357+
actual,
358+
schema("first(employer)", "string"),
359+
schema("last(email)", "string"),
360+
schema("count()", "bigint"),
361+
schema("avg(age)", "double"),
362+
schema("gender", "string"));
363+
verifyDataRows(
364+
actual,
365+
rows("Quility", "dillardmcpherson@quailcom.com", 3, 33.666666666666664, "F"),
366+
rows("Pyrami", "elinorratliff@scentric.com", 4, 34.25, "M"));
367+
}
368+
369+
@Test
370+
public void testFirstLastMixedFields() throws IOException {
371+
JSONObject actual =
372+
executeQuery(
373+
String.format(
374+
"source=%s | stats first(employer), last(balance), first(age)", TEST_INDEX_BANK));
375+
verifySchema(
376+
actual,
377+
schema("first(employer)", "string"),
378+
schema("last(balance)", "bigint"),
379+
schema("first(age)", "int"));
380+
verifyDataRows(actual, rows("Pyrami", 48086L, 32));
381+
}
382+
314383
@Test
315384
public void testFirstLastWithBirthdate() throws IOException {
316385
JSONObject actual =

‎integ-test/src/test/resources/expectedOutput/calcite/explain_earliest_latest.yaml‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -6,4 +6,4 @@ calcite:
66
LogicalProject(server=[$1], message=[$3], @timestamp=[$2])
77
CalciteLogicalIndexScan(table=[[OpenSearch, opensearch-sql_test_index_logs]])
88
physical: |
9-
CalciteEnumerableIndexScan(table=[[OpenSearch, opensearch-sql_test_index_logs]], PushDownContext=[[AGGREGATION->rel#:LogicalAggregate.NONE.[](input=RelSubset#,group={0},earliest_message=ARG_MIN($1, $2),latest_message=ARG_MAX($1, $2)), PROJECT->[earliest_message, latest_message, server], LIMIT->10000], OpenSearchRequestBuilder(sourceBuilder={"from":0,"size":0,"timeout":"1m","aggregations":{"composite_buckets":{"composite":{"size":1000,"sources":[{"server":{"terms":{"field":"server","missing_bucket":true,"missing_order":"first","order":"asc"}}}]},"aggregations":{"earliest_message":{"top_hits":{"from":0,"size":1,"version":false,"seq_no_primary_term":false,"explain":false,"fields":[{"field":"message.keyword"}],"sort":[{"@timestamp":{"order":"asc"}}]}},"latest_message":{"top_hits":{"from":0,"size":1,"version":false,"seq_no_primary_term":false,"explain":false,"fields":[{"field":"message.keyword"}],"sort":[{"@timestamp":{"order":"desc"}}]}}}}}}, requestedTotalSize=10000, pageSize=null, startFrom=0)])
9+
CalciteEnumerableIndexScan(table=[[OpenSearch, opensearch-sql_test_index_logs]], PushDownContext=[[AGGREGATION->rel#:LogicalAggregate.NONE.[](input=RelSubset#,group={0},earliest_message=ARG_MIN($1, $2),latest_message=ARG_MAX($1, $2)), PROJECT->[earliest_message, latest_message, server], LIMIT->10000], OpenSearchRequestBuilder(sourceBuilder={"from":0,"size":0,"timeout":"1m","aggregations":{"composite_buckets":{"composite":{"size":1000,"sources":[{"server":{"terms":{"field":"server","missing_bucket":true,"missing_order":"first","order":"asc"}}}]},"aggregations":{"earliest_message":{"top_hits":{"from":0,"size":1,"version":false,"seq_no_primary_term":false,"explain":false,"_source":false,"fields":[{"field":"message.keyword"}],"sort":[{"@timestamp":{"order":"asc"}}]}},"latest_message":{"top_hits":{"from":0,"size":1,"version":false,"seq_no_primary_term":false,"explain":false,"_source":false,"fields":[{"field":"message.keyword"}],"sort":[{"@timestamp":{"order":"desc"}}]}}}}}}, requestedTotalSize=10000, pageSize=null, startFrom=0)])

‎integ-test/src/test/resources/expectedOutput/calcite/explain_earliest_latest_custom_time.yaml‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -6,4 +6,4 @@ calcite:
66
LogicalProject(level=[$4], message=[$3], created_at=[$0])
77
CalciteLogicalIndexScan(table=[[OpenSearch, opensearch-sql_test_index_logs]])
88
physical: |
9-
CalciteEnumerableIndexScan(table=[[OpenSearch, opensearch-sql_test_index_logs]], PushDownContext=[[AGGREGATION->rel#:LogicalAggregate.NONE.[](input=RelSubset#,group={0},earliest_message=ARG_MIN($1, $2),latest_message=ARG_MAX($1, $2)), PROJECT->[earliest_message, latest_message, level], LIMIT->10000], OpenSearchRequestBuilder(sourceBuilder={"from":0,"size":0,"timeout":"1m","aggregations":{"composite_buckets":{"composite":{"size":1000,"sources":[{"level":{"terms":{"field":"level","missing_bucket":true,"missing_order":"first","order":"asc"}}}]},"aggregations":{"earliest_message":{"top_hits":{"from":0,"size":1,"version":false,"seq_no_primary_term":false,"explain":false,"fields":[{"field":"message.keyword"}],"sort":[{"created_at":{"order":"asc"}}]}},"latest_message":{"top_hits":{"from":0,"size":1,"version":false,"seq_no_primary_term":false,"explain":false,"fields":[{"field":"message.keyword"}],"sort":[{"created_at":{"order":"desc"}}]}}}}}}, requestedTotalSize=10000, pageSize=null, startFrom=0)])
9+
CalciteEnumerableIndexScan(table=[[OpenSearch, opensearch-sql_test_index_logs]], PushDownContext=[[AGGREGATION->rel#:LogicalAggregate.NONE.[](input=RelSubset#,group={0},earliest_message=ARG_MIN($1, $2),latest_message=ARG_MAX($1, $2)), PROJECT->[earliest_message, latest_message, level], LIMIT->10000], OpenSearchRequestBuilder(sourceBuilder={"from":0,"size":0,"timeout":"1m","aggregations":{"composite_buckets":{"composite":{"size":1000,"sources":[{"level":{"terms":{"field":"level","missing_bucket":true,"missing_order":"first","order":"asc"}}}]},"aggregations":{"earliest_message":{"top_hits":{"from":0,"size":1,"version":false,"seq_no_primary_term":false,"explain":false,"_source":false,"fields":[{"field":"message.keyword"}],"sort":[{"created_at":{"order":"asc"}}]}},"latest_message":{"top_hits":{"from":0,"size":1,"version":false,"seq_no_primary_term":false,"explain":false,"_source":false,"fields":[{"field":"message.keyword"}],"sort":[{"created_at":{"order":"desc"}}]}}}}}}, requestedTotalSize=10000, pageSize=null, startFrom=0)])

‎integ-test/src/test/resources/expectedOutput/calcite/explain_first_last.yaml‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -6,4 +6,4 @@ calcite:
66
LogicalProject(gender=[$4], firstname=[$1])
77
CalciteLogicalIndexScan(table=[[OpenSearch, opensearch-sql_test_index_bank]])
88
physical: |
9-
CalciteEnumerableIndexScan(table=[[OpenSearch, opensearch-sql_test_index_bank]], PushDownContext=[[AGGREGATION->rel#:LogicalAggregate.NONE.[](input=RelSubset#,group={0},first_name=FIRST($1),last_name=LAST($1)), PROJECT->[first_name, last_name, gender], LIMIT->10000], OpenSearchRequestBuilder(sourceBuilder={"from":0,"size":0,"timeout":"1m","aggregations":{"composite_buckets":{"composite":{"size":1000,"sources":[{"gender":{"terms":{"field":"gender.keyword","missing_bucket":true,"missing_order":"first","order":"asc"}}}]},"aggregations":{"first_name":{"top_hits":{"from":0,"size":1,"version":false,"seq_no_primary_term":false,"explain":false,"fields":[{"field":"firstname"}]}},"last_name":{"top_hits":{"from":0,"size":1,"version":false,"seq_no_primary_term":false,"explain":false,"fields":[{"field":"firstname"}],"sort":[{"_doc":{"order":"desc"}}]}}}}}}, requestedTotalSize=10000, pageSize=null, startFrom=0)])
9+
CalciteEnumerableIndexScan(table=[[OpenSearch, opensearch-sql_test_index_bank]], PushDownContext=[[AGGREGATION->rel#:LogicalAggregate.NONE.[](input=RelSubset#,group={0},first_name=FIRST($1),last_name=LAST($1)), PROJECT->[first_name, last_name, gender], LIMIT->10000], OpenSearchRequestBuilder(sourceBuilder={"from":0,"size":0,"timeout":"1m","aggregations":{"composite_buckets":{"composite":{"size":1000,"sources":[{"gender":{"terms":{"field":"gender.keyword","missing_bucket":true,"missing_order":"first","order":"asc"}}}]},"aggregations":{"first_name":{"top_hits":{"from":0,"size":1,"version":false,"seq_no_primary_term":false,"explain":false,"_source":{"includes":["firstname"],"excludes":[]}}},"last_name":{"top_hits":{"from":0,"size":1,"version":false,"seq_no_primary_term":false,"explain":false,"_source":{"includes":["firstname"],"excludes":[]},"sort":[{"_doc":{"order":"desc"}}]}}}}}}, requestedTotalSize=10000, pageSize=null, startFrom=0)])

‎integ-test/src/test/resources/expectedOutput/calcite/explain_first_last_text.yaml‎

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,9 @@
1+
calcite:
2+
logical: |
3+
LogicalSystemLimit(fetch=[10000], type=[QUERY_SIZE_LIMIT])
4+
LogicalProject(first_employer=[$1], last_employer=[$2], gender=[$0])
5+
LogicalAggregate(group=[{0}], first_employer=[FIRST($1)], last_employer=[LAST($1)])
6+
LogicalProject(gender=[$4], employer=[$8])
7+
CalciteLogicalIndexScan(table=[[OpenSearch, opensearch-sql_test_index_bank]])
8+
physical: |
9+
CalciteEnumerableIndexScan(table=[[OpenSearch, opensearch-sql_test_index_bank]], PushDownContext=[[AGGREGATION->rel#:LogicalAggregate.NONE.[](input=RelSubset#,group={0},first_employer=FIRST($1),last_employer=LAST($1)), PROJECT->[first_employer, last_employer, gender], LIMIT->10000], OpenSearchRequestBuilder(sourceBuilder={"from":0,"size":0,"timeout":"1m","aggregations":{"composite_buckets":{"composite":{"size":1000,"sources":[{"gender":{"terms":{"field":"gender.keyword","missing_bucket":true,"missing_order":"first","order":"asc"}}}]},"aggregations":{"first_employer":{"top_hits":{"from":0,"size":1,"version":false,"seq_no_primary_term":false,"explain":false,"_source":{"includes":["employer"],"excludes":[]}}},"last_employer":{"top_hits":{"from":0,"size":1,"version":false,"seq_no_primary_term":false,"explain":false,"_source":{"includes":["employer"],"excludes":[]},"sort":[{"_doc":{"order":"desc"}}]}}}}}}, requestedTotalSize=10000, pageSize=null, startFrom=0)])

‎integ-test/src/test/resources/expectedOutput/calcite/explain_max_string_field.yaml‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -6,4 +6,4 @@ calcite:
66
CalciteLogicalIndexScan(table=[[OpenSearch, opensearch-sql_test_index_account]])
77
physical: |
88
EnumerableLimit(fetch=[10000])
9-
CalciteEnumerableIndexScan(table=[[OpenSearch, opensearch-sql_test_index_account]], PushDownContext=[[AGGREGATION->rel#:LogicalAggregate.NONE.[](input=RelSubset#,group={},max(firstname)=MAX($0))], OpenSearchRequestBuilder(sourceBuilder={"from":0,"size":0,"timeout":"1m","aggregations":{"max(firstname)":{"top_hits":{"from":0,"size":1,"version":false,"seq_no_primary_term":false,"explain":false,"fields":[{"field":"firstname.keyword"}],"sort":[{"firstname.keyword":{"order":"desc"}}]}}}}, requestedTotalSize=2147483647, pageSize=null, startFrom=0)])
9+
CalciteEnumerableIndexScan(table=[[OpenSearch, opensearch-sql_test_index_account]], PushDownContext=[[AGGREGATION->rel#:LogicalAggregate.NONE.[](input=RelSubset#,group={},max(firstname)=MAX($0))], OpenSearchRequestBuilder(sourceBuilder={"from":0,"size":0,"timeout":"1m","aggregations":{"max(firstname)":{"top_hits":{"from":0,"size":1,"version":false,"seq_no_primary_term":false,"explain":false,"_source":false,"fields":[{"field":"firstname.keyword"}],"sort":[{"firstname.keyword":{"order":"desc"}}]}}}}, requestedTotalSize=2147483647, pageSize=null, startFrom=0)])

‎integ-test/src/test/resources/expectedOutput/calcite/explain_min_string_field.yaml‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -6,4 +6,4 @@ calcite:
66
CalciteLogicalIndexScan(table=[[OpenSearch, opensearch-sql_test_index_account]])
77
physical: |
88
EnumerableLimit(fetch=[10000])
9-
CalciteEnumerableIndexScan(table=[[OpenSearch, opensearch-sql_test_index_account]], PushDownContext=[[AGGREGATION->rel#:LogicalAggregate.NONE.[](input=RelSubset#,group={},min(firstname)=MIN($0))], OpenSearchRequestBuilder(sourceBuilder={"from":0,"size":0,"timeout":"1m","aggregations":{"min(firstname)":{"top_hits":{"from":0,"size":1,"version":false,"seq_no_primary_term":false,"explain":false,"fields":[{"field":"firstname.keyword"}],"sort":[{"firstname.keyword":{"order":"asc"}}]}}}}, requestedTotalSize=2147483647, pageSize=null, startFrom=0)])
9+
CalciteEnumerableIndexScan(table=[[OpenSearch, opensearch-sql_test_index_account]], PushDownContext=[[AGGREGATION->rel#:LogicalAggregate.NONE.[](input=RelSubset#,group={},min(firstname)=MIN($0))], OpenSearchRequestBuilder(sourceBuilder={"from":0,"size":0,"timeout":"1m","aggregations":{"min(firstname)":{"top_hits":{"from":0,"size":1,"version":false,"seq_no_primary_term":false,"explain":false,"_source":false,"fields":[{"field":"firstname.keyword"}],"sort":[{"firstname.keyword":{"order":"asc"}}]}}}}, requestedTotalSize=2147483647, pageSize=null, startFrom=0)])

0 commit comments

Comments
 (0)