mirror of
https://github.com/featurebasedb/featurebase.git
synced 2026-08-28 02:44:59 +00:00
Add more tracing and metdata to importRoaring
This commit is contained in:
Matt Jaffee
parent
e1e0d0cdfa
commit
61bf3d929d
7 changed files with 78 additions and 27 deletions
5
api.go
5
api.go
|
|
@ -290,6 +290,7 @@ func setUpImportOptions(opts ...ImportOption) (*ImportOptions, error) {
|
|||
// bitmap.
|
||||
func (api *API) ImportRoaring(ctx context.Context, indexName, fieldName string, shard uint64, remote bool, req *ImportRoaringRequest) (err error) {
|
||||
span, ctx := tracing.StartSpanFromContext(ctx, "API.ImportRoaring")
|
||||
span.LogKV("index", indexName, "field", fieldName)
|
||||
defer span.Finish()
|
||||
|
||||
if err = api.validate(apiField); err != nil {
|
||||
|
|
@ -325,7 +326,7 @@ func (api *API) ImportRoaring(ctx context.Context, indexName, fieldName string,
|
|||
}
|
||||
fileMagic := uint32(binary.LittleEndian.Uint16(viewData[0:2]))
|
||||
if fileMagic == roaring.MagicNumber { // if pilosa roaring format
|
||||
err = field.importRoaring(viewData, shard, viewName, req.Clear)
|
||||
err = field.importRoaring(ctx, viewData, shard, viewName, req.Clear)
|
||||
if err != nil {
|
||||
return errors.Wrap(err, "importing pilosa roaring")
|
||||
}
|
||||
|
|
@ -335,7 +336,7 @@ func (api *API) ImportRoaring(ctx context.Context, indexName, fieldName string,
|
|||
// field.importRoaring changes the standard roaring run format to pilosa roaring
|
||||
data := make([]byte, len(viewData))
|
||||
copy(data, viewData)
|
||||
err = field.importRoaring(data, shard, viewName, req.Clear)
|
||||
err = field.importRoaring(ctx, data, shard, viewName, req.Clear)
|
||||
if err != nil {
|
||||
return errors.Wrap(err, "importing standard roaring")
|
||||
}
|
||||
|
|
|
|||
10
field.go
10
field.go
|
|
@ -16,6 +16,7 @@ package pilosa
|
|||
|
||||
import (
|
||||
"bufio"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"io/ioutil"
|
||||
|
|
@ -32,6 +33,7 @@ import (
|
|||
"github.com/pilosa/pilosa/pql"
|
||||
"github.com/pilosa/pilosa/roaring"
|
||||
"github.com/pilosa/pilosa/stats"
|
||||
"github.com/pilosa/pilosa/tracing"
|
||||
"github.com/pkg/errors"
|
||||
)
|
||||
|
||||
|
|
@ -1218,10 +1220,14 @@ func (f *Field) importValue(columnIDs []uint64, values []int64, options *ImportO
|
|||
return nil
|
||||
}
|
||||
|
||||
func (f *Field) importRoaring(data []byte, shard uint64, viewName string, clear bool) error {
|
||||
func (f *Field) importRoaring(ctx context.Context, data []byte, shard uint64, viewName string, clear bool) error {
|
||||
span, ctx := tracing.StartSpanFromContext(ctx, "Field.importRoaring")
|
||||
defer span.Finish()
|
||||
|
||||
if viewName == "" {
|
||||
viewName = viewStandard
|
||||
}
|
||||
span.LogKV("view", viewName, "bytes", len(data), "shard", shard)
|
||||
view, err := f.createViewIfNotExists(viewName)
|
||||
if err != nil {
|
||||
return errors.Wrap(err, "creating view")
|
||||
|
|
@ -1232,7 +1238,7 @@ func (f *Field) importRoaring(data []byte, shard uint64, viewName string, clear
|
|||
return errors.Wrap(err, "creating fragment")
|
||||
}
|
||||
|
||||
if err := frag.importRoaring(data, clear); err != nil {
|
||||
if err := frag.importRoaring(ctx, data, clear); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
|
|
|
|||
19
fragment.go
19
fragment.go
|
|
@ -1811,11 +1811,15 @@ func (f *fragment) importValue(columnIDs, values []uint64, bitDepth uint, clear
|
|||
// importRoaring imports from the official roaring data format defined at
|
||||
// https://github.com/RoaringBitmap/RoaringFormatSpec or from pilosa's version
|
||||
// of the roaring format. The cache is updated to reflect the new data.
|
||||
func (f *fragment) importRoaring(data []byte, clear bool) error {
|
||||
func (f *fragment) importRoaring(ctx context.Context, data []byte, clear bool) error {
|
||||
span, ctx := tracing.StartSpanFromContext(ctx, "fragment.importRoaring")
|
||||
defer span.Finish()
|
||||
f.mu.Lock()
|
||||
defer f.mu.Unlock()
|
||||
bm := roaring.NewBTreeBitmap()
|
||||
span, ctx = tracing.StartSpanFromContext(ctx, "importRoaring.UnmarshalBinary")
|
||||
err := bm.UnmarshalBinary(data)
|
||||
span.Finish()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
|
@ -1851,16 +1855,25 @@ func (f *fragment) importRoaring(data []byte, clear bool) error {
|
|||
if clear {
|
||||
toSet, toClear = toClear, toSet
|
||||
}
|
||||
return f.importPositions(toSet, toClear, rowSet)
|
||||
span, _ = tracing.StartSpanFromContext(ctx, "importRoaring.ImportPositions")
|
||||
err := f.importPositions(toSet, toClear, rowSet)
|
||||
span.Finish()
|
||||
return err
|
||||
}
|
||||
|
||||
if clear {
|
||||
span, ctx = tracing.StartSpanFromContext(ctx, "importRoaringDifference")
|
||||
bm = f.storage.Difference(bm)
|
||||
span.Finish()
|
||||
} else if f.storage.Containers.Size() >= bm.Containers.Size() {
|
||||
span, ctx = tracing.StartSpanFromContext(ctx, "importRoaringStorageUIP")
|
||||
f.storage.UnionInPlace(bm)
|
||||
bm = f.storage
|
||||
span.Finish()
|
||||
} else {
|
||||
span, ctx = tracing.StartSpanFromContext(ctx, "importRoaringBitmapUIP")
|
||||
bm.UnionInPlace(f.storage)
|
||||
span.Finish()
|
||||
}
|
||||
|
||||
for rowID := range rowSet {
|
||||
|
|
@ -1869,7 +1882,9 @@ func (f *fragment) importRoaring(data []byte, clear bool) error {
|
|||
}
|
||||
f.cache.Recalculate()
|
||||
|
||||
span, _ = tracing.StartSpanFromContext(ctx, "importRoaring.WriteToFragment")
|
||||
err = unprotectedWriteToFragment(f, bm)
|
||||
span.Finish()
|
||||
return err
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -16,6 +16,7 @@ package pilosa
|
|||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"flag"
|
||||
"fmt"
|
||||
"io"
|
||||
|
|
@ -745,7 +746,7 @@ func BenchmarkFragment_RepeatedSmallImports(b *testing.B) {
|
|||
f := mustOpenFragment("i", "f", viewStandard, 0, "")
|
||||
f.MaxOpN = opN
|
||||
defer f.Clean(b)
|
||||
err := f.importRoaring(getZipfRowsSliceRoaring(uint64(numRows), 1, 0, ShardWidth), false)
|
||||
err := f.importRoaringT(getZipfRowsSliceRoaring(uint64(numRows), 1, 0, ShardWidth), false)
|
||||
if err != nil {
|
||||
b.Fatalf("importing base data for benchmark: %v", err)
|
||||
}
|
||||
|
|
@ -781,14 +782,14 @@ func BenchmarkFragment_RepeatedSmallImportsRoaring(b *testing.B) {
|
|||
f := mustOpenFragment("i", "f", viewStandard, 0, "")
|
||||
f.MaxOpN = opN
|
||||
defer f.Clean(b)
|
||||
err := f.importRoaring(getZipfRowsSliceRoaring(numRows, 1, 0, ShardWidth), false)
|
||||
err := f.importRoaringT(getZipfRowsSliceRoaring(numRows, 1, 0, ShardWidth), false)
|
||||
if err != nil {
|
||||
b.Fatalf("importing base data for benchmark: %v", err)
|
||||
}
|
||||
for i := 0; i < numUpdates; i++ {
|
||||
data := getUpdataRoaring(numRows, bitsPerUpdate, int64(i))
|
||||
b.StartTimer()
|
||||
err := f.importRoaring(data, false)
|
||||
err := f.importRoaringT(data, false)
|
||||
b.StopTimer()
|
||||
if err != nil {
|
||||
b.Fatalf("doing small roaring import: %v", err)
|
||||
|
|
@ -1024,7 +1025,7 @@ func TestFragment_TopN_Intersect_Large(t *testing.T) {
|
|||
if err != nil {
|
||||
t.Fatalf("writing to bytes: %v", err)
|
||||
}
|
||||
err = f.importRoaring(b.Bytes(), false)
|
||||
err = f.importRoaringT(b.Bytes(), false)
|
||||
if err != nil {
|
||||
t.Fatalf("importing data: %v", err)
|
||||
}
|
||||
|
|
@ -2008,7 +2009,7 @@ func BenchmarkImportRoaring(b *testing.B) {
|
|||
for i := 0; i < b.N; i++ {
|
||||
f := mustOpenFragment("i", fmt.Sprintf("r%dc%s", numRows, cacheType), viewStandard, 0, cacheType)
|
||||
b.StartTimer()
|
||||
err := f.importRoaring(data, false)
|
||||
err := f.importRoaringT(data, false)
|
||||
if err != nil {
|
||||
f.Clean(b)
|
||||
b.Fatalf("import error: %v", err)
|
||||
|
|
@ -2041,7 +2042,7 @@ func BenchmarkImportRoaringConcurrent(b *testing.B) {
|
|||
for j := 0; j < concurrency; j++ {
|
||||
j := j
|
||||
eg.Go(func() error {
|
||||
return frags[j].importRoaring(data, false)
|
||||
return frags[j].importRoaringT(data, false)
|
||||
})
|
||||
}
|
||||
err := eg.Wait()
|
||||
|
|
@ -2072,7 +2073,7 @@ func BenchmarkImportRoaringUpdateConcurrent(b *testing.B) {
|
|||
for i := 0; i < b.N; i++ {
|
||||
for j := 0; j < concurrency; j++ {
|
||||
frags[j] = mustOpenFragment("i", "f", viewStandard, uint64(j), CacheTypeRanked)
|
||||
err := frags[j].importRoaring(data, false)
|
||||
err := frags[j].importRoaringT(data, false)
|
||||
if err != nil {
|
||||
b.Fatalf("importing roaring: %v", err)
|
||||
}
|
||||
|
|
@ -2082,7 +2083,7 @@ func BenchmarkImportRoaringUpdateConcurrent(b *testing.B) {
|
|||
for j := 0; j < concurrency; j++ {
|
||||
j := j
|
||||
eg.Go(func() error {
|
||||
return frags[j].importRoaring(updata, false)
|
||||
return frags[j].importRoaringT(updata, false)
|
||||
})
|
||||
}
|
||||
err := eg.Wait()
|
||||
|
|
@ -2138,12 +2139,12 @@ func BenchmarkImportRoaringUpdate(b *testing.B) {
|
|||
b.StopTimer()
|
||||
for i := 0; i < b.N; i++ {
|
||||
f := mustOpenFragment("i", fmt.Sprintf("r%dc%s", numRows, cacheType), viewStandard, 0, cacheType)
|
||||
err := f.importRoaring(data, false)
|
||||
err := f.importRoaringT(data, false)
|
||||
if err != nil {
|
||||
b.Errorf("import error: %v", err)
|
||||
}
|
||||
b.StartTimer()
|
||||
err = f.importRoaring(updata, false)
|
||||
err = f.importRoaringT(updata, false)
|
||||
if err != nil {
|
||||
f.Clean(b)
|
||||
b.Errorf("import error: %v", err)
|
||||
|
|
@ -2174,12 +2175,12 @@ func BenchmarkUpdatePathological(b *testing.B) {
|
|||
for i := 0; i < b.N; i++ {
|
||||
b.StopTimer()
|
||||
f := mustOpenFragment("i", "f", viewStandard, 0, DefaultCacheType)
|
||||
err := f.importRoaring(exists, false)
|
||||
err := f.importRoaringT(exists, false)
|
||||
if err != nil {
|
||||
b.Fatalf("importing roaring: %v", err)
|
||||
}
|
||||
b.StartTimer()
|
||||
err = f.importRoaring(inc, false)
|
||||
err = f.importRoaringT(inc, false)
|
||||
if err != nil {
|
||||
b.Fatalf("importing second: %v", err)
|
||||
}
|
||||
|
|
@ -2196,7 +2197,7 @@ func initBigFrag() {
|
|||
for i := int64(0); i < 10; i++ {
|
||||
// 10 million rows, 1 bit per column, random seeded by i
|
||||
data := getZipfRowsSliceRoaring(10000000, i, 0, ShardWidth)
|
||||
err := f.importRoaring(data, false)
|
||||
err := f.importRoaringT(data, false)
|
||||
if err != nil {
|
||||
panic(fmt.Sprintf("setting up fragment data: %v", err))
|
||||
}
|
||||
|
|
@ -2273,7 +2274,7 @@ func BenchmarkImportRoaringIntoLargeFragment(b *testing.B) {
|
|||
b.Fatalf("opening fragment: %v", err)
|
||||
}
|
||||
b.StartTimer()
|
||||
err = nf.importRoaring(updata, false)
|
||||
err = nf.importRoaringT(updata, false)
|
||||
b.StopTimer()
|
||||
if err != nil {
|
||||
b.Fatalf("bulkImport: %v", err)
|
||||
|
|
@ -2286,7 +2287,7 @@ func BenchmarkImportRoaringIntoLargeFragment(b *testing.B) {
|
|||
func TestGetZipfRowsSliceRoaring(t *testing.T) {
|
||||
f := mustOpenFragment("i", "f", viewStandard, 0, DefaultCacheType)
|
||||
data := getZipfRowsSliceRoaring(10, 1, 0, ShardWidth)
|
||||
err := f.importRoaring(data, false)
|
||||
err := f.importRoaringT(data, false)
|
||||
if err != nil {
|
||||
t.Fatalf("importing roaring: %v", err)
|
||||
}
|
||||
|
|
@ -2431,6 +2432,11 @@ func (f *fragment) Clean(t testing.TB) {
|
|||
}
|
||||
}
|
||||
|
||||
// importRoaringT calls importRoaring with context.Background() for convenience
|
||||
func (f *fragment) importRoaringT(data []byte, clear bool) error {
|
||||
return f.importRoaring(context.Background(), data, clear)
|
||||
}
|
||||
|
||||
// CleanKeep is just like Clean(), but it doesn't remove the
|
||||
// fragment file (note that it DOES remove the cache file).
|
||||
func (f *fragment) CleanKeep(t testing.TB) {
|
||||
|
|
@ -2622,7 +2628,7 @@ func TestFragment_RoaringImport(t *testing.T) {
|
|||
if err != nil {
|
||||
t.Fatalf("writing to buffer: %v", err)
|
||||
}
|
||||
err = f.importRoaring(buf.Bytes(), false)
|
||||
err = f.importRoaringT(buf.Bytes(), false)
|
||||
if err != nil {
|
||||
t.Fatalf("importing roaring: %v", err)
|
||||
}
|
||||
|
|
@ -2697,7 +2703,7 @@ func TestFragment_RoaringImportTopN(t *testing.T) {
|
|||
if err != nil {
|
||||
t.Fatalf("writing to buffer: %v", err)
|
||||
}
|
||||
err = f.importRoaring(buf.Bytes(), false)
|
||||
err = f.importRoaringT(buf.Bytes(), false)
|
||||
if err != nil {
|
||||
t.Fatalf("importing roaring: %v", err)
|
||||
}
|
||||
|
|
|
|||
8
go.mod
8
go.mod
|
|
@ -5,27 +5,31 @@ replace github.com/hashicorp/memberlist => github.com/pilosa/memberlist v0.1.4-0
|
|||
require (
|
||||
github.com/CAFxX/gcnotifier v0.0.0-20190112062741-224a280d589d
|
||||
github.com/DataDog/datadog-go v0.0.0-20180822151419-281ae9f2d895
|
||||
github.com/StackExchange/wmi v0.0.0-20181212234831-e0a55b97c705 // indirect
|
||||
github.com/boltdb/bolt v1.3.1
|
||||
github.com/cespare/xxhash v1.1.0
|
||||
github.com/codahale/hdrhistogram v0.0.0-20161010025455-3a0bb77429bd // indirect
|
||||
github.com/davecgh/go-spew v1.1.1
|
||||
github.com/go-ole/go-ole v1.2.4 // indirect
|
||||
github.com/gogo/protobuf v1.2.0
|
||||
github.com/golang/protobuf v1.2.0
|
||||
github.com/google/go-cmp v0.2.0
|
||||
github.com/gorilla/handlers v1.3.0
|
||||
github.com/gorilla/mux v1.7.0
|
||||
github.com/hashicorp/memberlist v0.1.3
|
||||
github.com/inconshreveable/mousetrap v1.0.0 // indirect
|
||||
github.com/opentracing/opentracing-go v1.0.2
|
||||
github.com/pelletier/go-toml v1.2.0
|
||||
github.com/pkg/errors v0.8.1
|
||||
github.com/remyoudompheng/bigfft v0.0.0-20190321074620-2f0d2b0e0001 // indirect
|
||||
github.com/satori/go.uuid v1.2.0
|
||||
github.com/shirou/gopsutil v2.18.12+incompatible
|
||||
github.com/shirou/w32 v0.0.0-20160930032740-bb4de0191aa4 // indirect
|
||||
github.com/spf13/cobra v0.0.3
|
||||
github.com/spf13/pflag v1.0.3
|
||||
github.com/spf13/viper v1.3.1
|
||||
github.com/uber/jaeger-client-go v2.15.0+incompatible
|
||||
github.com/uber/jaeger-lib v1.5.0
|
||||
github.com/uber/jaeger-client-go v2.16.0+incompatible
|
||||
github.com/uber/jaeger-lib v2.0.0+incompatible // indirect
|
||||
golang.org/x/crypto v0.0.0-20190426145343-a29dc8fdc734 // indirect
|
||||
golang.org/x/net v0.0.0-20190424112056-4829fb13d2c6 // indirect
|
||||
golang.org/x/sync v0.0.0-20190423024810-112230192c58
|
||||
|
|
|
|||
10
go.sum
10
go.sum
|
|
@ -3,6 +3,8 @@ github.com/CAFxX/gcnotifier v0.0.0-20190112062741-224a280d589d/go.mod h1:Rn2zM2M
|
|||
github.com/DataDog/datadog-go v0.0.0-20180822151419-281ae9f2d895 h1:dmc/C8bpE5VkQn65PNbbyACDC8xw8Hpp/NEurdPmQDQ=
|
||||
github.com/DataDog/datadog-go v0.0.0-20180822151419-281ae9f2d895/go.mod h1:LButxg5PwREeZtORoXG3tL4fMGNddJ+vMq1mwgfaqoQ=
|
||||
github.com/OneOfOne/xxhash v1.2.2/go.mod h1:HSdplMjZKSmBqAxg5vPj2TmRDmfkzw+cTzAElWljhcU=
|
||||
github.com/StackExchange/wmi v0.0.0-20181212234831-e0a55b97c705 h1:UUppSQnhf4Yc6xGxSkoQpPhb7RVzuv5Nb1mwJ5VId9s=
|
||||
github.com/StackExchange/wmi v0.0.0-20181212234831-e0a55b97c705/go.mod h1:3eOhrUMpNV+6aFIbp5/iudMxNCF27Vw2OZgy4xEx0Fg=
|
||||
github.com/armon/consul-api v0.0.0-20180202201655-eb2c6b5be1b6/go.mod h1:grANhF5doyWs3UAsr3K4I6qtAmlQcZDesFNEHPZAzj8=
|
||||
github.com/armon/go-metrics v0.0.0-20180917152333-f0300d1749da h1:8GUt8eRujhVEGZFFEjBj46YV4rDjvGrNxb0KMWYkL2I=
|
||||
github.com/armon/go-metrics v0.0.0-20180917152333-f0300d1749da/go.mod h1:Q73ZrmVTwzkszR9V5SSuryQ31EELlFMUz1kKyl939pY=
|
||||
|
|
@ -19,6 +21,8 @@ github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c
|
|||
github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
|
||||
github.com/fsnotify/fsnotify v1.4.7 h1:IXs+QLmnXW2CcXuY+8Mzv/fWEsPGWxqefPtCP5CnV9I=
|
||||
github.com/fsnotify/fsnotify v1.4.7/go.mod h1:jwhsz4b93w/PPRr/qN1Yymfu8t87LnFCMoQvtojpjFo=
|
||||
github.com/go-ole/go-ole v1.2.4 h1:nNBDSCOigTSiarFpYE9J/KtEA1IOW4CNeqT9TQDqCxI=
|
||||
github.com/go-ole/go-ole v1.2.4/go.mod h1:XCwSNxSkXRo4vlyPy93sltvi/qJq0jqQhjqQNIwKuxM=
|
||||
github.com/gogo/protobuf v1.2.0 h1:xU6/SpYbvkNYiptHJYEDRseDLvYE7wSqhYYNy0QSUzI=
|
||||
github.com/gogo/protobuf v1.2.0/go.mod h1:r8qH/GZQm5c6nD/R0oafs1akxWv10x8SbQlK7atdtwQ=
|
||||
github.com/golang/protobuf v1.2.0 h1:P3YflyNX/ehuJFLhxviNdFxQPkGK5cDcApsge1SqnvM=
|
||||
|
|
@ -50,6 +54,8 @@ github.com/hashicorp/hcl v1.0.0 h1:0Anlzjpi4vEasTeNFn2mLJgTSwt0+6sfsiTG8qcWGx4=
|
|||
github.com/hashicorp/hcl v1.0.0/go.mod h1:E5yfLk+7swimpb2L/Alb/PJmXilQ/rhwaUYs4T20WEQ=
|
||||
github.com/hashicorp/memberlist v0.1.3 h1:EmmoJme1matNzb+hMpDuR/0sbJSUisxyqBGG676r31M=
|
||||
github.com/hashicorp/memberlist v0.1.3/go.mod h1:ajVTdAv/9Im8oMAAj5G31PhhMCZJV2pPBoIllUwCN7I=
|
||||
github.com/inconshreveable/mousetrap v1.0.0 h1:Z8tu5sraLXCXIcARxBp/8cbvlwVa7Z1NHg9XEKhtSvM=
|
||||
github.com/inconshreveable/mousetrap v1.0.0/go.mod h1:PxqpIevigyE2G7u3NXJIT2ANytuPF1OarO4DADm73n8=
|
||||
github.com/magiconair/properties v1.8.0 h1:LLgXmsheXeRoUOBOjtwPQCWIYqM/LU1ayDtDePerRcY=
|
||||
github.com/magiconair/properties v1.8.0/go.mod h1:PppfXfuXeibc/6YijjN8zIbojt8czPbwD3XqdrwzmxQ=
|
||||
github.com/miekg/dns v1.0.14 h1:9jZdLNd/P4+SfEJ0TNyxYpsK8N4GtfylBLqtbYN1sbA=
|
||||
|
|
@ -80,6 +86,8 @@ github.com/sean-/seed v0.0.0-20170313163322-e2103e2c3529 h1:nn5Wsu0esKSJiIVhscUt
|
|||
github.com/sean-/seed v0.0.0-20170313163322-e2103e2c3529/go.mod h1:DxrIzT+xaE7yg65j358z/aeFdxmN0P9QXhEzd20vsDc=
|
||||
github.com/shirou/gopsutil v2.18.12+incompatible h1:1eaJvGomDnH74/5cF4CTmTbLHAriGFsTZppLXDX93OM=
|
||||
github.com/shirou/gopsutil v2.18.12+incompatible/go.mod h1:5b4v6he4MtMOwMlS0TUMTu2PcXUg8+E1lC7eC3UO/RA=
|
||||
github.com/shirou/w32 v0.0.0-20160930032740-bb4de0191aa4 h1:udFKJ0aHUL60LboW/A+DfgoHVedieIzIXE8uylPue0U=
|
||||
github.com/shirou/w32 v0.0.0-20160930032740-bb4de0191aa4/go.mod h1:qsXQc7+bwAM3Q1u/4XEfrquwF8Lw7D7y5cD8CuHnfIc=
|
||||
github.com/spaolacci/murmur3 v0.0.0-20180118202830-f09979ecbc72/go.mod h1:JwIasOWyU6f++ZhiEuf87xNszmSA2myDM2Kzu9HwQUA=
|
||||
github.com/spf13/afero v1.1.2 h1:m8/z1t7/fwjysjQRYbP0RD+bUIF/8tJwPdEZsI83ACI=
|
||||
github.com/spf13/afero v1.1.2/go.mod h1:j4pytiNVoe2o6bmDsKpLACNPDBIoEAkihy7loJ1B0CQ=
|
||||
|
|
@ -96,6 +104,8 @@ github.com/spf13/viper v1.3.1/go.mod h1:ZiWeW+zYFKm7srdB9IoDzzZXaJaI5eL9QjNiN/DM
|
|||
github.com/stretchr/testify v1.2.2/go.mod h1:a8OnRcib4nhh0OaRAV+Yts87kKdq0PP7pXfy6kDkUVs=
|
||||
github.com/uber/jaeger-client-go v2.15.0+incompatible h1:NP3qsSqNxh8VYr956ur1N/1C1PjvOJnJykCzcD5QHbk=
|
||||
github.com/uber/jaeger-client-go v2.15.0+incompatible/go.mod h1:WVhlPFC8FDjOFMMWRy2pZqQJSXxYSwNYOkTr/Z6d3Kk=
|
||||
github.com/uber/jaeger-client-go v2.16.0+incompatible h1:Q2Pp6v3QYiocMxomCaJuwQGFt7E53bPYqEgug/AoBtY=
|
||||
github.com/uber/jaeger-client-go v2.16.0+incompatible/go.mod h1:WVhlPFC8FDjOFMMWRy2pZqQJSXxYSwNYOkTr/Z6d3Kk=
|
||||
github.com/uber/jaeger-lib v1.5.0 h1:OHbgr8l656Ub3Fw5k9SWnBfIEwvoHQ+W2y+Aa9D1Uyo=
|
||||
github.com/uber/jaeger-lib v1.5.0/go.mod h1:ComeNDZlWwrWnDv8aPp0Ba6+uUTzImX/AauajbLI56U=
|
||||
github.com/uber/jaeger-lib v2.0.0+incompatible h1:iMSCV0rmXEogjNWPh2D0xk9YVKvrtGoHJNe9ebLu/pw=
|
||||
|
|
|
|||
|
|
@ -1617,15 +1617,23 @@ func (h *Handler) handlePostImportRoaring(w http.ResponseWriter, r *http.Request
|
|||
remote = true
|
||||
}
|
||||
|
||||
ctx := r.Context()
|
||||
|
||||
// Read entire body.
|
||||
span, _ := tracing.StartSpanFromContext(ctx, "ioutil.ReadAll-Body")
|
||||
body, err := ioutil.ReadAll(r.Body)
|
||||
span.LogKV("bodySize", len(body))
|
||||
span.Finish()
|
||||
if err != nil {
|
||||
http.Error(w, err.Error(), http.StatusBadRequest)
|
||||
return
|
||||
}
|
||||
|
||||
req := &pilosa.ImportRoaringRequest{}
|
||||
if err := h.api.Serializer.Unmarshal(body, req); err != nil {
|
||||
span, _ = tracing.StartSpanFromContext(ctx, "Unmarshal")
|
||||
err = h.api.Serializer.Unmarshal(body, req)
|
||||
span.Finish()
|
||||
if err != nil {
|
||||
http.Error(w, err.Error(), http.StatusBadRequest)
|
||||
return
|
||||
}
|
||||
|
|
@ -1639,7 +1647,7 @@ func (h *Handler) handlePostImportRoaring(w http.ResponseWriter, r *http.Request
|
|||
|
||||
resp := &pilosa.ImportResponse{}
|
||||
// TODO give meaningful stats for import
|
||||
err = h.api.ImportRoaring(r.Context(), indexName, fieldName, shard, remote, req)
|
||||
err = h.api.ImportRoaring(ctx, indexName, fieldName, shard, remote, req)
|
||||
if err != nil {
|
||||
resp.Err = err.Error()
|
||||
if _, ok := err.(pilosa.BadRequestError); ok {
|
||||
|
|
@ -1648,6 +1656,7 @@ func (h *Handler) handlePostImportRoaring(w http.ResponseWriter, r *http.Request
|
|||
w.WriteHeader(http.StatusInternalServerError)
|
||||
}
|
||||
}
|
||||
|
||||
// Marshal response object.
|
||||
buf, err := h.api.Serializer.Marshal(resp)
|
||||
if err != nil {
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue