fix(readutil): preserve decimal scale in zonemap filters

pkg/container/types/decimal.go

@@ -199,6 +199,9 @@ func CompareDecimal128(x Decimal128, y Decimal128) int {
 }
 
 func CompareDecimal64WithScale(x, y Decimal64, scale1, scale2 int32) int {
+	if scale1 == scale2 {
+		return x.Compare(y)
+	}
 	if x.Sign() != y.Sign() {
 		if x.Sign() {
 			return -1
@@ -208,25 +211,29 @@ func CompareDecimal64WithScale(x, y Decimal64, scale1, scale2 int32) int {
 	}
 	var err error
 	if scale1 < scale2 {
-		x, err = x.Scale(scale2 - scale1)
-		if err != nil {
-			if x.Sign() {
-				return -1
-			} else {
-				return 1
-			}
-		}
-		return x.Compare(y)
+		scaled := x
+		scaled, err = scaled.Scale(scale2 - scale1)
+		if err != nil || scaled.Sign() != x.Sign() {
+			return CompareDecimal128WithScale(
+				Decimal128FromDecimal64(x, scale1),
+				Decimal128FromDecimal64(y, scale2),
+				scale1,
+				scale2,
+			)
+		}
+		return scaled.Compare(y)
 	} else {
-		y, err = y.Scale(scale1 - scale2)
-		if err != nil {
-			if x.Sign() {
-				return 1
-			} else {
-				return -1
-			}
-		}
-		return x.Compare(y)
+		scaled := y
+		scaled, err = scaled.Scale(scale1 - scale2)
+		if err != nil || scaled.Sign() != y.Sign() {
+			return CompareDecimal128WithScale(
+				Decimal128FromDecimal64(x, scale1),
+				Decimal128FromDecimal64(y, scale2),
+				scale1,
+				scale2,
+			)
+		}
+		return x.Compare(scaled)
 	}
 }
 

pkg/container/types/decimal_test.go

@@ -200,6 +200,32 @@ func TestCompare64(t *testing.T) {
 	}
 }
 
+func TestCompareDecimal64WithScaleFallsBackOnSignedOverflow(t *testing.T) {
+	positive, err := ParseDecimal64("9.99999999999999998", 18, 17)
+	require.NoError(t, err)
+	negative, err := ParseDecimal64("-9.99999999999999998", 18, 17)
+	require.NoError(t, err)
+	positiveBound, err := ParseDecimal64("100", 18, 0)
+	require.NoError(t, err)
+	negativeBound, err := ParseDecimal64("-100", 18, 0)
+	require.NoError(t, err)
+
+	require.Less(t, CompareDecimal64WithScale(positive, positiveBound, 17, 0), 0)
+	require.Greater(t, CompareDecimal64WithScale(negative, negativeBound, 17, 0), 0)
+	require.Greater(t, CompareDecimal64WithScale(positiveBound, positive, 0, 17), 0)
+	require.Less(t, CompareDecimal64WithScale(negativeBound, negative, 0, 17), 0)
+}
+
+func TestCompareDecimal64WithScaleFastPaths(t *testing.T) {
+	negative := Decimal64(1).Minus()
+
+	require.Less(t, CompareDecimal64WithScale(Decimal64(1), Decimal64(2), 2, 2), 0)
+	require.Less(t, CompareDecimal64WithScale(negative, Decimal64(1), 1, 2), 0)
+	require.Greater(t, CompareDecimal64WithScale(Decimal64(1), negative, 2, 1), 0)
+	require.Zero(t, CompareDecimal64WithScale(Decimal64(12), Decimal64(120), 1, 2))
+	require.Zero(t, CompareDecimal64WithScale(Decimal64(120), Decimal64(12), 2, 1))
+}
+
 func TestCompare128(t *testing.T) {
 	x := Decimal128{0, 0}
 	y := Decimal128{^x.B0_63, ^x.B64_127}