Feat/issue 89 support bigint (#90)

Author: yokofly

.github/workflows/run-tests.yml

@@ -15,39 +15,17 @@ jobs:
       fail-fast: true
       matrix:
         go:
-          - '1.16'
-          - '1.17'
-          - '1.18'
-          - '1.19'
-          - '1.20'
-          - '1.21'
+          - '1.22.x'
+          - '1.23.x'
         proton:
           - latest
 
-    services:
-      proton:
-        image: ghcr.io/timeplus-io/proton:${{ matrix.proton }}
-        ports:
-          - 3218:3218 # HTTP Streaming
-          - 8123:8123 # HTTP Snapshot
-          - 8463:8463 # TCP Streaming
-          - 5432:5432 # Postgres Snapshot
-          - 7587:7587 # TCP Snapshot
-        env:
-          MAX_CONCURRENT_QUERIES: 100               # Default: 100
-          MAX_CONCURRENT_SELECT_QUERIES: 100        # Default: 100
-          MAX_CONCURRENT_INSERT_QUERIES: 100        # Default: 100
-          MAX_CONCURRENT_STREAMING_QUERIES: 100     # Default: 100
-          MAX_SERVER_MEMORY_USAGE_TO_RAM_RATIO: 0.9 # Default: 0.9
-          MAX_SERVER_MEMORY_CACHE_TO_RAM_RATIO: 0.5 # Default: 0.5
-
     steps:
-      - uses: actions/checkout@v2
+      - uses: actions/checkout@v5
 
       - name: Install Go ${{ matrix.go }}
-        uses: actions/setup-go@v2.1.5
+        uses: actions/setup-go@v5
         with:
-          stable: false
           go-version: ${{ matrix.go }}
 
       - name: Run tests
@@ -56,6 +34,40 @@ jobs:
           sudo rm -rf /usr/local/lib/android
           sudo rm -rf /opt/ghc
           df -h
+          docker run -d --name proton \
+            -p 3218:3218 \
+            -p 8123:8123 \
+            -p 8463:8463 \
+            -p 5432:5432 \
+            -p 7587:7587 \
+            -e MAX_CONCURRENT_QUERIES=100 \
+            -e MAX_CONCURRENT_SELECT_QUERIES=100 \
+            -e MAX_CONCURRENT_INSERT_QUERIES=100 \
+            -e MAX_CONCURRENT_STREAMING_QUERIES=100 \
+            -e MAX_SERVER_MEMORY_USAGE_TO_RAM_RATIO=0.9 \
+            -e MAX_SERVER_MEMORY_CACHE_TO_RAM_RATIO=0.5 \
+            ghcr.io/timeplus-io/proton:${{ matrix.proton }} \
+            /bin/bash -c "sed -i'' 's/preallocate: true/preallocate: false/g' /etc/proton-server/config.yaml && exec /entrypoint.sh"
+          ready=0
+          for i in $(seq 1 60); do
+            if curl -fsS "http://127.0.0.1:8123/?query=SELECT%201" | grep -qx "1"; then
+              echo "Proton is ready"
+              ready=1
+              break
+            fi
+            sleep 1
+          done
+          if [ "$ready" -ne 1 ]; then
+            echo "Proton did not become ready within 60 seconds"
+            docker logs --tail=200 proton || true
+            exit 1
+          fi
           go test -v .
           go test -v ./tests
           go test -v ./lib/...
+
+      - name: Dump Proton logs (on failure)
+        if: failure()
+        run: |
+          docker ps -a
+          docker logs --tail=200 proton || true

lib/column/bigint.go

@@ -30,6 +30,7 @@ type BigInt struct {
 	data   []byte
 	chType Type
 	name   string
+	signed bool
 }
 
 func (col *BigInt) Name() string {
@@ -78,14 +79,18 @@ func (col *BigInt) Append(v interface{}) (nulls []uint8, err error) {
 	case []big.Int:
 		nulls = make([]uint8, len(v))
 		for _, v := range v {
-			col.append(&v)
+			if err := col.append(&v); err != nil {
+				return nil, err
+			}
 		}
 	case []*big.Int:
 		nulls = make([]uint8, len(v))
 		for i, v := range v {
 			switch {
 			case v != nil:
-				col.append(v)
+				if err := col.append(v); err != nil {
+					return nil, err
+				}
 			default:
 				col.data, nulls[i] = append(col.data, make([]byte, col.size)...), 1
 			}
@@ -103,11 +108,11 @@ func (col *BigInt) Append(v interface{}) (nulls []uint8, err error) {
 func (col *BigInt) AppendRow(v interface{}) error {
 	switch v := v.(type) {
 	case big.Int:
-		col.append(&v)
+		return col.append(&v)
 	case *big.Int:
 		switch {
 		case v != nil:
-			col.append(v)
+			return col.append(v)
 		default:
 			col.data = append(col.data, make([]byte, col.size)...)
 		}
@@ -133,40 +138,85 @@ func (col *BigInt) Encode(encoder *binary.Encoder) error {
 }
 
 func (col *BigInt) row(i int) *big.Int {
-	return rawToBigInt(col.data[i*col.size : (i+1)*col.size])
+	src := col.data[i*col.size : (i+1)*col.size]
+	if col.signed {
+		return rawToBigInt(src)
+	}
+	return rawToBigUInt(src)
 }
 
-func (col *BigInt) append(v *big.Int) {
+func (col *BigInt) append(v *big.Int) error {
 	dest := make([]byte, col.size)
-	bigIntToRaw(dest, new(big.Int).Set(v))
+	var err error
+	if col.signed {
+		err = bigIntToRawSigned(dest, new(big.Int).Set(v))
+	} else {
+		err = bigIntToRawUnsigned(dest, new(big.Int).Set(v))
+	}
+	if err != nil {
+		return &Error{
+			ColumnType: string(col.chType),
+			Err:        err,
+		}
+	}
 	col.data = append(col.data, dest...)
+	return nil
 }
 
-func bigIntToRaw(dest []byte, v *big.Int) {
-	var sign int
+func bigIntToRawUnsigned(dest []byte, v *big.Int) error {
 	if v.Sign() < 0 {
+		return fmt.Errorf("negative value for %s", "unsigned integer")
+	}
+	if v.BitLen() > len(dest)*8 {
+		return fmt.Errorf("value out of range for %d-bit unsigned integer", len(dest)*8)
+	}
+	v.FillBytes(dest)
+	endianSwap(dest, false)
+	return nil
+}
+
+func bigIntToRawSigned(dest []byte, v *big.Int) error {
+	bits := len(dest) * 8
+	switch v.Sign() {
+	case -1:
+		abs := new(big.Int).Abs(v)
+		if abs.BitLen() > bits || (abs.BitLen() == bits && abs.TrailingZeroBits() != uint(bits-1)) {
+			return fmt.Errorf("value out of range for %d-bit signed integer", bits)
+		}
 		v.Not(v).FillBytes(dest)
-		sign = -1
-	} else {
+		endianSwap(dest, true)
+		return nil
+	default:
+		if v.BitLen() > bits-1 {
+			return fmt.Errorf("value out of range for %d-bit signed integer", bits)
+		}
 		v.FillBytes(dest)
+		endianSwap(dest, false)
+		return nil
 	}
-	endianSwap(dest, sign < 0)
 }
 
-func rawToBigInt(v []byte) *big.Int {
+func rawToBigUInt(src []byte) *big.Int {
+	tmp := append([]byte(nil), src...)
+	endianSwap(tmp, false)
+	return new(big.Int).SetBytes(tmp)
+}
+
+func rawToBigInt(src []byte) *big.Int {
+	tmp := append([]byte(nil), src...)
 	// LittleEndian to BigEndian
-	endianSwap(v, false)
+	endianSwap(tmp, false)
 	var lt = new(big.Int)
-	if len(v) > 0 && v[0]&0x80 != 0 {
+	if len(tmp) > 0 && tmp[0]&0x80 != 0 {
 		// [0] ^ will +1
-		for i := 0; i < len(v); i++ {
-			v[i] = ^v[i]
+		for i := 0; i < len(tmp); i++ {
+			tmp[i] = ^tmp[i]
 		}
-		lt.SetBytes(v)
+		lt.SetBytes(tmp)
 		// neg ^ will -1
 		lt.Not(lt)
 	} else {
-		lt.SetBytes(v)
+		lt.SetBytes(tmp)
 	}
 	return lt
 }

lib/column/bigint_test.go

@@ -0,0 +1,74 @@
+package column
+
+import (
+	"bytes"
+	"math/big"
+	"testing"
+	"time"
+)
+
+func TestUInt128ColumnFactory(t *testing.T) {
+	c, err := Type("uint128").Column("col", time.UTC)
+	if err != nil {
+		t.Fatalf("Column(): %v", err)
+	}
+	bi, ok := c.(*BigInt)
+	if !ok {
+		t.Fatalf("expected *BigInt, got %T", c)
+	}
+	if bi.size != 16 {
+		t.Fatalf("expected size=16, got %d", bi.size)
+	}
+	if bi.signed {
+		t.Fatalf("expected unsigned BigInt for uint128")
+	}
+}
+
+func TestUInt128RoundTripHighBit(t *testing.T) {
+	v := new(big.Int).Lsh(big.NewInt(1), 127)
+	v.Add(v, big.NewInt(12345))
+
+	raw := make([]byte, 16)
+	if err := bigIntToRawUnsigned(raw, v); err != nil {
+		t.Fatalf("bigIntToRawUnsigned(): %v", err)
+	}
+	rawCopy := append([]byte(nil), raw...)
+
+	got := rawToBigUInt(raw)
+	if got.Cmp(v) != 0 {
+		t.Fatalf("round-trip mismatch: got=%s want=%s", got.String(), v.String())
+	}
+	if !bytes.Equal(raw, rawCopy) {
+		t.Fatalf("rawToBigUInt mutated input")
+	}
+}
+
+func TestUInt128AppendRejectsNegativeAndOverflow(t *testing.T) {
+	col := &BigInt{
+		size:   16,
+		chType: "uint128",
+		signed: false,
+	}
+
+	if err := col.AppendRow(big.NewInt(-1)); err == nil {
+		t.Fatalf("expected error for negative value")
+	}
+
+	tooBig := new(big.Int).Lsh(big.NewInt(1), 128)
+	if err := col.AppendRow(tooBig); err == nil {
+		t.Fatalf("expected error for out-of-range value")
+	}
+}
+
+func TestRawToBigIntDoesNotMutateInput(t *testing.T) {
+	raw := make([]byte, 16)
+	if err := bigIntToRawSigned(raw, big.NewInt(-128)); err != nil {
+		t.Fatalf("bigIntToRawSigned(): %v", err)
+	}
+	rawCopy := append([]byte(nil), raw...)
+
+	_ = rawToBigInt(raw)
+	if !bytes.Equal(raw, rawCopy) {
+		t.Fatalf("rawToBigInt mutated input")
+	}
+}

lib/column/codegen/column.tpl

@@ -45,18 +45,28 @@ func (t Type) Column(name string, tz *time.Location) (Interface, error) {
 			size:   16,
 			chType: t,
 			name:   name,
+			signed: true,
+		}, nil
+	case "uint128":
+		return &BigInt{
+			size:   16,
+			chType: t,
+			name:   name,
+			signed: false,
 		}, nil
 	case "int256":
 		return &BigInt{
 			size:   32,
 			chType: t,
 			name:   name,
+			signed: true,
 		}, nil
 	case "uint256":
 		return &BigInt{
 			size:   32,
 			chType: t,
 			name:   name,
+			signed: false,
 		}, nil
 	case "ipv4":
 		return &IPv4{name: name}, nil
@@ -281,4 +291,4 @@ func (col *{{ .ChType }}) AppendRow(v interface{}) error {
 	return nil
 }
 
-{{- end }}
+{{- end }}

lib/column/column_gen.go

@@ -236,7 +236,12 @@ func (col *Decimal) Encode(encoder *binary.Encoder) error {
 			default:
 				bi = v.BigInt()
 			}
-			bigIntToRaw(scratch[i*size:(i+1)*size], bi)
+			if err := bigIntToRawSigned(scratch[i*size:(i+1)*size], bi); err != nil {
+				return &Error{
+					ColumnType: string(col.chType),
+					Err:        err,
+				}
+			}
 		}
 		return encoder.Raw(scratch)
 	}

lib/column/decimal.go

@@ -41,7 +41,7 @@ func TestIssue164(t *testing.T) {
 			if !assert.NoError(t, err) {
 				return
 			}
-			if batch, err := scope.Prepare("INSERT INTO issue_164 (* except _tp_time)"); assert.NoError(t, err) {
+			if batch, err := scope.Prepare("INSERT INTO issue_164 (* except (_tp_time, _tp_sn))"); assert.NoError(t, err) {
 				stmtParams := make([]interface{}, 0)
 				stmtParams = append(stmtParams, sql.NamedArg{Name: "id", Value: int32(10)})
 				stmtParams = append(stmtParams, sql.NamedArg{Name: "anything", Value: nil})