From 513edeae9c03524d2259886b794de4cc38ebc22e Mon Sep 17 00:00:00 2001
From: Travis <travis@pilosa.com>
Date: Sat, 14 Mar 2020 22:32:47 -0500
Subject: [PATCH] fix min/max bug for decimal fields

---
 executor_test.go | 58 ++++++++++++++++++++++++++++++++++++++++++++++++
 field.go         | 16 ++++++++++---
 pql/decimal.go   | 14 ++++++++++++
 3 files changed, 85 insertions(+), 3 deletions(-)

diff --git a/executor_test.go b/executor_test.go
index 7926ea8c0..b695fcbe8 100644
--- a/executor_test.go
+++ b/executor_test.go
@@ -34,6 +34,7 @@ import (
 	"github.com/pilosa/pilosa/v2"
 	"github.com/pilosa/pilosa/v2/boltdb"
 	"github.com/pilosa/pilosa/v2/http"
+	"github.com/pilosa/pilosa/v2/pql"
 	"github.com/pilosa/pilosa/v2/server"
 	"github.com/pilosa/pilosa/v2/test"
 	"github.com/pkg/errors"
@@ -1463,6 +1464,63 @@ func TestExecutor_Execute_MinMax(t *testing.T) {
 				})
 			}
 		})
+
+		t.Run("Decimal", func(t *testing.T) {
+			c := test.MustRunCluster(t, 1)
+			defer c.Close()
+			hldr := test.Holder{Holder: c[0].Server.Holder()}
+
+			idx, err := hldr.CreateIndex("i", pilosa.IndexOptions{})
+			if err != nil {
+				t.Fatal(err)
+			}
+
+			tests := []struct {
+				scale int64
+				min   int64
+				max   int64
+				set   pql.Decimal
+			}{
+				{2, 10, 20, pql.Decimal{false, 115, 1}},
+				{2, -10, 20, pql.Decimal{false, 115, 1}},
+				{2, -10, 20, pql.Decimal{true, 95, 1}},
+				{2, -20, -10, pql.Decimal{true, 115, 1}},
+			}
+			for i, test := range tests {
+				fld := fmt.Sprintf("f%d", i)
+				t.Run("MinMaxField_"+fld, func(t *testing.T) {
+					if _, err := idx.CreateField(fld, pilosa.OptFieldTypeDecimal(test.scale, test.min, test.max)); err != nil {
+						t.Fatal(err)
+					}
+
+					if _, err := c[0].API.Query(context.Background(), &pilosa.QueryRequest{Index: "i", Query: fmt.Sprintf(`
+                Set(10, %s=%s)
+            `, fld, test.set)}); err != nil {
+						t.Fatal(err)
+					}
+
+					var pql string
+
+					t.Run("Min", func(t *testing.T) {
+						pql = fmt.Sprintf(`Min(field=%s)`, fld)
+						if result, err := c[0].API.Query(context.Background(), &pilosa.QueryRequest{Index: "i", Query: pql}); err != nil {
+							t.Fatal(err)
+						} else if !reflect.DeepEqual(result.Results[0], pilosa.ValCount{FloatVal: test.set.Float64(), Count: 1}) {
+							t.Fatalf("unexpected min result, test %d: %s", i, spew.Sdump(result))
+						}
+					})
+
+					t.Run("Max", func(t *testing.T) {
+						pql = fmt.Sprintf(`Max(field=%s)`, fld)
+						if result, err := c[0].API.Query(context.Background(), &pilosa.QueryRequest{Index: "i", Query: pql}); err != nil {
+							t.Fatal(err)
+						} else if !reflect.DeepEqual(result.Results[0], pilosa.ValCount{FloatVal: test.set.Float64(), Count: 1}) {
+							t.Fatalf("unexpected max result, test %d: %s", i, spew.Sdump(result))
+						}
+					})
+				})
+			}
+		})
 	})
 
 	t.Run("ColumnID", func(t *testing.T) {
diff --git a/field.go b/field.go
index e7573b2c6..aabcf0f0a 100644
--- a/field.go
+++ b/field.go
@@ -194,6 +194,10 @@ func OptFieldTypeDecimal(scale int64, minmax ...int64) FieldOption {
 		fo.Max = math.MaxInt64
 		if len(minmax) == 2 {
 			min, max := minmax[0], minmax[1]
+			if scale != 0 {
+				min = int64(float64(min) * math.Pow10(int(scale)))
+				max = int64(float64(max) * math.Pow10(int(scale)))
+			}
 			if min > max {
 				return errors.Errorf("decimal field min cannot be greater than max, got %d, %d", min, max)
 			}
@@ -202,7 +206,13 @@ func OptFieldTypeDecimal(scale int64, minmax ...int64) FieldOption {
 		} else if len(minmax) > 2 {
 			return errors.Errorf("unknown extra parameters beyond min and max: %v", minmax)
 		} else if len(minmax) == 1 {
-			fo.Min = minmax[0]
+			// It's not necessary to handle the scale==0 case separately,
+			// but it avoids the type conversion.
+			if scale == 0 {
+				fo.Min = minmax[0]
+			} else {
+				fo.Min = int64(float64(minmax[0]) * math.Pow10(int(scale)))
+			}
 		}
 		fo.Type = FieldTypeDecimal
 		fo.Base = bsiBase(fo.Min, fo.Max)
@@ -1447,7 +1457,7 @@ func (f *Field) MaxForShard(shard uint64, filter *Row) (ValCount, error) {
 	valCount := ValCount{Count: int64(cnt)}
 
 	if f.Options().Type == FieldTypeDecimal {
-		valCount.FloatVal = float64(max) / math.Pow10(int(bsig.Scale))
+		valCount.FloatVal = float64(max+bsig.Base) / math.Pow10(int(bsig.Scale))
 	} else {
 		valCount.Val = max + bsig.Base
 	}
@@ -1482,7 +1492,7 @@ func (f *Field) MinForShard(shard uint64, filter *Row) (ValCount, error) {
 	valCount := ValCount{Count: int64(cnt)}
 
 	if f.Options().Type == FieldTypeDecimal {
-		valCount.FloatVal = float64(min) / math.Pow10(int(bsig.Scale))
+		valCount.FloatVal = float64(min+bsig.Base) / math.Pow10(int(bsig.Scale))
 	} else {
 		valCount.Val = min + bsig.Base
 	}
diff --git a/pql/decimal.go b/pql/decimal.go
index 68b01f2fe..b0c6e58c3 100644
--- a/pql/decimal.go
+++ b/pql/decimal.go
@@ -54,6 +54,20 @@ func (d Decimal) ToInt64(scale int64) int64 {
 	return ret
 }
 
+// Float64 returns d as a float64.
+func (d Decimal) Float64() float64 {
+	var ret float64
+	if d.Scale == 0 {
+		ret = float64(d.Value)
+	} else {
+		ret = float64(d.Value) / math.Pow10(int(d.Scale))
+	}
+	if d.Sign {
+		ret *= -1
+	}
+	return ret
+}
+
 // String returns the string representation of the decimal.
 func (d Decimal) String() string {
 	var s string