Merge pull request #948 from molecula/randomquery

pilosa/cmd/random-query: generate random queries from existing schema/data

cmd/random-query/main.go

@@ -0,0 +1,377 @@
+// Copyright 2020 Pilosa Corp.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package main
+
+import (
+	"context"
+	"flag"
+	"fmt"
+	"math/rand"
+	nethttp "net/http"
+	"os"
+	"strings"
+	"time"
+
+	"github.com/pilosa/pilosa/v2"
+	"github.com/pilosa/pilosa/v2/http"
+)
+
+// RandomQueryConfig
+type RandomQueryConfig struct {
+
+	// user facing flags
+	HostPort   string // -hostport
+	TreeDepth  int    // -d
+	QueryCount int    // -n
+	Verbose    bool   // -v
+
+	IndexMap map[string]*Features
+
+	API  *pilosa.API
+	Info []*pilosa.IndexInfo
+
+	BitmapFunc []string
+
+	Rnd *rand.Rand
+}
+
+type API interface {
+
+	// InternalClient
+	Schema(ctx context.Context) ([]*pilosa.IndexInfo, error)
+	Query(ctx context.Context, index string, queryRequest *pilosa.QueryRequest) (*pilosa.QueryResponse, error)
+
+	// API for contrast; just a little different:
+	//Schema(ctx context.Context) []*IndexInfo
+	//Query(ctx context.Context, req *pilosa.QueryRequest) (pilosa.QueryResponse, error)
+}
+
+// have to wrap because the ugly little differences between InternalClient and API
+type wrapper struct {
+	api *pilosa.API
+}
+
+func (w *wrapper) Schema(ctx context.Context) ([]*pilosa.IndexInfo, error) {
+	return w.api.Schema(ctx), nil
+}
+
+func (w *wrapper) Query(ctx context.Context, index string, queryRequest *pilosa.QueryRequest) (*pilosa.QueryResponse, error) {
+	r, err := w.api.Query(ctx, queryRequest)
+	return &r, err
+}
+
+func wrapApiToInternalClient(api *pilosa.API) *wrapper {
+	return &wrapper{api: api}
+}
+
+// call DefineFlags before myflags.Parse()
+func (cfg *RandomQueryConfig) DefineFlags(fs *flag.FlagSet) {
+	fs.StringVar(&cfg.HostPort, "hostport", "localhost:10101", "host:port of pilosa to run random queries on.")
+	fs.IntVar(&cfg.TreeDepth, "d", 4, "depth of random queries to generate.")
+	fs.IntVar(&cfg.QueryCount, "n", 100, "number of random queries to generate. Set to 0 for inifinite queries.")
+	fs.BoolVar(&cfg.Verbose, "v", false, "show queries as they are generated")
+}
+
+// call c.ValidateConfig() after myflags.Parse()
+func (c *RandomQueryConfig) ValidateConfig() error {
+	if c.TreeDepth < 1 {
+		return fmt.Errorf("-d depth must be 1 or greater; saw %v", c.TreeDepth)
+	}
+	if c.QueryCount < 0 {
+		return fmt.Errorf("-n count must be 0 or greater; saw %v", c.QueryCount)
+	}
+	return nil
+}
+
+var ProgramName = "random-query"
+
+func main() {
+
+	myflags := flag.NewFlagSet(ProgramName, flag.ExitOnError)
+	cfg := NewRandomQueryConfig()
+	cfg.DefineFlags(myflags)
+
+	err := myflags.Parse(os.Args[1:])
+	if err != nil {
+		fmt.Fprintf(os.Stderr, "\n%v\n", err.Error())
+		os.Exit(1)
+	}
+	err = cfg.ValidateConfig()
+	if err != nil {
+		fmt.Fprintf(os.Stderr, "%s error: %s\n", ProgramName, err)
+		os.Exit(1)
+	}
+
+	err = cfg.Run()
+
+	if err != nil {
+		fmt.Fprintf(os.Stderr, "error: %v\n", err)
+		os.Exit(1)
+	}
+}
+
+func (cfg *RandomQueryConfig) Run() (err error) {
+	remoteClient := nethttp.DefaultClient
+	cli, err := http.NewInternalClient(cfg.HostPort, remoteClient)
+	if err != nil {
+		return err
+	}
+	ctx := context.Background()
+	totalQ := 0
+	loops := 0
+	t0 := time.Now()
+
+NewSetup:
+	err = cfg.Setup(cli)
+	if err != nil {
+		return err
+	}
+
+	if len(cfg.IndexMap) == 0 {
+		return fmt.Errorf("no rows to query")
+	}
+
+	var indexes []string
+	for index := range cfg.IndexMap {
+		indexes = append(indexes, index)
+	}
+
+	for j := 0; ; j++ {
+		if cfg.QueryCount > 0 {
+			if j >= cfg.QueryCount {
+				break
+			}
+		} else {
+			// else keep doing queries forever...
+			if loops > 0 && loops%500 == 0 {
+				// ...but account for any new data arrived by getting
+				// the schema and rows again every so often.
+				loops++
+				goto NewSetup
+			}
+		}
+		if totalQ%100 == 0 {
+			dur := time.Since(t0)
+			if dur > 0 {
+				qps := 1e9 * float64(totalQ) / float64(dur)
+				AlwaysPrintf("totalQueries run: %v   elapsed: %v   qps: %0.02f", totalQ, dur, qps)
+			}
+		}
+
+		index := indexes[rand.Intn(len(indexes))]
+
+		pql, err := cfg.GenQuery(index)
+		panicOn(err)
+
+		if cfg.Verbose {
+			fmt.Printf("pql = '%v'\n", pql)
+		}
+
+		// Query node0.
+		res, err := cli.Query(ctx, index, &pilosa.QueryRequest{Index: index, Query: pql})
+		if err != nil {
+			AlwaysPrintf("QUERY FAILED! queries before this=%v; err = '%v', pql='%v'", loops, err, pql)
+			return err
+		}
+		if cfg.Verbose {
+			fmt.Printf("success on pql = '%v'; res='%v'\n", pql, res.Results[0])
+		}
+		totalQ++
+		loops++
+	}
+
+	return nil
+}
+
+type Features struct {
+	Slc []IndexFieldRow
+}
+
+func NewRandomQueryConfig() *RandomQueryConfig {
+	return &RandomQueryConfig{
+		IndexMap: make(map[string]*Features),
+	}
+}
+
+type IndexFieldRow struct {
+	Index    string
+	Field    string
+	RowID    uint64
+	RowKey   string
+	IsRowKey bool
+}
+
+// Run a RandomQuery takes a list of RowIDFeatures and ColumnKeyObjects
+// and spits back a PQL query
+//
+func (cfg *RandomQueryConfig) Setup(api API) (err error) {
+
+	ctx := context.Background()
+	cfg.Info, err = api.Schema(ctx)
+	if err != nil {
+		return err
+	}
+	for i, ii := range cfg.Info {
+		_ = i
+		for k, fld := range ii.Fields {
+			_ = k
+			if fld.Options.Type == "set" {
+				pql := fmt.Sprintf("Rows(%v)", fld.Name)
+
+				res, err := api.Query(ctx, ii.Name, &pilosa.QueryRequest{Index: ii.Name, Query: pql})
+				panicOn(err)
+				if cfg.Verbose {
+					fmt.Printf("success on pql = '%v'; res='%v'\n", pql, res.Results[0])
+				}
+				// if the option is set to use RowKeys, then must get the Keys instead of the Rows from the RowIdentifiers.
+				// e.g.
+				// success on pql = 'Rows(aba)'; res='&pilosa.RowIdentifiers{Rows:[]uint64(nil), Keys:[]string{"aba1", "aba2"}
+				// success on pql = 'Rows(f)'; res='pilosa.RowIdentifiers{Rows:[]uint64{0x1}, Keys:[]string(nil), field:"f"}'
+
+				switch x := res.Results[0].(type) {
+				case *pilosa.RowIdentifiers:
+					// internalClient gets this
+					cfg.AddResponse(ii.Name, fld.Name, x)
+				case pilosa.RowIdentifiers:
+					// test gets this
+					cfg.AddResponse(ii.Name, fld.Name, &x)
+				}
+			}
+		}
+	}
+
+	cfg.BitmapFunc = []string{"Union", "Intersect", "Xor", "Not", "Difference"}
+	seed := int64(42)
+	cfg.Rnd = rand.New(rand.NewSource(seed))
+
+	return nil
+}
+
+func (cfg *RandomQueryConfig) AddResponse(index, field string, x *pilosa.RowIdentifiers) {
+	for _, rowID := range x.Rows {
+		cfg.AddFeature(index, field, rowID, "", false)
+	}
+	for _, rowKey := range x.Keys {
+		cfg.AddFeature(index, field, 0, rowKey, true)
+	}
+}
+
+func (cfg *RandomQueryConfig) GenQuery(index string) (pql string, err error) {
+
+	tree := cfg.GenTree(index, cfg.TreeDepth)
+
+	if cfg.Verbose {
+		fmt.Printf("%v\n", tree.StringIndent(0))
+	}
+	pql = tree.ToPQL()
+
+	// avoid using too much bandwidth, just count the final bitmap.
+	pql = fmt.Sprintf("Count(%v)", pql)
+	return
+}
+
+type Tree struct {
+	Chd []*Tree
+
+	S string
+}
+
+func (tr *Tree) StringIndent(ind int) (s string) {
+	spc := strings.Repeat("    ", ind)
+	spc1 := strings.Repeat("    ", ind+1)
+	var chds []string
+	leaf := true
+	if len(tr.Chd) == 0 {
+		// leaf
+	} else {
+		leaf = false
+		for _, chd := range tr.Chd {
+			chds = append(chds, chd.StringIndent(ind+1))
+		}
+	}
+	if leaf {
+		s += fmt.Sprintf("%v %v\n", spc1, tr.S)
+	} else {
+		for i, c := range chds {
+			if i == 0 {
+				s += fmt.Sprintf("%v %v\n%v", spc, tr.S, c)
+			} else {
+				s += fmt.Sprintf("%v", c)
+			}
+		}
+	}
+	return
+}
+
+func (cfg *RandomQueryConfig) GenTree(index string, depth int) (tr *Tree) {
+	if depth == 0 {
+		slc := cfg.IndexMap[index].Slc
+		//vv("depth is 0, slc = '%#v'", slc)
+		r := cfg.Rnd.Intn(len(slc))
+		fea := slc[r]
+		if fea.IsRowKey {
+			return &Tree{S: fmt.Sprintf("Row(%v='%v')", fea.Field, fea.RowKey)}
+		}
+		return &Tree{S: fmt.Sprintf("Row(%v=%v)", fea.Field, fea.RowID)}
+	}
+
+	r := cfg.Rnd.Intn(len(cfg.BitmapFunc))
+	f := cfg.BitmapFunc[r]
+	tr = &Tree{S: f}
+	numChild := 2
+	switch f {
+	case "Union", "Intersect", "Xor":
+		numChild = cfg.Rnd.Intn(8) + 2
+	case "Not":
+		numChild = 1
+	case "Difference":
+		numChild = 2
+	}
+	for i := 0; i < numChild; i++ {
+		tr.Chd = append(tr.Chd, cfg.GenTree(index, depth-1))
+	}
+	return
+}
+
+func (tr *Tree) ToPQL() (s string) {
+
+	if len(tr.Chd) == 0 {
+		// leaf
+		return tr.S
+	}
+
+	var chds []string
+	for _, c := range tr.Chd {
+		chds = append(chds, c.ToPQL())
+	}
+	all := strings.Join(chds, ", ")
+	return fmt.Sprintf("%v(%v)", tr.S, all)
+}
+
+func (cfg *RandomQueryConfig) AddFeature(index, field string, rowID uint64, rowKey string, isRowKey bool) {
+
+	f, ok := cfg.IndexMap[index]
+	if !ok {
+		f = &Features{}
+		cfg.IndexMap[index] = f
+	}
+	f.Slc = append(f.Slc, IndexFieldRow{
+		Index:    index,
+		Field:    field,
+		RowID:    rowID,
+		RowKey:   rowKey,
+		IsRowKey: isRowKey,
+	})
+}

cmd/random-query/main_test.go

@@ -0,0 +1,165 @@
+// Copyright 2020 Pilosa Corp.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package main
+
+import (
+	"context"
+	"math/rand"
+	"strconv"
+	"testing"
+
+	"github.com/pilosa/pilosa/v2"
+	"github.com/pilosa/pilosa/v2/boltdb"
+	"github.com/pilosa/pilosa/v2/http"
+	"github.com/pilosa/pilosa/v2/server"
+	"github.com/pilosa/pilosa/v2/test"
+)
+
+func Test_RandomQuery(t *testing.T) {
+
+	cfg := NewRandomQueryConfig()
+
+	nNodes := 1
+	nReplicas := 1
+
+	name := t.Name()
+	var nodeid []string
+	for i := 0; i < nNodes; i++ {
+		// work around a bug in the test.MustRunCluster that corrupts
+		// the .topology file if we only join name with one "_" underscore.
+		nodeid = append(nodeid, name+"__"+strconv.Itoa(i))
+	}
+
+	c := test.MustRunCluster(t, nNodes,
+		[]server.CommandOption{
+			server.OptCommandServerOptions(
+				pilosa.OptServerNodeID(nodeid[0]),
+				pilosa.OptServerOpenTranslateStore(boltdb.OpenTranslateStore),
+				pilosa.OptServerOpenTranslateReader(http.GetOpenTranslateReaderFunc(nil)),
+				pilosa.OptServerReplicaN(nReplicas),
+			)},
+	)
+	defer c.Close()
+
+	var nodes []*test.Command
+	var dirs []string
+	for i := 0; i < nNodes; i++ {
+		nd := c.GetNode(i)
+		nodes = append(nodes, nd)
+		dirs = append(dirs, nd.Server.Holder().Path())
+	}
+	_ = dirs
+
+	ctx := context.Background()
+
+	indexes := []string{"rick", "morty"}
+	fieldName := []string{"f", "flying_car"}
+	idx := make([]*pilosa.Index, len(indexes))
+	field := make([]*pilosa.Field, len(indexes))
+
+	var err error
+
+	for i := range indexes {
+
+		idx[i], err = nodes[0].API.CreateIndex(ctx, indexes[i], pilosa.IndexOptions{Keys: true, TrackExistence: true})
+		if err != nil {
+			t.Fatalf("creating index: %v", err)
+		}
+		if idx[i].CreatedAt() == 0 {
+			t.Fatal("index createdAt is empty")
+		}
+
+		field[i], err = nodes[0].API.CreateField(ctx, indexes[i], fieldName[i], pilosa.OptFieldTypeSet(pilosa.DefaultCacheType, 100))
+		if err != nil {
+			t.Fatalf("creating field: %v", err)
+		}
+		if field[i].CreatedAt() == 0 {
+			t.Fatal("field createdAt is empty")
+		}
+	}
+
+	timestamp := int64(0)
+
+	for i := range indexes {
+
+		// Generate some keyed records.
+		rowIDs := []uint64{}
+		timestamps := []int64{}
+		N := 10
+		for j := 1; j <= N; j++ {
+			rowIDs = append(rowIDs, uint64(j))
+			timestamps = append(timestamps, timestamp)
+		}
+
+		var colKeys []string
+		switch i {
+		case 0:
+			// Keys are sharded so ordering is not guaranteed.
+			colKeys = []string{"col10", "col8", "col9", "col6", "col7", "col4", "col5", "col2", "col3", "col1"}
+			colKeys = colKeys[:N]
+		case 1:
+			colKeys = []string{"col11", "col12"}
+			N = len(colKeys)
+			rowIDs = rowIDs[:N]
+			timestamps = timestamps[:N]
+		}
+
+		// Import data with keys to the coordinator (node0) and verify that it gets
+		// translated and forwarded to the owner of shard 0 (node1; because of offsetModHasher)
+		req := &pilosa.ImportRequest{
+			Index:          indexes[i],
+			IndexCreatedAt: idx[i].CreatedAt(),
+			Field:          fieldName[i],
+			FieldCreatedAt: field[i].CreatedAt(),
+
+			// even though this says Shard: 0, that won't matter. The column keys
+			// get hashed and that decides the actual shard.
+			Shard:      0,
+			RowIDs:     rowIDs,
+			ColumnKeys: colKeys,
+			Timestamps: timestamps,
+		}
+		//vv("rowIDs = '%#v'", rowIDs)
+		//vv("colKeys = '%#v'", colKeys)
+
+		qcx := nodes[0].API.Txf().NewQcx()
+
+		if err := nodes[0].API.Import(ctx, qcx, req); err != nil {
+			t.Fatal(err)
+		}
+		panicOn(qcx.Finish())
+		//qcx.Reset()
+	}
+	// end of setup.
+
+	panicOn(cfg.Setup(wrapApiToInternalClient(nodes[0].API)))
+
+	for j := 0; j < 4; j++ {
+		index := indexes[rand.Intn(len(indexes))]
+
+		pql, err := cfg.GenQuery(index)
+		panicOn(err)
+
+		//vv("pql = '%v'", pql)
+
+		// Query node0.
+		res, err := nodes[0].API.Query(ctx, &pilosa.QueryRequest{Index: index, Query: pql})
+		if err != nil {
+			t.Fatal(err)
+		}
+		_ = res
+		//vv("success on pql = '%v'; res='%v'", pql, res.Results[0])
+	}
+}

cmd/random-query/vprint.go

@@ -0,0 +1,177 @@
+// home: https://github.com/glycerine/vprint
+// Copyright 2019 Jason E. Aten, Ph.D. All rights reserved.
+// License: MIT
+//
+// MIT License
+//
+// Permission is hereby granted, free of charge, to any person obtaining a copy
+// of this software and associated documentation files (the "Software"), to deal
+// in the Software without restriction, including without limitation the rights
+// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+// copies of the Software, and to permit persons to whom the Software is
+// furnished to do so, subject to the following conditions:
+//
+// The above copyright notice and this permission notice shall be included in all
+// copies or substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+// SOFTWARE.
+
+package main
+
+import (
+	"fmt"
+	"io"
+	"os"
+	"path"
+	"runtime"
+	"runtime/debug"
+	"sync"
+	"time"
+)
+
+const RFC3339MsecTz0 = "2006-01-02T15:04:05.000Z07:00"
+const RFC3339UsecTz0 = "2006-01-02T15:04:05.000000Z07:00"
+
+// for tons of debug output
+var VerboseVerbose bool = false
+
+// convience functions for . import
+var pp = PP
+var vv = VV
+
+var panicOn = PanicOn
+
+func init() {
+	// keeper linter happy
+	_ = pp
+	_ = vv
+}
+
+func PanicOn(err error) {
+	if err != nil {
+		panic(err)
+	}
+}
+
+func PP(format string, a ...interface{}) {
+	if VerboseVerbose {
+		TSPrintf(format, a...)
+	}
+}
+
+func VV(format string, a ...interface{}) {
+	TSPrintf(format, a...)
+}
+
+func AlwaysPrintf(format string, a ...interface{}) {
+	TSPrintf(format, a...)
+}
+
+var tsPrintfMut sync.Mutex
+
+// time-stamped printf
+func TSPrintf(format string, a ...interface{}) {
+	tsPrintfMut.Lock()
+	Printf("# %s %s ", FileLine(3), ts())
+	Printf(format+"\n", a...)
+	tsPrintfMut.Unlock()
+}
+
+// get timestamp for logging purposes
+func ts() string {
+	return time.Now().Format(RFC3339UsecTz0)
+}
+
+// so we can multi write easily, use our own printf
+var OurStdout io.Writer = os.Stdout
+
+// Printf formats according to a format specifier and writes to standard output.
+// It returns the number of bytes written and any write error encountered.
+func Printf(format string, a ...interface{}) (n int, err error) {
+	return fmt.Fprintf(OurStdout, format, a...)
+}
+
+func FileLine(depth int) string {
+	_, fileName, fileLine, ok := runtime.Caller(depth)
+	var s string
+	if ok {
+		s = fmt.Sprintf("%s:%d", path.Base(fileName), fileLine)
+	} else {
+		s = ""
+	}
+	return s
+}
+
+func stack() string {
+	return string(debug.Stack())
+}
+
+func FileExists(name string) bool {
+	fi, err := os.Stat(name)
+	if err != nil {
+		return false
+	}
+	if fi.IsDir() {
+		return false
+	}
+	return true
+}
+
+func DirExists(name string) bool {
+	fi, err := os.Stat(name)
+	if err != nil {
+		return false
+	}
+	if fi.IsDir() {
+		return true
+	}
+	return false
+}
+
+func FileSize(name string) int64 {
+	fi, err := os.Stat(name)
+	if err != nil {
+		return 0
+	}
+	return fi.Size()
+}
+
+// Caller returns the name of the calling function.
+func Caller(upStack int) string {
+	// elide ourself and runtime.Callers
+	target := upStack + 2
+
+	pc := make([]uintptr, target+2)
+	n := runtime.Callers(0, pc)
+
+	f := runtime.Frame{Function: "unknown"}
+	if n > 0 {
+		frames := runtime.CallersFrames(pc[:n])
+		for i := 0; i <= target; i++ {
+			contender, more := frames.Next()
+			if i == target {
+				f = contender
+			}
+			if !more {
+				break
+			}
+		}
+	}
+	return f.Function
+}
+
+// happy linter:
+var _ = DirExists
+var _ = FileExists
+var _ = Caller
+var _ = stack
+var _ = RFC3339MsecTz0
+var _ = RFC3339UsecTz0
+var _ = AlwaysPrintf
+var _ = FileSize

license.exceptions

@@ -22,4 +22,4 @@
 ./proto/vdsm/vdsm.proto
 ./proto/vdsm/vdsm.pb.go
 ./cmd/pilosa-fsck/vprint.go
-
+./cmd/random-query/vprint.go