Skip to content

Commit 8068319

Browse files
LantaoJinLantaoJin
committed
Date/Time based Span aggregation should always not present null bucket (opensearch-project#4327)
(cherry picked from commit 2ab5002)
1 parent ce92b33 commit 8068319

33 files changed

Lines changed: 560 additions & 238 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
if (unit == null) return NONE;

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

Lines changed: 51 additions & 16 deletions
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;
@@ -893,27 +895,42 @@ public RelNode visitAggregation(Aggregation node, CalcitePlanContext context) {
893895
// The span column is always the first column in result whatever
894896
// the order of span in query is first or last one
895897
UnresolvedExpression span = node.getSpan();
896-
if (!Objects.isNull(span)) {
898+
if (Objects.nonNull(span)) {
897899
groupExprList.add(span);
900+
List<RexNode> timeSpanFilters =
901+
getTimeSpanField(span).stream()
902+
.map(f -> rexVisitor.analyze(f, context))
903+
.map(context.relBuilder::isNotNull)
904+
.collect(Collectors.toList());
905+
if (!timeSpanFilters.isEmpty()) {
906+
// add isNotNull filter before aggregation for time span
907+
context.relBuilder.filter(timeSpanFilters);
908+
}
898909
}
899910
groupExprList.addAll(node.getGroupExprList());
900-
Pair<List<RexNode>, List<AggCall>> aggregationAttributes =
901-
aggregateWithTrimming(groupExprList, aggExprList, context);
902-
// Add group by columns
903-
List<RexNode> aliasedGroupByList =
904-
aggregationAttributes.getLeft().stream()
905-
.map(this::extractAliasLiteral)
906-
.flatMap(Optional::stream)
907-
.map(ref -> ((RexLiteral) ref).getValueAs(String.class))
908-
.map(context.relBuilder::field)
909-
.map(f -> (RexNode) f)
910-
.collect(Collectors.toList());
911911

912912
// add stats hint to LogicalAggregation
913913
Argument.ArgumentMap statsArgs = Argument.ArgumentMap.of(node.getArgExprList());
914914
Boolean bucketNullable =
915915
(Boolean) statsArgs.getOrDefault(Argument.BUCKET_NULLABLE, Literal.TRUE).getValue();
916-
if (!bucketNullable && !aliasedGroupByList.isEmpty()) {
916+
boolean toAddHintsOnAggregate = false;
917+
if (!bucketNullable
918+
&& !groupExprList.isEmpty()
919+
&& !(groupExprList.size() == 1 && getTimeSpanField(span).isPresent())) {
920+
toAddHintsOnAggregate = true;
921+
// add isNotNull filter before aggregation for non-nullable buckets
922+
List<RexNode> groupByList =
923+
groupExprList.stream().map(expr -> rexVisitor.analyze(expr, context)).collect(Collectors.toList());
924+
context.relBuilder.filter(
925+
PlanUtils.getSelectColumns(groupByList).stream()
926+
.map(context.relBuilder::field)
927+
.map(context.relBuilder::isNotNull)
928+
.collect(Collectors.toList()));
929+
}
930+
931+
Pair<List<RexNode>, List<AggCall>> aggregationAttributes =
932+
aggregateWithTrimming(groupExprList, aggExprList, context);
933+
if (toAddHintsOnAggregate) {
917934
final RelHint statHits =
918935
RelHint.builder("stats_args").hintOption(Argument.BUCKET_NULLABLE, "false").build();
919936
assert context.relBuilder.peek() instanceof LogicalAggregate
@@ -930,9 +947,6 @@ public RelNode visitAggregation(Aggregation node, CalcitePlanContext context) {
930947
return rel instanceof LogicalAggregate;
931948
})
932949
.build());
933-
context.relBuilder.filter(
934-
aliasedGroupByList.stream().map(context.relBuilder::isNotNull)
935-
.collect(Collectors.toList()));
936950
}
937951

938952
// schema reordering
@@ -946,12 +960,33 @@ public RelNode visitAggregation(Aggregation node, CalcitePlanContext context) {
946960
List<RexNode> aggRexList =
947961
outputFields.subList(numOfOutputFields - numOfAggList, numOfOutputFields);
948962
reordered.addAll(aggRexList);
963+
// Add group by columns
964+
List<RexNode> aliasedGroupByList =
965+
aggregationAttributes.getLeft().stream()
966+
.map(this::extractAliasLiteral)
967+
.flatMap(Optional::stream)
968+
.map(ref -> ((RexLiteral) ref).getValueAs(String.class))
969+
.map(context.relBuilder::field)
970+
.map(f -> (RexNode) f)
971+
.collect(Collectors.toList());
949972
reordered.addAll(aliasedGroupByList);
950973
context.relBuilder.project(reordered);
951974

952975
return context.relBuilder.peek();
953976
}
954977

978+
private Optional<UnresolvedExpression> getTimeSpanField(UnresolvedExpression expr) {
979+
if (Objects.isNull(expr)) return Optional.empty();
980+
if (expr instanceof Span)
981+
if (SpanUnit.isTimeUnit(((Span)expr).getUnit())) {
982+
return Optional.of(((Span)expr).getField());
983+
}
984+
if (expr instanceof Alias) {
985+
return getTimeSpanField(((Alias)expr).getDelegated());
986+
}
987+
return Optional.empty();
988+
}
989+
955990
/** extract the RexLiteral of Alias from a node */
956991
private Optional<RexLiteral> extractAliasLiteral(RexNode node) {
957992
if (node == null) {

‎docs/user/ppl/cmd/stats.rst‎

Lines changed: 56 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -59,8 +59,7 @@ stats [bucket_nullable=bool] <aggregation>... [by-clause]
5959
* span-expression: optional, at most one.
6060

6161
* Syntax: span(field_expr, interval_expr)
62-
* Description: The unit of the interval expression is the natural unit by default. If the field is a date and time type field, and the interval is in date/time units, you will need to specify the unit in the interval expression. For example, to split the field ``age`` into buckets by 10 years, it looks like ``span(age, 10)``. And here is another example of time span, the span to split a ``timestamp`` field into hourly intervals, it looks like ``span(timestamp, 1h)``.
63-
62+
* Description: The unit of the interval expression is the natural unit by default. **If the field is a date/time type field, the aggregation results always ignore null bucket**. And the interval is in date/time units, you will need to specify the unit in the interval expression. For example, to split the field ``age`` into buckets by 10 years, it looks like ``span(age, 10)``. And here is another example of time span, the span to split a ``timestamp`` field into hourly intervals, it looks like ``span(timestamp, 1h)``.
6463
* Available time unit:
6564

6665
+----------------------------+
@@ -922,3 +921,58 @@ PPL query::
922921
|-------------------------------------|
923922
| ["Amber","Dale","Hattie","Nanette"] |
924923
+-------------------------------------+
924+
925+
926+
Example 17: Span on date/time field always ignore null bucket
927+
=============================================================
928+
929+
Index example data:
930+
931+
+-------+--------+------------+
932+
| Name | DEPTNO | birthday |
933+
+=======+========+============+
934+
| Alice | 1 | 2024-04-21 |
935+
+-------+--------+------------+
936+
| Bob | 2 | 2025-08-21 |
937+
+-------+--------+------------+
938+
| Jeff | null | 2025-04-22 |
939+
+-------+--------+------------+
940+
| Adam | 2 | null |
941+
+-------+--------+------------+
942+
943+
PPL query::
944+
945+
PPL> source=example | stats count() as cnt by span(birthday, 1y) as year;
946+
fetched rows / total rows = 3/3
947+
+-----+------------+
948+
| cnt | year |
949+
|-----+------------|
950+
| 1 | 2024-01-01 |
951+
| 2 | 2025-01-01 |
952+
+-----+------------+
953+
954+
955+
PPL query::
956+
957+
PPL> source=example | stats count() as cnt by span(birthday, 1y) as year, DEPTNO;
958+
fetched rows / total rows = 3/3
959+
+-----+------------+--------+
960+
| cnt | year | DEPTNO |
961+
|-----+------------+--------|
962+
| 1 | 2024-01-01 | 1 |
963+
| 1 | 2025-01-01 | 2 |
964+
| 1 | 2025-01-01 | null |
965+
+-----+------------+--------+
966+
967+
968+
PPL query::
969+
970+
PPL> source=example | stats bucket_nullable=false count() as cnt by span(birthday, 1y) as year, DEPTNO;
971+
fetched rows / total rows = 3/3
972+
+-----+------------+--------+
973+
| cnt | year | DEPTNO |
974+
|-----+------------+--------|
975+
| 1 | 2024-01-01 | 1 |
976+
| 1 | 2025-01-01 | 2 |
977+
+-----+------------+--------+
978+

‎integ-test/build.gradle‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -446,7 +446,7 @@ integTest {
446446

447447
dependsOn ':opensearch-sql-plugin:bundlePlugin'
448448
if(getOSFamilyType() != "windows") {
449-
dependsOn startPrometheus
449+
// dependsOn startPrometheus
450450
finalizedBy stopPrometheus
451451
}
452452

‎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/calcite/remote/CalcitePPLAggregationIT.java‎

Lines changed: 9 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -518,15 +518,20 @@ public void testCountByNullableTimeSpan() throws IOException {
518518
JSONObject actual =
519519
executeQuery(
520520
String.format(
521-
"source=%s | head 5 | stats count(datetime0), count(datetime1) by span(datetime1,"
522-
+ " 15 minute) as datetime_span",
521+
"source=%s | head 5 | stats count(datetime0), count(datetime1) by span(time1,"
522+
+ " 15 minute) as time_span",
523523
TEST_INDEX_CALCS));
524524
verifySchema(
525525
actual,
526-
schema("datetime_span", "timestamp"),
526+
schema("time_span", "time"),
527527
schema("count(datetime0)", "bigint"),
528528
schema("count(datetime1)", "bigint"));
529-
verifyDataRows(actual, rows(5, 0, null));
529+
verifyDataRows(
530+
actual,
531+
rows(1, 0, "19:30:00"),
532+
rows(1, 0, "02:00:00"),
533+
rows(1, 0, "09:30:00"),
534+
rows(1, 0, "22:45:00"));
530535
}
531536

532537
@Test

‎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
@@ -883,6 +883,11 @@ public enum Index {
883883
"time_data",
884884
getMappingFile("time_test_data_index_mapping.json"),
885885
"src/test/resources/time_test_data.json"),
886+
TIME_TEST_DATA_WITH_NULL(
887+
TestsConstants.TEST_INDEX_TIME_DATE_NULL,
888+
"time_data_with_null",
889+
getMappingFile("time_test_data_index_mapping.json"),
890+
"src/test/resources/time_test_data_with_null.json"),
886891
EVENTS(
887892
"events",
888893
"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
@@ -82,6 +82,7 @@ public class TestsConstants {
8282
public static final String TEST_INDEX_HDFS_LOGS = TEST_INDEX + "_hdfs_logs";
8383
public static final String TEST_INDEX_LOGS = TEST_INDEX + "_logs";
8484
public static final String TEST_INDEX_OTEL_LOGS = TEST_INDEX + "_otel_logs";
85+
public static final String TEST_INDEX_TIME_DATE_NULL = TEST_INDEX + "_time_date_null";
8586

8687
public static final String DATE_FORMAT = "yyyy-MM-dd'T'HH:mm:ss.SSS'Z'";
8788
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 & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -492,7 +492,6 @@ public void testStatsBySpan() throws IOException {
492492

493493
@Test
494494
public void testStatsBySpanNonBucketNullable() throws IOException {
495-
// TODO isNotNull(Span) pushdown to script, can be optimized to exist()
496495
String expected = loadExpectedPlan("explain_stats_by_span_non_bucket_nullable.json");
497496
assertJsonEqualsIgnoreId(
498497
expected,
@@ -515,6 +514,14 @@ public void testStatsByTimeSpan() throws IOException {
515514
expected,
516515
explainQueryToString(
517516
String.format("source=%s | stats count() by span(birthdate,1M)", TEST_INDEX_BANK)));
517+
518+
// bucket_nullable doesn't impact by-span-time
519+
assertJsonEqualsIgnoreId(
520+
expected,
521+
explainQueryToString(
522+
String.format(
523+
"source=%s | stats bucket_nullable=false count() by span(birthdate,1M)",
524+
TEST_INDEX_BANK)));
518525
}
519526

520527
@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
@@ -741,4 +743,25 @@ public void testDisableLegacyPreferred() throws IOException {
741743
rows(null, 36));
742744
});
743745
}
746+
747+
@Test
748+
public void testStatsBySpanTimeWithNullBucket() throws IOException {
749+
JSONObject response =
750+
executeQuery(
751+
String.format(
752+
"source=%s | stats percentile(value, 50) as p50 by span(@timestamp, 12h) as"
753+
+ " half_day",
754+
TEST_INDEX_TIME_DATE_NULL));
755+
verifySchema(response, schema("p50", null, "int"), schema("half_day", null, "timestamp"));
756+
verifyDataRows(
757+
response,
758+
rows(8407, "2025-07-28 00:00:00"),
759+
rows(7962, "2025-07-28 12:00:00"),
760+
rows(8006, "2025-07-29 00:00:00"),
761+
rows(7934, "2025-07-29 12:00:00"),
762+
rows(8089, "2025-07-30 00:00:00"),
763+
rows(8000, "2025-07-30 12:00:00"),
764+
rows(7931, "2025-07-31 00:00:00"),
765+
rows(8086, "2025-07-31 12:00:00"));
766+
}
744767
}

0 commit comments

Comments
 (0)