mirror of
https://github.com/featurebasedb/featurebase.git
synced 2026-08-28 10:54:59 +00:00
[FB-1831] distribute bulk insert to owning node (#2391)
* distribute bulk insert to owning node
(cherry picked from commit e8505d8a53)
This commit is contained in:
tgruben
Joe Friedrich
parent
74ee3ebf0e
commit
5ec31d4159
4 changed files with 212 additions and 67 deletions
|
|
@ -552,7 +552,6 @@ func (b *Batch) Add(rec Row) error {
|
|||
// empty string is not a valid value at this point (Pilosa refuses to translate it)
|
||||
if val == "" { //
|
||||
b.rowIDs[i] = append(rowIDs, nilSentinel)
|
||||
|
||||
} else if rowID, ok := b.getRowTranslation(field.Name, val); ok {
|
||||
b.rowIDs[i] = append(rowIDs, rowID)
|
||||
} else {
|
||||
|
|
|
|||
26
importer.go
26
importer.go
|
|
@ -32,12 +32,14 @@ var _ Importer = &onPremImporter{}
|
|||
// implemtation of the Importer interface does not use, and therefore they
|
||||
// intentionally no-op.
|
||||
type onPremImporter struct {
|
||||
api *API
|
||||
api *API
|
||||
client *InternalClient
|
||||
}
|
||||
|
||||
func NewOnPremImporter(api *API) *onPremImporter {
|
||||
return &onPremImporter{
|
||||
api: api,
|
||||
api: api,
|
||||
client: api.holder.executor.client,
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -63,7 +65,25 @@ func (i *onPremImporter) ImportRoaringBitmap(ctx context.Context, tid dax.TableI
|
|||
}
|
||||
|
||||
func (i *onPremImporter) ImportRoaringShard(ctx context.Context, tid dax.TableID, shard uint64, request *ImportRoaringShardRequest) error {
|
||||
return i.api.ImportRoaringShard(ctx, string(tid), shard, request)
|
||||
nodes, err := i.api.ShardNodes(ctx, string(tid), shard)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
eg := errgroup.Group{}
|
||||
for _, node := range nodes {
|
||||
node := node
|
||||
if node.ID == i.api.NodeID() { // local
|
||||
eg.Go(func() error {
|
||||
return i.api.ImportRoaringShard(ctx, string(tid), shard, request)
|
||||
})
|
||||
} else {
|
||||
eg.Go(func() error { // forward on
|
||||
return i.client.ImportRoaringShard(ctx, &node.URI, string(tid), shard, true, request)
|
||||
})
|
||||
}
|
||||
}
|
||||
err = eg.Wait()
|
||||
return errors.Wrap(err, "importing")
|
||||
}
|
||||
|
||||
func (i *onPremImporter) EncodeImportValues(ctx context.Context, tid dax.TableID, fld *dax.Field, shard uint64, vals []int64, ids []uint64, clear bool) (path string, data []byte, err error) {
|
||||
|
|
|
|||
|
|
@ -888,7 +888,6 @@ func (c *InternalClient) Import(ctx context.Context, qcx *Qcx, req *ImportReques
|
|||
func (c *InternalClient) ImportValue(ctx context.Context, qcx *Qcx, req *ImportValueRequest, options *ImportOptions) error {
|
||||
span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.Import")
|
||||
defer span.Finish()
|
||||
|
||||
if req.ColumnKeys != nil {
|
||||
req.Shard = ^uint64(0)
|
||||
}
|
||||
|
|
@ -926,35 +925,7 @@ func (c *InternalClient) ImportRoaring(ctx context.Context, uri *pnet.URI, index
|
|||
return errors.Wrap(err, "marshal import request")
|
||||
}
|
||||
|
||||
// Generate HTTP request.
|
||||
httpReq, err := http.NewRequest("POST", url, bytes.NewBuffer(data))
|
||||
if err != nil {
|
||||
return errors.Wrap(err, "creating request")
|
||||
}
|
||||
httpReq.Header.Set("Content-Type", "application/x-protobuf")
|
||||
httpReq.Header.Set("Accept", "application/x-protobuf")
|
||||
httpReq.Header.Set("X-Pilosa-Row", "roaring")
|
||||
httpReq.Header.Set("User-Agent", "pilosa/"+Version)
|
||||
AddAuthToken(ctx, &httpReq.Header)
|
||||
|
||||
// Execute request against the host.
|
||||
resp, err := c.executeRequest(httpReq.WithContext(ctx))
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
|
||||
dec := json.NewDecoder(resp.Body)
|
||||
rbody := &ImportResponse{}
|
||||
err = dec.Decode(rbody)
|
||||
// Decode can return EOF when no error occurred. helpful!
|
||||
if err != nil && err != io.EOF {
|
||||
return errors.Wrap(err, "decoding response body")
|
||||
}
|
||||
if rbody.Err != "" {
|
||||
return errors.Wrap(errors.New(rbody.Err), "importing roaring")
|
||||
}
|
||||
return nil
|
||||
return c.executeProtobufRequest(ctx, url, data)
|
||||
}
|
||||
|
||||
// ExportCSV bulk exports data for a single shard from a host to CSV format.
|
||||
|
|
@ -2459,3 +2430,58 @@ func (c *InternalClient) getDiskUsage(ctx context.Context, index string) (DiskUs
|
|||
|
||||
return sum, nil
|
||||
}
|
||||
|
||||
func (c *InternalClient) executeProtobufRequest(ctx context.Context, url string, data []byte) error {
|
||||
httpReq, err := http.NewRequest("POST", url, bytes.NewBuffer(data))
|
||||
if err != nil {
|
||||
return errors.Wrap(err, "creating request")
|
||||
}
|
||||
httpReq.Header.Set("Content-Type", "application/x-protobuf")
|
||||
httpReq.Header.Set("Accept", "application/x-protobuf")
|
||||
httpReq.Header.Set("X-Pilosa-Row", "roaring")
|
||||
httpReq.Header.Set("User-Agent", "pilosa/"+Version)
|
||||
AddAuthToken(ctx, &httpReq.Header)
|
||||
|
||||
// Execute request against the host.
|
||||
resp, err := c.executeRequest(httpReq.WithContext(ctx))
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
|
||||
dec := json.NewDecoder(resp.Body)
|
||||
rbody := &ImportResponse{}
|
||||
err = dec.Decode(rbody)
|
||||
// Decode can return EOF when no error occurred. helpful!
|
||||
if err != nil && err != io.EOF {
|
||||
return errors.Wrap(err, "decoding response body")
|
||||
}
|
||||
if rbody.Err != "" {
|
||||
return errors.Wrap(errors.New(rbody.Err), "importing roaring")
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// ImportRoaringShard(ctx, node, string(tid), shard, request
|
||||
func (c *InternalClient) ImportRoaringShard(ctx context.Context, uri *pnet.URI, index string, shard uint64, remote bool, req *ImportRoaringShardRequest) error {
|
||||
span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.ImportRoaringShard")
|
||||
defer span.Finish()
|
||||
|
||||
if index == "" {
|
||||
return ErrIndexRequired
|
||||
}
|
||||
if uri == nil {
|
||||
uri = c.defaultURI
|
||||
}
|
||||
|
||||
vals := url.Values{}
|
||||
vals.Set("remote", strconv.FormatBool(remote))
|
||||
url := fmt.Sprintf("%s%s/index/%s/shard/%d/import-roaring?%s", uri, c.prefix(), index, shard, vals.Encode())
|
||||
|
||||
// Marshal data to protobuf.
|
||||
data, err := c.serializer.Marshal(req)
|
||||
if err != nil {
|
||||
return errors.Wrap(err, "marshal import roaring shard request")
|
||||
}
|
||||
return c.executeProtobufRequest(ctx, url, data)
|
||||
}
|
||||
|
|
|
|||
|
|
@ -5,6 +5,8 @@ import (
|
|||
"context"
|
||||
"fmt"
|
||||
"os"
|
||||
"reflect"
|
||||
"sort"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
|
@ -25,7 +27,6 @@ import (
|
|||
)
|
||||
|
||||
func TestPlanner_Misc(t *testing.T) {
|
||||
|
||||
d, err := pql.ParseDecimal("12.345678")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
|
|
@ -308,6 +309,7 @@ func TestPlanner_Show(t *testing.T) {
|
|||
}
|
||||
})
|
||||
}
|
||||
|
||||
func TestPlanner_CoverCreateTable(t *testing.T) {
|
||||
c := test.MustRunCluster(t, 1)
|
||||
defer c.Close()
|
||||
|
|
@ -354,7 +356,7 @@ func TestPlanner_CoverCreateTable(t *testing.T) {
|
|||
_, _, err := sql_test.MustQueryRows(t, server, sql)
|
||||
if assert.Error(t, err) {
|
||||
assert.Equal(t, fld.expErr, err.Error())
|
||||
//sql3.SQLErrConflictingColumnConstraint.Message
|
||||
// sql3.SQLErrConflictingColumnConstraint.Message
|
||||
}
|
||||
}
|
||||
})
|
||||
|
|
@ -528,7 +530,7 @@ func TestPlanner_CoverCreateTable(t *testing.T) {
|
|||
|
||||
schema, err := api.Schema(ctx, false)
|
||||
assert.NoError(t, err)
|
||||
//spew.Dump(schema)
|
||||
// spew.Dump(schema)
|
||||
|
||||
// Get the fields from the FeatureBase schema.
|
||||
// fbFields is a map of fieldName to FieldInfo.
|
||||
|
|
@ -765,7 +767,6 @@ func TestPlanner_AlterTable(t *testing.T) {
|
|||
t.Fatal(diff)
|
||||
}
|
||||
})
|
||||
|
||||
}
|
||||
|
||||
func TestPlanner_DropTable(t *testing.T) {
|
||||
|
|
@ -825,7 +826,8 @@ func TestPlanner_ExpressionsInSelectListParen(t *testing.T) {
|
|||
Set(1, b=100)
|
||||
Set(2, a=20)
|
||||
Set(2, b=200)
|
||||
`}); err != nil {
|
||||
`,
|
||||
}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
|
|
@ -904,7 +906,8 @@ func TestPlanner_ExpressionsInSelectListLiterals(t *testing.T) {
|
|||
Set(1, d=10.3)
|
||||
Set(1, ts='2022-02-22T22:22:22Z')
|
||||
Set(1, str='foo')
|
||||
`}); err != nil {
|
||||
`,
|
||||
}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
|
|
@ -1050,7 +1053,8 @@ func TestPlanner_ExpressionsInSelectListCase(t *testing.T) {
|
|||
Set(1, d=10.3)
|
||||
Set(1, ts='2022-02-22T22:22:22Z')
|
||||
Set(1, str='foo')
|
||||
`}); err != nil {
|
||||
`,
|
||||
}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
|
|
@ -1133,7 +1137,8 @@ func TestPlanner_Select(t *testing.T) {
|
|||
Set(1, b=100)
|
||||
Set(2, a=20)
|
||||
Set(2, b=200)
|
||||
`}); err != nil {
|
||||
`,
|
||||
}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
|
|
@ -1321,7 +1326,8 @@ func TestPlanner_SelectOrderBy(t *testing.T) {
|
|||
Set(1, b=100)
|
||||
Set(2, a=20)
|
||||
Set(2, b=200)
|
||||
`}); err != nil {
|
||||
`,
|
||||
}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
|
|
@ -1506,7 +1512,6 @@ func TestPlanner_BulkInsert(t *testing.T) {
|
|||
})
|
||||
|
||||
t.Run("BulkCSVBadMap", func(t *testing.T) {
|
||||
|
||||
_, _, err = sql_test.MustQueryRows(t, c.GetNode(0).Server, `bulk insert into j (_id, a, b) map (0 id, 1 int, 10 int) from x'1,10,20
|
||||
2,11,21
|
||||
3,12,22
|
||||
|
|
@ -1523,7 +1528,6 @@ func TestPlanner_BulkInsert(t *testing.T) {
|
|||
})
|
||||
|
||||
t.Run("BulkCSVFileDefault", func(t *testing.T) {
|
||||
|
||||
tmpfile, err := os.CreateTemp("", "BulkCSVFileDefault.*.csv")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
|
|
@ -1546,7 +1550,6 @@ func TestPlanner_BulkInsert(t *testing.T) {
|
|||
})
|
||||
|
||||
t.Run("BulkCSVFileNoColumns", func(t *testing.T) {
|
||||
|
||||
tmpfile, err := os.CreateTemp("", "BulkCSVFileNoColumns.*.csv")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
|
|
@ -1576,7 +1579,6 @@ func TestPlanner_BulkInsert(t *testing.T) {
|
|||
})
|
||||
|
||||
t.Run("BulkCSVFileRowsLimit", func(t *testing.T) {
|
||||
|
||||
tmpfile, err := os.CreateTemp("", "BulkCSVFileDefault.*.csv")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
|
|
@ -1613,7 +1615,6 @@ func TestPlanner_BulkInsert(t *testing.T) {
|
|||
}, columns); diff != "" {
|
||||
t.Fatal(diff)
|
||||
}
|
||||
|
||||
})
|
||||
|
||||
t.Run("BulkCSVBlobDefault", func(t *testing.T) {
|
||||
|
|
@ -1624,7 +1625,6 @@ func TestPlanner_BulkInsert(t *testing.T) {
|
|||
})
|
||||
|
||||
t.Run("BulkNDJsonBlobDefault", func(t *testing.T) {
|
||||
|
||||
_, _, err = sql_test.MustQueryRows(t, c.GetNode(0).Server, `bulk insert into j (_id, a, b) map ('$._id' id, '$.a' int, '$.b' int)
|
||||
from x'{ "_id": 1, "a": 10, "b": 20 }
|
||||
{ "_id": 2, "a": 10, "b": 20 }
|
||||
|
|
@ -1659,7 +1659,6 @@ func TestPlanner_BulkInsert(t *testing.T) {
|
|||
})
|
||||
|
||||
t.Run("BulkNDJsonFileDefault", func(t *testing.T) {
|
||||
|
||||
tmpfile, err := os.CreateTemp("", "BulkNDJsonFileDefault.*.csv")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
|
|
@ -1683,7 +1682,6 @@ func TestPlanner_BulkInsert(t *testing.T) {
|
|||
})
|
||||
|
||||
t.Run("BulkNDJsonFileTransform", func(t *testing.T) {
|
||||
|
||||
tmpfile, err := os.CreateTemp("", "BulkNDJsonFileTransform.*.csv")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
|
|
@ -1707,7 +1705,6 @@ func TestPlanner_BulkInsert(t *testing.T) {
|
|||
})
|
||||
|
||||
t.Run("BulkNDJsonAllTypes", func(t *testing.T) {
|
||||
|
||||
_, _, err = sql_test.MustQueryRows(t, c.GetNode(0).Server, `bulk insert
|
||||
into alltypes (_id, id1, i1, ids1, ss1, ts1, s1, b1, d1)
|
||||
|
||||
|
|
@ -1727,7 +1724,6 @@ func TestPlanner_BulkInsert(t *testing.T) {
|
|||
})
|
||||
|
||||
t.Run("BulkNDJsonBadJsonPath", func(t *testing.T) {
|
||||
|
||||
_, _, err = sql_test.MustQueryRows(t, c.GetNode(0).Server, `bulk insert
|
||||
into alltypes (_id, id1, i1, ids1, ss1, ts1, s1, b1, d1)
|
||||
|
||||
|
|
@ -1747,7 +1743,6 @@ func TestPlanner_BulkInsert(t *testing.T) {
|
|||
})
|
||||
|
||||
t.Run("BulkNDJsonBadJson", func(t *testing.T) {
|
||||
|
||||
_, _, err = sql_test.MustQueryRows(t, c.GetNode(0).Server, `bulk insert
|
||||
into alltypes (_id, id1, i1, ids1, ss1, ts1, s1, b1, d1)
|
||||
|
||||
|
|
@ -1767,7 +1762,6 @@ func TestPlanner_BulkInsert(t *testing.T) {
|
|||
})
|
||||
|
||||
t.Run("BulkInsertDecimals", func(t *testing.T) {
|
||||
|
||||
_, _, err = sql_test.MustQueryRows(t, c.GetNode(0).Server, `create table iris (
|
||||
_id id,
|
||||
sepallength decimal(2),
|
||||
|
|
@ -1817,7 +1811,6 @@ func TestPlanner_BulkInsert(t *testing.T) {
|
|||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
})
|
||||
|
||||
t.Run("BulkInsertDupeColumnPlusNullsInJson", func(t *testing.T) {
|
||||
|
|
@ -1845,7 +1838,6 @@ func TestPlanner_BulkInsert(t *testing.T) {
|
|||
})
|
||||
|
||||
t.Run("BulkInsertCSVStringIDSet", func(t *testing.T) {
|
||||
|
||||
_, _, err = sql_test.MustQueryRows(t, c.GetNode(0).Server, `create table greg-test (
|
||||
_id STRING,
|
||||
id_col ID,
|
||||
|
|
@ -1904,7 +1896,6 @@ func TestPlanner_BulkInsert(t *testing.T) {
|
|||
})
|
||||
|
||||
t.Run("BulkInsertAllowMissingValues", func(t *testing.T) {
|
||||
|
||||
_, _, err = sql_test.MustQueryRows(t, c.GetNode(0).Server, `create table greg-test-amv (
|
||||
_id STRING,
|
||||
id_col ID,
|
||||
|
|
@ -1962,7 +1953,6 @@ func TestPlanner_BulkInsert(t *testing.T) {
|
|||
}
|
||||
})
|
||||
t.Run("BulkInsertNDJSONStringIDSet", func(t *testing.T) {
|
||||
|
||||
_, _, err = sql_test.MustQueryRows(t, c.GetNode(0).Server, `create table greg-test-01 (
|
||||
_id STRING,
|
||||
id_col ID,
|
||||
|
|
@ -2079,7 +2069,8 @@ func TestPlanner_SelectSelectSource(t *testing.T) {
|
|||
Set(1, b=100)
|
||||
Set(2, a=20)
|
||||
Set(2, b=200)
|
||||
`}); err != nil {
|
||||
`,
|
||||
}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
|
|
@ -2159,7 +2150,8 @@ func TestPlanner_In(t *testing.T) {
|
|||
Set(1, a=10)
|
||||
Set(2, a=20)
|
||||
Set(3, a=30)
|
||||
`}); err != nil {
|
||||
`,
|
||||
}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
|
|
@ -2174,15 +2166,16 @@ func TestPlanner_In(t *testing.T) {
|
|||
|
||||
Set(3, parentid=2)
|
||||
Set(3, x=300)
|
||||
`}); err != nil {
|
||||
`,
|
||||
}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
t.Run("Count", func(t *testing.T) {
|
||||
t.Skip("Need to add join conditions to get this to pass")
|
||||
results, columns, err := sql_test.MustQueryRows(t, c.GetNode(0).Server, fmt.Sprintf(`SELECT %j._id, %j.a, %k._id, %k.parentid, %k.x FROM %j INNER JOIN %k ON %j._id = %k.parentid`, c, c, c, c, c, c, c, c, c))
|
||||
//results, columns, err := sql_test.MustQueryRows(t, c.GetNode(0).Server, fmt.Sprintf(`SELECT COUNT(*) FROM %j INNER JOIN %k ON %j._id = %k.parentid`, c, c, c, c))
|
||||
//results, columns, err := sql_test.MustQueryRows(t, c.GetNode(0).Server, fmt.Sprintf(`SELECT a FROM %j where a = 20`, c)) // SELECT COUNT(*) FROM %j INNER JOIN %k ON %j._id = %k.parentid
|
||||
// results, columns, err := sql_test.MustQueryRows(t, c.GetNode(0).Server, fmt.Sprintf(`SELECT COUNT(*) FROM %j INNER JOIN %k ON %j._id = %k.parentid`, c, c, c, c))
|
||||
// results, columns, err := sql_test.MustQueryRows(t, c.GetNode(0).Server, fmt.Sprintf(`SELECT a FROM %j where a = 20`, c)) // SELECT COUNT(*) FROM %j INNER JOIN %k ON %j._id = %k.parentid
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
|
@ -2290,7 +2283,8 @@ func TestPlanner_Distinct(t *testing.T) {
|
|||
Set(1, a=10)
|
||||
Set(2, a=20)
|
||||
Set(3, a=30)
|
||||
`}); err != nil {
|
||||
`,
|
||||
}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
|
|
@ -2305,7 +2299,8 @@ func TestPlanner_Distinct(t *testing.T) {
|
|||
|
||||
Set(3, parentid=2)
|
||||
Set(3, x=300)
|
||||
`}); err != nil {
|
||||
`,
|
||||
}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
|
|
@ -2402,7 +2397,8 @@ func TestPlanner_SelectTop(t *testing.T) {
|
|||
Set(1, b=100)
|
||||
Set(2, b=200)
|
||||
Set(3, b=300)
|
||||
`}); err != nil {
|
||||
`,
|
||||
}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
|
|
@ -2460,6 +2456,7 @@ func wireQueryFieldID(name string) *pilosa.WireQueryField {
|
|||
BaseType: dax.BaseTypeID,
|
||||
}
|
||||
}
|
||||
|
||||
func wireQueryFieldBool(name string) *pilosa.WireQueryField {
|
||||
return &pilosa.WireQueryField{
|
||||
Name: dax.FieldName(name),
|
||||
|
|
@ -2467,6 +2464,7 @@ func wireQueryFieldBool(name string) *pilosa.WireQueryField {
|
|||
BaseType: dax.BaseTypeBool,
|
||||
}
|
||||
}
|
||||
|
||||
func wireQueryFieldString(name string) *pilosa.WireQueryField {
|
||||
return &pilosa.WireQueryField{
|
||||
Name: dax.FieldName(name),
|
||||
|
|
@ -2474,6 +2472,7 @@ func wireQueryFieldString(name string) *pilosa.WireQueryField {
|
|||
BaseType: dax.BaseTypeString,
|
||||
}
|
||||
}
|
||||
|
||||
func wireQueryFieldInt(name string) *pilosa.WireQueryField {
|
||||
return &pilosa.WireQueryField{
|
||||
Name: dax.FieldName(name),
|
||||
|
|
@ -2481,6 +2480,7 @@ func wireQueryFieldInt(name string) *pilosa.WireQueryField {
|
|||
BaseType: dax.BaseTypeInt,
|
||||
}
|
||||
}
|
||||
|
||||
func wireQueryFieldTimestamp(name string) *pilosa.WireQueryField {
|
||||
return &pilosa.WireQueryField{
|
||||
Name: dax.FieldName(name),
|
||||
|
|
@ -2488,6 +2488,7 @@ func wireQueryFieldTimestamp(name string) *pilosa.WireQueryField {
|
|||
BaseType: dax.BaseTypeTimestamp,
|
||||
}
|
||||
}
|
||||
|
||||
func wireQueryFieldDecimal(name string, scale int64) *pilosa.WireQueryField {
|
||||
return &pilosa.WireQueryField{
|
||||
Name: dax.FieldName(name),
|
||||
|
|
@ -2498,3 +2499,102 @@ func wireQueryFieldDecimal(name string, scale int64) *pilosa.WireQueryField {
|
|||
},
|
||||
}
|
||||
}
|
||||
|
||||
// This test verifies that data sent to all nodes shows up in the results
|
||||
func TestPlanner_BulkInsert_FB1831(t *testing.T) {
|
||||
c := test.MustRunCluster(t, 3)
|
||||
defer c.Close()
|
||||
|
||||
_, _, err := sql_test.MustQueryRows(t, c.GetNode(0).Server, `create table iris (_id id, sepallength decimal(2), sepalwidth decimal(2), petallength decimal(2), petalwidth decimal(2), species string cachetype ranked size 1000);`)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
_, _, err = sql_test.MustQueryRows(t, c.GetNode(0).Server, `bulk insert
|
||||
into iris (_id, sepallength, sepalwidth, petallength, petalwidth, species)
|
||||
map('id' id,
|
||||
'sepalLength' DECIMAL(2),
|
||||
'sepalWidth' DECIMAL(2),
|
||||
'petalLength' DECIMAL(2),
|
||||
'petalWidth' DECIMAL(2),
|
||||
'species' STRING)
|
||||
from
|
||||
x'{"id": 1, "sepalLength": "5.1", "sepalWidth": "3.5", "petalLength": "1.4", "petalWidth": "0.2", "species": "setosa"}
|
||||
{"id": 2, "sepalLength": "4.9", "sepalWidth": "3.0", "petalLength": "1.4", "petalWidth": "0.2", "species": "setosa"}
|
||||
{"id": 3, "sepalLength": "4.7", "sepalWidth": "3.2", "petalLength": "1.3", "petalWidth": "0.2", "species": "setosa"}'
|
||||
with
|
||||
format 'NDJSON'
|
||||
input 'STREAM';`)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
_, _, err = sql_test.MustQueryRows(t, c.GetNode(1).Server, `bulk insert
|
||||
into iris (_id, sepallength, sepalwidth, petallength, petalwidth, species)
|
||||
map('id' id,
|
||||
'sepalLength' DECIMAL(2),
|
||||
'sepalWidth' DECIMAL(2),
|
||||
'petalLength' DECIMAL(2),
|
||||
'petalWidth' DECIMAL(2),
|
||||
'species' STRING)
|
||||
from
|
||||
x'{"id": 4, "sepalLength": "5.1", "sepalWidth": "3.5", "petalLength": "1.4", "petalWidth": "0.2", "species": "setosa"}
|
||||
{"id": 5, "sepalLength": "4.9", "sepalWidth": "3.0", "petalLength": "1.4", "petalWidth": "0.2", "species": "setosa"}
|
||||
{"id": 6, "sepalLength": "4.7", "sepalWidth": "3.2", "petalLength": "1.3", "petalWidth": "0.2", "species": "setosa"}'
|
||||
with
|
||||
format 'NDJSON'
|
||||
input 'STREAM';`)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
_, _, err = sql_test.MustQueryRows(t, c.GetNode(2).Server, `bulk insert
|
||||
into iris (_id, sepallength, sepalwidth, petallength, petalwidth, species)
|
||||
map('id' id,
|
||||
'sepalLength' DECIMAL(2),
|
||||
'sepalWidth' DECIMAL(2),
|
||||
'petalLength' DECIMAL(2),
|
||||
'petalWidth' DECIMAL(2),
|
||||
'species' STRING)
|
||||
from
|
||||
x'{"id": 7, "sepalLength": "5.1", "sepalWidth": "3.5", "petalLength": "1.4", "petalWidth": "0.2", "species": "setosa"}
|
||||
{"id": 8, "sepalLength": "4.9", "sepalWidth": "3.0", "petalLength": "1.4", "petalWidth": "0.2", "species": "setosa"}
|
||||
{"id": 9, "sepalLength": "4.7", "sepalWidth": "3.2", "petalLength": "1.3", "petalWidth": "0.2", "species": "setosa"}'
|
||||
with
|
||||
format 'NDJSON'
|
||||
input 'STREAM';`)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
_, _, err = sql_test.MustQueryRows(t, c.GetNode(0).Server, `bulk insert
|
||||
into iris (_id, sepallength, sepalwidth, petallength, petalwidth, species)
|
||||
map('id' id,
|
||||
'sepalLength' DECIMAL(2),
|
||||
'sepalWidth' DECIMAL(2),
|
||||
'petalLength' DECIMAL(2),
|
||||
'petalWidth' DECIMAL(2),
|
||||
'species' STRING)
|
||||
from
|
||||
x'{"id": 1048577, "sepalLength": "5.1", "sepalWidth": "3.5", "petalLength": "1.4", "petalWidth": "0.2", "species": "setosa"}
|
||||
{"id": 2097153, "sepalLength": "4.9", "sepalWidth": "3.0", "petalLength": "1.4", "petalWidth": "0.2", "species": "setosa"}
|
||||
{"id": 3145729, "sepalLength": "4.7", "sepalWidth": "3.2", "petalLength": "1.3", "petalWidth": "0.2", "species": "setosa"}'
|
||||
with
|
||||
format 'NDJSON'
|
||||
input 'STREAM';`)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
results, _, err := sql_test.MustQueryRows(t, c.GetNode(0).Server, `select _id from iris`)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
got := make([]int64, 0)
|
||||
for i := range results {
|
||||
got = append(got, results[i][0].(int64))
|
||||
}
|
||||
sort.Slice(got, func(i, j int) bool {
|
||||
return got[i] < got[j]
|
||||
})
|
||||
expected := []int64{1, 2, 3, 4, 5, 6, 7, 8, 9, 1048577, 2097153, 3145729}
|
||||
if !reflect.DeepEqual(got, expected) {
|
||||
t.Fatal("Expecting to be equal")
|
||||
}
|
||||
}
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue