Skip to content

Commit 9807b2d

Browse files
LantaoJinLantaoJin
committed
By-Span-Date aggregation should always not represent null bucket
Signed-off-by: Lantao Jin <ltjin@amazon.com>
1 parent 0e5802b commit 9807b2d

22 files changed

Lines changed: 322 additions & 90 deletions

‎core/src/main/java/org/opensearch/sql/ast/expression/SpanUnit.java‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -42,6 +42,11 @@ public enum SpanUnit {
4242
SPAN_UNITS = builder.add(SpanUnit.values()).build();
4343
}
4444

45+
/** Util method to check if the unit is time unit. */
46+
public static boolean isTimeUnit(SpanUnit unit) {
47+
return unit != UNKNOWN && unit != NONE;
48+
}
49+
4550
/** Util method to get span unit given the unit name. */
4651
public static SpanUnit of(String unit) {
4752
switch (unit) {

‎core/src/main/java/org/opensearch/sql/calcite/CalciteRelNodeVisitor.java‎

Lines changed: 12 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -80,6 +80,8 @@
8080
import org.opensearch.sql.ast.expression.ParseMethod;
8181
import org.opensearch.sql.ast.expression.PatternMethod;
8282
import org.opensearch.sql.ast.expression.PatternMode;
83+
import org.opensearch.sql.ast.expression.Span;
84+
import org.opensearch.sql.ast.expression.SpanUnit;
8385
import org.opensearch.sql.ast.expression.UnresolvedExpression;
8486
import org.opensearch.sql.ast.expression.WindowFrame;
8587
import org.opensearch.sql.ast.expression.WindowFrame.FrameType;
@@ -913,7 +915,10 @@ public RelNode visitAggregation(Aggregation node, CalcitePlanContext context) {
913915
Argument.ArgumentMap statsArgs = Argument.ArgumentMap.of(node.getArgExprList());
914916
Boolean bucketNullable =
915917
(Boolean) statsArgs.getOrDefault(Argument.BUCKET_NULLABLE, Literal.TRUE).getValue();
916-
if (!bucketNullable && !aliasedGroupByList.isEmpty()) {
918+
// by-span-time (single) ignores the bucket_nullable arg
919+
if (!bucketNullable
920+
&& !aliasedGroupByList.isEmpty()
921+
&& !(aliasedGroupByList.size() == 1 && isTimeSpan(span))) {
917922
final RelHint statHits =
918923
RelHint.builder("stats_args").hintOption(Argument.BUCKET_NULLABLE, "false").build();
919924
assert context.relBuilder.peek() instanceof LogicalAggregate
@@ -951,6 +956,12 @@ public RelNode visitAggregation(Aggregation node, CalcitePlanContext context) {
951956
return context.relBuilder.peek();
952957
}
953958

959+
private boolean isTimeSpan(UnresolvedExpression expr) {
960+
return Objects.nonNull(expr)
961+
&& expr instanceof Span span
962+
&& SpanUnit.isTimeUnit(span.getUnit());
963+
}
964+
954965
/** extract the RexLiteral of Alias from a node */
955966
private Optional<RexLiteral> extractAliasLiteral(RexNode node) {
956967
if (node == null) {

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

Lines changed: 0 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -5,8 +5,6 @@
55

66
package org.opensearch.sql.calcite.remote;
77

8-
import static org.junit.Assert.assertTrue;
9-
import static org.opensearch.sql.legacy.TestUtils.*;
108
import static org.opensearch.sql.legacy.TestsConstants.TEST_INDEX_BANK;
119
import static org.opensearch.sql.legacy.TestsConstants.TEST_INDEX_LOGS;
1210
import static org.opensearch.sql.legacy.TestsConstants.TEST_INDEX_NESTED_SIMPLE;

‎integ-test/src/test/java/org/opensearch/sql/legacy/SQLIntegTestCase.java‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -867,6 +867,11 @@ public enum Index {
867867
"time_data",
868868
getMappingFile("time_test_data_index_mapping.json"),
869869
"src/test/resources/time_test_data.json"),
870+
TIME_TEST_DATA_WITH_NULL(
871+
TestsConstants.TEST_INDEX_TIME_DATE_NULL,
872+
"time_data_with_null",
873+
getMappingFile("time_test_data_index_mapping.json"),
874+
"src/test/resources/time_test_data_with_null.json"),
870875
EVENTS(
871876
"events",
872877
"events",

‎integ-test/src/test/java/org/opensearch/sql/legacy/TestsConstants.java‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -81,6 +81,7 @@ public class TestsConstants {
8181
public static final String TEST_INDEX_HDFS_LOGS = TEST_INDEX + "_hdfs_logs";
8282
public static final String TEST_INDEX_LOGS = TEST_INDEX + "_logs";
8383
public static final String TEST_INDEX_OTEL_LOGS = TEST_INDEX + "_otel_logs";
84+
public static final String TEST_INDEX_TIME_DATE_NULL = TEST_INDEX + "_time_date_null";
8485

8586
public static final String DATE_FORMAT = "yyyy-MM-dd'T'HH:mm:ss.SSS'Z'";
8687
public static final String TS_DATE_FORMAT = "yyyy-MM-dd HH:mm:ss.SSS";

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

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -478,6 +478,14 @@ public void testStatsByTimeSpan() throws IOException {
478478
expected,
479479
explainQueryToString(
480480
String.format("source=%s | stats count() by span(birthdate,1M)", TEST_INDEX_BANK)));
481+
482+
// bucket_nullable doesn't impact by-span-time
483+
assertJsonEqualsIgnoreId(
484+
expected,
485+
explainQueryToString(
486+
String.format(
487+
"source=%s | stats bucket_nullable=false count() by span(birthdate,1M)",
488+
TEST_INDEX_BANK)));
481489
}
482490

483491
@Ignore("https://github.com/opensearch-project/OpenSearch/issues/3725")

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

Lines changed: 23 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -8,6 +8,7 @@
88
import static org.opensearch.sql.legacy.TestsConstants.TEST_INDEX_ACCOUNT;
99
import static org.opensearch.sql.legacy.TestsConstants.TEST_INDEX_BANK;
1010
import static org.opensearch.sql.legacy.TestsConstants.TEST_INDEX_BANK_WITH_NULL_VALUES;
11+
import static org.opensearch.sql.legacy.TestsConstants.TEST_INDEX_TIME_DATE_NULL;
1112
import static org.opensearch.sql.util.MatcherUtils.rows;
1213
import static org.opensearch.sql.util.MatcherUtils.schema;
1314
import static org.opensearch.sql.util.MatcherUtils.verifyDataRows;
@@ -27,6 +28,7 @@ public void init() throws Exception {
2728
loadIndex(Index.ACCOUNT);
2829
loadIndex(Index.BANK_WITH_NULL_VALUES);
2930
loadIndex(Index.BANK);
31+
loadIndex(Index.TIME_TEST_DATA_WITH_NULL);
3032
}
3133

3234
@Test
@@ -749,4 +751,25 @@ public void testDisableLegacyPreferred() throws IOException {
749751
rows(null, 36));
750752
});
751753
}
754+
755+
@Test
756+
public void testStatsBySpanTimeWithNullBucket() throws IOException {
757+
JSONObject response =
758+
executeQuery(
759+
String.format(
760+
"source=%s | stats percentile(value, 50) as p50 by span(@timestamp, 12h) as"
761+
+ " half_day",
762+
TEST_INDEX_TIME_DATE_NULL));
763+
verifySchema(response, schema("p50", null, "int"), schema("half_day", null, "timestamp"));
764+
verifyDataRows(
765+
response,
766+
rows(8523, "2025-07-28 00:00:00"),
767+
rows(8094, "2025-07-28 12:00:00"),
768+
rows(8429, "2025-07-29 00:00:00"),
769+
rows(8216, "2025-07-29 12:00:00"),
770+
rows(8493, "2025-07-30 00:00:00"),
771+
rows(8426, "2025-07-30 12:00:00"),
772+
rows(8213, "2025-07-31 00:00:00"),
773+
rows(8490, "2025-07-31 12:00:00"));
774+
}
752775
}

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

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,6 @@
11
{
22
"calcite": {
33
"logical": "LogicalSystemLimit(sort0=[$0], dir0=[ASC-nulls-first], fetch=[10000], type=[QUERY_SIZE_LIMIT])\n LogicalSort(sort0=[$0], dir0=[ASC-nulls-first])\n LogicalProject(count()=[$1], state=[$0])\n LogicalFilter(condition=[IS NOT NULL($0)])\n LogicalAggregate(group=[{0}], count()=[COUNT()])\n LogicalProject(state=[$7])\n CalciteLogicalIndexScan(table=[[OpenSearch, opensearch-sql_test_index_account]])\n",
4-
"physical": "EnumerableLimit(fetch=[10000])\n EnumerableSort(sort0=[$0], dir0=[ASC-nulls-first])\n CalciteEnumerableIndexScan(table=[[OpenSearch, opensearch-sql_test_index_account]], PushDownContext=[[FILTER->IS NOT NULL($7), AGGREGATION->rel#:LogicalAggregate.NONE.[](input=RelSubset#,group={0},count()=COUNT()), PROJECT->[count(), state]], OpenSearchRequestBuilder(sourceBuilder={\"from\":0,\"size\":0,\"timeout\":\"1m\",\"query\":{\"exists\":{\"field\":\"state\",\"boost\":1.0}},\"sort\":[],\"aggregations\":{\"state\":{\"terms\":{\"field\":\"state.keyword\",\"size\":1000,\"min_doc_count\":1,\"shard_min_doc_count\":0,\"show_term_doc_count_error\":false,\"order\":{\"_key\":\"asc\"}},\"aggregations\":{\"count()\":{\"value_count\":{\"field\":\"_index\"}}}}}}, requestedTotalSize=2147483647, pageSize=null, startFrom=0)])\n"
4+
"physical": "EnumerableLimit(fetch=[10000])\n EnumerableSort(sort0=[$0], dir0=[ASC-nulls-first])\n CalciteEnumerableIndexScan(table=[[OpenSearch, opensearch-sql_test_index_account]], PushDownContext=[[FILTER->IS NOT NULL($7), AGGREGATION->rel#:LogicalAggregate.NONE.[](input=RelSubset#,group={0},count()=COUNT()), PROJECT->[count(), state]], OpenSearchRequestBuilder(sourceBuilder={\"from\":0,\"size\":0,\"timeout\":\"1m\",\"query\":{\"exists\":{\"field\":\"state\",\"boost\":1.0}},\"sort\":[],\"aggregations\":{\"composite_buckets\":{\"composite\":{\"size\":1000,\"sources\":[{\"state\":{\"terms\":{\"field\":\"state.keyword\",\"missing_bucket\":false,\"order\":\"asc\"}}}]},\"aggregations\":{\"count()\":{\"value_count\":{\"field\":\"_index\"}}}}}}, requestedTotalSize=2147483647, pageSize=null, startFrom=0)])\n"
55
}
66
}

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

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,6 @@
11
{
22
"calcite": {
33
"logical": "LogicalSystemLimit(sort0=[$0], dir0=[ASC-nulls-first], fetch=[10000], type=[QUERY_SIZE_LIMIT])\n LogicalSort(sort0=[$0], dir0=[ASC-nulls-first])\n LogicalProject(count()=[$2], gender=[$0], state=[$1])\n LogicalFilter(condition=[AND(IS NOT NULL($0), IS NOT NULL($1))])\n LogicalAggregate(group=[{0, 1}], count()=[COUNT()])\n LogicalProject(gender=[$4], state=[$7])\n CalciteLogicalIndexScan(table=[[OpenSearch, opensearch-sql_test_index_account]])\n",
4-
"physical": "EnumerableLimit(fetch=[10000])\n EnumerableSort(sort0=[$0], dir0=[ASC-nulls-first])\n CalciteEnumerableIndexScan(table=[[OpenSearch, opensearch-sql_test_index_account]], PushDownContext=[[FILTER->AND(IS NOT NULL($4), IS NOT NULL($7)), AGGREGATION->rel#:LogicalAggregate.NONE.[](input=RelSubset#,group={0, 1},count()=COUNT()), PROJECT->[count(), gender, state]], OpenSearchRequestBuilder(sourceBuilder={\"from\":0,\"size\":0,\"timeout\":\"1m\",\"query\":{\"bool\":{\"must\":[{\"exists\":{\"field\":\"gender\",\"boost\":1.0}},{\"exists\":{\"field\":\"state\",\"boost\":1.0}}],\"adjust_pure_negative\":true,\"boost\":1.0}},\"sort\":[],\"aggregations\":{\"composite_buckets\":{\"composite\":{\"size\":1000,\"sources\":[{\"gender\":{\"terms\":{\"field\":\"gender.keyword\",\"missing_bucket\":true,\"missing_order\":\"first\",\"order\":\"asc\"}}},{\"state\":{\"terms\":{\"field\":\"state.keyword\",\"missing_bucket\":true,\"missing_order\":\"first\",\"order\":\"asc\"}}}]},\"aggregations\":{\"count()\":{\"value_count\":{\"field\":\"_index\"}}}}}}, requestedTotalSize=2147483647, pageSize=null, startFrom=0)])\n"
4+
"physical": "EnumerableLimit(fetch=[10000])\n EnumerableSort(sort0=[$0], dir0=[ASC-nulls-first])\n CalciteEnumerableIndexScan(table=[[OpenSearch, opensearch-sql_test_index_account]], PushDownContext=[[FILTER->AND(IS NOT NULL($4), IS NOT NULL($7)), AGGREGATION->rel#:LogicalAggregate.NONE.[](input=RelSubset#,group={0, 1},count()=COUNT()), PROJECT->[count(), gender, state]], OpenSearchRequestBuilder(sourceBuilder={\"from\":0,\"size\":0,\"timeout\":\"1m\",\"query\":{\"bool\":{\"must\":[{\"exists\":{\"field\":\"gender\",\"boost\":1.0}},{\"exists\":{\"field\":\"state\",\"boost\":1.0}}],\"adjust_pure_negative\":true,\"boost\":1.0}},\"sort\":[],\"aggregations\":{\"composite_buckets\":{\"composite\":{\"size\":1000,\"sources\":[{\"gender\":{\"terms\":{\"field\":\"gender.keyword\",\"missing_bucket\":false,\"order\":\"asc\"}}},{\"state\":{\"terms\":{\"field\":\"state.keyword\",\"missing_bucket\":false,\"order\":\"asc\"}}}]},\"aggregations\":{\"count()\":{\"value_count\":{\"field\":\"_index\"}}}}}}, requestedTotalSize=2147483647, pageSize=null, startFrom=0)])\n"
55
}
66
}

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

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,6 @@
11
{
22
"calcite": {
33
"logical": "LogicalSystemLimit(fetch=[10000], type=[QUERY_SIZE_LIMIT])\n LogicalSort(offset=[10], fetch=[10])\n LogicalSort(fetch=[100])\n LogicalProject(count()=[$1], state=[$0])\n LogicalFilter(condition=[IS NOT NULL($0)])\n LogicalAggregate(group=[{0}], count()=[COUNT()])\n LogicalProject(state=[$7])\n CalciteLogicalIndexScan(table=[[OpenSearch, opensearch-sql_test_index_account]])\n",
4-
"physical": "EnumerableLimit(fetch=[10000])\n EnumerableCalc(expr#0..1=[{inputs}], count()=[$t1], state=[$t0])\n EnumerableLimit(offset=[10], fetch=[10])\n CalciteEnumerableIndexScan(table=[[OpenSearch, opensearch-sql_test_index_account]], PushDownContext=[[FILTER->IS NOT NULL($7), AGGREGATION->rel#:LogicalAggregate.NONE.[](input=RelSubset#,group={0},count()=COUNT()), LIMIT->100, LIMIT->[10 from 10]], OpenSearchRequestBuilder(sourceBuilder={\"from\":0,\"size\":0,\"timeout\":\"1m\",\"query\":{\"exists\":{\"field\":\"state\",\"boost\":1.0}},\"sort\":[],\"aggregations\":{\"state\":{\"terms\":{\"field\":\"state.keyword\",\"size\":20,\"min_doc_count\":1,\"shard_min_doc_count\":0,\"show_term_doc_count_error\":false,\"order\":{\"_key\":\"asc\"}},\"aggregations\":{\"count()\":{\"value_count\":{\"field\":\"_index\"}}}}}}, requestedTotalSize=2147483647, pageSize=null, startFrom=0)])\n"
4+
"physical": "EnumerableLimit(fetch=[10000])\n EnumerableCalc(expr#0..1=[{inputs}], count()=[$t1], state=[$t0])\n EnumerableLimit(offset=[10], fetch=[10])\n CalciteEnumerableIndexScan(table=[[OpenSearch, opensearch-sql_test_index_account]], PushDownContext=[[FILTER->IS NOT NULL($7), AGGREGATION->rel#:LogicalAggregate.NONE.[](input=RelSubset#,group={0},count()=COUNT()), LIMIT->100, LIMIT->[10 from 10]], OpenSearchRequestBuilder(sourceBuilder={\"from\":0,\"size\":0,\"timeout\":\"1m\",\"query\":{\"exists\":{\"field\":\"state\",\"boost\":1.0}},\"sort\":[],\"aggregations\":{\"composite_buckets\":{\"composite\":{\"size\":20,\"sources\":[{\"state\":{\"terms\":{\"field\":\"state.keyword\",\"missing_bucket\":false,\"order\":\"asc\"}}}]},\"aggregations\":{\"count()\":{\"value_count\":{\"field\":\"_index\"}}}}}}, requestedTotalSize=2147483647, pageSize=null, startFrom=0)])\n"
55
}
66
}

0 commit comments

Comments
 (0)