Works I think :D

Author: trzysiek

quesma/model/bucket_aggregations/auto_date_histogram.go

@@ -0,0 +1,54 @@
+// Copyright Quesma, licensed under the Elastic License 2.0.
+// SPDX-License-Identifier: Elastic-2.0
+package bucket_aggregations
+
+import (
+	"context"
+	"fmt"
+	"quesma/logger"
+	"quesma/model"
+	"time"
+)
+
+type AutoDateHistogram struct {
+	ctx       context.Context
+	field     model.Expr // name of the field, e.g. timestamp
+	bucketsNr int
+	key       int64
+}
+
+func NewAutoDateHistogram(ctx context.Context, field model.Expr, bucketsNr int) *AutoDateHistogram {
+	return &AutoDateHistogram{ctx: ctx, field: field, bucketsNr: bucketsNr}
+}
+
+func (query *AutoDateHistogram) AggregationType() model.AggregationType {
+	return model.BucketAggregation
+}
+
+func (query *AutoDateHistogram) TranslateSqlResponseToJson(rows []model.QueryResultRow) model.JsonMap {
+	fmt.Println(rows)
+	if len(rows) == 0 {
+		logger.WarnWithCtx(query.ctx).Msgf("no rows returned for %s", query.String())
+		return make(model.JsonMap, 0)
+	}
+	return model.JsonMap{
+		"buckets": []model.JsonMap{{
+			"key":           query.key,
+			"key_as_string": time.UnixMilli(query.key).Format("2006-01-02T15:04:05.000-07:00"),
+			"doc_count":     rows[0].LastColValue(),
+		}},
+		"interval": "100y",
+	}
+}
+
+func (query *AutoDateHistogram) String() string {
+	return fmt.Sprintf("auto_date_histogram(field: %v, bucketsNr: %d)", model.AsString(query.field), query.bucketsNr)
+}
+
+func (query *AutoDateHistogram) GetField() model.Expr {
+	return query.field
+}
+
+func (query *AutoDateHistogram) SetKey(key int64) {
+	query.key = key
+}

quesma/model/bucket_aggregations/date_histogram.go

@@ -85,7 +85,8 @@ func (query *DateHistogram) TranslateSqlResponseToJson(rows []model.QueryResultR
 	// Implement default when query.minDocCount == DefaultMinDocCount, we need to return
 	// all buckets between the first bucket that matches documents and the last one.
 
-	if query.minDocCount == 0 {
+	fmt.Println("query.minDocCount", query.minDocCount, "query.ebmin", query.ebmin)
+	if query.minDocCount == 0 || query.ebmin != 0 {
 		rows = query.NewRowsTransformer().Transform(query.ctx, rows)
 	}
 
@@ -99,7 +100,7 @@ func (query *DateHistogram) TranslateSqlResponseToJson(rows []model.QueryResultR
 		responseKey := query.calculateResponseKey(originalKey)
 
 		response = append(response, model.JsonMap{
-			//OriginalKeyName: originalKey,
+			OriginalKeyName: originalKey,
 			"key":           responseKey,
 			"doc_count":     docCount,
 			"key_as_string": query.calculateKeyAsString(responseKey),
@@ -119,6 +120,7 @@ func (query *DateHistogram) String() string {
 // only intervals <= days are needed
 func (query *DateHistogram) intervalAsDuration() time.Duration {
 	var intervalInHoursOrLess string
+	//fmt.Println("query.interval", query.interval)
 	if strings.HasSuffix(query.interval, "d") {
 		// time.ParseDuration doesn't accept > hours, we need to convert days to hours
 		daysNr, err := strconv.Atoi(strings.TrimSuffix(query.interval, "d"))
@@ -130,7 +132,9 @@ func (query *DateHistogram) intervalAsDuration() time.Duration {
 	} else {
 		intervalInHoursOrLess = query.interval
 	}
+	//fmt.Println("intervalInHoursOrLess", intervalInHoursOrLess)
 	duration, _ := time.ParseDuration(intervalInHoursOrLess)
+	//fmt.Println("duration", duration)
 	return duration
 }
 
@@ -228,7 +232,7 @@ func (query *DateHistogram) calculateResponseKey(originalKey int64) int64 {
 }
 
 func (query *DateHistogram) calculateKeyAsString(key int64) string {
-	return time.UnixMilli(key).In(query.wantedTimezone).Format("2006-01-02T15:04:05.000-07:00")
+	return time.UnixMilli(key).In(query.wantedTimezone).Format("2006/01/02 15:04:05")
 }
 
 func (query *DateHistogram) OriginalKeyToKeyAsString(originalKey any) string {
@@ -241,16 +245,14 @@ func (query *DateHistogram) SetMinDocCountToZero() {
 }
 
 func (query *DateHistogram) NewRowsTransformer() model.QueryRowsTransformer {
-	differenceBetweenTwoNextKeys := int64(1)
-	if query.intervalType == DateHistogramCalendarInterval {
-		duration, err := kibana.ParseInterval(query.interval)
-		if err == nil {
-			differenceBetweenTwoNextKeys = duration.Milliseconds()
-		} else {
-			logger.ErrorWithCtx(query.ctx).Err(err)
-			differenceBetweenTwoNextKeys = 0
-		}
+	duration, err := kibana.ParseInterval(query.interval)
+	var differenceBetweenTwoNextKeys int64
+	if err == nil {
+		differenceBetweenTwoNextKeys = duration.Milliseconds()
+	} else {
+		logger.ErrorWithCtx(query.ctx).Err(err)
 	}
+	fmt.Println("differenceBetweenTwoNextKeys", differenceBetweenTwoNextKeys)
 	return &DateHistogramRowsTransformer{MinDocCount: query.minDocCount, differenceBetweenTwoNextKeys: differenceBetweenTwoNextKeys, EmptyValue: 0, ebmin: query.ebmin, ebmax: query.ebmax}
 }
 
@@ -267,7 +269,7 @@ type DateHistogramRowsTransformer struct {
 // if MinDocCount == 0, and we have buckets e.g. [key, value1], [key+10, value2], we need to insert [key+1, 0], [key+2, 0]...
 // CAUTION: a different kind of postprocessing is needed for MinDocCount > 1, but I haven't seen any query with that yet, so not implementing it now.
 func (qt *DateHistogramRowsTransformer) Transform(ctx context.Context, rowsFromDB []model.QueryResultRow) []model.QueryResultRow {
-	if qt.MinDocCount != 0 || qt.differenceBetweenTwoNextKeys == 0 || len(rowsFromDB) < 1 {
+	if qt.MinDocCount != 0 || qt.differenceBetweenTwoNextKeys == 0 {
 		// we only add empty rows, when
 		// a) MinDocCount == 0
 		// b) we have valid differenceBetweenTwoNextKeys (>0)
@@ -281,7 +283,9 @@ func (qt *DateHistogramRowsTransformer) Transform(ctx context.Context, rowsFromD
 
 	emptyRowsAdded := 0
 	postprocessedRows := make([]model.QueryResultRow, 0, len(rowsFromDB))
-	postprocessedRows = append(postprocessedRows, rowsFromDB[0])
+	if len(rowsFromDB) > 0 {
+		postprocessedRows = append(postprocessedRows, rowsFromDB[0])
+	}
 	for i := 1; i < len(rowsFromDB); i++ {
 		if len(rowsFromDB[i-1].Cols) < 2 || len(rowsFromDB[i].Cols) < 2 {
 			logger.ErrorWithCtx(ctx).Msgf(
@@ -301,17 +305,31 @@ func (qt *DateHistogramRowsTransformer) Transform(ctx context.Context, rowsFromD
 		}
 		postprocessedRows = append(postprocessedRows, rowsFromDB[i])
 	}
-	/*
-		fmt.Println("postprocessedRows 1", postprocessedRows, qt.getKey(postprocessedRows[0])-qt.differenceBetweenTwoNextKeys, qt.ebmin, (qt.getKey(postprocessedRows[0])-qt.differenceBetweenTwoNextKeys)*qt.differenceBetweenTwoNextKeys)
-
-		for maybePreKey, i := qt.getKey(postprocessedRows[0])-qt.differenceBetweenTwoNextKeys, 0; i < 96; maybePreKey*qt.differenceBetweenTwoNextKeys >= qt.ebminmaybePreKey, i = maybePreKey-qt.differenceBetweenTwoNextKeys, i+1 {
-			preRow := postprocessedRows[0].Copy()
-			preRow.Cols[len(preRow.Cols)-2].Value = maybePreKey
-			preRow.Cols[len(preRow.Cols)-1].Value = qt.EmptyValue
-			postprocessedRows = append([]model.QueryResultRow{preRow}, postprocessedRows...)
-			emptyRowsAdded++
-		}
-	*/
+
+	//fmt.Println("postprocessedRows 1", postprocessedRows, qt.getKey(postprocessedRows[0])*qt.differenceBetweenTwoNextKeys-qt.differenceBetweenTwoNextKeys, qt.ebmin, qt.differenceBetweenTwoNextKeys)
+	fmt.Println("pre: ", len(postprocessedRows), emptyRowsAdded)
+	if qt.ebmin == 0 {
+		return postprocessedRows
+	}
+
+	if len(postprocessedRows) == 0 {
+		postprocessedRows = append(postprocessedRows, model.QueryResultRow{
+			Cols: []model.QueryResultCol{
+				{Value: (qt.ebmin+1000*60*60*2)/qt.differenceBetweenTwoNextKeys - 1},
+				{Value: qt.EmptyValue},
+			},
+		})
+	}
+	// gk*d-d = d(gk - 1)
+	// gk*d-2d = d (gk-2) = d(gk-1) - d
+	for maybePreKey := (qt.ebmin + 1000*60*60*2) / qt.differenceBetweenTwoNextKeys; maybePreKey*qt.differenceBetweenTwoNextKeys < qt.ebmax+1000*60*60*2; maybePreKey++ {
+		preRow := postprocessedRows[0].Copy()
+		preRow.Cols[len(preRow.Cols)-2].Value = maybePreKey
+		preRow.Cols[len(preRow.Cols)-1].Value = qt.EmptyValue
+		postprocessedRows = append(postprocessedRows, preRow)
+		emptyRowsAdded++
+	}
+	fmt.Println("post:", len(postprocessedRows), emptyRowsAdded)
 	return postprocessedRows
 }
 

quesma/model/bucket_aggregations/filter.go

@@ -4,6 +4,7 @@ package bucket_aggregations
 
 import (
 	"context"
+	"fmt"
 	"quesma/logger"
 	"quesma/model"
 )
@@ -18,14 +19,15 @@ func NewFilterAgg(ctx context.Context, whereClause model.Expr) FilterAgg {
 }
 
 func (query FilterAgg) AggregationType() model.AggregationType {
-	return model.BucketAggregation
+	return model.MetricsAggregation
 }
 
 func (query FilterAgg) TranslateSqlResponseToJson(rows []model.QueryResultRow) model.JsonMap {
 	if len(rows) == 0 {
 		logger.WarnWithCtx(query.ctx).Msg("no rows returned for filter aggregation")
 		return make(model.JsonMap, 0)
 	}
+	fmt.Println("filter_agg", query.String(), rows)
 	return model.JsonMap{"doc_count": rows[0].Cols[0].Value}
 }
 

quesma/model/bucket_aggregations/filters.go

@@ -28,6 +28,9 @@ type Filter struct {
 }
 
 func NewFilter(name string, sql model.SimpleQuery) Filter {
+	if sql.WhereClause == nil {
+		sql.WhereClause = model.NewLiteral("True")
+	}
 	return Filter{Name: name, Sql: sql}
 }
 
@@ -36,7 +39,7 @@ func (query Filters) AggregationType() model.AggregationType {
 }
 
 func (query Filters) TranslateSqlResponseToJson(rows []model.QueryResultRow) model.JsonMap {
-	var value any = nil
+	var value any = 0.0
 	if len(rows) > 0 {
 		if len(rows[0].Cols) > 0 {
 			value = rows[0].Cols[len(rows[0].Cols)-1].Value

quesma/model/metrics_aggregations/count.go

@@ -4,6 +4,7 @@ package metrics_aggregations
 
 import (
 	"context"
+	"fmt"
 	"quesma/logger"
 	"quesma/model"
 )
@@ -28,7 +29,11 @@ func (query Count) TranslateSqlResponseToJson(rows []model.QueryResultRow) model
 	if len(rows) > 1 {
 		logger.WarnWithCtx(query.ctx).Msg("More than one row returned for count aggregation")
 	}
-	return model.JsonMap{"doc_count": rows[0].Cols[0]}
+	if len(rows[0].Cols) == 0 {
+		return model.JsonMap{"doc_count": -1}
+	}
+	fmt.Println("COUNT", rows)
+	return model.JsonMap{"doc_count": rows[0].Cols[0].Value}
 }
 
 func (query Count) String() string {

quesma/model/pipeline_aggregations/bucket_script.go

@@ -5,55 +5,85 @@ package pipeline_aggregations
 import (
 	"context"
 	"fmt"
-	"quesma/logger"
 	"quesma/model"
 	"quesma/util"
+	"strings"
 )
 
 type BucketScript struct {
+	script string
 	*PipelineAggregation
 }
 
-func NewBucketScript(ctx context.Context) BucketScript {
-	return BucketScript{PipelineAggregation: newPipelineAggregation(ctx, "_count")}
+func NewBucketScript(ctx context.Context, script string) BucketScript {
+	return BucketScript{script: script, PipelineAggregation: newPipelineAggregation(ctx, "_count")}
 }
 
 func (query BucketScript) AggregationType() model.AggregationType {
-	return model.PipelineBucketAggregation // not sure
+	return model.PipelineMetricsAggregation // not sure
 }
 
 func (query BucketScript) TranslateSqlResponseToJson(rows []model.QueryResultRow) model.JsonMap {
-	fmt.Println("bucket_script", rows)
+	//fmt.Println("bucket_script", query.String(), rows)
 	if len(rows) == 0 {
-		logger.WarnWithCtx(query.ctx).Msg("no rows returned for bucket script aggregation")
+		//logger.WarnWithCtx(query.ctx).Msg("no rows returned for bucket script aggregation")
 	}
-	for _, row := range rows {
-		return model.JsonMap{"value": int64(util.ExtractInt64(row.LastColValue()))}
+	const defaultValue = 0.
+	switch query.script {
+	case "params.numerator != null && params.denominator != null && params.denominator != 0 ? params.numerator / params.denominator : 0":
+		numerator := query.findFilterValue("numerator", rows)
+		denominator := query.findFilterValue("denominator", rows)
+		if denominator == 0 {
+			return model.JsonMap{"value": defaultValue}
+		}
+		return model.JsonMap{"value": numerator / denominator}
+	default:
+		for _, row := range rows {
+			return model.JsonMap{"value": util.ExtractInt64(row.LastColValue())}
+		}
 	}
-	return model.JsonMap{"value": 0.}
+
+	return model.JsonMap{"value": 0.0}
 }
 
 func (query BucketScript) CalculateResultWhenMissing(parentRows []model.QueryResultRow) []model.QueryResultRow {
-	fmt.Println("bucket_script", parentRows)
+	//fmt.Println("bucket_script", query.String(), parentRows[:max(0, len(parentRows))])
 	if len(parentRows) == 0 {
-		logger.WarnWithCtx(query.ctx).Msg("no rows returned for bucket script aggregation")
+		//logger.WarnWithCtx(query.ctx).Msg("no rows returned for bucket script aggregation")
 		return parentRows
 	}
 	resultRows := make([]model.QueryResultRow, 0, len(parentRows))
 	for _, parentRow := range parentRows {
 		resultRow := parentRow.Copy()
 		resultRow.Cols[len(resultRow.Cols)-1].Value = float64(util.ExtractInt64(parentRow.LastColValue()))
-		fmt.Printf("last col %T %v", resultRow.LastColValue(), resultRow.LastColValue())
+		//fmt.Printf("last col %T %v", resultRow.LastColValue(), resultRow.LastColValue())
 		resultRows = append(resultRows, resultRow)
 	}
 	return resultRows
 }
 
 func (query BucketScript) String() string {
-	return fmt.Sprintf("bucket script(isCount: %v, parent: %s, pathToParent: %v, parentBucketAggregation: %v)",
-		query.isCount, query.Parent, query.PathToParent, query.parentBucketAggregation)
+	return fmt.Sprintf("bucket script(isCount: %v, parent: %s, pathToParent: %v, parentBucketAggregation: %v, script: %v)",
+		query.isCount, query.Parent, query.PathToParent, query.parentBucketAggregation, query.script)
 }
 
 func (query BucketScript) PipelineAggregationType() model.PipelineAggregationType {
-	return model.PipelineParentAggregation // not sure, maybe it's sibling.
+	return model.PipelineParentAggregation // not sure, maybe it's sibling. doesnt change the result
+}
+
+func (query BucketScript) findFilterValue(filterName string, rows []model.QueryResultRow) float64 {
+	for _, row := range rows {
+		for _, col := range row.Cols {
+			//fmt.Println("col", col)
+			colName := col.ColName
+			if !strings.HasSuffix(colName, "_col_0") {
+				continue
+			}
+			colName = strings.TrimSuffix(colName, "_col_0")
+			if strings.HasSuffix(colName, "-"+filterName) {
+				return float64(util.ExtractInt64(col.Value))
+			}
+		}
+	}
+	return 0
 }

quesma/model/where_visitor.go

@@ -0,0 +1,22 @@
+package model
+
+// TODO: it's not 100% full/proper implementation, but works in the client case
+func FindTimestampLowerBound(expr Expr) (InfixExpr, bool) {
+	candidates := make([]InfixExpr, 0)
+	visitor := NewBaseVisitor()
+	visitor.OverrideVisitInfix = func(visitor *BaseExprVisitor, e InfixExpr) interface{} {
+		if e.Op == ">=" {
+			candidates = append(candidates, e)
+		} else if e.Op == "AND" {
+			e.Left.Accept(visitor)
+			e.Right.Accept(visitor)
+		}
+		return nil
+	}
+
+	expr.Accept(visitor)
+	if len(candidates) == 1 {
+		return candidates[0], true
+	}
+	return InfixExpr{}, false
+}