Skip to content

Commit c6a3a70

Browse files
ahkcsahkcs
authored
Fix: Support Alias Fields in MIN, MAX, FIRST, LAST, and TAKE Aggregations (opensearch-project#4621)
1 parent 85dc8d9 commit c6a3a70

38 files changed

Lines changed: 357 additions & 146 deletions

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

Lines changed: 158 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,158 @@
1+
/*
2+
* Copyright OpenSearch Contributors
3+
* SPDX-License-Identifier: Apache-2.0
4+
*/
5+
6+
package org.opensearch.sql.calcite.remote;
7+
8+
import static org.opensearch.sql.util.MatcherUtils.rows;
9+
import static org.opensearch.sql.util.MatcherUtils.schema;
10+
import static org.opensearch.sql.util.MatcherUtils.verifyDataRows;
11+
import static org.opensearch.sql.util.MatcherUtils.verifySchema;
12+
import static org.opensearch.sql.util.MatcherUtils.verifySchemaInOrder;
13+
14+
import java.io.IOException;
15+
import java.util.List;
16+
import org.json.JSONObject;
17+
import org.junit.jupiter.api.Test;
18+
import org.opensearch.client.Request;
19+
import org.opensearch.client.ResponseException;
20+
import org.opensearch.sql.ppl.PPLIntegTestCase;
21+
22+
/**
23+
* Integration tests for aggregation functions (MIN, MAX, FIRST, LAST, TAKE) with alias fields.
24+
* Tests the fix for issue #4595.
25+
*/
26+
public class CalciteAliasFieldAggregationIT extends PPLIntegTestCase {
27+
28+
private static final String TEST_INDEX_ALIAS = "test_alias_bug";
29+
30+
@Override
31+
public void init() throws Exception {
32+
super.init();
33+
enableCalcite();
34+
createTestIndexWithAliasFields();
35+
}
36+
37+
/**
38+
* Create test index with alias fields mapping and insert sample data. This mirrors the
39+
* reproduction steps from issue #4595.
40+
*/
41+
private void createTestIndexWithAliasFields() throws IOException {
42+
// Delete the index if it exists (for test isolation)
43+
try {
44+
Request deleteIndex = new Request("DELETE", "/" + TEST_INDEX_ALIAS);
45+
client().performRequest(deleteIndex);
46+
} catch (ResponseException e) {
47+
// Index doesn't exist, which is fine
48+
}
49+
50+
// Create index with alias fields
51+
Request createIndex = new Request("PUT", "/" + TEST_INDEX_ALIAS);
52+
createIndex.setJsonEntity(
53+
"{\n"
54+
+ " \"mappings\": {\n"
55+
+ " \"properties\": {\n"
56+
+ " \"created_at\": {\"type\": \"date\"},\n"
57+
+ " \"@timestamp\": {\"type\": \"alias\", \"path\": \"created_at\"},\n"
58+
+ " \"value\": {\"type\": \"integer\"},\n"
59+
+ " \"value_alias\": {\"type\": \"alias\", \"path\": \"value\"}\n"
60+
+ " }\n"
61+
+ " }\n"
62+
+ "}");
63+
client().performRequest(createIndex);
64+
65+
// Insert test documents
66+
Request bulkRequest = new Request("POST", "/" + TEST_INDEX_ALIAS + "/_bulk?refresh=true");
67+
bulkRequest.setJsonEntity(
68+
"{\"index\":{}}\n"
69+
+ "{\"created_at\": \"2024-01-01T10:00:00Z\", \"value\": 100}\n"
70+
+ "{\"index\":{}}\n"
71+
+ "{\"created_at\": \"2024-01-02T10:00:00Z\", \"value\": 200}\n"
72+
+ "{\"index\":{}}\n"
73+
+ "{\"created_at\": \"2024-01-03T10:00:00Z\", \"value\": 300}\n");
74+
client().performRequest(bulkRequest);
75+
}
76+
77+
@Test
78+
public void testMinWithDateAliasField() throws IOException {
79+
JSONObject actual =
80+
executeQuery(String.format("source=%s | stats MIN(@timestamp)", TEST_INDEX_ALIAS));
81+
verifySchema(actual, schema("MIN(@timestamp)", "timestamp"));
82+
verifyDataRows(actual, rows("2024-01-01 10:00:00"));
83+
}
84+
85+
@Test
86+
public void testMaxWithDateAliasField() throws IOException {
87+
JSONObject actual =
88+
executeQuery(String.format("source=%s | stats MAX(@timestamp)", TEST_INDEX_ALIAS));
89+
verifySchema(actual, schema("MAX(@timestamp)", "timestamp"));
90+
verifyDataRows(actual, rows("2024-01-03 10:00:00"));
91+
}
92+
93+
@Test
94+
public void testMinMaxWithNumericAliasField() throws IOException {
95+
JSONObject actual =
96+
executeQuery(
97+
String.format(
98+
"source=%s | stats MIN(value_alias), MAX(value_alias)", TEST_INDEX_ALIAS));
99+
verifySchemaInOrder(
100+
actual, schema("MIN(value_alias)", "int"), schema("MAX(value_alias)", "int"));
101+
verifyDataRows(actual, rows(100, 300));
102+
}
103+
104+
@Test
105+
public void testFirstWithAliasField() throws IOException {
106+
JSONObject actual =
107+
executeQuery(
108+
String.format(
109+
"source=%s | sort @timestamp | stats FIRST(@timestamp)", TEST_INDEX_ALIAS));
110+
verifySchema(actual, schema("FIRST(@timestamp)", "timestamp"));
111+
verifyDataRows(actual, rows("2024-01-01 10:00:00"));
112+
}
113+
114+
@Test
115+
public void testLastWithAliasField() throws IOException {
116+
JSONObject actual =
117+
executeQuery(
118+
String.format(
119+
"source=%s | sort @timestamp | stats LAST(@timestamp)", TEST_INDEX_ALIAS));
120+
verifySchema(actual, schema("LAST(@timestamp)", "timestamp"));
121+
verifyDataRows(actual, rows("2024-01-03 10:00:00"));
122+
}
123+
124+
@Test
125+
public void testTakeWithAliasField() throws IOException {
126+
JSONObject actual =
127+
executeQuery(
128+
String.format(
129+
"source=%s | sort @timestamp | stats TAKE(@timestamp, 2)", TEST_INDEX_ALIAS));
130+
verifySchema(actual, schema("TAKE(@timestamp, 2)", "array"));
131+
verifyDataRows(actual, rows(List.of("2024-01-01T10:00:00.000Z", "2024-01-02T10:00:00.000Z")));
132+
}
133+
134+
@Test
135+
public void testAggregationsWithOriginalFieldsStillWork() throws IOException {
136+
JSONObject actual =
137+
executeQuery(
138+
String.format("source=%s | stats MIN(created_at), MAX(value)", TEST_INDEX_ALIAS));
139+
verifySchemaInOrder(
140+
actual, schema("MIN(created_at)", "timestamp"), schema("MAX(value)", "int"));
141+
verifyDataRows(actual, rows("2024-01-01 10:00:00", 300));
142+
}
143+
144+
@Test
145+
public void testUnaffectedAggregationsWithAliasFields() throws IOException {
146+
JSONObject actual =
147+
executeQuery(
148+
String.format(
149+
"source=%s | stats SUM(value_alias), AVG(value_alias), COUNT(value_alias)",
150+
TEST_INDEX_ALIAS));
151+
verifySchemaInOrder(
152+
actual,
153+
schema("SUM(value_alias)", "bigint"),
154+
schema("AVG(value_alias)", "double"),
155+
schema("COUNT(value_alias)", "bigint"));
156+
verifyDataRows(actual, rows(600, 200.0, 3));
157+
}
158+
}

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

Lines changed: 15 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -671,10 +671,10 @@ public void testExplainRegexMatchInEvalWithOutScriptPushdown() throws IOExceptio
671671
// Only for Calcite
672672
@Test
673673
public void testExplainOnEarliestLatest() throws IOException {
674-
String expected = loadExpectedPlan("explain_earliest_latest.json");
675-
assertJsonEqualsIgnoreId(
674+
String expected = loadExpectedPlan("explain_earliest_latest.yaml");
675+
assertYamlEqualsIgnoreId(
676676
expected,
677-
explainQueryToString(
677+
explainQueryYaml(
678678
String.format(
679679
"source=%s | stats earliest(message) as earliest_message, latest(message) as"
680680
+ " latest_message by server",
@@ -684,10 +684,10 @@ public void testExplainOnEarliestLatest() throws IOException {
684684
// Only for Calcite
685685
@Test
686686
public void testExplainOnEarliestLatestWithCustomTimeField() throws IOException {
687-
String expected = loadExpectedPlan("explain_earliest_latest_custom_time.json");
688-
assertJsonEqualsIgnoreId(
687+
String expected = loadExpectedPlan("explain_earliest_latest_custom_time.yaml");
688+
assertYamlEqualsIgnoreId(
689689
expected,
690-
explainQueryToString(
690+
explainQueryYaml(
691691
String.format(
692692
"source=%s | stats earliest(message, created_at) as earliest_message,"
693693
+ " latest(message, created_at) as latest_message by level",
@@ -697,10 +697,10 @@ public void testExplainOnEarliestLatestWithCustomTimeField() throws IOException
697697
// Only for Calcite
698698
@Test
699699
public void testExplainOnFirstLast() throws IOException {
700-
String expected = loadExpectedPlan("explain_first_last.json");
701-
assertJsonEqualsIgnoreId(
700+
String expected = loadExpectedPlan("explain_first_last.yaml");
701+
assertYamlEqualsIgnoreId(
702702
expected,
703-
explainQueryToString(
703+
explainQueryYaml(
704704
String.format(
705705
"source=%s | stats first(firstname) as first_name, last(firstname) as"
706706
+ " last_name by gender",
@@ -896,18 +896,18 @@ public void testPushdownLimitIntoAggregation() throws IOException {
896896

897897
@Test
898898
public void testExplainMaxOnStringField() throws IOException {
899-
String expected = loadExpectedPlan("explain_max_string_field.json");
900-
assertJsonEqualsIgnoreId(
899+
String expected = loadExpectedPlan("explain_max_string_field.yaml");
900+
assertYamlEqualsIgnoreId(
901901
expected,
902-
explainQueryToString("source=opensearch-sql_test_index_account | stats max(firstname)"));
902+
explainQueryYaml("source=opensearch-sql_test_index_account | stats max(firstname)"));
903903
}
904904

905905
@Test
906906
public void testExplainMinOnStringField() throws IOException {
907-
String expected = loadExpectedPlan("explain_min_string_field.json");
908-
assertJsonEqualsIgnoreId(
907+
String expected = loadExpectedPlan("explain_min_string_field.yaml");
908+
assertYamlEqualsIgnoreId(
909909
expected,
910-
explainQueryToString("source=opensearch-sql_test_index_account | stats min(firstname)"));
910+
explainQueryYaml("source=opensearch-sql_test_index_account | stats min(firstname)"));
911911
}
912912

913913
@Test

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

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -608,10 +608,10 @@ public void testDifferentFilterScriptPushDownBehaviorExplain() throws Exception
608608

609609
@Test
610610
public void testExplainOnTake() throws IOException {
611-
String expected = loadExpectedPlan("explain_take.json");
612-
assertJsonEqualsIgnoreId(
611+
String expected = loadExpectedPlan("explain_take.yaml");
612+
assertYamlEqualsIgnoreId(
613613
expected,
614-
explainQueryToString(
614+
explainQueryYaml(
615615
"source=opensearch-sql_test_index_account | stats take(firstname, 2) as take"));
616616
}
617617

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

Lines changed: 0 additions & 6 deletions
This file was deleted.

‎integ-test/src/test/resources/expectedOutput/calcite/explain_earliest_latest.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(earliest_message=[$1], latest_message=[$2], server=[$0])
5+
LogicalAggregate(group=[{0}], earliest_message=[ARG_MIN($1, $2)], latest_message=[ARG_MAX($1, $2)])
6+
LogicalProject(server=[$1], message=[$3], @timestamp=[$2])
7+
CalciteLogicalIndexScan(table=[[OpenSearch, opensearch-sql_test_index_logs]])
8+
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=2147483647, pageSize=null, startFrom=0)])

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

Lines changed: 0 additions & 6 deletions
This file was deleted.

‎integ-test/src/test/resources/expectedOutput/calcite/explain_earliest_latest_custom_time.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(earliest_message=[$1], latest_message=[$2], level=[$0])
5+
LogicalAggregate(group=[{0}], earliest_message=[ARG_MIN($1, $2)], latest_message=[ARG_MAX($1, $2)])
6+
LogicalProject(level=[$4], message=[$3], created_at=[$0])
7+
CalciteLogicalIndexScan(table=[[OpenSearch, opensearch-sql_test_index_logs]])
8+
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=2147483647, pageSize=null, startFrom=0)])

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

Lines changed: 0 additions & 6 deletions
This file was deleted.

‎integ-test/src/test/resources/expectedOutput/calcite/explain_first_last.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_name=[$1], last_name=[$2], gender=[$0])
5+
LogicalAggregate(group=[{0}], first_name=[FIRST($1)], last_name=[LAST($1)])
6+
LogicalProject(gender=[$4], firstname=[$1])
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_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=2147483647, pageSize=null, startFrom=0)])

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

Lines changed: 0 additions & 6 deletions
This file was deleted.

0 commit comments

Comments
 (0)