feat: support decimal

Author: RaphDal

README.md

@@ -188,6 +188,37 @@ will now also serialize ``float64`` (double-precision) columns as binary.
 You might see a performance uplift if this is a dominant data type in your
 ingestion workload.
 
+## Decimal columns
+
+QuestDB server version 9.2.0 and newer supports decimal columns with arbitrary precision and scale.
+The Go client converts supported decimal values to QuestDB's text/binary wire format automatically, pass any of the following to `DecimalColumn`:
+
+- `questdb.ScaledDecimal`, including helpers like `questdb.NewDecimalFromInt64` and `questdb.NewDecimal`.
+- Types implementing `questdb.DecimalMarshaler`.
+- `github.com/shopspring/decimal.Decimal` values or pointers.
+- `nil` or `questdb.NullDecimal()` to send a `NULL`.
+
+```go
+price := qdb.NewDecimalFromInt64(12345, 2) // 123.45 with scale 2
+commission := qdb.NewDecimal(big.NewInt(-750), 4) // -0.0750 with scale 4
+
+err = sender.
+    Table("trades").
+    Symbol("symbol", "ETH-USD").
+    DecimalColumn("price", price).
+    DecimalColumn("commission", commission).
+    AtNow(ctx)
+```
+
+To emit textual decimals, pass a validated string literal (without the trailing `d`—the client adds it):
+
+```go
+err = sender.
+    Table("quotes").
+    DecimalColumn("mid", "1.23456").
+    AtNow(ctx)
+```
+
 ## Pooled Line Senders
 
 **Warning: Experimental feature designed for use with HTTP senders ONLY**

buffer.go

@@ -573,6 +573,47 @@ func (b *buffer) Float64Column(name string, val float64) *buffer {
 	return b
 }
 
+func (b *buffer) DecimalColumn(name string, val any) *buffer {
+	if !b.prepareForField() {
+		return b
+	}
+	b.lastErr = b.writeColumnName(name)
+	if b.lastErr != nil {
+		return b
+	}
+	b.WriteByte('=')
+	if str, ok := val.(string); ok {
+		if err := validateDecimalText(str); err != nil {
+			b.lastErr = err
+			return b
+		}
+		b.WriteString(str)
+		b.WriteByte('d')
+		b.hasFields = true
+		return b
+	}
+
+	dec, err := normalizeDecimalValue(val)
+	if err != nil {
+		b.lastErr = err
+		return b
+	}
+	scale, payload, err := dec.toBinary()
+	if err != nil {
+		b.lastErr = err
+		return b
+	}
+	b.WriteByte('=')
+	b.WriteByte(decimalBinaryTypeCode)
+	b.WriteByte(scale)
+	b.WriteByte(byte(len(payload)))
+	if len(payload) > 0 {
+		b.Write(payload)
+	}
+	b.hasFields = true
+	return b
+}
+
 func (b *buffer) Float64ColumnBinary(name string, val float64) *buffer {
 	if !b.prepareForField() {
 		return b

buffer_test.go

@@ -37,8 +37,23 @@ import (
 	"github.com/stretchr/testify/assert"
 )
 
+const decimalTypeCode byte = 0x17
+
 type bufWriterFn func(b *qdb.Buffer) error
 
+type fakeShopspringDecimal struct {
+	coeff *big.Int
+	exp   int32
+}
+
+func (f fakeShopspringDecimal) Coefficient() *big.Int {
+	return f.coeff
+}
+
+func (f fakeShopspringDecimal) Exponent() int32 {
+	return f.exp
+}
+
 func newTestBuffer() qdb.Buffer {
 	return qdb.NewBuffer(128*1024, 1024*1024, 127)
 }
@@ -481,6 +496,132 @@ func TestFloat64ColumnBinary(t *testing.T) {
 	}
 }
 
+func TestDecimalColumnText(t *testing.T) {
+	prefix := []byte(testTable + " price==")
+	testCases := []struct {
+		name     string
+		value    any
+		expected []byte
+	}{
+		{
+			name:     "positive",
+			value:    qdb.NewDecimalFromInt64(12345, 2),
+			expected: append(prefix, 0x17, 0x02, 0x02, 0x30, 0x39, 0x0A),
+		},
+		{
+			name:     "negative",
+			value:    qdb.NewDecimal(big.NewInt(-12345), 3),
+			expected: append(prefix, 0x17, 0x03, 0x02, 0xCF, 0xC7, 0x0A),
+		},
+		{
+			name:     "zero with scale",
+			value:    qdb.NewDecimalFromInt64(0, 4),
+			expected: append(prefix, 0x17, 0x04, 0x01, 0x0, 0x0A),
+		},
+		{
+			name:     "null decimal",
+			value:    qdb.NullDecimal(),
+			expected: append(prefix, 0x17, 0x0, 0x0, 0x0A),
+		},
+		{
+			name:     "shopspring compatible",
+			value:    fakeShopspringDecimal{coeff: big.NewInt(123456), exp: -4},
+			expected: append(prefix, 0x17, 0x04, 0x03, 0x01, 0xE2, 0x40, 0x0A),
+		},
+		{
+			name:     "nil pointer treated as null",
+			value:    (*fakeShopspringDecimal)(nil),
+			expected: append(prefix, 0x17, 0x0, 0x0, 0x0A),
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			buf := newTestBuffer()
+			err := buf.Table(testTable).DecimalColumn("price", tc.value).At(time.Time{}, false)
+			assert.NoError(t, err)
+			assert.Equal(t, tc.expected, buf.Messages())
+		})
+	}
+}
+
+func TestDecimalColumnStringValidation(t *testing.T) {
+	t.Run("valid strings", func(t *testing.T) {
+		testCases := []struct {
+			name     string
+			value    string
+			expected string
+		}{
+			{"integer", "123", "123d"},
+			{"decimal", "123.450", "123.450d"},
+			{"negative", "-0.001", "-0.001d"},
+			{"exponent positive", "1.2e3", "1.2e3d"},
+			{"exponent negative", "-4.5E-2", "-4.5E-2d"},
+			{"nan token", "NaN", "NaNd"},
+			{"infinity token", "Infinity", "Infinityd"},
+		}
+		for _, tc := range testCases {
+			t.Run(tc.name, func(t *testing.T) {
+				buf := newTestBuffer()
+				err := buf.Table(testTable).DecimalColumn("price", tc.value).At(time.Time{}, false)
+				assert.NoError(t, err)
+				expected := []byte(testTable + " price=" + tc.expected + "\n")
+				assert.Equal(t, expected, buf.Messages())
+			})
+		}
+	})
+
+	t.Run("invalid strings", func(t *testing.T) {
+		testCases := []struct {
+			name  string
+			value string
+		}{
+			{"empty", ""},
+			{"sign only", "+"},
+			{"double dot", "12.3.4"},
+			{"invalid char", "12a3"},
+			{"exponent missing mantissa", "e10"},
+			{"exponent no digits", "1.2e"},
+			{"exponent sign no digits", "1.2e+"},
+		}
+		for _, tc := range testCases {
+			t.Run(tc.name, func(t *testing.T) {
+				buf := newTestBuffer()
+				err := buf.Table(testTable).DecimalColumn("price", tc.value).At(time.Time{}, false)
+				assert.Error(t, err)
+				assert.Contains(t, err.Error(), "decimal")
+				assert.Empty(t, buf.Messages())
+			})
+		}
+	})
+}
+
+func TestDecimalColumnErrors(t *testing.T) {
+	t.Run("invalid scale", func(t *testing.T) {
+		buf := newTestBuffer()
+		dec := qdb.NewDecimalFromInt64(1, 100)
+		err := buf.Table(testTable).DecimalColumn("price", dec).At(time.Time{}, false)
+		assert.ErrorContains(t, err, "decimal scale")
+		assert.Empty(t, buf.Messages())
+	})
+
+	t.Run("overflow", func(t *testing.T) {
+		buf := newTestBuffer()
+		bigVal := new(big.Int).Lsh(big.NewInt(1), 2100)
+		dec := qdb.NewDecimal(bigVal, 0)
+		err := buf.Table(testTable).DecimalColumn("price", dec).At(time.Time{}, false)
+		assert.ErrorContains(t, err, "exceeds 256-bit range")
+		assert.Empty(t, buf.Messages())
+	})
+
+	t.Run("unsupported type", func(t *testing.T) {
+		buf := newTestBuffer()
+		err := buf.Table(testTable).DecimalColumn("price", struct{}{}).At(time.Time{}, false)
+		assert.ErrorContains(t, err, "unsupported decimal column value type")
+		assert.Empty(t, buf.Messages())
+	})
+}
+
 func TestFloat64Array1DColumn(t *testing.T) {
 	testCases := []struct {
 		name   string

conf_parse.go

@@ -169,8 +169,8 @@ func confFromStr(conf string) (*lineSenderConfig, error) {
 					return nil, NewInvalidConfigStrError("invalid %s value, %q is not a valid int", k, v)
 				}
 				pVersion := protocolVersion(version)
-				if pVersion < ProtocolVersion1 || pVersion > ProtocolVersion2 {
-					return nil, NewInvalidConfigStrError("current client only supports protocol version 1 (text format for all datatypes), 2 (binary format for part datatypes) or explicitly unset")
+				if pVersion < ProtocolVersion1 || pVersion > ProtocolVersion3 {
+					return nil, NewInvalidConfigStrError("current client only supports protocol version 1 (text format for all datatypes), 2 (binary format for part datatypes), 3 (decimals) or explicitly unset")
 				}
 				senderConf.protocolVersion = pVersion
 			}