From cbbaba98cd357ba874df09ad99d5b1a55ad77cac Mon Sep 17 00:00:00 2001 From: tgruben Date: Thu, 2 Mar 2023 14:49:55 -0600 Subject: [PATCH] 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 | 7 ++-- cli/queryer.go | 2 ++ http_handler.go | 2 -- sql3/sql_complex_test.go | 60 ++++++++++++++++++++++++++++++++++ wire_response.go | 51 ++++++++++++++++++++++++++--- wireprotocol/wireprimitives.go | 8 ++--- 6 files changed, 113 insertions(+), 17 deletions(-) diff --git a/cli/cli.go b/cli/cli.go index cc8f2e0a3..e684961ed 100644 --- a/cli/cli.go +++ b/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") } diff --git a/cli/queryer.go b/cli/queryer.go index 6d7ec7865..9dd8b321a 100644 --- a/cli/queryer.go +++ b/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) diff --git a/http_handler.go b/http_handler.go index 012c4f73d..e77377350 100644 --- a/http_handler.go +++ b/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 { diff --git a/sql3/sql_complex_test.go b/sql3/sql_complex_test.go index 5aebc1bff..ff8bc5386 100644 --- a/sql3/sql_complex_test.go +++ b/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) +} diff --git a/wire_response.go b/wire_response.go index 17828b221..1eff5351e 100644 --- a/wire_response.go +++ b/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 } diff --git a/wireprotocol/wireprimitives.go b/wireprotocol/wireprimitives.go index be88ebd1d..4b9e530fd 100644 --- a/wireprotocol/wireprimitives.go +++ b/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 {