Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

semantics: Use a BitSet #11819

Merged
merged 7 commits into from
Nov 28, 2022
Merged
Show file tree
Hide file tree
Changes from 6 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion go/vt/vtgate/planbuilder/concatenateGen4.go
Original file line number Diff line number Diff line change
Expand Up @@ -104,7 +104,7 @@ func (c *concatenateGen4) Rewrite(inputs ...logicalPlan) error {
func (c *concatenateGen4) ContainsTables() semantics.TableSet {
var tableSet semantics.TableSet
for _, source := range c.sources {
tableSet.MergeInPlace(source.ContainsTables())
tableSet = tableSet.Merge(source.ContainsTables())
}
return tableSet
}
Expand Down
2 changes: 1 addition & 1 deletion go/vt/vtgate/planbuilder/operators/expressions.go
Original file line number Diff line number Diff line change
Expand Up @@ -36,7 +36,7 @@ func BreakExpressionInLHSandRHS(
switch node := cursor.Node().(type) {
case *sqlparser.ColName:
deps := ctx.SemTable.RecursiveDeps(node)
if deps.NumberOfTables() == 0 {
if deps.IsEmpty() {
err = vterrors.Errorf(vtrpcpb.Code_INTERNAL, "unknown column. has the AST been copied?")
return false
}
Expand Down
2 changes: 1 addition & 1 deletion go/vt/vtgate/planbuilder/operators/helpers.go
Original file line number Diff line number Diff line change
Expand Up @@ -71,7 +71,7 @@ type TableIDIntroducer interface {
func TableID(op ops.Operator) (result semantics.TableSet) {
_ = rewrite.Visit(op, func(this ops.Operator) error {
if tbl, ok := this.(TableIDIntroducer); ok {
result.MergeInPlace(tbl.Introduces())
result = result.Merge(tbl.Introduces())
}
return nil
})
Expand Down
23 changes: 3 additions & 20 deletions go/vt/vtgate/planbuilder/operators/subquery_planning.go
Original file line number Diff line number Diff line change
Expand Up @@ -100,21 +100,6 @@ func unresolvedAndSource(ctx *plancontext.PlanningContext, op ops.Operator) ([]s

func mergeSubQueryOp(ctx *plancontext.PlanningContext, outer *Route, inner *Route, subq *SubQueryInner) (*Route, error) {
subq.ExtractedSubquery.NeedsRewrite = true

// go over the subquery and add its tables to the one's solved by the route it is merged with
// this is needed to so that later when we try to push projections, we get the correct
// solved tableID from the route, since it also includes the tables from the subquery after merging
err := sqlparser.Walk(func(node sqlparser.SQLNode) (kontinue bool, err error) {
switch n := node.(type) {
case *sqlparser.AliasedTableExpr:
ts := TableID(outer)
ts.MergeInPlace(ctx.SemTable.TableSetFor(n))
}
return true, nil
}, subq.ExtractedSubquery.Subquery)
if err != nil {
return nil, err
}
outer.SysTableTableSchema = append(outer.SysTableTableSchema, inner.SysTableTableSchema...)
for k, v := range inner.SysTableTableName {
if outer.SysTableTableName == nil {
Expand Down Expand Up @@ -142,7 +127,7 @@ func mergeSubQueryOp(ctx *plancontext.PlanningContext, outer *Route, inner *Rout
}
}

err = outer.resetRoutingSelections(ctx)
err := outer.resetRoutingSelections(ctx)
if err != nil {
return nil, err
}
Expand Down Expand Up @@ -374,11 +359,9 @@ func rewriteColumnsInSubqueryOpForJoin(

// update the dependencies for the subquery by removing the dependencies from the innerOp
tableSet := ctx.SemTable.Direct[subQueryInner.ExtractedSubquery.Subquery]
tableSet.RemoveInPlace(TableID(resultInnerOp))
ctx.SemTable.Direct[subQueryInner.ExtractedSubquery.Subquery] = tableSet
ctx.SemTable.Direct[subQueryInner.ExtractedSubquery.Subquery] = tableSet.Remove(TableID(resultInnerOp))
tableSet = ctx.SemTable.Recursive[subQueryInner.ExtractedSubquery.Subquery]
tableSet.RemoveInPlace(TableID(resultInnerOp))
ctx.SemTable.Recursive[subQueryInner.ExtractedSubquery.Subquery] = tableSet
ctx.SemTable.Recursive[subQueryInner.ExtractedSubquery.Subquery] = tableSet.Remove(TableID(resultInnerOp))

// return any error while rewriting
return resultInnerOp, rewriteError
Expand Down
1 change: 1 addition & 0 deletions go/vt/vtgate/planbuilder/plan_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -242,6 +242,7 @@ func TestPlan(t *testing.T) {
testFile(t, "use_cases.json", testOutputTempDir, vschemaWrapper, false)
testFile(t, "set_cases.json", testOutputTempDir, vschemaWrapper, false)
testFile(t, "union_cases.json", testOutputTempDir, vschemaWrapper, false)
testFile(t, "large_union_cases.json", testOutputTempDir, vschemaWrapper, false)
testFile(t, "transaction_cases.json", testOutputTempDir, vschemaWrapper, false)
testFile(t, "lock_cases.json", testOutputTempDir, vschemaWrapper, false)
testFile(t, "large_cases.json", testOutputTempDir, vschemaWrapper, false)
Expand Down
2 changes: 1 addition & 1 deletion go/vt/vtgate/planbuilder/projection_pushing.go
Original file line number Diff line number Diff line change
Expand Up @@ -76,7 +76,7 @@ func pushProjectionIntoConcatenate(ctx *plancontext.PlanningContext, expr *sqlpa
if err != nil {
return 0, false, err
}
if added && ctx.SemTable.DirectDeps(expr.Expr).NumberOfTables() > 0 {
if added && ctx.SemTable.DirectDeps(expr.Expr).NonEmpty() {
return 0, false, vterrors.Errorf(vtrpcpb.Code_INTERNAL, "pushing projection %v on concatenate should reference an existing column", sqlparser.String(expr))
}
if added {
Expand Down
2,592 changes: 2,592 additions & 0 deletions go/vt/vtgate/planbuilder/testdata/large_union_cases.json

Large diffs are not rendered by default.

14 changes: 6 additions & 8 deletions go/vt/vtgate/semantics/binder.go
Original file line number Diff line number Diff line change
Expand Up @@ -94,7 +94,7 @@ func (b *binder) up(cursor *sqlparser.Cursor) error {
currentScope := b.scoper.currentScope()
deps, err := b.resolveColumn(node, currentScope, false)
if err != nil {
if deps.direct.NumberOfTables() == 0 ||
if deps.direct.IsEmpty() ||
!strings.HasSuffix(err.Error(), "is ambiguous") ||
!b.canRewriteUsingJoin(deps, node) {
return err
Expand Down Expand Up @@ -127,14 +127,14 @@ func (b *binder) bindCountStar(node *sqlparser.CountStar) {
case *vTableInfo:
for _, col := range tbl.cols {
if sqlparser.EqualsExpr(node, col) {
ts.MergeInPlace(b.recursive[col])
ts = ts.Merge(b.recursive[col])
}
}
default:
expr := tbl.getExpr()
if expr != nil {
setFor := b.tc.tableSetFor(expr)
ts.MergeInPlace(setFor)
ts = ts.Merge(setFor)
}
}
}
Expand Down Expand Up @@ -196,15 +196,13 @@ func (b *binder) setSubQueryDependencies(subq *sqlparser.Subquery, currScope *sc
sco := currScope
for sco != nil {
for _, table := range sco.tables {
tablesToKeep.MergeInPlace(table.getTableSet(b.org))
tablesToKeep = tablesToKeep.Merge(table.getTableSet(b.org))
}
sco = sco.parent
}

subqDirectDeps.KeepOnly(tablesToKeep)
subqRecursiveDeps.KeepOnly(tablesToKeep)
b.recursive[subq] = subqRecursiveDeps
b.direct[subq] = subqDirectDeps
b.recursive[subq] = subqRecursiveDeps.KeepOnly(tablesToKeep)
b.direct[subq] = subqDirectDeps.KeepOnly(tablesToKeep)
}

func (b *binder) createExtractedSubquery(cursor *sqlparser.Cursor, currScope *scope, subq *sqlparser.Subquery) (*sqlparser.ExtractedSubquery, error) {
Expand Down
241 changes: 241 additions & 0 deletions go/vt/vtgate/semantics/bitset/bitset.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,241 @@
/*
Copyright 2022 The Vitess Authors.

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 bitset

import (
"math/bits"
"unsafe"
)

// A Bitset is an immutable collection of bits. You can perform logical operations
// on it, but all mutable operations return a new Bitset.
// It is safe to compare directly using the comparison operator and to use as a map key.
type Bitset string

const bitsetWidth = 8

func bitsetWordSize(max int) int {
return max/bitsetWidth + 1
}

func toBitset(words []byte) Bitset {
if len(words) == 0 {
return ""
}
return *(*Bitset)(unsafe.Pointer(&words))
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Does this get the memory address from go runtime?

}

func minlen(a, b Bitset) int {
if len(a) < len(b) {
return len(a)
}
return len(b)
}

// Overlaps returns whether this Bitset and the input have any bits in common
func (bs Bitset) Overlaps(b2 Bitset) bool {
min := minlen(bs, b2)
for i := 0; i < min; i++ {
if bs[i]&b2[i] != 0 {
return true
}
}
return false
}

// Or returns the logical OR of the two Bitsets as a new Bitset
func (bs Bitset) Or(b2 Bitset) Bitset {
if len(bs) == 0 {
return b2
}
if len(b2) == 0 {
return bs
}

small, large := bs, b2
if len(small) > len(large) {
small, large = large, small
}

merged := make([]byte, len(large))
m := 0

for m < len(small) {
merged[m] = small[m] | large[m]
m++
}
for m < len(large) {
merged[m] = large[m]
m++
}
return toBitset(merged)
}

// AndNot returns the logical AND NOT of the two Bitsets as a new Bitset
func (bs Bitset) AndNot(b2 Bitset) Bitset {
if len(b2) == 0 {
return bs
}

merged := make([]byte, len(bs))
m := 0

for m = 0; m < len(bs); m++ {
if m < len(b2) {
merged[m] = bs[m] & ^b2[m]
} else {
merged[m] = bs[m]
}
}
for ; m > 0; m-- {
if merged[m-1] != 0 {
break
}
}
return toBitset(merged[:m])
}

// And returns the logical AND of the two bitsets as a new Bitset
func (bs Bitset) And(b2 Bitset) Bitset {
if len(bs) == 0 || len(b2) == 0 {
return ""
}

merged := make([]byte, minlen(bs, b2))
m := 0

for m = 0; m < len(merged); m++ {
merged[m] = bs[m] & b2[m]
}
for ; m > 0; m-- {
if merged[m-1] != 0 {
break
}
}
return toBitset(merged[:m])
}

// Set returns a copy of this Bitset where the bit at `offset` is set
func (bs Bitset) Set(offset int) Bitset {
alloc := len(bs)
if max := bitsetWordSize(offset); max > alloc {
alloc = max
}

words := make([]byte, alloc)
copy(words, bs)
words[offset/bitsetWidth] |= 1 << (offset % bitsetWidth)
return toBitset(words)
}

// SingleBit returns the position of the single bit that is set in this Bitset
// If the Bitset is empty, or contains more than one set bit, it returns -1
func (bs Bitset) SingleBit() int {
offset := -1
for i := 0; i < len(bs); i++ {
t := bs[i]
if t == 0 {
continue
}
if offset >= 0 || bits.OnesCount8(t) != 1 {
return -1
}
offset = i*bitsetWidth + bits.TrailingZeros8(t)
}
return offset
}

// IsContainedBy returns whether this Bitset is contained by the given Bitset
func (bs Bitset) IsContainedBy(b2 Bitset) bool {
if len(bs) > len(b2) {
return false
}
for i := 0; i < len(bs); i++ {
left := bs[i]
rigt := b2[i]
if left&rigt != left {
return false
}
}
return true
}

// Popcount returns the number of bits that are set in this Bitset
func (bs Bitset) Popcount() (count int) {
for i := 0; i < len(bs); i++ {
count += bits.OnesCount8(bs[i])
}
return
}

// ForEach calls the given callback with the position of each bit set in this Bitset
func (bs Bitset) ForEach(yield func(int)) {
for i := 0; i < len(bs); i++ {
bitset := bs[i]
for bitset != 0 {
t := bitset & -bitset
r := bits.TrailingZeros8(bitset)
yield(i*bitsetWidth + r)
bitset ^= t
}
}
}

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

an example of how this works will help. A general example for overall implementation will also work.

// Build creates a new immutable Bitset where all the given bits are set
func Build(bits ...int) Bitset {
if len(bits) == 0 {
return ""
}

max := bits[0]
for _, b := range bits[1:] {
if b > max {
max = b
}
}

words := make([]byte, bitsetWordSize(max))
for _, b := range bits {
words[b/bitsetWidth] |= 1 << (b % bitsetWidth)
}
return toBitset(words)
}

const singleton = "\x00\x00\x00\x01\x00\x00\x00\x02\x00\x00\x00\x04\x00\x00\x00\x08\x00\x00\x00\x10\x00\x00\x00\x20\x00\x00\x00\x40\x00\x00\x00\x80"

// Single returns a new Bitset where only the given bit is set.
// If the given bit is less than 32, Single does not allocate to create a new Bitset.
func Single(bit int) Bitset {
Comment on lines +231 to +235
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

what is the limit to the number of tableSet that can exist now?

Copy link
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

There is no limit! Any bitset larger than 32 will allocate, but you can make it as large as you want.

switch {
case bit < 8:
bit = (bit + 1) << 2
return Bitset(singleton[bit-1 : bit])
case bit < 16:
bit = (bit + 1 - 8) << 2
return Bitset(singleton[bit-2 : bit])
case bit < 24:
bit = (bit + 1 - 16) << 2
return Bitset(singleton[bit-3 : bit])
case bit < 32:
bit = (bit + 1 - 24) << 2
return Bitset(singleton[bit-4 : bit])
default:
words := make([]byte, bitsetWordSize(bit))
words[bit/bitsetWidth] |= 1 << (bit % bitsetWidth)
return toBitset(words)
}
}
Loading