mirror of
https://github.com/featurebasedb/featurebase.git
synced 2026-08-28 10:54:59 +00:00
Merge branch 'master' into roaring-import
This commit is contained in:
commit
5589324865
10 changed files with 73 additions and 40 deletions
|
|
@ -41,10 +41,12 @@ type InternalClient interface {
|
|||
ImportK(ctx context.Context, index, field string, bits []Bit) 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) error
|
||||
ImportValueK(ctx context.Context, index, field string, vals []FieldValue) 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
|
||||
FragmentBlocks(ctx context.Context, uri *URI, index, field, view string, shard uint64) ([]FragmentBlock, error)
|
||||
BlockData(ctx context.Context, uri *URI, index, field, view string, shard uint64, block int) ([]uint64, []uint64, error)
|
||||
ColumnAttrDiff(ctx context.Context, uri *URI, index string, blks []AttrBlock) (map[uint64]map[string]interface{}, error)
|
||||
|
|
@ -116,6 +118,9 @@ func (n nopInternalClient) EnsureIndex(ctx context.Context, name string, options
|
|||
func (n nopInternalClient) EnsureField(ctx context.Context, indexName string, fieldName string) error {
|
||||
return nil
|
||||
}
|
||||
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) error {
|
||||
return nil
|
||||
}
|
||||
|
|
@ -126,6 +131,9 @@ func (n nopInternalClient) ExportCSV(ctx context.Context, index, field string, s
|
|||
return nil
|
||||
}
|
||||
func (n nopInternalClient) CreateField(ctx context.Context, index, field string) error { return nil }
|
||||
func (n nopInternalClient) CreateFieldWithOptions(ctx context.Context, index, field string, opt FieldOptions) error {
|
||||
return nil
|
||||
}
|
||||
func (n nopInternalClient) FragmentBlocks(ctx context.Context, uri *URI, index, field, view string, shard uint64) ([]FragmentBlock, error) {
|
||||
return nil, nil
|
||||
}
|
||||
|
|
|
|||
|
|
@ -18,7 +18,6 @@ import (
|
|||
"context"
|
||||
"fmt"
|
||||
"io"
|
||||
"os"
|
||||
|
||||
"github.com/spf13/cobra"
|
||||
|
||||
|
|
@ -27,8 +26,8 @@ import (
|
|||
|
||||
var checker *ctl.CheckCommand
|
||||
|
||||
func newCheckCommand(_ io.Reader, _, _ io.Writer) *cobra.Command {
|
||||
checker = ctl.NewCheckCommand(os.Stdin, os.Stdout, os.Stderr)
|
||||
func newCheckCommand(stdin io.Reader, stdout io.Writer, stderr io.Writer) *cobra.Command {
|
||||
checker = ctl.NewCheckCommand(stdin, stdout, stderr)
|
||||
checkCmd := &cobra.Command{
|
||||
Use: "check <path> [path2]...",
|
||||
Short: "Do a consistency check on a pilosa data file.",
|
||||
|
|
|
|||
|
|
@ -17,7 +17,6 @@ package cmd
|
|||
import (
|
||||
"context"
|
||||
"io"
|
||||
"os"
|
||||
|
||||
"github.com/spf13/cobra"
|
||||
|
||||
|
|
@ -28,7 +27,7 @@ import (
|
|||
var conf *ctl.ConfigCommand
|
||||
|
||||
func newConfigCommand(stdin io.Reader, stdout, stderr io.Writer) *cobra.Command {
|
||||
conf = ctl.NewConfigCommand(os.Stdin, os.Stdout, os.Stderr)
|
||||
conf = ctl.NewConfigCommand(stdin, stdout, stderr)
|
||||
Server := server.NewCommand(stdin, stdout, stderr)
|
||||
confCmd := &cobra.Command{
|
||||
Use: "config",
|
||||
|
|
|
|||
|
|
@ -17,7 +17,6 @@ package cmd
|
|||
import (
|
||||
"context"
|
||||
"io"
|
||||
"os"
|
||||
|
||||
"github.com/spf13/cobra"
|
||||
|
||||
|
|
@ -26,8 +25,8 @@ import (
|
|||
|
||||
var Exporter *ctl.ExportCommand
|
||||
|
||||
func newExportCommand(_ io.Reader, _, _ io.Writer) *cobra.Command {
|
||||
Exporter = ctl.NewExportCommand(os.Stdin, os.Stdout, os.Stderr)
|
||||
func newExportCommand(stdin io.Reader, stdout, stderr io.Writer) *cobra.Command {
|
||||
Exporter = ctl.NewExportCommand(stdin, stdout, stderr)
|
||||
exportCmd := &cobra.Command{
|
||||
Use: "export",
|
||||
Short: "Export data from pilosa.",
|
||||
|
|
|
|||
|
|
@ -17,7 +17,6 @@ package cmd
|
|||
import (
|
||||
"context"
|
||||
"io"
|
||||
"os"
|
||||
|
||||
"github.com/spf13/cobra"
|
||||
|
||||
|
|
@ -26,8 +25,8 @@ import (
|
|||
|
||||
var generateConf *ctl.GenerateConfigCommand
|
||||
|
||||
func newGenerateConfigCommand(_ io.Reader, _, _ io.Writer) *cobra.Command {
|
||||
generateConf = ctl.NewGenerateConfigCommand(os.Stdin, os.Stdout, os.Stderr)
|
||||
func newGenerateConfigCommand(stdin io.Reader, stdout io.Writer, stderr io.Writer) *cobra.Command {
|
||||
generateConf = ctl.NewGenerateConfigCommand(stdin, stdout, stderr)
|
||||
confCmd := &cobra.Command{
|
||||
Use: "generate-config",
|
||||
Short: "Print the default configuration.",
|
||||
|
|
|
|||
|
|
@ -51,7 +51,8 @@ omitted. If it is present then its format should be YYYY-MM-DDTHH:MM.
|
|||
flags.StringVarP(&Importer.Host, "host", "", "localhost:10101", "host:port of Pilosa.")
|
||||
flags.StringVarP(&Importer.Index, "index", "i", "", "Pilosa index to import into.")
|
||||
flags.StringVarP(&Importer.Field, "field", "f", "", "Field to import into.")
|
||||
flags.BoolVar(&Importer.StringKeys, "string-keys", false, "REMOVED (key type is now determined by index/field configuration): Treat payload as string keys.")
|
||||
flags.BoolVar(&Importer.IndexKeys, "index-keys", false, "use keys=true when creating an index")
|
||||
flags.BoolVar(&Importer.FieldKeys, "field-keys", false, "use keys=true when creating a field")
|
||||
flags.IntVarP(&Importer.BufferSize, "buffer-size", "s", 10000000, "Number of bits to buffer/sort before importing.")
|
||||
flags.BoolVarP(&Importer.Sort, "sort", "", false, "Enables sorting before import.")
|
||||
flags.BoolVarP(&Importer.CreateSchema, "create", "e", false, "Create the schema if it does not exist before import.")
|
||||
|
|
|
|||
|
|
@ -18,7 +18,6 @@ import (
|
|||
"context"
|
||||
"fmt"
|
||||
"io"
|
||||
"os"
|
||||
|
||||
"github.com/spf13/cobra"
|
||||
|
||||
|
|
@ -27,8 +26,8 @@ import (
|
|||
|
||||
var inspector *ctl.InspectCommand
|
||||
|
||||
func newInspectCommand(_ io.Reader, _, _ io.Writer) *cobra.Command {
|
||||
inspector = ctl.NewInspectCommand(os.Stdin, os.Stdout, os.Stderr)
|
||||
func newInspectCommand(stdin io.Reader, stdout, stderr io.Writer) *cobra.Command {
|
||||
inspector = ctl.NewInspectCommand(stdin, stdout, stderr)
|
||||
|
||||
inspectCmd := &cobra.Command{
|
||||
Use: "inspect",
|
||||
|
|
|
|||
|
|
@ -40,15 +40,20 @@ type ImportCommand struct { // nolint: maligned
|
|||
Index string `json:"index"`
|
||||
Field string `json:"field"`
|
||||
|
||||
// Options for index & field to be created if they don't exist
|
||||
// Options for the index to be created if it doesn't exist
|
||||
indexOptions pilosa.IndexOptions
|
||||
|
||||
// Options for the field to be created if it doesn't exist
|
||||
fieldOptions pilosa.FieldOptions
|
||||
|
||||
// CreateSchema ensures the schema exists before import
|
||||
CreateSchema bool
|
||||
|
||||
// REMOVED: Indicates that the payload should be treated as string keys.
|
||||
// TODO: remove this in a future release
|
||||
StringKeys bool `json:"StringKeys"`
|
||||
// IndexKeys makes the import command use keys=true when creating an index
|
||||
IndexKeys bool `json:"indexKeys"`
|
||||
|
||||
// FieldKeys makes the import command use keys=true when creating a field
|
||||
FieldKeys bool `json:"fieldKeys"`
|
||||
|
||||
// Filenames to import from.
|
||||
Paths []string `json:"paths"`
|
||||
|
|
@ -80,11 +85,6 @@ func NewImportCommand(stdin io.Reader, stdout, stderr io.Writer) *ImportCommand
|
|||
func (cmd *ImportCommand) Run(ctx context.Context) error {
|
||||
logger := log.New(cmd.Stderr, "", log.LstdFlags)
|
||||
|
||||
// REMOVED: warning that --string-keys flag has been deprecated.
|
||||
if cmd.StringKeys {
|
||||
logger.Printf("REMOVED: The string-keys flag is no longer used.")
|
||||
}
|
||||
|
||||
// Validate arguments.
|
||||
// Index and field are validated early before the files are parsed.
|
||||
if cmd.Index == "" {
|
||||
|
|
@ -102,6 +102,12 @@ func (cmd *ImportCommand) Run(ctx context.Context) error {
|
|||
cmd.client = client
|
||||
|
||||
if cmd.CreateSchema {
|
||||
cmd.indexOptions = pilosa.IndexOptions{
|
||||
Keys: cmd.IndexKeys,
|
||||
}
|
||||
cmd.fieldOptions = pilosa.FieldOptions{
|
||||
Keys: cmd.FieldKeys,
|
||||
}
|
||||
err := cmd.ensureSchema(ctx)
|
||||
if err != nil {
|
||||
return errors.Wrap(err, "ensuring schema")
|
||||
|
|
@ -146,7 +152,7 @@ func (cmd *ImportCommand) ensureSchema(ctx context.Context) error {
|
|||
if err != nil {
|
||||
return fmt.Errorf("Error Creating Index: %s", err)
|
||||
}
|
||||
err = cmd.client.EnsureField(ctx, cmd.Index, cmd.Field)
|
||||
err = cmd.client.EnsureFieldWithOptions(ctx, cmd.Index, cmd.Field, cmd.fieldOptions)
|
||||
if err != nil {
|
||||
return fmt.Errorf("Error Creating Field: %s", err)
|
||||
}
|
||||
|
|
|
|||
|
|
@ -372,7 +372,11 @@ func (c *InternalClient) EnsureIndex(ctx context.Context, name string, options p
|
|||
}
|
||||
|
||||
func (c *InternalClient) EnsureField(ctx context.Context, indexName string, fieldName string) error {
|
||||
err := c.CreateField(ctx, indexName, fieldName)
|
||||
return c.EnsureFieldWithOptions(ctx, indexName, fieldName, pilosa.FieldOptions{})
|
||||
}
|
||||
|
||||
func (c *InternalClient) EnsureFieldWithOptions(ctx context.Context, indexName string, fieldName string, opt pilosa.FieldOptions) error {
|
||||
err := c.CreateFieldWithOptions(ctx, indexName, fieldName, opt)
|
||||
if err == nil || errors.Cause(err) == pilosa.ErrFieldExists {
|
||||
return nil
|
||||
}
|
||||
|
|
@ -666,16 +670,35 @@ func (c *InternalClient) backupShardNode(ctx context.Context, index, field strin
|
|||
return resp.Body, nil
|
||||
}
|
||||
|
||||
// CreateField creates a new field on the server.
|
||||
func (c *InternalClient) CreateField(ctx context.Context, index, field string) error {
|
||||
return c.CreateFieldWithOptions(ctx, index, field, pilosa.FieldOptions{})
|
||||
}
|
||||
|
||||
// CreateField creates a new field on the server.
|
||||
func (c *InternalClient) CreateFieldWithOptions(ctx context.Context, index, field string, opt pilosa.FieldOptions) error {
|
||||
if index == "" {
|
||||
return pilosa.ErrIndexRequired
|
||||
}
|
||||
|
||||
// convert pilosa.FieldOptions to fieldOptions
|
||||
fieldOpt := fieldOptions{
|
||||
Type: opt.Type,
|
||||
Keys: &opt.Keys,
|
||||
}
|
||||
if fieldOpt.Type == "set" {
|
||||
fieldOpt.CacheType = &opt.CacheType
|
||||
fieldOpt.CacheSize = &opt.CacheSize
|
||||
} else if fieldOpt.Type == "int" {
|
||||
fieldOpt.Min = &opt.Min
|
||||
fieldOpt.Max = &opt.Max
|
||||
} else if fieldOpt.Type == "time" {
|
||||
fieldOpt.TimeQuantum = &opt.TimeQuantum
|
||||
}
|
||||
|
||||
// TODO: remove buf completely? (depends on whether importer needs to create specific field types)
|
||||
// Encode query request.
|
||||
buf, err := json.Marshal(&postFieldRequest{
|
||||
//Options: opt,
|
||||
Options: fieldOpt,
|
||||
})
|
||||
if err != nil {
|
||||
return errors.Wrap(err, "marshaling")
|
||||
|
|
|
|||
|
|
@ -129,40 +129,40 @@ func (m *Command) Reopen() error {
|
|||
|
||||
// MustCreateIndex uses this command's API to create an index and fails the test
|
||||
// if there is an error.
|
||||
func (m *Command) MustCreateIndex(t *testing.T, name string, opts pilosa.IndexOptions) *pilosa.Index {
|
||||
func (m *Command) MustCreateIndex(tb testing.TB, name string, opts pilosa.IndexOptions) *pilosa.Index {
|
||||
idx, err := m.API.CreateIndex(context.Background(), name, opts)
|
||||
if err != nil {
|
||||
t.Fatalf("creating index: %v with options: %v, err: %v", name, opts, err)
|
||||
tb.Fatalf("creating index: %v with options: %v, err: %v", name, opts, err)
|
||||
}
|
||||
return idx
|
||||
}
|
||||
|
||||
// MustCreateField uses this command's API to create the field. The index must
|
||||
// already exist - it fails the test if there is an error.
|
||||
func (m *Command) MustCreateField(t *testing.T, index, field string, opts ...pilosa.FieldOption) *pilosa.Field {
|
||||
func (m *Command) MustCreateField(tb testing.TB, index, field string, opts ...pilosa.FieldOption) *pilosa.Field {
|
||||
f, err := m.API.CreateField(context.Background(), index, field, opts...)
|
||||
if err != nil {
|
||||
t.Fatalf("creating field: %s in index: %s err: %v", field, index, err)
|
||||
tb.Fatalf("creating field: %s in index: %s err: %v", field, index, err)
|
||||
}
|
||||
return f
|
||||
}
|
||||
|
||||
// MustQuery uses this command's API to execute the given query request, failing
|
||||
// if Query returns a non-nil error, otherwise returning the QueryResponse.
|
||||
func (m *Command) MustQuery(t *testing.T, req *pilosa.QueryRequest) pilosa.QueryResponse {
|
||||
func (m *Command) MustQuery(tb testing.TB, req *pilosa.QueryRequest) pilosa.QueryResponse {
|
||||
resp, err := m.API.Query(context.Background(), req)
|
||||
if err != nil {
|
||||
t.Fatalf("making query: %v, err: %v", req, err)
|
||||
tb.Fatalf("making query: %v, err: %v", req, err)
|
||||
}
|
||||
return resp
|
||||
}
|
||||
|
||||
// MustRecalculateCaches calls RecalculateCaches on the command's API, and fails
|
||||
// if there is an error.
|
||||
func (m *Command) MustRecalculateCaches(t *testing.T) {
|
||||
func (m *Command) MustRecalculateCaches(tb testing.TB) {
|
||||
err := m.API.RecalculateCaches(context.Background())
|
||||
if err != nil {
|
||||
t.Fatalf("recalcluating caches: %v", err)
|
||||
tb.Fatalf("recalcluating caches: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -224,10 +224,10 @@ func (c Cluster) Close() error {
|
|||
}
|
||||
|
||||
// MustNewCluster creates a new cluster
|
||||
func MustNewCluster(t *testing.T, size int, opts ...[]server.CommandOption) Cluster {
|
||||
func MustNewCluster(tb testing.TB, size int, opts ...[]server.CommandOption) Cluster {
|
||||
c, err := newCluster(size, opts...)
|
||||
if err != nil {
|
||||
t.Fatalf("new cluster: %v", err)
|
||||
tb.Fatalf("new cluster: %v", err)
|
||||
}
|
||||
return c
|
||||
}
|
||||
|
|
@ -271,10 +271,10 @@ func runCluster(size int, opts ...[]server.CommandOption) (Cluster, error) {
|
|||
}
|
||||
|
||||
// MustRunCluster creates and starts a new cluster
|
||||
func MustRunCluster(t *testing.T, size int, opts ...[]server.CommandOption) Cluster {
|
||||
func MustRunCluster(tb testing.TB, size int, opts ...[]server.CommandOption) Cluster {
|
||||
c, err := runCluster(size, opts...)
|
||||
if err != nil {
|
||||
t.Fatalf("run cluster: %v", err)
|
||||
tb.Fatalf("run cluster: %v", err)
|
||||
}
|
||||
return c
|
||||
}
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue