From 7935624549104ccad3578eae9401073c6c5ddd0f Mon Sep 17 00:00:00 2001 From: Matthew Jaffee Date: Fri, 10 Dec 2021 15:21:29 -0600 Subject: [PATCH 1/5] implement percentiles on timestamp/decimal, still needs tests --- executor.go | 33 ++++++++++++++++++++++++++++++--- field.go | 34 +++++++++++++++++++--------------- field_internal_test.go | 16 ++++++++-------- 3 files changed, 57 insertions(+), 26 deletions(-) diff --git a/executor.go b/executor.go index 226759b55..90db6c74a 100644 --- a/executor.go +++ b/executor.go @@ -537,6 +537,11 @@ func (e *executor) execute(ctx context.Context, qcx *Qcx, index string, q *pql.Q return nil, err } + if vc, ok := v.(ValCount); ok { + vc.cleanup() + v = vc + } + results = append(results, v) // Some Calls can have significant data associated with them // that gets generated during processing, such as Precomputed @@ -547,6 +552,22 @@ func (e *executor) execute(ctx context.Context, qcx *Qcx, index string, q *pql.Q return results, nil } +// cleanup removes the integer value (Val) from the ValCount if one of +// the other fields is in use. +// +// ValCounts are normally holding data which is stored as a BSI +// (integer) under the hood. Sometimes it's convenient to be able to +// compare the underlying integer values rather than their +// interpretation as decimal, timestamp, etc, so the lower level +// functions may return both integer and the interpreted value, but we +// don't want to pass that all the way back to the client, so we +// remove it here. +func (vc *ValCount) cleanup() { + if vc.Val != 0 && (vc.FloatVal != 0 || !vc.TimestampVal.IsZero() || vc.DecimalVal != nil) { + vc.Val = 0 + } +} + // preprocessQuery expands any calls that need preprocessing. func (e *executor) preprocessQuery(ctx context.Context, qcx *Qcx, index string, c *pql.Call, shards []uint64, opt *execOptions) (*pql.Call, error) { switch c.Name { @@ -1211,6 +1232,10 @@ func (e *executor) executePercentile(ctx context.Context, qcx *Qcx, index string if err != nil { return ValCount{}, errors.New("Percentile(): field required") } + field := e.Holder.Field(index, fieldName) + if field == nil { + return ValCount{}, ErrFieldNotFound + } // filter call for min & max var filterCall *pql.Call @@ -1231,7 +1256,7 @@ func (e *executor) executePercentile(ctx context.Context, qcx *Qcx, index string return ValCount{}, errors.Wrap(err, "executing Min call for Percentile") } if nthFloat == 0.0 { - return ValCount{Val: minVal.Val, Count: minVal.Count}, nil + return minVal, nil } // get max @@ -1298,11 +1323,11 @@ func (e *executor) executePercentile(ctx context.Context, qcx *Qcx, index string } else if leftCountWeighted < rightCount { min = possibleNthVal + 1 } else { - return ValCount{Val: possibleNthVal, Count: 1}, nil + return field.valCountize(possibleNthVal, 1, nil) } } - return ValCount{Val: min, Count: 1}, nil + return field.valCountize(min, 1, nil) } @@ -8057,6 +8082,8 @@ func getScaledInt(f *Field, v interface{}) (int64, error) { switch tv := v.(type) { case time.Time: value = tv.UnixNano() / TimeUnitNanos(f.options.TimeUnit) + case int64: + value = tv default: return 0, errors.Errorf("unexpected timestamp value type %T, val %v", tv, tv) } diff --git a/field.go b/field.go index 2aa865a1f..bcbc79c6e 100644 --- a/field.go +++ b/field.go @@ -1385,18 +1385,7 @@ func (f *Field) MaxForShard(tx Tx, shard uint64, filter *Row) (ValCount, error) return ValCount{}, errors.Wrap(err, "calling fragment.max") } - valCount := ValCount{Count: int64(cnt)} - - if f.Options().Type == FieldTypeDecimal { - dec := pql.NewDecimal(max+bsig.Base, bsig.Scale) - valCount.DecimalVal = &dec - } else if f.Options().Type == FieldTypeTimestamp { - valCount.TimestampVal = time.Unix(0, (max+bsig.Base)*TimeUnitNanos(f.options.TimeUnit)).UTC() - } else { - valCount.Val = max + bsig.Base - } - - return valCount, nil + return f.valCountize(max, cnt, bsig) } // MinForShard returns the minimum value which appears in this shard @@ -1431,6 +1420,23 @@ func (f *Field) MinForShard(tx Tx, shard uint64, filter *Row) (ValCount, error) return ValCount{}, errors.Wrap(err, "calling fragment.min") } + return f.valCountize(min, cnt, bsig) +} + +// valCountize takes the "raw" min value and count we get from the +// fragment and calculates the cooked values for this field +// (timestamping, decimaling, or just adding in the base). It always +// includes the int64 "Val\" value to make comparisons easier in the +// executor (at time of writing, Percentile takes advantage of this, +// but we might be able to simplify logic in other places as well). +func (f *Field) valCountize(min int64, cnt uint64, bsig *bsiGroup) (ValCount, error) { + if bsig == nil { + bsig = f.bsiGroup(f.name) + if bsig == nil { + return ValCount{}, ErrBSIGroupNotFound + } + + } valCount := ValCount{Count: int64(cnt)} if f.Options().Type == FieldTypeDecimal { @@ -1438,10 +1444,8 @@ func (f *Field) MinForShard(tx Tx, shard uint64, filter *Row) (ValCount, error) valCount.DecimalVal = &dec } else if f.Options().Type == FieldTypeTimestamp { valCount.TimestampVal = time.Unix(0, (min+bsig.Base)*TimeUnitNanos(f.options.TimeUnit)).UTC() - } else { - valCount.Val = min + bsig.Base } - + valCount.Val = min + bsig.Base return valCount, nil } diff --git a/field_internal_test.go b/field_internal_test.go index f1219f7e2..f7e0fadfd 100644 --- a/field_internal_test.go +++ b/field_internal_test.go @@ -748,29 +748,29 @@ func TestDecimalField_MinMaxForShard(t *testing.T) { name: "single", columnIDs: []uint64{1}, values: []float64{10.1}, - expMax: ValCount{DecimalVal: &pql.Decimal{Value: 10100, Scale: 3}, Count: 1}, - expMin: ValCount{DecimalVal: &pql.Decimal{Value: 10100, Scale: 3}, Count: 1}, + expMax: ValCount{Val: 10100, DecimalVal: &pql.Decimal{Value: 10100, Scale: 3}, Count: 1}, + expMin: ValCount{Val: 10100, DecimalVal: &pql.Decimal{Value: 10100, Scale: 3}, Count: 1}, }, { name: "twovals", columnIDs: []uint64{1, 2}, values: []float64{10.1, 20.2}, - expMax: ValCount{DecimalVal: &pql.Decimal{Value: 20200, Scale: 3}, Count: 1}, - expMin: ValCount{DecimalVal: &pql.Decimal{Value: 10100, Scale: 3}, Count: 1}, + expMax: ValCount{Val: 20200, DecimalVal: &pql.Decimal{Value: 20200, Scale: 3}, Count: 1}, + expMin: ValCount{Val: 10100, DecimalVal: &pql.Decimal{Value: 10100, Scale: 3}, Count: 1}, }, { name: "multiplecounts", columnIDs: []uint64{1, 2, 3, 4, 5}, values: []float64{10.1, 20.2, 10.1, 10.1, 20.2}, - expMax: ValCount{DecimalVal: &pql.Decimal{Value: 20200, Scale: 3}, Count: 2}, - expMin: ValCount{DecimalVal: &pql.Decimal{Value: 10100, Scale: 3}, Count: 3}, + expMax: ValCount{Val: 20200, DecimalVal: &pql.Decimal{Value: 20200, Scale: 3}, Count: 2}, + expMin: ValCount{Val: 10100, DecimalVal: &pql.Decimal{Value: 10100, Scale: 3}, Count: 3}, }, { name: "middlevals", columnIDs: []uint64{1, 2, 3, 4, 5, 6, 7, 8, 9, 10}, values: []float64{10.1, 20.2, 10.1, 10.1, 20.2, 11, 12, 11, 13, 11}, - expMax: ValCount{DecimalVal: &pql.Decimal{Value: 20200, Scale: 3}, Count: 2}, - expMin: ValCount{DecimalVal: &pql.Decimal{Value: 10100, Scale: 3}, Count: 3}, + expMax: ValCount{Val: 20200, DecimalVal: &pql.Decimal{Value: 20200, Scale: 3}, Count: 2}, + expMin: ValCount{Val: 10100, DecimalVal: &pql.Decimal{Value: 10100, Scale: 3}, Count: 3}, }, } { t.Run(test.name+strconv.Itoa(i), func(t *testing.T) { From 99f6a1c113f2fad7ee5e9b900e1f05d031332649 Mon Sep 17 00:00:00 2001 From: reesporte Date: Mon, 3 Jan 2022 10:45:23 -0600 Subject: [PATCH 2/5] change min to val bc it could be used for things besides mins --- field.go | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/field.go b/field.go index bcbc79c6e..a726514e7 100644 --- a/field.go +++ b/field.go @@ -1429,7 +1429,7 @@ func (f *Field) MinForShard(tx Tx, shard uint64, filter *Row) (ValCount, error) // includes the int64 "Val\" value to make comparisons easier in the // executor (at time of writing, Percentile takes advantage of this, // but we might be able to simplify logic in other places as well). -func (f *Field) valCountize(min int64, cnt uint64, bsig *bsiGroup) (ValCount, error) { +func (f *Field) valCountize(val int64, cnt uint64, bsig *bsiGroup) (ValCount, error) { if bsig == nil { bsig = f.bsiGroup(f.name) if bsig == nil { @@ -1440,12 +1440,12 @@ func (f *Field) valCountize(min int64, cnt uint64, bsig *bsiGroup) (ValCount, er valCount := ValCount{Count: int64(cnt)} if f.Options().Type == FieldTypeDecimal { - dec := pql.NewDecimal(min+bsig.Base, bsig.Scale) + dec := pql.NewDecimal(val+bsig.Base, bsig.Scale) valCount.DecimalVal = &dec } else if f.Options().Type == FieldTypeTimestamp { - valCount.TimestampVal = time.Unix(0, (min+bsig.Base)*TimeUnitNanos(f.options.TimeUnit)).UTC() + valCount.TimestampVal = time.Unix(0, (val+bsig.Base)*TimeUnitNanos(f.options.TimeUnit)).UTC() } - valCount.Val = min + bsig.Base + valCount.Val = val + bsig.Base return valCount, nil } From fa2391b948784edd51dafa1b81151effc6113c77 Mon Sep 17 00:00:00 2001 From: reesporte Date: Mon, 3 Jan 2022 11:11:27 -0600 Subject: [PATCH 3/5] explicitly test untested path of valcountize --- field_internal_test.go | 17 +++++++++++++++++ 1 file changed, 17 insertions(+) diff --git a/field_internal_test.go b/field_internal_test.go index f7e0fadfd..d9471db91 100644 --- a/field_internal_test.go +++ b/field_internal_test.go @@ -182,6 +182,23 @@ func TestBSIGroup_BaseValue(t *testing.T) { }) } +func TestField_ValCountize(t *testing.T) { + f := OpenField(t, OptFieldTypeDefault()) + defer f.Close() + // check that you get an empty val count and err + // BSIGroupNotFound on nil bsig from + // f.bsiGroup(f.name) + f.bsiGroups = []*bsiGroup{} + v, err := f.valCountize(42, 42, nil) + if !reflect.DeepEqual(v, ValCount{}) { + t.Errorf("expected %v, got %v", ValCount{}, v) + } + if err != ErrBSIGroupNotFound { + t.Errorf("expected %v, got %v", ErrBSIGroupNotFound, err) + } + +} + // Ensure field can open and retrieve a view. func TestField_DeleteView(t *testing.T) { f := OpenField(t, OptFieldTypeDefault()) From 6e3ce01ecb6b06fb86497a2add2cf56b02dd2890 Mon Sep 17 00:00:00 2001 From: reesporte Date: Mon, 3 Jan 2022 11:11:54 -0600 Subject: [PATCH 4/5] explicitly test that getScaledInt works with timestamps --- executor_internal_test.go | 15 +++++++++++++++ 1 file changed, 15 insertions(+) diff --git a/executor_internal_test.go b/executor_internal_test.go index 5c5ed9314..79a9cfe72 100644 --- a/executor_internal_test.go +++ b/executor_internal_test.go @@ -489,3 +489,18 @@ func TestExecutorSafeCopyDistinctTimestamp(t *testing.T) { t.Fatalf("Did not copy results. got %+v, want %+v", copied.Results, response.Results) } } + +func TestGetScaledInt(t *testing.T) { + f := OpenField(t, OptFieldTypeTimestamp(time.Now(), "ms")) + defer f.Close() + // check that fields with type timestamp return the int64 passed in to getScaledInt with nil err + v := time.Now().Unix() + res, err := getScaledInt(f.Field, v) + if err != nil { + t.Errorf("got error %v, expected nil", err) + } + if !reflect.DeepEqual(res, v) { + t.Errorf("expected %v, got %v", v, res) + } + +} From b13538e4266aaa908ace8143526e04ee38e8ec79 Mon Sep 17 00:00:00 2001 From: reesporte Date: Mon, 3 Jan 2022 11:16:14 -0600 Subject: [PATCH 5/5] update doc comment --- field.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/field.go b/field.go index 6435634f2..e45e82d6e 100644 --- a/field.go +++ b/field.go @@ -1426,7 +1426,7 @@ func (f *Field) MinForShard(tx Tx, shard uint64, filter *Row) (ValCount, error) return f.valCountize(min, cnt, bsig) } -// valCountize takes the "raw" min value and count we get from the +// valCountize takes the "raw" value and count we get from the // fragment and calculates the cooked values for this field // (timestamping, decimaling, or just adding in the base). It always // includes the int64 "Val\" value to make comparisons easier in the