implement percentiles on timestamp/decimal, still needs tests

This commit is contained in:
Matthew Jaffee 2021-12-10 15:21:29 -06:00
parent 9283d52141
commit 7935624549
3 changed files with 57 additions and 26 deletions

View file

@ -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)
}

View file

@ -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
}

View file

@ -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) {