Add datetime aggregation fixes #144

Yury-Fridlyand · 2022-10-26T17:06:53Z

Signed-off-by: Yury-Fridlyand yuryf@bitquilltech.com

Description

Make listed aggregations work with datetime types.

Fixes

min
max
avg

Not included

var_samp
var_pop
stddev_samp
stddev_pop
sum [1]

Implementation details:

Presicion

Aggregation done with milliseconds precision - this follows OpenSearch approach. See code snippets: one, two.
We convert datetimes to millis and back on our own (see below), so we can scale up to nanoseconds. Such fix requires additional changes - few tests fail.

`ExpressionAggregationScript::execute`

A callback function, called by OpenSearch node to extract a value during processing aggregation script.
This added for backward compatibility:

opensearch-project-sql/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/script/aggregation/ExpressionAggregationScript.java

Lines 53 to 55 in 272bf67

    
           if (expr.type() instanceof OpenSearchDataType) { 
        
             return expr.value(); 
        
           }

This - to extract value. Fortunately, toEpochMilli() returns negative values for pre-Epoch timestamps, so we are able to use it for group/compare any values.

opensearch-project-sql/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/script/aggregation/ExpressionAggregationScript.java

Lines 56 to 67 in 272bf67

    
           switch ((ExprCoreType)expr.type()) { 
        
             case TIME: 
        
               // workaround for session context issue 
        
               // TODO remove once fixed 
        
               return MILLIS.between(LocalTime.MIN, expr.timeValue()); 
        
             case DATE: 
        
             case DATETIME: 
        
             case TIMESTAMP: 
        
               return expr.timestampValue().toEpochMilli(); 
        
             default: 
        
               return expr.value(); 
        
           }

OpenSearch accepts dates in Joda lib types and converts to milliseconds since Epoch. We have java datetime types, to I do conversion there.
https://github.com/opensearch-project/OpenSearch/blob/140e8d3e6c91519edc47be07b4cd053fdfac1769/server/src/main/java/org/opensearch/search/aggregations/support/values/ScriptDoubleValues.java#L123

`OpenSearchExprValueFactory` `parseTimestamp` and `constructTimestamp`

OpenSearch returns aggregated values as strings, even milliseconds since Epoch. I have to extract it, so I combined these functions together.

opensearch-project-sql/opensearch/src/main/java/org/opensearch/sql/opensearch/data/value/OpenSearchExprValueFactory.java

Lines 178 to 198 in 975317c

    
           private ExprValue parseTimestamp(Content value) { 
        
             if (value.isString()) { 
        
               // value may contain epoch millis as a string, trying to extract it 
        
               try { 
        
                 return parseTimestamp(new ObjectContent(Long.parseLong(value.stringValue()))); 
        
               } catch (NumberFormatException ignored) { /* nothing to do, try another format */ } 
        
               try { 
        
                 return new ExprTimestampValue( 
        
                     // Using OpenSearch DateFormatters for now. 
        
                     DateFormatters.from(DATE_TIME_FORMATTER.parse(value.stringValue())).toInstant()); 
        
               } catch (DateTimeParseException e) { 
        
                 throw new IllegalStateException(String.format( 
        
                     "Construct ExprTimestampValue from \"%s\" failed, unsupported date format.", 
        
                     value.stringValue()), e); 
        
               } 
        
             } 
        
             if (value.isNumber()) { 
        
               return new ExprTimestampValue(Instant.ofEpochMilli(value.longValue())); 
        
             } 
        
             return new ExprTimestampValue((Instant) value.objectValue()); 
        
           }

`ExprValueUtils`

These two methods used by avg in-memory aggregation for datetime types.

opensearch-project-sql/core/src/main/java/org/opensearch/sql/data/model/ExprValueUtils.java

Line 190 in 272bf67

public static long extractEpochMilliFromAnyDateTimeType(ExprValue value) {

opensearch-project-sql/core/src/main/java/org/opensearch/sql/data/model/ExprValueUtils.java

Line 213 in 272bf67

    
           public static ExprValue convertEpochMilliToDateTimeType(long value, ExprCoreType type) {

`AvgAggregator`

Added new field to store which type is being aggregated
AvgState was renamed to DoubleAvgState

DateTimeAvgState does the same logic, but for datetime types.

opensearch-project-sql/core/src/main/java/org/opensearch/sql/expression/aggregation/AvgAggregator.java

Lines 105 to 117 in 272bf67

    
           @Override 
        
           public ExprValue result() { 
        
             if (count == 0) { 
        
               return ExprNullValue.of(); 
        
             } 
        
             return ExprValueUtils.convertEpochMilliToDateTimeType(Math.round(total / count), dataType); 
        
           } 
        
           @Override 
        
           protected AvgState iterate(ExprValue value) { 
        
             total += ExprValueUtils.extractEpochMilliFromAnyDateTimeType(value); 
        
             return super.iterate(value); 
        
           }

Added abstract class AvgState

`AggregatorFunction`

New signature were added.

opensearch-project-sql/core/src/main/java/org/opensearch/sql/expression/aggregation/AggregatorFunction.java

Lines 69 to 76 in 272bf67

    
           .put(new FunctionSignature(functionName, Collections.singletonList(DATE)), 
        
               arguments -> new AvgAggregator(arguments, DATE)) 
        
           .put(new FunctionSignature(functionName, Collections.singletonList(DATETIME)), 
        
               arguments -> new AvgAggregator(arguments, DATETIME)) 
        
           .put(new FunctionSignature(functionName, Collections.singletonList(TIME)), 
        
               arguments -> new AvgAggregator(arguments, TIME)) 
        
           .put(new FunctionSignature(functionName, Collections.singletonList(TIMESTAMP)), 
        
               arguments -> new AvgAggregator(arguments, TIMESTAMP))

Actually, plugin was able to do push down aggregation on datetime types, but this weren't accepted.

[1] It works on MySQL, but I don't see any reason to implement this for datetime types.

Test queries

In-memory aggregation

opensearch-project-sql/integ-test/src/test/java/org/opensearch/sql/sql/AggregationIT.java

Lines 196 to 199 in 17e1ae9

    
           // Playing with 'over (PARTITION BY `datetime1`)' - `datetime1` column has the same value for all rows 
        
           // so partitioning by this column has no sense and doesn't (shouldn't) affect the results 
        
           // Aggregations with `OVER` clause are executed in memory (in SQL plugin memory), 
        
           // Aggregations without it are performed the OpenSearch node itself (pushed down to opensearch)

SELECT avg(CAST(time1 AS time)) OVER(PARTITION BY datetime1) from calcs;

(cast required to ensure that you're working with TIME).
To try other types, use

CAST(date0 AS date)            
datetime(CAST(time0 AS STRING))
CAST(time1 AS time)            
CAST(datetime0 AS timestamp)

Push Down aggregation (`metric`)

SELECT min(CAST(date0 AS date)) from calcs;

Push Down aggregation (`composite_buckets`)

SELECT CAST(datetime0 AS timestamp) FROM bank GROUP BY 1;

Issues Resolved

opensearch-project#645

Check List

New functionality includes testing.
- All tests pass, including unit test, integration test and doctest
New functionality has been documented.
- New functionality has javadoc added
- New functionality has user manual doc added
Commits are signed per the DCO using --signoff

By submitting this pull request, I confirm that my contribution is made under the terms of the Apache 2.0 license.
For more information on following Developer Certificate of Origin and signing off your commits, please check here.

Yury-Fridlyand · 2022-10-26T17:39:55Z

...va/org/opensearch/sql/opensearch/storage/script/aggregation/ExpressionAggregationScript.java

+      case TIME:
+        return MILLIS.between(LocalTime.MIN, expr.timeValue());


Workaround for sesstion context issue

please add a TODO to fix this

opensearch/src/main/java/org/opensearch/sql/opensearch/response/agg/Utils.java

acarbonetto · 2022-10-26T18:16:33Z

...rg/opensearch/sql/opensearch/storage/script/aggregation/ExpressionAggregationScriptTest.java

+        .shouldMatch(new ExprTimestampValue("1984-03-17 22:16:42")
+            .timestampValue().toEpochMilli());
+  }
+


should we add the all-nulls case here too?

I'm planning to add ITs that cover too.

Existing UT framework is very limited and can't properly cover such case. There is a test for that and it didn't detect such case:

opensearch-project-sql/opensearch/src/test/java/org/opensearch/sql/opensearch/storage/script/aggregation/ExpressionAggregationScriptTest.java

Lines 102 to 108 in 3eba57e

@Test

void can_execute_expression_with_null_field() {

assertThat()

.docValues("age", null)

.evaluate(ref("age", INTEGER))

.shouldMatch(null);

}

acarbonetto · 2022-10-27T01:38:01Z

core/src/main/java/org/opensearch/sql/expression/aggregation/AvgAggregator.java

+      case TIME:
+        state.total += MILLIS.between(LocalTime.MIN, value.timeValue());
+        break;
+      case DOUBLE:


we cannot compute AVG for SHORT, INTEGER, FLOAT, etc...?

These types are casted to DOUBLE automatically

Should the automatic casting of types to double be tested in AvgAggregatorTest? I think only INTEGER and DOUBLE are only in there for number values.

core/src/test/java/org/opensearch/sql/expression/aggregation/AvgAggregatorTest.java

docs/user/dql/aggregations.rst

acarbonetto · 2022-11-01T21:13:45Z

core/src/main/java/org/opensearch/sql/data/model/ExprValueUtils.java

+  public static long extractEpochMilliFromAnyDateTimeType(ExprValue value) {
+    switch ((ExprCoreType)value.type()) {
+      case DATE:
+        return MILLIS.between(LocalDateTime.of(1970, 1, 1, 0, 0), value.datetimeValue());


I may be wrong, but it seems like we could use the Instant.toEpochMilli for this call too. Something like:

value.datetimeValue().toInstant().toEpochMilli();

Nice catch!
Initially case DATE: fallen back with TIMESTAMP and DATETIME, conversion thru timestamp (Instant) returned wrong result, because timestamp changes TZ to UTC.
DATE -> MILLIS -> DATE -> MILLIS check failed.

Thank you. While thinking on your comment I found a mistake. Fixed in 478141e.

GumpacG · 2022-11-03T15:57:27Z

I think I may have found a bug where min of a datetime column containing nulls does not return null.

Yury-Fridlyand · 2022-11-03T16:49:48Z

I think I may have found a bug where min of a datetime column containing nulls does not return null.

Nice catch! It is fixed in #145 (not merged yet on upstream - opensearch-project#1000).

GumpacG · 2022-11-03T16:54:15Z

I think I may have found a bug where min of a datetime column containing nulls does not return null.

Nice catch! It is fixed in #145 (not merged yet on upstream - opensearch-project#1000).

I see. Thanks. LGTM!

GumpacG · 2022-11-04T21:26:54Z

I think I may have found a bug where min of a datetime column containing nulls does not return null.

Nice catch! It is fixed in #145 (not merged yet on upstream - opensearch-project#1000).

I think we should add integration tests for these cases once the PR upstream is merged.

* Add aggregator fixes and some useless unit tests. * Add `avg` aggregation on datetime types. * Rework in-memory `AVG`. Fix parsing value returned from the OpenSearch node. Signed-off-by: Yury-Fridlyand <yuryf@bitquilltech.com>

Yury-Fridlyand · 2022-11-04T23:29:10Z

I think I may have found a bug where min of a datetime column containing nulls does not return null.

Nice catch! It is fixed in #145 (not merged yet on upstream - opensearch-project#1000).

I think we should add integration tests for these cases once the PR upstream is merged.

Done! Thank you for the idea.

forestmvey

LGTM

MitchellGale

Validation looks good.

This comment was marked as spam.

Sign in to view

Yury-Fridlyand commented Oct 26, 2022

View reviewed changes

acarbonetto approved these changes Oct 26, 2022

View reviewed changes

GabeFernandez310 approved these changes Oct 26, 2022

View reviewed changes

Yury-Fridlyand marked this pull request as ready for review October 27, 2022 01:24

Yury-Fridlyand requested review from acarbonetto, forestmvey, GabeFernandez310, GumpacG, margarit-h, MaxKsyunz and MitchellGale October 27, 2022 01:24

acarbonetto reviewed Oct 27, 2022

View reviewed changes

core/src/test/java/org/opensearch/sql/expression/aggregation/AvgAggregatorTest.java Outdated Show resolved Hide resolved

acarbonetto approved these changes Oct 27, 2022

View reviewed changes

docs/user/dql/aggregations.rst Outdated Show resolved Hide resolved

GumpacG approved these changes Oct 27, 2022

View reviewed changes

forestmvey approved these changes Oct 27, 2022

View reviewed changes

Yury-Fridlyand marked this pull request as draft October 27, 2022 21:20

Yury-Fridlyand mentioned this pull request Oct 28, 2022

SQL ingestion and ops - standup opensearch-project/sql#791

Open

MaxKsyunz approved these changes Oct 28, 2022

View reviewed changes

MaxKsyunz self-requested a review October 28, 2022 18:53

Yury-Fridlyand marked this pull request as ready for review November 1, 2022 02:35

Yury-Fridlyand requested review from forestmvey, GumpacG and acarbonetto November 1, 2022 02:37

acarbonetto approved these changes Nov 1, 2022

View reviewed changes

GumpacG approved these changes Nov 3, 2022

View reviewed changes

Yury-Fridlyand force-pushed the dev-datetime-aggreg branch from db98434 to 435e940 Compare November 4, 2022 23:19

Update aggregation to support datetime types.

31a72d4

* Add aggregator fixes and some useless unit tests. * Add `avg` aggregation on datetime types. * Rework in-memory `AVG`. Fix parsing value returned from the OpenSearch node. Signed-off-by: Yury-Fridlyand <yuryf@bitquilltech.com>

Yury-Fridlyand force-pushed the dev-datetime-aggreg branch from 435e940 to 31a72d4 Compare November 4, 2022 23:28

MaxKsyunz approved these changes Nov 9, 2022

View reviewed changes

Yury-Fridlyand merged commit 4a9dfda into integ-datetime-aggreg Nov 9, 2022

Yury-Fridlyand deleted the dev-datetime-aggreg branch November 9, 2022 16:59

Yury-Fridlyand mentioned this pull request Nov 9, 2022

Datetime aggregation fixes. opensearch-project/sql#1061

Merged

6 tasks

forestmvey reviewed Nov 9, 2022

View reviewed changes

MitchellGale reviewed Nov 10, 2022

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Add datetime aggregation fixes #144

Add datetime aggregation fixes #144

Yury-Fridlyand commented Oct 26, 2022 •

edited

Loading

This comment was marked as spam.

Yury-Fridlyand Oct 26, 2022

acarbonetto Oct 27, 2022

acarbonetto Oct 26, 2022

Yury-Fridlyand Oct 26, 2022

Yury-Fridlyand Oct 26, 2022

acarbonetto Oct 27, 2022

Yury-Fridlyand Oct 27, 2022

GumpacG Oct 27, 2022

acarbonetto Nov 1, 2022

Yury-Fridlyand Nov 1, 2022

Yury-Fridlyand Nov 2, 2022

GumpacG commented Nov 3, 2022

Yury-Fridlyand commented Nov 3, 2022 •

edited

Loading

GumpacG commented Nov 3, 2022

GumpacG commented Nov 4, 2022

Yury-Fridlyand commented Nov 4, 2022

forestmvey left a comment

MitchellGale left a comment

	if (expr.type() instanceof OpenSearchDataType) {
	return expr.value();
	}

	switch ((ExprCoreType)expr.type()) {
	case TIME:
	// workaround for session context issue
	// TODO remove once fixed
	return MILLIS.between(LocalTime.MIN, expr.timeValue());
	case DATE:
	case DATETIME:
	case TIMESTAMP:
	return expr.timestampValue().toEpochMilli();
	default:
	return expr.value();
	}

	private ExprValue parseTimestamp(Content value) {
	if (value.isString()) {
	// value may contain epoch millis as a string, trying to extract it
	try {
	return parseTimestamp(new ObjectContent(Long.parseLong(value.stringValue())));
	} catch (NumberFormatException ignored) { /* nothing to do, try another format */ }
	try {
	return new ExprTimestampValue(
	// Using OpenSearch DateFormatters for now.
	DateFormatters.from(DATE_TIME_FORMATTER.parse(value.stringValue())).toInstant());
	} catch (DateTimeParseException e) {
	throw new IllegalStateException(String.format(
	"Construct ExprTimestampValue from \"%s\" failed, unsupported date format.",
	value.stringValue()), e);
	}
	}
	if (value.isNumber()) {
	return new ExprTimestampValue(Instant.ofEpochMilli(value.longValue()));
	}
	return new ExprTimestampValue((Instant) value.objectValue());
	}

	@Override
	public ExprValue result() {
	if (count == 0) {
	return ExprNullValue.of();
	}
	return ExprValueUtils.convertEpochMilliToDateTimeType(Math.round(total / count), dataType);
	}

	@Override
	protected AvgState iterate(ExprValue value) {
	total += ExprValueUtils.extractEpochMilliFromAnyDateTimeType(value);
	return super.iterate(value);
	}

	.put(new FunctionSignature(functionName, Collections.singletonList(DATE)),
	arguments -> new AvgAggregator(arguments, DATE))
	.put(new FunctionSignature(functionName, Collections.singletonList(DATETIME)),
	arguments -> new AvgAggregator(arguments, DATETIME))
	.put(new FunctionSignature(functionName, Collections.singletonList(TIME)),
	arguments -> new AvgAggregator(arguments, TIME))
	.put(new FunctionSignature(functionName, Collections.singletonList(TIMESTAMP)),
	arguments -> new AvgAggregator(arguments, TIMESTAMP))

	// Playing with 'over (PARTITION BY `datetime1`)' - `datetime1` column has the same value for all rows
	// so partitioning by this column has no sense and doesn't (shouldn't) affect the results
	// Aggregations with `OVER` clause are executed in memory (in SQL plugin memory),
	// Aggregations without it are performed the OpenSearch node itself (pushed down to opensearch)

	@Test
	void can_execute_expression_with_null_field() {
	assertThat()
	.docValues("age", null)
	.evaluate(ref("age", INTEGER))
	.shouldMatch(null);
	}

Add datetime aggregation fixes #144

Add datetime aggregation fixes #144

Conversation

Yury-Fridlyand commented Oct 26, 2022 • edited Loading

Description

Fixes

Not included

Implementation details:

Presicion

ExpressionAggregationScript::execute

OpenSearchExprValueFactory parseTimestamp and constructTimestamp

ExprValueUtils

AvgAggregator

AggregatorFunction

Test queries

In-memory aggregation

Push Down aggregation (metric)

Push Down aggregation (composite_buckets)

Issues Resolved

Check List

This comment was marked as spam.

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

GumpacG commented Nov 3, 2022

Yury-Fridlyand commented Nov 3, 2022 • edited Loading

GumpacG commented Nov 3, 2022

GumpacG commented Nov 4, 2022

Yury-Fridlyand commented Nov 4, 2022

forestmvey left a comment

Choose a reason for hiding this comment

MitchellGale left a comment

Choose a reason for hiding this comment

Yury-Fridlyand commented Oct 26, 2022 •

edited

Loading

`ExpressionAggregationScript::execute`

`OpenSearchExprValueFactory` `parseTimestamp` and `constructTimestamp`

`ExprValueUtils`

`AvgAggregator`

`AggregatorFunction`

Push Down aggregation (`metric`)

Push Down aggregation (`composite_buckets`)

Yury-Fridlyand commented Nov 3, 2022 •

edited

Loading