From e9b92e1cd45c4a0d67fcdc61a74d93770e2c9502 Mon Sep 17 00:00:00 2001 From: Nia Weiss Date: Thu, 25 Mar 2021 13:13:20 -0400 Subject: [PATCH 1/3] add ExternalLookup query --- .circleci/config.yml | 17 +++ Makefile | 2 + ctl/server.go | 3 + executor.go | 236 ++++++++++++++++++++++++++++++++++++ executor_test.go | 277 +++++++++++++++++++++++++++++++++++++++++++ holder.go | 24 ++++ pql/ast.go | 6 + server.go | 10 ++ server/config.go | 3 + server/server.go | 4 + 10 files changed, 582 insertions(+) diff --git a/.circleci/config.yml b/.circleci/config.yml index 82e8f175b..0b37e827b 100644 --- a/.circleci/config.yml +++ b/.circleci/config.yml @@ -108,6 +108,19 @@ jobs: - run: command: make << parameters.test_make_target >> SHARD_WIDTH=<< parameters.shard_width >> GOARCH=<< parameters.goarch >> no_output_timeout: 30m + test-external-lookup: + docker: + - image: circleci/golang:1.15.8 + - image: circleci/postgres:13.2-ram + environment: + POSTGRES_PASSWORD=password + steps: + - checkout-plus + - run: sudo apt-get install postgresql-client + - run: (for i in `seq 1 20`; do pg_isready -h localhost && exit 0 || sleep 1; done; exit 1) + - run: + command: make test-external-lookup EXTERNAL_LOOKUP_DSN=postgresql://postgres:password@localhost/circle_test?sslmode=disable + no_output_timeout: 30m cluster-tests: executor: name: golang @@ -226,6 +239,10 @@ workflows: resource_class: large requires: - setup + - test-external-lookup: + context: molecula + requires: + - setup - cluster-tests: context: molecula requires: diff --git a/Makefile b/Makefile index dd9b03606..f4efc6784 100644 --- a/Makefile +++ b/Makefile @@ -341,3 +341,5 @@ test-txstore-rbf: test-txstore-rbf_bolt: PILOSA_STORAGE_BACKEND=rbf_bolt $(MAKE) testv-race +test-external-lookup: + $(GO) test . -tags='$(BUILD_TAGS) $(TEST_TAGS)' $(TESTFLAGS) -run ^TestExternalLookup$$ -externalLookupDSN $(EXTERNAL_LOOKUP_DSN) \ No newline at end of file diff --git a/ctl/server.go b/ctl/server.go index fd5365bc8..c451b5c78 100644 --- a/ctl/server.go +++ b/ctl/server.go @@ -66,6 +66,9 @@ func BuildServerFlags(cmd *cobra.Command, srv *server.Command) { flags.StringVar(&srv.Config.Etcd.ClusterURL, "etcd.cluster-url", srv.Config.Etcd.ClusterURL, "Cluster URL to join.") flags.StringVar(&srv.Config.Etcd.InitCluster, "etcd.initial-cluster", srv.Config.Etcd.InitCluster, "Initial cluster name1=apurl1,name2=apurl2") + // External DB + flags.StringVar(&srv.Config.ExternalDB, "externaldb", "", "external (postgres) database DSN to use for ExternalQuery calls") + // AntiEntropy flags.DurationVar((*time.Duration)(&srv.Config.AntiEntropy.Interval), "anti-entropy.interval", (time.Duration)(srv.Config.AntiEntropy.Interval), "Interval at which to run anti-entropy routine.") diff --git a/executor.go b/executor.go index bc9b3014e..4ae6044fc 100644 --- a/executor.go +++ b/executor.go @@ -16,10 +16,12 @@ package pilosa import ( "context" + "database/sql" "encoding/json" "fmt" "math" "math/bits" + "reflect" "sort" "strings" "sync" @@ -28,6 +30,7 @@ import ( "golang.org/x/sync/errgroup" + "github.com/lib/pq" "github.com/pilosa/pilosa/v2/disco" "github.com/pilosa/pilosa/v2/pql" "github.com/pilosa/pilosa/v2/proto" @@ -799,6 +802,10 @@ func (e *executor) executeCall(ctx context.Context, qcx *Qcx, index string, c *p statFn() res, err := e.executeRows(ctx, qcx, index, c, shards, opt) return res, errors.Wrapf(err, "executeRows %v", shardSlice(shards)) + case "ExternalLookup": + statFn() + res, err := e.executeExternalLookup(ctx, qcx, index, c, shards, opt) + return res, errors.Wrapf(err, "executeExternalLookup %v", shardSlice(shards)) case "Extract": statFn() res, err := e.executeExtract(ctx, qcx, index, c, shards, opt) @@ -3992,6 +3999,235 @@ func (e *ExtractedIDMatrix) Append(m ExtractedIDMatrix) { } } +var ( + typeSQLNullString = reflect.TypeOf(sql.NullString{}) + typeSQLNullBool = reflect.TypeOf(sql.NullBool{}) + typeSQLNullInt32 = reflect.TypeOf(sql.NullInt32{}) + typeSQLNullInt64 = reflect.TypeOf(sql.NullInt64{}) +) + +func (e *executor) executeExternalLookup(ctx context.Context, qcx *Qcx, index string, c *pql.Call, shards []uint64, opt *execOptions) (res ExtractedTable, err error) { + if e.Holder.externalDB == nil { + return ExtractedTable{}, errors.New("external DB connection is not configured") + } + + idx := e.Holder.Index(index) + if idx == nil { + return ExtractedTable{}, errors.Errorf("index not found: %q", index) + } + + query, ok, err := c.StringArg("query") + if err != nil { + return ExtractedTable{}, errors.Wrap(err, "looking up query") + } + if !ok { + return ExtractedTable{}, errors.New("missing query") + } + + switch len(c.Children) { + case 0: + return ExtractedTable{}, errors.New("missing lookup input") + case 1: + default: + return ExtractedTable{}, errors.New("too many inputs to lookup query") + } + + rawArg, err := e.executeCall(ctx, qcx, index, c.Children[0], shards, opt) + if err != nil { + return ExtractedTable{}, errors.Wrapf(err, "evaluating SQL argument call %q", c.String()) + } + + qr := []interface{}{rawArg} + err = e.translateResults(ctx, index, idx, c.Children, qr) + if err != nil { + return ExtractedTable{}, errors.Wrap(err, "translating query result") + } + argRow := qr[0].(*Row) + if !argRow.Any() { + // If we attempt to substitute in an empty slice, the substitution will fail. + // Do not attempt to execute the query. + return ExtractedTable{}, nil + } + + var arg interface{} + if argRow.Keys != nil { + arg = argRow.Keys + } else { + arg = argRow.Columns() + } + + result, err := e.Holder.externalDB.QueryContext(ctx, query, pq.Array(arg)) + if err != nil { + return ExtractedTable{}, errors.Wrapf(err, "SQL query failed") + } + defer func() { + cerr := result.Close() + if cerr != nil && err == nil { + err = cerr + } + }() + if !result.Next() { + return ExtractedTable{}, errors.Wrap(result.Err(), "reading SQL query result") + } + + colTypes, err := result.ColumnTypes() + if err != nil { + return ExtractedTable{}, errors.Wrapf(err, "fetching SQL query result types") + } + + scanSlots := make([]interface{}, len(colTypes)) + scanMapper := make([]func() interface{}, len(colTypes)) + header := make([]ExtractedTableField, len(colTypes)) + for i, colType := range colTypes { + scanType := colType.ScanType() + setupScan: + if scanType.PkgPath() != "" { + // This is a named type. + switch scanType { + case typeSQLNullString: + var dst sql.NullString + scanSlots[i] = &dst + scanMapper[i] = func() interface{} { + if !dst.Valid { + return nil + } + + return dst.String + } + header[i] = ExtractedTableField{ + Name: colType.Name(), + Type: "string", + } + + case typeSQLNullBool: + var dst sql.NullBool + scanSlots[i] = &dst + scanMapper[i] = func() interface{} { + if !dst.Valid { + return nil + } + + return dst.Bool + } + header[i] = ExtractedTableField{ + Name: colType.Name(), + Type: "bool", + } + + case typeSQLNullInt32: + var dst sql.NullInt32 + scanSlots[i] = &dst + scanMapper[i] = func() interface{} { + if !dst.Valid { + return nil + } + + return int64(dst.Int32) + } + header[i] = ExtractedTableField{ + Name: colType.Name(), + Type: "int64", + } + + case typeSQLNullInt64: + var dst sql.NullInt64 + scanSlots[i] = &dst + scanMapper[i] = func() interface{} { + if !dst.Valid { + return nil + } + + return dst.Int64 + } + header[i] = ExtractedTableField{ + Name: colType.Name(), + Type: "int64", + } + + default: + return ExtractedTable{}, errors.Errorf("unable to process result type %v from SQL query %q", scanType, query) + } + continue + } + + switch scanType.Kind() { + case reflect.Uint, reflect.Uint8, reflect.Uint16, reflect.Uint32, reflect.Uint64: + var dst uint64 + scanSlots[i] = &dst + scanMapper[i] = func() interface{} { return dst } + header[i] = ExtractedTableField{ + Name: colType.Name(), + Type: "uint64", + } + + case reflect.Int, reflect.Int8, reflect.Int16, reflect.Int32, reflect.Int64: + // The database may have lied and this may be nullable. + scanType = typeSQLNullInt64 + goto setupScan + + case reflect.String: + // The database may have lied and this may be nullable. + scanType = typeSQLNullString + goto setupScan + + case reflect.Bool: + // The database may have lied and this may be nullable. + scanType = typeSQLNullBool + goto setupScan + + default: + return ExtractedTable{}, errors.Errorf("unable to process result type %v from SQL query %q", scanType, query) + } + } + + var columns []ExtractedTableColumn + for { + err := result.Scan(scanSlots...) + if err != nil { + return ExtractedTable{}, errors.Wrap(err, "scanning SQL result") + } + + var col KeyOrID + switch v := scanMapper[0]().(type) { + case nil: + return ExtractedTable{}, errors.Errorf("missing primary key in result") + case uint64: + col.ID = v + case int64: + col.ID = uint64(v) + case string: + col.Keyed = true + col.Key = v + default: + return ExtractedTable{}, errors.Wrap(err, "cannot use %v of type %T as primary key for result table") + } + + data := make([]interface{}, len(scanMapper)-1) + for i, m := range scanMapper[1:] { + data[i] = m() + } + + columns = append(columns, ExtractedTableColumn{ + Column: col, + Rows: data, + }) + + if !result.Next() { + break + } + } + + err = result.Err() + if err != nil { + return ExtractedTable{}, errors.Wrap(err, "reading SQL result") + } + + return ExtractedTable{ + Fields: header[1:], + Columns: columns, + }, nil +} + func (e *executor) executeExtract(ctx context.Context, qcx *Qcx, index string, c *pql.Call, shards []uint64, opt *execOptions) (ExtractedIDMatrix, error) { // Extract the column filter call. if len(c.Children) < 1 { diff --git a/executor_test.go b/executor_test.go index 3f33c8415..8f2abd09c 100644 --- a/executor_test.go +++ b/executor_test.go @@ -17,6 +17,7 @@ package pilosa_test import ( "bytes" "context" + "database/sql" "encoding/csv" "encoding/json" "flag" @@ -7888,3 +7889,279 @@ func tableResponseToCSVString(m *proto.TableResponse) (string, error) { } return buf.String(), nil } + +var dbDSN string + +func init() { + flag.StringVar(&dbDSN, "externalLookupDSN", "", "SQL DSN to use for external database access") +} + +func TestExternalLookup(t *testing.T) { + // Set up access to a SQL database. + if dbDSN == "" { + t.Skip("no database provided") + } + db, err := sql.Open("postgres", dbDSN) + if err != nil { + t.Fatalf("failed to set up test database: %v", err) + } + defer func() { + if cerr := db.Close(); cerr != nil { + t.Errorf("failed to close test database: %v", cerr) + } + }() + + // Set up some data to use in the SQL DB. + // This creates 3 tables: + // - "lookup" - which stores an id->string mapping + // - "misc" - which has misc non-nullable fields + // - "nullable" - to test handling of nullable fields + func() { + // Set up a write transaction. + // This uses a function scope so that the defer is scoped appropriately. + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + tx, err := db.BeginTx(ctx, nil) + if err != nil { + t.Fatalf("failed to create transaction: %v", err) + } + var ok bool + defer func() { + if !ok { + rerr := tx.Rollback() + if rerr != nil { + t.Errorf("failed to roll-back transaction: %v", rerr) + } + } + }() + + // Delete all tables so that we start fresh. + _, err = tx.Exec(`DROP INDEX IF EXISTS lookupIndex`) + if err != nil { + t.Fatalf("deleting lookup index: %v", err) + } + _, err = tx.Exec(`DROP TABLE IF EXISTS lookup`) + if err != nil { + t.Fatalf("deleting lookup table: %v", err) + } + _, err = tx.Exec(`DROP TABLE IF EXISTS misc`) + if err != nil { + t.Fatalf("deleting misc table: %v", err) + } + _, err = tx.Exec(`DROP TABLE IF EXISTS nullable`) + if err != nil { + t.Fatalf("deleting nullable table: %v", err) + } + + _, err = tx.Exec(`CREATE TABLE lookup ( + id int NOT NULL, + data text NOT NULL + )`) + if err != nil { + t.Fatalf("failed to create lookup table: %v", err) + } + _, err = tx.Exec(`INSERT INTO lookup (id, data) VALUES + (0, 'h'), + (1, 'xyzzy'), + (2, 'plugh'), + (3, 'wow') + `) + if err != nil { + t.Fatalf("failed to populate lookup table: %v", err) + } + _, err = tx.Exec(`CREATE UNIQUE INDEX lookupIndex ON lookup (id);`) + if err != nil { + t.Fatalf("failed to index lookup table: %v", err) + } + + _, err = tx.Exec(`CREATE TABLE misc ( + id int NOT NULL, + stringval text NOT NULL, + boolval boolean NOT NULL, + intval int NOT NULL + )`) + if err != nil { + t.Fatalf("failed to create misc table: %v", err) + } + _, err = tx.Exec(`INSERT INTO misc (id, stringval, boolval, intval) VALUES + (0, 'h', true, 4), + (1, 'y', false, 11) + `) + if err != nil { + t.Fatalf("failed to populate misc table: %v", err) + } + + _, err = tx.Exec(`CREATE TABLE nullable ( + id int NOT NULL, + stringval text, + boolval boolean, + intval int + )`) + if err != nil { + t.Fatalf("failed to create nullable table: %v", err) + } + _, err = tx.Exec(`INSERT INTO nullable (id, stringval, boolval, intval) VALUES + (0, 'h', true, 4), + (1, 'y', false, 11), + (2, null, null, null), + (3, null, true, null), + (4, 'plugh', null, 0) + ;`) + if err != nil { + t.Fatalf("failed to populate nullable table: %v", err) + } + + err = tx.Commit() + if err != nil { + t.Fatalf("failed to commit DB setup transaction: %v", err) + } + + ok = true + }() + + // Start up a Pilosa cluster with access to the DB. + c := test.MustRunCluster(t, 3, []server.CommandOption{server.OptCommandServerOptions(pilosa.OptServerExternalDB(dbDSN))}) + defer c.Close() + + // Populate a field with some data that can be used in queries. + c.CreateField(t, "i", pilosa.IndexOptions{}, "f") + c.ImportBits(t, "i", "f", [][2]uint64{ + {1, 1}, + {1, 3}, + {2, 2}, + {2, 3}, + }) + + cases := []struct { + name string + query string + expect pilosa.QueryResponse + }{ + { + name: "Empty", + query: `ExternalLookup(Union(), query="select * from lookup where id = ANY($1)")`, + expect: pilosa.QueryResponse{ + Results: []interface{}{ + pilosa.ExtractedTable{}, + }, + }, + }, + { + name: "ConstRowLookup", + query: `ExternalLookup(ConstRow(columns=[1, 3]), query="select id, data from lookup where id = ANY($1)")`, + expect: pilosa.QueryResponse{ + Results: []interface{}{ + pilosa.ExtractedTable{ + Fields: []pilosa.ExtractedTableField{ + {Name: "data", Type: "string"}, + }, + Columns: []pilosa.ExtractedTableColumn{ + { + Column: pilosa.KeyOrID{ID: 1}, + Rows: []interface{}{"xyzzy"}, + }, + { + Column: pilosa.KeyOrID{ID: 3}, + Rows: []interface{}{"wow"}, + }, + }, + }, + }, + }, + }, + { + name: "ComputedFilter", + query: `ExternalLookup(Intersect(Row(f=1), Row(f=2)), query="select id, data from lookup where id = ANY($1)")`, + expect: pilosa.QueryResponse{ + Results: []interface{}{ + pilosa.ExtractedTable{ + Fields: []pilosa.ExtractedTableField{ + {Name: "data", Type: "string"}, + }, + Columns: []pilosa.ExtractedTableColumn{ + { + Column: pilosa.KeyOrID{ID: 3}, + Rows: []interface{}{"wow"}, + }, + }, + }, + }, + }, + }, + { + name: "Misc", + query: `ExternalLookup(ConstRow(columns=[0, 1]), query="select id, stringval, boolval, intval from misc where id = ANY($1)")`, + expect: pilosa.QueryResponse{ + Results: []interface{}{ + pilosa.ExtractedTable{ + Fields: []pilosa.ExtractedTableField{ + {Name: "stringval", Type: "string"}, + {Name: "boolval", Type: "bool"}, + {Name: "intval", Type: "int64"}, + }, + Columns: []pilosa.ExtractedTableColumn{ + { + Column: pilosa.KeyOrID{ID: 0}, + Rows: []interface{}{"h", true, int64(4)}, + }, + { + Column: pilosa.KeyOrID{ID: 1}, + Rows: []interface{}{"y", false, int64(11)}, + }, + }, + }, + }, + }, + }, + { + name: "Nullable", + query: `ExternalLookup(ConstRow(columns=[0, 1, 2, 3, 4]), query="select id, stringval, boolval, intval from nullable where id = ANY($1)")`, + expect: pilosa.QueryResponse{ + Results: []interface{}{ + pilosa.ExtractedTable{ + Fields: []pilosa.ExtractedTableField{ + {Name: "stringval", Type: "string"}, + {Name: "boolval", Type: "bool"}, + {Name: "intval", Type: "int64"}, + }, + Columns: []pilosa.ExtractedTableColumn{ + { + Column: pilosa.KeyOrID{ID: 0}, + Rows: []interface{}{"h", true, int64(4)}, + }, + { + Column: pilosa.KeyOrID{ID: 1}, + Rows: []interface{}{"y", false, int64(11)}, + }, + { + Column: pilosa.KeyOrID{ID: 2}, + Rows: []interface{}{nil, nil, nil}, + }, + { + Column: pilosa.KeyOrID{ID: 3}, + Rows: []interface{}{nil, true, nil}, + }, + { + Column: pilosa.KeyOrID{ID: 4}, + Rows: []interface{}{"plugh", nil, int64(0)}, + }, + }, + }, + }, + }, + }, + } + t.Run("Query", func(t *testing.T) { + for _, tc := range cases { + tc := tc + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + + result := c.Query(t, "i", tc.query) + if !reflect.DeepEqual(result, tc.expect) { + t.Errorf("expected %v but got %v", tc.expect, result) + } + }) + } + }) +} diff --git a/holder.go b/holder.go index 8df98105d..fa9efe254 100644 --- a/holder.go +++ b/holder.go @@ -16,6 +16,7 @@ package pilosa import ( "context" + "database/sql" "fmt" "os" "path/filepath" @@ -150,6 +151,8 @@ type Holder struct { txf *TxFactory + externalDB *sql.DB + // a separate lock out for indexes, to avoid the deadlock/race dilema // on holding mu. imu sync.RWMutex @@ -239,6 +242,8 @@ type HolderConfig struct { StorageConfig *storage.Config RBFConfig *rbfcfg.Config AntiEntropyInterval time.Duration + + ExternalDB string } func DefaultHolderConfig() *HolderConfig { @@ -728,6 +733,17 @@ func (h *Holder) Open() error { h.txf.blueGreenOnIfRunningBlueGreen() + if h.cfg.ExternalDB != "" { + h.Logger.Printf("connecting to external DB") + + db, err := sql.Open("postgres", h.cfg.ExternalDB) + if err != nil { + return errors.Wrap(err, "connecting to external database") + } + + h.externalDB = db + } + h.Logger.Printf("open holder: complete") return nil @@ -838,6 +854,14 @@ func (h *Holder) Close() error { h.SnapshotQueue = nil } + if h.externalDB != nil { + err := h.externalDB.Close() + if err != nil { + return errors.Wrap(err, "closing DB") + } + h.externalDB = nil + } + _ = testhook.Closed(h.Auditor, h, nil) return nil diff --git a/pql/ast.go b/pql/ast.go index e6155ac7f..f0c0d38b3 100644 --- a/pql/ast.go +++ b/pql/ast.go @@ -400,6 +400,12 @@ var callInfoByFunc = map[string]callInfo{ "Union": {allowUnknown: false}, "UnionRows": {allowUnknown: false, callType: PrecallGlobal}, "Extract": {allowUnknown: false}, + "ExternalLookup": { + allowUnknown: false, + prototypes: map[string]interface{}{ + "query": "", + }, + }, "Limit": { allowUnknown: false, prototypes: map[string]interface{}{ diff --git a/server.go b/server.go index fcd70ef13..6bbfa2ba4 100644 --- a/server.go +++ b/server.go @@ -40,6 +40,8 @@ import ( "github.com/pilosa/pilosa/v2/topology" "github.com/pkg/errors" "golang.org/x/sync/errgroup" + + _ "github.com/lib/pq" ) // Default server settings. @@ -398,6 +400,14 @@ func OptServerDisCo(disCo disco.DisCo, } } +// OptServerExternalDB configures a connection to an external postgres database. +func OptServerExternalDB(dsn string) ServerOption { + return func(s *Server) error { + s.holderConfig.ExternalDB = dsn + return nil + } +} + // NewServer returns a new instance of Server. func NewServer(opts ...ServerOption) (*Server, error) { cluster := newCluster() diff --git a/server/config.go b/server/config.go index edb69d5ec..9081e197c 100644 --- a/server/config.go +++ b/server/config.go @@ -222,6 +222,9 @@ type Config struct { // for the /query-history endpoint. This parameter is per-node, and the // result combines the history from all nodes. QueryHistoryLength int `toml:"query-history-length"` + + // ExternalDB is an external database to connect to for `ExternalLookup` queries. + ExternalDB string `toml:"external-db"` } // MustValidate checks that all ports in a Config are unique and not zero. diff --git a/server/server.go b/server/server.go index d4b47a6e1..4422858b0 100644 --- a/server/server.go +++ b/server/server.go @@ -429,6 +429,10 @@ func (m *Command) SetupServer() error { discoOpt, } + if m.Config.ExternalDB != "" { + serverOptions = append(serverOptions, pilosa.OptServerExternalDB(m.Config.ExternalDB)) + } + serverOptions = append(serverOptions, m.serverOptions...) m.Server, err = pilosa.NewServer(serverOptions...) From c4aad290ed38952ed44eb2642d300c72a9e9f361 Mon Sep 17 00:00:00 2001 From: Nia Weiss Date: Mon, 29 Mar 2021 08:22:41 -0400 Subject: [PATCH 2/3] address ExternalLookup review comments --- ctl/server.go | 2 +- executor.go | 7 ++++--- holder.go | 6 +++--- server.go | 2 +- server/config.go | 4 ++-- server/server.go | 4 ++-- 6 files changed, 13 insertions(+), 12 deletions(-) diff --git a/ctl/server.go b/ctl/server.go index c451b5c78..b43c8d907 100644 --- a/ctl/server.go +++ b/ctl/server.go @@ -67,7 +67,7 @@ func BuildServerFlags(cmd *cobra.Command, srv *server.Command) { flags.StringVar(&srv.Config.Etcd.InitCluster, "etcd.initial-cluster", srv.Config.Etcd.InitCluster, "Initial cluster name1=apurl1,name2=apurl2") // External DB - flags.StringVar(&srv.Config.ExternalDB, "externaldb", "", "external (postgres) database DSN to use for ExternalQuery calls") + flags.StringVar(&srv.Config.ExternalDBDSN, "external-db-dsn", "", "external (postgres) database DSN to use for ExternalLookup calls") // AntiEntropy flags.DurationVar((*time.Duration)(&srv.Config.AntiEntropy.Interval), "anti-entropy.interval", (time.Duration)(srv.Config.AntiEntropy.Interval), "Interval at which to run anti-entropy routine.") diff --git a/executor.go b/executor.go index 4ae6044fc..c6252c7ba 100644 --- a/executor.go +++ b/executor.go @@ -4042,10 +4042,11 @@ func (e *executor) executeExternalLookup(ctx context.Context, qcx *Qcx, index st if err != nil { return ExtractedTable{}, errors.Wrap(err, "translating query result") } - argRow := qr[0].(*Row) + argRow, ok := qr[0].(*Row) + if !ok { + return ExtractedTable{}, errors.Errorf("argument call result is a %T but expected a row", qr[0]) + } if !argRow.Any() { - // If we attempt to substitute in an empty slice, the substitution will fail. - // Do not attempt to execute the query. return ExtractedTable{}, nil } diff --git a/holder.go b/holder.go index fa9efe254..3f960433f 100644 --- a/holder.go +++ b/holder.go @@ -243,7 +243,7 @@ type HolderConfig struct { RBFConfig *rbfcfg.Config AntiEntropyInterval time.Duration - ExternalDB string + ExternalDBDSN string } func DefaultHolderConfig() *HolderConfig { @@ -733,10 +733,10 @@ func (h *Holder) Open() error { h.txf.blueGreenOnIfRunningBlueGreen() - if h.cfg.ExternalDB != "" { + if h.cfg.ExternalDBDSN != "" { h.Logger.Printf("connecting to external DB") - db, err := sql.Open("postgres", h.cfg.ExternalDB) + db, err := sql.Open("postgres", h.cfg.ExternalDBDSN) if err != nil { return errors.Wrap(err, "connecting to external database") } diff --git a/server.go b/server.go index 6bbfa2ba4..38a77dd39 100644 --- a/server.go +++ b/server.go @@ -403,7 +403,7 @@ func OptServerDisCo(disCo disco.DisCo, // OptServerExternalDB configures a connection to an external postgres database. func OptServerExternalDB(dsn string) ServerOption { return func(s *Server) error { - s.holderConfig.ExternalDB = dsn + s.holderConfig.ExternalDBDSN = dsn return nil } } diff --git a/server/config.go b/server/config.go index 9081e197c..484a4b0f1 100644 --- a/server/config.go +++ b/server/config.go @@ -223,8 +223,8 @@ type Config struct { // result combines the history from all nodes. QueryHistoryLength int `toml:"query-history-length"` - // ExternalDB is an external database to connect to for `ExternalLookup` queries. - ExternalDB string `toml:"external-db"` + // ExternalDBDSN is an external database to connect to for `ExternalLookup` queries. + ExternalDBDSN string `toml:"external-db-dsn"` } // MustValidate checks that all ports in a Config are unique and not zero. diff --git a/server/server.go b/server/server.go index 4422858b0..556e7b6b6 100644 --- a/server/server.go +++ b/server/server.go @@ -429,8 +429,8 @@ func (m *Command) SetupServer() error { discoOpt, } - if m.Config.ExternalDB != "" { - serverOptions = append(serverOptions, pilosa.OptServerExternalDB(m.Config.ExternalDB)) + if m.Config.ExternalDBDSN != "" { + serverOptions = append(serverOptions, pilosa.OptServerExternalDB(m.Config.ExternalDBDSN)) } serverOptions = append(serverOptions, m.serverOptions...) From 9bc1b23b7e9fca369b8a2ef2c592aa2f2216b3a0 Mon Sep 17 00:00:00 2001 From: Nia Weiss Date: Mon, 29 Mar 2021 12:54:29 -0400 Subject: [PATCH 3/3] change "External" DB to "Lookup" DB --- ctl/server.go | 4 ++-- executor.go | 4 ++-- executor_test.go | 2 +- holder.go | 20 ++++++++++---------- server.go | 6 +++--- server/config.go | 4 ++-- server/server.go | 4 ++-- 7 files changed, 22 insertions(+), 22 deletions(-) diff --git a/ctl/server.go b/ctl/server.go index b43c8d907..6c885d213 100644 --- a/ctl/server.go +++ b/ctl/server.go @@ -66,8 +66,8 @@ func BuildServerFlags(cmd *cobra.Command, srv *server.Command) { flags.StringVar(&srv.Config.Etcd.ClusterURL, "etcd.cluster-url", srv.Config.Etcd.ClusterURL, "Cluster URL to join.") flags.StringVar(&srv.Config.Etcd.InitCluster, "etcd.initial-cluster", srv.Config.Etcd.InitCluster, "Initial cluster name1=apurl1,name2=apurl2") - // External DB - flags.StringVar(&srv.Config.ExternalDBDSN, "external-db-dsn", "", "external (postgres) database DSN to use for ExternalLookup calls") + // External postgres database for ExternalLookup + flags.StringVar(&srv.Config.LookupDBDSN, "lookup-db-dsn", "", "external (postgres) database DSN to use for ExternalLookup calls") // AntiEntropy flags.DurationVar((*time.Duration)(&srv.Config.AntiEntropy.Interval), "anti-entropy.interval", (time.Duration)(srv.Config.AntiEntropy.Interval), "Interval at which to run anti-entropy routine.") diff --git a/executor.go b/executor.go index c6252c7ba..9cd5cccf9 100644 --- a/executor.go +++ b/executor.go @@ -4007,7 +4007,7 @@ var ( ) func (e *executor) executeExternalLookup(ctx context.Context, qcx *Qcx, index string, c *pql.Call, shards []uint64, opt *execOptions) (res ExtractedTable, err error) { - if e.Holder.externalDB == nil { + if e.Holder.lookupDB == nil { return ExtractedTable{}, errors.New("external DB connection is not configured") } @@ -4057,7 +4057,7 @@ func (e *executor) executeExternalLookup(ctx context.Context, qcx *Qcx, index st arg = argRow.Columns() } - result, err := e.Holder.externalDB.QueryContext(ctx, query, pq.Array(arg)) + result, err := e.Holder.lookupDB.QueryContext(ctx, query, pq.Array(arg)) if err != nil { return ExtractedTable{}, errors.Wrapf(err, "SQL query failed") } diff --git a/executor_test.go b/executor_test.go index 8f2abd09c..48ce5200d 100644 --- a/executor_test.go +++ b/executor_test.go @@ -8020,7 +8020,7 @@ func TestExternalLookup(t *testing.T) { }() // Start up a Pilosa cluster with access to the DB. - c := test.MustRunCluster(t, 3, []server.CommandOption{server.OptCommandServerOptions(pilosa.OptServerExternalDB(dbDSN))}) + c := test.MustRunCluster(t, 3, []server.CommandOption{server.OptCommandServerOptions(pilosa.OptServerLookupDB(dbDSN))}) defer c.Close() // Populate a field with some data that can be used in queries. diff --git a/holder.go b/holder.go index 3f960433f..766d82a7d 100644 --- a/holder.go +++ b/holder.go @@ -151,7 +151,7 @@ type Holder struct { txf *TxFactory - externalDB *sql.DB + lookupDB *sql.DB // a separate lock out for indexes, to avoid the deadlock/race dilema // on holding mu. @@ -243,7 +243,7 @@ type HolderConfig struct { RBFConfig *rbfcfg.Config AntiEntropyInterval time.Duration - ExternalDBDSN string + LookupDBDSN string } func DefaultHolderConfig() *HolderConfig { @@ -733,15 +733,15 @@ func (h *Holder) Open() error { h.txf.blueGreenOnIfRunningBlueGreen() - if h.cfg.ExternalDBDSN != "" { - h.Logger.Printf("connecting to external DB") + if h.cfg.LookupDBDSN != "" { + h.Logger.Printf("connecting to lookup DB") - db, err := sql.Open("postgres", h.cfg.ExternalDBDSN) + db, err := sql.Open("postgres", h.cfg.LookupDBDSN) if err != nil { - return errors.Wrap(err, "connecting to external database") + return errors.Wrap(err, "connecting to lookup database") } - h.externalDB = db + h.lookupDB = db } h.Logger.Printf("open holder: complete") @@ -854,12 +854,12 @@ func (h *Holder) Close() error { h.SnapshotQueue = nil } - if h.externalDB != nil { - err := h.externalDB.Close() + if h.lookupDB != nil { + err := h.lookupDB.Close() if err != nil { return errors.Wrap(err, "closing DB") } - h.externalDB = nil + h.lookupDB = nil } _ = testhook.Closed(h.Auditor, h, nil) diff --git a/server.go b/server.go index 38a77dd39..018d13eec 100644 --- a/server.go +++ b/server.go @@ -400,10 +400,10 @@ func OptServerDisCo(disCo disco.DisCo, } } -// OptServerExternalDB configures a connection to an external postgres database. -func OptServerExternalDB(dsn string) ServerOption { +// OptServerLookupDB configures a connection to an external postgres database for ExternalLookup queries. +func OptServerLookupDB(dsn string) ServerOption { return func(s *Server) error { - s.holderConfig.ExternalDBDSN = dsn + s.holderConfig.LookupDBDSN = dsn return nil } } diff --git a/server/config.go b/server/config.go index 484a4b0f1..074dbdc07 100644 --- a/server/config.go +++ b/server/config.go @@ -223,8 +223,8 @@ type Config struct { // result combines the history from all nodes. QueryHistoryLength int `toml:"query-history-length"` - // ExternalDBDSN is an external database to connect to for `ExternalLookup` queries. - ExternalDBDSN string `toml:"external-db-dsn"` + // LookupDBDSN is an external database to connect to for `ExternalLookup` queries. + LookupDBDSN string `toml:"lookup-db-dsn"` } // MustValidate checks that all ports in a Config are unique and not zero. diff --git a/server/server.go b/server/server.go index 556e7b6b6..58827eca0 100644 --- a/server/server.go +++ b/server/server.go @@ -429,8 +429,8 @@ func (m *Command) SetupServer() error { discoOpt, } - if m.Config.ExternalDBDSN != "" { - serverOptions = append(serverOptions, pilosa.OptServerExternalDB(m.Config.ExternalDBDSN)) + if m.Config.LookupDBDSN != "" { + serverOptions = append(serverOptions, pilosa.OptServerLookupDB(m.Config.LookupDBDSN)) } serverOptions = append(serverOptions, m.serverOptions...)