syncbase: Enable watching a set of glob patterns.
Watch API was changed to accept one or more CollectionRowPattern
structures, with separate SQL LIKE-style glob patterns for collection
blessing, name, and row key. Helper function allows easily watching
a single prefix (as in old functionality).
Watch method no longer returns an error; the error can be checked on the
returned stream instead, same as in Scan.
Refactored LIKE pattern parsing out of internal query packages into a
separate package.
Collection and row filtering refactored into separate package, can be
used for Scan and other RPCs in the future. Implementation is currently
inefficient, scanning over all collections and over all rows in matched
collections.
MultiPart: 1/4
Change-Id: Ia2c9f4311a3c728b58b3f3cea8d9c1fe97e4c800
diff --git a/query/engine/internal/conversions/conversions.go b/query/engine/internal/conversions/conversions.go
index 24f00de..0b8ecd9 100644
--- a/query/engine/internal/conversions/conversions.go
+++ b/query/engine/internal/conversions/conversions.go
@@ -20,8 +20,8 @@
switch o.Type {
case query_parser.TypStr:
c.Str = o.Str
- c.Regex = o.Regex // non-empty for rhs of like expressions
- c.CompRegex = o.CompRegex // non-nil for rhs of like expressions
+ c.Prefix = o.Prefix // non-nil for rhs of like expressions
+ c.Pattern = o.Pattern // non-nil for rhs of like expressions
default:
return nil, errors.New("Cannot convert operand to string.")
}
diff --git a/query/engine/internal/eval.go b/query/engine/internal/eval.go
index 2338bbe..b2a628d 100644
--- a/query/engine/internal/eval.go
+++ b/query/engine/internal/eval.go
@@ -288,9 +288,9 @@
case query_parser.GreaterThanOrEqual:
return lhsValue.Str >= rhsValue.Str
case query_parser.Like:
- return rhsValue.CompRegex.MatchString(lhsValue.Str)
+ return rhsValue.Pattern.MatchString(lhsValue.Str)
case query_parser.NotLike:
- return !rhsValue.CompRegex.MatchString(lhsValue.Str)
+ return !rhsValue.Pattern.MatchString(lhsValue.Str)
default:
// TODO(jkline): Log this logic error and all other similar cases.
return false
diff --git a/query/engine/internal/query_checker/query_checker.go b/query/engine/internal/query_checker/query_checker.go
index a4bf7b1..d3bce52 100644
--- a/query/engine/internal/query_checker/query_checker.go
+++ b/query/engine/internal/query_checker/query_checker.go
@@ -5,14 +5,12 @@
package query_checker
import (
- "bytes"
- "regexp"
"sort"
- "strings"
ds "v.io/v23/query/engine/datasource"
"v.io/v23/query/engine/internal/query_functions"
"v.io/v23/query/engine/internal/query_parser"
+ "v.io/v23/query/pattern"
"v.io/v23/query/syncql"
"v.io/v23/vdl"
)
@@ -136,19 +134,15 @@
if e.Operand2.Type != query_parser.TypStr {
return syncql.NewErrLikeExpressionsRequireRhsString(db.GetContext(), e.Operand2.Off)
}
- prefix, err := computePrefix(db, e.Operand2.Off, e.Operand2.Str, ec)
+ // Compile the like pattern now to to check for errors.
+ p, err := parseLikePattern(db, e.Operand2.Off, e.Operand2.Str, ec)
if err != nil {
return err
}
- e.Operand2.Prefix = prefix
- // Compute the regular expression now to to check for errors.
- // Save the regex (testing) and the compiled regex (for later use in evaluation).
- regex, compRegex, foundWildcard, err := computeRegex(db, e.Operand2.Off, e.Operand2.Str, ec)
- if err != nil {
- return err
- }
+ fixedPrefix, noWildcards := p.FixedPrefix()
+ e.Operand2.Prefix = fixedPrefix
// Optimization: If like/not like argument contains no wildcards, convert the expression to equals/not equals.
- if !foundWildcard {
+ if noWildcards {
if e.Operator.Type == query_parser.Like {
e.Operator.Type = query_parser.Equal
} else { // not like
@@ -156,10 +150,10 @@
}
// Since this is no longer a like expression, we need to unescape
// any escaped chars.
- e.Operand2.Str = unescapeLikeExpression(e.Operand2.Str, ec)
+ e.Operand2.Str = fixedPrefix
}
- e.Operand2.Regex = regex
- e.Operand2.CompRegex = compRegex
+ // Save the compiled pattern for later use in evaluation.
+ e.Operand2.Pattern = p
}
// Is/IsNot expressions require operand1 to be a (value or function) and operand2 to be nil.
@@ -239,137 +233,16 @@
return nil
}
-// Only include up to (but not including) a wildcard character ('%', '_').
-func computePrefix(db ds.Database, off int64, s string, ec *query_parser.EscapeClause) (string, error) {
- if strings.Index(s, "%") == -1 && strings.Index(s, "_") == -1 && (ec == nil || strings.IndexRune(s, ec.EscapeChar.Value) == -1) {
- return s, nil
- }
- var s2 string
- escapedChar := false
- for _, c := range s {
- if escapedChar {
- switch c {
- case '%':
- s2 += string(c)
- case '_':
- s2 += string(c)
- default:
- return "", syncql.NewErrInvalidEscapeSequence(db.GetContext(), off)
- }
- escapedChar = false
- } else {
- // Hit an unescaped wildcard, we are done
- if c == '%' || c == '_' {
- return s2, nil
- } else if ec != nil && c == ec.EscapeChar.Value {
- escapedChar = true
- } else {
- s2 += string(c)
- }
- }
- }
- if escapedChar {
- return "", syncql.NewErrInvalidEscapeSequence(db.GetContext(), off)
- }
- return s2, nil
-}
-
-// Convert Like expression to a regex. That is, convert:
-// % to .*?
-// _ to .
-// Escape everything that would be incorrectly interpreted as a regex.
-//
-// The approach this function takes is to collect characters to be escaped
-// into toBeEscapedBuf. When a wildcard is encountered, first toBeEscapedBuf
-// is escaped and written to the regex buffer, next the wildcard is translated
-// to regex (either ".*?" or ".") and written to the regex buffer.
-// At the end, any remaining chars in toBeEscapedBuf are written.
-//
-// Return values are:
-// 1. string: uncompiled regular expression
-// 2. *Regexp: compiled regular expression
-// 3. bool: true if wildcards were found (if false, like is converted to equal)
-// 4. error: non-nil if error encountered
-func computeRegex(db ds.Database, off int64, s string, ec *query_parser.EscapeClause) (string, *regexp.Regexp, bool, error) {
- var buf bytes.Buffer // buffer for return regex
- var toBeEscapedBuf bytes.Buffer // buffer to hold characters waiting to be escaped
-
- buf.WriteString("^") // '^<regex_str>$'
- escapedMode := false
- foundWildcard := false
-
- escChar := ' ' // blank will be ignored as an escape char
+func parseLikePattern(db ds.Database, off int64, s string, ec *query_parser.EscapeClause) (*pattern.Pattern, error) {
+ escChar := '\x00' // nul is ignored as an escape char
if ec != nil {
escChar = ec.EscapeChar.Value
}
-
- for _, c := range s {
- switch c {
- case '%', '_':
- if escapedMode {
- toBeEscapedBuf.WriteString(string(c))
- } else {
- // Write out any chars waiting to be escaped, then
- // write ".*?' or '.'.
- buf.WriteString(regexp.QuoteMeta(toBeEscapedBuf.String()))
- toBeEscapedBuf.Reset()
- if c == '%' {
- buf.WriteString(".*?")
- } else {
- buf.WriteString(".")
- }
- foundWildcard = true
- }
- escapedMode = false
- case escChar:
- if escChar != ' ' {
- if escapedMode {
- toBeEscapedBuf.WriteString(string(c))
- }
- escapedMode = !escapedMode
- } else {
- // not an escape char, treat same as default
- toBeEscapedBuf.WriteString(string(c))
- }
- default:
- toBeEscapedBuf.WriteString(string(c))
- }
- }
- // Write any remaining chars in toBeEscapedBuf.
- buf.WriteString(regexp.QuoteMeta(toBeEscapedBuf.String()))
- buf.WriteString("$") // '^<regex_str>$'
- regex := buf.String()
- compRegex, err := regexp.Compile(regex)
+ p, err := pattern.ParseWithEscapeChar(s, escChar)
if err != nil {
- return "", nil, false, syncql.NewErrErrorCompilingRegularExpression(db.GetContext(), off, regex, err)
+ return nil, syncql.NewErrInvalidLikePattern(db.GetContext(), off, err)
}
- return regex, compRegex, foundWildcard, nil
-}
-
-// Unescape any escaped % and _ chars as this is being converted to an equals expression.
-func unescapeLikeExpression(s string, ec *query_parser.EscapeClause) string {
- var buf bytes.Buffer // buffer for returned unescaped string
-
- if ec == nil {
- // there is nothing to unescape
- return s
- }
-
- escapedMode := false
-
- for _, c := range s {
- switch c {
- case ec.EscapeChar.Value:
- if escapedMode {
- buf.WriteString(string(c))
- }
- escapedMode = !escapedMode
- default:
- buf.WriteString(string(c))
- escapedMode = false
- }
- }
- return buf.String()
+ return p, nil
}
func IsLogicalOperator(o *query_parser.BinaryOperator) bool {
@@ -748,15 +621,15 @@
sort.Sort(*fieldRanges)
}
-// Check escape clause. Escape char cannot be '\'.
+// Check escape clause. Escape char cannot be '\', ' ', or a wildcard.
// Return bool (true if escape char defined), escape char, error.
func checkEscapeClause(db ds.Database, e *query_parser.EscapeClause) error {
if e == nil {
return nil
}
- switch e.EscapeChar.Value {
- case ' ', '\\':
- return syncql.NewErrInvalidEscapeChar(db.GetContext(), e.EscapeChar.Off)
+ switch ec := e.EscapeChar.Value; ec {
+ case '\x00', '_', '%', ' ', '\\':
+ return syncql.NewErrInvalidEscapeChar(db.GetContext(), e.EscapeChar.Off, string(ec))
default:
return nil
}
diff --git a/query/engine/internal/query_checker/query_checker_test.go b/query/engine/internal/query_checker/query_checker_test.go
index 045a0c5..4da899a 100644
--- a/query/engine/internal/query_checker/query_checker_test.go
+++ b/query/engine/internal/query_checker/query_checker_test.go
@@ -15,6 +15,7 @@
ds "v.io/v23/query/engine/datasource"
"v.io/v23/query/engine/internal/query_checker"
"v.io/v23/query/engine/internal/query_parser"
+ "v.io/v23/query/pattern"
"v.io/v23/query/syncql"
"v.io/v23/vdl"
"v.io/v23/verror"
@@ -94,9 +95,8 @@
indexRanges []*ds.IndexRanges
}
-type regularExpressionsTest struct {
+type likePatternsTest struct {
query string
- regex string
matches []string
nonMatches []string
}
@@ -440,7 +440,7 @@
},
},
{
- // Note: 'like "Foo"' is optimized to '= "Foo"
+ // Note: 'like "Foo"' is optimized to '= "Foo"'
"select k, v from Customer where k = \"Foo.Bar\" or k like \"Foo\" or k like \"abc%\" limit 100 offset 200",
&ds.IndexRanges{
FieldName: "k",
@@ -679,7 +679,7 @@
},
},
{
- // Note: 'like "Foo"' is optimized to '= "Foo"
+ // Note: 'like "Foo"' is optimized to '= "Foo"'
"delete from Customer where k = \"Foo.Bar\" or k like \"Foo\" or k like \"abc%\" limit 100",
&ds.IndexRanges{
FieldName: "k",
@@ -1072,7 +1072,7 @@
},
},
{
- // Note: 'like "Foo"' is optimized to '= "Foo"
+ // Note: 'like "Foo"' is optimized to '= "Foo"'
"select k, v from Customer where v.Foo = \"Foo.Bar\" or v.Foo like \"Foo\" or v.Foo like \"abc%\" limit 100 offset 200",
[]string{"v.Foo"},
[]*ds.IndexRanges{
@@ -1450,7 +1450,7 @@
},
},
{
- // Note: 'like "Foo"' is optimized to '= "Foo"
+ // Note: 'like "Foo"' is optimized to '= "Foo"'
"delete from Customer where v.Foo = \"Foo.Bar\" or v.Foo like \"Foo\" or v.Foo like \"abc%\" limit 100",
[]string{"v.Foo"},
[]*ds.IndexRanges{
@@ -1566,103 +1566,87 @@
}
}
-func TestRegularExpressions(t *testing.T) {
- basic := []regularExpressionsTest{
+func TestLikePatterns(t *testing.T) {
+ basic := []likePatternsTest{
// Select
{
"select v from Customer where v like \"abc%\"",
- "^abc.*?$",
[]string{"abc", "abcd", "abcabc"},
[]string{"xabcd"},
},
{
"select v from Customer where v like \"abc_\"",
- "^abc.$",
[]string{"abcd", "abc1"},
[]string{"abc", "xabcd", "abcde"},
},
{
"select v from Customer where v like \"*%*_%\" escape '*'",
- "^%_.*?$",
[]string{"%_", "%_abc"},
[]string{"%a", "abc%_abc"},
},
{
"select v from Customer where v like \"abc^_\" escape '^'",
- "^abc_$",
[]string{"abc_"},
[]string{"abc", "xabcd", "abcde"},
},
{
"select v from Customer where v like \"abc^%\" escape '^'",
- "^abc%$",
[]string{"abc%"},
[]string{"abc", "xabcd", "abcde"},
},
{
"select v from Customer where v like \"abc_efg\"",
- "^abc.efg$",
[]string{"abcdefg"},
[]string{"abc", "xabcd", "abcde", "abcdefgh"},
},
{
"select v from Customer where v like \"abc%def\"",
- "^abc.*?def$",
[]string{"abcdefdef", "abcdef", "abcdefghidef"},
[]string{"abcdefg", "abcdefde"},
},
{
"select v from Customer where v like \"[0-9]*abc%def\"",
- "^\\[0-9\\]\\*abc.*?def$",
[]string{"[0-9]*abcdefdef", "[0-9]*abcdef", "[0-9]*abcdefghidef"},
[]string{"0abcdefg", "9abcdefde", "[0-9]abcdefg", "[0-9]abcdefg", "[0-9]abcdefg"},
},
// Delete
{
"delete from Customer where v like \"abc%\"",
- "^abc.*?$",
[]string{"abc", "abcd", "abcabc"},
[]string{"xabcd"},
},
{
"delete from Customer where v like \"abc_\"",
- "^abc.$",
[]string{"abcd", "abc1"},
[]string{"abc", "xabcd", "abcde"},
},
{
"delete from Customer where v like \"*%*_%\" escape '*'",
- "^%_.*?$",
[]string{"%_", "%_abc"},
[]string{"%a", "abc%_abc"},
},
{
"delete from Customer where v like \"abc^_\" escape '^'",
- "^abc_$",
[]string{"abc_"},
[]string{"abc", "xabcd", "abcde"},
},
{
"delete from Customer where v like \"abc^%\" escape '^'",
- "^abc%$",
[]string{"abc%"},
[]string{"abc", "xabcd", "abcde"},
},
{
"delete from Customer where v like \"abc_efg\"",
- "^abc.efg$",
[]string{"abcdefg"},
[]string{"abc", "xabcd", "abcde", "abcdefgh"},
},
{
"delete from Customer where v like \"abc%def\"",
- "^abc.*?def$",
[]string{"abcdefdef", "abcdef", "abcdefghidef"},
[]string{"abcdefg", "abcdefde"},
},
{
"delete from Customer where v like \"[0-9]*abc%def\"",
- "^\\[0-9\\]\\*abc.*?def$",
[]string{"[0-9]*abcdefdef", "[0-9]*abcdef", "[0-9]*abcdefghidef"},
[]string{"0abcdefg", "9abcdefde", "[0-9]abcdefg", "[0-9]abcdefg", "[0-9]abcdefg"},
},
@@ -1680,39 +1664,33 @@
switch sel := (*s).(type) {
case query_parser.SelectStatement:
// We know there is exactly one like expression and operand2 contains
- // a regex and compiled regex.
- if sel.Where.Expr.Operand2.Regex != test.regex {
- t.Errorf("query: %s;\nGOT %s\nWANT %s", test.query, sel.Where.Expr.Operand2.Regex, test.regex)
- }
- regexp := sel.Where.Expr.Operand2.CompRegex
- // Make sure all matches actually match
+ // a prefix and compiled pattern.
+ p := sel.Where.Expr.Operand2.Pattern
+ // Make sure all matches actually match.
for _, m := range test.matches {
- if !regexp.MatchString(m) {
+ if !p.MatchString(m) {
t.Errorf("query: %s;Expected match: %s; \nGOT false\nWANT true", test.query, m)
}
}
- // Make sure all nonMatches actually don't match
+ // Make sure all nonMatches actually don't match.
for _, n := range test.nonMatches {
- if regexp.MatchString(n) {
+ if p.MatchString(n) {
t.Errorf("query: %s;Expected nonMatch: %s; \nGOT true\nWANT false", test.query, n)
}
}
case query_parser.DeleteStatement:
// We know there is exactly one like expression and operand2 contains
- // a regex and compiled regex.
- if sel.Where.Expr.Operand2.Regex != test.regex {
- t.Errorf("query: %s;\nGOT %s\nWANT %s", test.query, sel.Where.Expr.Operand2.Regex, test.regex)
- }
- regexp := sel.Where.Expr.Operand2.CompRegex
- // Make sure all matches actually match
+ // a prefix and compiled pattern.
+ p := sel.Where.Expr.Operand2.Pattern
+ // Make sure all matches actually match.
for _, m := range test.matches {
- if !regexp.MatchString(m) {
+ if !p.MatchString(m) {
t.Errorf("query: %s;Expected match: %s; \nGOT false\nWANT true", test.query, m)
}
}
- // Make sure all nonMatches actually don't match
+ // Make sure all nonMatches actually don't match.
for _, n := range test.nonMatches {
- if regexp.MatchString(n) {
+ if p.MatchString(n) {
t.Errorf("query: %s;Expected nonMatch: %s; \nGOT true\nWANT false", test.query, n)
}
}
@@ -1730,7 +1708,7 @@
{"select v from Customer where a=1", syncql.NewErrBadFieldInWhere(db.GetContext(), 29)},
{"select v from Customer limit 0", syncql.NewErrLimitMustBeGt0(db.GetContext(), 29)},
{"select v.z from Customer where v.x like v.y", syncql.NewErrLikeExpressionsRequireRhsString(db.GetContext(), 40)},
- {"select v.z from Customer where k like \"a^bc%\" escape '^'", syncql.NewErrInvalidEscapeSequence(db.GetContext(), 38)},
+ {"select v.z from Customer where k like \"a^bc%\" escape '^'", syncql.NewErrInvalidLikePattern(db.GetContext(), 38, pattern.NewErrInvalidEscape(nil, "b"))},
{"select v from Customer where v.A > false", syncql.NewErrBoolInvalidExpression(db.GetContext(), 33)},
{"select v from Customer where true <= v.A", syncql.NewErrBoolInvalidExpression(db.GetContext(), 34)},
{"select v from Customer where Foo(\"2015/07/22\", true, 3.14157) = true", syncql.NewErrFunctionNotFound(db.GetContext(), 29, "Foo")},
@@ -1768,7 +1746,7 @@
{"delete from Customer where a=1", syncql.NewErrBadFieldInWhere(db.GetContext(), 27)},
{"delete from Customer limit 0", syncql.NewErrLimitMustBeGt0(db.GetContext(), 27)},
{"delete from Customer where v.x like v.y", syncql.NewErrLikeExpressionsRequireRhsString(db.GetContext(), 36)},
- {"delete from Customer where k like \"a^bc%\" escape '^'", syncql.NewErrInvalidEscapeSequence(db.GetContext(), 34)},
+ {"delete from Customer where k like \"a^bc%\" escape '^'", syncql.NewErrInvalidLikePattern(db.GetContext(), 34, pattern.NewErrInvalidEscape(nil, "b"))},
{"delete from Customer where v.A > false", syncql.NewErrBoolInvalidExpression(db.GetContext(), 31)},
{"delete from Customer where true <= v.A", syncql.NewErrBoolInvalidExpression(db.GetContext(), 32)},
{"delete from Customer where Foo(\"2015/07/22\", true, 3.14157) = true", syncql.NewErrFunctionNotFound(db.GetContext(), 27, "Foo")},
diff --git a/query/engine/internal/query_parser/query_parser.go b/query/engine/internal/query_parser/query_parser.go
index 2b242b5..388739c 100644
--- a/query/engine/internal/query_parser/query_parser.go
+++ b/query/engine/internal/query_parser/query_parser.go
@@ -7,7 +7,6 @@
import (
"fmt"
"math/big"
- "regexp"
"strconv"
"strings"
"text/scanner"
@@ -15,6 +14,7 @@
"unicode/utf8"
ds "v.io/v23/query/engine/datasource"
+ "v.io/v23/query/pattern"
"v.io/v23/query/syncql"
"v.io/v23/vdl"
)
@@ -110,27 +110,26 @@
TypParameter
TypStr
TypTime
- TypObject // Only as the result of a ResolveOperand.
- TypUint // Only as a result of a ResolveOperand
+ TypObject // Only as the result of a ResolveOperand
+ TypUint // Only as the result of a ResolveOperand
)
type Operand struct {
- Type OperandType
- BigInt *big.Int
- BigRat *big.Rat
- Bool bool
- Column *Field
- Float float64
- Function *Function
- Int int64
- Str string
- Time time.Time
- Prefix string // Computed by checker for Like expressions
- Regex string // Computed by checker for Like expressions
- Uint uint64
- CompRegex *regexp.Regexp
- Expr *Expression
- Object *vdl.Value
+ Type OperandType
+ BigInt *big.Int
+ BigRat *big.Rat
+ Bool bool
+ Column *Field
+ Float float64
+ Function *Function
+ Int int64
+ Str string
+ Time time.Time
+ Prefix string // Computed by checker for Like expressions
+ Pattern *pattern.Pattern // Computed by checker for Like expressions
+ Uint uint64
+ Expr *Expression
+ Object *vdl.Value
Node
}
diff --git a/query/engine/internal/query_test.go b/query/engine/internal/query_test.go
index 09e7592..0e44958 100644
--- a/query/engine/internal/query_test.go
+++ b/query/engine/internal/query_test.go
@@ -3349,7 +3349,7 @@
},
{
"select v from Customer where k like \"abc %\" escape ' '",
- syncql.NewErrInvalidEscapeChar(db.GetContext(), 51),
+ syncql.NewErrInvalidEscapeChar(db.GetContext(), 51, string(' ')),
},
}
diff --git a/query/engine/query_test.go b/query/engine/query_test.go
index bb06864..9537340 100644
--- a/query/engine/query_test.go
+++ b/query/engine/query_test.go
@@ -3166,7 +3166,7 @@
},
{
"select v from Customer where k like \"abc %\" escape ' '",
- syncql.NewErrInvalidEscapeChar(db.GetContext(), 51),
+ syncql.NewErrInvalidEscapeChar(db.GetContext(), 51, string(' ')),
},
{
"select v.Foo from",
diff --git a/query/pattern/.api b/query/pattern/.api
new file mode 100644
index 0000000..0812d41
--- /dev/null
+++ b/query/pattern/.api
@@ -0,0 +1,12 @@
+pkg pattern, const DefaultEscapeChar ideal-char
+pkg pattern, func Escape(string) string
+pkg pattern, func EscapeWithEscapeChar(string, rune) string
+pkg pattern, func NewErrIllegalEscapeChar(*context.T) error
+pkg pattern, func NewErrInvalidEscape(*context.T, string) error
+pkg pattern, func Parse(string) (*Pattern, error)
+pkg pattern, func ParseWithEscapeChar(string, rune) (*Pattern, error)
+pkg pattern, method (*Pattern) FixedPrefix() (string, bool)
+pkg pattern, method (*Pattern) MatchString(string) bool
+pkg pattern, type Pattern struct
+pkg pattern, var ErrIllegalEscapeChar unknown-type
+pkg pattern, var ErrInvalidEscape unknown-type
diff --git a/query/pattern/errors.vdl b/query/pattern/errors.vdl
new file mode 100644
index 0000000..56b7608
--- /dev/null
+++ b/query/pattern/errors.vdl
@@ -0,0 +1,14 @@
+// Copyright 2016 The Vanadium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style
+// license that can be found in the LICENSE file.
+
+package pattern
+
+error (
+ IllegalEscapeChar() {
+ "en": "'%' and '_' cannot be used as escape characters",
+ }
+ InvalidEscape(escaped string) {
+ "en": "only '%', '_', and the escape character are allowed to be escaped, found '\\\\{escaped}'",
+ }
+)
diff --git a/query/pattern/pattern.go b/query/pattern/pattern.go
new file mode 100644
index 0000000..8e5d7a7
--- /dev/null
+++ b/query/pattern/pattern.go
@@ -0,0 +1,161 @@
+// Copyright 2016 The Vanadium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style
+// license that can be found in the LICENSE file.
+
+// Package pattern handles parsing and matching SQL LIKE-style glob patterns.
+package pattern
+
+import (
+ "bytes"
+ "fmt"
+ "regexp"
+
+ "v.io/v23/verror"
+)
+
+const (
+ DefaultEscapeChar = '\\'
+)
+
+// Pattern is a parsed LIKE-style glob pattern.
+type Pattern struct {
+ // regular expression equivalent to the original like pattern
+ regex *regexp.Regexp
+ // fixed prefix that all pattern matches must start with
+ fixedPrefix string
+ // true if pattern contains no unescaped wildcards; in this case, fixedPrefix
+ // is the entire unescaped expression
+ noWildcards bool
+}
+
+// Parse parses a LIKE-style glob pattern assuming '\' as escape character.
+// See ParseWithEscapeChar().
+func Parse(pattern string) (*Pattern, error) {
+ return ParseWithEscapeChar(pattern, DefaultEscapeChar)
+}
+
+// ParseWithEscapeChar parses a LIKE-style glob pattern.
+// Supported wildcards are '_' (match any one character) and '%' (match zero or
+// more characters). They can be escaped by escChar; escChar can also escape
+// itself. '_' and '%' cannot be used as escChar; '\x00' escChar disables
+// escaping.
+func ParseWithEscapeChar(pattern string, escChar rune) (*Pattern, error) {
+ if escChar == '%' || escChar == '_' {
+ return nil, NewErrIllegalEscapeChar(nil)
+ }
+
+ // The LIKE-style pattern is converted to a regex, converting:
+ // % to .*?
+ // _ to .
+ // Everything else that would be incorrectly interpreted as a regex is escaped.
+ // The approach this function takes is to collect characters to be escaped
+ // into toBeEscapedBuf. When a wildcard is encountered, first toBeEscapedBuf
+ // is escaped and written to the regex buffer, next the wildcard is translated
+ // to regex (either ".*?" or ".") and written to the regex buffer.
+ // At the end, any remaining chars in toBeEscapedBuf are written.
+ var buf bytes.Buffer // buffer for return regex
+ var toBeEscapedBuf bytes.Buffer // buffer to hold characters waiting to be escaped
+ // Even though regexp.Regexp provides a LiteralPrefix() method, it doesn't
+ // always return the longest fixed prefix, so we save it while parsing.
+ var fixedPrefix string
+ foundWildcard := false
+
+ buf.WriteString("^") // '^<regex_str>$'
+ escapedMode := false
+ for _, c := range pattern {
+ if escapedMode {
+ switch c {
+ case '%', '_', escChar:
+ toBeEscapedBuf.WriteRune(c)
+ default:
+ return nil, NewErrInvalidEscape(nil, string(c))
+ }
+ escapedMode = false
+ } else {
+ switch c {
+ case '%', '_':
+ // Write out any chars waiting to be escaped, then write ".*?' or '.'.
+ buf.WriteString(regexp.QuoteMeta(toBeEscapedBuf.String()))
+ if !foundWildcard {
+ // First wildcard found, fixedPrefix is the pattern up to it.
+ fixedPrefix = toBeEscapedBuf.String()
+ foundWildcard = true
+ }
+ toBeEscapedBuf.Reset()
+ if c == '%' {
+ buf.WriteString(".*?")
+ } else {
+ buf.WriteString(".")
+ }
+ case escChar:
+ if escChar != '\x00' {
+ escapedMode = true
+ } else {
+ // nul is never an escape char, treat same as default.
+ toBeEscapedBuf.WriteRune(c)
+ }
+ default:
+ toBeEscapedBuf.WriteRune(c)
+ }
+ }
+ }
+ if escapedMode {
+ return nil, NewErrInvalidEscape(nil, "<end>")
+ }
+ // Write any remaining chars in toBeEscapedBuf.
+ buf.WriteString(regexp.QuoteMeta(toBeEscapedBuf.String()))
+ if !foundWildcard {
+ // No wildcard found, fixedPrefix is the entire pattern.
+ fixedPrefix = toBeEscapedBuf.String()
+ }
+ buf.WriteString("$") // '^<regex_str>$'
+
+ regex := buf.String()
+ compRegex, err := regexp.Compile(regex)
+ if err != nil {
+ // TODO(ivanpi): Should never happen. Panic here?
+ return nil, verror.New(verror.ErrInternal, nil, fmt.Sprintf("failed to compile pattern %q (regular expression %q): %v", pattern, regex, err))
+ }
+ return &Pattern{
+ regex: compRegex,
+ fixedPrefix: fixedPrefix,
+ noWildcards: !foundWildcard,
+ }, nil
+}
+
+// MatchString returns true iff the pattern matches the entire string.
+func (p *Pattern) MatchString(s string) bool {
+ return p.regex.MatchString(s)
+}
+
+// FixedPrefix returns the unescaped fixed prefix that all matching strings must
+// start with, and whether the prefix is the whole pattern.
+func (p *Pattern) FixedPrefix() (string, bool) {
+ return p.fixedPrefix, p.noWildcards
+}
+
+// Escape escapes a literal string for inclusion in a LIKE-style pattern
+// assuming '\' as escape character.
+// See EscapeWithEscapeChar().
+func Escape(s string) string {
+ return EscapeWithEscapeChar(s, DefaultEscapeChar)
+}
+
+// EscapeWithEscapeChar escapes a literal string for inclusion in a LIKE-style
+// pattern. It inserts escChar before each '_', '%', and escChar in the string.
+func EscapeWithEscapeChar(s string, escChar rune) string {
+ if escChar == '\x00' {
+ panic(verror.New(verror.ErrBadArg, nil, "'\x00' disables escaping, cannot be used in EscapeWithEscapeChar"))
+ }
+ if escChar == '%' || escChar == '_' {
+ panic(NewErrIllegalEscapeChar(nil))
+ }
+ var buf bytes.Buffer
+ for _, c := range s {
+ if c == '%' || c == '_' || c == escChar {
+ buf.WriteRune(escChar)
+ }
+ buf.WriteRune(c)
+ }
+ return buf.String()
+}
diff --git a/query/pattern/pattern.vdl.go b/query/pattern/pattern.vdl.go
new file mode 100644
index 0000000..f7732eb
--- /dev/null
+++ b/query/pattern/pattern.vdl.go
@@ -0,0 +1,62 @@
+// Copyright 2016 The Vanadium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style
+// license that can be found in the LICENSE file.
+
+// This file was auto-generated by the vanadium vdl tool.
+// Package: pattern
+
+package pattern
+
+import (
+ "v.io/v23/context"
+ "v.io/v23/i18n"
+ "v.io/v23/verror"
+)
+
+var _ = __VDLInit() // Must be first; see __VDLInit comments for details.
+
+//////////////////////////////////////////////////
+// Error definitions
+
+var (
+ ErrIllegalEscapeChar = verror.Register("v.io/v23/query/pattern.IllegalEscapeChar", verror.NoRetry, "{1:}{2:} '%' and '_' cannot be used as escape characters")
+ ErrInvalidEscape = verror.Register("v.io/v23/query/pattern.InvalidEscape", verror.NoRetry, "{1:}{2:} only '%', '_', and the escape character are allowed to be escaped, found '\\{3}'")
+)
+
+// NewErrIllegalEscapeChar returns an error with the ErrIllegalEscapeChar ID.
+func NewErrIllegalEscapeChar(ctx *context.T) error {
+ return verror.New(ErrIllegalEscapeChar, ctx)
+}
+
+// NewErrInvalidEscape returns an error with the ErrInvalidEscape ID.
+func NewErrInvalidEscape(ctx *context.T, escaped string) error {
+ return verror.New(ErrInvalidEscape, ctx, escaped)
+}
+
+var __VDLInitCalled bool
+
+// __VDLInit performs vdl initialization. It is safe to call multiple times.
+// If you have an init ordering issue, just insert the following line verbatim
+// into your source files in this package, right after the "package foo" clause:
+//
+// var _ = __VDLInit()
+//
+// The purpose of this function is to ensure that vdl initialization occurs in
+// the right order, and very early in the init sequence. In particular, vdl
+// registration and package variable initialization needs to occur before
+// functions like vdl.TypeOf will work properly.
+//
+// This function returns a dummy value, so that it can be used to initialize the
+// first var in the file, to take advantage of Go's defined init order.
+func __VDLInit() struct{} {
+ if __VDLInitCalled {
+ return struct{}{}
+ }
+ __VDLInitCalled = true
+
+ // Set error format strings.
+ i18n.Cat().SetWithBase(i18n.LangID("en"), i18n.MsgID(ErrIllegalEscapeChar.ID), "{1:}{2:} '%' and '_' cannot be used as escape characters")
+ i18n.Cat().SetWithBase(i18n.LangID("en"), i18n.MsgID(ErrInvalidEscape.ID), "{1:}{2:} only '%', '_', and the escape character are allowed to be escaped, found '\\{3}'")
+
+ return struct{}{}
+}
diff --git a/query/pattern/pattern_test.go b/query/pattern/pattern_test.go
new file mode 100644
index 0000000..d4c5fbc
--- /dev/null
+++ b/query/pattern/pattern_test.go
@@ -0,0 +1,202 @@
+// Copyright 2016 The Vanadium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style
+// license that can be found in the LICENSE file.
+
+package pattern
+
+import (
+ "testing"
+
+ "v.io/v23/verror"
+)
+
+type patternTest struct {
+ pattern string
+ escChar rune
+ wantRegex string
+ wantFixedPrefix string
+ wantNoWildcards bool
+ matches []string
+ nonMatches []string
+}
+
+func TestPatternParse(t *testing.T) {
+ tests := []patternTest{
+ {
+ "abc", '\\',
+ "^abc$",
+ "abc", true,
+ []string{"abc"},
+ []string{"a", "abcd", "xabc"},
+ },
+ {
+ "abc%", '\\',
+ "^abc.*?$",
+ "abc", false,
+ []string{"abc", "abcd", "abcabc"},
+ []string{"xabcd"},
+ },
+ {
+ "abc_", '\\',
+ "^abc.$",
+ "abc", false,
+ []string{"abcd", "abc1"},
+ []string{"abc", "xabcd", "abcde"},
+ },
+ {
+ "*%*_%", '*',
+ "^%_.*?$",
+ "%_", false,
+ []string{"%_", "%_abc"},
+ []string{"%a", "abc%_abc"},
+ },
+ {
+ "abc^_", '^',
+ "^abc_$",
+ "abc_", true,
+ []string{"abc_"},
+ []string{"abc", "xabcd", "abcde"},
+ },
+ {
+ "abc^%", '^',
+ "^abc%$",
+ "abc%", true,
+ []string{"abc%"},
+ []string{"abc", "xabcd", "abcde"},
+ },
+ {
+ "abc^^%", '^',
+ "^abc\\^.*?$",
+ "abc^", false,
+ []string{"abc^", "abc^de", "abc^%"},
+ []string{"abc", "abcd", "xabc^d"},
+ },
+ {
+ "abc_efg", '\\',
+ "^abc.efg$",
+ "abc", false,
+ []string{"abcdefg"},
+ []string{"abc", "xabcd", "abcde", "abcdefgh"},
+ },
+ {
+ "abc%def", '\\',
+ "^abc.*?def$",
+ "abc", false,
+ []string{"abcdefdef", "abcdef", "abcdefghidef"},
+ []string{"abcdefg", "abcdefde"},
+ },
+ {
+ "[0-9]*abc%def", '\\',
+ "^\\[0-9\\]\\*abc.*?def$",
+ "[0-9]*abc", false,
+ []string{"[0-9]*abcdefdef", "[0-9]*abcdef", "[0-9]*abcdefghidef"},
+ []string{"0abcdefg", "9abcdefde", "[0-9]abcdefg", "[0-9]abcdefg", "[0-9]abcdefg"},
+ },
+ {
+ "abc\x00%def\x00", '\x00', // nul character doesn't escape
+ "^abc\x00.*?def\x00$",
+ "abc\x00", false,
+ []string{"abc\x00defdef\x00", "abc\x00def\x00", "abc\x00defghidef\x00"},
+ []string{"abcdef\x00", "abc%def", "abc\x00defg", "abc\x00def\x00de", "abc\x00defdef"},
+ },
+ }
+
+ for _, test := range tests {
+ p, err := ParseWithEscapeChar(test.pattern, test.escChar)
+ if err != nil {
+ t.Fatalf("failed to parse pattern %q: %v", test.pattern, err)
+ }
+ if got, want := p.regex.String(), test.wantRegex; got != want {
+ t.Errorf("pattern %q: expected regex %q, got %q", test.pattern, want, got)
+ }
+ if got, want := p.fixedPrefix, test.wantFixedPrefix; got != want {
+ t.Errorf("pattern %q: expected fixedPrefix %q, got %q", test.pattern, want, got)
+ }
+ if got, want := p.noWildcards, test.wantNoWildcards; got != want {
+ t.Errorf("pattern %q: expected noWildcards %v, got %v", test.pattern, want, got)
+ }
+ // Make sure all expected matches actually match.
+ for _, m := range test.matches {
+ if !p.MatchString(m) {
+ t.Errorf("pattern %q: expected %q to match", test.pattern, m)
+ }
+ }
+ // Make sure all expected nonMatches actually don't match.
+ for _, n := range test.nonMatches {
+ if p.MatchString(n) {
+ t.Errorf("pattern %q: expected %q to not match", test.pattern, n)
+ }
+ }
+ }
+}
+
+func TestPatternParseError(t *testing.T) {
+ escChar := '#'
+ errorPatterns := []string{
+ "#",
+ "abc#",
+ "a#bc",
+ "a%bc##de#_f#gh",
+ "a%bc##de#_f#",
+ "###",
+ "###a_",
+ "a#bc%de",
+ }
+
+ for _, p := range errorPatterns {
+ if _, err := ParseWithEscapeChar(p, escChar); verror.ErrorID(err) != ErrInvalidEscape.ID {
+ t.Errorf("pattern %q: expected ErrInvalidEscape, got %v", p, err)
+ }
+ }
+}
+
+type escapeTest struct {
+ literal string
+ escChar rune
+ wantPattern string
+}
+
+func TestPatternEscape(t *testing.T) {
+ tests := []escapeTest{
+ {
+ "abc", '#',
+ "abc",
+ },
+ {
+ "abcabc", 'b',
+ "abbcabbc",
+ },
+ {
+ "ab%c", '#',
+ "ab#%c",
+ },
+ {
+ "_abc%&", '&',
+ "&_abc&%&&",
+ },
+ {
+ "___", '^',
+ "^_^_^_",
+ },
+ {
+ "abc%", '\\',
+ "abc\\%",
+ },
+ {
+ "_同步数据库_", '钒',
+ "钒_同步数据库钒_",
+ },
+ }
+
+ for _, test := range tests {
+ ep := EscapeWithEscapeChar(test.literal, test.escChar)
+ if got, want := ep, test.wantPattern; got != want {
+ t.Errorf("literal %q: expected escape %q, got %q", test.literal, want, got)
+ }
+ if p, err := ParseWithEscapeChar(ep, test.escChar); err != nil {
+ t.Fatalf("failed to parse escaped pattern %q: %v", ep, err)
+ } else if !p.noWildcards {
+ t.Errorf("literal %q: escaped pattern contains unescaped wildcards", test.literal)
+ }
+ }
+}
diff --git a/query/syncql/.api b/query/syncql/.api
index 9a0794d..0b037be 100644
--- a/query/syncql/.api
+++ b/query/syncql/.api
@@ -11,7 +11,6 @@
pkg syncql, func NewErrDidYouMeanLowercaseK(*context.T, int64) error
pkg syncql, func NewErrDidYouMeanLowercaseV(*context.T, int64) error
pkg syncql, func NewErrDotNotationDisallowedForKey(*context.T, int64) error
-pkg syncql, func NewErrErrorCompilingRegularExpression(*context.T, int64, string, error) error
pkg syncql, func NewErrExecOfUnknownStatementType(*context.T, int64, string) error
pkg syncql, func NewErrExpected(*context.T, int64, string) error
pkg syncql, func NewErrExpectedFrom(*context.T, int64, string) error
@@ -27,9 +26,9 @@
pkg syncql, func NewErrFunctionTypeInvalidArg(*context.T, int64) error
pkg syncql, func NewErrIndexKindNotSupported(*context.T, int64, string, string, string) error
pkg syncql, func NewErrIntConversionError(*context.T, int64, error) error
-pkg syncql, func NewErrInvalidEscapeChar(*context.T, int64) error
-pkg syncql, func NewErrInvalidEscapeSequence(*context.T, int64) error
+pkg syncql, func NewErrInvalidEscapeChar(*context.T, int64, string) error
pkg syncql, func NewErrInvalidIndexField(*context.T, int64, string, string) error
+pkg syncql, func NewErrInvalidLikePattern(*context.T, int64, error) error
pkg syncql, func NewErrInvalidSelectField(*context.T, int64) error
pkg syncql, func NewErrIsIsNotRequireLhsValue(*context.T, int64) error
pkg syncql, func NewErrIsIsNotRequireRhsNil(*context.T, int64) error
@@ -73,7 +72,6 @@
pkg syncql, var ErrDidYouMeanLowercaseK unknown-type
pkg syncql, var ErrDidYouMeanLowercaseV unknown-type
pkg syncql, var ErrDotNotationDisallowedForKey unknown-type
-pkg syncql, var ErrErrorCompilingRegularExpression unknown-type
pkg syncql, var ErrExecOfUnknownStatementType unknown-type
pkg syncql, var ErrExpected unknown-type
pkg syncql, var ErrExpectedFrom unknown-type
@@ -90,8 +88,8 @@
pkg syncql, var ErrIndexKindNotSupported unknown-type
pkg syncql, var ErrIntConversionError unknown-type
pkg syncql, var ErrInvalidEscapeChar unknown-type
-pkg syncql, var ErrInvalidEscapeSequence unknown-type
pkg syncql, var ErrInvalidIndexField unknown-type
+pkg syncql, var ErrInvalidLikePattern unknown-type
pkg syncql, var ErrInvalidSelectField unknown-type
pkg syncql, var ErrIsIsNotRequireLhsValue unknown-type
pkg syncql, var ErrIsIsNotRequireRhsNil unknown-type
diff --git a/query/syncql/errors.vdl b/query/syncql/errors.vdl
index 0460edf..cd44a8e 100644
--- a/query/syncql/errors.vdl
+++ b/query/syncql/errors.vdl
@@ -20,9 +20,6 @@
DotNotationDisallowedForKey(off int64) {
"en": "[{off}]Dot notation may not be used on a key field.",
}
- ErrorCompilingRegularExpression(off int64, regex string, err error) {
- "en": "[{off}]The following error encountered compiling regex '{regex}': {err}",
- }
ExecOfUnknownStatementType(off int64, statementType string) {
"en": "[{off}]Cannot execute unknown statement type: {statementType}.",
}
@@ -98,8 +95,8 @@
IsIsNotRequireRhsNil(off int64) {
"en": "[{off}]'Is/is not' expressions require right operand to be nil.",
}
- InvalidEscapeSequence(off int64) {
- "en": "[{off}Expected percent, or underscore after escape character.]",
+ InvalidLikePattern(off int64, err error) {
+ "en": "[{off}]Invalid like pattern: {err}.",
}
InvalidSelectField(off int64) {
"en": "[{off}]Select field must be 'k' or 'v[{.<ident>}...]'.",
@@ -140,8 +137,8 @@
UnknownIdentifier(off int64, found string) {
"en": "[{off}]Unknown identifier: {found}.",
}
- InvalidEscapeChar(off int64) {
- "en": "[{off}]Invalid escape character cannot be space or backslash.",
+ InvalidEscapeChar(off int64, escChar string) {
+ "en": "[{off}]'{escChar}' is not a valid escape character.",
}
DidYouMeanLowercaseK(off int64) {
"en": "[{off}]Did you mean: 'k'?",
diff --git a/query/syncql/syncql.vdl.go b/query/syncql/syncql.vdl.go
index 4c94b80..ecfcea8 100644
--- a/query/syncql/syncql.vdl.go
+++ b/query/syncql/syncql.vdl.go
@@ -24,7 +24,6 @@
ErrCheckOfUnknownStatementType = verror.Register("v.io/v23/query/syncql.CheckOfUnknownStatementType", verror.NoRetry, "{1:}{2:} [{3}]Cannot semantically check unknown statement type.")
ErrCouldNotConvert = verror.Register("v.io/v23/query/syncql.CouldNotConvert", verror.NoRetry, "{1:}{2:} [{3}]Could not convert {4} to {5}.")
ErrDotNotationDisallowedForKey = verror.Register("v.io/v23/query/syncql.DotNotationDisallowedForKey", verror.NoRetry, "{1:}{2:} [{3}]Dot notation may not be used on a key field.")
- ErrErrorCompilingRegularExpression = verror.Register("v.io/v23/query/syncql.ErrorCompilingRegularExpression", verror.NoRetry, "{1:}{2:} [{3}]The following error encountered compiling regex '{4}': {5}")
ErrExecOfUnknownStatementType = verror.Register("v.io/v23/query/syncql.ExecOfUnknownStatementType", verror.NoRetry, "{1:}{2:} [{3}]Cannot execute unknown statement type: {4}.")
ErrExpected = verror.Register("v.io/v23/query/syncql.Expected", verror.NoRetry, "{1:}{2:} [{3}]Expected '{4}'.")
ErrExpectedFrom = verror.Register("v.io/v23/query/syncql.ExpectedFrom", verror.NoRetry, "{1:}{2:} [{3}]Expected 'from', found {4}.")
@@ -50,7 +49,7 @@
ErrIntConversionError = verror.Register("v.io/v23/query/syncql.IntConversionError", verror.NoRetry, "{1:}{2:} [{3}]Can't convert to int: {4}.")
ErrIsIsNotRequireLhsValue = verror.Register("v.io/v23/query/syncql.IsIsNotRequireLhsValue", verror.NoRetry, "{1:}{2:} [{3}]'Is/is not' expressions require left operand to be a value operand.")
ErrIsIsNotRequireRhsNil = verror.Register("v.io/v23/query/syncql.IsIsNotRequireRhsNil", verror.NoRetry, "{1:}{2:} [{3}]'Is/is not' expressions require right operand to be nil.")
- ErrInvalidEscapeSequence = verror.Register("v.io/v23/query/syncql.InvalidEscapeSequence", verror.NoRetry, "{1:}{2:} [{3}Expected percent, or underscore after escape character.]")
+ ErrInvalidLikePattern = verror.Register("v.io/v23/query/syncql.InvalidLikePattern", verror.NoRetry, "{1:}{2:} [{3}]Invalid like pattern: {4}.")
ErrInvalidSelectField = verror.Register("v.io/v23/query/syncql.InvalidSelectField", verror.NoRetry, "{1:}{2:} [{3}]Select field must be 'k' or 'v[{.<ident>}...]'.")
ErrKeyExpressionLiteral = verror.Register("v.io/v23/query/syncql.KeyExpressionLiteral", verror.NoRetry, "{1:}{2:} [{3}]Key (i.e., 'k') compares against literals must be string literal.")
ErrKeyValueStreamError = verror.Register("v.io/v23/query/syncql.KeyValueStreamError", verror.NoRetry, "{1:}{2:} [{3}]KeyValueStream error: {4}.")
@@ -64,7 +63,7 @@
ErrUnexpected = verror.Register("v.io/v23/query/syncql.Unexpected", verror.NoRetry, "{1:}{2:} [{3}]Unexpected: {4}.")
ErrUnexpectedEndOfStatement = verror.Register("v.io/v23/query/syncql.UnexpectedEndOfStatement", verror.NoRetry, "{1:}{2:} [{3}]Unexpected end of statement.")
ErrUnknownIdentifier = verror.Register("v.io/v23/query/syncql.UnknownIdentifier", verror.NoRetry, "{1:}{2:} [{3}]Unknown identifier: {4}.")
- ErrInvalidEscapeChar = verror.Register("v.io/v23/query/syncql.InvalidEscapeChar", verror.NoRetry, "{1:}{2:} [{3}]Invalid escape character cannot be space or backslash.")
+ ErrInvalidEscapeChar = verror.Register("v.io/v23/query/syncql.InvalidEscapeChar", verror.NoRetry, "{1:}{2:} [{3}]'{4}' is not a valid escape character.")
ErrDidYouMeanLowercaseK = verror.Register("v.io/v23/query/syncql.DidYouMeanLowercaseK", verror.NoRetry, "{1:}{2:} [{3}]Did you mean: 'k'?")
ErrDidYouMeanLowercaseV = verror.Register("v.io/v23/query/syncql.DidYouMeanLowercaseV", verror.NoRetry, "{1:}{2:} [{3}]Did you mean: 'v'?")
ErrDidYouMeanFunction = verror.Register("v.io/v23/query/syncql.DidYouMeanFunction", verror.NoRetry, "{1:}{2:} [{3}]Did you mean: '{4}'?")
@@ -102,11 +101,6 @@
return verror.New(ErrDotNotationDisallowedForKey, ctx, off)
}
-// NewErrErrorCompilingRegularExpression returns an error with the ErrErrorCompilingRegularExpression ID.
-func NewErrErrorCompilingRegularExpression(ctx *context.T, off int64, regex string, err error) error {
- return verror.New(ErrErrorCompilingRegularExpression, ctx, off, regex, err)
-}
-
// NewErrExecOfUnknownStatementType returns an error with the ErrExecOfUnknownStatementType ID.
func NewErrExecOfUnknownStatementType(ctx *context.T, off int64, statementType string) error {
return verror.New(ErrExecOfUnknownStatementType, ctx, off, statementType)
@@ -232,9 +226,9 @@
return verror.New(ErrIsIsNotRequireRhsNil, ctx, off)
}
-// NewErrInvalidEscapeSequence returns an error with the ErrInvalidEscapeSequence ID.
-func NewErrInvalidEscapeSequence(ctx *context.T, off int64) error {
- return verror.New(ErrInvalidEscapeSequence, ctx, off)
+// NewErrInvalidLikePattern returns an error with the ErrInvalidLikePattern ID.
+func NewErrInvalidLikePattern(ctx *context.T, off int64, err error) error {
+ return verror.New(ErrInvalidLikePattern, ctx, off, err)
}
// NewErrInvalidSelectField returns an error with the ErrInvalidSelectField ID.
@@ -303,8 +297,8 @@
}
// NewErrInvalidEscapeChar returns an error with the ErrInvalidEscapeChar ID.
-func NewErrInvalidEscapeChar(ctx *context.T, off int64) error {
- return verror.New(ErrInvalidEscapeChar, ctx, off)
+func NewErrInvalidEscapeChar(ctx *context.T, off int64, escChar string) error {
+ return verror.New(ErrInvalidEscapeChar, ctx, off, escChar)
}
// NewErrDidYouMeanLowercaseK returns an error with the ErrDidYouMeanLowercaseK ID.
@@ -384,7 +378,6 @@
i18n.Cat().SetWithBase(i18n.LangID("en"), i18n.MsgID(ErrCheckOfUnknownStatementType.ID), "{1:}{2:} [{3}]Cannot semantically check unknown statement type.")
i18n.Cat().SetWithBase(i18n.LangID("en"), i18n.MsgID(ErrCouldNotConvert.ID), "{1:}{2:} [{3}]Could not convert {4} to {5}.")
i18n.Cat().SetWithBase(i18n.LangID("en"), i18n.MsgID(ErrDotNotationDisallowedForKey.ID), "{1:}{2:} [{3}]Dot notation may not be used on a key field.")
- i18n.Cat().SetWithBase(i18n.LangID("en"), i18n.MsgID(ErrErrorCompilingRegularExpression.ID), "{1:}{2:} [{3}]The following error encountered compiling regex '{4}': {5}")
i18n.Cat().SetWithBase(i18n.LangID("en"), i18n.MsgID(ErrExecOfUnknownStatementType.ID), "{1:}{2:} [{3}]Cannot execute unknown statement type: {4}.")
i18n.Cat().SetWithBase(i18n.LangID("en"), i18n.MsgID(ErrExpected.ID), "{1:}{2:} [{3}]Expected '{4}'.")
i18n.Cat().SetWithBase(i18n.LangID("en"), i18n.MsgID(ErrExpectedFrom.ID), "{1:}{2:} [{3}]Expected 'from', found {4}.")
@@ -410,7 +403,7 @@
i18n.Cat().SetWithBase(i18n.LangID("en"), i18n.MsgID(ErrIntConversionError.ID), "{1:}{2:} [{3}]Can't convert to int: {4}.")
i18n.Cat().SetWithBase(i18n.LangID("en"), i18n.MsgID(ErrIsIsNotRequireLhsValue.ID), "{1:}{2:} [{3}]'Is/is not' expressions require left operand to be a value operand.")
i18n.Cat().SetWithBase(i18n.LangID("en"), i18n.MsgID(ErrIsIsNotRequireRhsNil.ID), "{1:}{2:} [{3}]'Is/is not' expressions require right operand to be nil.")
- i18n.Cat().SetWithBase(i18n.LangID("en"), i18n.MsgID(ErrInvalidEscapeSequence.ID), "{1:}{2:} [{3}Expected percent, or underscore after escape character.]")
+ i18n.Cat().SetWithBase(i18n.LangID("en"), i18n.MsgID(ErrInvalidLikePattern.ID), "{1:}{2:} [{3}]Invalid like pattern: {4}.")
i18n.Cat().SetWithBase(i18n.LangID("en"), i18n.MsgID(ErrInvalidSelectField.ID), "{1:}{2:} [{3}]Select field must be 'k' or 'v[{.<ident>}...]'.")
i18n.Cat().SetWithBase(i18n.LangID("en"), i18n.MsgID(ErrKeyExpressionLiteral.ID), "{1:}{2:} [{3}]Key (i.e., 'k') compares against literals must be string literal.")
i18n.Cat().SetWithBase(i18n.LangID("en"), i18n.MsgID(ErrKeyValueStreamError.ID), "{1:}{2:} [{3}]KeyValueStream error: {4}.")
@@ -424,7 +417,7 @@
i18n.Cat().SetWithBase(i18n.LangID("en"), i18n.MsgID(ErrUnexpected.ID), "{1:}{2:} [{3}]Unexpected: {4}.")
i18n.Cat().SetWithBase(i18n.LangID("en"), i18n.MsgID(ErrUnexpectedEndOfStatement.ID), "{1:}{2:} [{3}]Unexpected end of statement.")
i18n.Cat().SetWithBase(i18n.LangID("en"), i18n.MsgID(ErrUnknownIdentifier.ID), "{1:}{2:} [{3}]Unknown identifier: {4}.")
- i18n.Cat().SetWithBase(i18n.LangID("en"), i18n.MsgID(ErrInvalidEscapeChar.ID), "{1:}{2:} [{3}]Invalid escape character cannot be space or backslash.")
+ i18n.Cat().SetWithBase(i18n.LangID("en"), i18n.MsgID(ErrInvalidEscapeChar.ID), "{1:}{2:} [{3}]'{4}' is not a valid escape character.")
i18n.Cat().SetWithBase(i18n.LangID("en"), i18n.MsgID(ErrDidYouMeanLowercaseK.ID), "{1:}{2:} [{3}]Did you mean: 'k'?")
i18n.Cat().SetWithBase(i18n.LangID("en"), i18n.MsgID(ErrDidYouMeanLowercaseV.ID), "{1:}{2:} [{3}]Did you mean: 'v'?")
i18n.Cat().SetWithBase(i18n.LangID("en"), i18n.MsgID(ErrDidYouMeanFunction.ID), "{1:}{2:} [{3}]Did you mean: '{4}'?")
diff --git a/services/syncbase/.api b/services/syncbase/.api
index a8b1487..36bafaf 100644
--- a/services/syncbase/.api
+++ b/services/syncbase/.api
@@ -70,6 +70,7 @@
pkg syncbase, method (*BlobManagerPutBlobServerCallStub) RecvStream() interface { Advance() bool; Value() []byte; Err() error;}
pkg syncbase, method (*BlobRef) VDLRead(vdl.Decoder) error
pkg syncbase, method (*CollectionRow) VDLRead(vdl.Decoder) error
+pkg syncbase, method (*CollectionRowPattern) VDLRead(vdl.Decoder) error
pkg syncbase, method (*CollectionScanServerCallStub) Init(rpc.StreamServerCall)
pkg syncbase, method (*CollectionScanServerCallStub) SendStream() interface { Send(item KeyValue) error;}
pkg syncbase, method (*ConflictInfo) VDLRead(vdl.Decoder) error
@@ -80,6 +81,8 @@
pkg syncbase, method (*CrRule) VDLRead(vdl.Decoder) error
pkg syncbase, method (*DatabaseExecServerCallStub) Init(rpc.StreamServerCall)
pkg syncbase, method (*DatabaseExecServerCallStub) SendStream() interface { Send(item []*vom.RawBytes) error;}
+pkg syncbase, method (*DatabaseWatcherWatchPatternsServerCallStub) Init(rpc.StreamServerCall)
+pkg syncbase, method (*DatabaseWatcherWatchPatternsServerCallStub) SendStream() interface { Send(item watch.Change) error;}
pkg syncbase, method (*DevModeUpdateVClockOpts) VDLRead(vdl.Decoder) error
pkg syncbase, method (*Id) VDLRead(vdl.Decoder) error
pkg syncbase, method (*KeyValue) VDLRead(vdl.Decoder) error
@@ -116,6 +119,8 @@
pkg syncbase, method (BlobRef) VDLWrite(vdl.Encoder) error
pkg syncbase, method (CollectionRow) VDLIsZero() bool
pkg syncbase, method (CollectionRow) VDLWrite(vdl.Encoder) error
+pkg syncbase, method (CollectionRowPattern) VDLIsZero() bool
+pkg syncbase, method (CollectionRowPattern) VDLWrite(vdl.Encoder) error
pkg syncbase, method (ConflictDataBatch) Index() int
pkg syncbase, method (ConflictDataBatch) Interface() interface{}
pkg syncbase, method (ConflictDataBatch) Name() string
@@ -303,6 +308,10 @@
pkg syncbase, type CollectionRow struct
pkg syncbase, type CollectionRow struct, CollectionId Id
pkg syncbase, type CollectionRow struct, Row string
+pkg syncbase, type CollectionRowPattern struct
+pkg syncbase, type CollectionRowPattern struct, CollectionBlessing string
+pkg syncbase, type CollectionRowPattern struct, CollectionName string
+pkg syncbase, type CollectionRowPattern struct, RowKey string
pkg syncbase, type CollectionScanClientCall interface { Finish, RecvStream }
pkg syncbase, type CollectionScanClientCall interface, Finish() error
pkg syncbase, type CollectionScanClientCall interface, RecvStream() interface { Advance() bool;; Value() KeyValue;; Err() error;}
@@ -418,6 +427,7 @@
pkg syncbase, type DatabaseClientMethods interface, SetSyncgroupSpec(*context.T, Id, SyncgroupSpec, string, ...rpc.CallOpt) error
pkg syncbase, type DatabaseClientMethods interface, StartConflictResolver(*context.T, ...rpc.CallOpt) (ConflictManagerStartConflictResolverClientCall, error)
pkg syncbase, type DatabaseClientMethods interface, UnpinBlob(*context.T, BlobRef, ...rpc.CallOpt) error
+pkg syncbase, type DatabaseClientMethods interface, WatchPatterns(*context.T, watch.ResumeMarker, []CollectionRowPattern, ...rpc.CallOpt) (DatabaseWatcherWatchPatternsClientCall, error)
pkg syncbase, type DatabaseClientMethods interface, unexported methods
pkg syncbase, type DatabaseClientStub interface, Abort(*context.T, BatchHandle, ...rpc.CallOpt) error
pkg syncbase, type DatabaseClientStub interface, BeginBatch(*context.T, BatchOptions, ...rpc.CallOpt) (BatchHandle, error)
@@ -452,6 +462,7 @@
pkg syncbase, type DatabaseClientStub interface, SetSyncgroupSpec(*context.T, Id, SyncgroupSpec, string, ...rpc.CallOpt) error
pkg syncbase, type DatabaseClientStub interface, StartConflictResolver(*context.T, ...rpc.CallOpt) (ConflictManagerStartConflictResolverClientCall, error)
pkg syncbase, type DatabaseClientStub interface, UnpinBlob(*context.T, BlobRef, ...rpc.CallOpt) error
+pkg syncbase, type DatabaseClientStub interface, WatchPatterns(*context.T, watch.ResumeMarker, []CollectionRowPattern, ...rpc.CallOpt) (DatabaseWatcherWatchPatternsClientCall, error)
pkg syncbase, type DatabaseClientStub interface, unexported methods
pkg syncbase, type DatabaseExecClientCall interface { Finish, RecvStream }
pkg syncbase, type DatabaseExecClientCall interface, Finish() error
@@ -497,6 +508,7 @@
pkg syncbase, type DatabaseServerMethods interface, SetSyncgroupSpec(*context.T, rpc.ServerCall, Id, SyncgroupSpec, string) error
pkg syncbase, type DatabaseServerMethods interface, StartConflictResolver(*context.T, ConflictManagerStartConflictResolverServerCall) error
pkg syncbase, type DatabaseServerMethods interface, UnpinBlob(*context.T, rpc.ServerCall, BlobRef) error
+pkg syncbase, type DatabaseServerMethods interface, WatchPatterns(*context.T, DatabaseWatcherWatchPatternsServerCall, watch.ResumeMarker, []CollectionRowPattern) error
pkg syncbase, type DatabaseServerMethods interface, unexported methods
pkg syncbase, type DatabaseServerStub interface, Abort(*context.T, rpc.ServerCall, BatchHandle) error
pkg syncbase, type DatabaseServerStub interface, BeginBatch(*context.T, rpc.ServerCall, BatchOptions) (BatchHandle, error)
@@ -521,6 +533,7 @@
pkg syncbase, type DatabaseServerStub interface, ResumeSync(*context.T, rpc.ServerCall) error
pkg syncbase, type DatabaseServerStub interface, StartConflictResolver(*context.T, *ConflictManagerStartConflictResolverServerCallStub) error
pkg syncbase, type DatabaseServerStub interface, UnpinBlob(*context.T, rpc.ServerCall, BlobRef) error
+pkg syncbase, type DatabaseServerStub interface, WatchPatterns(*context.T, *DatabaseWatcherWatchPatternsServerCallStub, watch.ResumeMarker, []CollectionRowPattern) error
pkg syncbase, type DatabaseServerStub interface, unexported methods
pkg syncbase, type DatabaseServerStubMethods interface, Abort(*context.T, rpc.ServerCall, BatchHandle) error
pkg syncbase, type DatabaseServerStubMethods interface, BeginBatch(*context.T, rpc.ServerCall, BatchOptions) (BatchHandle, error)
@@ -544,18 +557,35 @@
pkg syncbase, type DatabaseServerStubMethods interface, ResumeSync(*context.T, rpc.ServerCall) error
pkg syncbase, type DatabaseServerStubMethods interface, StartConflictResolver(*context.T, *ConflictManagerStartConflictResolverServerCallStub) error
pkg syncbase, type DatabaseServerStubMethods interface, UnpinBlob(*context.T, rpc.ServerCall, BlobRef) error
+pkg syncbase, type DatabaseServerStubMethods interface, WatchPatterns(*context.T, *DatabaseWatcherWatchPatternsServerCallStub, watch.ResumeMarker, []CollectionRowPattern) error
pkg syncbase, type DatabaseServerStubMethods interface, unexported methods
pkg syncbase, type DatabaseWatcherClientMethods interface, GetResumeMarker(*context.T, BatchHandle, ...rpc.CallOpt) (watch.ResumeMarker, error)
+pkg syncbase, type DatabaseWatcherClientMethods interface, WatchPatterns(*context.T, watch.ResumeMarker, []CollectionRowPattern, ...rpc.CallOpt) (DatabaseWatcherWatchPatternsClientCall, error)
pkg syncbase, type DatabaseWatcherClientMethods interface, unexported methods
pkg syncbase, type DatabaseWatcherClientStub interface, GetResumeMarker(*context.T, BatchHandle, ...rpc.CallOpt) (watch.ResumeMarker, error)
+pkg syncbase, type DatabaseWatcherClientStub interface, WatchPatterns(*context.T, watch.ResumeMarker, []CollectionRowPattern, ...rpc.CallOpt) (DatabaseWatcherWatchPatternsClientCall, error)
pkg syncbase, type DatabaseWatcherClientStub interface, unexported methods
pkg syncbase, type DatabaseWatcherServerMethods interface, GetResumeMarker(*context.T, rpc.ServerCall, BatchHandle) (watch.ResumeMarker, error)
+pkg syncbase, type DatabaseWatcherServerMethods interface, WatchPatterns(*context.T, DatabaseWatcherWatchPatternsServerCall, watch.ResumeMarker, []CollectionRowPattern) error
pkg syncbase, type DatabaseWatcherServerMethods interface, unexported methods
pkg syncbase, type DatabaseWatcherServerStub interface, Describe__() []rpc.InterfaceDesc
pkg syncbase, type DatabaseWatcherServerStub interface, GetResumeMarker(*context.T, rpc.ServerCall, BatchHandle) (watch.ResumeMarker, error)
+pkg syncbase, type DatabaseWatcherServerStub interface, WatchPatterns(*context.T, *DatabaseWatcherWatchPatternsServerCallStub, watch.ResumeMarker, []CollectionRowPattern) error
pkg syncbase, type DatabaseWatcherServerStub interface, unexported methods
pkg syncbase, type DatabaseWatcherServerStubMethods interface, GetResumeMarker(*context.T, rpc.ServerCall, BatchHandle) (watch.ResumeMarker, error)
+pkg syncbase, type DatabaseWatcherServerStubMethods interface, WatchPatterns(*context.T, *DatabaseWatcherWatchPatternsServerCallStub, watch.ResumeMarker, []CollectionRowPattern) error
pkg syncbase, type DatabaseWatcherServerStubMethods interface, unexported methods
+pkg syncbase, type DatabaseWatcherWatchPatternsClientCall interface { Finish, RecvStream }
+pkg syncbase, type DatabaseWatcherWatchPatternsClientCall interface, Finish() error
+pkg syncbase, type DatabaseWatcherWatchPatternsClientCall interface, RecvStream() interface { Advance() bool;; Value() watch.Change;; Err() error;}
+pkg syncbase, type DatabaseWatcherWatchPatternsClientStream interface { RecvStream }
+pkg syncbase, type DatabaseWatcherWatchPatternsClientStream interface, RecvStream() interface { Advance() bool;; Value() watch.Change;; Err() error;}
+pkg syncbase, type DatabaseWatcherWatchPatternsServerCall interface, SendStream() interface { Send(item watch.Change) error;}
+pkg syncbase, type DatabaseWatcherWatchPatternsServerCall interface, unexported methods
+pkg syncbase, type DatabaseWatcherWatchPatternsServerCallStub struct
+pkg syncbase, type DatabaseWatcherWatchPatternsServerCallStub struct, embedded rpc.StreamServerCall
+pkg syncbase, type DatabaseWatcherWatchPatternsServerStream interface { SendStream }
+pkg syncbase, type DatabaseWatcherWatchPatternsServerStream interface, SendStream() interface { Send(item watch.Change) error;}
pkg syncbase, type DevModeUpdateVClockOpts struct
pkg syncbase, type DevModeUpdateVClockOpts struct, DoLocalUpdate bool
pkg syncbase, type DevModeUpdateVClockOpts struct, DoNtpUpdate bool
diff --git a/services/syncbase/service.vdl b/services/syncbase/service.vdl
index 0cc627d..204c088 100644
--- a/services/syncbase/service.vdl
+++ b/services/syncbase/service.vdl
@@ -350,16 +350,15 @@
// DatabaseWatcher allows a client to watch for updates to the database. For
// each watch request, the client will receive a reliable stream of watch events
-// without re-ordering. See watch.GlobWatcher for a detailed explanation of the
-// behavior.
-// TODO(rogulenko): Currently the only supported watch patterns are
-// "<collectionId>/<rowPrefix>*". Consider changing that.
+// without re-ordering. Only rows matching at least one of the patterns are
+// returned. Rows in collections with no Read access are also filtered out.
//
// Watching is done by starting a streaming RPC. The RPC takes a ResumeMarker
// argument that points to a particular place in the database event log. If an
// empty ResumeMarker is provided, the WatchStream will begin with a Change
// batch containing the initial state. Otherwise, the WatchStream will contain
-// only changes since the provided ResumeMarker.
+// only changes since the provided ResumeMarker. See watch.GlobWatcher for a
+// detailed explanation of the behavior.
//
// The result stream consists of a never-ending sequence of Change messages
// (until the call fails or is canceled). Each Change contains the Name field in
@@ -372,11 +371,15 @@
// resolution. However, changes from a single original batch will always appear
// in the same Change batch.
type DatabaseWatcher interface {
- watch.GlobWatcher
-
// GetResumeMarker returns the ResumeMarker that points to the current end
// of the event log. GetResumeMarker() can be called on a batch.
GetResumeMarker(bh BatchHandle) (watch.ResumeMarker | error) {access.Read}
+
+ // WatchPatterns returns a stream of changes that match any of the specified
+ // patterns. At least one pattern must be specified.
+ WatchPatterns(resumeMarker watch.ResumeMarker, patterns []CollectionRowPattern) stream<_, watch.Change> error {access.Resolve}
+
+ watch.GlobWatcher
}
error (
diff --git a/services/syncbase/syncbase.vdl.go b/services/syncbase/syncbase.vdl.go
index 01167c9..381c621 100644
--- a/services/syncbase/syncbase.vdl.go
+++ b/services/syncbase/syncbase.vdl.go
@@ -3095,8 +3095,133 @@
}
}
+// CollectionRowPattern contains SQL LIKE-style glob patterns ('%' and '_'
+// wildcards, '\' as escape character) for matching rows and collections by
+// name components.
+// Collection blessing and name patterns are not allowed to be empty, but the
+// row key pattern is (for matching only collections and no rows).
+type CollectionRowPattern struct {
+ CollectionBlessing string
+ CollectionName string
+ RowKey string
+}
+
+func (CollectionRowPattern) __VDLReflect(struct {
+ Name string `vdl:"v.io/v23/services/syncbase.CollectionRowPattern"`
+}) {
+}
+
+func (x CollectionRowPattern) VDLIsZero() bool {
+ return x == CollectionRowPattern{}
+}
+
+func (x CollectionRowPattern) VDLWrite(enc vdl.Encoder) error {
+ if err := enc.StartValue(__VDLType_struct_36); err != nil {
+ return err
+ }
+ if x.CollectionBlessing != "" {
+ if err := enc.NextField("CollectionBlessing"); err != nil {
+ return err
+ }
+ if err := enc.StartValue(vdl.StringType); err != nil {
+ return err
+ }
+ if err := enc.EncodeString(x.CollectionBlessing); err != nil {
+ return err
+ }
+ if err := enc.FinishValue(); err != nil {
+ return err
+ }
+ }
+ if x.CollectionName != "" {
+ if err := enc.NextField("CollectionName"); err != nil {
+ return err
+ }
+ if err := enc.StartValue(vdl.StringType); err != nil {
+ return err
+ }
+ if err := enc.EncodeString(x.CollectionName); err != nil {
+ return err
+ }
+ if err := enc.FinishValue(); err != nil {
+ return err
+ }
+ }
+ if x.RowKey != "" {
+ if err := enc.NextField("RowKey"); err != nil {
+ return err
+ }
+ if err := enc.StartValue(vdl.StringType); err != nil {
+ return err
+ }
+ if err := enc.EncodeString(x.RowKey); err != nil {
+ return err
+ }
+ if err := enc.FinishValue(); err != nil {
+ return err
+ }
+ }
+ if err := enc.NextField(""); err != nil {
+ return err
+ }
+ return enc.FinishValue()
+}
+
+func (x *CollectionRowPattern) VDLRead(dec vdl.Decoder) error {
+ *x = CollectionRowPattern{}
+ if err := dec.StartValue(__VDLType_struct_36); err != nil {
+ return err
+ }
+ for {
+ f, err := dec.NextField()
+ if err != nil {
+ return err
+ }
+ switch f {
+ case "":
+ return dec.FinishValue()
+ case "CollectionBlessing":
+ if err := dec.StartValue(vdl.StringType); err != nil {
+ return err
+ }
+ var err error
+ if x.CollectionBlessing, err = dec.DecodeString(); err != nil {
+ return err
+ }
+ if err := dec.FinishValue(); err != nil {
+ return err
+ }
+ case "CollectionName":
+ if err := dec.StartValue(vdl.StringType); err != nil {
+ return err
+ }
+ var err error
+ if x.CollectionName, err = dec.DecodeString(); err != nil {
+ return err
+ }
+ if err := dec.FinishValue(); err != nil {
+ return err
+ }
+ case "RowKey":
+ if err := dec.StartValue(vdl.StringType); err != nil {
+ return err
+ }
+ var err error
+ if x.RowKey, err = dec.DecodeString(); err != nil {
+ return err
+ }
+ if err := dec.FinishValue(); err != nil {
+ return err
+ }
+ default:
+ if err := dec.SkipValue(); err != nil {
+ return err
+ }
+ }
+ }
+}
+
// StoreChange is the new value for a watched entity.
-// TODO(rogulenko): Consider adding the Shell state.
type StoreChange struct {
// Value is the new value for the row if the Change state equals to Exists,
// otherwise the Value is nil.
@@ -3123,7 +3248,7 @@
}
func (x StoreChange) VDLWrite(enc vdl.Encoder) error {
- if err := enc.StartValue(__VDLType_struct_36); err != nil {
+ if err := enc.StartValue(__VDLType_struct_37); err != nil {
return err
}
if x.Value != nil && !x.Value.VDLIsZero() {
@@ -3158,7 +3283,7 @@
*x = StoreChange{
Value: vom.RawBytesOf(vdl.ZeroValue(vdl.AnyType)),
}
- if err := dec.StartValue(__VDLType_struct_36); err != nil {
+ if err := dec.StartValue(__VDLType_struct_37); err != nil {
return err
}
for {
@@ -3512,16 +3637,15 @@
//
// DatabaseWatcher allows a client to watch for updates to the database. For
// each watch request, the client will receive a reliable stream of watch events
-// without re-ordering. See watch.GlobWatcher for a detailed explanation of the
-// behavior.
-// TODO(rogulenko): Currently the only supported watch patterns are
-// "<collectionId>/<rowPrefix>*". Consider changing that.
+// without re-ordering. Only rows matching at least one of the patterns are
+// returned. Rows in collections with no Read access are also filtered out.
//
// Watching is done by starting a streaming RPC. The RPC takes a ResumeMarker
// argument that points to a particular place in the database event log. If an
// empty ResumeMarker is provided, the WatchStream will begin with a Change
// batch containing the initial state. Otherwise, the WatchStream will contain
-// only changes since the provided ResumeMarker.
+// only changes since the provided ResumeMarker. See watch.GlobWatcher for a
+// detailed explanation of the behavior.
//
// The result stream consists of a never-ending sequence of Change messages
// (until the call fails or is canceled). Each Change contains the Name field in
@@ -3540,6 +3664,9 @@
// GetResumeMarker returns the ResumeMarker that points to the current end
// of the event log. GetResumeMarker() can be called on a batch.
GetResumeMarker(_ *context.T, bh BatchHandle, _ ...rpc.CallOpt) (watch.ResumeMarker, error)
+ // WatchPatterns returns a stream of changes that match any of the specified
+ // patterns. At least one pattern must be specified.
+ WatchPatterns(_ *context.T, resumeMarker watch.ResumeMarker, patterns []CollectionRowPattern, _ ...rpc.CallOpt) (DatabaseWatcherWatchPatternsClientCall, error)
}
// DatabaseWatcherClientStub adds universal methods to DatabaseWatcherClientMethods.
@@ -3564,21 +3691,98 @@
return
}
+func (c implDatabaseWatcherClientStub) WatchPatterns(ctx *context.T, i0 watch.ResumeMarker, i1 []CollectionRowPattern, opts ...rpc.CallOpt) (ocall DatabaseWatcherWatchPatternsClientCall, err error) {
+ var call rpc.ClientCall
+ if call, err = v23.GetClient(ctx).StartCall(ctx, c.name, "WatchPatterns", []interface{}{i0, i1}, opts...); err != nil {
+ return
+ }
+ ocall = &implDatabaseWatcherWatchPatternsClientCall{ClientCall: call}
+ return
+}
+
+// DatabaseWatcherWatchPatternsClientStream is the client stream for DatabaseWatcher.WatchPatterns.
+type DatabaseWatcherWatchPatternsClientStream interface {
+ // RecvStream returns the receiver side of the DatabaseWatcher.WatchPatterns client stream.
+ RecvStream() interface {
+ // Advance stages an item so that it may be retrieved via Value. Returns
+ // true iff there is an item to retrieve. Advance must be called before
+ // Value is called. May block if an item is not available.
+ Advance() bool
+ // Value returns the item that was staged by Advance. May panic if Advance
+ // returned false or was not called. Never blocks.
+ Value() watch.Change
+ // Err returns any error encountered by Advance. Never blocks.
+ Err() error
+ }
+}
+
+// DatabaseWatcherWatchPatternsClientCall represents the call returned from DatabaseWatcher.WatchPatterns.
+type DatabaseWatcherWatchPatternsClientCall interface {
+ DatabaseWatcherWatchPatternsClientStream
+ // Finish blocks until the server is done, and returns the positional return
+ // values for call.
+ //
+ // Finish returns immediately if the call has been canceled; depending on the
+ // timing the output could either be an error signaling cancelation, or the
+ // valid positional return values from the server.
+ //
+ // Calling Finish is mandatory for releasing stream resources, unless the call
+ // has been canceled or any of the other methods return an error. Finish should
+ // be called at most once.
+ Finish() error
+}
+
+type implDatabaseWatcherWatchPatternsClientCall struct {
+ rpc.ClientCall
+ valRecv watch.Change
+ errRecv error
+}
+
+func (c *implDatabaseWatcherWatchPatternsClientCall) RecvStream() interface {
+ Advance() bool
+ Value() watch.Change
+ Err() error
+} {
+ return implDatabaseWatcherWatchPatternsClientCallRecv{c}
+}
+
+type implDatabaseWatcherWatchPatternsClientCallRecv struct {
+ c *implDatabaseWatcherWatchPatternsClientCall
+}
+
+func (c implDatabaseWatcherWatchPatternsClientCallRecv) Advance() bool {
+ c.c.valRecv = watch.Change{}
+ c.c.errRecv = c.c.Recv(&c.c.valRecv)
+ return c.c.errRecv == nil
+}
+func (c implDatabaseWatcherWatchPatternsClientCallRecv) Value() watch.Change {
+ return c.c.valRecv
+}
+func (c implDatabaseWatcherWatchPatternsClientCallRecv) Err() error {
+ if c.c.errRecv == io.EOF {
+ return nil
+ }
+ return c.c.errRecv
+}
+func (c *implDatabaseWatcherWatchPatternsClientCall) Finish() (err error) {
+ err = c.ClientCall.Finish()
+ return
+}
+
// DatabaseWatcherServerMethods is the interface a server writer
// implements for DatabaseWatcher.
//
// DatabaseWatcher allows a client to watch for updates to the database. For
// each watch request, the client will receive a reliable stream of watch events
-// without re-ordering. See watch.GlobWatcher for a detailed explanation of the
-// behavior.
-// TODO(rogulenko): Currently the only supported watch patterns are
-// "<collectionId>/<rowPrefix>*". Consider changing that.
+// without re-ordering. Only rows matching at least one of the patterns are
+// returned. Rows in collections with no Read access are also filtered out.
//
// Watching is done by starting a streaming RPC. The RPC takes a ResumeMarker
// argument that points to a particular place in the database event log. If an
// empty ResumeMarker is provided, the WatchStream will begin with a Change
// batch containing the initial state. Otherwise, the WatchStream will contain
-// only changes since the provided ResumeMarker.
+// only changes since the provided ResumeMarker. See watch.GlobWatcher for a
+// detailed explanation of the behavior.
//
// The result stream consists of a never-ending sequence of Change messages
// (until the call fails or is canceled). Each Change contains the Name field in
@@ -3597,6 +3801,9 @@
// GetResumeMarker returns the ResumeMarker that points to the current end
// of the event log. GetResumeMarker() can be called on a batch.
GetResumeMarker(_ *context.T, _ rpc.ServerCall, bh BatchHandle) (watch.ResumeMarker, error)
+ // WatchPatterns returns a stream of changes that match any of the specified
+ // patterns. At least one pattern must be specified.
+ WatchPatterns(_ *context.T, _ DatabaseWatcherWatchPatternsServerCall, resumeMarker watch.ResumeMarker, patterns []CollectionRowPattern) error
}
// DatabaseWatcherServerStubMethods is the server interface containing
@@ -3610,6 +3817,9 @@
// GetResumeMarker returns the ResumeMarker that points to the current end
// of the event log. GetResumeMarker() can be called on a batch.
GetResumeMarker(_ *context.T, _ rpc.ServerCall, bh BatchHandle) (watch.ResumeMarker, error)
+ // WatchPatterns returns a stream of changes that match any of the specified
+ // patterns. At least one pattern must be specified.
+ WatchPatterns(_ *context.T, _ *DatabaseWatcherWatchPatternsServerCallStub, resumeMarker watch.ResumeMarker, patterns []CollectionRowPattern) error
}
// DatabaseWatcherServerStub adds universal methods to DatabaseWatcherServerStubMethods.
@@ -3647,6 +3857,10 @@
return s.impl.GetResumeMarker(ctx, call, i0)
}
+func (s implDatabaseWatcherServerStub) WatchPatterns(ctx *context.T, call *DatabaseWatcherWatchPatternsServerCallStub, i0 watch.ResumeMarker, i1 []CollectionRowPattern) error {
+ return s.impl.WatchPatterns(ctx, call, i0, i1)
+}
+
func (s implDatabaseWatcherServerStub) Globber() *rpc.GlobState {
return s.gs
}
@@ -3662,7 +3876,7 @@
var descDatabaseWatcher = rpc.InterfaceDesc{
Name: "DatabaseWatcher",
PkgPath: "v.io/v23/services/syncbase",
- Doc: "// DatabaseWatcher allows a client to watch for updates to the database. For\n// each watch request, the client will receive a reliable stream of watch events\n// without re-ordering. See watch.GlobWatcher for a detailed explanation of the\n// behavior.\n// TODO(rogulenko): Currently the only supported watch patterns are\n// \"<collectionId>/<rowPrefix>*\". Consider changing that.\n//\n// Watching is done by starting a streaming RPC. The RPC takes a ResumeMarker\n// argument that points to a particular place in the database event log. If an\n// empty ResumeMarker is provided, the WatchStream will begin with a Change\n// batch containing the initial state. Otherwise, the WatchStream will contain\n// only changes since the provided ResumeMarker.\n//\n// The result stream consists of a never-ending sequence of Change messages\n// (until the call fails or is canceled). Each Change contains the Name field in\n// the form \"<collectionId>/<rowKey>\" and the Value field of the StoreChange\n// type. If the client has no access to a row specified in a change, that change\n// is excluded from the result stream.\n//\n// Note: A single Watch Change batch may contain changes from more than one\n// batch as originally committed on a remote Syncbase or obtained from conflict\n// resolution. However, changes from a single original batch will always appear\n// in the same Change batch.",
+ Doc: "// DatabaseWatcher allows a client to watch for updates to the database. For\n// each watch request, the client will receive a reliable stream of watch events\n// without re-ordering. Only rows matching at least one of the patterns are\n// returned. Rows in collections with no Read access are also filtered out.\n//\n// Watching is done by starting a streaming RPC. The RPC takes a ResumeMarker\n// argument that points to a particular place in the database event log. If an\n// empty ResumeMarker is provided, the WatchStream will begin with a Change\n// batch containing the initial state. Otherwise, the WatchStream will contain\n// only changes since the provided ResumeMarker. See watch.GlobWatcher for a\n// detailed explanation of the behavior.\n//\n// The result stream consists of a never-ending sequence of Change messages\n// (until the call fails or is canceled). Each Change contains the Name field in\n// the form \"<collectionId>/<rowKey>\" and the Value field of the StoreChange\n// type. If the client has no access to a row specified in a change, that change\n// is excluded from the result stream.\n//\n// Note: A single Watch Change batch may contain changes from more than one\n// batch as originally committed on a remote Syncbase or obtained from conflict\n// resolution. However, changes from a single original batch will always appear\n// in the same Change batch.",
Embeds: []rpc.EmbedDesc{
{"GlobWatcher", "v.io/v23/services/watch", "// GlobWatcher allows a client to receive updates for changes to objects\n// that match a pattern. See the package comments for details."},
},
@@ -3678,9 +3892,61 @@
},
Tags: []*vdl.Value{vdl.ValueOf(access.Tag("Read"))},
},
+ {
+ Name: "WatchPatterns",
+ Doc: "// WatchPatterns returns a stream of changes that match any of the specified\n// patterns. At least one pattern must be specified.",
+ InArgs: []rpc.ArgDesc{
+ {"resumeMarker", ``}, // watch.ResumeMarker
+ {"patterns", ``}, // []CollectionRowPattern
+ },
+ Tags: []*vdl.Value{vdl.ValueOf(access.Tag("Resolve"))},
+ },
},
}
+// DatabaseWatcherWatchPatternsServerStream is the server stream for DatabaseWatcher.WatchPatterns.
+type DatabaseWatcherWatchPatternsServerStream interface {
+ // SendStream returns the send side of the DatabaseWatcher.WatchPatterns server stream.
+ SendStream() interface {
+ // Send places the item onto the output stream. Returns errors encountered
+ // while sending. Blocks if there is no buffer space; will unblock when
+ // buffer space is available.
+ Send(item watch.Change) error
+ }
+}
+
+// DatabaseWatcherWatchPatternsServerCall represents the context passed to DatabaseWatcher.WatchPatterns.
+type DatabaseWatcherWatchPatternsServerCall interface {
+ rpc.ServerCall
+ DatabaseWatcherWatchPatternsServerStream
+}
+
+// DatabaseWatcherWatchPatternsServerCallStub is a wrapper that converts rpc.StreamServerCall into
+// a typesafe stub that implements DatabaseWatcherWatchPatternsServerCall.
+type DatabaseWatcherWatchPatternsServerCallStub struct {
+ rpc.StreamServerCall
+}
+
+// Init initializes DatabaseWatcherWatchPatternsServerCallStub from rpc.StreamServerCall.
+func (s *DatabaseWatcherWatchPatternsServerCallStub) Init(call rpc.StreamServerCall) {
+ s.StreamServerCall = call
+}
+
+// SendStream returns the send side of the DatabaseWatcher.WatchPatterns server stream.
+func (s *DatabaseWatcherWatchPatternsServerCallStub) SendStream() interface {
+ Send(item watch.Change) error
+} {
+ return implDatabaseWatcherWatchPatternsServerCallSend{s}
+}
+
+type implDatabaseWatcherWatchPatternsServerCallSend struct {
+ s *DatabaseWatcherWatchPatternsServerCallStub
+}
+
+func (s implDatabaseWatcherWatchPatternsServerCallSend) Send(item watch.Change) error {
+ return s.s.Send(item)
+}
+
// SyncgroupManagerClientMethods is the client interface
// containing SyncgroupManager methods.
//
@@ -5295,16 +5561,15 @@
permissions.ObjectClientMethods
// DatabaseWatcher allows a client to watch for updates to the database. For
// each watch request, the client will receive a reliable stream of watch events
- // without re-ordering. See watch.GlobWatcher for a detailed explanation of the
- // behavior.
- // TODO(rogulenko): Currently the only supported watch patterns are
- // "<collectionId>/<rowPrefix>*". Consider changing that.
+ // without re-ordering. Only rows matching at least one of the patterns are
+ // returned. Rows in collections with no Read access are also filtered out.
//
// Watching is done by starting a streaming RPC. The RPC takes a ResumeMarker
// argument that points to a particular place in the database event log. If an
// empty ResumeMarker is provided, the WatchStream will begin with a Change
// batch containing the initial state. Otherwise, the WatchStream will contain
- // only changes since the provided ResumeMarker.
+ // only changes since the provided ResumeMarker. See watch.GlobWatcher for a
+ // detailed explanation of the behavior.
//
// The result stream consists of a never-ending sequence of Change messages
// (until the call fails or is canceled). Each Change contains the Name field in
@@ -5589,16 +5854,15 @@
permissions.ObjectServerMethods
// DatabaseWatcher allows a client to watch for updates to the database. For
// each watch request, the client will receive a reliable stream of watch events
- // without re-ordering. See watch.GlobWatcher for a detailed explanation of the
- // behavior.
- // TODO(rogulenko): Currently the only supported watch patterns are
- // "<collectionId>/<rowPrefix>*". Consider changing that.
+ // without re-ordering. Only rows matching at least one of the patterns are
+ // returned. Rows in collections with no Read access are also filtered out.
//
// Watching is done by starting a streaming RPC. The RPC takes a ResumeMarker
// argument that points to a particular place in the database event log. If an
// empty ResumeMarker is provided, the WatchStream will begin with a Change
// batch containing the initial state. Otherwise, the WatchStream will contain
- // only changes since the provided ResumeMarker.
+ // only changes since the provided ResumeMarker. See watch.GlobWatcher for a
+ // detailed explanation of the behavior.
//
// The result stream consists of a never-ending sequence of Change messages
// (until the call fails or is canceled). Each Change contains the Name field in
@@ -5737,16 +6001,15 @@
permissions.ObjectServerStubMethods
// DatabaseWatcher allows a client to watch for updates to the database. For
// each watch request, the client will receive a reliable stream of watch events
- // without re-ordering. See watch.GlobWatcher for a detailed explanation of the
- // behavior.
- // TODO(rogulenko): Currently the only supported watch patterns are
- // "<collectionId>/<rowPrefix>*". Consider changing that.
+ // without re-ordering. Only rows matching at least one of the patterns are
+ // returned. Rows in collections with no Read access are also filtered out.
//
// Watching is done by starting a streaming RPC. The RPC takes a ResumeMarker
// argument that points to a particular place in the database event log. If an
// empty ResumeMarker is provided, the WatchStream will begin with a Change
// batch containing the initial state. Otherwise, the WatchStream will contain
- // only changes since the provided ResumeMarker.
+ // only changes since the provided ResumeMarker. See watch.GlobWatcher for a
+ // detailed explanation of the behavior.
//
// The result stream consists of a never-ending sequence of Change messages
// (until the call fails or is canceled). Each Change contains the Name field in
@@ -5932,7 +6195,7 @@
Doc: "// Database represents a set of Collections. Batches, queries, syncgroups, and\n// watch all operate at the Database level.\n// Database.Glob operates over Collection ids.",
Embeds: []rpc.EmbedDesc{
{"Object", "v.io/v23/services/permissions", "// Object provides access control for Vanadium objects.\n//\n// Vanadium services implementing dynamic access control would typically embed\n// this interface and tag additional methods defined by the service with one of\n// Admin, Read, Write, Resolve etc. For example, the VDL definition of the\n// object would be:\n//\n// package mypackage\n//\n// import \"v.io/v23/security/access\"\n// import \"v.io/v23/services/permissions\"\n//\n// type MyObject interface {\n// permissions.Object\n// MyRead() (string, error) {access.Read}\n// MyWrite(string) error {access.Write}\n// }\n//\n// If the set of pre-defined tags is insufficient, services may define their\n// own tag type and annotate all methods with this new type.\n//\n// Instead of embedding this Object interface, define SetPermissions and\n// GetPermissions in their own interface. Authorization policies will typically\n// respect annotations of a single type. For example, the VDL definition of an\n// object would be:\n//\n// package mypackage\n//\n// import \"v.io/v23/security/access\"\n//\n// type MyTag string\n//\n// const (\n// Blue = MyTag(\"Blue\")\n// Red = MyTag(\"Red\")\n// )\n//\n// type MyObject interface {\n// MyMethod() (string, error) {Blue}\n//\n// // Allow clients to change access via the access.Object interface:\n// SetPermissions(perms access.Permissions, version string) error {Red}\n// GetPermissions() (perms access.Permissions, version string, err error) {Blue}\n// }"},
- {"DatabaseWatcher", "v.io/v23/services/syncbase", "// DatabaseWatcher allows a client to watch for updates to the database. For\n// each watch request, the client will receive a reliable stream of watch events\n// without re-ordering. See watch.GlobWatcher for a detailed explanation of the\n// behavior.\n// TODO(rogulenko): Currently the only supported watch patterns are\n// \"<collectionId>/<rowPrefix>*\". Consider changing that.\n//\n// Watching is done by starting a streaming RPC. The RPC takes a ResumeMarker\n// argument that points to a particular place in the database event log. If an\n// empty ResumeMarker is provided, the WatchStream will begin with a Change\n// batch containing the initial state. Otherwise, the WatchStream will contain\n// only changes since the provided ResumeMarker.\n//\n// The result stream consists of a never-ending sequence of Change messages\n// (until the call fails or is canceled). Each Change contains the Name field in\n// the form \"<collectionId>/<rowKey>\" and the Value field of the StoreChange\n// type. If the client has no access to a row specified in a change, that change\n// is excluded from the result stream.\n//\n// Note: A single Watch Change batch may contain changes from more than one\n// batch as originally committed on a remote Syncbase or obtained from conflict\n// resolution. However, changes from a single original batch will always appear\n// in the same Change batch."},
+ {"DatabaseWatcher", "v.io/v23/services/syncbase", "// DatabaseWatcher allows a client to watch for updates to the database. For\n// each watch request, the client will receive a reliable stream of watch events\n// without re-ordering. Only rows matching at least one of the patterns are\n// returned. Rows in collections with no Read access are also filtered out.\n//\n// Watching is done by starting a streaming RPC. The RPC takes a ResumeMarker\n// argument that points to a particular place in the database event log. If an\n// empty ResumeMarker is provided, the WatchStream will begin with a Change\n// batch containing the initial state. Otherwise, the WatchStream will contain\n// only changes since the provided ResumeMarker. See watch.GlobWatcher for a\n// detailed explanation of the behavior.\n//\n// The result stream consists of a never-ending sequence of Change messages\n// (until the call fails or is canceled). Each Change contains the Name field in\n// the form \"<collectionId>/<rowKey>\" and the Value field of the StoreChange\n// type. If the client has no access to a row specified in a change, that change\n// is excluded from the result stream.\n//\n// Note: A single Watch Change batch may contain changes from more than one\n// batch as originally committed on a remote Syncbase or obtained from conflict\n// resolution. However, changes from a single original batch will always appear\n// in the same Change batch."},
{"SyncgroupManager", "v.io/v23/services/syncbase", "// SyncgroupManager is the interface for syncgroup operations.\n// TODO(hpucha): Add blessings to create/join and add a refresh method."},
{"BlobManager", "v.io/v23/services/syncbase", "// BlobManager is the interface for blob operations.\n//\n// Description of API for resumable blob creation (append-only):\n// - Up until commit, a BlobRef may be used with PutBlob, GetBlobSize,\n// DeleteBlob, and CommitBlob. Blob creation may be resumed by obtaining the\n// current blob size via GetBlobSize and appending to the blob via PutBlob.\n// - After commit, a blob is immutable, at which point PutBlob and CommitBlob\n// may no longer be used.\n// - All other methods (GetBlob, FetchBlob, PinBlob, etc.) may only be used\n// after commit."},
{"SchemaManager", "v.io/v23/services/syncbase", "// SchemaManager implements the API for managing schema metadata attached\n// to a Database."},
@@ -6692,6 +6955,7 @@
__VDLType_enum_34 *vdl.Type
__VDLType_struct_35 *vdl.Type
__VDLType_struct_36 *vdl.Type
+ __VDLType_struct_37 *vdl.Type
)
var __VDLInitCalled bool
@@ -6743,6 +7007,7 @@
vdl.Register((*BlobRef)(nil))
vdl.Register((*BlobFetchState)(nil))
vdl.Register((*BlobFetchStatus)(nil))
+ vdl.Register((*CollectionRowPattern)(nil))
vdl.Register((*StoreChange)(nil))
// Initialize type definitions.
@@ -6781,7 +7046,8 @@
__VDLType_string_33 = vdl.TypeOf((*BlobRef)(nil))
__VDLType_enum_34 = vdl.TypeOf((*BlobFetchState)(nil))
__VDLType_struct_35 = vdl.TypeOf((*BlobFetchStatus)(nil)).Elem()
- __VDLType_struct_36 = vdl.TypeOf((*StoreChange)(nil)).Elem()
+ __VDLType_struct_36 = vdl.TypeOf((*CollectionRowPattern)(nil)).Elem()
+ __VDLType_struct_37 = vdl.TypeOf((*StoreChange)(nil)).Elem()
// Set error format strings.
i18n.Cat().SetWithBase(i18n.LangID("en"), i18n.MsgID(ErrNotInDevMode.ID), "{1:}{2:} not running with --dev=true")
diff --git a/services/syncbase/types.vdl b/services/syncbase/types.vdl
index 092ef8e..cf7251e 100644
--- a/services/syncbase/types.vdl
+++ b/services/syncbase/types.vdl
@@ -388,8 +388,18 @@
Total int64 // Blob size.
}
+// CollectionRowPattern contains SQL LIKE-style glob patterns ('%' and '_'
+// wildcards, '\' as escape character) for matching rows and collections by
+// name components.
+// Collection blessing and name patterns are not allowed to be empty, but the
+// row key pattern is (for matching only collections and no rows).
+type CollectionRowPattern struct {
+ CollectionBlessing string
+ CollectionName string
+ RowKey string
+}
+
// StoreChange is the new value for a watched entity.
-// TODO(rogulenko): Consider adding the Shell state.
type StoreChange struct {
// Value is the new value for the row if the Change state equals to Exists,
// otherwise the Value is nil.
diff --git a/syncbase/.api b/syncbase/.api
index 1e7cea3..ecea5e1 100644
--- a/syncbase/.api
+++ b/syncbase/.api
@@ -12,11 +12,6 @@
pkg syncbase, method (*ConflictRow) VDLRead(vdl.Decoder) error
pkg syncbase, method (*ConflictRowSet) VDLRead(vdl.Decoder) error
pkg syncbase, method (*ConflictScanSet) VDLRead(vdl.Decoder) error
-pkg syncbase, method (*InvalidScanStream) Advance() bool
-pkg syncbase, method (*InvalidScanStream) Cancel()
-pkg syncbase, method (*InvalidScanStream) Err() error
-pkg syncbase, method (*InvalidScanStream) Key() string
-pkg syncbase, method (*InvalidScanStream) Value(interface{}) error
pkg syncbase, method (*Resolution) VDLRead(vdl.Decoder) error
pkg syncbase, method (*ResolvedRow) VDLRead(vdl.Decoder) error
pkg syncbase, method (*Value) Get(interface{}) error
@@ -123,7 +118,7 @@
pkg syncbase, type Database interface, ResumeSync(*context.T) error
pkg syncbase, type Database interface, Syncgroup(*context.T, string) Syncgroup
pkg syncbase, type Database interface, SyncgroupForId(wire.Id) Syncgroup
-pkg syncbase, type Database interface, Watch(*context.T, wire.Id, string, watch.ResumeMarker) (WatchStream, error)
+pkg syncbase, type Database interface, Watch(*context.T, watch.ResumeMarker, []wire.CollectionRowPattern) WatchStream
pkg syncbase, type Database interface, unexported methods
pkg syncbase, type DatabaseHandle interface { Collection, CollectionForId, Exec, FullName, GetResumeMarker, Id, ListCollections }
pkg syncbase, type DatabaseHandle interface, Collection(*context.T, string) Collection
@@ -133,8 +128,6 @@
pkg syncbase, type DatabaseHandle interface, GetResumeMarker(*context.T) (watch.ResumeMarker, error)
pkg syncbase, type DatabaseHandle interface, Id() wire.Id
pkg syncbase, type DatabaseHandle interface, ListCollections(*context.T) ([]wire.Id, error)
-pkg syncbase, type InvalidScanStream struct
-pkg syncbase, type InvalidScanStream struct, Error error
pkg syncbase, type PrefixRange interface { Limit, Prefix, Start }
pkg syncbase, type PrefixRange interface, Limit() string
pkg syncbase, type PrefixRange interface, Prefix() string
diff --git a/syncbase/benchmark_test.go b/syncbase/benchmark_test.go
index 3befa52..c63b919 100644
--- a/syncbase/benchmark_test.go
+++ b/syncbase/benchmark_test.go
@@ -35,6 +35,7 @@
wire "v.io/v23/services/syncbase"
"v.io/v23/services/watch"
"v.io/v23/syncbase"
+ "v.io/v23/syncbase/util"
_ "v.io/x/ref/runtime/factories/roaming"
tu "v.io/x/ref/services/syncbase/testutil"
)
@@ -186,9 +187,11 @@
defer cleanup()
b.ResetTimer()
for i := 0; i < b.N; i++ {
- w, err := d.Watch(ctx, wire.Id{"u", "c"}, "", watch.ResumeMarker("now"))
- if err != nil {
- b.Fatalf("watch error: %v", err)
+ w := d.Watch(ctx, watch.ResumeMarker("now"), []wire.CollectionRowPattern{
+ util.RowPrefixPattern(wire.Id{"u", "c"}, ""),
+ })
+ if w.Err() != nil {
+ b.Fatalf("watch error: %v", w.Err())
}
done := make(chan struct{})
go func() {
@@ -213,11 +216,13 @@
b.Skip("Hangs on occasion, for unknown reasons - v.io/i/1134")
ctx, d, c, cleanup := prepare(b)
defer cleanup()
- w, err := d.Watch(ctx, wire.Id{"u", "c"}, "", watch.ResumeMarker("now"))
- row := make(chan struct{})
- if err != nil {
- b.Fatalf("watch error: %v", err)
+ w := d.Watch(ctx, watch.ResumeMarker("now"), []wire.CollectionRowPattern{
+ util.RowPrefixPattern(wire.Id{"u", "c"}, ""),
+ })
+ if w.Err() != nil {
+ b.Fatalf("watch error: %v", w.Err())
}
+ row := make(chan struct{})
go func() {
seen := 0
for seen < b.N && w.Advance() {
diff --git a/syncbase/client_test.go b/syncbase/client_test.go
index c4bf5a8..4505b41 100644
--- a/syncbase/client_test.go
+++ b/syncbase/client_test.go
@@ -17,6 +17,7 @@
wire "v.io/v23/services/syncbase"
"v.io/v23/services/watch"
"v.io/v23/syncbase"
+ "v.io/v23/syncbase/util"
"v.io/v23/verror"
"v.io/v23/vom"
_ "v.io/x/ref/runtime/factories/roaming"
@@ -541,7 +542,7 @@
}
}
-// TestWatchBasic test the basic client watch functionality: no perms,
+// TestWatchBasic tests the basic client watch functionality: no perms,
// no batches.
func TestWatchBasic(t *testing.T) {
ctx, sName, cleanup := tu.SetupOrDie(nil)
@@ -557,8 +558,8 @@
t.Fatalf("d.GetResumeMarker() failed: %v", err)
}
resumeMarkers = append(resumeMarkers, resumeMarker)
- // Put "abc".
- r := c.Row("abc")
+ // Put "abc%".
+ r := c.Row("abc%")
if err := r.Put(ctx, "value"); err != nil {
t.Fatalf("r.Put() failed: %v", err)
}
@@ -566,7 +567,7 @@
t.Fatalf("d.GetResumeMarker() failed: %v", err)
}
resumeMarkers = append(resumeMarkers, resumeMarker)
- // Delete "abc".
+ // Delete "abc%".
if err := r.Delete(ctx); err != nil {
t.Fatalf("r.Delete() failed: %v", err)
}
@@ -589,7 +590,7 @@
tu.WatchChangeTest{
WatchChange: syncbase.WatchChange{
Collection: tu.CxId("c"),
- Row: "abc",
+ Row: "abc%",
ChangeType: syncbase.PutChange,
ResumeMarker: resumeMarkers[1],
},
@@ -598,7 +599,7 @@
tu.WatchChangeTest{
WatchChange: syncbase.WatchChange{
Collection: tu.CxId("c"),
- Row: "abc",
+ Row: "abc%",
ChangeType: syncbase.DeleteChange,
ResumeMarker: resumeMarkers[2],
},
@@ -615,21 +616,21 @@
}
ctxWithTimeout, cancel := context.WithTimeout(ctx, 10*time.Second)
defer cancel()
- wstream, _ := d.Watch(ctxWithTimeout, tu.CxId("c"), "a", resumeMarkers[0])
+ wstream := d.Watch(ctxWithTimeout, resumeMarkers[0], []wire.CollectionRowPattern{util.RowPrefixPattern(tu.CxId("c"), "a")})
tu.CheckWatch(t, wstream, allChanges)
- wstream, _ = d.Watch(ctxWithTimeout, tu.CxId("c"), "a", resumeMarkers[1])
+ wstream = d.Watch(ctxWithTimeout, resumeMarkers[1], []wire.CollectionRowPattern{util.RowPrefixPattern(tu.CxId("c"), "a")})
tu.CheckWatch(t, wstream, allChanges[1:])
- wstream, _ = d.Watch(ctxWithTimeout, tu.CxId("c"), "a", resumeMarkers[2])
+ wstream = d.Watch(ctxWithTimeout, resumeMarkers[2], []wire.CollectionRowPattern{util.RowPrefixPattern(tu.CxId("c"), "a")})
tu.CheckWatch(t, wstream, allChanges[2:])
- wstream, _ = d.Watch(ctxWithTimeout, tu.CxId("c"), "abc", resumeMarkers[0])
+ wstream = d.Watch(ctxWithTimeout, resumeMarkers[0], []wire.CollectionRowPattern{util.RowPrefixPattern(tu.CxId("c"), "abc")})
tu.CheckWatch(t, wstream, allChanges[:2])
- wstream, _ = d.Watch(ctxWithTimeout, tu.CxId("c"), "abc", resumeMarkers[1])
+ wstream = d.Watch(ctxWithTimeout, resumeMarkers[1], []wire.CollectionRowPattern{util.RowPrefixPattern(tu.CxId("c"), "abc")})
tu.CheckWatch(t, wstream, allChanges[1:2])
}
-// TestWatchWithBatchAndInitialState test that the client watch correctly
-// handles batches and fetching initial state on empty resume marker.
+// TestWatchWithBatchAndInitialState tests that the client watch correctly
+// handles batches, perms, and fetching initial state on empty resume marker.
func TestWatchWithBatchAndInitialState(t *testing.T) {
ctx, adminCtx, sName, rootp, cleanup := tu.SetupOrDieCustom("admin", "server", nil)
defer cleanup()
@@ -666,21 +667,40 @@
ctxWithTimeout, cancel := context.WithTimeout(clientCtx, 10*time.Second)
defer cancel()
+ adminCtxWithTimeout, cancelAdmin := context.WithTimeout(adminCtx, 10*time.Second)
+ defer cancelAdmin()
// Start watches with empty resume marker.
- // TODO(ivanpi): Empty prefix watch should watch both collections using both
- // admin and client contexts, checking that chidden updates are visible only
- // to admin.
- wstreamAll, _ := d.Watch(ctxWithTimeout, tu.CxId("cpublic"), "", nil)
- wstreamD, _ := d.Watch(ctxWithTimeout, tu.CxId("cpublic"), "d", nil)
+ wstreamAll := d.Watch(ctxWithTimeout, nil, []wire.CollectionRowPattern{
+ util.RowPrefixPattern(tu.CxId("cpublic"), ""),
+ util.RowPrefixPattern(tu.CxId("chidden"), ""),
+ })
+ wstreamD := d.Watch(ctxWithTimeout, nil, []wire.CollectionRowPattern{
+ util.RowPrefixPattern(tu.CxId("cpublic"), "d"),
+ })
+ wstreamAllAdmin := d.Watch(adminCtxWithTimeout, nil, []wire.CollectionRowPattern{
+ util.RowPrefixPattern(tu.CxId("cpublic"), ""),
+ util.RowPrefixPattern(tu.CxId("chidden"), ""),
+ })
resumeMarkerInitial, err := d.GetResumeMarker(clientCtx)
if err != nil {
t.Fatalf("d.GetResumeMarker() failed: %v", err)
}
valueBytes, _ := vom.RawBytesFromValue("value")
+
initialChanges := []tu.WatchChangeTest{
tu.WatchChangeTest{
WatchChange: syncbase.WatchChange{
+ Collection: tu.CxId("chidden"),
+ Row: "b/1",
+ ChangeType: syncbase.PutChange,
+ ResumeMarker: nil,
+ Continued: true,
+ },
+ ValueBytes: valueBytes,
+ },
+ tu.WatchChangeTest{
+ WatchChange: syncbase.WatchChange{
Collection: tu.CxId("cpublic"),
Row: "a/1",
ChangeType: syncbase.PutChange,
@@ -701,35 +721,38 @@
}
// Watch with empty prefix should have seen the initial state as one batch,
// omitting the row in the secret collection.
- tu.CheckWatch(t, wstreamAll, initialChanges)
+ tu.CheckWatch(t, wstreamAll, initialChanges[1:])
+ // Admin watch with empty prefix should have seen the full initial state as
+ // one batch.
+ tu.CheckWatch(t, wstreamAllAdmin, initialChanges)
// More writes.
- // Put cp:"a/2" and cs:"b/2" in a batch.
+ // Put ch:"b/2" and cp:"a/2" in a batch.
if err := syncbase.RunInBatch(adminCtx, d, wire.BatchOptions{}, func(b syncbase.BatchDatabase) error {
- cp := b.CollectionForId(tu.CxId("cpublic"))
- if err := cp.Put(adminCtx, "a/2", "value"); err != nil {
+ cp := b.CollectionForId(tu.CxId("chidden"))
+ if err := cp.Put(adminCtx, "b/2", "value"); err != nil {
return err
}
- ch := b.CollectionForId(tu.CxId("chidden"))
- return ch.Put(adminCtx, "b/2", "value")
+ ch := b.CollectionForId(tu.CxId("cpublic"))
+ return ch.Put(adminCtx, "a/2", "value")
}); err != nil {
t.Fatalf("RunInBatch failed: %v", err)
}
- resumeMarkerAfterA2B2, err := d.GetResumeMarker(clientCtx)
+ resumeMarkerAfterB2A2, err := d.GetResumeMarker(clientCtx)
if err != nil {
t.Fatalf("d.GetResumeMarker() failed: %v", err)
}
- // Put cp:"a/3" and cp:"d/1" in a batch.
+ // Put cp:"a/1" (overwrite) and cp:"d/1" in a batch.
if err := syncbase.RunInBatch(adminCtx, d, wire.BatchOptions{}, func(b syncbase.BatchDatabase) error {
cp := b.CollectionForId(tu.CxId("cpublic"))
- if err := cp.Put(adminCtx, "a/3", "value"); err != nil {
+ if err := cp.Put(adminCtx, "a/1", "value"); err != nil {
return err
}
return cp.Put(adminCtx, "d/1", "value")
}); err != nil {
t.Fatalf("RunInBatch failed: %v", err)
}
- resumeMarkerAfterA3D1, err := d.GetResumeMarker(clientCtx)
+ resumeMarkerAfterA1rD1, err := d.GetResumeMarker(clientCtx)
if err != nil {
t.Fatalf("d.GetResumeMarker() failed: %v", err)
}
@@ -737,17 +760,27 @@
continuedChanges := []tu.WatchChangeTest{
tu.WatchChangeTest{
WatchChange: syncbase.WatchChange{
- Collection: tu.CxId("cpublic"),
- Row: "a/2",
+ Collection: tu.CxId("chidden"),
+ Row: "b/2",
ChangeType: syncbase.PutChange,
- ResumeMarker: resumeMarkerAfterA2B2,
+ ResumeMarker: nil,
+ Continued: true,
},
ValueBytes: valueBytes,
},
tu.WatchChangeTest{
WatchChange: syncbase.WatchChange{
Collection: tu.CxId("cpublic"),
- Row: "a/3",
+ Row: "a/2",
+ ChangeType: syncbase.PutChange,
+ ResumeMarker: resumeMarkerAfterB2A2,
+ },
+ ValueBytes: valueBytes,
+ },
+ tu.WatchChangeTest{
+ WatchChange: syncbase.WatchChange{
+ Collection: tu.CxId("cpublic"),
+ Row: "a/1",
ChangeType: syncbase.PutChange,
ResumeMarker: nil,
Continued: true,
@@ -759,17 +792,20 @@
Collection: tu.CxId("cpublic"),
Row: "d/1",
ChangeType: syncbase.PutChange,
- ResumeMarker: resumeMarkerAfterA3D1,
+ ResumeMarker: resumeMarkerAfterA1rD1,
},
ValueBytes: valueBytes,
},
}
// Watch with empty prefix should have seen the continued changes as separate
// batches, omitting rows in the secret collection.
- tu.CheckWatch(t, wstreamAll, continuedChanges)
+ tu.CheckWatch(t, wstreamAll, continuedChanges[1:])
// Watch with prefix "d" should have seen only the last change; its initial
// state was empty.
- tu.CheckWatch(t, wstreamD, continuedChanges[2:])
+ tu.CheckWatch(t, wstreamD, continuedChanges[3:])
+ // Admin watch with empty prefix should have seen all the continued changes as
+ // separate batches.
+ tu.CheckWatch(t, wstreamAllAdmin, continuedChanges)
}
// TestBlockingWatch tests that the server side of the client watch correctly
@@ -786,7 +822,9 @@
}
ctxWithTimeout, cancel := context.WithTimeout(ctx, 10*time.Second)
defer cancel()
- wstream, _ := d.Watch(ctxWithTimeout, tu.CxId("c"), "a", resumeMarker)
+ wstream := d.Watch(ctxWithTimeout, resumeMarker, []wire.CollectionRowPattern{
+ util.RowPrefixPattern(tu.CxId("c"), "a"),
+ })
valueBytes, _ := vom.RawBytesFromValue("value")
for i := 0; i < 10; i++ {
// Put "abc".
@@ -828,8 +866,10 @@
}
ctxWithTimeout, cancel := context.WithTimeout(ctx, 10*time.Second)
defer cancel()
- wstream, err := d.Watch(ctxWithTimeout, tu.CxId("c"), "a", resumeMarker)
- if err != nil {
+ wstream := d.Watch(ctxWithTimeout, resumeMarker, []wire.CollectionRowPattern{
+ util.RowPrefixPattern(tu.CxId("c"), "a"),
+ })
+ if err := wstream.Err(); err != nil {
t.Fatalf("d.Watch() failed: %v", err)
}
time.AfterFunc(500*time.Millisecond, wstream.Cancel)
@@ -840,3 +880,292 @@
t.Errorf("Unexpected wstream error ID: got %v, want %v", got, want)
}
}
+
+// TestWatchMulti tests that watch properly filters collections and includes
+// matching rows only once per update.
+func TestWatchMulti(t *testing.T) {
+ ctx, sName, cleanup := tu.SetupOrDie(nil)
+ defer cleanup()
+ d := tu.CreateDatabase(t, ctx, syncbase.NewService(sName), "d")
+ // Create three collections.
+ cAFoo := d.CollectionForId(wire.Id{"root:alice", "foo"})
+ if err := cAFoo.Create(ctx, nil); err != nil {
+ t.Fatalf("cAFoo.Create() failed: %v", err)
+ }
+ cAFoobar := d.CollectionForId(wire.Id{"root:alice", "foobar"})
+ if err := cAFoobar.Create(ctx, nil); err != nil {
+ t.Fatalf("cAFoobar.Create() failed: %v", err)
+ }
+ cBFoo := d.CollectionForId(wire.Id{"root:%", "foo"})
+ if err := cBFoo.Create(ctx, nil); err != nil {
+ t.Fatalf("cBFoo.Create() failed: %v", err)
+ }
+
+ // Prepopulate data.
+ for _, key := range []string{"a", "abc", "abcd", "cd", "ef", "zcd_"} {
+ if err := cAFoo.Put(ctx, key, "value"); err != nil {
+ t.Fatalf("cAFoo.Put() failed: %v", err)
+ }
+ }
+ for _, key := range []string{"a", "abc", "cd", "ef%", "xv"} {
+ if err := cAFoobar.Put(ctx, key, "value"); err != nil {
+ t.Fatalf("cAFoobar.Put() failed: %v", err)
+ }
+ }
+ for _, key := range []string{"ab", "ef", "efg", "pq", "x\\yz"} {
+ if err := cBFoo.Put(ctx, key, "value"); err != nil {
+ t.Fatalf("cBFoo.Put() failed: %v", err)
+ }
+ }
+
+ ctxWithTimeout, cancel := context.WithTimeout(ctx, 10*time.Second)
+ defer cancel()
+ // Start watches with empty resume marker.
+ // non-overlapping - collections 'foo' and '%bar'
+ wstream1 := d.Watch(ctxWithTimeout, nil, []wire.CollectionRowPattern{
+ {"%:alice", "foo", "%c_"},
+ {"%", "%bar", "a%"},
+ })
+ // prefix pattern - only literal 'root:\%'
+ wstream2 := d.Watch(ctxWithTimeout, nil, []wire.CollectionRowPattern{
+ util.RowPrefixPattern(wire.Id{"root:%", "foo"}, "e"),
+ })
+ // partially overlapping - '%:alice' and 'root:%', '%cd' and 'ab%'
+ wstream3 := d.Watch(ctxWithTimeout, nil, []wire.CollectionRowPattern{
+ {"%:alice", "%", "%cd"},
+ {"root:%", "%", "ab%"},
+ {"root:\\%", "%", "x%"},
+ })
+
+ resumeMarkerInitial, err := d.GetResumeMarker(ctxWithTimeout)
+ if err != nil {
+ t.Fatalf("d.GetResumeMarker() failed: %v", err)
+ }
+ valueBytes, _ := vom.RawBytesFromValue("value")
+
+ initialChanges1 := []tu.WatchChangeTest{
+ tu.WatchChangeTest{
+ WatchChange: syncbase.WatchChange{
+ Collection: wire.Id{"root:alice", "foobar"}, Row: "a",
+ ChangeType: syncbase.PutChange, Continued: true,
+ },
+ ValueBytes: valueBytes,
+ },
+ tu.WatchChangeTest{
+ WatchChange: syncbase.WatchChange{
+ Collection: wire.Id{"root:alice", "foobar"}, Row: "abc",
+ ChangeType: syncbase.PutChange, Continued: true,
+ },
+ ValueBytes: valueBytes,
+ },
+ tu.WatchChangeTest{
+ WatchChange: syncbase.WatchChange{
+ Collection: wire.Id{"root:alice", "foo"}, Row: "abcd",
+ ChangeType: syncbase.PutChange, Continued: true,
+ },
+ ValueBytes: valueBytes,
+ },
+ tu.WatchChangeTest{
+ WatchChange: syncbase.WatchChange{
+ Collection: wire.Id{"root:alice", "foo"}, Row: "cd",
+ ChangeType: syncbase.PutChange, ResumeMarker: resumeMarkerInitial,
+ },
+ ValueBytes: valueBytes,
+ },
+ }
+ tu.CheckWatch(t, wstream1, initialChanges1)
+
+ initialChanges2 := []tu.WatchChangeTest{
+ tu.WatchChangeTest{
+ WatchChange: syncbase.WatchChange{
+ Collection: wire.Id{"root:%", "foo"}, Row: "ef",
+ ChangeType: syncbase.PutChange, Continued: true,
+ },
+ ValueBytes: valueBytes,
+ },
+ tu.WatchChangeTest{
+ WatchChange: syncbase.WatchChange{
+ Collection: wire.Id{"root:%", "foo"}, Row: "efg",
+ ChangeType: syncbase.PutChange, ResumeMarker: resumeMarkerInitial,
+ },
+ ValueBytes: valueBytes,
+ },
+ }
+ tu.CheckWatch(t, wstream2, initialChanges2)
+
+ initialChanges3 := []tu.WatchChangeTest{
+ tu.WatchChangeTest{
+ WatchChange: syncbase.WatchChange{
+ Collection: wire.Id{"root:%", "foo"}, Row: "ab",
+ ChangeType: syncbase.PutChange, Continued: true,
+ },
+ ValueBytes: valueBytes,
+ },
+ tu.WatchChangeTest{
+ WatchChange: syncbase.WatchChange{
+ Collection: wire.Id{"root:%", "foo"}, Row: "x\\yz",
+ ChangeType: syncbase.PutChange, Continued: true,
+ },
+ ValueBytes: valueBytes,
+ },
+ tu.WatchChangeTest{
+ WatchChange: syncbase.WatchChange{
+ Collection: wire.Id{"root:alice", "foobar"}, Row: "abc",
+ ChangeType: syncbase.PutChange, Continued: true,
+ },
+ ValueBytes: valueBytes,
+ },
+ tu.WatchChangeTest{
+ WatchChange: syncbase.WatchChange{
+ Collection: wire.Id{"root:alice", "foobar"}, Row: "cd",
+ ChangeType: syncbase.PutChange, Continued: true,
+ },
+ ValueBytes: valueBytes,
+ },
+ tu.WatchChangeTest{
+ WatchChange: syncbase.WatchChange{
+ Collection: wire.Id{"root:alice", "foo"}, Row: "abc",
+ ChangeType: syncbase.PutChange, Continued: true,
+ },
+ ValueBytes: valueBytes,
+ },
+ tu.WatchChangeTest{
+ WatchChange: syncbase.WatchChange{
+ Collection: wire.Id{"root:alice", "foo"}, Row: "abcd",
+ ChangeType: syncbase.PutChange, Continued: true,
+ },
+ ValueBytes: valueBytes,
+ },
+ tu.WatchChangeTest{
+ WatchChange: syncbase.WatchChange{
+ Collection: wire.Id{"root:alice", "foo"}, Row: "cd",
+ ChangeType: syncbase.PutChange, ResumeMarker: resumeMarkerInitial,
+ },
+ ValueBytes: valueBytes,
+ },
+ }
+ tu.CheckWatch(t, wstream3, initialChanges3)
+
+ // More writes.
+ // Put root:alice,foo:"abcd" and root:alice,foobar:"abcd", delete root:%,foo:"ef%" in a batch.
+ if err := syncbase.RunInBatch(ctxWithTimeout, d, wire.BatchOptions{}, func(b syncbase.BatchDatabase) error {
+ if err := b.CollectionForId(wire.Id{"root:alice", "foo"}).Put(ctxWithTimeout, "abcd", "value"); err != nil {
+ return err
+ }
+ if err := b.CollectionForId(wire.Id{"root:alice", "foobar"}).Put(ctxWithTimeout, "abcd", "value"); err != nil {
+ return err
+ }
+ return b.CollectionForId(wire.Id{"root:%", "foo"}).Delete(ctxWithTimeout, "ef%")
+ }); err != nil {
+ t.Fatalf("RunInBatch failed: %v", err)
+ }
+ resumeMarkerAfterBatch1, err := d.GetResumeMarker(ctxWithTimeout)
+ if err != nil {
+ t.Fatalf("d.GetResumeMarker() failed: %v", err)
+ }
+ // Create collection root:bob,foobar and put "xyz", "acd", "abcd", root:alice,foo:"bcd" in a batch.
+ if err := syncbase.RunInBatch(ctxWithTimeout, d, wire.BatchOptions{}, func(b syncbase.BatchDatabase) error {
+ cNew := b.CollectionForId(wire.Id{"root:bob", "foobar"})
+ if err := cNew.Create(ctxWithTimeout, nil); err != nil {
+ return err
+ }
+ if err := cNew.Put(ctxWithTimeout, "xyz", "value"); err != nil {
+ return err
+ }
+ if err := cNew.Put(ctxWithTimeout, "acd", "value"); err != nil {
+ return err
+ }
+ if err := cNew.Put(ctxWithTimeout, "abcd", "value"); err != nil {
+ return err
+ }
+ return b.CollectionForId(wire.Id{"root:alice", "foo"}).Put(ctxWithTimeout, "bcd", "value")
+ }); err != nil {
+ t.Fatalf("RunInBatch failed: %v", err)
+ }
+ resumeMarkerAfterBatch2, err := d.GetResumeMarker(ctxWithTimeout)
+ _ = resumeMarkerAfterBatch2
+ if err != nil {
+ t.Fatalf("d.GetResumeMarker() failed: %v", err)
+ }
+
+ continuedChanges1 := []tu.WatchChangeTest{
+ tu.WatchChangeTest{
+ WatchChange: syncbase.WatchChange{
+ Collection: wire.Id{"root:alice", "foo"}, Row: "abcd",
+ ChangeType: syncbase.PutChange, Continued: true,
+ },
+ ValueBytes: valueBytes,
+ },
+ tu.WatchChangeTest{
+ WatchChange: syncbase.WatchChange{
+ Collection: wire.Id{"root:alice", "foobar"}, Row: "abcd",
+ ChangeType: syncbase.PutChange, ResumeMarker: resumeMarkerAfterBatch1,
+ },
+ ValueBytes: valueBytes,
+ },
+ tu.WatchChangeTest{
+ WatchChange: syncbase.WatchChange{
+ Collection: wire.Id{"root:bob", "foobar"}, Row: "acd",
+ ChangeType: syncbase.PutChange, Continued: true,
+ },
+ ValueBytes: valueBytes,
+ },
+ tu.WatchChangeTest{
+ WatchChange: syncbase.WatchChange{
+ Collection: wire.Id{"root:bob", "foobar"}, Row: "abcd",
+ ChangeType: syncbase.PutChange, Continued: true,
+ },
+ ValueBytes: valueBytes,
+ },
+ tu.WatchChangeTest{
+ WatchChange: syncbase.WatchChange{
+ Collection: wire.Id{"root:alice", "foo"}, Row: "bcd",
+ ChangeType: syncbase.PutChange, ResumeMarker: resumeMarkerAfterBatch2,
+ },
+ ValueBytes: valueBytes,
+ },
+ }
+ tu.CheckWatch(t, wstream1, continuedChanges1)
+
+ continuedChanges2 := []tu.WatchChangeTest{
+ tu.WatchChangeTest{
+ WatchChange: syncbase.WatchChange{
+ Collection: wire.Id{"root:%", "foo"}, Row: "ef%",
+ ChangeType: syncbase.DeleteChange, ResumeMarker: resumeMarkerAfterBatch1,
+ },
+ },
+ }
+ tu.CheckWatch(t, wstream2, continuedChanges2)
+
+ continuedChanges3 := []tu.WatchChangeTest{
+ tu.WatchChangeTest{
+ WatchChange: syncbase.WatchChange{
+ Collection: wire.Id{"root:alice", "foo"}, Row: "abcd",
+ ChangeType: syncbase.PutChange, Continued: true,
+ },
+ ValueBytes: valueBytes,
+ },
+ tu.WatchChangeTest{
+ WatchChange: syncbase.WatchChange{
+ Collection: wire.Id{"root:alice", "foobar"}, Row: "abcd",
+ ChangeType: syncbase.PutChange, ResumeMarker: resumeMarkerAfterBatch1,
+ },
+ ValueBytes: valueBytes,
+ },
+ tu.WatchChangeTest{
+ WatchChange: syncbase.WatchChange{
+ Collection: wire.Id{"root:bob", "foobar"}, Row: "abcd",
+ ChangeType: syncbase.PutChange, Continued: true,
+ },
+ ValueBytes: valueBytes,
+ },
+ tu.WatchChangeTest{
+ WatchChange: syncbase.WatchChange{
+ Collection: wire.Id{"root:alice", "foo"}, Row: "bcd",
+ ChangeType: syncbase.PutChange, ResumeMarker: resumeMarkerAfterBatch2,
+ },
+ ValueBytes: valueBytes,
+ },
+ }
+ tu.CheckWatch(t, wstream3, continuedChanges3)
+}
diff --git a/syncbase/collection.go b/syncbase/collection.go
index 711b31f..27e6833 100644
--- a/syncbase/collection.go
+++ b/syncbase/collection.go
@@ -96,7 +96,7 @@
ctx, cancel := context.WithCancel(ctx)
call, err := c.c.Scan(ctx, c.bh, []byte(r.Start()), []byte(r.Limit()))
if err != nil {
- return &InvalidScanStream{Error: err}
+ return &invalidScanStream{invalidStream{err: err}}
}
return newScanStream(cancel, call)
}
diff --git a/syncbase/database.go b/syncbase/database.go
index ed66afa..7fe5e0a 100644
--- a/syncbase/database.go
+++ b/syncbase/database.go
@@ -9,7 +9,6 @@
"time"
"v.io/v23/context"
- "v.io/v23/naming"
"v.io/v23/security/access"
wire "v.io/v23/services/syncbase"
"v.io/v23/services/watch"
@@ -105,19 +104,13 @@
}
// Watch implements Database.Watch.
-func (d *database) Watch(ctx *context.T, collection wire.Id, prefix string, resumeMarker watch.ResumeMarker) (WatchStream, error) {
- if err := util.ValidateId(collection); err != nil {
- return nil, verror.New(wire.ErrInvalidName, ctx, collection, err)
- }
+func (d *database) Watch(ctx *context.T, resumeMarker watch.ResumeMarker, patterns []wire.CollectionRowPattern) WatchStream {
ctx, cancel := context.WithCancel(ctx)
- call, err := d.c.WatchGlob(ctx, watch.GlobRequest{
- Pattern: naming.Join(util.EncodeId(collection), prefix+"*"),
- ResumeMarker: resumeMarker,
- })
+ call, err := d.c.WatchPatterns(ctx, resumeMarker, patterns)
if err != nil {
- return nil, err
+ return &invalidWatchStream{invalidStream{err: err}}
}
- return newWatchStream(cancel, call), nil
+ return newWatchStream(cancel, call)
}
// SyncgroupForId implements Database.SyncgroupForId.
diff --git a/syncbase/exec_test.go b/syncbase/exec_test.go
index f40abff..112d7c3 100644
--- a/syncbase/exec_test.go
+++ b/syncbase/exec_test.go
@@ -13,6 +13,7 @@
"time"
"v.io/v23/context"
+ "v.io/v23/query/pattern"
"v.io/v23/query/syncql"
"v.io/v23/syncbase"
"v.io/v23/syncbase/testdata"
@@ -2451,11 +2452,11 @@
},
{
"select k from Customer where v.Name like \"a^b%\" escape '^'",
- syncql.NewErrInvalidEscapeSequence(ctx, 41),
+ syncql.NewErrInvalidLikePattern(ctx, 41, pattern.NewErrInvalidEscape(nil, "b")),
},
{
"select k from Customer where v.Name like \"John^Smith\" escape '^'",
- syncql.NewErrInvalidEscapeSequence(ctx, 41),
+ syncql.NewErrInvalidLikePattern(ctx, 41, pattern.NewErrInvalidEscape(nil, "S")),
},
{
"select a from Customer",
@@ -2502,7 +2503,7 @@
},
{
"select k from Customer escape ' '",
- syncql.NewErrInvalidEscapeChar(ctx, 30),
+ syncql.NewErrInvalidEscapeChar(ctx, 30, " "),
},
{
"select K from Customer",
diff --git a/syncbase/featuretests/client_v23_test.go b/syncbase/featuretests/client_v23_test.go
index 323cc97..7ff0e4e 100644
--- a/syncbase/featuretests/client_v23_test.go
+++ b/syncbase/featuretests/client_v23_test.go
@@ -9,7 +9,9 @@
"time"
"v.io/v23/context"
+ wire "v.io/v23/services/syncbase"
"v.io/v23/syncbase"
+ "v.io/v23/syncbase/util"
"v.io/x/ref/services/syncbase/syncbaselib"
"v.io/x/ref/test/v23test"
)
@@ -58,10 +60,7 @@
// Do a watch from the resume marker before the put operation.
ctxWithTimeout, cancel := context.WithTimeout(ctx, 10*time.Second)
defer cancel()
- stream, err := d.Watch(ctxWithTimeout, testCx, "", marker)
- if err != nil {
- t.Fatalf("unable to start a watch %v", err)
- }
+ stream := d.Watch(ctxWithTimeout, marker, []wire.CollectionRowPattern{util.RowPrefixPattern(testCx, "")})
if !stream.Advance() {
t.Fatalf("watch stream unexpectedly reached the end: %v", stream.Err())
}
diff --git a/syncbase/featuretests/ping_pong_test.go b/syncbase/featuretests/ping_pong_test.go
index c30e2d6..eabed2a 100644
--- a/syncbase/featuretests/ping_pong_test.go
+++ b/syncbase/featuretests/ping_pong_test.go
@@ -14,6 +14,7 @@
wire "v.io/v23/services/syncbase"
"v.io/v23/services/watch"
"v.io/v23/syncbase"
+ "v.io/v23/syncbase/util"
"v.io/x/ref/test/v23test"
)
@@ -78,10 +79,14 @@
// Set up the watch streams (watching the other syncbase's prefix).
prefix0, prefix1 := "prefix0", "prefix1"
- w0, err := db0.Watch(sbs[0].clientCtx, testCx, prefix1, watch.ResumeMarker("now"))
- ok(b, err)
- w1, err := db1.Watch(sbs[1].clientCtx, testCx, prefix0, watch.ResumeMarker("now"))
- ok(b, err)
+ w0 := db0.Watch(sbs[0].clientCtx, watch.ResumeMarker("now"), []wire.CollectionRowPattern{
+ util.RowPrefixPattern(testCx, prefix1),
+ })
+ ok(b, w0.Err())
+ w1 := db1.Watch(sbs[1].clientCtx, watch.ResumeMarker("now"), []wire.CollectionRowPattern{
+ util.RowPrefixPattern(testCx, prefix0),
+ })
+ ok(b, w0.Err())
// The join has succeeded, so make sure sync is initialized.
// The strategy is: s0 sends to s1, and then s1 responds.
diff --git a/syncbase/featuretests/restartability_v23_test.go b/syncbase/featuretests/restartability_v23_test.go
index b1084f8..316884b 100644
--- a/syncbase/featuretests/restartability_v23_test.go
+++ b/syncbase/featuretests/restartability_v23_test.go
@@ -17,6 +17,7 @@
"v.io/v23/context"
wire "v.io/v23/services/syncbase"
"v.io/v23/syncbase"
+ "v.io/v23/syncbase/util"
"v.io/v23/verror"
"v.io/x/ref/services/syncbase/syncbaselib"
tu "v.io/x/ref/services/syncbase/testutil"
@@ -314,10 +315,7 @@
// Watch for the row change.
timeout, _ := context.WithTimeout(clientCtx, time.Second)
- stream, err := d.Watch(timeout, testCx, "r", marker)
- if err != nil {
- t.Fatalf("unexpected error: %v", err)
- }
+ stream := d.Watch(timeout, marker, []wire.CollectionRowPattern{util.RowPrefixPattern(testCx, "r")})
if !stream.Advance() {
cleanup(os.Interrupt)
t.Fatalf("expected to be able to Advance: %v", stream.Err())
@@ -348,10 +346,7 @@
// Resume the watch from after the first Put. We should see only the second
// Put.
- stream, err = d.Watch(clientCtx, testCx, "", marker)
- if err != nil {
- t.Fatalf("unexpected error: %v", err)
- }
+ stream = d.Watch(clientCtx, marker, []wire.CollectionRowPattern{util.RowPrefixPattern(testCx, "")})
if !stream.Advance() {
t.Fatalf("expected to be able to Advance: %v", stream.Err())
}
diff --git a/syncbase/featuretests/sync_v23_test.go b/syncbase/featuretests/sync_v23_test.go
index e87b758..d036255 100644
--- a/syncbase/featuretests/sync_v23_test.go
+++ b/syncbase/featuretests/sync_v23_test.go
@@ -20,6 +20,7 @@
wire "v.io/v23/services/syncbase"
"v.io/v23/services/watch"
"v.io/v23/syncbase"
+ "v.io/v23/syncbase/util"
"v.io/v23/verror"
_ "v.io/x/ref/runtime/factories/roaming"
"v.io/x/ref/services/syncbase/syncbaselib"
@@ -555,18 +556,17 @@
ctxWithTimeout, cancel := context.WithTimeout(ctx, 10*time.Second)
defer cancel()
- stream, err := d.Watch(ctxWithTimeout, collectionId, keyPrefix, beforeSyncMarker)
- if err != nil {
- return fmt.Errorf("watch error: %v\n", err)
- }
+ stream := d.Watch(ctxWithTimeout, beforeSyncMarker, []wire.CollectionRowPattern{
+ util.RowPrefixPattern(collectionId, keyPrefix),
+ })
var changes []syncbase.WatchChange
- for i := 0; stream.Advance() && i < count; i++ {
- if err := stream.Err(); err != nil {
- return fmt.Errorf("watch stream error: %v\n", err)
- }
+ for i := 0; i < count && stream.Advance(); i++ {
changes = append(changes, stream.Change())
}
+ if err := stream.Err(); err != nil {
+ return fmt.Errorf("watch stream error: %v\n", err)
+ }
sort.Sort(ByRow(changes))
diff --git a/syncbase/invalid_types.go b/syncbase/invalid_types.go
index c5d221e..fde5c81 100644
--- a/syncbase/invalid_types.go
+++ b/syncbase/invalid_types.go
@@ -4,39 +4,52 @@
package syncbase
-// InvalidScanStream is a ScanStream for which all methods return errors.
-// TODO(sadovsky): Make InvalidScanStream private.
-type InvalidScanStream struct {
- Error error // returned by all methods
+// invalidStream is a Stream for which all methods return errors.
+type invalidStream struct {
+ err error // returned by all methods
}
-var (
- _ ScanStream = (*InvalidScanStream)(nil)
-)
-
-////////////////////////////////////////////////////////////
-// InvalidScanStream
+var _ Stream = (*invalidStream)(nil)
// Advance implements the Stream interface.
-func (s *InvalidScanStream) Advance() bool {
+func (_ *invalidStream) Advance() bool {
return false
}
// Err implements the Stream interface.
-func (s *InvalidScanStream) Err() error {
- return s.Error
+func (s *invalidStream) Err() error {
+ return s.err
}
// Cancel implements the Stream interface.
-func (s *InvalidScanStream) Cancel() {
+func (_ *invalidStream) Cancel() {
}
+// invalidScanStream is a ScanStream for which all methods return errors.
+type invalidScanStream struct {
+ invalidStream
+}
+
+var _ ScanStream = (*invalidScanStream)(nil)
+
// Key implements the ScanStream interface.
-func (s *InvalidScanStream) Key() string {
- panic(s.Error)
+func (s *invalidScanStream) Key() string {
+ panic(s.err)
}
// Value implements the ScanStream interface.
-func (s *InvalidScanStream) Value(value interface{}) error {
- panic(s.Error)
+func (s *invalidScanStream) Value(value interface{}) error {
+ panic(s.err)
+}
+
+// invalidWatchStream is a WatchStream for which all methods return errors.
+type invalidWatchStream struct {
+ invalidStream
+}
+
+var _ WatchStream = (*invalidWatchStream)(nil)
+
+// Change implements the WatchStream interface.
+func (s *invalidWatchStream) Change() WatchChange {
+ panic(s.err)
}
diff --git a/syncbase/model.go b/syncbase/model.go
index 86e7b79..ab22e0b 100644
--- a/syncbase/model.go
+++ b/syncbase/model.go
@@ -142,18 +142,17 @@
// interface.
util.AccessController
- // Watch allows a client to watch for updates to the database. For each watch
- // request, the client will receive a reliable stream of watch events without
- // reordering. See watch.GlobWatcher for a detailed explanation of the
- // behavior.
+ // Watch allows a client to watch for updates to the database. At least one
+ // pattern must be specified. For each watch request, the client will receive
+ // a reliable stream of watch events without reordering. Only rows matching at
+ // least one of the patterns are returned. Rows in collections with no Read
+ // access are also filtered out.
//
// If a nil ResumeMarker is provided, the WatchStream will begin with a Change
// batch containing the initial state. Otherwise, the WatchStream will contain
// only changes since the provided ResumeMarker.
- //
- // TODO(sadovsky): Watch should return just a WatchStream, similar to how Scan
- // returns just a ScanStream.
- Watch(ctx *context.T, collection wire.Id, prefix string, resumeMarker watch.ResumeMarker) (WatchStream, error)
+ // See watch.GlobWatcher for a detailed explanation of the behavior.
+ Watch(ctx *context.T, resumeMarker watch.ResumeMarker, patterns []wire.CollectionRowPattern) WatchStream
// SyncgroupForId returns a handle to the syncgroup with the given Id.
SyncgroupForId(id wire.Id) Syncgroup
diff --git a/syncbase/util/.api b/syncbase/util/.api
index 3e388b2..7210dd2 100644
--- a/syncbase/util/.api
+++ b/syncbase/util/.api
@@ -8,6 +8,7 @@
pkg util, func ParseCollectionRowPair(*context.T, string) (wire.Id, string, error)
pkg util, func PrefixRangeLimit(string) string
pkg util, func PrefixRangeStart(string) string
+pkg util, func RowPrefixPattern(wire.Id, string) wire.CollectionRowPattern
pkg util, func SortIds([]wire.Id)
pkg util, func UserBlessingFromContext(*context.T) (string, error)
pkg util, func ValidateId(wire.Id) error
diff --git a/syncbase/util/util.go b/syncbase/util/util.go
index 6d4365d..f6e0982 100644
--- a/syncbase/util/util.go
+++ b/syncbase/util/util.go
@@ -11,6 +11,7 @@
"v.io/v23/context"
"v.io/v23/naming"
+ "v.io/v23/query/pattern"
"v.io/v23/security"
"v.io/v23/security/access"
wire "v.io/v23/services/syncbase"
@@ -142,6 +143,16 @@
return collection, row, nil
}
+// RowPrefixPattern returns a CollectionRowPattern matching a single key prefix
+// in a single collection.
+func RowPrefixPattern(cxId wire.Id, keyPrefix string) wire.CollectionRowPattern {
+ return wire.CollectionRowPattern{
+ CollectionBlessing: pattern.Escape(cxId.Blessing),
+ CollectionName: pattern.Escape(cxId.Name),
+ RowKey: pattern.Escape(keyPrefix) + "%",
+ }
+}
+
// PrefixRangeStart returns the start of the row range for the given prefix.
func PrefixRangeStart(p string) string {
return p
