Merge pull request #1218 from jaten-molecula/short_txkey

short_txkey elides index and shard from the txkey

bluegreentx.go

@@ -18,9 +18,7 @@ import (
 	"bytes"
 	"fmt"
 	"io"
-	"math"
 	"reflect"
-	"sort"
 	"sync"
 
 	"github.com/pilosa/pilosa/v2/roaring"
@@ -909,53 +907,6 @@ func (c *blueGreenTx) Sn() int64 {
 	return bsn
 }
 
-func (c *blueGreenTx) SliceOfShards(index, field, view, optionalViewPath string) (sliceOfShards []uint64, err error) {
-	// doesn't change state, so we don't really need see() call here. And we don't have a single shard for it.
-	//c.checker.see(index, field, view, shard) // don't have shard.
-	defer func() {
-		if r := recover(); r != nil {
-			c.Dump(c.short, math.MaxUint64)
-			AlwaysPrintf("see SliceOfShards() panic '%v' at '%v'", r, stack())
-			panic(r)
-		}
-	}()
-	slcA, errA := c.a.SliceOfShards(index, field, view, optionalViewPath)
-	slcB, errB := c.b.SliceOfShards(index, field, view, optionalViewPath)
-
-	if !c.o.blueGreenOff {
-		compareErrors(errA, errB)
-
-		// sort order may be different, and that's ok.
-		cpa := append([]uint64{}, slcA...)
-		cpb := append([]uint64{}, slcB...)
-		sort.Slice(cpa, func(i, j int) bool { return cpa[i] < cpa[j] })
-		sort.Slice(cpb, func(i, j int) bool { return cpb[i] < cpb[j] })
-
-		if !reflect.DeepEqual(cpa, cpb) {
-			// report the first difference
-			ma := make(map[uint64]bool)
-			for _, ka := range slcA {
-				ma[ka] = true
-			}
-			for _, kb := range slcB {
-				if !ma[kb] {
-					//vv("blueGreenTx SliceOfShards diference! B(%v) had shard %v, but A(%v) did not. cpa='%#v'; cpb='%#v'; in the SliceOfShards returned slice.", c.bs, kb, c.as, cpa, cpb)
-					c.Dump(c.short, math.MaxUint64)
-					panic(fmt.Sprintf("blueGreenTx SliceOfShards diference! B(%v) had shard %v, but A(%v) did not. cpa='%#v'; cpb='%#v'; in the SliceOfShards returned slice.", c.bs, kb, c.as, cpa, cpb))
-				}
-				delete(ma, kb)
-			}
-			if len(ma) != 0 {
-				for firstDifference := range ma {
-					panic(fmt.Sprintf("blueGreenTx SliceOfShards diference! A(%v) had %v, but B(%v) did not. cpa='%#v'; cpb='%#v'; in the SliceOfShards returned slice.", c.as, firstDifference, c.bs, cpa, cpb))
-				}
-			}
-			panic(fmt.Sprintf("blueGreenTx SliceOfShards diference \n slcA(%v)='%#v';\n slcB(%v)='%#v';\n", c.as, cpa, c.bs, cpb))
-		}
-	}
-	return slcB, errB
-}
-
 // MultiReaderB is returned by RoaringBitmapReader. It verifies
 // that identical byte streams are read from its two members.
 type MultiReaderB struct {

bolt.go

@@ -728,33 +728,6 @@ func (tx *BoltTx) Contains(index, field, view string, shard uint64, key uint64)
 	return exists, err
 }
 
-func (tx *BoltTx) SliceOfShards(index, field, view, optionalViewPath string) (sliceOfShards []uint64, err error) {
-
-	prefix := txkey.AllShardPrefix(index, field, view)
-
-	bi := NewBoltIterator(tx, prefix)
-	defer bi.Close()
-
-	lastShard := uint64(0)
-	firstDone := false
-	for bi.Next() {
-		shard := txkey.ShardFromKey(bi.lastKey)
-		if firstDone {
-			if shard != lastShard {
-				sliceOfShards = append(sliceOfShards, shard)
-			}
-			lastShard = shard
-		} else {
-			// first time
-			lastShard = shard
-			firstDone = true
-			sliceOfShards = append(sliceOfShards, shard)
-		}
-
-	}
-	return
-}
-
 // key is the container key for the first roaring Container
 // roaring docs: Iterator returns a ContainterIterator which *after* a call to Next(), a call to Value() will
 // return the first container at or after key. found will be true if a

bolt_test.go

@@ -1239,29 +1239,6 @@ func TestBolt_DeleteIndex_over100k(t *testing.T) {
 	}
 }
 
-func TestBolt_SliceOfShards(t *testing.T) {
-
-	dbwrap, clean := mustOpenEmptyBoltWrapper("TestBolt_SliceOfShards")
-	defer clean()
-	defer dbwrap.Close()
-	index, field, view := "i", "f", "v"
-	shards := []uint64{0, 1, 2, 3, 1000001, 2000001}
-	putme := uint64(179)
-	for _, shard := range shards {
-		BoltMustSetBitvalue(dbwrap, index, field, view, shard, putme)
-	}
-	tx, _ := dbwrap.NewTx(!writable, index, Txo{})
-	defer tx.Rollback()
-
-	slc, err := tx.SliceOfShards(index, field, view, "")
-	panicOn(err)
-	for i := range shards {
-		if shards[i] != slc[i] {
-			panic(fmt.Sprintf("expected at i=%v that slc[i]=%v = shards[i]=%v", i, slc[i], shards[i]))
-		}
-	}
-}
-
 func TestBolt_HasData(t *testing.T) {
 
 	db, clean := mustOpenEmptyBoltWrapper("TestBolt_SliceOfShards")

catcher.go

@@ -294,15 +294,6 @@ func (c *catcherTx) RoaringBitmapReader(index, field, view string, shard uint64,
 func (c *catcherTx) Type() string {
 	return c.b.Type()
 }
-func (c *catcherTx) SliceOfShards(index, field, view, optionalViewPath string) (sliceOfShards []uint64, err error) {
-	defer func() {
-		if r := recover(); r != nil {
-			AlwaysPrintf("see SliceOfShards() panic '%v' at '%v'", r, stack())
-			panic(r)
-		}
-	}()
-	return c.b.SliceOfShards(index, field, view, optionalViewPath)
-}
 
 func (c *catcherTx) Group() *TxGroup {
 	return c.b.Group()

rbf.go

@@ -29,7 +29,7 @@ import (
 	"github.com/pilosa/pilosa/v2/rbf"
 	rbfcfg "github.com/pilosa/pilosa/v2/rbf/cfg"
 	"github.com/pilosa/pilosa/v2/roaring"
-	"github.com/pilosa/pilosa/v2/txkey"
+	txkey "github.com/pilosa/pilosa/v2/short_txkey"
 	"github.com/pkg/errors"
 )
 
@@ -399,26 +399,6 @@ func (tx *RBFTx) RoaringBitmapReader(index, field, view string, shard uint64, fr
 	return ioutil.NopCloser(&buf), sz, err
 }
 
-func (tx *RBFTx) SliceOfShards(index, field, view, optionalViewPath string) (sliceOfShards []uint64, err error) {
-
-	prefix := string(txkey.AllShardPrefix(index, field, view))
-
-	names, err := tx.tx.BitmapNames()
-	if err != nil {
-		return nil, err
-	}
-
-	// Iterate over shard names and collect shards from matching field/view prefix.
-	for _, name := range names {
-		if !strings.HasPrefix(name, prefix) {
-			continue
-		}
-		shard := txkey.ShardFromPrefix([]byte(name))
-		sliceOfShards = append(sliceOfShards, shard)
-	}
-	return sliceOfShards, nil
-}
-
 func (tx *RBFTx) NewTxIterator(index, field, view string, shard uint64) *roaring.Iterator {
 	b, err := tx.RoaringBitmap(index, field, view, shard)
 	panicOn(err)

rbf/ingest_test.go

@@ -27,7 +27,7 @@ import (
 
 	"github.com/pilosa/pilosa/v2/rbf/cfg"
 	"github.com/pilosa/pilosa/v2/roaring"
-	"github.com/pilosa/pilosa/v2/txkey"
+	txkey "github.com/pilosa/pilosa/v2/short_txkey"
 )
 
 func rbfName(index, field, view string, shard uint64) string {

rbf/tx.go

@@ -25,7 +25,7 @@ import (
 	"github.com/benbjohnson/immutable"
 	"github.com/pilosa/pilosa/v2/hash"
 	"github.com/pilosa/pilosa/v2/roaring"
-	"github.com/pilosa/pilosa/v2/txkey"
+	txkey "github.com/pilosa/pilosa/v2/short_txkey"
 )
 
 var _ = txkey.ToString

rbf/tx_test.go

@@ -22,7 +22,7 @@ import (
 	"time"
 
 	"github.com/pilosa/pilosa/v2/rbf"
-	"github.com/pilosa/pilosa/v2/txkey"
+	txkey "github.com/pilosa/pilosa/v2/short_txkey"
 )
 
 func TestTx_CommitRollback(t *testing.T) {

rbf/util.go

@@ -18,7 +18,7 @@ import (
 	"io"
 	"strings"
 
-	"github.com/pilosa/pilosa/v2/txkey"
+	txkey "github.com/pilosa/pilosa/v2/short_txkey"
 )
 
 func (tx *Tx) dumpAllPages(showLeaves bool) error {

short_txkey/txkey.go

@@ -0,0 +1,192 @@
+// 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 txkey consolidates in one place the use of keys to index into our
+// various storage/txn back-ends. The short_txkey version omits the
+// index and shard, since these are implicitly part of our database-per-shard
+// in an index scheme. In other words, every database is only in exactly
+// one shard of one index already. There is no need to repeat the index
+// and shard in these keys.
+package short_txkey
+
+import (
+	"encoding/binary"
+	"fmt"
+)
+
+// Key produces the bytes that we use as a key to query the storage/tx engine.
+// The roaringContainerKey argument to Key() is a container key into a roaring Container.
+// The return value from Key() is constructed as follows:
+//
+//     ~field;view<ckey#
+//
+// where ckey is always exactly 8 bytes, uint64 big-endian encoded.
+//
+// Keys always start with either '~' or '>'. Keys always end with '#'.
+// Keys always contain exactly one each of ';' and '<', in that order.
+// The field is between the '~' and the ';'. It must be at least 1 byte long.
+// The view is between the ';' and the '<'. It must be at least 1 byte long.
+// The ckey is the 8 bytes between the '<' and the '#'.
+// The Prefix of a key ends at, and includes, the '<'. It is at least 13 bytes long.
+// The index, field, and view are not allowed to contain these reserved bytes:
+//  {'~', '>', ';', ':', '<', '#', '$', '%', '^', '(', ')', '*', '!'}
+//
+// The bytes {'+', '/', '-', '_', '.', and '=' can be used in index, field, and view; to enable
+// base-64 encoding.
+//
+// The shortest possible key is 14 bytes. It would be laid out like this:
+//           ~f;v<12345678#
+//           12345678901234
+//
+// keys starting with '~' are regular value keys.
+// keys starting with '>' are symlink keys.
+//
+// NB must be kept in sync with Prefix() and KeyExtractContainerKey().
+//
+func Key(index, field, view string, shard, roaringContainerKey uint64) (r []byte) {
+
+	prefix := Prefix(index, field, view, shard)
+
+	var ckey [9]byte
+	binary.BigEndian.PutUint64(ckey[:8], roaringContainerKey)
+	ckey[8] = byte('#')
+	return append(prefix, ckey[:]...)
+}
+
+// KeyAndPrefix returns the equivalent of Key() and Prefix() calls.
+func KeyAndPrefix(index, field, view string, shard, roaringContainerKey uint64) (key, prefix []byte) {
+	prefix = Prefix(index, field, view, shard)
+
+	var ckey [9]byte
+	binary.BigEndian.PutUint64(ckey[:8], roaringContainerKey)
+	ckey[8] = byte('#')
+	key = append(prefix, ckey[:]...)
+	return
+}
+
+var _ = KeyAndPrefix // keep linter happy
+
+func MustValidateKey(bkey []byte) {
+	n := len(bkey)
+	if n < 14 {
+		panic(fmt.Sprintf("bkey too short, must have at least 14 bytes: '%v'", string(bkey)))
+	}
+	typ := bkey[0]
+	if typ != '~' && typ != '>' {
+		panic(fmt.Sprintf("bkey did not start with '~' for value nor '>' for symlink: '%v'", string(bkey)))
+	}
+	if bkey[n-10] != '<' {
+		panic(fmt.Sprintf("bkey did not have '<' at 9 bytes from the end: '%v'", string(bkey)))
+	}
+	if bkey[n-1] != '#' {
+		panic(fmt.Sprintf("bkey did not end in '#': '%v'", string(bkey)))
+	}
+}
+
+// KeyExtractContainerKey extracts the containerKey from bkey.
+// key example: field;view<ckey
+// shortest: ~f;v<12345678#
+//           1234567890123456789012345
+//  numbering len(bkey) - i:
+//           5432109876543210987654321
+func KeyExtractContainerKey(bkey []byte) (containerKey uint64) {
+	n := len(bkey)
+	MustValidateKey(bkey)
+	containerKey = binary.BigEndian.Uint64(bkey[(n - 9):(n - 1)])
+	return
+}
+
+func AllShardPrefix(index, field, view string) (r []byte) {
+	r = make([]byte, 0, 64)
+	r = append(r, '~')
+	r = append(r, []byte(field)...)
+	r = append(r, ';')
+	r = append(r, []byte(view)...)
+	r = append(r, '<')
+	return
+}
+
+// Prefix returns everything from Key up to and
+// including the '<' byte in a Key. The prefix excludes the roaring container key itself.
+// NB must be kept in sync with Key() and KeyExtractContainerKey().
+func Prefix(index, field, view string, shard uint64) (r []byte) {
+	r = make([]byte, 0, 32)
+	r = append(r, '~')
+	r = append(r, []byte(field)...)
+	r = append(r, ';')
+	r = append(r, []byte(view)...)
+	r = append(r, '<')
+	return
+}
+
+// IndexOnlyPrefix returns a "~" prefix suitable for DeleteIndex and a key-scan to
+// remove all storage. We assume only one index in this database, so delete everything.
+//
+func IndexOnlyPrefix(indexName string) (r []byte) {
+	return []byte("~")
+}
+
+// same for deleting a whole field.
+func FieldPrefix(index, field string) (r []byte) {
+	r = make([]byte, 0, 16)
+	r = append(r, '~')
+	r = append(r, []byte(field)...)
+	r = append(r, ';')
+	return
+}
+
+// PrefixFromKey key example: ~field;view<ckey#
+//                  n-9    n-1
+// ... : 01234567 < 01234567  #
+//          view       ckey
+func PrefixFromKey(bkey []byte) (prefix []byte) {
+	n := len(bkey)
+	return bkey[:(n - 9)]
+}
+
+func ToString(bkey []byte) (r string) {
+	field, view, ckey := Split(bkey)
+	return fmt.Sprintf("fld:'%v';vw:'%v';ckey@%020d", field, view, ckey)
+}
+
+func PrefixToString(pre []byte) (r string) {
+	field, view := SplitPrefix(pre)
+	return fmt.Sprintf("fld:'%v';vw:'%v';", field, view)
+}
+
+func Split(bkey []byte) (field, view string, ckey uint64) {
+	ckey = KeyExtractContainerKey(bkey)
+	n := len(bkey)
+	field, view = SplitPrefix(bkey[:(n - 9)])
+	return
+}
+
+// full key: ~field;view<ckey#
+// prefix  : ~field;view<
+func SplitPrefix(pre []byte) (field, view string) {
+	n := len(pre)
+
+	// prefix: ~field;view<
+	beg := 1
+	for i := 1; i < n; i++ {
+		switch pre[i] {
+		case ';':
+			field = string(pre[beg:i])
+			beg = i + 1
+			view = string(pre[beg:(n - 1)])
+			return
+		}
+	}
+	panic(fmt.Sprintf("malformed prefix '%v' / '%#v', could not Split", string(pre), pre))
+}

short_txkey/txkey_test.go

@@ -0,0 +1,94 @@
+// 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 short_txkey
+
+import (
+	"bytes"
+	"encoding/binary"
+	"fmt"
+	"testing"
+)
+
+func Test_KeyPrefix(t *testing.T) {
+
+	// Prefix() must agree with Key(), but not have the key at the end.
+	// This is important for iteration over containers.
+
+	index, field, view := "i", "f", "v"
+
+	needle := Key(index, field, view, 0, 0)
+
+	// prefix example: i%f;v:12345678<
+	prefix := Prefix(index, field, view, 0)
+
+	//fmt.Printf("needle = '%v'\n", string(needle))
+	//fmt.Printf("prefix = '%v'\n", string(prefix))
+
+	if !bytes.HasPrefix(needle, prefix) {
+		panic(fmt.Sprintf("Prefix() output '%v'was not a prefix of Key() '%v'", string(needle), string(prefix)))
+	}
+	npre := len(prefix)
+	nneed := len(needle)
+	if npre+9 != nneed {
+		panic(fmt.Sprintf("Prefix() output len %v '%v' was not 9 characters shorter than Key() len %v '%v'", npre, string(prefix), nneed, string(needle)))
+	}
+
+	// verify panic on submitting a prefix
+	func() {
+		defer func() {
+			r := recover()
+			if r == nil {
+				panic(fmt.Sprintf("should have seen panic on call to KeyExtractContainerKey(prefix='%v')", prefix))
+			}
+		}()
+		KeyExtractContainerKey(prefix) // should panic.
+	}()
+}
+
+func Test_PrefixFromKey(t *testing.T) {
+	k := []byte("~f;v<12345678#")
+	x := []byte("~f;v<")
+	pre := PrefixFromKey(k)
+	if !bytes.Equal(pre, x) {
+		nx := len(x)
+		npre := len(pre)
+		if nx != npre {
+			panic(fmt.Sprintf("nx=%v, npre=%v; expected '%v', observed '%v'", nx, npre, string(x), string(pre)))
+		}
+		for i := 0; i < nx; i++ {
+			if x[i] != pre[i] {
+				panic(fmt.Sprintf("first diff at index %v, expected '%v', observed '%v'", i, string(x[:i]), string(pre[:i])))
+			}
+		}
+		panic(fmt.Sprintf("expected:\n%v\n, observed:\n%v\n", string(x), string(pre)))
+	}
+}
+
+func Test_Split(t *testing.T) {
+	bkey := []byte("~f;v<12345678#")
+	var xckey uint64 = 43
+	binary.BigEndian.PutUint64(bkey[5:13], xckey)
+
+	f, v, ckey := Split(bkey)
+	if f != "f" {
+		panic("wrong field")
+	}
+	if v != "v" {
+		panic("wrong view")
+	}
+	if ckey != xckey {
+		panic("wrong ckey")
+	}
+}

stattx.go

@@ -667,22 +667,6 @@ func (c *statTx) Type() string {
 	return c.b.Type()
 }
 
-func (c *statTx) SliceOfShards(index, field, view, optionalViewPath string) (sliceOfShards []uint64, err error) {
-	me := kSliceOfShards
-
-	t0 := time.Now()
-	defer func() {
-		c.stats.add(me, time.Since(t0))
-	}()
-	defer func() {
-		if r := recover(); r != nil {
-			AlwaysPrintf("see SliceOfShards() panic '%v' at '%v'", r, stack())
-			panic(r)
-		}
-	}()
-	return c.b.SliceOfShards(index, field, view, optionalViewPath)
-}
-
 // Sn retreives the serial number of the Tx.
 func (c *statTx) Sn() int64 {
 	return c.b.Sn()

tx.go

@@ -193,12 +193,6 @@ type Tx interface {
 
 	RoaringBitmapReader(index, field, view string, shard uint64, fragmentPathForRoaring string) (r io.ReadCloser, sz int64, err error)
 
-	// SliceOfShards returns all of the shards for the specified index, field, view triple.
-	// Use within pilosa supposes a new read-only transaction was created just
-	// for the SliceOfShards() call. The legacy RoaringTx version is the only
-	// one that needs optionalViewPath; any other Tx implementation can ignore that.
-	SliceOfShards(index, field, view, optionalViewPath string) (sliceOfShards []uint64, err error)
-
 	// Group returns nil or the TxGroup that this Tx is a part of.
 	Group() *TxGroup