scriptoriumadmin/featurebase
1
0
Fork 0
mirror of https://github.com/featurebasedb/featurebase.git synced 2026-09-11 23:31:03 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Merge pull request #642 from jaddr2line/query-extract

Browse source 
Add an "Extract" query
This commit is contained in:
Jaden Weiss 2020-08-10 19:32:36 -04:00 • committed by GitHub
commit 96c2364c1b
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
7 changed files with 3681 additions and 219 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

35
docs/query-language.md
View file

@ -850,6 +850,41 @@ Rows(job, like="%t")
{"rows":null,"keys":["management","student"]}
```
#### Extract
**Spec:**
```
Extract(<ROW_CALL>, [<ROWS_CALL>...])
```
**Description:**
Extract intersects a set of columns with a set of rows in order to extract a subset of the index.
The result is a table consisting of the matched columns and the rows which they intersect.
This is similar to a select query in a SQL database.
**Result Type:** Object with an array of the selected fields and an array of the selected columns.
The column array contains objects containing a column identifier and an array of field values.
Field values are typed as such:
- Bool Field - boolean or null
- Mutex Field (unkeyed) - 64-bit unsigned integer or null
- Mutex Field (keyed) - string or null
- Integer Field - 64-bit signed integer or null
- Decimal Field - Pilosa decimal value or null
- Set Field (unkeyed) - array of 64-bit unsigned integers
- Set Field (keyed) - array of strings
- Time Field - same as the equivalent Set
**Examples:**
List all stargazers who have starred repository 1, and the full set of repositories they have starred:
```request
Extract(Row(stargazer=1), Rows(stargazer))
```
```response
{"fields":[{"name":"stargazer","type":"set"}],"columns":[{"column":3,"rows":[[1, 2, 3]]}]}
```
#### Group By
**Spec:**

181
encoding/proto/proto.go
View file

@ -512,12 +512,18 @@ func (s Serializer) encodeQueryResponse(m *pilosa.QueryResponse) *internal.Query
case pilosa.RowIDs:
pb.Results[i].Type = queryResultTypeRowIDs
pb.Results[i].RowIDs = result
case pilosa.ExtractedIDMatrix:
pb.Results[i].Type = queryResultTypeExtractedIDMatrix
pb.Results[i].ExtractedIDMatrix = s.endcodeExtractedIDMatrix(result)
case []pilosa.GroupCount:
pb.Results[i].Type = queryResultTypeGroupCounts
pb.Results[i].GroupCounts = s.encodeGroupCounts(result)
case pilosa.RowIdentifiers:
pb.Results[i].Type = queryResultTypeRowIdentifiers
pb.Results[i].RowIdentifiers = s.encodeRowIdentifiers(result)
case pilosa.ExtractedTable:
pb.Results[i].Type = queryResultTypeExtractedTable
pb.Results[i].ExtractedTable = s.encodeExtractedTable(result)
case pilosa.Pair:
pb.Results[i].Type = queryResultTypePair
pb.Results[i].Pairs = []*internal.Pair{s.encodePair(result)}
@ -1313,6 +1319,8 @@ const (
queryResultTypePair
queryResultTypePairField
queryResultTypeSignedRow
queryResultTypeExtractedIDMatrix
queryResultTypeExtractedTable
)
func (s Serializer) decodeQueryResult(pb *internal.QueryResult) interface{} {
@ -1343,6 +1351,10 @@ func (s Serializer) decodeQueryResult(pb *internal.QueryResult) interface{} {
return s.decodePair(pb.Pairs[0])
case queryResultTypePairField:
return s.decodePairField(pb.PairField)
case queryResultTypeExtractedIDMatrix:
return s.decodeExtractedIDMatrix(pb.ExtractedIDMatrix)
case queryResultTypeExtractedTable:
return s.decodeExtractedTable(pb.ExtractedTable)
}
panic(fmt.Sprintf("unknown type: %d", pb.Type))
}
@ -1410,6 +1422,83 @@ func (s Serializer) decodeAttr(attr *internal.Attr) (key string, value interface
}
}
func (s Serializer) decodeExtractedIDMatrix(m *internal.ExtractedIDMatrix) pilosa.ExtractedIDMatrix {
cols := make([]pilosa.ExtractedIDColumn, len(m.Columns))
for i, c := range m.Columns {
rows := make([][]uint64, len(c.Vals))
for j, r := range c.Vals {
rows[j] = r.IDs
}
cols[i] = pilosa.ExtractedIDColumn{
ColumnID: c.ID,
Rows: rows,
}
}
return pilosa.ExtractedIDMatrix{
Fields: m.Fields,
Columns: cols,
}
}
func (s Serializer) decodeExtractedTable(t *internal.ExtractedTable) pilosa.ExtractedTable {
fields := make([]pilosa.ExtractedTableField, len(t.Fields))
for i, f := range t.Fields {
fields[i] = pilosa.ExtractedTableField{
Name: f.Name,
Type: f.Type,
}
}
columns := make([]pilosa.ExtractedTableColumn, len(t.Columns))
for i, c := range t.Columns {
var col pilosa.KeyOrID
switch kid := c.KeyOrID.(type) {
case *internal.ExtractedTableColumn_ID:
col = pilosa.KeyOrID{
ID: kid.ID,
}
case *internal.ExtractedTableColumn_Key:
col = pilosa.KeyOrID{
Keyed: true,
Key: kid.Key,
}
}
rows := make([]interface{}, len(c.Values))
for j, v := range rows {
var val interface{}
switch v := v.(type) {
case *internal.ExtractedTableValue_IDs:
val = v.IDs.IDs
case *internal.ExtractedTableValue_Keys:
val = v.Keys.Keys
case *internal.ExtractedTableValue_BSIValue:
val = v.BSIValue
case *internal.ExtractedTableValue_MutexID:
val = v.MutexID
case *internal.ExtractedTableValue_MutexKey:
val = v.MutexKey
case *internal.ExtractedTableValue_Bool:
val = v.Bool
}
rows[j] = val
}
columns[i] = pilosa.ExtractedTableColumn{
Column: col,
Rows: rows,
}
}
return pilosa.ExtractedTable{
Fields: fields,
Columns: columns,
}
}
func (s Serializer) decodeRowIdentifiers(a *internal.RowIdentifiers) *pilosa.RowIdentifiers {
return &pilosa.RowIdentifiers{
Rows: a.Rows,
@ -1580,6 +1669,98 @@ func (s Serializer) encodeFieldRows(a []pilosa.FieldRow) []*internal.FieldRow {
return other
}
func (s Serializer) endcodeExtractedIDMatrix(m pilosa.ExtractedIDMatrix) *internal.ExtractedIDMatrix {
cols := make([]*internal.ExtractedIDColumn, len(m.Columns))
for i, v := range m.Columns {
vals := make([]*internal.IDList, len(v.Rows))
for j, f := range v.Rows {
vals[j] = &internal.IDList{IDs: f}
}
cols[i] = &internal.ExtractedIDColumn{
ID: v.ColumnID,
Vals: vals,
}
}
return &internal.ExtractedIDMatrix{
Fields: m.Fields,
Columns: cols,
}
}
func (s Serializer) encodeExtractedTable(t pilosa.ExtractedTable) *internal.ExtractedTable {
fields := make([]*internal.ExtractedTableField, len(t.Fields))
for i, f := range t.Fields {
fields[i] = &internal.ExtractedTableField{
Name: f.Name,
Type: f.Type,
}
}
cols := make([]*internal.ExtractedTableColumn, len(t.Columns))
for i, c := range t.Columns {
var col internal.ExtractedTableColumn
if c.Column.Keyed {
col.KeyOrID = &internal.ExtractedTableColumn_Key{Key: c.Column.Key}
} else {
col.KeyOrID = &internal.ExtractedTableColumn_ID{ID: c.Column.ID}
}
rows := make([]*internal.ExtractedTableValue, len(c.Rows))
for j, v := range c.Rows {
switch v := v.(type) {
case []uint64:
rows[j] = &internal.ExtractedTableValue{
Value: &internal.ExtractedTableValue_IDs{
IDs: &internal.IDList{
IDs: v,
},
},
}
case []string:
rows[j] = &internal.ExtractedTableValue{
Value: &internal.ExtractedTableValue_Keys{
Keys: &internal.KeyList{
Keys: v,
},
},
}
case int64:
rows[j] = &internal.ExtractedTableValue{
Value: &internal.ExtractedTableValue_BSIValue{
BSIValue: v,
},
}
case uint64:
rows[j] = &internal.ExtractedTableValue{
Value: &internal.ExtractedTableValue_MutexID{
MutexID: v,
},
}
case string:
rows[j] = &internal.ExtractedTableValue{
Value: &internal.ExtractedTableValue_MutexKey{
MutexKey: v,
},
}
case bool:
rows[j] = &internal.ExtractedTableValue{
Value: &internal.ExtractedTableValue_Bool{
Bool: v,
},
}
}
}
col.Values = rows
cols[i] = &col
}
return &internal.ExtractedTable{
Fields: fields,
Columns: cols,
}
}
func (s Serializer) encodePairs(a pilosa.Pairs) []*internal.Pair {
other := make([]*internal.Pair, len(a))
for i := range a {

468
executor.go
View file

@ -289,6 +289,10 @@ func (e *executor) safeCopy(resp QueryResponse) (out QueryResponse) {
out.Results = append(out.Results, x)
case []GroupCount:
out.Results = append(out.Results, x)
case ExtractedTable:
out.Results = append(out.Results, x)
case ExtractedIDMatrix:
out.Results = append(out.Results, x)
case RowIdentifiers:
// no bitmap material, so should be ok to skip Clone()
out.Results = append(out.Results, x)
@ -714,6 +718,9 @@ func (e *executor) executeCall(ctx context.Context, tx Tx, index string, c *pql.
case "Rows":
statFn()
return e.executeRows(ctx, tx, index, c, shards, opt)
case "Extract":
statFn()
return e.executeExtract(ctx, tx, index, c, shards, opt)
case "GroupBy":
statFn()
return e.executeGroupBy(ctx, tx, index, c, shards, opt)
@ -2712,6 +2719,298 @@ func (e *executor) executeRowsShard(ctx context.Context, tx Tx, index string, fi
return rowIDs, nil
}
type ExtractedTableField struct {
Name string `json:"name"`
Type string `json:"type"`
}
type KeyOrID struct {
ID uint64
Key string
Keyed bool
}
func (kid KeyOrID) MarshalJSON() ([]byte, error) {
if kid.Keyed {
return json.Marshal(kid.Key)
}
return json.Marshal(kid.ID)
}
type ExtractedTableColumn struct {
Column KeyOrID `json:"column"`
Rows []interface{} `json:"rows"`
}
type ExtractedTable struct {
Fields []ExtractedTableField `json:"fields"`
Columns []ExtractedTableColumn `json:"columns"`
}
type ExtractedIDColumn struct {
ColumnID uint64
Rows [][]uint64
}
type ExtractedIDMatrix struct {
Fields []string
Columns []ExtractedIDColumn
}
func (e *ExtractedIDMatrix) Append(m ExtractedIDMatrix) {
e.Columns = append(e.Columns, m.Columns...)
if e.Fields == nil {
e.Fields = m.Fields
}
}
func (e *executor) executeExtract(ctx context.Context, tx Tx, index string, c *pql.Call, shards []uint64, opt *execOptions) (ExtractedIDMatrix, error) {
// Extract the column filter call.
if len(c.Children) < 1 {
return ExtractedIDMatrix{}, errors.New("missing column filter in Extract")
}
filter := c.Children[0]
// Extract fields from rows calls.
fields := make([]string, len(c.Children)-1)
for i, rows := range c.Children[1:] {
if rows.Name != "Rows" {
return ExtractedIDMatrix{}, errors.Errorf("child call of Extract is %q but expected Rows", rows.Name)
}
var fieldName string
var ok bool
for k, v := range rows.Args {
switch k {
case "field", "_field":
fieldName = v.(string)
ok = true
default:
return ExtractedIDMatrix{}, errors.Errorf("unsupported Rows argument for Extract: %q", k)
}
}
if !ok {
return ExtractedIDMatrix{}, errors.New("missing field specification in Rows")
}
fields[i] = fieldName
}
// Execute calls in bulk on each remote node and merge.
mapFn := func(ctx context.Context, shard uint64) (interface{}, error) {
return e.executeExtractShard(ctx, tx, index, fields, filter, shard)
}
// Merge returned results at coordinating node.
reduceFn := func(ctx context.Context, prev, v interface{}) interface{} {
other, _ := prev.(ExtractedIDMatrix)
if err := ctx.Err(); err != nil {
return err
}
other.Append(v.(ExtractedIDMatrix))
return other
}
// Get full result set.
other, err := e.mapReduce(ctx, index, shards, c, opt, mapFn, reduceFn)
if err != nil {
return ExtractedIDMatrix{}, err
}
results, _ := other.(ExtractedIDMatrix)
sort.Slice(results.Columns, func(i, j int) bool {
return results.Columns[i].ColumnID < results.Columns[j].ColumnID
})
return results, nil
}
func mergeBits(bits *Row, mask uint64, out map[uint64]uint64) {
for _, v := range bits.Columns() {
out[v] |= mask
}
}
var trueRowFakeID = []uint64{1}
var falseRowFakeID = []uint64{0}
func (e *executor) executeExtractShard(ctx context.Context, tx Tx, index string, fields []string, filter *pql.Call, shard uint64) (ExtractedIDMatrix, error) {
// Execute filter.
colsBitmap, err := e.executeBitmapCallShard(ctx, tx, index, filter, shard)
if err != nil {
return ExtractedIDMatrix{}, errors.Wrap(err, "failed to get extraction column filter")
}
// Fetch index.
idx := e.Holder.Index(index)
if idx == nil {
return ExtractedIDMatrix{}, ErrIndexNotFound
}
// Decompress columns bitmap.
cols := colsBitmap.Columns()
// Generate a matrix to stuff the results into.
m := make([]ExtractedIDColumn, len(cols))
{
rowsBuf := make([][]uint64, len(m)*len(fields))
for i, c := range cols {
m[i] = ExtractedIDColumn{
ColumnID: c,
Rows: rowsBuf[i*len(fields) : (i+1)*len(fields) : (i+1)*len(fields)],
}
}
}
if len(m) == 0 {
return ExtractedIDMatrix{
Fields: fields,
Columns: m,
}, nil
}
mLookup := make(map[uint64]int)
for i, j := range cols {
mLookup[j] = i
}
// Process fields.
for i, name := range fields {
// Look up the field.
field := idx.Field(name)
if field == nil {
return ExtractedIDMatrix{}, ErrFieldNotFound
}
switch field.Type() {
case FieldTypeSet, FieldTypeMutex, FieldTypeTime:
// Handle a set field by listing the rows and then intersecting them with the filter.
// Extract the standard view fragment.
fragment := e.Holder.fragment(index, name, viewStandard, shard)
if fragment == nil {
// There is nothing here.
continue
}
// List all rows in the standard view.
rows, err := fragment.rows(ctx, tx, 0)
if err != nil {
return ExtractedIDMatrix{}, errors.Wrap(err, "listing rows in set field")
}
// Loop over each row and scan the intersection with the filter.
for _, rowID := range rows {
// Load row from fragment.
row, err := fragment.row(tx, rowID)
if err != nil {
return ExtractedIDMatrix{}, errors.Wrap(err, "loading row from fragment")
}
// Apply column filter to row.
row = row.Intersect(colsBitmap)
// Rotate vector into the matrix.
for _, columnID := range row.Columns() {
fieldSlot := &m[mLookup[columnID]].Rows[i]
*fieldSlot = append(*fieldSlot, rowID)
}
}
case FieldTypeBool:
// Handle bool fields by scanning the true and false rows and assigning an integer.
// Extract the standard view fragment.
fragment := e.Holder.fragment(index, name, viewStandard, shard)
if fragment == nil {
// There is nothing here.
continue
}
// Fetch true and false rows.
trueRow, err := fragment.row(tx, trueRowID)
if err != nil {
return ExtractedIDMatrix{}, errors.Wrap(err, "loading true row from fragment")
}
falseRow, err := fragment.row(tx, falseRowID)
if err != nil {
return ExtractedIDMatrix{}, errors.Wrap(err, "loading true row from fragment")
}
// Fetch values by column.
for j := range m {
col := m[j].ColumnID
switch {
case trueRow.Includes(col):
m[j].Rows[i] = trueRowFakeID
case falseRow.Includes(col):
m[j].Rows[i] = falseRowFakeID
}
}
case FieldTypeInt, FieldTypeDecimal:
// Handle an int/decimal field by rotating a BSI matrix.
// Extract the BSI view fragment.
fragment := e.Holder.fragment(index, name, viewBSIGroupPrefix+name, shard)
if fragment == nil {
// There is nothing here.
continue
}
// Load the BSI group.
bsig := field.bsiGroup(name)
if bsig == nil {
return ExtractedIDMatrix{}, ErrBSIGroupNotFound
}
// Load the BSI exists bit.
exists, err := fragment.row(tx, bsiExistsBit)
if err != nil {
return ExtractedIDMatrix{}, errors.Wrap(err, "loading BSI exists bit from fragment")
}
// Filter BSI exists bit by selected columns.
exists = exists.Intersect(colsBitmap)
if !exists.Any() {
// No relevant BSI values are present in this fragment.
continue
}
// Populate a map with the BSI data.
data := make(map[uint64]uint64)
mergeBits(exists, 0, data)
// Copy in the sign bit.
sign, err := fragment.row(tx, bsiSignBit)
if err != nil {
return ExtractedIDMatrix{}, errors.Wrap(err, "loading BSI sign bit from fragment")
}
sign = sign.Intersect(exists)
mergeBits(sign, 1<<63, data)
// Copy in the significand.
for i := uint(0); i < bsig.BitDepth; i++ {
bits, err := fragment.row(tx, bsiOffsetBit+uint64(i))
if err != nil {
return ExtractedIDMatrix{}, errors.Wrap(err, "loading BSI significand bit from fragment")
}
bits = bits.Intersect(exists)
mergeBits(bits, 1<<i, data)
}
// Store the results back into the matrix.
for columnID, val := range data {
// Convert to two's complement.
val = uint64((2*(int64(val)>>63) + 1) * int64(val&^(1<<63)))
m[mLookup[columnID]].Rows[i] = []uint64{val}
}
}
}
// Emit the final matrix.
// Like RowIDs, this is an internal type and will need to be converted.
return ExtractedIDMatrix{
Fields: fields,
Columns: m,
}, nil
}
func (e *executor) executeRowShard(ctx context.Context, tx Tx, index string, c *pql.Call, shard uint64) (*Row, error) {
span, _ := tracing.StartSpanFromContext(ctx, "Executor.executeRowShard")
@ -4452,18 +4751,24 @@ func (e *executor) translateResults(ctx context.Context, index string, idx *Inde
}
func (e *executor) collectResultIDs(index string, idx *Index, call *pql.Call, result interface{}, idSet map[uint64]struct{}) error {
row, ok := result.(*Row)
if !ok {
return nil
} else if !idx.Keys() {
return nil
}
switch result := result.(type) {
case *Row:
if !idx.Keys() {
return nil
}
for _, segment := range row.Segments() {
for _, col := range segment.Columns() {
idSet[col] = struct{}{}
for _, segment := range result.Segments() {
for _, col := range segment.Columns() {
idSet[col] = struct{}{}
}
}
case ExtractedIDMatrix:
for _, col := range result.Columns {
idSet[col.ColumnID] = struct{}{}
}
}
return nil
}
@ -4630,6 +4935,151 @@ func (e *executor) translateResult(ctx context.Context, index string, idx *Index
}
return other, nil
case ExtractedIDMatrix:
type fieldMapper = func([]uint64) (interface{}, error)
fields := make([]ExtractedTableField, len(result.Fields))
mappers := make([]fieldMapper, len(result.Fields))
for i, v := range result.Fields {
field := idx.Field(v)
if field == nil {
return nil, ErrFieldNotFound
}
typ := field.Type()
fields[i] = ExtractedTableField{
Name: v,
Type: typ,
}
var mapper fieldMapper
switch typ {
case FieldTypeBool:
mapper = func(ids []uint64) (interface{}, error) {
switch len(ids) {
case 0:
return nil, nil
case 1:
switch ids[0] {
case 0:
return false, nil
case 1:
return true, nil
default:
return nil, errors.Errorf("invalid ID for boolean %q: %d", field.Name(), ids[0])
}
default:
return nil, errors.Errorf("boolean %q has too many values: %v", field.Name(), ids)
}
}
case FieldTypeSet, FieldTypeTime:
if field.Keys() {
translator := field.TranslateStore()
mapper = func(ids []uint64) (interface{}, error) {
return translator.TranslateIDs(ids)
}
} else {
mapper = func(ids []uint64) (interface{}, error) {
if ids == nil {
ids = []uint64{}
}
return ids, nil
}
}
case FieldTypeMutex:
if field.Keys() {
translator := field.TranslateStore()
mapper = func(ids []uint64) (interface{}, error) {
switch len(ids) {
case 0:
return nil, nil
case 1:
return translator.TranslateID(ids[0])
default:
return nil, errors.Errorf("mutex %q has too many values: %v", field.Name(), ids)
}
}
} else {
mapper = func(ids []uint64) (interface{}, error) {
switch len(ids) {
case 0:
return nil, nil
case 1:
return ids[0], nil
default:
return nil, errors.Errorf("mutex %q has too many values: %v", field.Name(), ids)
}
}
}
case FieldTypeInt:
mapper = func(ids []uint64) (interface{}, error) {
switch len(ids) {
case 0:
return nil, nil
case 1:
return int64(ids[0]), nil
default:
return nil, errors.Errorf("BSI field %q has too many values: %v", field.Name(), ids)
}
}
case FieldTypeDecimal:
scale := field.Options().Scale
mapper = func(ids []uint64) (interface{}, error) {
switch len(ids) {
case 0:
return nil, nil
case 1:
return pql.NewDecimal(int64(ids[0]), scale), nil
default:
return nil, errors.Errorf("BSI field %q has too many values: %v", field.Name(), ids)
}
}
default:
return nil, errors.Errorf("field type %q not yet supported", typ)
}
mappers[i] = mapper
}
var translateCol func(uint64) (KeyOrID, error)
if idx.keys {
translateCol = func(id uint64) (KeyOrID, error) {
return KeyOrID{Keyed: true, Key: idSet[id]}, nil
}
} else {
translateCol = func(id uint64) (KeyOrID, error) {
return KeyOrID{ID: id}, nil
}
}
cols := make([]ExtractedTableColumn, len(result.Columns))
colData := make([]interface{}, len(cols)*len(result.Fields))
for i, col := range result.Columns {
data := colData[i*len(result.Fields) : (i+1)*len(result.Fields) : (i+1)*len(result.Fields)]
for j, rows := range col.Rows {
v, err := mappers[j](rows)
if err != nil {
return nil, errors.Wrap(err, "translating extracted table value")
}
data[j] = v
}
colTrans, err := translateCol(col.ColumnID)
if err != nil {
return nil, errors.Wrap(err, "translating column ID in extracted table")
}
cols[i] = ExtractedTableColumn{
Column: colTrans,
Rows: data,
}
}
return ExtractedTable{
Fields: fields,
Columns: cols,
}, nil
}
return result, nil

285
executor_test.go
View file

@ -4233,6 +4233,291 @@ func benchmarkExistence(nn bool, b *testing.B) {
func BenchmarkExecutor_Existence_True(b *testing.B) { benchmarkExistence(true, b) }
func BenchmarkExecutor_Existence_False(b *testing.B) { benchmarkExistence(false, b) }
func TestExecutor_Execute_Extract(t *testing.T) {
c := test.MustRunCluster(t, 3)
defer c.Close()
c.CreateField(t, "i", pilosa.IndexOptions{TrackExistence: true}, "set")
c.ImportBits(t, "i", "set", [][2]uint64{
{0, 1},
{0, 2},
{3, 1},
{4, 1},
{4, 4 * ShardWidth},
{5, ShardWidth},
})
c.Query(t, "i", fmt.Sprintf("Clear(%d, set=5)", ShardWidth))
c.CreateField(t, "i", pilosa.IndexOptions{TrackExistence: true}, "keyset", pilosa.OptFieldKeys())
c.Query(t, "i", `
Set(0, keyset="h")
Set(1, keyset="xyzzy")
Set(0, keyset="plugh")
`)
c.CreateField(t, "i", pilosa.IndexOptions{TrackExistence: true}, "mutex", pilosa.OptFieldTypeMutex(pilosa.CacheTypeRanked, 5000))
c.ImportBits(t, "i", "mutex", [][2]uint64{
{0, 1},
{0, 2},
{4, 4 * ShardWidth},
})
c.CreateField(t, "i", pilosa.IndexOptions{TrackExistence: true}, "keymutex", pilosa.OptFieldKeys(), pilosa.OptFieldTypeMutex(pilosa.CacheTypeRanked, 5000))
c.Query(t, "i", `
Set(0, keymutex="h")
Set(1, keymutex="xyzzy")
Set(3, keymutex="plugh")
`)
c.CreateField(t, "i", pilosa.IndexOptions{TrackExistence: true}, "time", pilosa.OptFieldTypeTime("YMDH"))
c.Query(t, "i", `
Set(0, time=1, 2016-01-01T00:00)
Set(1, time=2, 2017-01-01T00:00)
Set(3, time=3, 2018-01-01T00:00)
`)
c.CreateField(t, "i", pilosa.IndexOptions{TrackExistence: true}, "keytime", pilosa.OptFieldKeys(), pilosa.OptFieldTypeTime("YMDH"))
c.Query(t, "i", `
Set(0, keytime="h", 2016-01-01T00:00)
Set(1, keytime="xyzzy", 2017-01-01T00:00)
Set(0, keytime="plugh", 2018-01-01T00:00)
`)
c.CreateField(t, "i", pilosa.IndexOptions{TrackExistence: true}, "bsint", pilosa.OptFieldTypeInt(-100, 100))
c.Query(t, "i", `
Set(0, bsint=1)
Set(1, bsint=-1)
Set(3, bsint=2)
`)
c.CreateField(t, "i", pilosa.IndexOptions{TrackExistence: true}, "bsidecimal", pilosa.OptFieldTypeDecimal(2))
c.Query(t, "i", `
Set(0, bsidecimal=0.01)
Set(1, bsidecimal=1.00)
Set(3, bsidecimal=-1.01)
`)
c.CreateField(t, "i", pilosa.IndexOptions{TrackExistence: true}, "bool", pilosa.OptFieldTypeBool())
c.Query(t, "i", `
Set(0, bool=true)
Set(1, bool=false)
Set(3, bool=true)
`)
resp := c.Query(t, "i", `Extract(All(), Rows(set), Rows(keyset), Rows(mutex), Rows(keymutex), Rows(time), Rows(keytime), Rows(bsint), Rows(bsidecimal), Rows(bool))`)
expect := []interface{}{
pilosa.ExtractedTable{
Fields: []pilosa.ExtractedTableField{
{
Name: "set",
Type: pilosa.FieldTypeSet,
},
{
Name: "keyset",
Type: pilosa.FieldTypeSet,
},
{
Name: "mutex",
Type: pilosa.FieldTypeMutex,
},
{
Name: "keymutex",
Type: pilosa.FieldTypeMutex,
},
{
Name: "time",
Type: pilosa.FieldTypeTime,
},
{
Name: "keytime",
Type: pilosa.FieldTypeTime,
},
{
Name: "bsint",
Type: pilosa.FieldTypeInt,
},
{
Name: "bsidecimal",
Type: pilosa.FieldTypeDecimal,
},
{
Name: "bool",
Type: pilosa.FieldTypeBool,
},
},
Columns: []pilosa.ExtractedTableColumn{
{
Column: pilosa.KeyOrID{ID: 0},
Rows: []interface{}{
[]uint64{},
[]string{
"h",
"plugh",
},
nil,
"h",
[]uint64{
1,
},
[]string{
"h",
"plugh",
},
int64(1),
pql.NewDecimal(1, 2),
true,
},
},
{
Column: pilosa.KeyOrID{ID: 1},
Rows: []interface{}{
[]uint64{
0,
3,
4,
},
[]string{
"xyzzy",
},
uint64(0),
"xyzzy",
[]uint64{
2,
},
[]string{
"xyzzy",
},
int64(-1),
pql.NewDecimal(100, 2),
false,
},
},
{
Column: pilosa.KeyOrID{ID: 2},
Rows: []interface{}{
[]uint64{
0,
},
[]string{},
uint64(0),
nil,
[]uint64{},
[]string{},
nil,
nil,
nil,
},
},
{
Column: pilosa.KeyOrID{ID: 3},
Rows: []interface{}{
[]uint64{},
[]string{},
nil,
"plugh",
[]uint64{
3,
},
[]string{},
int64(2),
pql.NewDecimal(-101, 2),
true,
},
},
{
Column: pilosa.KeyOrID{ID: ShardWidth},
Rows: []interface{}{
[]uint64{},
[]string{},
nil,
nil,
[]uint64{},
[]string{},
nil,
nil,
nil,
},
},
{
Column: pilosa.KeyOrID{ID: 4 * ShardWidth},
Rows: []interface{}{
[]uint64{
4,
},
[]string{},
uint64(4),
nil,
[]uint64{},
[]string{},
nil,
nil,
nil,
},
},
},
},
}
if !reflect.DeepEqual(expect, resp.Results) {
t.Errorf("expected %v but got %v", expect, resp.Results)
}
}
func TestExecutor_Execute_Extract_Keyed(t *testing.T) {
c := test.MustRunCluster(t, 3)
defer c.Close()
c.CreateField(t, "i", pilosa.IndexOptions{TrackExistence: true, Keys: true}, "set")
c.Query(t, "i", `
Set("h", set=1)
Set("h", set=2)
Set("xyzzy", set=2)
Set("plugh", set=1)
Clear("plugh", set=1)
`)
resp := c.Query(t, "i", `Extract(All(), Rows(set))`)
expect := []interface{}{
pilosa.ExtractedTable{
Fields: []pilosa.ExtractedTableField{
{
Name: "set",
Type: "set",
},
},
Columns: []pilosa.ExtractedTableColumn{
{
Column: pilosa.KeyOrID{Keyed: true, Key: "plugh"},
Rows: []interface{}{
[]uint64{},
},
},
{
Column: pilosa.KeyOrID{Keyed: true, Key: "h"},
Rows: []interface{}{
[]uint64{
1,
2,
},
},
},
{
Column: pilosa.KeyOrID{Keyed: true, Key: "xyzzy"},
Rows: []interface{}{
[]uint64{
2,
},
},
},
},
},
}
if !reflect.DeepEqual(expect, resp.Results) {
t.Errorf("expected %v but got %v", expect, resp.Results)
}
}
func TestExecutor_Execute_Rows(t *testing.T) {
c := test.MustRunCluster(t, 3)
defer c.Close()

2881
internal/public.pb.go
View file

File diff suppressed because it is too large Load diff

49
internal/public.proto
View file

@ -19,6 +19,53 @@ message RowIdentifiers {
repeated string Keys = 2;
}
message IDList {
repeated uint64 IDs = 1;
}
message ExtractedIDColumn {
uint64 ID = 1;
repeated IDList Vals = 2;
}
message ExtractedIDMatrix {
repeated string Fields = 1;
repeated ExtractedIDColumn Columns = 2;
}
message KeyList {
repeated string Keys = 1;
}
message ExtractedTableValue {
oneof Value {
IDList IDs = 1;
KeyList Keys = 2;
int64 BSIValue = 3;
uint64 MutexID = 4;
string MutexKey = 5;
bool Bool = 6;
}
}
message ExtractedTableColumn {
oneof KeyOrID {
string Key = 1;
uint64 ID = 2;
}
repeated ExtractedTableValue Values = 3;
}
message ExtractedTableField {
string Name = 1;
string Type = 2;
}
message ExtractedTable {
repeated ExtractedTableField Fields = 1;
repeated ExtractedTableColumn Columns = 2;
}
message Pair {
uint64 ID = 1;
string Key = 3;
@ -112,6 +159,8 @@ message QueryResult {
SignedRow SignedRow = 10;
PairsField PairsField = 11;
PairField PairField = 12;
ExtractedIDMatrix ExtractedIDMatrix = 13;
ExtractedTable ExtractedTable = 14;
}
message ImportRequest {

1
pql/ast.go
View file

@ -393,6 +393,7 @@ var callInfoByFunc = map[string]callInfo{
},
"Union": {allowUnknown: false},
"UnionRows": {allowUnknown: false},
"Extract": {allowUnknown: false},
"Xor": {allowUnknown: false},
// things that take _field