mirror of
https://github.com/featurebasedb/featurebase.git
synced 2026-08-28 10:54:59 +00:00
cf97a0dcb8
We switch everything to use QueryContext/QueryRead/etc instead of Qcx/Tx. We drop the short_txkey subpackage (it's now handled by either keys or querycontext). We drop all the dbshard stuff, and all the tx/txfactory stuff. We remove all the things that related to the old "Block" concept, which was mostly used by the anti-entropy code, but had one fragmentary usage left in the ImportRoaringOverwrite case of ImportRoaring. That's replaced by using a rewriter that deletes all bits (not just bits in specific columns) from an existing thing, but writes in new bits. Actually we could probably do that better with a custom "eradicate-rewriter" that doesn't try to be clever, and just eliminates things. This includes a number of minor bug fixes that were exposed by getting the testing to work. For example: * When checking whether an operation "requires write", we now consider a Delete a kind of a Write, because it is. * Several tests were relying on the fact that writes through Qcx were being committed whether or not the Qcx was ever told to finish. With QueryContext, you actually have to reach a Commit() or the writes don't happen (except for special cases in Delete). * Replaced a lot of panics with t.Fatalf in tests. There's also some minor staticcheck fixes, like deleting the unused "db" member of a boltdb transaction wrapper.
1594 lines
48 KiB
Go
1594 lines
48 KiB
Go
// Copyright 2021 Molecula Corp. All rights reserved.
|
|
package pilosa_test
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"encoding/hex"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"io"
|
|
"math"
|
|
"math/rand"
|
|
"net/http"
|
|
"net/http/httptest"
|
|
"os"
|
|
"path/filepath"
|
|
"reflect"
|
|
"sort"
|
|
"strings"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/golang-jwt/jwt"
|
|
pilosa "github.com/molecula/featurebase/v3"
|
|
"github.com/molecula/featurebase/v3/authn"
|
|
qc "github.com/molecula/featurebase/v3/querycontext"
|
|
"github.com/molecula/featurebase/v3/roaring"
|
|
"github.com/molecula/featurebase/v3/server"
|
|
"github.com/molecula/featurebase/v3/shardwidth"
|
|
"github.com/molecula/featurebase/v3/test"
|
|
. "github.com/molecula/featurebase/v3/vprint" // nolint:staticcheck
|
|
"github.com/stretchr/testify/require"
|
|
|
|
"golang.org/x/sync/errgroup"
|
|
)
|
|
|
|
// mustQueryContext gets a write query context for the API and the
|
|
// specified index, or index-and-shards, or fails
|
|
// the test. The query context will be released at the end of the test.
|
|
func mustIndexQueryContext(tb testing.TB, api *pilosa.API, index string, shards ...uint64) qc.QueryContext {
|
|
tb.Helper()
|
|
// disregard a leading ^0, because that's idiomatic for "all shards"
|
|
if len(shards) > 0 && shards[0] == ^uint64(0) {
|
|
shards = shards[1:]
|
|
}
|
|
qcx, err := api.NewIndexQueryContext(context.Background(), index, shards...)
|
|
if err != nil {
|
|
tb.Fatalf("creating query context: %v", err)
|
|
}
|
|
tb.Cleanup(qcx.Release)
|
|
return qcx
|
|
}
|
|
|
|
// mustQueryContext gets a read-only query context for the API, or fails
|
|
// the test. The query context will be released at the end of the test.
|
|
func mustQueryContext(tb testing.TB, api *pilosa.API) qc.QueryContext {
|
|
tb.Helper()
|
|
qcx, err := api.NewQueryContext(context.Background())
|
|
if err != nil {
|
|
tb.Fatalf("creating query context: %v", err)
|
|
}
|
|
tb.Cleanup(qcx.Release)
|
|
return qcx
|
|
}
|
|
|
|
func TestAPI_Import(t *testing.T) {
|
|
c := test.MustRunCluster(t, 3)
|
|
defer c.Close()
|
|
|
|
m0 := c.GetNode(0)
|
|
m1 := c.GetNode(1)
|
|
|
|
indexNames := map[bool]string{false: c.Idx("u"), true: c.Idx("k")}
|
|
fieldNames := map[bool]string{false: "f", true: "kf"}
|
|
|
|
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)
|
|
}
|
|
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)
|
|
}
|
|
}
|
|
}
|
|
|
|
N := 10
|
|
|
|
// 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
|
|
req := &pilosa.ImportRequest{
|
|
Index: indexNames[true],
|
|
Field: fieldNames[false],
|
|
Shard: 0, // inaccurate, but keys override it
|
|
RowIDs: rowIDs,
|
|
ColumnKeys: colKeys,
|
|
}
|
|
|
|
qcx := mustIndexQueryContext(t, m0.API, req.Index)
|
|
|
|
if err := m0.API.Import(ctx, qcx, req); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
require.Nil(t, qcx.Commit())
|
|
|
|
pql := fmt.Sprintf("Row(%s=%d)", fieldNames[false], rowIDs[0])
|
|
|
|
// Query node0.
|
|
var keys []string
|
|
if res, err := m0.API.Query(ctx, &pilosa.QueryRequest{Index: indexNames[true], Query: pql}); err != nil {
|
|
t.Fatal(err)
|
|
} 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: indexNames[true], Query: pql}); err != nil {
|
|
return err
|
|
} 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
|
|
}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
})
|
|
t.Run("ExpectedErrors", func(t *testing.T) {
|
|
t.Skip("partitioning strategy changed, test not supported") // skipping due to change partitioning strategy
|
|
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 := mustIndexQueryContext(t, m0.API, req.Index)
|
|
defer qcx.Release()
|
|
err := m0.API.Import(ctx, qcx, req.Clone())
|
|
require.Nil(t, qcx.Commit())
|
|
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.
|
|
t.Run("SchemaHasNoExists", func(t *testing.T) {
|
|
schema, err := m1.API.Schema(context.Background(), false)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
|
|
for _, f := range schema[0].Fields {
|
|
if f.Name == "_exists" {
|
|
t.Fatalf("found _exists field in schema")
|
|
}
|
|
if strings.HasPrefix(f.Name, "_") {
|
|
t.Fatalf("found internal field '%s' in schema output", f.Name)
|
|
}
|
|
}
|
|
})
|
|
}
|
|
|
|
func TestAPI_ImportValue(t *testing.T) {
|
|
c := test.MustRunCluster(t, 3)
|
|
defer c.Close()
|
|
|
|
coord := c.GetPrimary()
|
|
m0 := c.GetNode(0)
|
|
m1 := c.GetNode(1)
|
|
m2 := c.GetNode(2)
|
|
|
|
t.Run("ValColumnKey", func(t *testing.T) {
|
|
ctx := context.Background()
|
|
index := c.Idx("valck")
|
|
field := "f"
|
|
|
|
_, err := coord.API.CreateIndex(ctx, index, pilosa.IndexOptions{Keys: true})
|
|
if err != nil {
|
|
t.Fatalf("creating index: %v", err)
|
|
}
|
|
_, err = coord.API.CreateField(ctx, index, field, pilosa.OptFieldTypeInt(math.MinInt64, math.MaxInt64))
|
|
if err != nil {
|
|
t.Fatalf("creating field: %v", err)
|
|
}
|
|
|
|
// Generate some keyed records.
|
|
values := []int64{}
|
|
for i := 1; i <= 10; i++ {
|
|
values = append(values, int64(i))
|
|
}
|
|
|
|
// Column keys are sharded so their order is not guaranteed.
|
|
colKeys := []string{"col10", "col8", "col9", "col6", "col7", "col4", "col5", "col2", "col3", "col1"}
|
|
|
|
// Import data with keys to the primary and verify that it gets
|
|
// translated and forwarded to the owner of shard 0
|
|
req := &pilosa.ImportValueRequest{
|
|
Index: index,
|
|
Field: field,
|
|
ColumnKeys: colKeys,
|
|
Values: values,
|
|
Shard: 0, // inaccurate but keys override it
|
|
}
|
|
|
|
qcx := mustIndexQueryContext(t, coord.API, req.Index)
|
|
if err := coord.API.ImportValue(ctx, qcx, req); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
require.Nil(t, qcx.Commit())
|
|
|
|
pql := fmt.Sprintf("Row(%s>0)", field)
|
|
|
|
// Query node0.
|
|
res, err := m0.API.Query(ctx, &pilosa.QueryRequest{Index: index, Query: pql})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
keys := res.Results[0].(*pilosa.Row).Keys
|
|
if !sameStringSlice(keys, colKeys) {
|
|
t.Fatalf("unexpected column keys: %+v", keys)
|
|
}
|
|
|
|
// Query node1.
|
|
if err := test.RetryUntil(5*time.Second, func() error {
|
|
res, err := m1.API.Query(ctx, &pilosa.QueryRequest{Index: index, Query: pql})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
keys := res.Results[0].(*pilosa.Row).Keys
|
|
if !sameStringSlice(keys, colKeys) {
|
|
t.Fatalf("unexpected column keys: %+v", keys)
|
|
}
|
|
return nil
|
|
}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
})
|
|
|
|
t.Run("ValIntEmpty", func(t *testing.T) {
|
|
ctx := context.Background()
|
|
index := c.Idx("valintempty")
|
|
field := "fld"
|
|
createIndexForTest(index, coord, t)
|
|
createFieldForTest(index, field, coord, t)
|
|
|
|
// Column keys are sharded so their order is not guaranteed.
|
|
colKeys := []string{"col2", "col1", "col3"}
|
|
values := []int64{1, 2, 3, 4}
|
|
|
|
// Import without data, verify that it succeeds
|
|
req := &pilosa.ImportValueRequest{
|
|
Index: index,
|
|
Field: field,
|
|
}
|
|
qcx1 := mustIndexQueryContext(t, coord.API, req.Index)
|
|
defer qcx1.Release()
|
|
|
|
// Import with empty request, should succeed
|
|
if err := coord.API.ImportValue(ctx, qcx1, req); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
require.Nil(t, qcx1.Commit())
|
|
|
|
// Import without data but with columnkeys, verify that it errors
|
|
req.ColumnKeys = colKeys
|
|
qcx2 := mustIndexQueryContext(t, coord.API, req.Index)
|
|
defer qcx2.Release()
|
|
if err := coord.API.ImportValue(ctx, qcx2, req); err == nil {
|
|
t.Fatal("expected error but succeeded")
|
|
}
|
|
require.Nil(t, qcx2.Commit())
|
|
|
|
// Import with mismatch column and value lengths
|
|
req.Values = values
|
|
qcx3 := mustIndexQueryContext(t, coord.API, req.Index)
|
|
defer qcx3.Release()
|
|
if err := coord.API.ImportValue(ctx, qcx3, req); err == nil {
|
|
t.Fatal("expected error but succeeded")
|
|
}
|
|
require.Nil(t, qcx3.Commit())
|
|
|
|
// Import with data but no columns
|
|
req.ColumnKeys = make([]string, 0)
|
|
qcx4 := mustIndexQueryContext(t, coord.API, req.Index)
|
|
defer qcx4.Release()
|
|
if err := coord.API.ImportValue(ctx, qcx4, req); err == nil {
|
|
t.Fatal("expected error but succeeded")
|
|
}
|
|
require.Nil(t, qcx4.Commit())
|
|
|
|
})
|
|
|
|
t.Run("ValDecimalField", func(t *testing.T) {
|
|
t.Skip() // skipping due to change partitioning strategy
|
|
ctx := context.Background()
|
|
index := "valdec"
|
|
field := "fdec"
|
|
_, err := m2.API.CreateIndex(ctx, index, pilosa.IndexOptions{})
|
|
if err != nil {
|
|
t.Fatalf("creating index: %v", err)
|
|
}
|
|
_, err = m2.API.CreateField(ctx, index, field, pilosa.OptFieldTypeDecimal(1))
|
|
if err != nil {
|
|
t.Fatalf("creating field: %v", err)
|
|
}
|
|
// Generate some records.
|
|
values := []float64{}
|
|
colIDs := []uint64{}
|
|
for i := 0; i < 10; i++ {
|
|
values = append(values, float64(i)+0.1)
|
|
colIDs = append(colIDs, uint64(i))
|
|
}
|
|
// Import data with keys to node1 and verify that it gets translated and
|
|
// forwarded to the owner of shard 0
|
|
req := &pilosa.ImportValueRequest{
|
|
Index: index,
|
|
Field: field,
|
|
ColumnIDs: colIDs,
|
|
FloatValues: values,
|
|
}
|
|
qcx := mustIndexQueryContext(t, m0.API, req.Index)
|
|
if err := m0.API.ImportValue(ctx, qcx, req); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
require.Nil(t, qcx.Commit())
|
|
query := fmt.Sprintf("Row(%s>6)", field)
|
|
// Query node0.
|
|
if res, err := m0.API.Query(ctx, &pilosa.QueryRequest{Index: index, Query: query}); err != nil {
|
|
t.Fatal(err)
|
|
} else if ids := res.Results[0].(*pilosa.Row).Columns(); !reflect.DeepEqual(ids, colIDs[6:]) {
|
|
t.Fatalf("unexpected column keys: observerd %+v; expected '%+v'", ids, colIDs[6:])
|
|
}
|
|
})
|
|
|
|
t.Run("ValDecimalFieldNegativeScale", func(t *testing.T) {
|
|
ctx := context.Background()
|
|
index := c.Idx("valdecneg")
|
|
field := "fdecneg"
|
|
|
|
_, err := m0.API.CreateIndex(ctx, index, pilosa.IndexOptions{})
|
|
if err != nil {
|
|
t.Fatalf("creating index: %v", err)
|
|
}
|
|
_, err = m0.API.CreateField(ctx, index, field, pilosa.OptFieldTypeDecimal(-1))
|
|
if err == nil {
|
|
t.Fatal("expected error creating field")
|
|
}
|
|
})
|
|
|
|
t.Run("ValTimestampField", func(t *testing.T) {
|
|
t.Skip("partition strategy change invalidated") // skipping due to change partitioning strategy
|
|
ctx := context.Background()
|
|
index := c.Idx("valts")
|
|
field := "fts"
|
|
|
|
_, err := m1.API.CreateIndex(ctx, index, pilosa.IndexOptions{})
|
|
if err != nil {
|
|
t.Fatalf("creating index: %v", err)
|
|
}
|
|
_, err = m1.API.CreateField(ctx, index, field, pilosa.OptFieldTypeTimestamp(pilosa.DefaultEpoch, pilosa.TimeUnitSeconds))
|
|
if err != nil {
|
|
t.Fatalf("creating field: %v", err)
|
|
}
|
|
|
|
// Generate some records.
|
|
values := []time.Time{}
|
|
colIDs := []uint64{}
|
|
for i := 0; i < 10; i++ {
|
|
values = append(values, pilosa.MinTimestamp.Add(time.Duration(i)*time.Second))
|
|
colIDs = append(colIDs, uint64(i))
|
|
}
|
|
|
|
// Import data with keys to node1 and verify that it gets translated and
|
|
// forwarded to the owner of shard 0
|
|
req := &pilosa.ImportValueRequest{
|
|
Index: index,
|
|
Field: field,
|
|
ColumnIDs: colIDs,
|
|
TimestampValues: values,
|
|
}
|
|
|
|
qcx := mustIndexQueryContext(t, m2.API, req.Index)
|
|
if err := m2.API.ImportValue(ctx, qcx, req); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
require.Nil(t, qcx.Commit())
|
|
|
|
query := fmt.Sprintf("Row(%s>='1833-11-24T17:31:50Z')", field) // 6s after MinTimestamp
|
|
|
|
// Query node0.
|
|
if res, err := m0.API.Query(ctx, &pilosa.QueryRequest{Index: index, Query: query}); err != nil {
|
|
t.Fatal(err)
|
|
} else if ids := res.Results[0].(*pilosa.Row).Columns(); !reflect.DeepEqual(ids, colIDs[6:]) {
|
|
t.Fatalf("unexpected column keys: observerd %+v; expected '%+v'", ids, colIDs[6:])
|
|
}
|
|
})
|
|
|
|
t.Run("ValStringField", func(t *testing.T) {
|
|
t.Skip("partition strategy change invalidated") // skipping due to change partitioning strategy
|
|
ctx := context.Background()
|
|
index := c.Idx("valstr")
|
|
field := "fstr"
|
|
|
|
fgnIndex := "fgnvalstr"
|
|
|
|
_, err := coord.API.CreateIndex(ctx, index, pilosa.IndexOptions{})
|
|
if err != nil {
|
|
t.Fatalf("creating index: %v", err)
|
|
}
|
|
|
|
_, err = coord.API.CreateIndex(ctx, fgnIndex, pilosa.IndexOptions{Keys: true})
|
|
if err != nil {
|
|
t.Fatalf("creating foreign index: %v", err)
|
|
}
|
|
_, err = coord.API.CreateField(ctx, index, field,
|
|
pilosa.OptFieldTypeInt(0, math.MaxInt64),
|
|
pilosa.OptFieldForeignIndex(fgnIndex),
|
|
)
|
|
if err != nil {
|
|
t.Fatalf("creating field: %v", err)
|
|
}
|
|
|
|
// Generate some keyed records.
|
|
values := []string{}
|
|
colIDs := []uint64{}
|
|
for i := 0; i < 10; i++ {
|
|
value := fmt.Sprintf("strval-%d", (i)*100+10)
|
|
values = append(values, value)
|
|
colIDs = append(colIDs, uint64(i))
|
|
}
|
|
|
|
// Import data with keys to the node0 and verify that it gets translated
|
|
// and forwarded to the owner of shard 0
|
|
req := &pilosa.ImportValueRequest{
|
|
Index: index,
|
|
Field: field,
|
|
ColumnIDs: colIDs,
|
|
StringValues: values,
|
|
}
|
|
qcx := mustIndexQueryContext(t, m0.API, req.Index)
|
|
if err := m0.API.ImportValue(ctx, qcx, req); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
require.Nil(t, qcx.Commit())
|
|
|
|
pql := fmt.Sprintf(`Row(%s=="strval-110")`, field)
|
|
|
|
// Query node1.
|
|
if res, err := m1.API.Query(ctx, &pilosa.QueryRequest{Index: index, Query: pql}); err != nil {
|
|
t.Fatal(err)
|
|
} else if ids := res.Results[0].(*pilosa.Row).Columns(); !reflect.DeepEqual(ids, []uint64{1}) {
|
|
t.Fatalf("unexpected columns: observerd %+v; expected '%+v'", ids, []uint64{1})
|
|
}
|
|
})
|
|
}
|
|
|
|
func TestAPI_Ingest(t *testing.T) {
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
defer cancel()
|
|
c := test.MustRunCluster(t, 1)
|
|
defer c.Close()
|
|
coord := c.GetPrimary()
|
|
index := c.Idx()
|
|
setField := "set"
|
|
timeField := "tq"
|
|
intField := "int"
|
|
|
|
_, err := coord.API.CreateIndex(ctx, index, pilosa.IndexOptions{Keys: false, TrackExistence: true})
|
|
if err != nil {
|
|
t.Fatalf("creating index: %v", err)
|
|
}
|
|
_, err = coord.API.CreateField(ctx, index, setField, pilosa.OptFieldTypeSet("none", 0))
|
|
if err != nil {
|
|
t.Fatalf("creating field: %v", err)
|
|
}
|
|
_, err = coord.API.CreateField(ctx, index, timeField, pilosa.OptFieldTypeTime("YMD", "0"))
|
|
if err != nil {
|
|
t.Fatalf("creating field: %v", err)
|
|
}
|
|
_, err = coord.API.CreateField(ctx, index, intField, pilosa.OptFieldTypeInt(0, 100000))
|
|
if err != nil {
|
|
t.Fatalf("creating field: %v", err)
|
|
}
|
|
|
|
t.Run("ImportRoaringShard", func(t *testing.T) {
|
|
setBuf := &bytes.Buffer{}
|
|
setBits := roaring.NewBitmap(7, pilosa.ShardWidth+7)
|
|
_, _ = setBits.WriteTo(setBuf) // bytes.Buffer never errors
|
|
intBuf := &bytes.Buffer{}
|
|
intBits := roaring.NewBitmap(7, pilosa.ShardWidth*2+7)
|
|
_, _ = intBits.WriteTo(intBuf) // bytes.Buffer never errors
|
|
request := &pilosa.ImportRoaringShardRequest{
|
|
Remote: true,
|
|
Views: []pilosa.RoaringUpdate{
|
|
{
|
|
Field: setField,
|
|
View: "standard",
|
|
Set: setBuf.Bytes(),
|
|
},
|
|
{
|
|
Field: intField,
|
|
View: "bsig_" + intField,
|
|
Set: intBuf.Bytes(),
|
|
},
|
|
},
|
|
}
|
|
if err := coord.API.ImportRoaringShard(context.Background(), c.Idx(), 8, request); err != nil {
|
|
t.Fatalf("ingesting: %v", err)
|
|
}
|
|
|
|
mustQuery := func(t *testing.T, index, query string) pilosa.QueryResponse {
|
|
res, err := coord.API.Query(context.Background(), &pilosa.QueryRequest{Index: index, Query: query})
|
|
if err != nil {
|
|
t.Fatalf("querying: %v", err)
|
|
}
|
|
return res
|
|
}
|
|
|
|
res := mustQuery(t, c.Idx(), "Row(set=0)")
|
|
r := res.Results[0].(*pilosa.Row).Columns()
|
|
if len(r) != 1 || r[0] != pilosa.ShardWidth*8+7 {
|
|
t.Fatalf("expected row with pilosa.ShardWidth*8+7 set, got %d", r)
|
|
}
|
|
|
|
res = mustQuery(t, c.Idx(), "Row(set=1)")
|
|
r = res.Results[0].(*pilosa.Row).Columns()
|
|
if len(r) != 1 || r[0] != pilosa.ShardWidth*8+7 {
|
|
t.Fatalf("expected row with pilosa.ShardWidth*8+7 set, got %d", r)
|
|
}
|
|
|
|
res = mustQuery(t, c.Idx(), "Row(int==1)")
|
|
r = res.Results[0].(*pilosa.Row).Columns()
|
|
if len(r) != 1 || r[0] != pilosa.ShardWidth*8+7 {
|
|
t.Fatalf("expected row with, pilosa.ShardWidth*8+7 set, got %d", r)
|
|
}
|
|
|
|
request = &pilosa.ImportRoaringShardRequest{
|
|
Remote: true,
|
|
Views: []pilosa.RoaringUpdate{
|
|
{
|
|
Field: setField,
|
|
View: "standard",
|
|
Clear: setBuf.Bytes(),
|
|
},
|
|
{
|
|
Field: intField,
|
|
View: "bsig_" + intField,
|
|
Clear: intBuf.Bytes(),
|
|
},
|
|
},
|
|
}
|
|
if err := coord.API.ImportRoaringShard(context.Background(), c.Idx(), 8, request); err != nil {
|
|
t.Fatalf("ingesting: %v", err)
|
|
}
|
|
|
|
res = mustQuery(t, c.Idx(), "Row(set=0)")
|
|
r = res.Results[0].(*pilosa.Row).Columns()
|
|
if len(r) != 0 {
|
|
t.Fatalf("expected no values after clearing, got: %v", r)
|
|
}
|
|
|
|
res = mustQuery(t, c.Idx(), "Row(set=1)")
|
|
r = res.Results[0].(*pilosa.Row).Columns()
|
|
if len(r) != 0 {
|
|
t.Fatalf("expected no values after clearing, got: %v", r)
|
|
}
|
|
|
|
res = mustQuery(t, c.Idx(), "Row(int==1)")
|
|
r = res.Results[0].(*pilosa.Row).Columns()
|
|
if len(r) != 0 {
|
|
t.Fatalf("expected no values after clearing, got: %v", r)
|
|
}
|
|
|
|
})
|
|
}
|
|
|
|
func TestAPI_ClearFlagForImportAndImportValues(t *testing.T) {
|
|
c := test.MustRunCluster(t, 1)
|
|
defer c.Close()
|
|
|
|
// plan:
|
|
// 1. set a bit
|
|
// 2. clear with Import() using the ImportRequest.Clear flag
|
|
// 3. verifiy the clear is done.
|
|
// repeat for ImportValueRequest and ImportValues()
|
|
|
|
m0 := c.GetNode(0)
|
|
m0api := m0.API
|
|
|
|
ctx := context.Background()
|
|
index := c.Idx()
|
|
fieldAcct0 := "acct0"
|
|
|
|
opts := pilosa.OptFieldTypeInt(-1000, 1000)
|
|
|
|
_, err := m0api.CreateIndex(ctx, index, pilosa.IndexOptions{})
|
|
if err != nil {
|
|
t.Fatalf("creating index: %v", err)
|
|
}
|
|
_, err = m0api.CreateField(ctx, index, fieldAcct0, opts)
|
|
if err != nil {
|
|
t.Fatalf("creating fieldAcct0: %v", err)
|
|
}
|
|
|
|
iraField := "ira" // set field.
|
|
iraRowID := uint64(3)
|
|
_, err = m0api.CreateField(ctx, index, iraField)
|
|
if err != nil {
|
|
t.Fatalf("creating fieldIRA: %v", err)
|
|
}
|
|
|
|
acctOwnerID := uint64(78) // ColumnID
|
|
shard := acctOwnerID / ShardWidth
|
|
acct0bal := int64(500)
|
|
|
|
ivr0 := &pilosa.ImportValueRequest{
|
|
Index: index,
|
|
Field: fieldAcct0,
|
|
Shard: shard,
|
|
ColumnIDs: []uint64{acctOwnerID},
|
|
Values: []int64{acct0bal},
|
|
}
|
|
ir0 := &pilosa.ImportRequest{
|
|
Index: index,
|
|
Field: iraField,
|
|
Shard: shard,
|
|
ColumnIDs: []uint64{acctOwnerID},
|
|
RowIDs: []uint64{iraRowID},
|
|
}
|
|
|
|
qcx := mustIndexQueryContext(t, m0api, index)
|
|
if err := m0api.Import(ctx, qcx, ir0.Clone()); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err := m0api.ImportValue(ctx, qcx, ivr0.Clone()); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
require.Nil(t, qcx.Commit())
|
|
|
|
bitIsSet := func() bool {
|
|
query := fmt.Sprintf("Row(%v=%v)", iraField, iraRowID)
|
|
res, err := m0api.Query(context.Background(), &pilosa.QueryRequest{Index: index, Query: query})
|
|
PanicOn(err)
|
|
cols := res.Results[0].(*pilosa.Row).Columns()
|
|
for i := range cols {
|
|
if cols[i] == acctOwnerID {
|
|
return true
|
|
}
|
|
}
|
|
return false
|
|
}
|
|
|
|
if !bitIsSet() {
|
|
PanicOn("IRA bit should have been set")
|
|
}
|
|
|
|
queryAcct := func(m0api *pilosa.API, acctOwnerID uint64, fieldAcct0, index string) (acctBal int64) {
|
|
query := fmt.Sprintf("FieldValue(field=%v, column=%v)", fieldAcct0, acctOwnerID)
|
|
res, err := m0api.Query(context.Background(), &pilosa.QueryRequest{Index: index, Query: query})
|
|
PanicOn(err)
|
|
|
|
if len(res.Results) == 0 {
|
|
return 0
|
|
}
|
|
valCount := res.Results[0].(pilosa.ValCount)
|
|
return valCount.Val
|
|
}
|
|
|
|
bal := queryAcct(m0api, acctOwnerID, fieldAcct0, index)
|
|
|
|
if bal != acct0bal {
|
|
PanicOn(fmt.Sprintf("expected %v, observed %v starting acct0 balance", acct0bal, bal))
|
|
}
|
|
|
|
// clear the bit
|
|
qcx = mustIndexQueryContext(t, m0api, index)
|
|
ir0.Clear = true
|
|
if err := m0api.Import(ctx, qcx, ir0); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
require.Nil(t, qcx.Commit())
|
|
|
|
if bitIsSet() {
|
|
PanicOn("IRA bit should have been cleared")
|
|
}
|
|
|
|
// clear the BSI
|
|
qcx = mustIndexQueryContext(t, m0api, index)
|
|
ivr0.Clear = true
|
|
if err := m0api.ImportValue(ctx, qcx, ivr0); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
require.Nil(t, qcx.Commit())
|
|
|
|
bal = queryAcct(m0api, acctOwnerID, fieldAcct0, index)
|
|
if bal != 0 {
|
|
PanicOn(fmt.Sprintf("expected %v, observed %v starting acct0 balance", acct0bal, 0))
|
|
}
|
|
}
|
|
|
|
func TestAPI_IDAlloc(t *testing.T) {
|
|
c := test.MustRunCluster(t, 3)
|
|
defer c.Close()
|
|
|
|
primary := c.GetPrimary().API
|
|
|
|
t.Run("Normal", func(t *testing.T) {
|
|
key := pilosa.IDAllocKey{
|
|
Index: "normal",
|
|
Key: "key",
|
|
}
|
|
var session [32]byte
|
|
_, err := rand.Read(session[:])
|
|
if err != nil {
|
|
t.Fatalf("obtaining random bytes: %v", err)
|
|
}
|
|
|
|
const toReserve = 2
|
|
|
|
ids, err := primary.ReserveIDs(key, session, ^uint64(0), toReserve)
|
|
if err != nil {
|
|
t.Fatalf("reserving IDs: %v", err)
|
|
}
|
|
|
|
var numIds uint64
|
|
for _, idr := range ids {
|
|
numIds += (idr.Last - idr.First) + 1
|
|
}
|
|
if numIds != toReserve {
|
|
t.Errorf("expected %d ids but got %d: %v", toReserve, numIds, ids)
|
|
}
|
|
|
|
err = primary.CommitIDs(key, session, numIds)
|
|
if err != nil {
|
|
t.Fatalf("committing IDs: %v", err)
|
|
}
|
|
|
|
err = primary.ResetIDAlloc(key.Index)
|
|
if err != nil {
|
|
t.Fatalf("resetting ID alloc: %v", err)
|
|
}
|
|
})
|
|
t.Run("Offset", func(t *testing.T) {
|
|
key := pilosa.IDAllocKey{
|
|
Index: "offset",
|
|
Key: "key",
|
|
}
|
|
var session [32]byte
|
|
_, err := rand.Read(session[:])
|
|
if err != nil {
|
|
t.Fatalf("obtaining random bytes: %v", err)
|
|
}
|
|
|
|
ids, err := primary.ReserveIDs(key, session, 0, 2)
|
|
if err != nil {
|
|
t.Fatalf("reserving IDs: %v", err)
|
|
}
|
|
|
|
{
|
|
var numIds uint64
|
|
for _, idr := range ids {
|
|
numIds += (idr.Last - idr.First) + 1
|
|
}
|
|
if numIds != 2 {
|
|
t.Errorf("expected %d ids but got %d: %v", 2, numIds, ids)
|
|
}
|
|
}
|
|
|
|
_, err = rand.Read(session[:])
|
|
if err != nil {
|
|
t.Fatalf("obtaining random bytes: %v", err)
|
|
}
|
|
ids2, err := primary.ReserveIDs(key, session, 1, 2)
|
|
if err != nil {
|
|
t.Fatalf("reserving IDs with partially increased offset: %v", err)
|
|
}
|
|
|
|
var numIds uint64
|
|
for _, idr := range ids2 {
|
|
numIds += (idr.Last - idr.First) + 1
|
|
}
|
|
if numIds != 2 {
|
|
t.Errorf("expected %d ids but got %d: %v", 2, numIds, ids2)
|
|
}
|
|
|
|
if prevEnd, newStart := ids[len(ids)-1].Last, ids2[0].First; prevEnd != newStart {
|
|
t.Errorf("expected reuse of last ID (%d), but started with %d", prevEnd, newStart)
|
|
}
|
|
|
|
err = primary.CommitIDs(key, session, numIds)
|
|
if err != nil {
|
|
t.Errorf("committing IDs: %v", err)
|
|
}
|
|
|
|
_, err = rand.Read(session[:])
|
|
if err != nil {
|
|
t.Fatalf("obtaining random bytes: %v", err)
|
|
}
|
|
ids3, err := primary.ReserveIDs(key, session, 0, 2)
|
|
var esync pilosa.ErrIDOffsetDesync
|
|
if errors.As(err, &esync) {
|
|
if esync.Requested != 0 {
|
|
t.Errorf("incorrect requested offset in error: provided %d but got %d", 0, esync.Requested)
|
|
}
|
|
if esync.Base != 3 {
|
|
t.Errorf("incorrect base offset: expected %d but got %d", 3, esync.Base)
|
|
}
|
|
} else if err == nil {
|
|
t.Errorf("successfully re-reserved at a committed offset: %v", ids3)
|
|
} else {
|
|
t.Fatalf("unexpected error when reserving committed IDs: %v", err)
|
|
}
|
|
|
|
err = primary.ResetIDAlloc(key.Index)
|
|
if err != nil {
|
|
t.Fatalf("resetting ID alloc: %v", err)
|
|
}
|
|
})
|
|
}
|
|
|
|
type mutexCheckIndex struct {
|
|
index *pilosa.Index
|
|
indexName string
|
|
createdAt int64
|
|
fields map[bool]mutexCheckField
|
|
}
|
|
|
|
type mutexCheckField struct {
|
|
fieldName string
|
|
field *pilosa.Field
|
|
createdAt int64
|
|
}
|
|
|
|
func TestAPI_MutexCheck(t *testing.T) {
|
|
// Can't share this one, because it has to get custom option settings and needs
|
|
// replication.
|
|
c := test.MustUnsharedCluster(t, 3)
|
|
for _, c := range c.Nodes {
|
|
c.Config.Cluster.ReplicaN = 2
|
|
}
|
|
if err := c.Start(); err != nil {
|
|
t.Fatalf("starting cluster: %v", err)
|
|
}
|
|
defer c.Close()
|
|
|
|
m0 := c.GetNode(0)
|
|
nodesByID := make(map[string]*test.Command, len(c.Nodes))
|
|
qcxsByID := make(map[string]qc.QueryContext, len(c.Nodes))
|
|
for i := 0; i < len(c.Nodes); i++ {
|
|
node := c.GetNode(i)
|
|
id := node.API.NodeID()
|
|
nodesByID[id] = node
|
|
}
|
|
|
|
indexes := make(map[bool]mutexCheckIndex)
|
|
|
|
ctx := context.Background()
|
|
for _, keyedIndex := range []bool{false, true} {
|
|
indexName := c.Idx(map[bool]string{false: "u", true: "k"}[keyedIndex])
|
|
index, err := m0.API.CreateIndex(ctx, indexName, pilosa.IndexOptions{Keys: keyedIndex, TrackExistence: true})
|
|
if err != nil {
|
|
t.Fatalf("creating index: %v", err)
|
|
}
|
|
if index.CreatedAt() == 0 {
|
|
t.Fatal("index createdAt is empty")
|
|
}
|
|
indexData := mutexCheckIndex{indexName: indexName, index: index, fields: make(map[bool]mutexCheckField), createdAt: index.CreatedAt()}
|
|
for _, keyedField := range []bool{false, true} {
|
|
fieldName := fmt.Sprintf("f%t", keyedField)
|
|
var field *pilosa.Field
|
|
if keyedField {
|
|
field, err = m0.API.CreateField(ctx, indexName, fieldName, pilosa.OptFieldTypeMutex(pilosa.CacheTypeNone, 0), pilosa.OptFieldKeys())
|
|
} else {
|
|
field, err = m0.API.CreateField(ctx, indexName, fieldName, pilosa.OptFieldTypeMutex(pilosa.CacheTypeNone, 0))
|
|
}
|
|
if err != nil {
|
|
t.Fatalf("creating field: %v", err)
|
|
}
|
|
if field.CreatedAt() == 0 {
|
|
t.Fatal("field createdAt is empty")
|
|
}
|
|
indexData.fields[keyedField] = mutexCheckField{fieldName: fieldName, field: field, createdAt: field.CreatedAt()}
|
|
}
|
|
indexes[keyedIndex] = indexData
|
|
}
|
|
|
|
rowIDs := []uint64{0, 1, 2, 3}
|
|
colIDs := []uint64{0, 1, 2, 3}
|
|
rowKeysBase := []string{"v0", "v1", "v2", "v3"}
|
|
colKeysBase := []string{"c0", "c1", "c2", "c3"}
|
|
|
|
const nShards = 9
|
|
|
|
// now, try the same thing for each combination of keyed/unkeyed. we
|
|
// share code between keyed/unkeyed fields, but for indexes, the logic
|
|
// is fundamentally different because we can't know shards in advance.
|
|
indexData := indexes[false]
|
|
for keyedField, fieldData := range indexData.fields {
|
|
t.Run(fmt.Sprintf("%s-%s", indexData.indexName, fieldData.fieldName), func(t *testing.T) {
|
|
for id, node := range nodesByID {
|
|
qcxsByID[id] = mustIndexQueryContext(t, node.API, indexData.indexName)
|
|
}
|
|
for shard := uint64(0); shard < nShards; shard++ {
|
|
// restore row/col ID values which can get altered by imports
|
|
for i := range rowIDs {
|
|
rowIDs[i] = uint64(i)
|
|
colIDs[i] = (shard << shardwidth.Exponent) + uint64(i) + (shard % 4)
|
|
}
|
|
req := &pilosa.ImportRequest{
|
|
Index: indexData.indexName,
|
|
IndexCreatedAt: indexData.createdAt,
|
|
Field: fieldData.fieldName,
|
|
FieldCreatedAt: fieldData.createdAt,
|
|
Shard: shard,
|
|
ColumnIDs: colIDs,
|
|
}
|
|
if keyedField {
|
|
req.RowKeys = rowKeysBase
|
|
} else {
|
|
req.RowIDs = rowIDs
|
|
}
|
|
nodesForShard, err := m0.API.ShardNodes(ctx, indexData.indexName, shard)
|
|
if err != nil {
|
|
t.Fatalf("obtaining shard list: %v", err)
|
|
}
|
|
if len(nodesForShard) < 1 {
|
|
t.Fatalf("no nodes for shard %d", shard)
|
|
}
|
|
node := nodesByID[nodesForShard[0].ID]
|
|
if err := node.API.Import(ctx, qcxsByID[nodesForShard[0].ID], req); err != nil {
|
|
t.Fatalf("importing data: %v", err)
|
|
}
|
|
}
|
|
// and then we break the mutex and close the Qcxs
|
|
for id, node := range nodesByID {
|
|
field, err := node.API.Field(ctx, indexData.indexName, fieldData.fieldName)
|
|
if err != nil {
|
|
t.Fatalf("requesting field %s from node %s: %v", fieldData.fieldName, id, err)
|
|
}
|
|
pilosa.CorruptAMutex(t, field, qcxsByID[id])
|
|
require.Nil(t, qcxsByID[id].Commit())
|
|
}
|
|
qcx := mustQueryContext(t, m0.API)
|
|
defer qcx.Release()
|
|
|
|
// first two shards of each group of 4 should have a collision in
|
|
// position 1
|
|
expected := map[uint64]bool{
|
|
(0 << shardwidth.Exponent) + 1: true,
|
|
(1 << shardwidth.Exponent) + 1: true,
|
|
(4 << shardwidth.Exponent) + 1: true,
|
|
(5 << shardwidth.Exponent) + 1: true,
|
|
(8 << shardwidth.Exponent) + 1: true,
|
|
// So, nShards used to be 10. If you run a complete test,
|
|
// with go test -race, and you have the sample input for the
|
|
// unrelated TestImportMutexSampleData configured to use 64K
|
|
// bit density and 2K rows, everything is fine. If you run a
|
|
// partial test, everything is fine. If you run a complete
|
|
// test with -race, but you skip TestImportMutexSampleData,
|
|
// or reduce either the bit density or the row count, you get
|
|
// a very strange panic where the go panic handler panics
|
|
// trying to report what happened so we don't get a valid
|
|
// stack dump. On Macs. This is as much as I could debug it
|
|
// after about 6 hours. Since there's no special reason to
|
|
// think we need all 10 shards, and 9 still tests the
|
|
// behavior, we're leaving this one a mystery.
|
|
// (9 << shardwidth.Exponent) + 1: true,
|
|
}
|
|
|
|
results, err := m0.API.MutexCheck(ctx, qcx, indexData.indexName, fieldData.fieldName, true, 0)
|
|
if err != nil {
|
|
t.Fatalf("checking mutexes: %v", err)
|
|
}
|
|
|
|
if keyedField {
|
|
mapped, ok := results.(map[uint64][]string)
|
|
if !ok {
|
|
t.Fatalf("expected map[uint64][]string, got %T", results)
|
|
}
|
|
seen := 0
|
|
for k, v := range mapped {
|
|
seen++
|
|
if !expected[k] {
|
|
t.Fatalf("expected all collisions to be 1 shards (s %% 4 in [0,1]), got %d", k)
|
|
}
|
|
if len(v) != 2 {
|
|
t.Fatalf("expected exactly two collisions")
|
|
}
|
|
}
|
|
if seen != len(expected) {
|
|
t.Fatalf("expected exactly %d records to have collisions", len(expected))
|
|
}
|
|
} else {
|
|
mapped, ok := results.(map[uint64][]uint64)
|
|
if !ok {
|
|
t.Fatalf("expected map[uint64][]uint64, got %T", results)
|
|
}
|
|
seen := 0
|
|
for k, v := range mapped {
|
|
seen++
|
|
if !expected[k] {
|
|
t.Fatalf("expected all collisions to be 1 shards (s %% 4 in [0,1]), got %d", k)
|
|
}
|
|
if len(v) != 2 {
|
|
t.Fatalf("expected exactly two collisions")
|
|
}
|
|
}
|
|
if seen != len(expected) {
|
|
t.Fatalf("expected exactly %d records to have collisions, got %d", len(expected), seen)
|
|
}
|
|
}
|
|
|
|
// and let's try with no details and a limit of 3...
|
|
results, err = m0.API.MutexCheck(ctx, qcx, indexData.indexName, fieldData.fieldName, false, 3)
|
|
if err != nil {
|
|
t.Fatalf("checking mutexes: %v", err)
|
|
}
|
|
mapped, ok := results.([]uint64)
|
|
if !ok {
|
|
t.Fatalf("expected []uint64, got %T", results)
|
|
}
|
|
seen := 0
|
|
for _, k := range mapped {
|
|
seen++
|
|
if !expected[k] {
|
|
t.Fatalf("expected all collisions to be position 1 in shards (s %% 4 in [0,1]), got %d", k)
|
|
}
|
|
}
|
|
if seen != 3 {
|
|
t.Fatalf("expected results limited to 3, got %d", seen)
|
|
}
|
|
})
|
|
}
|
|
indexData = indexes[true]
|
|
for keyedField, fieldData := range indexData.fields {
|
|
t.Run(fmt.Sprintf("%s-%s", indexData.indexName, fieldData.fieldName), func(t *testing.T) {
|
|
req := &pilosa.ImportRequest{
|
|
Index: indexData.indexName,
|
|
IndexCreatedAt: indexData.createdAt,
|
|
Field: fieldData.fieldName,
|
|
FieldCreatedAt: fieldData.createdAt,
|
|
Shard: 0, // ignored when using keys
|
|
}
|
|
rowKeys := make([]string, 0, len(rowKeysBase)*nShards)
|
|
colKeys := make([]string, 0, len(rowKeysBase)*nShards)
|
|
rowIDs = rowIDs[:0]
|
|
for shard := uint64(0); shard < nShards; shard++ {
|
|
for i := range rowKeysBase {
|
|
colKeys = append(colKeys, fmt.Sprintf("s%d-%s", shard, colKeysBase[i]))
|
|
if keyedField {
|
|
rowKeys = append(rowKeys, rowKeysBase[i])
|
|
} else {
|
|
rowIDs = append(rowIDs, uint64(i))
|
|
}
|
|
}
|
|
}
|
|
req.ColumnKeys = colKeys
|
|
if keyedField {
|
|
req.RowKeys = rowKeys
|
|
} else {
|
|
req.RowIDs = rowIDs
|
|
}
|
|
var node *test.Command
|
|
for _, node = range nodesByID {
|
|
break
|
|
}
|
|
if err := node.API.Import(ctx, nil, req); err != nil {
|
|
t.Fatalf("importing data: %v", err)
|
|
}
|
|
expected, err := node.API.FindIndexKeys(ctx, indexData.indexName, colKeys...)
|
|
if err != nil {
|
|
t.Fatalf("looking up index keys: %v", err)
|
|
}
|
|
for key, id := range expected {
|
|
// CorruptAMutex should only corrupt things in position 1 of their
|
|
// shards...
|
|
if id%(1<<shardwidth.Exponent) != 1 {
|
|
delete(expected, key)
|
|
}
|
|
}
|
|
if keyedField {
|
|
fieldValues, err := node.API.FindFieldKeys(ctx, indexData.indexName, fieldData.fieldName, rowKeys...)
|
|
if err != nil {
|
|
t.Fatalf("looking up field keys: %v", err)
|
|
}
|
|
// Figure out which key got the value 3, delete any records
|
|
// which would have had that key, because they won't be
|
|
// conflicts.
|
|
for key, value := range fieldValues {
|
|
if value == 3 {
|
|
for offset, baseKey := range rowKeysBase {
|
|
if baseKey == key {
|
|
for i := offset; i < len(rowKeys); i += len(rowKeysBase) {
|
|
delete(expected, colKeys[i])
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
} else {
|
|
// we set rowKeys to 0-1-2-... for rowKeysBase items, which
|
|
// tells us which keys we expect to be 3 already.
|
|
for i := 3; i < len(rowIDs); i += len(rowKeysBase) {
|
|
delete(expected, colKeys[i])
|
|
}
|
|
}
|
|
// and then we break the mutex and close the Qcxs
|
|
for id, node := range nodesByID {
|
|
field, err := node.API.Field(ctx, indexData.indexName, fieldData.fieldName)
|
|
if err != nil {
|
|
t.Fatalf("requesting field %s from node %s: %v", fieldData.fieldName, id, err)
|
|
}
|
|
qcx := mustIndexQueryContext(t, node.API, indexData.indexName)
|
|
pilosa.CorruptAMutex(t, field, qcx)
|
|
require.Nil(t, qcx.Commit())
|
|
}
|
|
qcx := mustQueryContext(t, m0.API)
|
|
if err != nil {
|
|
t.Fatalf("creating query context: %v", err)
|
|
}
|
|
defer qcx.Release()
|
|
|
|
results, err := m0.API.MutexCheck(ctx, qcx, indexData.indexName, fieldData.fieldName, true, 0)
|
|
if err != nil {
|
|
t.Fatalf("checking mutexes: %v", err)
|
|
}
|
|
if keyedField {
|
|
mapped, ok := results.(map[string][]string)
|
|
if !ok {
|
|
t.Fatalf("expected map[string][]string, got %T", results)
|
|
}
|
|
seen := 0
|
|
for k, v := range mapped {
|
|
seen++
|
|
if _, ok := expected[k]; !ok {
|
|
t.Fatalf("unexpected collision on key %q", k)
|
|
}
|
|
if len(v) != 2 {
|
|
t.Fatalf("expected exactly two collisions")
|
|
}
|
|
}
|
|
if seen != len(expected) {
|
|
t.Fatalf("expected exactly %d records to have collisions, got %d", len(expected), seen)
|
|
}
|
|
} else {
|
|
mapped, ok := results.(map[string][]uint64)
|
|
if !ok {
|
|
t.Fatalf("expected map[string][]uint64, got %T", results)
|
|
}
|
|
seen := 0
|
|
for k, v := range mapped {
|
|
seen++
|
|
if _, ok := expected[k]; !ok {
|
|
t.Fatalf("unexpected collision on key %q", k)
|
|
}
|
|
if len(v) != 2 {
|
|
t.Fatalf("expected exactly two collisions")
|
|
}
|
|
}
|
|
if seen != len(expected) {
|
|
t.Fatalf("expected exactly %d records to have collisions, got %d", len(expected), seen)
|
|
}
|
|
}
|
|
|
|
results, err = m0.API.MutexCheck(ctx, qcx, indexData.indexName, fieldData.fieldName, false, 3)
|
|
if err != nil {
|
|
t.Fatalf("checking mutexes: %v", err)
|
|
}
|
|
mapped, ok := results.([]string)
|
|
if !ok {
|
|
t.Fatalf("expected []string, got %T", results)
|
|
}
|
|
seen := 0
|
|
for _, k := range mapped {
|
|
seen++
|
|
if _, ok := expected[k]; !ok {
|
|
t.Fatalf("unexpected collision on key %q", k)
|
|
}
|
|
}
|
|
if seen != 3 {
|
|
t.Fatalf("expected results limited to 3, got %d", len(expected))
|
|
}
|
|
})
|
|
}
|
|
}
|
|
|
|
func createIndexForTest(index string, coord *test.Command, t *testing.T) {
|
|
ctx := context.Background()
|
|
_, err := coord.API.CreateIndex(ctx, index, pilosa.IndexOptions{Keys: true})
|
|
if err != nil {
|
|
t.Fatalf("creating index: %v", err)
|
|
}
|
|
}
|
|
|
|
func createFieldForTest(index string, field string, coord *test.Command, t *testing.T) {
|
|
ctx := context.Background()
|
|
_, err := coord.API.CreateField(ctx, index, field, pilosa.OptFieldTypeInt(math.MinInt64, math.MaxInt64))
|
|
if err != nil {
|
|
t.Fatalf("creating field: %v", err)
|
|
}
|
|
}
|
|
|
|
func TestVariousApiTranslateCalls(t *testing.T) {
|
|
for i := 1; i < 8; i += 3 {
|
|
c := test.MustRunCluster(t, i)
|
|
defer c.Close()
|
|
node := c.GetNode(0)
|
|
api := node.API
|
|
// this should never actually get used because we're testing for errors here
|
|
r := strings.NewReader("")
|
|
// test index
|
|
idx, err := api.Holder().CreateIndex(c.Idx(), "", pilosa.IndexOptions{})
|
|
if err != nil {
|
|
t.Fatalf("%v: could not create test index", err)
|
|
}
|
|
if _, err = idx.CreateFieldIfNotExistsWithOptions("field", "", &pilosa.FieldOptions{Keys: false}); err != nil {
|
|
t.Fatalf("creating field: %v", err)
|
|
}
|
|
t.Run("translateIndexDbOnNilIndex",
|
|
func(t *testing.T) {
|
|
err := api.TranslateIndexDB(context.Background(), "nonExistentIndex", 0, r)
|
|
expected := fmt.Sprintf("index %q not found", "nonExistentIndex")
|
|
if err == nil || err.Error() != expected {
|
|
t.Fatalf("expected '%#v', got '%#v'", expected, err)
|
|
}
|
|
})
|
|
|
|
t.Run("translateIndexDbOnNilTranslateStore",
|
|
func(t *testing.T) {
|
|
err := api.TranslateIndexDB(context.Background(), c.Idx(), 0, r)
|
|
expected := fmt.Sprintf("index %q has no translate store", c.Idx())
|
|
if err == nil || err.Error() != expected {
|
|
t.Fatalf("expected '%#v', got '%#v'", expected, err)
|
|
}
|
|
})
|
|
|
|
t.Run("translateFieldDbOnNilIndex",
|
|
func(t *testing.T) {
|
|
err := api.TranslateFieldDB(context.Background(), "nonExistentIndex", "field", r)
|
|
expected := fmt.Sprintf("index %q not found", "nonExistentIndex")
|
|
if err == nil || err.Error() != expected {
|
|
t.Fatalf("expected '%#v', got '%#v'", expected, err)
|
|
}
|
|
})
|
|
|
|
t.Run("translateFieldDbOnNilField",
|
|
func(t *testing.T) {
|
|
err := api.TranslateFieldDB(context.Background(), c.Idx(), "nonExistentField", r)
|
|
expected := fmt.Sprintf("field %q/%q not found", c.Idx(), "nonExistentField")
|
|
if err == nil || err.Error() != expected {
|
|
t.Fatalf("expected '%#v', got '%#v'", expected, err)
|
|
}
|
|
})
|
|
|
|
t.Run("translateFieldDbNilField_keys",
|
|
func(t *testing.T) {
|
|
err := api.TranslateFieldDB(context.Background(), c.Idx(), "_keys", r)
|
|
if err != nil {
|
|
t.Fatalf("expected 'nil', got '%#v'", err)
|
|
}
|
|
})
|
|
/*
|
|
TODO: this test will break, bc currently all fields create translate
|
|
stores, which is a bug, but one that we will eventually fix. when we do, this
|
|
test might come in handy t.Run("translateFieldDbOnNilTranslateStore",
|
|
func(t *testing.T) {
|
|
err := api.TranslateFieldDB(context.Background(), c.Idx(), "field", r)
|
|
expected := fmt.Errorf("field %q/%q has no translate store", c.Idx(), "field")
|
|
if !reflect.DeepEqual(err, expected) {
|
|
t.Fatalf("expected '%#v', got '%#v'", expected, err)
|
|
}
|
|
})
|
|
*/
|
|
}
|
|
}
|
|
|
|
func TestAPI_CreateField(t *testing.T) {
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
defer cancel()
|
|
c := test.MustRunCluster(t, 3)
|
|
defer c.Close()
|
|
|
|
nodes := make([]*test.Command, 3)
|
|
for i := range nodes {
|
|
nodes[i] = c.GetNode(i)
|
|
}
|
|
|
|
if _, err := nodes[0].API.CreateIndex(ctx, c.Idx(), pilosa.IndexOptions{}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
eg, ctx := errgroup.WithContext(context.Background())
|
|
for _, n := range nodes {
|
|
node := n
|
|
eg.Go(func() error {
|
|
for i := 0; i < 10; i++ {
|
|
_, err := node.API.CreateField(ctx, c.Idx(), fmt.Sprintf("f%d", i))
|
|
if err != nil && !errors.Is(err, pilosa.ErrFieldExists) {
|
|
return err
|
|
}
|
|
}
|
|
return nil
|
|
})
|
|
}
|
|
err := eg.Wait()
|
|
if err != nil {
|
|
if errors.Is(err, pilosa.ErrFieldExists) {
|
|
t.Fatalf("conflict error: %v", err)
|
|
}
|
|
t.Fatalf("unexpected error: %T %v", err, err)
|
|
}
|
|
}
|
|
|
|
// makeUser makes an authnUserInfo from groups and a name and a secret key
|
|
func makeUser(t *testing.T, groups []authn.Group, name, secret string) *authn.UserInfo {
|
|
tkn := jwt.New(jwt.SigningMethodHS256)
|
|
claims := tkn.Claims.(jwt.MapClaims)
|
|
claims["oid"] = "42"
|
|
claims["name"] = name
|
|
secretKey, _ := hex.DecodeString(secret)
|
|
|
|
validToken, err := tkn.SignedString(secretKey)
|
|
if err != nil {
|
|
t.Fatalf("signing string %v", err)
|
|
}
|
|
|
|
return &authn.UserInfo{
|
|
UserID: "fake" + name,
|
|
UserName: name,
|
|
Groups: groups,
|
|
Token: validToken,
|
|
Expiry: time.Time{},
|
|
}
|
|
}
|
|
|
|
func TestAuth_MultiNode(t *testing.T) {
|
|
// create permissions file
|
|
permissions := `
|
|
"user-groups":
|
|
"dca35310-ecda-4f23-86cd-876aee55906b":
|
|
"test": "read"
|
|
"dca35310-ecda-4f23-86cd-876aee55906f":
|
|
"test": "write"
|
|
admin: "ac97c9e2-346b-42a2-b6da-18bcb61a32fe"`
|
|
adminUser := makeUser(t, []authn.Group{{GroupID: "ac97c9e2-346b-42a2-b6da-18bcb61a32fe", GroupName: "adminGroup"}}, "admin", "DEADBEEFDEADBEEFDEADBEEFDEADBEEFDEADBEEFDEADBEEFDEADBEEFDEADBEEF")
|
|
adminCtx := authn.WithUserInfo(context.Background(), adminUser)
|
|
readUser := makeUser(t, []authn.Group{{GroupID: "dca35310-ecda-4f23-86cd-876aee55906b", GroupName: "readGroup"}}, "reader", "DEADBEEFDEADBEEFDEADBEEFDEADBEEFDEADBEEFDEADBEEFDEADBEEFDEADBEEF")
|
|
readCtx := authn.WithUserInfo(context.Background(), readUser)
|
|
writeUser := makeUser(t, []authn.Group{{GroupID: "dca35310-ecda-4f23-86cd-876aee55906f", GroupName: "writeGroup"}}, "writer", "DEADBEEFDEADBEEFDEADBEEFDEADBEEFDEADBEEFDEADBEEFDEADBEEFDEADBEED")
|
|
writeCtx := authn.WithUserInfo(context.Background(), writeUser)
|
|
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
|
token, ok := r.Header["Authorization"]
|
|
if !ok || len(token) == 0 {
|
|
http.Error(w, "BAD REQUEST", http.StatusBadRequest)
|
|
return
|
|
}
|
|
g := []authn.Group{}
|
|
switch strings.TrimPrefix(token[0], "Bearer ") {
|
|
case adminUser.Token:
|
|
g = adminUser.Groups
|
|
case readUser.Token:
|
|
g = readUser.Groups
|
|
case writeUser.Token:
|
|
g = writeUser.Groups
|
|
}
|
|
if err := json.NewEncoder(w).Encode(authn.Groups{Groups: g}); err != nil {
|
|
t.Fatalf("unexpected error marshalling groups response: %v", err)
|
|
}
|
|
}))
|
|
|
|
// authentication on
|
|
auth := server.Auth{
|
|
Enable: true,
|
|
ClientId: "e9088663-eb08-41d7-8f65-efb5f54bbb71",
|
|
ClientSecret: "DEADBEEFDEADBEEFDEADBEEFDEADBEEFDEADBEEFDEADBEEFDEADBEEFDEADBEEF",
|
|
AuthorizeURL: "https://login.microsoftonline.com/4a137d66-d161-4ae4-b1e6-07e9920874b8/oauth2/v2.0/authorize",
|
|
TokenURL: "https://login.microsoftonline.com/4a137d66-d161-4ae4-b1e6-07e9920874b8/oauth2/v2.0/token",
|
|
GroupEndpointURL: srv.URL,
|
|
RedirectBaseURL: "https://localhost:10101",
|
|
LogoutURL: "https://login.microsoftonline.com/common/oauth2/v2.0/logout",
|
|
Scopes: []string{"https://graph.microsoft.com/.default", "offline_access"},
|
|
SecretKey: "DEADBEEFDEADBEEFDEADBEEFDEADBEEFDEADBEEFDEADBEEFDEADBEEFDEADBEEF",
|
|
PermissionsFile: writeTestFile(t, "permissions.yaml", permissions),
|
|
QueryLogPath: writeTestFile(t, "queryLog.log", ""),
|
|
}
|
|
|
|
config := server.NewConfig()
|
|
config.Auth = auth
|
|
|
|
// set up TLS certificates
|
|
localhostCert := `-----BEGIN CERTIFICATE-----
|
|
MIICEzCCAXygAwIBAgIQMIMChMLGrR+QvmQvpwAU6zANBgkqhkiG9w0BAQsFADAS
|
|
MRAwDgYDVQQKEwdBY21lIENvMCAXDTcwMDEwMTAwMDAwMFoYDzIwODQwMTI5MTYw
|
|
MDAwWjASMRAwDgYDVQQKEwdBY21lIENvMIGfMA0GCSqGSIb3DQEBAQUAA4GNADCB
|
|
iQKBgQDuLnQAI3mDgey3VBzWnB2L39JUU4txjeVE6myuDqkM/uGlfjb9SjY1bIw4
|
|
iA5sBBZzHi3z0h1YV8QPuxEbi4nW91IJm2gsvvZhIrCHS3l6afab4pZBl2+XsDul
|
|
rKBxKKtD1rGxlG4LjncdabFn9gvLZad2bSysqz/qTAUStTvqJQIDAQABo2gwZjAO
|
|
BgNVHQ8BAf8EBAMCAqQwEwYDVR0lBAwwCgYIKwYBBQUHAwEwDwYDVR0TAQH/BAUw
|
|
AwEB/zAuBgNVHREEJzAlggtleGFtcGxlLmNvbYcEfwAAAYcQAAAAAAAAAAAAAAAA
|
|
AAAAATANBgkqhkiG9w0BAQsFAAOBgQCEcetwO59EWk7WiJsG4x8SY+UIAA+flUI9
|
|
tyC4lNhbcF2Idq9greZwbYCqTTTr2XiRNSMLCOjKyI7ukPoPjo16ocHj+P3vZGfs
|
|
h1fIw3cSS2OolhloGw/XM6RWPWtPAlGykKLciQrBru5NAPvCMsb/I1DAceTiotQM
|
|
fblo6RBxUQ==
|
|
-----END CERTIFICATE-----`
|
|
|
|
localhostKey := `-----BEGIN RSA PRIVATE KEY-----
|
|
MIICXgIBAAKBgQDuLnQAI3mDgey3VBzWnB2L39JUU4txjeVE6myuDqkM/uGlfjb9
|
|
SjY1bIw4iA5sBBZzHi3z0h1YV8QPuxEbi4nW91IJm2gsvvZhIrCHS3l6afab4pZB
|
|
l2+XsDulrKBxKKtD1rGxlG4LjncdabFn9gvLZad2bSysqz/qTAUStTvqJQIDAQAB
|
|
AoGAGRzwwir7XvBOAy5tM/uV6e+Zf6anZzus1s1Y1ClbjbE6HXbnWWF/wbZGOpet
|
|
3Zm4vD6MXc7jpTLryzTQIvVdfQbRc6+MUVeLKwZatTXtdZrhu+Jk7hx0nTPy8Jcb
|
|
uJqFk541aEw+mMogY/xEcfbWd6IOkp+4xqjlFLBEDytgbIECQQDvH/E6nk+hgN4H
|
|
qzzVtxxr397vWrjrIgPbJpQvBsafG7b0dA4AFjwVbFLmQcj2PprIMmPcQrooz8vp
|
|
jy4SHEg1AkEA/v13/5M47K9vCxmb8QeD/asydfsgS5TeuNi8DoUBEmiSJwma7FXY
|
|
fFUtxuvL7XvjwjN5B30pNEbc6Iuyt7y4MQJBAIt21su4b3sjXNueLKH85Q+phy2U
|
|
fQtuUE9txblTu14q3N7gHRZB4ZMhFYyDy8CKrN2cPg/Fvyt0Xlp/DoCzjA0CQQDU
|
|
y2ptGsuSmgUtWj3NM9xuwYPm+Z/F84K6+ARYiZ6PYj013sovGKUFfYAqVXVlxtIX
|
|
qyUBnu3X9ps8ZfjLZO7BAkEAlT4R5Yl6cGhaJQYZHOde3JEMhNRcVFMO8dJDaFeo
|
|
f9Oeos0UUothgiDktdQHxdNEwLjQf7lJJBzV+5OtwswCWA==
|
|
-----END RSA PRIVATE KEY-----`
|
|
|
|
config.TLS.CertificateKeyPath = writeTestFile(t, "certKey.pem", localhostKey)
|
|
config.TLS.CertificatePath = writeTestFile(t, "cert.pem", localhostCert)
|
|
|
|
c := test.MustRunUnsharedCluster(t, 3,
|
|
[]server.CommandOption{
|
|
server.OptCommandServerOptions(
|
|
pilosa.OptServerNodeID("node0"),
|
|
),
|
|
server.OptCommandConfig(config),
|
|
},
|
|
[]server.CommandOption{
|
|
server.OptCommandServerOptions(
|
|
pilosa.OptServerNodeID("node1"),
|
|
),
|
|
server.OptCommandConfig(config),
|
|
},
|
|
[]server.CommandOption{
|
|
server.OptCommandServerOptions(
|
|
pilosa.OptServerNodeID("node2"),
|
|
),
|
|
server.OptCommandConfig(config),
|
|
},
|
|
)
|
|
defer c.Close()
|
|
|
|
primaryAPI := c.GetPrimary().API
|
|
|
|
// needs internal/cluster/message
|
|
// Note: This indexName wouldn't be safe on a shared cluster, but we have to use an
|
|
// unshared cluster to set up the auth config anyway.
|
|
indexName := "test"
|
|
_, err := primaryAPI.CreateIndex(adminCtx, indexName, pilosa.IndexOptions{})
|
|
if err != nil {
|
|
t.Fatalf("creating index: %v", err)
|
|
}
|
|
// needs internal/translate/data
|
|
fieldName := "f"
|
|
_, err = primaryAPI.CreateField(adminCtx, indexName, fieldName, pilosa.OptFieldTypeSet(pilosa.DefaultCacheType, 100))
|
|
if err != nil {
|
|
t.Fatalf("creating field: %v", err)
|
|
}
|
|
|
|
_, err = primaryAPI.Query(readCtx, &pilosa.QueryRequest{
|
|
Index: indexName,
|
|
Query: fmt.Sprintf(`Set(1, %s=1)`, fieldName),
|
|
})
|
|
if err == nil {
|
|
t.Fatalf("readCtx should not be able to set bits")
|
|
}
|
|
|
|
_, err = primaryAPI.Query(writeCtx, &pilosa.QueryRequest{
|
|
Index: indexName,
|
|
Query: fmt.Sprintf(`Set(1, %s=1)`, fieldName),
|
|
})
|
|
if err != nil {
|
|
t.Fatalf("writeCtx should be able to set bits: %v", err)
|
|
}
|
|
|
|
_, err = primaryAPI.Query(adminCtx, &pilosa.QueryRequest{
|
|
Index: indexName,
|
|
Query: fmt.Sprintf(`Set(1, %s=1)`, fieldName),
|
|
})
|
|
if err != nil {
|
|
t.Fatalf("adminCtx should be able to set bits: %v", err)
|
|
}
|
|
|
|
_, err = primaryAPI.Query(readCtx, &pilosa.QueryRequest{
|
|
Index: indexName,
|
|
Query: fmt.Sprintf(`Count(Row(%s=1))`, fieldName),
|
|
})
|
|
if err != nil {
|
|
t.Fatalf("readCtx should be able read: %v", err)
|
|
}
|
|
|
|
_, err = primaryAPI.Query(writeCtx, &pilosa.QueryRequest{
|
|
Index: indexName,
|
|
Query: fmt.Sprintf(`Count(Row(%s=1))`, fieldName),
|
|
})
|
|
if err != nil {
|
|
t.Fatalf("writeCtx should be able read: %v", err)
|
|
}
|
|
|
|
_, err = primaryAPI.Query(adminCtx, &pilosa.QueryRequest{
|
|
Index: indexName,
|
|
Query: fmt.Sprintf(`Count(Row(%s=1))`, fieldName),
|
|
})
|
|
if err != nil {
|
|
t.Fatalf("adminCtx should be able read: %v", err)
|
|
}
|
|
}
|
|
|
|
func writeTestFile(t *testing.T, filename, content string) string {
|
|
t.Helper()
|
|
fname := filepath.Join(t.TempDir(), filename)
|
|
f, err := os.Create(fname)
|
|
if err != nil {
|
|
t.Fatalf("could not create file %v with err %v", filename, err)
|
|
}
|
|
_, err = io.WriteString(f, content)
|
|
if err != nil {
|
|
t.Fatalf("could not write string %v", err)
|
|
}
|
|
if err := f.Close(); err != nil {
|
|
t.Fatalf("could not close file %v", err)
|
|
}
|
|
return fname
|
|
}
|