Use json.Number decoder to handle large ints in sql wire protocol (#2285)

* Use json.Number decoder to handle large ints in sql wireprotocol

cli/cli.go

@@ -32,11 +32,9 @@ var (
 	Stderr io.Writer     = os.Stderr
 )
 
-var (
-	splash string = fmt.Sprintf(`FeatureBase CLI (%s)
+var splash string = fmt.Sprintf(`FeatureBase CLI (%s)
 Type "\q" to quit.
 `, featurebase.Version)
-)
 
 // Ensure type implments interfaces.
 var _ printer = (*Command)(nil)
@@ -297,7 +295,7 @@ func (cmd *Command) Run(ctx context.Context) error {
 			}
 			return nil
 		default:
-			//pass
+			// pass
 		}
 	}
 }
@@ -329,7 +327,6 @@ func (cmd *Command) executeAndWriteQuery(qry query) error {
 	if err != nil {
 		return errors.Wrap(err, "making query")
 	}
-
 	if err := writeTable(queryResponse, cmd.writeOptions, cmd.output, cmd.Stdout, cmd.Stderr); err != nil {
 		return errors.Wrap(err, "writing out response")
 	}

cli/queryer.go

@@ -45,8 +45,10 @@ func (qryr *standardQueryer) Query(org string, db string, sql io.Reader) (*featu
 	if err != nil {
 		return nil, errors.Wrap(err, "reading response")
 	}
+
 	sqlResponse := &featurebase.WireQueryResponse{}
 	// TODO(tlt): switch this back once all responses are typed
+	// TODO(twg) 2023/03/01 using json.Number to decode large ints so care must be made
 	// if err := json.Unmarshal(fullbod, sqlResponse); err != nil {
 	if err := sqlResponse.UnmarshalJSONTyped(fullbod, true); err != nil {
 		return nil, errors.Wrapf(err, "unmarshaling query response, body:\n'%s'\n", fullbod)

http_handler.go

@@ -1384,7 +1384,6 @@ func (h *Handler) writeBadRequest(w http.ResponseWriter, r *http.Request, err er
 // we do not track these requests as user requests
 // TODO(pok) - thus is there anything we need here to align with how we do this for other nodes
 func (h *Handler) handlePostSQLPlanOperator(w http.ResponseWriter, r *http.Request) {
-
 	writeError := func(err error) {
 		if err != nil {
 			w.Write(wireprotocol.WriteError(err))
@@ -1439,7 +1438,6 @@ func (h *Handler) handlePostSQLPlanOperator(w http.ResponseWriter, r *http.Reque
 // supports a ?plan=true|false parameter to send back the plan in the
 // query response
 func (h *Handler) handlePostSQL(w http.ResponseWriter, r *http.Request) {
-
 	includePlan := false
 	includePlanValue := r.URL.Query().Get("plan")
 	if len(includePlanValue) > 0 {

sql3/sql_complex_test.go

@@ -2978,3 +2978,63 @@ func TestPlanner_BulkInsertParquet(t *testing.T) {
 		assert.Equal(t, row[1], row[2])
 	})
 }
+
+func TestPlanner_BulkInsert_FP1916(t *testing.T) {
+	c := test.MustRunCluster(t, 3)
+	defer c.Close()
+	// 8924809397503602651 is larger than 2^53 which is the largest integer value representable in float64
+	node := c.GetNode(0).Server
+	_, _, _, err := sql_test.MustQueryRows(t, node, `create table greg-test (
+		_id STRING,
+		id_col ID,
+		string_col STRING cachetype ranked size 1000,
+		int_col int,
+		decimal_col DECIMAL(2),
+		bool_col BOOL
+		time_col TIMESTAMP,
+		stringset_col STRINGSET,
+		ideset_col IDSET
+	);`)
+	assert.NoError(t, err)
+
+	_, _, _, err = sql_test.MustQueryRows(t, node, `BULK INSERT INTO greg-test (
+		_id,
+		id_col,
+		string_col,
+		int_col,
+		decimal_col,
+		bool_col,
+		time_col,
+		stringset_col,
+		ideset_col)
+		map (
+		0 ID,
+		1 STRING,
+		2 INT,
+		3 DECIMAL(2),
+		4 BOOL,
+		5 TIMESTAMP,
+		6 STRINGSET,
+		7 IDSET)
+		transform(
+		@1,
+		@0,
+		@1,
+		@2,
+		@3,
+		@4,
+		@5,
+		@6,
+		@7)
+		FROM x'1,TEST2,8924809397503602651,31.2,1,"2014-07-15T01:18:46Z",stringset1, 1'
+		with
+			BATCHSIZE 10000
+			format 'CSV'
+			input 'STREAM';`)
+	assert.NoError(t, err)
+	results, _, _, err := sql_test.MustQueryRows(t, node, `select int_col from greg-test`)
+	assert.NoError(t, err)
+	got := results[0][0].(int64)
+	expected := int64(8924809397503602651)
+	assert.Equal(t, got, expected)
+}

wire_response.go

@@ -1,6 +1,7 @@
 package pilosa
 
 import (
+	"bytes"
 	"encoding/json"
 	"fmt"
 	"log"
@@ -44,6 +45,24 @@ func (s *WireQueryResponse) UnmarshalJSON(in []byte) error {
 	return s.UnmarshalJSONTyped(in, false)
 }
 
+// decodeDataWithNumber allows access to integers larger than 2^53 which is normally
+// not supported in JSON, this extra parse is inefficient and should be re-examined
+// once all the typing is settled
+func (s *WireQueryResponse) decodeDataWithNumber(in []byte) ([]interface{}, error) {
+	dat := make(map[string]interface{})
+	d := json.NewDecoder(bytes.NewBuffer(in))
+	d.UseNumber()
+	if err := d.Decode(&dat); err != nil {
+		return nil, err
+	}
+	if a, ok := dat["data"].([]interface{}); ok {
+		return a, nil
+	}
+	return nil, noArrayPresent
+}
+
+var noArrayPresent = errors.New("no data in response")
+
 // UnmarshalJSONTyped is a temporary until we send typed values back in sql
 // responses. At that point, we can get rid of the typed=false path. In order to
 // do that, we need sql3 to return typed values, and we need the sql3/test/defs
@@ -56,6 +75,12 @@ func (s *WireQueryResponse) UnmarshalJSONTyped(in []byte, typed bool) error {
 	if err := json.Unmarshal(in, &aux); err != nil {
 		return err
 	}
+	bypass, err := s.decodeDataWithNumber(in)
+	if err != nil {
+		if err != noArrayPresent { // no data is not an error but just allows the compiler
+			return err
+		}
+	}
 
 	*s = WireQueryResponse(aux)
 
@@ -85,22 +110,38 @@ func (s *WireQueryResponse) UnmarshalJSONTyped(in []byte, typed bool) error {
 		for j, hdr := range s.Schema.Fields {
 			switch hdr.BaseType {
 			case dax.BaseTypeID, dax.BaseTypeInt:
-				if _, ok := s.Data[i][j].(float64); ok {
-					s.Data[i][j] = int64(s.Data[i][j].(float64))
+				jn := bypass[i].([]interface{})[j] // forced to do this to handle ints bigger than 2^53
+				if v, ok := jn.(json.Number); ok {
+					if x, err := v.Int64(); err == nil {
+						s.Data[i][j] = x
+					} else {
+						return errors.Wrap(err, "can't be decoded as int64")
+					}
 				}
 
 			case dax.BaseTypeIDSet:
-				if src, ok := s.Data[i][j].([]interface{}); ok {
+				if _, ok := s.Data[i][j].([]interface{}); ok {
+					src := bypass[i].([]interface{})[j].([]interface{})
 					if typed {
 						val := make(IDSet, len(src))
 						for k := range src {
-							val[k] = int64(src[k].(float64))
+							v := src[k].(json.Number)
+							if x, err := v.Int64(); err == nil {
+								val[k] = x
+							} else {
+								return errors.Wrap(err, "can't be decoded as int64")
+							}
 						}
 						s.Data[i][j] = val
 					} else {
 						val := make([]int64, len(src))
 						for k := range src {
-							val[k] = int64(src[k].(float64))
+							v := src[k].(json.Number)
+							if x, err := v.Int64(); err == nil {
+								val[k] = x
+							} else {
+								return errors.Wrap(err, "can't be decoded as int64")
+							}
 						}
 						s.Data[i][j] = val
 					}

wireprotocol/wireprimitives.go

@@ -4,9 +4,8 @@ import (
 	"bufio"
 	"bytes"
 	"encoding/binary"
-	"time"
-
 	"io"
+	"time"
 
 	"github.com/featurebasedb/featurebase/v3/errors"
 	"github.com/featurebasedb/featurebase/v3/pql"
@@ -344,7 +343,6 @@ func WriteRow(row types.Row, schema types.Schema) ([]byte, error) {
 }
 
 func ReadRow(reader io.Reader, schema types.Schema) (types.Row, error) {
-
 	row := make(types.Row, len(schema))
 
 	for idx, s := range schema {
@@ -427,7 +425,7 @@ func ReadRow(reader io.Reader, schema types.Schema) (types.Row, error) {
 				row[idx] = nil
 			} else {
 				set := make([]int64, len)
-				for j, _ := range set {
+				for j := range set {
 					var value int64
 					err := binary.Read(reader, binary.BigEndian, &value)
 					if err != nil {
@@ -465,7 +463,7 @@ func ReadRow(reader io.Reader, schema types.Schema) (types.Row, error) {
 				row[idx] = nil
 			} else {
 				set := make([]string, len)
-				for j, _ := range set {
+				for j := range set {
 					var vlen int16
 					err = binary.Read(reader, binary.BigEndian, &vlen)
 					if err != nil {