diff --git a/api.go b/api.go index 42d7e3b0d..65fe6d8db 100644 --- a/api.go +++ b/api.go @@ -1454,13 +1454,6 @@ func OptImportOptionsPresorted(b bool) ImportOption { } } -func optImportOptionsFullySorted(b bool) ImportOption { - return func(o *ImportOptions) error { - o.fullySorted = b - return nil - } -} - var ErrAborted = fmt.Errorf("error: update was aborted") func (api *API) ImportAtomicRecord(ctx context.Context, qcx *Qcx, req *AtomicRecord, opts ...ImportOption) error { @@ -1491,14 +1484,20 @@ func (api *API) ImportAtomicRecord(ctx context.Context, qcx *Qcx, req *AtomicRec qcx.StartAtomicWriteTx(Txo{Write: writable, Index: idx, Shard: req.Shard}) tot := 0 + options, err := setUpImportOptions(opts...) + if err != nil { + return errors.Wrap(err, "setting up import options") + } + // BSIs (Values) for _, ivr := range req.Ivr { tot++ if simPowerLoss && tot > lossAfter { return ErrAborted } - opts0 := append(opts, OptImportOptionsClear(ivr.Clear)) - err := api.ImportValueWithTx(ctx, qcx, ivr, opts0...) + subOpts := *options + subOpts.Clear = ivr.Clear + err = api.ImportValueWithTx(ctx, qcx, ivr, &subOpts) if err != nil { return errors.Wrap(err, "ImportAtomicRecord ImportValueWithTx") } @@ -1510,8 +1509,9 @@ func (api *API) ImportAtomicRecord(ctx context.Context, qcx *Qcx, req *AtomicRec if simPowerLoss && tot > lossAfter { return ErrAborted } - opts0 := append(opts, OptImportOptionsClear(ir.Clear)) - err := api.ImportWithTx(ctx, qcx, ir, opts0...) + subOpts := *options + subOpts.Clear = ir.Clear + err := api.ImportWithTx(ctx, qcx, ir, &subOpts) if err != nil { return errors.Wrap(err, "ImportAtomicRecord ImportWithTx") } @@ -1539,7 +1539,12 @@ func (api *API) Import(ctx context.Context, qcx *Qcx, req *ImportRequest, opts . if req.Clear { opts = addClearToImportOptions(opts) } - err = api.ImportWithTx(ctx, qcx, req, opts...) + // Set up import options. + options, err := setUpImportOptions(opts...) + if err != nil { + return errors.Wrap(err, "setting up import options") + } + err = api.ImportWithTx(ctx, qcx, req, options) if err != nil { return err } @@ -1547,7 +1552,7 @@ func (api *API) Import(ctx context.Context, qcx *Qcx, req *ImportRequest, opts . } // ImportWithTx bulk imports data into a particular index,field,shard. -func (api *API) ImportWithTx(ctx context.Context, qcx *Qcx, req *ImportRequest, opts ...ImportOption) error { +func (api *API) ImportWithTx(ctx context.Context, qcx *Qcx, req *ImportRequest, options *ImportOptions) error { span, _ := tracing.StartSpanFromContext(ctx, "API.Import") defer span.Finish() @@ -1564,11 +1569,6 @@ func (api *API) ImportWithTx(ctx context.Context, qcx *Qcx, req *ImportRequest, return errors.Wrap(err, "validating import value request") } - // Set up import options. - options, err := setUpImportOptions(opts...) - if err != nil { - return errors.Wrap(err, "setting up import options") - } span.LogKV( "index", req.Index, "field", req.Field) @@ -1586,6 +1586,8 @@ func (api *API) ImportWithTx(ctx context.Context, qcx *Qcx, req *ImportRequest, if req.RowIDs, err = api.cluster.translateFieldKeys(ctx, field, req.RowKeys, true); err != nil { return errors.Wrapf(err, "translating field keys") } + } else if len(req.RowKeys) != 0 { + return errors.New("value keys cannot be used because field uses integer IDs") } // Translate column keys. @@ -1597,44 +1599,36 @@ func (api *API) ImportWithTx(ctx context.Context, qcx *Qcx, req *ImportRequest, if req.ColumnIDs, err = api.cluster.translateIndexKeys(ctx, req.Index, req.ColumnKeys, true); err != nil { return errors.Wrap(err, "translating columns") } - } - - // For translated data, map the columnIDs to shards. If - // this node does not own the shard, forward to the node that does. - if idx.Keys() || field.Keys() { - m := make(map[uint64][]Bit) - - for i, colID := range req.ColumnIDs { - shard := colID / ShardWidth - if _, ok := m[shard]; !ok { - m[shard] = make([]Bit, 0) - } - bit := Bit{ - RowID: req.RowIDs[i], - ColumnID: colID, - } - if len(req.Timestamps) > 0 { - bit.Timestamp = req.Timestamps[i] - } - m[shard] = append(m[shard], bit) - } - - // Signal to the receiving nodes to ignore checking for key translation. - opts = append(opts, OptImportOptionsIgnoreKeyCheck(true)) - - var eg errgroup.Group - for shard, bits := range m { - // TODO: if local node owns this shard we don't need to go through the client - shard := shard - bits := bits - eg.Go(func() error { - return api.server.defaultClient.Import(ctx, req.Index, req.Field, shard, bits, opts...) - }) - } - return eg.Wait() + // mark this request as having an unknown shard, meaning it will + // be sorted and served out to multiple nodes. + req.Shard = ^uint64(0) + } else if len(req.ColumnKeys) != 0 { + return errors.New("record keys cannot be used because field uses integer IDs") } } + // if you specify a shard of ^0, we try to split this out. If we did any + // key translation, we set it to ^0 already above. + if req.Shard == ^uint64(0) { + reqs := req.SortToShards() + + // Signal to the receiving nodes to ignore checking for key translation. + options.IgnoreKeyCheck = true + + var eg errgroup.Group + for _, subReq := range reqs { + // TODO: if local node owns this shard we don't need to go through the client + subReq := subReq + eg.Go(func() error { + return api.server.defaultClient.Import(ctx, qcx, subReq, options) + }) + } + return eg.Wait() + } + + // otherwise, this has to be a shard that we have, and everything has + // to be for that shard. + // Validate shard ownership. if err := api.validateShardOwnership(req.Index, req.Shard); err != nil { return errors.Wrap(err, "validating shard ownership") @@ -1649,8 +1643,6 @@ func (api *API) ImportWithTx(ctx context.Context, qcx *Qcx, req *ImportRequest, } // Import columnIDs into existence field. - // Note: req.Shard may not be the only shard imported into here, - // so don't expect it to be invariant. if !options.Clear { if err := importExistenceColumns(qcx, idx, req.ColumnIDs, req.Shard); err != nil { api.server.logger.Errorf("import existence error: index=%s, field=%s, shard=%d, columns=%d, err=%s", req.Index, req.Field, req.Shard, len(req.ColumnIDs), err) @@ -1662,7 +1654,7 @@ func (api *API) ImportWithTx(ctx context.Context, qcx *Qcx, req *ImportRequest, } // Import into fragment. - err = field.Import(qcx, req.RowIDs, req.ColumnIDs, timestamps, req.Shard, opts...) + err = field.Import(qcx, req.RowIDs, req.ColumnIDs, timestamps, req.Shard, options) if err != nil { api.server.logger.Errorf("import error: index=%s, field=%s, shard=%d, columns=%d, err=%s", req.Index, req.Field, req.Shard, len(req.ColumnIDs), err) return errors.Wrap(err, "importing") @@ -1676,11 +1668,16 @@ func (api *API) ImportValue(ctx context.Context, qcx *Qcx, req *ImportValueReque if req.Clear { opts = addClearToImportOptions(opts) } - return api.ImportValueWithTx(ctx, qcx, req, opts...) + // Set up import options. + options, err := setUpImportOptions(opts...) + if err != nil { + return errors.Wrap(err, "setting up import options") + } + return api.ImportValueWithTx(ctx, qcx, req, options) } // ImportValueWithTx bulk imports values into a particular field. -func (api *API) ImportValueWithTx(ctx context.Context, qcx *Qcx, req *ImportValueRequest, opts ...ImportOption) (err0 error) { +func (api *API) ImportValueWithTx(ctx context.Context, qcx *Qcx, req *ImportValueRequest, options *ImportOptions) (err0 error) { span, _ := tracing.StartSpanFromContext(ctx, "API.ImportValue") defer span.Finish() @@ -1706,12 +1703,6 @@ func (api *API) ImportValueWithTx(ctx context.Context, qcx *Qcx, req *ImportValu return errors.Wrap(err, "validating import value request") } - // Set up import options. - options, err := setUpImportOptions(opts...) - if err != nil { - return errors.Wrap(err, "setting up import options") - } - idx, field, err = api.indexField(req.Index, req.Field, req.Shard) if err != nil { return errors.Wrap(err, "getting index and field") @@ -1826,7 +1817,7 @@ func (api *API) ImportValueWithTx(ctx context.Context, qcx *Qcx, req *ImportValu } eg.Go(func() error { - return api.server.defaultClient.ImportValue2(ctx, subreq, options) + return api.server.defaultClient.ImportValue(ctx, qcx, subreq, options) }) start = i shard = colID / ShardWidth @@ -1847,7 +1838,7 @@ func (api *API) ImportValueWithTx(ctx context.Context, qcx *Qcx, req *ImportValu // TODO we should elevate the logic for figuring out which // node(s) to send to into API instead of having those details // in the client implementation. - return api.server.defaultClient.ImportValue2(ctx, subreq, options) + return api.server.defaultClient.ImportValue(ctx, qcx, subreq, options) }) err = eg.Wait() if err != nil { @@ -2018,7 +2009,6 @@ func (api *API) applyOperations(ctx context.Context, qcx *Qcx, index *Index, sha // For each operation, we may have a set of records/fields to clear, and then // also a set of fields to set/remove specific bits in. opts := &ImportOptions{Presorted: true, IgnoreKeyCheck: true, fullySorted: true} - funcOpts := []ImportOption{OptImportOptionsIgnoreKeyCheck(true), OptImportOptionsPresorted(true), optImportOptionsFullySorted(true), OptImportOptionsClear(true)} for _, op := range ops { // ClearRecordIDs should exist only for delete, clear, and write. For clear and write, // we'll have a list of fields, for delete, it should be all the fields. @@ -2067,12 +2057,6 @@ func (api *API) applyOperations(ctx context.Context, qcx *Qcx, index *Index, sha } } opts.Clear = (op.OpType == ingest.OpRemove) - // reslice this rather than regenerating it. i'm so efficient. - if opts.Clear { - funcOpts = funcOpts[:4] - } else { - funcOpts = funcOpts[:3] - } // for "set" and "write" ops, we'll be setting bits, for // "remove" ops we'll be clearing them, and for "clear" ops // there shouldn't be anything here. @@ -2088,7 +2072,7 @@ func (api *API) applyOperations(ctx context.Context, qcx *Qcx, index *Index, sha } switch field.Type() { case "set", "time", "mutex", "bool": - err = field.Import(qcx, fieldOp.Values, fieldOp.RecordIDs, fieldOp.Signed, shard, funcOpts...) + err = field.Import(qcx, fieldOp.Values, fieldOp.RecordIDs, fieldOp.Signed, shard, opts) case "int", "timestamp", "decimal": err = field.importValue(qcx, fieldOp.RecordIDs, fieldOp.Signed, shard, opts) default: @@ -2115,7 +2099,8 @@ func importExistenceColumns(qcx *Qcx, index *Index, columnIDs []uint64, shard ui // existence field write. columnCopy := make([]uint64, len(columnIDs)) copy(columnCopy, columnIDs) - return ef.Import(qcx, existenceRowIDs, columnCopy, nil, shard) + options := ImportOptions{} + return ef.Import(qcx, existenceRowIDs, columnCopy, nil, shard, &options) } func clearExistenceColumns(qcx *Qcx, index *Index, columnIDs []uint64, shard uint64) error { @@ -2131,7 +2116,8 @@ func clearExistenceColumns(qcx *Qcx, index *Index, columnIDs []uint64, shard uin // existence field write. columnCopy := make([]uint64, len(columnIDs)) copy(columnCopy, columnIDs) - return ef.Import(qcx, existenceRowIDs, columnCopy, nil, shard, OptImportOptionsClear(true)) + options := ImportOptions{Clear: true} + return ef.Import(qcx, existenceRowIDs, columnCopy, nil, shard, &options) } // ShardDistribution returns an object representing the distribution of shards diff --git a/api_test.go b/api_test.go index 08e1b49e5..e5989ccb4 100644 --- a/api_test.go +++ b/api_test.go @@ -22,6 +22,7 @@ import ( "math" "math/rand" "reflect" + "sort" "strings" "testing" "time" @@ -64,55 +65,53 @@ func TestAPI_Import(t *testing.T) { m0 := c.GetNode(0) m1 := c.GetNode(1) - t.Run("RowIDColumnKey", func(t *testing.T) { - ctx := context.Background() - indexName := "rick" - fieldName := "f" + indexNames := map[bool]string{false: "i", true: "ki"} + fieldNames := map[bool]string{false: "f", true: "kf"} - index, err := m0.API.CreateIndex(ctx, indexName, pilosa.IndexOptions{Keys: true, TrackExistence: true}) + ctx := context.Background() + for ik, indexName := range indexNames { + _, err := m0.API.CreateIndex(ctx, indexName, pilosa.IndexOptions{Keys: ik, TrackExistence: true}) if err != nil { t.Fatalf("creating index: %v", err) } - if index.CreatedAt() == 0 { - t.Fatal("index createdAt is empty") + for fk, fieldName := range fieldNames { + if fk { + _, err = m0.API.CreateField(ctx, indexName, fieldName, pilosa.OptFieldTypeSet(pilosa.DefaultCacheType, 100), pilosa.OptFieldKeys()) + } else { + _, err = m0.API.CreateField(ctx, indexName, fieldName, pilosa.OptFieldTypeSet(pilosa.DefaultCacheType, 100)) + + } + if err != nil { + t.Fatalf("creating field: %v", err) + } } + } - field, err := m0.API.CreateField(ctx, indexName, fieldName, pilosa.OptFieldTypeSet(pilosa.DefaultCacheType, 100)) - if err != nil { - t.Fatalf("creating field: %v", err) - } - if field.CreatedAt() == 0 { - t.Fatal("field createdAt is empty") - } + N := 10 - rowID := uint64(1) - timestamp := int64(0) - - // Generate some keyed records. - rowIDs := []uint64{} - timestamps := []int64{} - N := 10 - for i := 1; i <= N; i++ { - rowIDs = append(rowIDs, rowID) - timestamps = append(timestamps, timestamp) - } - - // Keys are sharded so ordering is not guaranteed. - colKeys := []string{"col10", "col8", "col9", "col6", "col7", "col4", "col5", "col2", "col3", "col1"} - - colKeys = colKeys[:N] + // Keys are sharded so ordering is not guaranteed. + colKeys := make([]string, N) + rowKeys := make([]string, N) + rowIDs := make([]uint64, N) + colIDs := make([]uint64, N) + for i := range colKeys { + colKeys[i] = fmt.Sprintf("col%d", i) + rowKeys[i] = fmt.Sprintf("row%d", i) + colIDs[i] = (uint64(i) + 1) * 3 + rowIDs[i] = 1 + } + sort.Strings(colKeys) + sort.Strings(rowKeys) + t.Run("RowIDColumnKey", func(t *testing.T) { // Import data with keys to the primary and verify that it gets // translated and forwarded to the owner of shard 0 (node1; because of offsetModHasher) req := &pilosa.ImportRequest{ - Index: indexName, - IndexCreatedAt: index.CreatedAt(), - Field: fieldName, - FieldCreatedAt: field.CreatedAt(), - Shard: 0, // import is all on shard 0, why are we making lots of other shards? b/c this is not a restriction. - RowIDs: rowIDs, - ColumnKeys: colKeys, - Timestamps: timestamps, + Index: indexNames[true], + Field: fieldNames[false], + Shard: 0, // inaccurate, but keys override it + RowIDs: rowIDs, + ColumnKeys: colKeys, } qcx := m0.API.Txf().NewQcx() @@ -122,20 +121,30 @@ func TestAPI_Import(t *testing.T) { } PanicOn(qcx.Finish()) - pql := fmt.Sprintf("Row(%s=%d)", fieldName, rowID) + pql := fmt.Sprintf("Row(%s=%d)", fieldNames[false], rowIDs[0]) // Query node0. - if res, err := m0.API.Query(ctx, &pilosa.QueryRequest{Index: indexName, Query: pql}); err != nil { + var keys []string + if res, err := m0.API.Query(ctx, &pilosa.QueryRequest{Index: indexNames[true], Query: pql}); err != nil { t.Fatal(err) - } else if keys := res.Results[0].(*pilosa.Row).Keys; !reflect.DeepEqual(keys, colKeys) { + } else { + keys = res.Results[0].(*pilosa.Row).Keys + } + sort.Strings(keys) + if !reflect.DeepEqual(keys, colKeys) { t.Fatalf("expected colKeys='%#v'; observed column keys: %#v", colKeys, keys) } // Query node1. if err := test.RetryUntil(5*time.Second, func() error { - if res, err := m1.API.Query(ctx, &pilosa.QueryRequest{Index: indexName, Query: pql}); err != nil { + if res, err := m1.API.Query(ctx, &pilosa.QueryRequest{Index: indexNames[true], Query: pql}); err != nil { return err - } else if keys := res.Results[0].(*pilosa.Row).Keys; !reflect.DeepEqual(keys, colKeys) { + } else { + keys = res.Results[0].(*pilosa.Row).Keys + + } + sort.Strings(keys) + if !reflect.DeepEqual(keys, colKeys) { return fmt.Errorf("unexpected column keys: %#v", keys) } return nil @@ -143,6 +152,57 @@ func TestAPI_Import(t *testing.T) { t.Fatal(err) } }) + t.Run("ExpectedErrors", func(t *testing.T) { + ctx := context.Background() + for ik, indexName := range indexNames { + for fk, fieldName := range fieldNames { + req := pilosa.ImportRequest{ + Index: indexName, + Field: fieldName, + Shard: 0, + } + for rik := range indexNames { + if rik { + req.ColumnKeys = colKeys + req.ColumnIDs = nil + } else { + req.ColumnKeys = nil + req.ColumnIDs = colIDs + } + for rfk := range fieldNames { + if rfk { + req.RowKeys = rowKeys + req.RowIDs = nil + } else { + req.RowKeys = nil + req.RowIDs = rowIDs + } + err := func() error { + qcx := m0.API.Txf().NewQcx() + defer qcx.Abort() + err := m0.API.Import(ctx, qcx, req.Clone()) + e2 := qcx.Finish() + if e2 != nil { + t.Fatalf("unexpected error committing: %v", e2) + } + return err + }() + if err != nil { + if rfk == fk && rik == ik { + t.Errorf("unexpected error: schema keys %t/%t, req keys %t/%t: %v", + ik, fk, rik, rfk, err) + } + } else { + if rfk != fk || rik != ik { + t.Errorf("unexpected no error: schema keys %t/%t, req keys %t/%t, req %#v", + ik, fk, rik, rfk, req) + } + } + } + } + } + } + }) // Relies on the previous test creating an index with TrackExistence and // adding some data. @@ -221,6 +281,7 @@ func TestAPI_ImportValue(t *testing.T) { Field: field, ColumnKeys: colKeys, Values: values, + Shard: 0, // inaccurate but keys override it } qcx := coord.API.Txf().NewQcx() diff --git a/client.go b/client.go index 8726c440a..c35a67895 100644 --- a/client.go +++ b/client.go @@ -63,14 +63,11 @@ type InternalClient interface { PartitionNodes(ctx context.Context, partitionID int) ([]*topology.Node, error) Nodes(ctx context.Context) ([]*topology.Node, error) Query(ctx context.Context, index string, queryRequest *QueryRequest) (*QueryResponse, error) - Import(ctx context.Context, index, field string, shard uint64, bits []Bit, opts ...ImportOption) error - ImportK(ctx context.Context, index, field string, bits []Bit, opts ...ImportOption) error + Import(ctx context.Context, qcx *Qcx, req *ImportRequest, options *ImportOptions) error EnsureIndex(ctx context.Context, name string, options IndexOptions) error EnsureField(ctx context.Context, indexName string, fieldName string) error EnsureFieldWithOptions(ctx context.Context, index, field string, opt FieldOptions) error - ImportValue(ctx context.Context, index, field string, shard uint64, vals []FieldValue, opts ...ImportOption) error - ImportValueK(ctx context.Context, index, field string, vals []FieldValue, opts ...ImportOption) error - ImportValue2(ctx context.Context, req *ImportValueRequest, options *ImportOptions) error + ImportValue(ctx context.Context, qcx *Qcx, req *ImportValueRequest, options *ImportOptions) error ExportCSV(ctx context.Context, index, field string, shard uint64, w io.Writer) error CreateField(ctx context.Context, index, field string) error CreateFieldWithOptions(ctx context.Context, index, field string, opt FieldOptions) error @@ -98,6 +95,10 @@ type InternalClient interface { ImportFieldKeys(ctx context.Context, uri *pnet.URI, index, field string, remote bool, rddbdata io.Reader) error ImportIndexKeys(ctx context.Context, uri *pnet.URI, index string, partitionID int, remote bool, rddbdata io.Reader) error + + // SetInternalAPI tells the client the API it should use for internal/loopback ops + // where applicable. + SetInternalAPI(api *API) } //=============== @@ -202,13 +203,10 @@ func (n nopInternalClient) Nodes(ctx context.Context) ([]*topology.Node, error) func (n nopInternalClient) Query(ctx context.Context, index string, queryRequest *QueryRequest) (*QueryResponse, error) { return nil, nil } -func (n nopInternalClient) Import(ctx context.Context, index, field string, shard uint64, bits []Bit, opts ...ImportOption) error { +func (n nopInternalClient) Import(ctx context.Context, qcx *Qcx, req *ImportRequest, options *ImportOptions) error { return nil } -func (n nopInternalClient) ImportK(ctx context.Context, index, field string, bits []Bit, opts ...ImportOption) error { - return nil -} -func (n nopInternalClient) ImportValue2(ctx context.Context, req *ImportValueRequest, options *ImportOptions) error { +func (n nopInternalClient) ImportValue(ctx context.Context, qcx *Qcx, req *ImportValueRequest, options *ImportOptions) error { return nil } @@ -249,12 +247,6 @@ func (n nopInternalClient) EnsureField(ctx context.Context, indexName string, fi func (n nopInternalClient) EnsureFieldWithOptions(ctx context.Context, index, field string, opt FieldOptions) error { return nil } -func (n nopInternalClient) ImportValue(ctx context.Context, index, field string, shard uint64, vals []FieldValue, opts ...ImportOption) error { - return nil -} -func (n nopInternalClient) ImportValueK(ctx context.Context, index, field string, vals []FieldValue, opts ...ImportOption) error { - return nil -} func (n nopInternalClient) ExportCSV(ctx context.Context, index, field string, shard uint64, w io.Writer) error { return nil } @@ -305,3 +297,6 @@ func (c nopInternalClient) ImportFieldKeys(ctx context.Context, uri *pnet.URI, i func (c nopInternalClient) ImportIndexKeys(ctx context.Context, uri *pnet.URI, index string, partitionID int, remote bool, rddbdata io.Reader) error { return nil } + +func (c nopInternalClient) SetInternalAPI(api *API) { +} diff --git a/ctl/import.go b/ctl/import.go index f53ee8e19..a616bbb67 100644 --- a/ctl/import.go +++ b/ctl/import.go @@ -19,15 +19,12 @@ import ( "encoding/csv" "fmt" "io" - "log" "math" "os" - "sort" "strconv" "time" "github.com/molecula/featurebase/v2" - "github.com/molecula/featurebase/v2/http" "github.com/molecula/featurebase/v2/pql" "github.com/molecula/featurebase/v2/server" "github.com/pkg/errors" @@ -175,8 +172,6 @@ func (cmd *ImportCommand) importPath(ctx context.Context, fieldType string, useC // bufferBits buffers slices of bits to be imported as a batch. func (cmd *ImportCommand) bufferBits(ctx context.Context, useColumnKeys, useRowKeys bool, path string) error { - a := make([]pilosa.Bit, 0, cmd.BufferSize) - var r *csv.Reader if path != "-" { @@ -195,6 +190,13 @@ func (cmd *ImportCommand) bufferBits(ctx context.Context, useColumnKeys, useRowK r.FieldsPerRecord = -1 rnum := 0 + req := &pilosa.ImportRequest{ + Index: cmd.Index, + Field: cmd.Field, + Shard: ^uint64(0), + } + batchRecs := 0 // records in this batch + lastTime := 0 // last record that had a timestamp for { rnum++ @@ -213,25 +215,28 @@ func (cmd *ImportCommand) bufferBits(ctx context.Context, useColumnKeys, useRowK return fmt.Errorf("bad column count on row %d: col=%d", rnum, len(record)) } - var bit pilosa.Bit - // Parse row id. if useRowKeys { - bit.RowKey = record[0] + req.RowKeys = append(req.RowKeys, record[0]) } else { - if bit.RowID, err = strconv.ParseUint(record[0], 10, 64); err != nil { + var id uint64 + if id, err = strconv.ParseUint(record[0], 10, 64); err != nil { return fmt.Errorf("invalid row id on row %d: %q", rnum, record[0]) } + req.RowIDs = append(req.RowIDs, id) } // Parse column id. if useColumnKeys { - bit.ColumnKey = record[1] + req.ColumnKeys = append(req.ColumnKeys, record[1]) } else { - if bit.ColumnID, err = strconv.ParseUint(record[1], 10, 64); err != nil { + var id uint64 + if id, err = strconv.ParseUint(record[1], 10, 64); err != nil { return fmt.Errorf("invalid column id on row %d: %q", rnum, record[1]) } + req.ColumnIDs = append(req.ColumnIDs, id) } + batchRecs++ // Parse time, if exists. if len(record) > 2 && record[2] != "" { @@ -239,54 +244,46 @@ func (cmd *ImportCommand) bufferBits(ctx context.Context, useColumnKeys, useRowK if err != nil { return fmt.Errorf("invalid timestamp on row %d: %q", rnum, record[2]) } - bit.Timestamp = t.UnixNano() + if lastTime < batchRecs { + req.Timestamps = append(req.Timestamps, make([]int64, batchRecs-lastTime)...) + } + req.Timestamps = append(req.Timestamps, t.UnixNano()) + lastTime = batchRecs } - a = append(a, bit) - // If we've reached the buffer size then import bits. - if len(a) == cmd.BufferSize { - if err := cmd.importBits(ctx, useColumnKeys, useRowKeys, a); err != nil { + if batchRecs == cmd.BufferSize { + // pad timestamps out with 0s + if lastTime > 0 && lastTime < batchRecs { + req.Timestamps = append(req.Timestamps, make([]int64, batchRecs-lastTime)...) + } + if err := cmd.importBits(ctx, req); err != nil { return err } - a = a[:0] + req.ColumnIDs = req.ColumnIDs[:0] + req.RowIDs = req.RowIDs[:0] + req.ColumnKeys = req.ColumnKeys[:0] + req.RowKeys = req.RowKeys[:0] + req.Timestamps = req.Timestamps[:0] + lastTime = 0 + batchRecs = 0 } } // If there are still bits in the buffer then flush them. - return cmd.importBits(ctx, useColumnKeys, useRowKeys, a) + if batchRecs == 0 { + return nil + } + if lastTime > 0 && lastTime < batchRecs { + req.Timestamps = append(req.Timestamps, make([]int64, batchRecs-lastTime)...) + } + return cmd.importBits(ctx, req) } // importBits sends batches of bits to the server. -func (cmd *ImportCommand) importBits(ctx context.Context, useColumnKeys, useRowKeys bool, bits []pilosa.Bit) error { - logger := log.New(cmd.Stderr, "", log.LstdFlags) - - // If keys are used, all bits are sent to the primary translate store. - if useColumnKeys || useRowKeys { - logger.Printf("importing keys: n=%d", len(bits)) - if err := cmd.client.ImportK(ctx, cmd.Index, cmd.Field, bits, pilosa.OptImportOptionsClear(cmd.Clear)); err != nil { - return errors.Wrap(err, "importing keys") - } - return nil - } - - // Group bits by shard. - logger.Printf("grouping %d bits", len(bits)) - bitsByShard := http.Bits(bits).GroupByShard() - - // Parse path into bits. - for shard, chunk := range bitsByShard { - if cmd.Sort { - sort.Sort(http.BitsByPos(chunk)) - } - - logger.Printf("importing shard: %d, n=%d", shard, len(chunk)) - if err := cmd.client.Import(ctx, cmd.Index, cmd.Field, shard, chunk, pilosa.OptImportOptionsClear(cmd.Clear)); err != nil { - return errors.Wrap(err, "importing") - } - } - - return nil +func (cmd *ImportCommand) importBits(ctx context.Context, req *pilosa.ImportRequest) error { + req.Shard = ^uint64(0) + return cmd.client.Import(ctx, nil, req, &pilosa.ImportOptions{Clear: cmd.Clear}) } // bufferValues buffers slices of record identifiers and values to be imported as a batch. @@ -360,7 +357,7 @@ func (cmd *ImportCommand) bufferValues(ctx context.Context, useColumnKeys, parse // If we've reached the buffer size then import the batch. if len(req.ColumnKeys) == cmd.BufferSize || len(req.ColumnIDs) == cmd.BufferSize { - if err := cmd.client.ImportValue2(ctx, req, &pilosa.ImportOptions{}); err != nil { + if err := cmd.client.ImportValue(ctx, nil, req, &pilosa.ImportOptions{Clear: cmd.Clear}); err != nil { return errors.Wrap(err, "importing values") } req.ColumnIDs = req.ColumnIDs[:0] @@ -371,7 +368,7 @@ func (cmd *ImportCommand) bufferValues(ctx context.Context, useColumnKeys, parse } // If there are still values in the buffer then flush them. - return errors.Wrap(cmd.client.ImportValue2(ctx, req, &pilosa.ImportOptions{}), "importing values") + return errors.Wrap(cmd.client.ImportValue(ctx, nil, req, &pilosa.ImportOptions{Clear: cmd.Clear}), "importing values") } func (cmd *ImportCommand) TLSHost() string { diff --git a/executor_test.go b/executor_test.go index 24395df8c..b6c4b993a 100644 --- a/executor_test.go +++ b/executor_test.go @@ -4131,13 +4131,11 @@ func TestExecutor_Execute_All(t *testing.T) { m0 := c.GetNode(0) // the request gets altered by the Import operation now... - reqs, err := req.Clone().ShardSplit() - if err != nil { - t.Fatalf("splitting request into shards: %v", err) - } - + reqs := req.Clone().SortToShards() qcx := m0.API.Txf().NewQcx() for _, r := range reqs { + // we can ignore the key (which is the shard) because each req + // also got its internal key set. if err := m0.API.Import(context.Background(), qcx, r); err != nil { t.Fatal(err) } diff --git a/field.go b/field.go index 519a3c0d5..e38626df4 100644 --- a/field.go +++ b/field.go @@ -1483,17 +1483,7 @@ func (f *Field) Range(qcx *Qcx, name string, op pql.Token, predicate int64) (*Ro } // Import bulk imports data. -func (f *Field) Import(qcx *Qcx, rowIDs, columnIDs []uint64, timestamps []int64, shard uint64, opts ...ImportOption) (err0 error) { - - // Set up import options. - options := &ImportOptions{} - for _, opt := range opts { - err := opt(options) - if err != nil { - return errors.Wrap(err, "applying option") - } - } - +func (f *Field) Import(qcx *Qcx, rowIDs, columnIDs []uint64, timestamps []int64, shard uint64, options *ImportOptions) (err0 error) { // Determine quantum if timestamps are set. q := f.TimeQuantum() if len(timestamps) > 0 { diff --git a/handler.go b/handler.go index a07c3d6c9..a7630d3c6 100644 --- a/handler.go +++ b/handler.go @@ -18,7 +18,7 @@ import ( "encoding/json" "time" - "github.com/molecula/featurebase/v2/shardwidth" + "github.com/molecula/featurebase/v2/ingest" "github.com/molecula/featurebase/v2/tracing" "github.com/pkg/errors" ) @@ -307,40 +307,31 @@ func (ir *ImportRequest) Clone() *ImportRequest { return newIR } -// ShardSplit splits the request into a slice of import requests. It requires -// that the original request have all elements sorted, and already have -// column IDs, not column keys. -func (ir *ImportRequest) ShardSplit() ([]*ImportRequest, error) { - if ir == nil { - return nil, nil +// SortToShards takes an import request which has been translated, but may +// not be sorted, and turns it into a map from shard IDs to individual import +// requests. We don't sort the entries within each shard because the correct +// sorting depends on the field type and we don't want to deal with that +// here. +func (ir *ImportRequest) SortToShards() map[uint64]*ImportRequest { + // cheat: use ingest + fo := ingest.FieldOperation{ + RecordIDs: ir.ColumnIDs, + Values: ir.RowIDs, + Signed: ir.Timestamps, } - // fix shard - if len(ir.ColumnIDs) < 2 { - ir.Shard = ir.ColumnIDs[0] >> shardwidth.Exponent - return []*ImportRequest{ir}, nil + sharded := fo.SortToShards() + output := make(map[uint64]*ImportRequest, len(sharded)) + for shard, shardOp := range sharded { + shardReq := *ir + shardReq.ColumnKeys = nil + shardReq.RowKeys = nil + shardReq.Shard = shard + shardReq.ColumnIDs = shardOp.RecordIDs + shardReq.RowIDs = shardOp.Values + shardReq.Timestamps = shardOp.Signed + output[shard] = &shardReq } - shards, ends := shardwidth.FindShards(ir.ColumnIDs) - out := make([]*ImportRequest, len(shards)) - prev := 0 - for i, shard := range shards { - next := ends[i] - newIR := &ImportRequest{} - *newIR = *ir - newIR.ColumnIDs = ir.ColumnIDs[prev:next:next] - if ir.RowIDs != nil { - newIR.RowIDs = ir.RowIDs[prev:next:next] - } - if ir.RowKeys != nil { - newIR.RowKeys = ir.RowKeys[prev:next:next] - } - if ir.Timestamps != nil { - newIR.Timestamps = ir.Timestamps[prev:next:next] - } - newIR.Shard = shard - out[i] = newIR - prev = next - } - return out, nil + return output } // ValidateWithTimestamp ensures that the payload of the request is valid. diff --git a/http/client.go b/http/client.go index ca3c52c11..12f87d160 100644 --- a/http/client.go +++ b/http/client.go @@ -46,9 +46,13 @@ type InternalClient struct { // The client to use for HTTP communication. httpClient *http.Client + // the local node's API, used for operations that we can short-circuit that way + api *pilosa.API } // NewInternalClient returns a new instance of InternalClient to connect to host. +// If api is non-nil, the client uses it for some same-host operations instead +// of going through http. func NewInternalClient(host string, remoteClient *http.Client) (*InternalClient, error) { if host == "" { return nil, pilosa.ErrHostRequired @@ -542,47 +546,6 @@ func (c *InternalClient) QueryNode(ctx context.Context, uri *pnet.URI, index str return qresp, nil } -// Import bulk imports bits for a single shard to a host. -func (c *InternalClient) Import(ctx context.Context, index, field string, shard uint64, bits []pilosa.Bit, opts ...pilosa.ImportOption) error { - span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.Import") - defer span.Finish() - - if index == "" { - return pilosa.ErrIndexRequired - } else if field == "" { - return pilosa.ErrFieldRequired - } - - // Set up import options. - options := &pilosa.ImportOptions{} - for _, opt := range opts { - err := opt(options) - if err != nil { - return errors.Wrap(err, "applying option") - } - } - - buf, err := c.marshalImportPayload(index, field, shard, bits) - if err != nil { - return fmt.Errorf("Error Creating Payload: %s", err) - } - - // Retrieve a list of nodes that own the shard. - nodes, err := c.FragmentNodes(ctx, index, shard) - if err != nil { - return fmt.Errorf("shard nodes: %s", err) - } - - // Import to each node. - for _, node := range nodes { - if err := c.importNode(ctx, node, index, field, buf, options); err != nil { - return fmt.Errorf("import node: host=%s, err=%s", node.URI, err) - } - } - - return nil -} - func getPrimaryNode(nodes []*topology.Node) *topology.Node { for _, node := range nodes { if node.IsPrimary { @@ -592,57 +555,6 @@ func getPrimaryNode(nodes []*topology.Node) *topology.Node { return nil } -// ImportK bulk imports bits specified by string keys to a host. -func (c *InternalClient) ImportK(ctx context.Context, index, field string, bits []pilosa.Bit, opts ...pilosa.ImportOption) error { - span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.ImportK") - defer span.Finish() - - if index == "" { - return pilosa.ErrIndexRequired - } else if field == "" { - return pilosa.ErrFieldRequired - } - - // Set up import options. - options := &pilosa.ImportOptions{} - for _, opt := range opts { - err := opt(options) - if err != nil { - return errors.Wrap(err, "applying option") - } - } - - buf, err := c.marshalImportPayload(index, field, 0, bits) - if err != nil { - return fmt.Errorf("Error Creating Payload: %s", err) - } - - // Get the primary node; all bits are sent to the - // primary translate store (i.e. primary). - // TODO... is that right^^? - // RESPONSE: It looks like in ctl/import.go, we could change the - // logic in ImportCommand.importBits() to only use ImportK - // when useRowKeys = true. It's no longer necessary to - // send column key translations to the primary (although - // it should still work). As far as I know, the only thing - // that uses ImportK is the pilosa import sub-command. - nodes, err := c.Nodes(ctx) - if err != nil { - return fmt.Errorf("getting nodes: %s", err) - } - coord := getPrimaryNode(nodes) - if coord == nil { - return fmt.Errorf("could not find the primary node") - } - - // Import to node. - if err := c.importNode(ctx, coord, index, field, buf, options); err != nil { - return fmt.Errorf("import node: host=%s, err=%s", coord.URI, err) - } - - return nil -} - func (c *InternalClient) EnsureIndex(ctx context.Context, name string, options pilosa.IndexOptions) error { span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.EnsureIndex") defer span.Finish() @@ -670,32 +582,6 @@ func (c *InternalClient) EnsureFieldWithOptions(ctx context.Context, indexName s return err } -// marshalImportPayload marshalls the import parameters into a protobuf byte slice. -func (c *InternalClient) marshalImportPayload(index, field string, shard uint64, bits []pilosa.Bit) ([]byte, error) { - // Separate row and column IDs to reduce allocations. - rowIDs := Bits(bits).RowIDs() - rowKeys := Bits(bits).RowKeys() - columnIDs := Bits(bits).ColumnIDs() - columnKeys := Bits(bits).ColumnKeys() - timestamps := Bits(bits).Timestamps() - - // Marshal data to protobuf. - buf, err := c.serializer.Marshal(&pilosa.ImportRequest{ - Index: index, - Field: field, - Shard: shard, - RowIDs: rowIDs, - RowKeys: rowKeys, - ColumnIDs: columnIDs, - ColumnKeys: columnKeys, - Timestamps: timestamps, - }) - if err != nil { - return nil, fmt.Errorf("marshal import request: %s", err) - } - return buf, nil -} - // importNode sends a pre-marshaled import request to a node. func (c *InternalClient) importNode(ctx context.Context, node *topology.Node, index, field string, buf []byte, opts *pilosa.ImportOptions) error { span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.importNode") @@ -747,136 +633,212 @@ func (c *InternalClient) importNode(ctx context.Context, node *topology.Node, in return nil } -// ImportValue bulk imports field values for a single shard to a host. -func (c *InternalClient) ImportValue(ctx context.Context, index, field string, shard uint64, vals []pilosa.FieldValue, opts ...pilosa.ImportOption) error { - span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.ImportValue") +// Import imports values using an ImportRequest, whether or not it's keyed. +// It may modify the contents of req. +// +// If a request comes in with Shard -1, it will be sent to only one node, +// which will translate if necessary, split into shards, and loop back +// through this for each sub-request. If a request uses record keys, +// it will be set to use shard = -1 unconditionally, because we know +// that it has to be translated and possibly reshuffled. Value keys +// don't override the shard. +// +// If we get a non-nil qcx, and have an associated API, we'll use that API +// directly for the local shard. +func (c *InternalClient) Import(ctx context.Context, qcx *pilosa.Qcx, req *pilosa.ImportRequest, options *pilosa.ImportOptions) error { + span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.Import") defer span.Finish() - if index == "" { - return pilosa.ErrIndexRequired - } else if field == "" { - return pilosa.ErrFieldRequired + if req.ColumnKeys != nil { + req.Shard = ^uint64(0) } - - // Set up import options. - options := &pilosa.ImportOptions{} - for _, opt := range opts { - err := opt(options) - if err != nil { - return errors.Wrap(err, "applying option") + // If we don't actually know what shards we're sending to, and we have + // a local API, we can short-circuit: We just send this request to the + // local API, which will break it out and call its defaultClient for + // the data for individual shards. + var nodes []*topology.Node + var err error + if req.Shard == ^uint64(0) { + if qcx != nil && c.api != nil { + err = c.api.ImportWithTx(ctx, qcx, req, options) + // Note that Wrap(nil, ...) is still nil. + return errors.Wrap(err, "local import") } } - - buf, err := c.marshalImportValuePayload(index, field, shard, vals) - if err != nil { - return fmt.Errorf("Error Creating Payload: %s", err) - } - - // Retrieve a list of nodes that own the shard. - nodes, err := c.FragmentNodes(ctx, index, shard) - if err != nil { - return fmt.Errorf("shard nodes: %s", err) - } - - // Import to each node. - for _, node := range nodes { - if err := c.importNode(ctx, node, index, field, buf, options); err != nil { - return fmt.Errorf("import node: host=%s, err=%s", node.URI, err) - } - } - - return nil -} - -// ImportValue2 is a simplified ImportValue method which just uses the -// ImportValueRequest instead of splitting up ImportValue and -// ImportValueK... it also supports importing float values. The idea -// being that (assuming it works) this will become the default (and be -// renamed) for 2.0, and we can deprecate the other methods. -func (c *InternalClient) ImportValue2(ctx context.Context, req *pilosa.ImportValueRequest, options *pilosa.ImportOptions) error { - span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.NewImportValue") - defer span.Finish() - - buf, err := c.serializer.Marshal(req) - if err != nil { - return errors.Errorf("marshal import request: %s", err) - } - - // Retrieve a list of nodes that own the shard. - nodes, err := c.FragmentNodes(ctx, req.Index, req.Shard) + nodes, err = c.FragmentNodes(ctx, req.Index, req.Shard) if err != nil { return errors.Errorf("shard nodes: %s", err) } + // "us" is a usable local node if any, "them" is every node that we need + // to process which isn't that node. We start out with us == nil and + // them = the whole set of nodes. + var us *topology.Node + var them []*topology.Node = nodes - // Import to each node. - for _, node := range nodes { - if err := c.importNode(ctx, node, req.Index, req.Field, buf, options); err != nil { - return errors.Errorf("import node: host=%s, err=%s", node.URI, err) + // If we have an API, we know what node we are. Even if we don't have + // a Qcx, we still care, because looping back to the local node will + // be faster than going to another node. + if c.api != nil { + myID := c.api.NodeID() + for i, node := range nodes { + if myID == node.ID { + // swap our node into the first position + nodes[i], nodes[0] = nodes[0], nodes[i] + // If we have a qcx, we'll treat our node even MORE + // specially. + if qcx != nil { + us, them = nodes[0], nodes[1:] + } + break + } + } + } + + // If we had a valid API and Qcx, and shard was ^0, we'd have handled + // it previously. So if we get here, we don't have both a Qcx and an API, + // but we might have an API, in which case we'll have shuffled our node + // into the first position. Otherwise we're just taking whatever the first + // node is. + if req.Shard == ^uint64(0) { + them = them[:1] + } + + // We handle remote nodes first, for two distinct reasons. One is that + // the local API ImportWithTx is allowed to modify its inputs, and if we + // ran that before serializing, we'd get corrupt data serialized. + // The other is that if we were to hold a write lock that started with + // that import and ended when we hit the end of our Qcx, we wouldn't want + // to hold it during all our requests to the remote nodes. + if len(them) > 0 { + buf, err := c.serializer.Marshal(req) + if err != nil { + return errors.Errorf("marshal import request: %s", err) + } + // We process remote nodes first so we won't be actually holding our + // write lock yet, in theory. This doesn't actually matter yet, but is + // helpful for future planned refactoring. + for _, node := range them { + if err = c.importNode(ctx, node, req.Index, req.Field, buf, options); err != nil { + return errors.Wrap(err, "remote import") + } + } + } + + // Write to the local node if we have one. + if us != nil { + // WARNING: ImportWithTx can alter its inputs. However, we can + // only ever do this once, and if we're going to need a marshalled + // form, we already made it. + if err = c.api.ImportWithTx(ctx, qcx, req, options); err != nil { + return errors.Wrap(err, "local import") } } return nil } -// ImportValueK bulk imports keyed field values to a host. -func (c *InternalClient) ImportValueK(ctx context.Context, index, field string, vals []pilosa.FieldValue, opts ...pilosa.ImportOption) error { - span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.ImportValueK") +// ImportValue imports values using an ImportValueRequest, whether or not it's +// keyed. It may modify the contents of req. +// +// If a request comes in with Shard -1, it will be sent to only one node, +// which will translate if necessary, split into shards, and loop back +// through this for each sub-request. If a request uses record keys, +// it will be set to use shard = -1 unconditionally, because we know +// that it has to be translated and possibly reshuffled. Value keys +// don't override the shard. +// +// If we get a non-nil qcx, and have an associated API, we'll use that API +// directly for the local shard. +func (c *InternalClient) ImportValue(ctx context.Context, qcx *pilosa.Qcx, req *pilosa.ImportValueRequest, options *pilosa.ImportOptions) error { + span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.ImportValue") defer span.Finish() - buf, err := c.marshalImportValuePayload(index, field, 0, vals) - if err != nil { - return fmt.Errorf("Error Creating Payload: %s", err) + if req.ColumnKeys != nil { + req.Shard = ^uint64(0) } + // If we don't actually know what shards we're sending to, and we have + // a local API, we can short-circuit: We just send this request to the + // local API, which will break it out and call its defaultClient for + // the data for individual shards. + var nodes []*topology.Node + var err error + if req.Shard == ^uint64(0) { + if qcx != nil && c.api != nil { + err = c.api.ImportValueWithTx(ctx, qcx, req, options) + // Note that Wrap(nil, ...) is still nil. + return errors.Wrap(err, "local import") + } + } + nodes, err = c.FragmentNodes(ctx, req.Index, req.Shard) + if err != nil { + return errors.Errorf("shard nodes: %s", err) + } + // "us" is a usable local node if any, "them" is every node that we need + // to process which isn't that node. We start out with us == nil and + // them = the whole set of nodes. + var us *topology.Node + var them []*topology.Node = nodes - // Set up import options. - options := &pilosa.ImportOptions{} - for _, opt := range opts { - err := opt(options) - if err != nil { - return errors.Wrap(err, "applying option") + // If we have an API, we know what node we are. Even if we don't have + // a Qcx, we still care, because looping back to the local node will + // be faster than going to another node. + if c.api != nil { + myID := c.api.NodeID() + for i, node := range nodes { + if myID == node.ID { + // swap our node into the first position + nodes[i], nodes[0] = nodes[0], nodes[i] + // If we have a qcx, we'll treat our node even MORE + // specially. + if qcx != nil { + us, them = nodes[0], nodes[1:] + } + break + } } } - // Get the primary node; all bits are sent to the - // primary translate store. - nodes, err := c.Nodes(ctx) - if err != nil { - return fmt.Errorf("getting nodes: %s", err) - } - coord := getPrimaryNode(nodes) - if coord == nil { - return fmt.Errorf("could not find the primary node") + // If we had a valid API and Qcx, and shard was ^0, we'd have handled + // it previously. So if we get here, we don't have both a Qcx and an API, + // but we might have an API, in which case we'll have shuffled our node + // into the first position. Otherwise we're just taking whatever the first + // node is. + if req.Shard == ^uint64(0) { + them = them[:1] } - // Import to node. - if err := c.importNode(ctx, coord, index, field, buf, options); err != nil { - return fmt.Errorf("import node: host=%s, err=%s", coord.URI, err) + // We handle remote nodes first, for two distinct reasons. One is that + // the local API ImportWithTx is allowed to modify its inputs, and if we + // ran that before serializing, we'd get corrupt data serialized. + // The other is that if we were to hold a write lock that started with + // that import and ended when we hit the end of our Qcx, we wouldn't want + // to hold it during all our requests to the remote nodes. + if len(them) > 0 { + buf, err := c.serializer.Marshal(req) + if err != nil { + return errors.Errorf("marshal import request: %s", err) + } + // We process remote nodes first so we won't be actually holding our + // write lock yet, in theory. This doesn't actually matter yet, but is + // helpful for future planned refactoring. + for _, node := range them { + if err = c.importNode(ctx, node, req.Index, req.Field, buf, options); err != nil { + return errors.Wrap(err, "remote import") + } + } } + // Write to the local node if we have one. + if us != nil { + // WARNING: ImportWithTx can alter its inputs. However, we can + // only ever do this once, and if we're going to need a marshalled + // form, we already made it. + if err = c.api.ImportValueWithTx(ctx, qcx, req, options); err != nil { + return errors.Wrap(err, "local import") + } + } return nil } -// marshalImportValuePayload marshalls the import parameters into a protobuf byte slice. -func (c *InternalClient) marshalImportValuePayload(index, field string, shard uint64, vals []pilosa.FieldValue) ([]byte, error) { - // Separate row and column IDs to reduce allocations. - columnIDs := FieldValues(vals).ColumnIDs() - columnKeys := FieldValues(vals).ColumnKeys() - values := FieldValues(vals).Values() - - // Marshal data to protobuf. - buf, err := c.serializer.Marshal(&pilosa.ImportValueRequest{ - Index: index, - Field: field, - Shard: shard, - ColumnIDs: columnIDs, - ColumnKeys: columnKeys, - Values: values, - }) - if err != nil { - return nil, fmt.Errorf("marshal import request: %s", err) - } - return buf, nil -} - // ImportRoaring does fast import of raw bits in roaring format (pilosa or // official format, see API.ImportRoaring). func (c *InternalClient) ImportRoaring(ctx context.Context, uri *pnet.URI, index, field string, shard uint64, remote bool, req *pilosa.ImportRoaringRequest) error { @@ -2336,3 +2298,7 @@ func (c *InternalClient) PartitionNodes(ctx context.Context, partitionID int) ([ } return a, nil } + +func (c *InternalClient) SetInternalAPI(api *pilosa.API) { + c.api = api +} diff --git a/http/client_test.go b/http/client_test.go index d78b04e08..a87c776e9 100644 --- a/http/client_test.go +++ b/http/client_test.go @@ -202,7 +202,6 @@ func TestClient_Export(t *testing.T) { cmd.MustCreateField(t, "unkeyed", "unkeyedf", pilosa.OptFieldTypeSet(pilosa.CacheTypeRanked, 1000)) c := MustNewClient(host, http.GetHTTPClient(nil)) - data := []pilosa.Bit{ {RowID: 1, ColumnID: 100, RowKey: "row1", ColumnKey: "col100"}, {RowID: 1, ColumnID: 101, RowKey: "row1", ColumnKey: "col101"}, @@ -367,49 +366,155 @@ func TestClient_Export(t *testing.T) { // Ensure client can bulk import data. func TestClient_Import(t *testing.T) { - cluster := test.MustRunCluster(t, 1) + // Need a cluster to verify hitting multiple nodes + cluster := test.MustRunCluster(t, 3) defer cluster.Close() cmd := cluster.GetNode(0) host := cmd.URL() - holder := cmd.Server.Holder() - hldr := test.Holder{Holder: holder} + api := cmd.API - // Load bitmap into cache to ensure cache gets updated. - hldr.SetBit("i", "f", 1, 0) // set a bit so the view gets created. - hldr.Row("i", "f", 0) + cmd.MustCreateIndex(t, "keyed", pilosa.IndexOptions{Keys: true}) + cmd.MustCreateIndex(t, "unkeyed", pilosa.IndexOptions{Keys: false}) + + cmd.MustCreateField(t, "keyed", "keyedf", pilosa.OptFieldTypeSet(pilosa.CacheTypeNone, 0), pilosa.OptFieldKeys()) + cmd.MustCreateField(t, "keyed", "unkeyedf", pilosa.OptFieldTypeSet(pilosa.CacheTypeNone, 0)) + cmd.MustCreateField(t, "unkeyed", "keyedf", pilosa.OptFieldTypeSet(pilosa.CacheTypeNone, 0), pilosa.OptFieldKeys()) + cmd.MustCreateField(t, "unkeyed", "unkeyedf", pilosa.OptFieldTypeSet(pilosa.CacheTypeNone, 0)) + + indexes := map[bool]string{true: "keyed", false: "unkeyed"} + fields := map[bool]string{true: "keyedf", false: "unkeyedf"} + + recKeys := []string{"rec-a", "rec-b", "rec-c"} + valueKeys := []string{"val-a", "val-b", "val-c"} + recIDs := []uint64{0, 3, 7} + valueIDs := []uint64{0, 3, 7} - // Send import request. c := MustNewClient(host, http.GetHTTPClient(nil)) - if err := c.Import(context.Background(), "i", "f", 0, []pilosa.Bit{ - {RowID: 0, ColumnID: 1}, - {RowID: 0, ColumnID: 5}, - {RowID: 200, ColumnID: 6}, - }); err != nil { - t.Fatal(err) + // set API to point at the local node + c.SetInternalAPI(cmd.API) + + checkResults := func(results pilosa.QueryResponse, keyed bool, maxN int) { + for i, r := range results.Results { + row, ok := r.(*pilosa.Row) + if !ok { + t.Fatalf("expected row, got %T", r) + } + if i >= maxN { + // we cleared these, so they should be empty + if keyed { + vals := row.Keys + if len(vals) != 0 { + t.Fatalf("expected empty row, got %d result(s) back, first result %q", + len(vals), vals[0]) + } + } else { + vals := row.Columns() + if len(vals) != 0 { + t.Fatalf("expected empty row, got %d result(s) back, first result %d", + len(vals), vals[0]) + } + } + return + } + if keyed { + vals := row.Keys + if len(vals) != 1 { + t.Fatalf("expected one result, didn't get it") + } + if vals[0] != recKeys[i] { + t.Fatalf("expected %q, got %q", recKeys[i], vals[0]) + } + } else { + vals := row.Columns() + if len(vals) != 1 { + t.Fatalf("expected one result, didn't get it") + } + if vals[0] != recIDs[i] { + t.Fatalf("expected %d, got %d", recIDs[i], vals[0]) + } + } + } } - // Verify data. - if a := hldr.Row("i", "f", 0).Columns(); !reflect.DeepEqual(a, []uint64{1, 5}) { - t.Fatalf("unexpected columns: %+v", a) - } - if a := hldr.Row("i", "f", 200).Columns(); !reflect.DeepEqual(a, []uint64{6}) { - t.Fatalf("unexpected columns: %+v", a) - } + for keyed, indexName := range indexes { + for _, fieldName := range fields { + req := pilosa.ImportRequest{ + Index: indexName, + Field: fieldName, + } + if indexName == "keyed" { + req.ColumnKeys = recKeys + req.Shard = ^uint64(0) + } else { + req.ColumnIDs = recIDs + req.Shard = 0 + } + if fieldName == "keyedf" { + req.RowKeys = valueKeys + } else { + req.RowIDs = valueIDs + } + // clone request because some imports will modify their input parameters + if err := c.Import(context.Background(), nil, req.Clone(), &pilosa.ImportOptions{}); err != nil { + t.Fatalf("%s/%s: %v", + indexName, fieldName, err) + } - // Clear some data. - if err := c.Import(context.Background(), "i", "f", 0, []pilosa.Bit{ - {RowID: 0, ColumnID: 5}, - {RowID: 200, ColumnID: 6}, - }, pilosa.OptImportOptionsClear(true)); err != nil { - t.Fatal(err) - } + // Now do a query to see whether it worked... + var pql string + if fieldName == "keyedf" { + pql = `Row(keyedf="val-a") Row(keyedf="val-b") Row(keyedf="val-c")` + } else { + pql = `Row(unkeyedf=0) Row(unkeyedf=3) Row(unkeyedf=7)` + } + results, err := api.Query(context.Background(), &pilosa.QueryRequest{Index: indexName, Query: pql}) + if err != nil { + t.Fatal(err) + } + checkResults(results, keyed, 3) - // Verify data. - if a := hldr.Row("i", "f", 0).Columns(); !reflect.DeepEqual(a, []uint64{1}) { - t.Fatalf("unexpected columns: %+v", a) - } - if a := hldr.Row("i", "f", 200).Columns(); !reflect.DeepEqual(a, []uint64{}) { - t.Fatalf("unexpected columns: %+v", a) + // Now clear a bit... + req = pilosa.ImportRequest{ + Index: indexName, + Field: fieldName, + } + if indexName == "keyed" { + req.ColumnKeys = recKeys[2:] + req.Shard = ^uint64(0) + } else { + req.ColumnIDs = recIDs[2:] + req.Shard = 0 + } + if fieldName == "keyedf" { + req.RowKeys = valueKeys[2:] + } else { + req.RowIDs = valueIDs[2:] + } + // do a clear. also, do the clear with a Qcx. + func() { + // inner function so the deferred abort isn't delayed a lot + qcx := api.Txf().NewQcx() + defer qcx.Abort() + if err := c.Import(context.Background(), qcx, req.Clone(), &pilosa.ImportOptions{Clear: true}); err != nil { + t.Fatalf("%s/%s: %v", + indexName, fieldName, err) + } + if err := qcx.Finish(); err != nil { + t.Fatalf("committing write: %v", err) + } + }() + // Now do a query to see whether it worked... + if fieldName == "keyedf" { + pql = `Row(keyedf="val-a") Row(keyedf="val-b") Row(keyedf="val-c")` + } else { + pql = `Row(unkeyedf=0) Row(unkeyedf=3) Row(unkeyedf=7)` + } + results, err = api.Query(context.Background(), &pilosa.QueryRequest{Index: indexName, Query: pql}) + if err != nil { + t.Fatal(err) + } + checkResults(results, keyed, 2) + } } } @@ -547,15 +652,17 @@ func TestClient_ImportRoaring_MultiView(t *testing.T) { } defer cluster.Close() - _, err = cluster.GetNode(0).API.CreateIndex(context.Background(), "i", pilosa.IndexOptions{}) + api := cluster.GetNode(0).API + + _, err = api.CreateIndex(context.Background(), "i", pilosa.IndexOptions{}) if err != nil { t.Fatalf("creating index: %v", err) } - _, err = cluster.GetNode(0).API.CreateField(context.Background(), "i", "f", pilosa.OptFieldTypeSet(pilosa.CacheTypeRanked, 100)) + _, err = api.CreateField(context.Background(), "i", "f", pilosa.OptFieldTypeSet(pilosa.CacheTypeRanked, 100)) if err != nil { t.Fatalf("creating field: %v", err) } - _, err = cluster.GetNode(0).API.Query(context.Background(), &pilosa.QueryRequest{Index: "i", Query: "Set(0, f=1)"}) + _, err = api.Query(context.Background(), &pilosa.QueryRequest{Index: "i", Query: "Set(0, f=1)"}) if err != nil { t.Fatalf("querying: %v", err) } @@ -588,16 +695,21 @@ func TestClient_ImportKeys(t *testing.T) { // Send import request. c := MustNewClient(host, http.GetHTTPClient(nil)) + baseReq := &pilosa.ImportRequest{ + Index: "keyed", + Field: "keyedf", + ColumnKeys: []string{"eve", "alice", "bob", "eve", "alice", "eve"}, + ColumnIDs: []uint64{1, 2, 3, 1, 2, 1}, + RowKeys: []string{"green", "green", "green", "blue", "blue", "purple"}, + RowIDs: []uint64{1, 1, 1, 2, 2, 3}, + } t.Run("Import keyed,keyed", func(t *testing.T) { - if err := c.Import(context.Background(), "keyed", "keyedf", 0, []pilosa.Bit{ - {RowKey: "green", ColumnKey: "eve"}, - {RowKey: "green", ColumnKey: "alice"}, - {RowKey: "green", ColumnKey: "bob"}, - {RowKey: "blue", ColumnKey: "eve"}, - {RowKey: "blue", ColumnKey: "alice"}, - {RowKey: "purple", ColumnKey: "eve"}, - }); err != nil { + req := baseReq.Clone() + req.Index = "keyed" + req.Field = "keyedf" + req.ColumnIDs, req.RowIDs = nil, nil + if err := c.Import(context.Background(), nil, req, &pilosa.ImportOptions{}); err != nil { t.Fatal(err) } resp := cmd.QueryAPI(t, &pilosa.QueryRequest{ @@ -616,14 +728,11 @@ func TestClient_ImportKeys(t *testing.T) { }) t.Run("Import keyed,unkeyedf", func(t *testing.T) { - if err := c.Import(context.Background(), "keyed", "unkeyedf", 0, []pilosa.Bit{ - {RowID: 1, ColumnKey: "eve"}, - {RowID: 1, ColumnKey: "alice"}, - {RowID: 1, ColumnKey: "bob"}, - {RowID: 2, ColumnKey: "eve"}, - {RowID: 2, ColumnKey: "alice"}, - {RowID: 3, ColumnKey: "eve"}, - }); err != nil { + req := baseReq.Clone() + req.Index = "keyed" + req.Field = "unkeyedf" + req.ColumnIDs, req.RowKeys = nil, nil + if err := c.Import(context.Background(), nil, req, &pilosa.ImportOptions{}); err != nil { t.Fatal(err) } resp := cmd.QueryAPI(t, &pilosa.QueryRequest{ @@ -642,14 +751,11 @@ func TestClient_ImportKeys(t *testing.T) { }) t.Run("Import unkeyed,keyed", func(t *testing.T) { - if err := c.Import(context.Background(), "unkeyed", "keyedf", 0, []pilosa.Bit{ - {RowKey: "green", ColumnID: 1}, - {RowKey: "green", ColumnID: 2}, - {RowKey: "green", ColumnID: 3}, - {RowKey: "blue", ColumnID: 1}, - {RowKey: "blue", ColumnID: 2}, - {RowKey: "purple", ColumnID: 1}, - }); err != nil { + req := baseReq.Clone() + req.Index = "unkeyed" + req.Field = "keyedf" + req.ColumnKeys, req.RowIDs = nil, nil + if err := c.Import(context.Background(), nil, req, &pilosa.ImportOptions{}); err != nil { t.Fatal(err) } resp := cmd.QueryAPI(t, &pilosa.QueryRequest{ @@ -686,14 +792,13 @@ func TestClient_ImportKeys(t *testing.T) { // Import to node0. t.Run("Import node0", func(t *testing.T) { - if err := c0.ImportK(context.Background(), "keyed", "keyedf0", []pilosa.Bit{ - {RowKey: "green", ColumnKey: "eve"}, - {RowKey: "green", ColumnKey: "alice"}, - {RowKey: "green", ColumnKey: "bob"}, - {RowKey: "blue", ColumnKey: "eve"}, - {RowKey: "blue", ColumnKey: "alice"}, - {RowKey: "purple", ColumnKey: "eve"}, - }); err != nil { + req := &pilosa.ImportRequest{ + Index: "keyed", + Field: "keyedf0", + ColumnKeys: []string{"eve", "alice", "bob", "eve", "alice", "eve"}, + RowKeys: []string{"green", "green", "green", "blue", "blue", "purple"}, + } + if err := c0.Import(context.Background(), nil, req, &pilosa.ImportOptions{}); err != nil { t.Fatal(err) } resp := cmd0.QueryAPI(t, &pilosa.QueryRequest{ @@ -713,14 +818,13 @@ func TestClient_ImportKeys(t *testing.T) { // Import to node1 (ensure import is routed to primary for translation). t.Run("Import node1", func(t *testing.T) { - if err := c1.ImportK(context.Background(), "keyed", "keyedf1", []pilosa.Bit{ - {RowKey: "green", ColumnKey: "eve"}, - {RowKey: "green", ColumnKey: "alice"}, - {RowKey: "green", ColumnKey: "bob"}, - {RowKey: "blue", ColumnKey: "eve"}, - {RowKey: "blue", ColumnKey: "alice"}, - {RowKey: "purple", ColumnKey: "eve"}, - }); err != nil { + req := &pilosa.ImportRequest{ + Index: "keyed", + Field: "keyedf1", + ColumnKeys: []string{"eve", "alice", "bob", "eve", "alice", "eve"}, + RowKeys: []string{"green", "green", "green", "blue", "blue", "purple"}, + } + if err := c1.Import(context.Background(), nil, req, &pilosa.ImportOptions{}); err != nil { t.Fatal(err) } @@ -762,11 +866,13 @@ func TestClient_ImportKeys(t *testing.T) { // Send import request. c := MustNewClient(host, http.GetHTTPClient(nil)) - if err := c.ImportValue(context.Background(), "i", "f", 0, []pilosa.FieldValue{ - {ColumnKey: "col1", Value: -10}, - {ColumnKey: "col2", Value: 20}, - {ColumnKey: "col3", Value: 40}, - }); err != nil { + req := &pilosa.ImportValueRequest{ + Index: "i", + Field: "f", + ColumnKeys: []string{"col1", "col2", "col3"}, + Values: []int64{-10, 20, 40}, + } + if err := c.ImportValue(context.Background(), nil, req, &pilosa.ImportOptions{}); err != nil { t.Fatal(err) } @@ -786,9 +892,13 @@ func TestClient_ImportKeys(t *testing.T) { } // Clear data. - if err := c.ImportValue(context.Background(), "i", "f", 0, []pilosa.FieldValue{ - {ColumnKey: "col2", Value: 20}, - }, pilosa.OptImportOptionsClear(true)); err != nil { + req = &pilosa.ImportValueRequest{ + Index: "i", + Field: "f", + ColumnKeys: []string{"col2"}, + Values: []int64{20}, + } + if err := c.ImportValue(context.Background(), nil, req, &pilosa.ImportOptions{Clear: true}); err != nil { t.Fatal(err) } @@ -835,9 +945,13 @@ func TestClient_ImportIDs(t *testing.T) { // Send import request. c := MustNewClient(host, http.GetHTTPClient(nil)) - if err := c.ImportValue(context.Background(), idxName, fldName, 0, []pilosa.FieldValue{ - {ColumnID: 2, Value: 1}, - }); err != nil { + req := &pilosa.ImportValueRequest{ + Index: idxName, + Field: fldName, + ColumnIDs: []uint64{2}, + Values: []int64{1}, + } + if err := c.ImportValue(context.Background(), nil, req, &pilosa.ImportOptions{}); err != nil { t.Fatal(err) } @@ -857,9 +971,13 @@ func TestClient_ImportIDs(t *testing.T) { } // Send import request. - if err := c.ImportValue(context.Background(), idxName, fldName, 0, []pilosa.FieldValue{ - {ColumnID: 1000, Value: 1}, - }); err != nil { + req = &pilosa.ImportValueRequest{ + Index: idxName, + Field: fldName, + ColumnIDs: []uint64{1000}, + Values: []int64{1}, + } + if err := c.ImportValue(context.Background(), nil, req, &pilosa.ImportOptions{}); err != nil { t.Fatal(err) } @@ -895,11 +1013,13 @@ func TestClient_ImportValue(t *testing.T) { // Send import request. c := MustNewClient(host, http.GetHTTPClient(nil)) - if err := c.ImportValue(context.Background(), "i", "f", 0, []pilosa.FieldValue{ - {ColumnID: 1, Value: -10}, - {ColumnID: 2, Value: 20}, - {ColumnID: 3, Value: 40}, - }); err != nil { + req := &pilosa.ImportValueRequest{ + Index: "i", + Field: "f", + ColumnIDs: []uint64{1, 2, 3}, + Values: []int64{-10, 20, 40}, + } + if err := c.ImportValue(context.Background(), nil, req, &pilosa.ImportOptions{}); err != nil { t.Fatal(err) } @@ -922,10 +1042,13 @@ func TestClient_ImportValue(t *testing.T) { } // Send import request. - if err := c.ImportValue(context.Background(), "i", "f", 0, []pilosa.FieldValue{ - {ColumnID: 1, Value: -10}, - {ColumnID: 3, Value: 40}, - }, pilosa.OptImportOptionsClear(true)); err != nil { + req = &pilosa.ImportValueRequest{ + Index: "i", + Field: "f", + ColumnIDs: []uint64{1, 3}, + Values: []int64{-10, 40}, + } + if err := c.ImportValue(context.Background(), nil, req, &pilosa.ImportOptions{Clear: true}); err != nil { t.Fatal(err) } @@ -969,11 +1092,13 @@ func TestClient_ImportExistence(t *testing.T) { // Send import request. c := MustNewClient(host, http.GetHTTPClient(nil)) - if err := c.Import(context.Background(), idxName, fldName, 0, []pilosa.Bit{ - {RowID: 0, ColumnID: 1}, - {RowID: 0, ColumnID: 5}, - {RowID: 200, ColumnID: 6}, - }); err != nil { + req := &pilosa.ImportRequest{ + Index: "iset", + Field: "fset", + ColumnIDs: []uint64{1, 5, 6}, + RowIDs: []uint64{0, 0, 200}, + } + if err := c.Import(context.Background(), nil, req, &pilosa.ImportOptions{}); err != nil { t.Fatal(err) } @@ -1003,11 +1128,13 @@ func TestClient_ImportExistence(t *testing.T) { // Send import request. c := MustNewClient(host, http.GetHTTPClient(nil)) - if err := c.ImportValue(context.Background(), idxName, fldName, 0, []pilosa.FieldValue{ - {ColumnID: 1, Value: -10}, - {ColumnID: 2, Value: 20}, - {ColumnID: 3, Value: 40}, - }); err != nil { + req := &pilosa.ImportValueRequest{ + Index: "iint", + Field: "fint", + ColumnIDs: []uint64{1, 2, 3}, + Values: []int64{-10, 20, 40}, + } + if err := c.ImportValue(context.Background(), nil, req, &pilosa.ImportOptions{}); err != nil { t.Fatal(err) } @@ -1087,7 +1214,7 @@ func TestClient_CreateDecimalField(t *testing.T) { t.Fatalf("expected Scale 1, got: %+v", fld.Options()) } - err = c.ImportValue2(context.Background(), &pilosa.ImportValueRequest{Index: index, Field: field, ColumnIDs: []uint64{1, 2, 3}, Shard: 0, FloatValues: []float64{1.1, 2.2, 3.3}}, &pilosa.ImportOptions{}) + err = c.ImportValue(context.Background(), nil, &pilosa.ImportValueRequest{Index: index, Field: field, ColumnIDs: []uint64{1, 2, 3}, Shard: 0, FloatValues: []float64{1.1, 2.2, 3.3}}, &pilosa.ImportOptions{}) if err != nil { t.Fatalf("importing float values: %v", err) } diff --git a/internal/clustertests/cluster_test.go b/internal/clustertests/cluster_test.go index 3726bcc3b..df8390fb4 100644 --- a/internal/clustertests/cluster_test.go +++ b/internal/clustertests/cluster_test.go @@ -53,13 +53,19 @@ func TestClusterStuff(t *testing.T) { t.Fatalf("creating field: %v", err) } - data := make([]pilosa.Bit, 10) + req := &pilosa.ImportRequest{ + Index: "testidx", + Field: "testf", + } + req.ColumnIDs = make([]uint64, 10) + req.RowIDs = make([]uint64, 10) + for i := 0; i < 1000; i++ { - data[i%10].RowID = 0 - data[i%10].ColumnID = uint64((i/10)*pilosa.ShardWidth + i%10) - shard := uint64(i / 10) + req.RowIDs[i%10] = 0 + req.ColumnIDs[i%10] = uint64((i/10)*pilosa.ShardWidth + i%10) + req.Shard = uint64(i / 10) if i%10 == 9 { - err = cli1.Import(context.Background(), "testidx", "testf", shard, data) + err = cli1.Import(context.Background(), nil, req, &pilosa.ImportOptions{}) if err != nil { t.Fatalf("importing: %v", err) } diff --git a/server.go b/server.go index 63077302b..8058c25f4 100644 --- a/server.go +++ b/server.go @@ -531,6 +531,10 @@ func (s *Server) GRPCURI() pnet.URI { return s.grpcURI } +func (s *Server) SetAPI(api *API) { + s.defaultClient.SetInternalAPI(api) +} + // UpAndDown brings the server up minimally and shuts it down // again; basically, it exists for testing holder open and close. func (s *Server) UpAndDown() error { diff --git a/server/server.go b/server/server.go index 929404852..3feaf4582 100644 --- a/server/server.go +++ b/server/server.go @@ -516,6 +516,8 @@ func (m *Command) SetupServer() error { if err != nil { return errors.Wrap(err, "new api") } + // Tell server about its new API, which its client will need. + m.Server.SetAPI(m.API) m.grpcServer, err = NewGRPCServer( OptGRPCServerAPI(m.API), diff --git a/server/server_test.go b/server/server_test.go index d7f26a36d..5e2a47dfa 100644 --- a/server/server_test.go +++ b/server/server_test.go @@ -67,6 +67,7 @@ func TestMain_Set_Quick(t *testing.T) { // Create client. client, err := http.NewInternalClient(m.API.Node().URI.HostPort(), http.GetHTTPClient(nil)) + client.SetInternalAPI(m.API) if err != nil { t.Fatal(err) } @@ -831,6 +832,7 @@ func TestQueryingWithQuotesAndStuff(t *testing.T) { defer m.Close() client, err := http.NewInternalClient(m.API.Node().URI.HostPort(), http.GetHTTPClient(nil)) + client.SetInternalAPI(m.API) if err != nil { t.Fatal(err) }