mirror of
https://github.com/featurebasedb/featurebase.git
synced 2026-09-05 16:15:56 +00:00
411 lines
12 KiB
Go
411 lines
12 KiB
Go
package query
|
|
|
|
import (
|
|
"encoding/gob"
|
|
"fmt"
|
|
"math/rand"
|
|
"pilosa/db"
|
|
|
|
"tux21b.org/v1/gocql/uuid"
|
|
)
|
|
|
|
// A single step in the query plan.
|
|
type QueryStep struct {
|
|
id *uuid.UUID
|
|
operation string
|
|
inputs []QueryInput
|
|
location *db.Location
|
|
destination *db.Location
|
|
}
|
|
|
|
func (q QueryStep) StringHOLD() string {
|
|
return fmt.Sprintf("%s %s %s, LOC: %s, DEST: %s", q.operation, q.id.String(), q.inputs, q.location, q.destination)
|
|
}
|
|
|
|
type PortableQueryStep interface {
|
|
GetId() *uuid.UUID
|
|
GetLocation() *db.Location
|
|
}
|
|
|
|
///////////////////////////////////////////////////////////////////////////////////////////////////
|
|
// BASE
|
|
///////////////////////////////////////////////////////////////////////////////////////////////////
|
|
|
|
type BaseQueryStep struct {
|
|
Id *uuid.UUID
|
|
Operation string
|
|
Location *db.Location
|
|
Destination *db.Location
|
|
}
|
|
|
|
func (self *BaseQueryStep) GetId() *uuid.UUID {
|
|
return self.Id
|
|
}
|
|
func (self *BaseQueryStep) GetLocation() *db.Location {
|
|
return self.Location
|
|
}
|
|
|
|
func (self *BaseQueryStep) LocIsDest() bool {
|
|
if self.Location.ProcessId == self.Destination.ProcessId && self.Location.FragmentId == self.Destination.FragmentId {
|
|
return true
|
|
}
|
|
return false
|
|
}
|
|
|
|
type BaseQueryResult struct {
|
|
Id *uuid.UUID
|
|
Data interface{}
|
|
}
|
|
|
|
func (self *BaseQueryResult) ResultId() *uuid.UUID {
|
|
return self.Id
|
|
}
|
|
|
|
func (self *BaseQueryResult) ResultData() interface{} {
|
|
return self.Data
|
|
}
|
|
|
|
///////////////////////////////////////////////////////////////////////////////////////////////////
|
|
// COUNT
|
|
///////////////////////////////////////////////////////////////////////////////////////////////////
|
|
type CountQueryStep struct {
|
|
*BaseQueryStep
|
|
Input *uuid.UUID
|
|
}
|
|
|
|
type CountQueryResult struct {
|
|
*BaseQueryResult
|
|
}
|
|
|
|
// QueryTree for COUNT queries
|
|
type CountQueryTree struct {
|
|
subquery QueryTree
|
|
location *db.Location
|
|
}
|
|
|
|
// Uses consistent hashing function to select node containing data for GET operation
|
|
func (qt *CountQueryTree) getLocation(d *db.Database) *db.Location {
|
|
return qt.subquery.getLocation(d)
|
|
}
|
|
|
|
///////////////////////////////////////////////////////////////////////////////////////////////////
|
|
// UNION
|
|
///////////////////////////////////////////////////////////////////////////////////////////////////
|
|
type UnionQueryStep struct {
|
|
*BaseQueryStep
|
|
Inputs []*uuid.UUID
|
|
}
|
|
|
|
type UnionQueryResult struct {
|
|
*BaseQueryResult
|
|
}
|
|
|
|
// QueryTree for UNION queries
|
|
type UnionQueryTree struct {
|
|
subqueries []QueryTree
|
|
location *db.Location
|
|
}
|
|
|
|
// Uses consistent hashing function to select node containing data for GET operation
|
|
func (qt *UnionQueryTree) getLocation(d *db.Database) *db.Location {
|
|
if qt.location == nil {
|
|
subqueryLength := len(qt.subqueries)
|
|
if subqueryLength > 0 {
|
|
locationIndex := rand.Intn(subqueryLength)
|
|
subquery := qt.subqueries[locationIndex]
|
|
qt.location = subquery.getLocation(d)
|
|
}
|
|
}
|
|
return qt.location
|
|
}
|
|
|
|
///////////////////////////////////////////////////////////////////////////////////////////////////
|
|
// INTERSECT
|
|
///////////////////////////////////////////////////////////////////////////////////////////////////
|
|
type IntersectQueryStep struct {
|
|
*BaseQueryStep
|
|
Inputs []*uuid.UUID
|
|
}
|
|
|
|
type IntersectQueryResult struct {
|
|
*BaseQueryResult
|
|
}
|
|
|
|
// QueryTree for UNION queries
|
|
type IntersectQueryTree struct {
|
|
subqueries []QueryTree
|
|
location *db.Location
|
|
}
|
|
|
|
// Uses consistent hashing function to select node containing data for GET operation
|
|
func (qt *IntersectQueryTree) getLocation(d *db.Database) *db.Location {
|
|
if qt.location == nil {
|
|
subqueryLength := len(qt.subqueries)
|
|
if subqueryLength > 0 {
|
|
locationIndex := rand.Intn(subqueryLength)
|
|
subquery := qt.subqueries[locationIndex]
|
|
qt.location = subquery.getLocation(d)
|
|
}
|
|
}
|
|
return qt.location
|
|
}
|
|
|
|
///////////////////////////////////////////////////////////////////////////////////////////////////
|
|
// CAT
|
|
///////////////////////////////////////////////////////////////////////////////////////////////////
|
|
type CatQueryStep struct {
|
|
*BaseQueryStep
|
|
Inputs []*uuid.UUID
|
|
}
|
|
|
|
type CatQueryResult struct {
|
|
*BaseQueryResult
|
|
}
|
|
|
|
// QueryTree for CAT queries
|
|
type CatQueryTree struct {
|
|
subqueries []QueryTree
|
|
location *db.Location
|
|
}
|
|
|
|
// Uses consistent hashing function to select node containing data for GET operation
|
|
func (qt *CatQueryTree) getLocation(d *db.Database) *db.Location {
|
|
if qt.location == nil {
|
|
subqueryLength := len(qt.subqueries)
|
|
if subqueryLength > 0 {
|
|
locationIndex := rand.Intn(subqueryLength)
|
|
subquery := qt.subqueries[locationIndex]
|
|
qt.location = subquery.getLocation(d)
|
|
}
|
|
}
|
|
return qt.location
|
|
}
|
|
|
|
///////////////////////////////////////////////////////////////////////////////////////////////////
|
|
// GET
|
|
///////////////////////////////////////////////////////////////////////////////////////////////////
|
|
type GetQueryStep struct {
|
|
*BaseQueryStep
|
|
Bitmap *db.Bitmap
|
|
Slice int
|
|
}
|
|
|
|
type GetQueryResult struct {
|
|
*BaseQueryResult
|
|
}
|
|
|
|
// QueryTree for GET queries
|
|
type GetQueryTree struct {
|
|
bitmap *db.Bitmap
|
|
slice int
|
|
}
|
|
|
|
// Uses consistent hashing function to select node containing data for GET operation
|
|
func (qt *GetQueryTree) getLocation(d *db.Database) *db.Location {
|
|
slice := d.GetOrCreateSlice(qt.slice) // TODO: this should probably be just GetSlice (no create)
|
|
fragment, err := d.GetFragmentForBitmap(slice, qt.bitmap)
|
|
if err != nil {
|
|
panic(err)
|
|
}
|
|
return fragment.GetLocation()
|
|
}
|
|
|
|
///////////////////////////////////////////////////////////////////////////////////////////////////
|
|
// SET
|
|
///////////////////////////////////////////////////////////////////////////////////////////////////
|
|
type SetQueryStep struct {
|
|
*BaseQueryStep
|
|
Bitmap *db.Bitmap
|
|
ProfileId uint64
|
|
}
|
|
|
|
type SetQueryResult struct {
|
|
*BaseQueryResult
|
|
}
|
|
|
|
// QueryTree for SET queries
|
|
type SetQueryTree struct {
|
|
bitmap *db.Bitmap
|
|
profile_id uint64
|
|
}
|
|
|
|
// Uses consistent hashing function to select node containing data for GET operation
|
|
func (qt *SetQueryTree) getLocation(d *db.Database) *db.Location {
|
|
slice, err := d.GetSliceForProfile(qt.profile_id)
|
|
fragment, err := d.GetFragmentForBitmap(slice, qt.bitmap)
|
|
if err != nil {
|
|
panic(err)
|
|
}
|
|
return fragment.GetLocation()
|
|
}
|
|
|
|
///////////////////////////////////////////////////////////////////////////////////////////////////
|
|
|
|
func init() {
|
|
gob.Register(SetQueryResult{})
|
|
gob.Register(GetQueryResult{})
|
|
gob.Register(CatQueryResult{})
|
|
gob.Register(UnionQueryResult{})
|
|
gob.Register(IntersectQueryResult{})
|
|
gob.Register(CountQueryResult{})
|
|
|
|
gob.Register(SetQueryStep{})
|
|
gob.Register(GetQueryStep{})
|
|
gob.Register(CatQueryStep{})
|
|
gob.Register(UnionQueryStep{})
|
|
gob.Register(IntersectQueryStep{})
|
|
gob.Register(CountQueryStep{})
|
|
}
|
|
|
|
///////////////////////////////////////////////////////////////////////////////////////////////////
|
|
|
|
// This is the output of the query planner. Contains a list of steps which can be performed in parallel
|
|
//type QueryPlan []QueryStep
|
|
type QueryPlan []interface{}
|
|
|
|
type QueryPlanner struct {
|
|
Database *db.Database
|
|
}
|
|
type QueryTree interface {
|
|
getLocation(d *db.Database) *db.Location
|
|
}
|
|
|
|
// QueryTree for UNION and INTERSECT queries
|
|
type CompositeQueryTree struct {
|
|
operation string
|
|
subqueries []QueryTree
|
|
location *db.Location
|
|
}
|
|
|
|
// Randomly select location from subqueries (so subqueries roll up into composite queries while minimizing inter-node data traffic)
|
|
func (qt *CompositeQueryTree) getLocation(d *db.Database) *db.Location {
|
|
if qt.location == nil {
|
|
subqueryLength := len(qt.subqueries)
|
|
if subqueryLength > 0 {
|
|
locationIndex := rand.Intn(subqueryLength)
|
|
subquery := qt.subqueries[locationIndex]
|
|
qt.location = subquery.getLocation(d)
|
|
}
|
|
}
|
|
return qt.location
|
|
}
|
|
|
|
// Builds QueryTree object from Query. Pass slice=-1 to perform operation on all slices
|
|
func (qp *QueryPlanner) buildTree(query *Query, slice int) QueryTree {
|
|
var tree QueryTree
|
|
|
|
// handle SET operation regardless of the slice
|
|
if query.Operation == "set" {
|
|
tree = &SetQueryTree{query.Inputs[0].(*db.Bitmap), query.ProfileId}
|
|
return tree
|
|
}
|
|
|
|
// handle the remaining operations, taking slice into consideration
|
|
if slice == -1 {
|
|
tree = &CatQueryTree{}
|
|
numSlices, err := qp.Database.NumSlices()
|
|
if err != nil {
|
|
panic(err)
|
|
}
|
|
for slice := 0; slice < numSlices; slice++ {
|
|
subtree := qp.buildTree(query, slice)
|
|
composite := tree.(*CatQueryTree)
|
|
composite.subqueries = append(composite.subqueries, subtree)
|
|
}
|
|
} else {
|
|
if query.Operation == "get" {
|
|
tree = &GetQueryTree{query.Inputs[0].(*db.Bitmap), slice}
|
|
return tree
|
|
} else if query.Operation == "count" {
|
|
subquery := qp.buildTree(query.Inputs[0].(*Query), slice)
|
|
tree = &CountQueryTree{subquery: subquery}
|
|
} else if query.Operation == "union" {
|
|
subqueries := make([]QueryTree, len(query.Inputs))
|
|
for i, input := range query.Inputs {
|
|
subqueries[i] = qp.buildTree(input.(*Query), slice)
|
|
}
|
|
tree = &UnionQueryTree{subqueries: subqueries}
|
|
} else if query.Operation == "intersect" {
|
|
subqueries := make([]QueryTree, len(query.Inputs))
|
|
for i, input := range query.Inputs {
|
|
subqueries[i] = qp.buildTree(input.(*Query), slice)
|
|
}
|
|
tree = &IntersectQueryTree{subqueries: subqueries}
|
|
} else {
|
|
subqueries := make([]QueryTree, len(query.Inputs))
|
|
for i, input := range query.Inputs {
|
|
subqueries[i] = qp.buildTree(input.(*Query), slice)
|
|
}
|
|
tree = &CompositeQueryTree{operation: query.Operation, subqueries: subqueries}
|
|
}
|
|
}
|
|
return tree
|
|
}
|
|
|
|
// Produces flattened QueryPlan from QueryTree input
|
|
func (qp *QueryPlanner) flatten(qt QueryTree, id *uuid.UUID, location *db.Location) *QueryPlan {
|
|
plan := QueryPlan{}
|
|
if composite, ok := qt.(*CompositeQueryTree); ok {
|
|
inputs := make([]QueryInput, len(composite.subqueries))
|
|
step := QueryStep{id, composite.operation, inputs, composite.getLocation(qp.Database), location}
|
|
for index, subq := range composite.subqueries {
|
|
sub_id := uuid.RandomUUID()
|
|
step.inputs[index] = &sub_id
|
|
subq_steps := qp.flatten(subq, &sub_id, composite.getLocation(qp.Database))
|
|
plan = append(plan, *subq_steps...)
|
|
}
|
|
plan = append(plan, step)
|
|
} else if cat, ok := qt.(*CatQueryTree); ok {
|
|
inputs := make([]*uuid.UUID, len(cat.subqueries))
|
|
step := CatQueryStep{&BaseQueryStep{id, "cat", cat.getLocation(qp.Database), location}, inputs}
|
|
for index, subq := range cat.subqueries {
|
|
sub_id := uuid.RandomUUID()
|
|
step.Inputs[index] = &sub_id
|
|
subq_steps := qp.flatten(subq, &sub_id, cat.getLocation(qp.Database))
|
|
plan = append(plan, *subq_steps...)
|
|
}
|
|
plan = append(plan, step)
|
|
} else if union, ok := qt.(*UnionQueryTree); ok {
|
|
inputs := make([]*uuid.UUID, len(union.subqueries))
|
|
step := UnionQueryStep{&BaseQueryStep{id, "union", union.getLocation(qp.Database), location}, inputs}
|
|
for index, subq := range union.subqueries {
|
|
sub_id := uuid.RandomUUID()
|
|
step.Inputs[index] = &sub_id
|
|
subq_steps := qp.flatten(subq, &sub_id, union.getLocation(qp.Database))
|
|
plan = append(plan, *subq_steps...)
|
|
}
|
|
plan = append(plan, step)
|
|
} else if intersect, ok := qt.(*IntersectQueryTree); ok {
|
|
inputs := make([]*uuid.UUID, len(intersect.subqueries))
|
|
step := IntersectQueryStep{&BaseQueryStep{id, "intersect", intersect.getLocation(qp.Database), location}, inputs}
|
|
for index, subq := range intersect.subqueries {
|
|
sub_id := uuid.RandomUUID()
|
|
step.Inputs[index] = &sub_id
|
|
subq_steps := qp.flatten(subq, &sub_id, intersect.getLocation(qp.Database))
|
|
plan = append(plan, *subq_steps...)
|
|
}
|
|
plan = append(plan, step)
|
|
} else if get, ok := qt.(*GetQueryTree); ok {
|
|
step := GetQueryStep{&BaseQueryStep{id, "get", get.getLocation(qp.Database), location}, get.bitmap, get.slice}
|
|
plan := QueryPlan{step}
|
|
return &plan
|
|
} else if set, ok := qt.(*SetQueryTree); ok {
|
|
step := SetQueryStep{&BaseQueryStep{id, "set", set.getLocation(qp.Database), location}, set.bitmap, set.profile_id}
|
|
plan := QueryPlan{step}
|
|
return &plan
|
|
} else if cnt, ok := qt.(*CountQueryTree); ok {
|
|
sub_id := uuid.RandomUUID()
|
|
step := &CountQueryStep{&BaseQueryStep{id, "count", cnt.getLocation(qp.Database), location}, &sub_id}
|
|
subq_steps := qp.flatten(cnt.subquery, &sub_id, cnt.getLocation(qp.Database))
|
|
plan = append(plan, *subq_steps...)
|
|
plan = append(plan, step)
|
|
}
|
|
return &plan
|
|
}
|
|
|
|
// Transforms Query into QueryTree and flattens to QueryPlan object
|
|
func (qp *QueryPlanner) Plan(query *Query, id *uuid.UUID, destination *db.Location) *QueryPlan {
|
|
queryTree := qp.buildTree(query, -1)
|
|
//return qp.flatten(queryTree, id, destination) // TODO: remove the "id" parameter, since we are using the query.Id as the value
|
|
return qp.flatten(queryTree, query.Id, destination)
|
|
}
|