diff --git a/dax/queryer/queryer.go b/dax/queryer/queryer.go index 205c10e9f..58352b696 100644 --- a/dax/queryer/queryer.go +++ b/dax/queryer/queryer.go @@ -39,6 +39,8 @@ type Queryer struct { controller dax.Controller + systemLayer *systemlayer.SystemLayer + logger logger.Logger } @@ -47,6 +49,7 @@ func New(cfg Config) *Queryer { q := &Queryer{ controller: dax.NewNopController(), orchestrators: make(map[dax.QualifiedDatabaseID]*qualifiedOrchestrator), + systemLayer: systemlayer.NewSystemLayer(), logger: logger.NopLogger, } @@ -138,14 +141,18 @@ func (q *Queryer) QuerySQL(ctx context.Context, qdbid dax.QualifiedDatabaseID, s applyExecutionTime() } - // Peek at the first character of sql. If it's "[", then handle this as PQL. + // Peek at the first 4k characters of sql. This is limited to 4k to avoid + // reading in a very large BULK INSERT statement. If the first character is + // "[", then handle this as PQL. We keep more than the first character + // because the ExecutionPlanner as currently written expects a sql string + // (which it saves in the ExecutionRequests API). var isPQL bool - peekSize := 1 + peekSize := 4096 peeker := io.LimitReader(sql, int64(peekSize)) peek, err := io.ReadAll(peeker) if err != nil { return nil, errors.Wrap(err, "reading peeker") - } else if len(peek) == 1 && peek[0] == byte('[') { + } else if len(peek) > 0 && peek[0] == byte('[') { isPQL = true } @@ -182,6 +189,9 @@ func (q *Queryer) QuerySQL(ctx context.Context, qdbid dax.QualifiedDatabaseID, s // put the requestId in the context ctx = fbcontext.WithRequestID(ctx, requestID.String()) + userID := "travis" + ctx = fbcontext.WithUserID(ctx, userID) + st, err := parser.NewParser(multiReader).ParseStatement() if err != nil { applyError(errors.Wrap(err, "parsing sql")) @@ -198,13 +208,14 @@ func (q *Queryer) QuerySQL(ctx context.Context, qdbid dax.QualifiedDatabaseID, s // SystemAPI. sysapi := &featurebase.NopSystemAPI{} - systemLayer := systemlayer.NewSystemLayer() + // systemLayer.ExecutionRequests().AddRequest("reqid", "userid", time.Now(), "select foo") + // systemLayer.ExecutionRequests(). // We intentionally don't pass the sql argument here because we're working // with an io.Reader rather than a string and it's just not necessary to // send it as a string to this method. Also, what happens if the sql is a // large BULK INSERT? - pl := planner.NewExecutionPlanner(q.Orchestrator(qdbid), sapi, sysapi, systemLayer, imp, q.logger, "") + pl := planner.NewExecutionPlanner(q.Orchestrator(qdbid), sapi, sysapi, q.systemLayer, imp, q.logger, string(peek)) planOp, err := pl.CompilePlan(ctx, st) if err != nil { diff --git a/sql3/planner/opquery.go b/sql3/planner/opquery.go index a38b54cf2..1ef857a28 100644 --- a/sql3/planner/opquery.go +++ b/sql3/planner/opquery.go @@ -4,6 +4,7 @@ import ( "context" "encoding/json" "fmt" + "log" "time" pilosa "github.com/featurebasedb/featurebase/v3" @@ -122,6 +123,7 @@ func (i *queryIterator) Next(ctx context.Context) (types.Row, error) { userId := "" userId, _ = fbcontext.UserID(ctx) + log.Printf("DEEBUG: AddRequest: %s, %s, %s", requestId, userId, i.query.sql) i.requests.AddRequest(requestId, userId, time.Now(), i.query.sql) i.hasStarted = &struct{}{} } diff --git a/sql3/planner/opsystemtable.go b/sql3/planner/opsystemtable.go index c6c2c74f4..58123d76b 100644 --- a/sql3/planner/opsystemtable.go +++ b/sql3/planner/opsystemtable.go @@ -6,6 +6,7 @@ import ( "bytes" "context" "fmt" + "log" "sort" pilosa "github.com/featurebasedb/featurebase/v3" @@ -450,6 +451,7 @@ type fbExecRequestsRowIter struct { var _ types.RowIterator = (*fbExecRequestsRowIter)(nil) func (i *fbExecRequestsRowIter) Next(ctx context.Context) (types.Row, error) { + log.Printf("DEEBUG: fbExecRequestsRowIter.Next() ") if i.result == nil { var err error i.result, err = i.planner.systemLayerAPI.ExecutionRequests().ListRequests() diff --git a/systemlayer/executionrequests.go b/systemlayer/executionrequests.go index 29b5ff7b4..411e72280 100644 --- a/systemlayer/executionrequests.go +++ b/systemlayer/executionrequests.go @@ -2,6 +2,7 @@ package systemlayer import ( "fmt" + "log" "sync" "time" @@ -122,6 +123,7 @@ func (e *ExecutionRequests) UpdateRequest(requestID string, // ListRequests returns the content of the ExecutionRequests struct as copies func (e *ExecutionRequests) ListRequests() ([]pilosa.ExecutionRequest, error) { + log.Printf("DEEBUG: ListRequests() called, %p", e) e.RLock() defer e.RUnlock()