Merge branch 'master' into 30trial

api.go

@@ -816,48 +816,88 @@ func (api *API) Node() *Node {
 
 // NodeUsage represents all usage measurements for one node.
 type NodeUsage struct {
-	Disk DiskUsage `json:"bytesOnDisk"`
+	Disk   DiskUsage   `json:"diskUsage"`
+	Memory MemoryUsage `json:"memoryUsage"`
 }
 
 // DiskUsage represents the storage space used on disk by one node.
 type DiskUsage struct {
-	Capacity uint64           `json:"capacity,omitempty"`
-	TotalUse int64            `json:"totalInUse"`
-	Indexes  map[string]int64 `json:"indexes"`
+	Capacity   uint64                `json:"capacity,omitempty"`
+	TotalUse   uint64                `json:"totalInUse"`
+	IndexUsage map[string]IndexUsage `json:"indexes"`
 }
 
-// Usage gets the disk usage per index, in a map[nodeID]NodeUsage
+// IndexUsage represents the storage space used on disk by one index, on one node.
+type IndexUsage struct {
+	Total          uint64                `json:"total"`
+	IndexKeys      uint64                `json:"indexKeys"`
+	FieldKeysTotal uint64                `json:"fieldKeysTotal"`
+	Fragments      uint64                `json:"fragments"`
+	Metadata       uint64                `json:"metadata"`
+	Fields         map[string]FieldUsage `json:"fields"`
+}
+
+// FieldUsage represents the storage space used on disk by one field, on one node
+type FieldUsage struct {
+	Total     uint64 `json:"total"`
+	Fragments uint64 `json:"fragments"`
+	Keys      uint64 `json:"keys"`
+	Metadata  uint64 `json:"metadata"`
+}
+
+// MemoryUsage represents the memory used by one node.
+type MemoryUsage struct {
+	Capacity uint64 `json:"capacity"`
+	TotalUse uint64 `json:"totalInUse"`
+}
+
+// Usage gets the resource usage per index, in a map[nodeID]NodeUsage
 func (api *API) Usage(ctx context.Context, remote bool) (map[string]NodeUsage, error) {
 	span, _ := tracing.StartSpanFromContext(ctx, "API.Usage")
 	defer span.Finish()
 
 	nodeUsages := make(map[string]NodeUsage)
 
-	indexSizes, err := api.holder.Txf().IndexSizes()
+	indexDetails, nodeMetadataBytes, err := api.holder.Txf().IndexUsageDetails()
 	if err != nil {
-		return nil, errors.Wrap(err, "getting index usage")
+		return nil, errors.Wrap(err, "getting node usage")
 	}
-	var totalSize int64
-	for _, s := range indexSizes {
-		totalSize += s
+	totalSize := nodeMetadataBytes
+	for _, s := range indexDetails {
+		totalSize += s.Total
 	}
 
-	capacity, err := api.server.systemInfo.DiskCapacity(api.holder.path)
+	// NOTE: these errors are ignored in api.Info(), but checked here
+	si := api.server.systemInfo
+	diskCapacity, err := si.DiskCapacity(api.holder.path)
 	if err != nil {
 		api.server.logger.Printf("couldn't read disk capacity: %s", err)
 	}
 
+	memoryCapacity, err := si.MemTotal()
+	if err != nil {
+		api.server.logger.Printf("couldn't read memory capacity: %s", err)
+	}
+	memoryUse, err := si.MemUsed()
+	if err != nil {
+		api.server.logger.Printf("couldn't read memory usage: %s", err)
+	}
+
 	// Insert into result.
 	nodeUsage := NodeUsage{
 		Disk: DiskUsage{
-			Capacity: capacity,
-			TotalUse: totalSize,
-			Indexes:  indexSizes,
+			Capacity:   diskCapacity,
+			TotalUse:   totalSize,
+			IndexUsage: indexDetails,
+		},
+		Memory: MemoryUsage{
+			Capacity: memoryCapacity,
+			TotalUse: memoryUse,
 		},
 	}
 	nodeUsages[api.server.nodeID] = nodeUsage
 
-	// Collect size on disk from remote nodes
+	// Collect usage from remote nodes
 	if !remote {
 		nodes := api.cluster.Nodes()
 		for _, node := range nodes {

bluegreentx.go

@@ -663,6 +663,10 @@ func (c *blueGreenTx) ContainerIterator(index, field, view string, shard uint64,
 	return bgi, bfound, errB
 }
 
+func (tx *blueGreenTx) GetFieldSizeBytes(index, field string) (uint64, error) {
+	return 0, nil
+}
+
 func NewBlueGreenIterator(tx *blueGreenTx, ait, bit roaring.ContainerIterator) *blueGreenIterator {
 	return &blueGreenIterator{
 		tx:  tx,

bolt.go

@@ -759,6 +759,10 @@ func (tx *BoltTx) ContainerIterator(index, field, view string, shard uint64, fir
 	return bi, bytes.Equal(bi.lastKey, needle), nil
 }
 
+func (tx *BoltTx) GetFieldSizeBytes(index, field string) (uint64, error) {
+	return 0, nil
+}
+
 // BoltIterator is the iterator returned from a BoltTx.ContainerIterator() call.
 // It implements the roaring.ContainerIterator interface.
 type BoltIterator struct {

catcher.go

@@ -317,3 +317,7 @@ func (c *catcherTx) ApplyFilter(index, field, view string, shard uint64, ckey ui
 func (c *catcherTx) GetSortedFieldViewList(idx *Index, shard uint64) (fvs []txkey.FieldView, err error) {
 	return c.b.GetSortedFieldViewList(idx, shard)
 }
+
+func (tx *catcherTx) GetFieldSizeBytes(index, field string) (uint64, error) {
+	return 0, nil
+}

executor.go

@@ -143,7 +143,6 @@ func (e *executor) Close() error {
 
 // Execute executes a PQL query.
 func (e *executor) Execute(ctx context.Context, index string, q *pql.Query, shards []uint64, opt *execOptions) (QueryResponse, error) {
-
 	span, ctx := tracing.StartSpanFromContext(ctx, "Executor.Execute")
 	span.LogKV("pql", q.String())
 	defer span.Finish()
@@ -556,11 +555,22 @@ func (e *executor) execute(ctx context.Context, qcx *Qcx, index string, q *pql.Q
 		// about the positive values, because only positive values
 		// are valid column IDs. So we don't actually eat top-level
 		// pre calls.
-		err := e.handlePreCallChildren(ctx, qcx, index, call, shards, opt)
-		if err != nil {
-			return nil, err
+		if call.Name == "Count" {
+			// Handle count specially, skipping the level directly underneath it.
+			for _, child := range call.Children {
+				err := e.handlePreCallChildren(ctx, qcx, index, child, shards, opt)
+				if err != nil {
+					return nil, err
+				}
+			}
+		} else {
+			err := e.handlePreCallChildren(ctx, qcx, index, call, shards, opt)
+			if err != nil {
+				return nil, err
+			}
 		}
 		var v interface{}
+		var err error
 		// Top-level calls don't need to precompute cross-index things,
 		// because we can just pick whatever index we want, but we
 		// still need to handle them. Since everything else was
@@ -1505,7 +1515,18 @@ func executeDistinctShardSet(ctx context.Context, qcx *Qcx, idx *Index, fieldNam
 	fragData, _, err := tx.ContainerIterator(index, fieldName, "standard", shard, 0)
 	switch errors.Cause(err) {
 	case ViewNotFound, FragmentNotFound:
-		return nil, nil
+		// It may seem reasonable to return `nil` here in the case where the
+		// fragment for this shard does not exist. The problem with doing that
+		// is that if this operation is being performed on a remote node, then
+		// this result is going to get serialized as a QueryResponse and sent
+		// back to the original, non-remote node. When this happens, the
+		// encodeRow/decodeRow logic replaces `nil` with an empty Row. An empty
+		// Row will cause problems during the union step of the reduce phase if
+		// it is the "left" side of the union, because then the resulting Row
+		// after the union will have blank Index and Field values. Here, we
+		// ensure that we send a non-nil Row with valid Index and Field values
+		// so that the union step doesn't cause problems.
+		return &Row{Index: index, Field: fieldName}, nil
 	case nil:
 	default:
 		return nil, errors.Wrap(err, "getting fragment data")
@@ -4618,28 +4639,21 @@ func (e *executor) executeCount(ctx context.Context, qcx *Qcx, index string, c *
 
 	child := c.Children[0]
 
-	// If the child is precomputed, we'll bypass mapreduce, ignore
-	// shards, and just count the number of bits.
-	if child.Name == "Precomputed" {
-		count := uint64(0)
-		for _, irow := range child.Precomputed {
-			switch row := irow.(type) {
-			case *Row:
-				for _, seg := range row.segments {
-					count += seg.n
-				}
-			case SignedRow:
-				for _, seg := range row.Pos.segments {
-					count += seg.n
-				}
-				for _, seg := range row.Neg.segments {
-					count += seg.n
-				}
-			default:
-				return 0, errors.Errorf("unexpected precomputed value type inside count: %+v", row)
-			}
+	// If the child is distinct/similar, execute it directly here and count the result.
+	if child.Type == pql.PrecallGlobal {
+		result, err := e.executeCall(ctx, qcx, index, child, shards, opt)
+		if err != nil {
+			return 0, err
+		}
+
+		switch row := result.(type) {
+		case *Row:
+			return row.Count(), nil
+		case SignedRow:
+			return row.Pos.Count() + row.Neg.Count(), nil
+		default:
+			return 0, errors.Errorf("cannot count result of type %T from call %q", row, child.String())
 		}
-		return count, nil
 	}
 
 	// Execute calls in bulk on each remote node and merge.
@@ -6562,6 +6576,9 @@ func (e *executor) translateResult(ctx context.Context, index string, idx *Index
 
 			if field.Keys() {
 				rslt := result.Pos
+				if rslt == nil {
+					return &SignedRow{Pos: &Row{}}, nil
+				}
 				other := &Row{Attrs: rslt.Attrs}
 				for _, segment := range rslt.Segments() {
 					keys, err := e.Cluster.translateIndexIDs(context.Background(), field.ForeignIndex(), segment.Columns())

executor_test.go

@@ -5397,6 +5397,19 @@ func TestExecutor_ForeignIndex(t *testing.T) {
 		pilosa.OptFieldKeys(),
 	)
 
+	// stepchild/other field needs to have usesKeys=true
+	crashSchemaJson := `{"indexes": [{"name": "stepparent","createdAt": 1611247966371721700,"options": {"keys": true,"trackExistence": true},"shardWidth": 1048576},{"name": "stepchild","createdAt": 1611247953796662800,"options": {"keys": true,"trackExistence": true},"shardWidth": 1048576,"fields": [{"name": "parent_id","createdAt": 1611247953797265700,"options": {"type": "int","base": 0,"bitDepth": 28,"min": -9223372036854776000,"max": 9223372036854776000,"keys": false,"foreignIndex": "stepparent"}},{"name": "other","createdAt": 1611247953796814000,"options": {"type": "int","base": 0,"bitDepth": 17,"min": -9223372036854776000,"max": 9223372036854776000,"keys": true,"foreignIndex": ""}}]}]}`
+
+	crashSchema := &pilosa.Schema{}
+	err := json.Unmarshal([]byte(crashSchemaJson), &crashSchema)
+	if err != nil {
+		t.Fatalf("json unmarshall: %v", err)
+	}
+	err = c.GetNode(0).API.ApplySchema(context.Background(), crashSchema, false)
+	if err != nil {
+		t.Fatalf("applying JSON schema: %v", err)
+	}
+
 	// Populate parent data.
 	c.Query(t, "parent", fmt.Sprintf(`
 			Set("one", general=1)
@@ -5442,6 +5455,12 @@ func TestExecutor_ForeignIndex(t *testing.T) {
 		t.Fatalf("unexpected keys: %v", row.Keys)
 	}
 
+	crash := c.Query(t, "stepchild", `Distinct(Row(parent_id=3), field=other)`).Results[0].(pilosa.SignedRow)
+	if !sameStringSlice(crash.Pos.Keys, []string{}) {
+		// empty result; error condition does not require data
+		t.Fatalf("unexpected columns: %v", crash.Pos.Keys)
+	}
+
 	eq := c.Query(t, "child", `Row(parent_id=="one")`).Results[0].(*pilosa.Row)
 	if !reflect.DeepEqual(eq.Columns(), []uint64{1, ShardWidth}) {
 		t.Fatalf("unexpected columns: %v", eq.Columns())
@@ -6818,6 +6837,8 @@ func TestMissingKeyRegression(t *testing.T) {
 func TestVariousQueries(t *testing.T) {
 	for _, clusterSize := range []int{1, 3, 4, 7} {
 		t.Run(fmt.Sprintf("%d-node", clusterSize), func(t *testing.T) {
+			t.Parallel()
+
 			variousQueries(t, clusterSize)
 		})
 	}
@@ -6897,8 +6918,7 @@ func variousQueries(t *testing.T, clusterSize int) {
 		{Val: 0, Key: "userE"},
 	})
 
-	// Create and populate "affinity" int field with negative, positive, zero and null values.
-
+	// Create and populate "net_worth" int field with positive values.
 	c.CreateField(t, "users", pilosa.IndexOptions{Keys: true, TrackExistence: true}, "net_worth", pilosa.OptFieldTypeInt(-100000000, 100000000))
 	c.ImportIntKey(t, "users", "net_worth", []test.IntKey{
 		{Val: 1, Key: "userA"},
@@ -7032,6 +7052,15 @@ toronto,2,11
 			},
 			csvVerifier: "-10\n-5\n0\n5\n10\n",
 		},
+		{
+			query: "Count(Distinct(field=affinity))",
+			qrVerifier: func(t *testing.T, resp pilosa.QueryResponse) {
+				if resp.Results[0].(uint64) != 5 {
+					t.Errorf("wrong number of values: %+v", resp.Results[0])
+				}
+			},
+			csvVerifier: "5\n",
+		},
 		{
 			query: "Distinct(Row(affinity>=0),field=affinity)",
 			qrVerifier: func(t *testing.T, resp pilosa.QueryResponse) {

fragment.go

@@ -2651,7 +2651,13 @@ func (f *fragment) importValueSmallWrite(tx Tx, columnIDs []uint64, values []int
 		}
 		return nil
 	}(); err != nil {
-		_ = f.openStorage(true)
+		errOpenStorage := f.openStorage(true)
+		if errOpenStorage != nil {
+			f.Logger.Printf("failed to import data into fragment: %v", err)
+			f.Logger.Printf("recovery with openStorage failed for fragment: %v", errOpenStorage)
+			f.Logger.Debugf("%s", debug.Stack())
+			os.Exit(1)
+		}
 		return err
 	}
 	rowSet := make(map[uint64]struct{}, bitDepth+1)
@@ -2705,7 +2711,13 @@ func (f *fragment) importValue(tx Tx, columnIDs []uint64, values []int64, bitDep
 		// Flush changes in bulk back to the transaction.
 		return txb.Flush()
 	}(); err != nil {
-		_ = f.openStorage(true)
+		errOpenStorage := f.openStorage(true)
+		if errOpenStorage != nil {
+			f.Logger.Printf("failed to import data into fragment: %v", err)
+			f.Logger.Printf("recovery with openStorage failed for fragment: %v", errOpenStorage)
+			f.Logger.Debugf("%s", debug.Stack())
+			os.Exit(1)
+		}
 		return err
 	}
 	// Keep stats accurate. We don't call incrementOpN here because it may

holder.go

@@ -1277,10 +1277,8 @@ func (h *Holder) LoadNodeID() (string, error) {
 
 // Log startup time and version to $DATA_DIR/.startup.log
 func (h *Holder) logStartup() error {
-	time, err := time.Now().MarshalText()
-	if err != nil {
-		return errors.Wrap(err, "creating timestamp")
-	}
+	RFC3339NanoFixedWidth := "2006-01-02T15:04:05.000000 07:00"
+	time := time.Now().Format(RFC3339NanoFixedWidth)
 	logLine := fmt.Sprintf("%s\t%s\n", time, Version)
 
 	f, err := os.OpenFile(h.path+"/.startup.log", os.O_APPEND|os.O_WRONLY|os.O_CREATE, 0600)

lattice

@@ -1 +1 @@
-Subproject commit 28c2313ecfcd7e083d42d4e409483e968b4c421b
+Subproject commit fa773628a276e2590785a87fbc236c7e88ea6284

pql/ast.go

@@ -274,14 +274,17 @@ type callStackElem struct {
 type CallType byte
 
 const (
-	// Normal calls can be executed per shard.
+	// PrecallNone calls can be executed per shard.
 	PrecallNone = CallType(iota)
-	// PreCallGlobal indicates a call which must be run globally *before*
+
+	// PrecallGlobal indicates a call which must be run globally *before*
 	// distributing the call to other shards. Example: A Distinct query,
 	// where every shard could potentially produce results for any shard,
 	// so you have to produce the results up front.
+	// These are processed directly when inside of a count operation.
 	PrecallGlobal
-	// PreCallPerNode indicates a call which needs to be run per-shard
+
+	// PrecallPerNode indicates a call which needs to be run per-shard
 	// in a way that lets it be done on each shard, but where it should
 	// be done prior to spawning per-shard goroutines. Example:
 	// A cross-index query, where each local shard may or may not need

rbf.go

@@ -435,6 +435,10 @@ func (tx *RBFTx) GetSortedFieldViewList(idx *Index, shard uint64) (fvs []txkey.F
 	return tx.tx.GetSortedFieldViewList()
 }
 
+func (tx *RBFTx) GetFieldSizeBytes(index, field string) (uint64, error) {
+	return tx.tx.GetSizeBytesWithPrefix(string(txkey.FieldPrefix(index, field)))
+}
+
 // rbfName returns a NULL-separated key used for identifying bitmap maps in RBF.
 func rbfName(index, field, view string, shard uint64) string {
 	return string(txkey.Prefix(index, field, view, shard))

rbf/tx.go

@@ -839,6 +839,33 @@ func (tx *Tx) inusePageSet() (map[uint32]struct{}, error) {
 	return m, nil
 }
 
+// GetSizeBytesWithPrefix returns the size of bitmaps with a given key prefix.
+func (tx *Tx) GetSizeBytesWithPrefix(prefix string) (n uint64, err error) {
+	records, err := tx.RootRecords()
+	if err != nil {
+		return 0, err
+	}
+
+	// Loop over each bitmap in the database.
+	for itr := records.Iterator(); !itr.Done(); {
+		name, pgno := itr.Next()
+
+		// Skip over any bitmaps that don't have a matching prefix.
+		if !strings.HasPrefix(name.(string), prefix) {
+			continue
+		}
+
+		// Traverse the bitmap's b-tree and count the bytes for each page.
+		if err := tx.walkTree(pgno.(uint32), 0, func(pgno, parent, typ uint32) error {
+			n += PageSize
+			return nil
+		}); err != nil {
+			return 0, err
+		}
+	}
+	return n, nil
+}
+
 // walkTree recursively iterates over a page and all its children.
 func (tx *Tx) walkTree(pgno, parent uint32, fn func(pgno, parent, typ uint32) error) error {
 	// Read page and iterate over children.
@@ -964,6 +991,7 @@ func (tx *Tx) deallocateTree(pgno uint32) error {
 
 func (tx *Tx) readPage(pgno uint32) (_ []byte, isHeap bool, err error) {
 	// Meta page is always cached on the transaction.
+	//fmt.Printf("readPage %d\n", pgno)
 	if pgno == 0 {
 		return tx.meta[:], false, nil
 	}
@@ -971,7 +999,7 @@ func (tx *Tx) readPage(pgno uint32) (_ []byte, isHeap bool, err error) {
 	// Verify page number requested is within current size of database.
 	pageN := readMetaPageN(tx.meta[:])
 	if pgno > pageN {
-		return nil, false, fmt.Errorf("rbf: page read out of bounds: pgno=%d max=%d", pgno, pageN)
+		return nil, false, fmt.Errorf("rbf: page read out of bounds: pgno=%d max=%d", pgno, pageN-1)
 	}
 
 	// Check if page has been updated in this tx.

rrtx.go

@@ -589,6 +589,10 @@ func (tx *RoaringTx) GetSortedFieldViewList(idx *Index, shard uint64) (fvs []txk
 	return
 }
 
+func (tx *RoaringTx) GetFieldSizeBytes(index, field string) (uint64, error) {
+	return 0, nil
+}
+
 //////// registrar and wrapper machinery
 
 // roaringRegistrar mirrors the machinery expected

scripts/bench_read.sh

@@ -21,7 +21,7 @@ SHA=$(git -C $PILOSA_SRC rev-parse HEAD)
 # Format current date.
 DATE=$(date '+%Y%m%d')
 
-for TYPE in row row-bsi row-range count intersect union difference xor groupby topk 
+for TYPE in row row-bsi row-range count count-keyed intersect union difference xor groupby topk 
 do
 	WORKFLOW_PATH="${BASH_SOURCE%/*}/etc/gloat/query.${TYPE}.yml"
 	WORKFLOW_NAME="$(gloat workflow name $WORKFLOW_PATH)"

scripts/bench_write.sh

@@ -21,19 +21,22 @@ SHA=$(git -C $PILOSA_SRC rev-parse HEAD)
 # Format current date.
 DATE=$(date '+%Y%m%d')
 
-WORKFLOW_PATH="${BASH_SOURCE%/*}/etc/gloat/gh.1m.yml"
-WORKFLOW_NAME="$(gloat workflow name $WORKFLOW_PATH)"
-TITLE="RBF vs Roaring, $WORKFLOW_NAME, $DATE ($SHA)"
+for FILENAME in gh.1m.yml gh.issues.keyed.yml gh.issues.unkeyed.yml 
+do
+	WORKFLOW_PATH="${BASH_SOURCE%/*}/etc/gloat/${FILENAME}"
+	WORKFLOW_NAME="$(gloat workflow name $WORKFLOW_PATH)"
+	TITLE="RBF vs Roaring, $WORKFLOW_NAME, $DATE ($SHA)"
 
-# Execute RBF/Roaring benchmark.
-RBF_PATH=gloat/data/1m/rbf/${DATE}.tar.gz
-TXSRC=rbf gloat run -v -o $RBF_PATH $WORKFLOW_PATH
+	# Execute RBF/Roaring benchmark.
+	RBF_PATH=gloat/data/1m/rbf/${DATE}.tar.gz
+	TXSRC=rbf gloat run -v -o $RBF_PATH $WORKFLOW_PATH
 
-ROARING_PATH=gloat/data/1m/roaring/${DATE}.tar.gz
-TXSRC=roaring gloat run -v -o $ROARING_PATH $WORKFLOW_PATH
+	ROARING_PATH=gloat/data/1m/roaring/${DATE}.tar.gz
+	TXSRC=roaring gloat run -v -o $ROARING_PATH $WORKFLOW_PATH
 
-# Generate graph from results.
-gloat graph -layout 2,5 -size 5120,820 -title "$TITLE" -name utime,stime,heap_alloc,heap_inuse,heap_objects,num_gc,rchar,wchar,syscr,syscw -series rbf,roaring -o /tmp/output.png $RBF_PATH $ROARING_PATH
+	# Generate graph from results.
+	gloat graph -layout 2,5 -size 5120,820 -title "$TITLE" -name utime,stime,heap_alloc,heap_inuse,heap_objects,num_gc,rchar,wchar,syscr,syscw -series rbf,roaring -o /tmp/output.png $RBF_PATH $ROARING_PATH
 
-# Post graph to Slack with SHA.
-curl -F file=@/tmp/output.png -F channels=C01HBFKRLGH -F "initial_comment=$TITLE" -H "Authorization: Bearer $SLACK_OAUTH_TOKEN" https://slack.com/api/files.upload
+	# Post graph to Slack with SHA.
+	curl -F file=@/tmp/output.png -F channels=C01HBFKRLGH -F "initial_comment=$TITLE" -H "Authorization: Bearer $SLACK_OAUTH_TOKEN" https://slack.com/api/files.upload
+done

scripts/etc/gloat/gh.issues.keyed.yml

@@ -0,0 +1,9 @@
+name: "GitHub Issues Import Load Testing (1 month, keyed)"
+
+main: "pilosa server --data-dir ${TMPDIR} --txsrc ${TXSRC}"
+load: "molecula-consumer-github -i issues -r url --record-type issue --batch-size=100000 --start-time 2020-01-01T00:00:00Z --end-time 2020-01-13T23:00:00Z --cache-dir ~/.githubarchive" 
+
+health_url: "http://localhost:10101/status"
+vars_urls:
+  - http://localhost:10101/debug/vars
+  - http://localhost:7070/debug/vars

scripts/etc/gloat/gh.issues.unkeyed.yml

@@ -0,0 +1,9 @@
+name: "GitHub Issues Import Load Testing (1 month, unkeyed)"
+
+main: "pilosa server --data-dir ${TMPDIR} --txsrc ${TXSRC}"
+load: "molecula-consumer-github -i issues -d id --record-type issue --batch-size=100000 --start-time 2020-01-01T00:00:00Z --end-time 2020-01-13T23:00:00Z --cache-dir ~/.githubarchive" 
+
+health_url: "http://localhost:10101/status"
+vars_urls:
+  - http://localhost:10101/debug/vars
+  - http://localhost:7070/debug/vars

scripts/etc/gloat/query.count.keyed.yml

@@ -0,0 +1,11 @@
+name: "Count() Load Testing w/ Keys"
+
+main: "pilosa server --data-dir ~/pilosa.query.keyed.${TXSRC} --txsrc ${TXSRC}"
+load: "pilosa-bench -type count -rate 100 -n 3000" 
+
+health_url:    "http://localhost:10101/status"
+health_regexp: "NORMAL"
+
+vars_urls:
+  - http://localhost:10101/debug/vars
+  - http://localhost:7070/debug/vars

scripts/populate_query_db.keyed.sh

@@ -0,0 +1,28 @@
+#!/bin/bash
+set -e
+
+# This script generates data query load testing to be run against.
+#
+# Environment variables:
+#   - TXSRC: Transaction store type ("roaring", "rbf")
+#   - CACHEDIR: Path to local GitHub Archive data, if available.
+
+# Require environment variables.
+: "${TXSRC:?Must set TXSRC environment variable}"
+: "${GHCACHEDIR:''}"
+
+echo "Starting pilosa"
+pilosa server --data-dir ~/pilosa.query.keyed.${TXSRC} --txsrc ${TXSRC} & pid_pilosa=$!
+sleep 5
+
+echo ""
+echo "Importing GitHub Archive"
+molecula-consumer-github -i issues -r url --record-type issue --batch-size=100000 \
+	--start-time 2020-01-01T00:00:00Z --end-time 2020-01-31T23:00:00Z \
+	--cache-dir "$GHCACHEDIR"
+
+echo ""
+echo "Import complete, shutting down pilosa"
+
+sleep 5
+kill $pid_pilosa

server/grpc.go

@@ -20,6 +20,7 @@ import (
 	"fmt"
 	"net"
 	"net/http"
+	"strconv"
 	"strings"
 	"sync"
 	"time"
@@ -34,6 +35,7 @@ import (
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/credentials"
+	"google.golang.org/grpc/metadata"
 	"google.golang.org/grpc/reflection"
 	"google.golang.org/grpc/status"
 )
@@ -146,6 +148,13 @@ func (h *GRPCHandler) QuerySQL(req *pb.QuerySQLRequest, stream pb.Pilosa_QuerySQ
 		return err
 	}
 
+	err = stream.SendHeader(metadata.New(map[string]string{
+		"duration": strconv.Itoa(int(duration)),
+	}))
+	if err != nil {
+		return errors.Wrap(err, "sending header")
+	}
+
 	err = newDurationRowser(results, duration).ToRows(stream.Send)
 	if err != nil {
 		return errors.Wrap(err, "streaming result")
@@ -183,7 +192,15 @@ func (h *GRPCHandler) QuerySQLUnary(ctx context.Context, req *pb.QuerySQLRequest
 	if err != nil {
 		return nil, err
 	}
-	table.Duration = int64(time.Since(start))
+	duration := time.Since(start)
+	table.Duration = int64(duration)
+	err = grpc.SendHeader(ctx, metadata.New(map[string]string{
+		"duration": strconv.Itoa(int(duration)),
+	}))
+	if err != nil {
+		return nil, errors.Wrap(err, "sending header")
+	}
+
 	return table, nil
 }
 
@@ -197,6 +214,7 @@ func (h *GRPCHandler) QueryPQL(req *pb.QueryPQLRequest, stream pb.Pilosa_QueryPQ
 	t := time.Now()
 	resp, err := h.api.Query(stream.Context(), &query)
 	durQuery := time.Since(t)
+
 	// TODO: what about resp.CollumnAttrSets?
 	if err != nil {
 		return errToStatusError(err)
@@ -215,6 +233,13 @@ func (h *GRPCHandler) QueryPQL(req *pb.QueryPQLRequest, stream pb.Pilosa_QueryPQ
 		return errors.Wrap(err, "wrapping as type ToRowser")
 	}
 
+	err = stream.SendHeader(metadata.New(map[string]string{
+		"duration": strconv.Itoa(int(durQuery)),
+	}))
+	if err != nil {
+		return errors.Wrap(err, "sending header")
+	}
+
 	t = time.Now()
 	if err := newDurationRowser(toRowser, durQuery).ToRows(stream.Send); err != nil {
 		return errToStatusError(err)
@@ -262,7 +287,14 @@ func (h *GRPCHandler) QueryPQLUnary(ctx context.Context, req *pb.QueryPQLRequest
 	}
 	durFormat := time.Since(t)
 
-	table.Duration = int64(durQuery + durFormat)
+	duration := durQuery + durFormat
+	table.Duration = int64(duration)
+	err = grpc.SendHeader(ctx, metadata.New(map[string]string{
+		"duration": strconv.Itoa(int(duration)),
+	}))
+	if err != nil {
+		return nil, errors.Wrap(err, "sending header")
+	}
 
 	h.stats.Timing(pilosa.MetricGRPCUnaryQueryDurationSeconds, durQuery, 0.1)
 	h.stats.Timing(pilosa.MetricGRPCUnaryFormatDurationSeconds, durFormat, 0.1)

server/grpc_test.go

@@ -29,7 +29,9 @@ import (
 	"github.com/pilosa/pilosa/v2/sql"
 	"github.com/pilosa/pilosa/v2/test"
 	"github.com/pkg/errors"
+	"google.golang.org/grpc"
 	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/metadata"
 	"google.golang.org/grpc/status"
 )
 
@@ -354,9 +356,11 @@ func TestQueryPQLUnary(t *testing.T) {
 
 	i := m.MustCreateIndex(t, "i", pilosa.IndexOptions{})
 	m.MustCreateField(t, i.Name(), "f", pilosa.OptFieldKeys())
-	ctx := context.Background()
 	gh := server.NewGRPCHandler(m.API)
 
+	stream := &MockServerTransportStream{}
+	ctx := grpc.NewContextWithServerTransportStream(context.Background(), stream)
+
 	resp, err := gh.QueryPQLUnary(ctx, &pb.QueryPQLRequest{
 		Index: i.Name(),
 		Pql:   `Set(0, f="zero")`,
@@ -369,6 +373,10 @@ func TestQueryPQLUnary(t *testing.T) {
 	if resp.Duration == 0 {
 		t.Fatal("duration not recorded")
 	}
+	duration, err := stream.GetDuration()
+	if duration == 0 || err != nil {
+		t.Fatal("duration header not recorded")
+	}
 
 	_, err = gh.QueryPQLUnary(ctx, &pb.QueryPQLRequest{
 		Index: i.Name(),
@@ -400,6 +408,11 @@ func TestQueryPQL(t *testing.T) {
 		t.Fatal(err)
 	}
 
+	duration, err := mock.GetDuration()
+	if duration == 0 || err != nil {
+		t.Fatal("duration header not recorded")
+	}
+
 	if len(mock.Results) != 1 {
 		t.Fatal("expecting one result")
 	}
@@ -481,7 +494,9 @@ type (
 
 func TestQuerySQL(t *testing.T) {
 
-	ctx := context.Background()
+	stream := &MockServerTransportStream{}
+	ctx := grpc.NewContextWithServerTransportStream(context.Background(), stream)
+
 	gh, tearDownFunc := setUpTestQuerySQLUnary(ctx, t)
 	defer tearDownFunc()
 
@@ -924,6 +939,11 @@ func TestQuerySQL(t *testing.T) {
 				if resp.Duration == 0 {
 					t.Fatal("duration not recorded")
 				}
+				duration, err := stream.GetDuration()
+				if duration == 0 || err != nil {
+					t.Fatal("duration header not recorded")
+				}
+				stream.ClearMD()
 				tr := toTableResponse(resp)
 				if err := test.eq(test.exp, tr); err != nil {
 					t.Fatalf("sql: %s, error: %+v", test.sql, err)
@@ -942,6 +962,10 @@ func TestQuerySQL(t *testing.T) {
 				if mock.Results[0].Duration == 0 {
 					t.Fatal("duration not recorded")
 				}
+				duration, err := mock.GetDuration()
+				if duration == 0 || err != nil {
+					t.Fatal("duration header not recorded")
+				}
 				if len(mock.Results) > 1 && mock.Results[1].Duration != 0 {
 					t.Fatal("duration on second result expected to be zero")
 				}
@@ -953,7 +977,8 @@ func TestQuerySQL(t *testing.T) {
 
 func TestQuerySQLUnaryWithError(t *testing.T) {
 
-	ctx := context.Background()
+	stream := &MockServerTransportStream{}
+	ctx := grpc.NewContextWithServerTransportStream(context.Background(), stream)
 	gh, tearDownFunc := setUpTestQuerySQLUnary(ctx, t)
 	defer tearDownFunc()
 
@@ -999,7 +1024,8 @@ func TestCRUDIndexes(t *testing.T) {
 	m := test.RunCommand(t)
 	defer m.Close()
 
-	ctx := context.Background()
+	stream := &MockServerTransportStream{}
+	ctx := grpc.NewContextWithServerTransportStream(context.Background(), stream)
 	gh := server.NewGRPCHandler(m.API)
 
 	t.Run("CreateIndex", func(t *testing.T) {
@@ -1383,7 +1409,43 @@ func equalUnordered(exp tableResponse, got tableResponse) error {
 	return nil
 }
 
+type MockServerTransportStream struct {
+	header metadata.MD
+}
+
+func (stream *MockServerTransportStream) Method() string {
+	return ""
+}
+
+func (stream *MockServerTransportStream) SetHeader(md metadata.MD) error {
+	// Should probably merge md with value of stream.header, but this works since we have only one metadata value
+	stream.header = md
+	return nil
+}
+
+func (stream *MockServerTransportStream) SendHeader(md metadata.MD) error {
+	stream.header = md
+	return nil
+}
+
+func (stream *MockServerTransportStream) SetTrailer(md metadata.MD) error {
+	return nil
+}
+
+func (stream *MockServerTransportStream) GetDuration() (int, error) {
+	duration, ok := stream.header["duration"]
+	if ok {
+		return strconv.Atoi(duration[0])
+	}
+	return 0, errors.New("duration not recorded")
+}
+
+func (stream *MockServerTransportStream) ClearMD() {
+	stream.header = metadata.New(map[string]string{})
+}
+
 type mockPilosa_QuerySQLServer struct {
+	MockServerTransportStream
 	pb.Pilosa_QuerySQLServer
 	Results []*pb.RowResponse
 }
@@ -1393,6 +1455,18 @@ func (m *mockPilosa_QuerySQLServer) Send(result *pb.RowResponse) error {
 	return nil
 }
 
+func (m *mockPilosa_QuerySQLServer) SendHeader(md metadata.MD) error {
+	return m.MockServerTransportStream.SendHeader(md)
+}
+
+func (m *mockPilosa_QuerySQLServer) SetHeader(md metadata.MD) error {
+	return m.MockServerTransportStream.SetHeader(md)
+}
+
+func (m *mockPilosa_QuerySQLServer) SetTrailer(md metadata.MD) {
+	_ = m.MockServerTransportStream.SetTrailer(md)
+}
+
 func (m *mockPilosa_QuerySQLServer) Context() context.Context {
 	return context.Background()
 }

server/handler_test.go

@@ -40,6 +40,7 @@ import (
 	pb "github.com/pilosa/pilosa/v2/proto"
 	"github.com/pilosa/pilosa/v2/server"
 	"github.com/pilosa/pilosa/v2/test"
+	"google.golang.org/grpc"
 )
 
 func TestHandler_PostSchemaCluster(t *testing.T) {
@@ -396,10 +397,22 @@ func TestHandler_Endpoints(t *testing.T) {
 		}
 
 		for _, nodeUsage := range nodeUsages {
-			if len(nodeUsage.Disk.Indexes) != 2 {
-				t.Fatalf("wrong length index size list: %#v", nodeUsage.Disk.Indexes)
+			numIndexes := len(nodeUsage.Disk.IndexUsage)
+			if nodeUsage.Disk.TotalUse < 75000 || nodeUsage.Disk.TotalUse > 300000 {
+				// Usage measurements are not consistent between machines, or
+				// over time, as features and implementations change, so checking
+				// for a range of sizes may be most useful way to test the details of this.
+				t.Fatalf("expected 75k < total < 300k, got %d", nodeUsage.Disk.TotalUse)
+			}
+			if numIndexes != 2 {
+				t.Fatalf("wrong length index usage list: expected %d, got %d", 2, numIndexes)
+			}
+			numFields := len(nodeUsage.Disk.IndexUsage["i1"].Fields)
+			if numFields != len(i1.Fields()) {
+				t.Fatalf("wrong length field usage list: expected %d, got %d", len(i1.Fields()), numFields)
 			}
 		}
+
 	})
 
 	t.Run("UI/shard-distribution", func(t *testing.T) {
@@ -1505,7 +1518,9 @@ func TestQueryHistory(t *testing.T) {
 	test.Do(t, "POST", cmd.URL()+"/index/i0/field/f0", "")
 
 	gh := server.NewGRPCHandler(cmd.API)
-	_, err = gh.QuerySQLUnary(context.Background(), &pb.QuerySQLRequest{
+	stream := &MockServerTransportStream{}
+	ctx := grpc.NewContextWithServerTransportStream(context.Background(), stream)
+	_, err = gh.QuerySQLUnary(ctx, &pb.QuerySQLRequest{
 		Sql: `select * from i0`,
 	})
 

stattx.go

@@ -681,3 +681,7 @@ func (c *statTx) Sn() int64 {
 func (c *statTx) GetSortedFieldViewList(idx *Index, shard uint64) (fvs []txkey.FieldView, err error) {
 	return c.b.GetSortedFieldViewList(idx, shard)
 }
+
+func (tx *statTx) GetFieldSizeBytes(index, field string) (uint64, error) {
+	return 0, nil
+}

test/pilosa.go

@@ -41,6 +41,13 @@ type Command struct {
 	commandOptions []server.CommandOption
 }
 
+func OptTxSrc(src string) server.CommandOption {
+	return func(m *server.Command) error {
+		m.Config.Txsrc = src
+		return nil
+	}
+}
+
 func OptAllowedOrigins(origins []string) server.CommandOption {
 	return func(m *server.Command) error {
 		m.Config.Handler.AllowedOrigins = origins

tx.go

@@ -219,6 +219,8 @@ type Tx interface {
 
 	// GetSortedFieldViewList gets the set of FieldView(s)
 	GetSortedFieldViewList(idx *Index, shard uint64) (fvs []txkey.FieldView, err error)
+
+	GetFieldSizeBytes(index, field string) (uint64, error)
 }
 
 // Closer is used by Finders

txfactory.go

@@ -593,40 +593,156 @@ func (f *TxFactory) DumpAll() {
 	f.dbPerShard.DumpAll()
 }
 
-func (f *TxFactory) IndexSizes() (index2bytes map[string]int64, err error) {
-	// Open storage directory.
-	index2bytes = make(map[string]int64)
-	dirName, err := expandDirName(f.holder.path)
+// IndexUsageDetails computes the sum of filesizes used by the node, broken down
+// by index, field, fragments and keys.
+func (f *TxFactory) IndexUsageDetails() (map[string]IndexUsage, uint64, error) {
+	indexUsage := make(map[string]IndexUsage)
+	holderPath, err := expandDirName(f.holder.path)
 	if err != nil {
-		return index2bytes, errors.Wrap(err, "expanding data directory")
+		return indexUsage, 0, errors.Wrap(err, "expanding data directory")
 	}
 
 	idxs := f.holder.Indexes()
 
+	qcx := f.NewQcx()
+	defer qcx.Abort()
 	for _, idx := range idxs {
 		index := idx.name
-		fullName := path.Join(dirName, index)
-		roaringAndMeta, err := directoryUsage(fullName)
-		if err != nil {
-			return index2bytes, errors.Wrap(err, "getting disk usage for roaring and meta")
+		indexPath := path.Join(holderPath, index)
+
+		// field usage
+		fieldUsages := make(map[string]FieldUsage)
+		fragmentsTotal := uint64(0)
+		fieldKeysTotal := uint64(0)
+		fieldMetaBytesTotal := uint64(0)
+		fieldsTotal := uint64(0)
+		flds := idx.Fields()
+		for _, fld := range flds {
+			field := fld.Name()
+			if field == "_keys" {
+				continue
+			}
+			fUsage, err := f.fieldUsage(indexPath, fld)
+			if err != nil {
+				return indexUsage, 0, errors.Wrapf(err, "getting disk usage for index (%s)", index)
+			}
+
+			// non-roaring field usage
+			fragmentUsage := uint64(0)
+
+			for _, shard := range fld.AvailableShards(true).Slice() {
+				if err := func() error {
+					tx, finisher, err := qcx.GetTx(Txo{Write: !writable, Index: idx, Shard: shard})
+					if err != nil {
+						return errors.Wrap(err, "qcx.GetTx")
+					}
+					defer finisher(nil)
+
+					fieldBytes, err := tx.GetFieldSizeBytes(index, field)
+					if err != nil {
+						return errors.Wrapf(err, "getting disk usage for non-roaring fragments (%s)", field)
+					}
+					fragmentUsage += fieldBytes
+					return nil
+				}(); err != nil {
+					return indexUsage, 0, err
+				}
+			}
+
+			// add non-roaring to roaring
+			fUsage.Fragments += fragmentUsage
+			fUsage.Total += fragmentUsage
+
+			// add to running total
+			fieldMetaBytesTotal += fUsage.Metadata
+			fieldKeysTotal += fUsage.Keys
+			fragmentsTotal += fUsage.Fragments
+			fieldsTotal += fUsage.Total
+
+			fieldUsages[field] = fUsage
 		}
-		fullName += ".index.txstores@@@"
-		rbfOrLmdb, err := directoryUsage(fullName)
+
+		// index metadata, e.g. columnAttrs
+		indexMetaBytes, err := directoryUsage(indexPath, false)
 		if err != nil {
-			return index2bytes, errors.Wrap(err, "getting disk usage for backend")
+			return indexUsage, 0, errors.Wrapf(err, "getting disk usage for index metadata (%s)", index)
+		}
+
+		// index keys usage
+		indexKeysBytes := uint64(0)
+		if idx.keys {
+			keysPath := path.Join(indexPath, translateStoreDir)
+			indexKeysBytes, _ = directoryUsage(keysPath, true) // if directory doesn't exist, size = 0
+		}
+
+		indexUsage[index] = IndexUsage{
+			Total:          indexMetaBytes + indexKeysBytes + fieldsTotal,
+			Metadata:       indexMetaBytes + fieldMetaBytesTotal,
+			IndexKeys:      indexKeysBytes,
+			FieldKeysTotal: fieldKeysTotal,
+			Fragments:      fragmentsTotal,
+			Fields:         fieldUsages,
 		}
-		index2bytes[index] = roaringAndMeta + rbfOrLmdb
 	}
 
-	return index2bytes, nil
+	// node metadata, e.g. id allocator
+	nodeMetaBytes, err := directoryUsage(holderPath, false)
+	if err != nil {
+		return indexUsage, 0, errors.Wrapf(err, "getting disk usage for node metadata")
+	}
+
+	return indexUsage, nodeMetaBytes, nil
 }
 
-func directoryUsage(fname string) (int64, error) {
-	if !DirExists(fname) {
-		return 0, nil
+// fieldUsage computes the sum of filesizes used by a field in
+// the filesystem tree (roaring storage), broken down by keys and fragments.
+func (f *TxFactory) fieldUsage(indexPath string, fld *Field) (FieldUsage, error) {
+	fieldUsage := FieldUsage{}
+
+	field := fld.name
+
+	// row keys
+	keysBytes := int64(0)
+	var err error
+	keysBytes, err = fileSize(fld.TranslateStorePath())
+	if err != nil {
+		// if file doesn't exist, size = 0
+		keysBytes = 0
 	}
 
-	var size int64
+	// field metadata, e.g. rowAttrs
+	fieldPath := path.Join(indexPath, field)
+	metaBytes, err := directoryUsage(fieldPath, false) // this includes keys
+	if err != nil {
+		return fieldUsage, errors.Wrapf(err, "getting disk usage for field meta (%s)", field)
+	}
+
+	// fragment data
+	viewsPath := path.Join(fieldPath, "views")
+	fragmentBytes := uint64(0)
+	if dirExists(viewsPath) {
+		fragmentBytes, err = directoryUsage(viewsPath, true)
+		if err != nil {
+			return fieldUsage, errors.Wrapf(err, "getting disk usage for field fragments (%s)", field)
+		}
+	}
+
+	fieldUsage = FieldUsage{
+		Total:     metaBytes + fragmentBytes, // metaBytes includes keys
+		Metadata:  metaBytes - uint64(keysBytes),
+		Fragments: fragmentBytes,
+		Keys:      uint64(keysBytes),
+	}
+
+	return fieldUsage, nil
+}
+
+func directoryUsage(fname string, recursive bool) (uint64, error) {
+	if !dirExists(fname) {
+		return 0, errors.Errorf("directory does not exist (%s)", fname)
+	}
+
+	var size uint64
 
 	dir, err := os.Open(fname)
 	if err != nil {
@@ -640,14 +756,14 @@ func directoryUsage(fname string) (int64, error) {
 	}
 
 	for _, file := range files {
-		if file.IsDir() {
-			sz, err := directoryUsage(path.Join(fname, file.Name()))
+		if recursive && file.IsDir() {
+			sz, err := directoryUsage(path.Join(fname, file.Name()), true)
 			if err != nil {
 				return 0, err
 			}
 			size += sz
 		} else {
-			size += file.Size()
+			size += uint64(file.Size()) // NOTE this cast is safe for regular files, not necessarily others
 		}
 	}