Add expected database schema version to RPCs.
1) Client side
- Minimum changes to client API. Database version
is piped around internally for use in RPC.
- Schema version added to BatchOptions during
creation of batch.
- RPC API change do pass a int64 schema version number
as param for all read write methods of database, table
and row.
2) Server side
- schema version number is stored in databaseData object
- schema version is read from db for each write operation
that needs to check version. In future leveldb is expected
to have internal caching layer making the read op
inexpensive.
MultiPart: 1/2
Change-Id: Ic7443e9087cf51fb96cb35b30918fa24665dbcbc
diff --git a/v23/services/syncbase/nosql/service.vdl b/v23/services/syncbase/nosql/service.vdl
index 0fc865e..465ca69 100644
--- a/v23/services/syncbase/nosql/service.vdl
+++ b/v23/services/syncbase/nosql/service.vdl
@@ -14,45 +14,47 @@
// Database represents a collection of Tables. Batches, queries, sync, watch,
// etc. all operate at the Database level.
// Database.Glob operates over Table names.
+// Param schemaVersion is the version number that the client expects the database
+// to be at. To disable schema version checking, pass -1.
//
// TODO(sadovsky): Add Watch method.
type Database interface {
// Create creates this Database.
// If perms is nil, we inherit (copy) the App perms.
// Create requires the caller to have Write permission at the App.
- Create(perms access.Permissions, metadata ?SchemaMetadata) error {access.Write}
+ Create(metadata ?SchemaMetadata, perms access.Permissions) error {access.Write}
// Delete deletes this Database.
- Delete() error {access.Write}
+ Delete(schemaVersion int32) error {access.Write}
// Exists returns true only if this Database exists. Insufficient permissions
// cause Exists to return false instead of an error.
// TODO(ivanpi): Exists may fail with an error if higher levels of hierarchy
// do not exist.
- Exists() (bool | error) {access.Read}
+ Exists(schemaVersion int32) (bool | error) {access.Read}
// BeginBatch creates a new batch. It returns an App-relative name for a
// Database handle bound to this batch. If this Database is already bound to a
// batch, BeginBatch() will fail with ErrBoundToBatch. Concurrency semantics
// are documented in model.go.
// TODO(sadovsky): make BatchOptions optional
- BeginBatch(bo BatchOptions) (string | error) {access.Read}
+ BeginBatch(schemaVersion int32, bo BatchOptions) (string | error) {access.Read}
// Commit persists the pending changes to the database.
// If this Database is not bound to a batch, Commit() will fail with
// ErrNotBoundToBatch.
- Commit() error {access.Read}
+ Commit(schemaVersion int32) error {access.Read}
// Exec executes a syncQL query and returns all results as specified by in the
// query's select clause. Concurrency semantics are documented in model.go.
- Exec(query string) stream<_, []any> error {access.Read}
+ Exec(schemaVersion int32, query string) stream<_, []any> error {access.Read}
// Abort notifies the server that any pending changes can be discarded.
// It is not strictly required, but it may allow the server to release locks
// or other resources sooner than if it was not called.
// If this Database is not bound to a batch, Abort() will fail with
// ErrNotBoundToBatch.
- Abort() error {access.Read}
+ Abort(schemaVersion int32) error {access.Read}
// SetPermissions and GetPermissions are included from the Object interface.
permissions.Object
@@ -75,35 +77,37 @@
// Table represents a collection of Rows.
// Table.Glob operates over the primary keys of Rows in the Table.
+// SchemaVersion is the version number that the client expects the database
+// to be at. To disable schema version checking, pass -1.
type Table interface {
// Create creates this Table.
// If perms is nil, we inherit (copy) the Database perms.
- Create(perms access.Permissions) error {access.Write}
+ Create(schemaVersion int32, perms access.Permissions) error {access.Write}
// Delete deletes this Table.
- Delete() error {access.Write}
+ Delete(schemaVersion int32) error {access.Write}
// Exists returns true only if this Table exists. Insufficient permissions
// cause Exists to return false instead of an error.
// TODO(ivanpi): Exists may fail with an error if higher levels of hierarchy
// do not exist.
- Exists() (bool | error) {access.Read}
+ Exists(schemaVersion int32) (bool | error) {access.Read}
// Delete deletes all rows in the given half-open range [start, limit). If
// limit is "", all rows with keys >= start are included.
- DeleteRowRange(start, limit []byte) error {access.Write}
+ DeleteRowRange(schemaVersion int32, start, limit []byte) error {access.Write}
// Scan returns all rows in the given half-open range [start, limit). If limit
// is "", all rows with keys >= start are included. Concurrency semantics are
// documented in model.go.
- Scan(start, limit []byte) stream<_, KeyValue> error {access.Read}
+ Scan(schemaVersion int32, start, limit []byte) stream<_, KeyValue> error {access.Read}
// GetPermissions returns an array of (prefix, perms) pairs. The array is
// sorted from longest prefix to shortest, so element zero is the one that
// applies to the row with the given key. The last element is always the
// prefix "" which represents the table's permissions -- the array will always
// have at least one element.
- GetPermissions(key string) ([]PrefixPermissions | error) {access.Admin}
+ GetPermissions(schemaVersion int32, key string) ([]PrefixPermissions | error) {access.Admin}
// SetPermissions sets the permissions for all current and future rows with
// the given prefix. If the prefix overlaps with an existing prefix, the
@@ -112,17 +116,19 @@
// SetPermissions(ctx, Prefix("a/b/c"), perms2)
// The permissions for row "a/b/1" are perms1, and the permissions for row
// "a/b/c/1" are perms2.
- SetPermissions(prefix string, perms access.Permissions) error {access.Admin}
+ SetPermissions(schemaVersion int32, prefix string, perms access.Permissions) error {access.Admin}
// DeletePermissions deletes the permissions for the specified prefix. Any
// rows covered by this prefix will use the next longest prefix's permissions
// (see the array returned by GetPermissions).
- DeletePermissions(prefix string) error {access.Admin}
+ DeletePermissions(schemaVersion int32, prefix string) error {access.Admin}
}
// Row represents a single row in a Table.
// All access checks are performed against the most specific matching prefix
// permissions in the Table.
+// SchemaVersion is the version number that the client expects the database
+// to be at. To disable schema version checking, pass -1.
// NOTE(sadovsky): Currently we send []byte values over the wire for Get, Put,
// and Scan. If there's a way to avoid encoding/decoding on the server side, we
// can use vdl.Value everywhere without sacrificing performance.
@@ -131,16 +137,16 @@
// cause Exists to return false instead of an error.
// TODO(ivanpi): Exists may fail with an error if higher levels of hierarchy
// do not exist.
- Exists() (bool | error) {access.Read}
+ Exists(schemaVersion int32) (bool | error) {access.Read}
// Get returns the value for this Row.
- Get() ([]byte | error) {access.Read}
+ Get(schemaVersion int32) ([]byte | error) {access.Read}
// Put writes the given value for this Row.
- Put(value []byte) error {access.Write}
+ Put(schemaVersion int32, value []byte) error {access.Write}
// Delete deletes this Row.
- Delete() error {access.Write}
+ Delete(schemaVersion int32) error {access.Write}
}
// SyncGroupManager is the interface for SyncGroup operations.
@@ -314,4 +320,5 @@
NotBoundToBatch() {"en": "not bound to batch"}
ReadOnlyBatch() {"en": "batch is read-only"}
ConcurrentBatch() {"en": "concurrent batch"}
+ SchemaVersionMismatch() {"en": "actual schema version does not match the provided one"}
)
diff --git a/v23/services/syncbase/nosql/service.vdl.go b/v23/services/syncbase/nosql/service.vdl.go
index c5af13f..da6bf6d 100644
--- a/v23/services/syncbase/nosql/service.vdl.go
+++ b/v23/services/syncbase/nosql/service.vdl.go
@@ -25,10 +25,11 @@
)
var (
- ErrBoundToBatch = verror.Register("v.io/syncbase/v23/services/syncbase/nosql.BoundToBatch", verror.NoRetry, "{1:}{2:} bound to batch")
- ErrNotBoundToBatch = verror.Register("v.io/syncbase/v23/services/syncbase/nosql.NotBoundToBatch", verror.NoRetry, "{1:}{2:} not bound to batch")
- ErrReadOnlyBatch = verror.Register("v.io/syncbase/v23/services/syncbase/nosql.ReadOnlyBatch", verror.NoRetry, "{1:}{2:} batch is read-only")
- ErrConcurrentBatch = verror.Register("v.io/syncbase/v23/services/syncbase/nosql.ConcurrentBatch", verror.NoRetry, "{1:}{2:} concurrent batch")
+ ErrBoundToBatch = verror.Register("v.io/syncbase/v23/services/syncbase/nosql.BoundToBatch", verror.NoRetry, "{1:}{2:} bound to batch")
+ ErrNotBoundToBatch = verror.Register("v.io/syncbase/v23/services/syncbase/nosql.NotBoundToBatch", verror.NoRetry, "{1:}{2:} not bound to batch")
+ ErrReadOnlyBatch = verror.Register("v.io/syncbase/v23/services/syncbase/nosql.ReadOnlyBatch", verror.NoRetry, "{1:}{2:} batch is read-only")
+ ErrConcurrentBatch = verror.Register("v.io/syncbase/v23/services/syncbase/nosql.ConcurrentBatch", verror.NoRetry, "{1:}{2:} concurrent batch")
+ ErrSchemaVersionMismatch = verror.Register("v.io/syncbase/v23/services/syncbase/nosql.SchemaVersionMismatch", verror.NoRetry, "{1:}{2:} actual schema version does not match the provided one")
)
func init() {
@@ -36,6 +37,7 @@
i18n.Cat().SetWithBase(i18n.LangID("en"), i18n.MsgID(ErrNotBoundToBatch.ID), "{1:}{2:} not bound to batch")
i18n.Cat().SetWithBase(i18n.LangID("en"), i18n.MsgID(ErrReadOnlyBatch.ID), "{1:}{2:} batch is read-only")
i18n.Cat().SetWithBase(i18n.LangID("en"), i18n.MsgID(ErrConcurrentBatch.ID), "{1:}{2:} concurrent batch")
+ i18n.Cat().SetWithBase(i18n.LangID("en"), i18n.MsgID(ErrSchemaVersionMismatch.ID), "{1:}{2:} actual schema version does not match the provided one")
}
// NewErrBoundToBatch returns an error with the ErrBoundToBatch ID.
@@ -58,6 +60,11 @@
return verror.New(ErrConcurrentBatch, ctx)
}
+// NewErrSchemaVersionMismatch returns an error with the ErrSchemaVersionMismatch ID.
+func NewErrSchemaVersionMismatch(ctx *context.T) error {
+ return verror.New(ErrSchemaVersionMismatch, ctx)
+}
+
// DatabaseWatcherClientMethods is the client interface
// containing DatabaseWatcher methods.
//
@@ -1431,6 +1438,8 @@
// Database represents a collection of Tables. Batches, queries, sync, watch,
// etc. all operate at the Database level.
// Database.Glob operates over Table names.
+// Param schemaVersion is the version number that the client expects the database
+// to be at. To disable schema version checking, pass -1.
//
// TODO(sadovsky): Add Watch method.
type DatabaseClientMethods interface {
@@ -1511,33 +1520,33 @@
// Create creates this Database.
// If perms is nil, we inherit (copy) the App perms.
// Create requires the caller to have Write permission at the App.
- Create(ctx *context.T, perms access.Permissions, metadata *SchemaMetadata, opts ...rpc.CallOpt) error
+ Create(ctx *context.T, metadata *SchemaMetadata, perms access.Permissions, opts ...rpc.CallOpt) error
// Delete deletes this Database.
- Delete(*context.T, ...rpc.CallOpt) error
+ Delete(ctx *context.T, schemaVersion int32, opts ...rpc.CallOpt) error
// Exists returns true only if this Database exists. Insufficient permissions
// cause Exists to return false instead of an error.
// TODO(ivanpi): Exists may fail with an error if higher levels of hierarchy
// do not exist.
- Exists(*context.T, ...rpc.CallOpt) (bool, error)
+ Exists(ctx *context.T, schemaVersion int32, opts ...rpc.CallOpt) (bool, error)
// BeginBatch creates a new batch. It returns an App-relative name for a
// Database handle bound to this batch. If this Database is already bound to a
// batch, BeginBatch() will fail with ErrBoundToBatch. Concurrency semantics
// are documented in model.go.
// TODO(sadovsky): make BatchOptions optional
- BeginBatch(ctx *context.T, bo BatchOptions, opts ...rpc.CallOpt) (string, error)
+ BeginBatch(ctx *context.T, schemaVersion int32, bo BatchOptions, opts ...rpc.CallOpt) (string, error)
// Commit persists the pending changes to the database.
// If this Database is not bound to a batch, Commit() will fail with
// ErrNotBoundToBatch.
- Commit(*context.T, ...rpc.CallOpt) error
+ Commit(ctx *context.T, schemaVersion int32, opts ...rpc.CallOpt) error
// Exec executes a syncQL query and returns all results as specified by in the
// query's select clause. Concurrency semantics are documented in model.go.
- Exec(ctx *context.T, query string, opts ...rpc.CallOpt) (DatabaseExecClientCall, error)
+ Exec(ctx *context.T, schemaVersion int32, query string, opts ...rpc.CallOpt) (DatabaseExecClientCall, error)
// Abort notifies the server that any pending changes can be discarded.
// It is not strictly required, but it may allow the server to release locks
// or other resources sooner than if it was not called.
// If this Database is not bound to a batch, Abort() will fail with
// ErrNotBoundToBatch.
- Abort(*context.T, ...rpc.CallOpt) error
+ Abort(ctx *context.T, schemaVersion int32, opts ...rpc.CallOpt) error
}
// DatabaseClientStub adds universal methods to DatabaseClientMethods.
@@ -1561,42 +1570,42 @@
SchemaManagerClientStub
}
-func (c implDatabaseClientStub) Create(ctx *context.T, i0 access.Permissions, i1 *SchemaMetadata, opts ...rpc.CallOpt) (err error) {
+func (c implDatabaseClientStub) Create(ctx *context.T, i0 *SchemaMetadata, i1 access.Permissions, opts ...rpc.CallOpt) (err error) {
err = v23.GetClient(ctx).Call(ctx, c.name, "Create", []interface{}{i0, i1}, nil, opts...)
return
}
-func (c implDatabaseClientStub) Delete(ctx *context.T, opts ...rpc.CallOpt) (err error) {
- err = v23.GetClient(ctx).Call(ctx, c.name, "Delete", nil, nil, opts...)
+func (c implDatabaseClientStub) Delete(ctx *context.T, i0 int32, opts ...rpc.CallOpt) (err error) {
+ err = v23.GetClient(ctx).Call(ctx, c.name, "Delete", []interface{}{i0}, nil, opts...)
return
}
-func (c implDatabaseClientStub) Exists(ctx *context.T, opts ...rpc.CallOpt) (o0 bool, err error) {
- err = v23.GetClient(ctx).Call(ctx, c.name, "Exists", nil, []interface{}{&o0}, opts...)
+func (c implDatabaseClientStub) Exists(ctx *context.T, i0 int32, opts ...rpc.CallOpt) (o0 bool, err error) {
+ err = v23.GetClient(ctx).Call(ctx, c.name, "Exists", []interface{}{i0}, []interface{}{&o0}, opts...)
return
}
-func (c implDatabaseClientStub) BeginBatch(ctx *context.T, i0 BatchOptions, opts ...rpc.CallOpt) (o0 string, err error) {
- err = v23.GetClient(ctx).Call(ctx, c.name, "BeginBatch", []interface{}{i0}, []interface{}{&o0}, opts...)
+func (c implDatabaseClientStub) BeginBatch(ctx *context.T, i0 int32, i1 BatchOptions, opts ...rpc.CallOpt) (o0 string, err error) {
+ err = v23.GetClient(ctx).Call(ctx, c.name, "BeginBatch", []interface{}{i0, i1}, []interface{}{&o0}, opts...)
return
}
-func (c implDatabaseClientStub) Commit(ctx *context.T, opts ...rpc.CallOpt) (err error) {
- err = v23.GetClient(ctx).Call(ctx, c.name, "Commit", nil, nil, opts...)
+func (c implDatabaseClientStub) Commit(ctx *context.T, i0 int32, opts ...rpc.CallOpt) (err error) {
+ err = v23.GetClient(ctx).Call(ctx, c.name, "Commit", []interface{}{i0}, nil, opts...)
return
}
-func (c implDatabaseClientStub) Exec(ctx *context.T, i0 string, opts ...rpc.CallOpt) (ocall DatabaseExecClientCall, err error) {
+func (c implDatabaseClientStub) Exec(ctx *context.T, i0 int32, i1 string, opts ...rpc.CallOpt) (ocall DatabaseExecClientCall, err error) {
var call rpc.ClientCall
- if call, err = v23.GetClient(ctx).StartCall(ctx, c.name, "Exec", []interface{}{i0}, opts...); err != nil {
+ if call, err = v23.GetClient(ctx).StartCall(ctx, c.name, "Exec", []interface{}{i0, i1}, opts...); err != nil {
return
}
ocall = &implDatabaseExecClientCall{ClientCall: call}
return
}
-func (c implDatabaseClientStub) Abort(ctx *context.T, opts ...rpc.CallOpt) (err error) {
- err = v23.GetClient(ctx).Call(ctx, c.name, "Abort", nil, nil, opts...)
+func (c implDatabaseClientStub) Abort(ctx *context.T, i0 int32, opts ...rpc.CallOpt) (err error) {
+ err = v23.GetClient(ctx).Call(ctx, c.name, "Abort", []interface{}{i0}, nil, opts...)
return
}
@@ -1674,6 +1683,8 @@
// Database represents a collection of Tables. Batches, queries, sync, watch,
// etc. all operate at the Database level.
// Database.Glob operates over Table names.
+// Param schemaVersion is the version number that the client expects the database
+// to be at. To disable schema version checking, pass -1.
//
// TODO(sadovsky): Add Watch method.
type DatabaseServerMethods interface {
@@ -1754,33 +1765,33 @@
// Create creates this Database.
// If perms is nil, we inherit (copy) the App perms.
// Create requires the caller to have Write permission at the App.
- Create(ctx *context.T, call rpc.ServerCall, perms access.Permissions, metadata *SchemaMetadata) error
+ Create(ctx *context.T, call rpc.ServerCall, metadata *SchemaMetadata, perms access.Permissions) error
// Delete deletes this Database.
- Delete(*context.T, rpc.ServerCall) error
+ Delete(ctx *context.T, call rpc.ServerCall, schemaVersion int32) error
// Exists returns true only if this Database exists. Insufficient permissions
// cause Exists to return false instead of an error.
// TODO(ivanpi): Exists may fail with an error if higher levels of hierarchy
// do not exist.
- Exists(*context.T, rpc.ServerCall) (bool, error)
+ Exists(ctx *context.T, call rpc.ServerCall, schemaVersion int32) (bool, error)
// BeginBatch creates a new batch. It returns an App-relative name for a
// Database handle bound to this batch. If this Database is already bound to a
// batch, BeginBatch() will fail with ErrBoundToBatch. Concurrency semantics
// are documented in model.go.
// TODO(sadovsky): make BatchOptions optional
- BeginBatch(ctx *context.T, call rpc.ServerCall, bo BatchOptions) (string, error)
+ BeginBatch(ctx *context.T, call rpc.ServerCall, schemaVersion int32, bo BatchOptions) (string, error)
// Commit persists the pending changes to the database.
// If this Database is not bound to a batch, Commit() will fail with
// ErrNotBoundToBatch.
- Commit(*context.T, rpc.ServerCall) error
+ Commit(ctx *context.T, call rpc.ServerCall, schemaVersion int32) error
// Exec executes a syncQL query and returns all results as specified by in the
// query's select clause. Concurrency semantics are documented in model.go.
- Exec(ctx *context.T, call DatabaseExecServerCall, query string) error
+ Exec(ctx *context.T, call DatabaseExecServerCall, schemaVersion int32, query string) error
// Abort notifies the server that any pending changes can be discarded.
// It is not strictly required, but it may allow the server to release locks
// or other resources sooner than if it was not called.
// If this Database is not bound to a batch, Abort() will fail with
// ErrNotBoundToBatch.
- Abort(*context.T, rpc.ServerCall) error
+ Abort(ctx *context.T, call rpc.ServerCall, schemaVersion int32) error
}
// DatabaseServerStubMethods is the server interface containing
@@ -1865,33 +1876,33 @@
// Create creates this Database.
// If perms is nil, we inherit (copy) the App perms.
// Create requires the caller to have Write permission at the App.
- Create(ctx *context.T, call rpc.ServerCall, perms access.Permissions, metadata *SchemaMetadata) error
+ Create(ctx *context.T, call rpc.ServerCall, metadata *SchemaMetadata, perms access.Permissions) error
// Delete deletes this Database.
- Delete(*context.T, rpc.ServerCall) error
+ Delete(ctx *context.T, call rpc.ServerCall, schemaVersion int32) error
// Exists returns true only if this Database exists. Insufficient permissions
// cause Exists to return false instead of an error.
// TODO(ivanpi): Exists may fail with an error if higher levels of hierarchy
// do not exist.
- Exists(*context.T, rpc.ServerCall) (bool, error)
+ Exists(ctx *context.T, call rpc.ServerCall, schemaVersion int32) (bool, error)
// BeginBatch creates a new batch. It returns an App-relative name for a
// Database handle bound to this batch. If this Database is already bound to a
// batch, BeginBatch() will fail with ErrBoundToBatch. Concurrency semantics
// are documented in model.go.
// TODO(sadovsky): make BatchOptions optional
- BeginBatch(ctx *context.T, call rpc.ServerCall, bo BatchOptions) (string, error)
+ BeginBatch(ctx *context.T, call rpc.ServerCall, schemaVersion int32, bo BatchOptions) (string, error)
// Commit persists the pending changes to the database.
// If this Database is not bound to a batch, Commit() will fail with
// ErrNotBoundToBatch.
- Commit(*context.T, rpc.ServerCall) error
+ Commit(ctx *context.T, call rpc.ServerCall, schemaVersion int32) error
// Exec executes a syncQL query and returns all results as specified by in the
// query's select clause. Concurrency semantics are documented in model.go.
- Exec(ctx *context.T, call *DatabaseExecServerCallStub, query string) error
+ Exec(ctx *context.T, call *DatabaseExecServerCallStub, schemaVersion int32, query string) error
// Abort notifies the server that any pending changes can be discarded.
// It is not strictly required, but it may allow the server to release locks
// or other resources sooner than if it was not called.
// If this Database is not bound to a batch, Abort() will fail with
// ErrNotBoundToBatch.
- Abort(*context.T, rpc.ServerCall) error
+ Abort(ctx *context.T, call rpc.ServerCall, schemaVersion int32) error
}
// DatabaseServerStub adds universal methods to DatabaseServerStubMethods.
@@ -1933,32 +1944,32 @@
gs *rpc.GlobState
}
-func (s implDatabaseServerStub) Create(ctx *context.T, call rpc.ServerCall, i0 access.Permissions, i1 *SchemaMetadata) error {
+func (s implDatabaseServerStub) Create(ctx *context.T, call rpc.ServerCall, i0 *SchemaMetadata, i1 access.Permissions) error {
return s.impl.Create(ctx, call, i0, i1)
}
-func (s implDatabaseServerStub) Delete(ctx *context.T, call rpc.ServerCall) error {
- return s.impl.Delete(ctx, call)
+func (s implDatabaseServerStub) Delete(ctx *context.T, call rpc.ServerCall, i0 int32) error {
+ return s.impl.Delete(ctx, call, i0)
}
-func (s implDatabaseServerStub) Exists(ctx *context.T, call rpc.ServerCall) (bool, error) {
- return s.impl.Exists(ctx, call)
+func (s implDatabaseServerStub) Exists(ctx *context.T, call rpc.ServerCall, i0 int32) (bool, error) {
+ return s.impl.Exists(ctx, call, i0)
}
-func (s implDatabaseServerStub) BeginBatch(ctx *context.T, call rpc.ServerCall, i0 BatchOptions) (string, error) {
- return s.impl.BeginBatch(ctx, call, i0)
+func (s implDatabaseServerStub) BeginBatch(ctx *context.T, call rpc.ServerCall, i0 int32, i1 BatchOptions) (string, error) {
+ return s.impl.BeginBatch(ctx, call, i0, i1)
}
-func (s implDatabaseServerStub) Commit(ctx *context.T, call rpc.ServerCall) error {
- return s.impl.Commit(ctx, call)
+func (s implDatabaseServerStub) Commit(ctx *context.T, call rpc.ServerCall, i0 int32) error {
+ return s.impl.Commit(ctx, call, i0)
}
-func (s implDatabaseServerStub) Exec(ctx *context.T, call *DatabaseExecServerCallStub, i0 string) error {
- return s.impl.Exec(ctx, call, i0)
+func (s implDatabaseServerStub) Exec(ctx *context.T, call *DatabaseExecServerCallStub, i0 int32, i1 string) error {
+ return s.impl.Exec(ctx, call, i0, i1)
}
-func (s implDatabaseServerStub) Abort(ctx *context.T, call rpc.ServerCall) error {
- return s.impl.Abort(ctx, call)
+func (s implDatabaseServerStub) Abort(ctx *context.T, call rpc.ServerCall, i0 int32) error {
+ return s.impl.Abort(ctx, call, i0)
}
func (s implDatabaseServerStub) Globber() *rpc.GlobState {
@@ -1976,7 +1987,7 @@
var descDatabase = rpc.InterfaceDesc{
Name: "Database",
PkgPath: "v.io/syncbase/v23/services/syncbase/nosql",
- Doc: "// Database represents a collection of Tables. Batches, queries, sync, watch,\n// etc. all operate at the Database level.\n// Database.Glob operates over Table names.\n//\n// TODO(sadovsky): Add Watch method.",
+ Doc: "// Database represents a collection of Tables. Batches, queries, sync, watch,\n// etc. all operate at the Database level.\n// Database.Glob operates over Table names.\n// Param schemaVersion is the version number that the client expects the database\n// to be at. To disable schema version checking, pass -1.\n//\n// TODO(sadovsky): Add Watch method.",
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/syncbase/v23/services/syncbase/nosql", "// Watch allows a client to watch for updates in the database. For each watched\n// request, the client will receive a reliable stream of watch events without\n// re-ordering. See watch.GlobWatcher for a detailed explanation of the\n// behavior.\n//\n// The watching is done by starting a streaming RPC. The argument to the RPC\n// contains the ResumeMarker that points to a particular place in the database\n// event log. The result stream consists of a never-ending sequence of Change\n// messages (until the call fails or is canceled). Each Change contains the\n// Name field in the form \"<tableName>/<rowKey>\" and the Value field of the\n// StoreChange type. If the client has no access to a row specified in a change,\n// that change is excluded from the result stream.\n//\n// The DatabaseWatcher is designed to be used in the following way:\n// 1) begin a read-only batch\n// 2) read all information your app needs\n// 3) read the ResumeMarker\n// 4) abort the batch\n// 5) start watching changes to the data using the ResumeMarker\n// In this configuration the client doesn't miss any changes."},
@@ -1989,19 +2000,25 @@
Name: "Create",
Doc: "// Create creates this Database.\n// If perms is nil, we inherit (copy) the App perms.\n// Create requires the caller to have Write permission at the App.",
InArgs: []rpc.ArgDesc{
- {"perms", ``}, // access.Permissions
{"metadata", ``}, // *SchemaMetadata
+ {"perms", ``}, // access.Permissions
},
Tags: []*vdl.Value{vdl.ValueOf(access.Tag("Write"))},
},
{
Name: "Delete",
Doc: "// Delete deletes this Database.",
+ InArgs: []rpc.ArgDesc{
+ {"schemaVersion", ``}, // int32
+ },
Tags: []*vdl.Value{vdl.ValueOf(access.Tag("Write"))},
},
{
Name: "Exists",
Doc: "// Exists returns true only if this Database exists. Insufficient permissions\n// cause Exists to return false instead of an error.\n// TODO(ivanpi): Exists may fail with an error if higher levels of hierarchy\n// do not exist.",
+ InArgs: []rpc.ArgDesc{
+ {"schemaVersion", ``}, // int32
+ },
OutArgs: []rpc.ArgDesc{
{"", ``}, // bool
},
@@ -2011,7 +2028,8 @@
Name: "BeginBatch",
Doc: "// BeginBatch creates a new batch. It returns an App-relative name for a\n// Database handle bound to this batch. If this Database is already bound to a\n// batch, BeginBatch() will fail with ErrBoundToBatch. Concurrency semantics\n// are documented in model.go.\n// TODO(sadovsky): make BatchOptions optional",
InArgs: []rpc.ArgDesc{
- {"bo", ``}, // BatchOptions
+ {"schemaVersion", ``}, // int32
+ {"bo", ``}, // BatchOptions
},
OutArgs: []rpc.ArgDesc{
{"", ``}, // string
@@ -2021,19 +2039,26 @@
{
Name: "Commit",
Doc: "// Commit persists the pending changes to the database.\n// If this Database is not bound to a batch, Commit() will fail with\n// ErrNotBoundToBatch.",
+ InArgs: []rpc.ArgDesc{
+ {"schemaVersion", ``}, // int32
+ },
Tags: []*vdl.Value{vdl.ValueOf(access.Tag("Read"))},
},
{
Name: "Exec",
Doc: "// Exec executes a syncQL query and returns all results as specified by in the\n// query's select clause. Concurrency semantics are documented in model.go.",
InArgs: []rpc.ArgDesc{
- {"query", ``}, // string
+ {"schemaVersion", ``}, // int32
+ {"query", ``}, // string
},
Tags: []*vdl.Value{vdl.ValueOf(access.Tag("Read"))},
},
{
Name: "Abort",
Doc: "// Abort notifies the server that any pending changes can be discarded.\n// It is not strictly required, but it may allow the server to release locks\n// or other resources sooner than if it was not called.\n// If this Database is not bound to a batch, Abort() will fail with\n// ErrNotBoundToBatch.",
+ InArgs: []rpc.ArgDesc{
+ {"schemaVersion", ``}, // int32
+ },
Tags: []*vdl.Value{vdl.ValueOf(access.Tag("Read"))},
},
},
@@ -2087,30 +2112,32 @@
//
// Table represents a collection of Rows.
// Table.Glob operates over the primary keys of Rows in the Table.
+// SchemaVersion is the version number that the client expects the database
+// to be at. To disable schema version checking, pass -1.
type TableClientMethods interface {
// Create creates this Table.
// If perms is nil, we inherit (copy) the Database perms.
- Create(ctx *context.T, perms access.Permissions, opts ...rpc.CallOpt) error
+ Create(ctx *context.T, schemaVersion int32, perms access.Permissions, opts ...rpc.CallOpt) error
// Delete deletes this Table.
- Delete(*context.T, ...rpc.CallOpt) error
+ Delete(ctx *context.T, schemaVersion int32, opts ...rpc.CallOpt) error
// Exists returns true only if this Table exists. Insufficient permissions
// cause Exists to return false instead of an error.
// TODO(ivanpi): Exists may fail with an error if higher levels of hierarchy
// do not exist.
- Exists(*context.T, ...rpc.CallOpt) (bool, error)
+ Exists(ctx *context.T, schemaVersion int32, opts ...rpc.CallOpt) (bool, error)
// Delete deletes all rows in the given half-open range [start, limit). If
// limit is "", all rows with keys >= start are included.
- DeleteRowRange(ctx *context.T, start []byte, limit []byte, opts ...rpc.CallOpt) error
+ DeleteRowRange(ctx *context.T, schemaVersion int32, start []byte, limit []byte, opts ...rpc.CallOpt) error
// Scan returns all rows in the given half-open range [start, limit). If limit
// is "", all rows with keys >= start are included. Concurrency semantics are
// documented in model.go.
- Scan(ctx *context.T, start []byte, limit []byte, opts ...rpc.CallOpt) (TableScanClientCall, error)
+ Scan(ctx *context.T, schemaVersion int32, start []byte, limit []byte, opts ...rpc.CallOpt) (TableScanClientCall, error)
// GetPermissions returns an array of (prefix, perms) pairs. The array is
// sorted from longest prefix to shortest, so element zero is the one that
// applies to the row with the given key. The last element is always the
// prefix "" which represents the table's permissions -- the array will always
// have at least one element.
- GetPermissions(ctx *context.T, key string, opts ...rpc.CallOpt) ([]PrefixPermissions, error)
+ GetPermissions(ctx *context.T, schemaVersion int32, key string, opts ...rpc.CallOpt) ([]PrefixPermissions, error)
// SetPermissions sets the permissions for all current and future rows with
// the given prefix. If the prefix overlaps with an existing prefix, the
// longest prefix that matches a row applies. For example:
@@ -2118,11 +2145,11 @@
// SetPermissions(ctx, Prefix("a/b/c"), perms2)
// The permissions for row "a/b/1" are perms1, and the permissions for row
// "a/b/c/1" are perms2.
- SetPermissions(ctx *context.T, prefix string, perms access.Permissions, opts ...rpc.CallOpt) error
+ SetPermissions(ctx *context.T, schemaVersion int32, prefix string, perms access.Permissions, opts ...rpc.CallOpt) error
// DeletePermissions deletes the permissions for the specified prefix. Any
// rows covered by this prefix will use the next longest prefix's permissions
// (see the array returned by GetPermissions).
- DeletePermissions(ctx *context.T, prefix string, opts ...rpc.CallOpt) error
+ DeletePermissions(ctx *context.T, schemaVersion int32, prefix string, opts ...rpc.CallOpt) error
}
// TableClientStub adds universal methods to TableClientMethods.
@@ -2140,47 +2167,47 @@
name string
}
-func (c implTableClientStub) Create(ctx *context.T, i0 access.Permissions, opts ...rpc.CallOpt) (err error) {
- err = v23.GetClient(ctx).Call(ctx, c.name, "Create", []interface{}{i0}, nil, opts...)
+func (c implTableClientStub) Create(ctx *context.T, i0 int32, i1 access.Permissions, opts ...rpc.CallOpt) (err error) {
+ err = v23.GetClient(ctx).Call(ctx, c.name, "Create", []interface{}{i0, i1}, nil, opts...)
return
}
-func (c implTableClientStub) Delete(ctx *context.T, opts ...rpc.CallOpt) (err error) {
- err = v23.GetClient(ctx).Call(ctx, c.name, "Delete", nil, nil, opts...)
+func (c implTableClientStub) Delete(ctx *context.T, i0 int32, opts ...rpc.CallOpt) (err error) {
+ err = v23.GetClient(ctx).Call(ctx, c.name, "Delete", []interface{}{i0}, nil, opts...)
return
}
-func (c implTableClientStub) Exists(ctx *context.T, opts ...rpc.CallOpt) (o0 bool, err error) {
- err = v23.GetClient(ctx).Call(ctx, c.name, "Exists", nil, []interface{}{&o0}, opts...)
+func (c implTableClientStub) Exists(ctx *context.T, i0 int32, opts ...rpc.CallOpt) (o0 bool, err error) {
+ err = v23.GetClient(ctx).Call(ctx, c.name, "Exists", []interface{}{i0}, []interface{}{&o0}, opts...)
return
}
-func (c implTableClientStub) DeleteRowRange(ctx *context.T, i0 []byte, i1 []byte, opts ...rpc.CallOpt) (err error) {
- err = v23.GetClient(ctx).Call(ctx, c.name, "DeleteRowRange", []interface{}{i0, i1}, nil, opts...)
+func (c implTableClientStub) DeleteRowRange(ctx *context.T, i0 int32, i1 []byte, i2 []byte, opts ...rpc.CallOpt) (err error) {
+ err = v23.GetClient(ctx).Call(ctx, c.name, "DeleteRowRange", []interface{}{i0, i1, i2}, nil, opts...)
return
}
-func (c implTableClientStub) Scan(ctx *context.T, i0 []byte, i1 []byte, opts ...rpc.CallOpt) (ocall TableScanClientCall, err error) {
+func (c implTableClientStub) Scan(ctx *context.T, i0 int32, i1 []byte, i2 []byte, opts ...rpc.CallOpt) (ocall TableScanClientCall, err error) {
var call rpc.ClientCall
- if call, err = v23.GetClient(ctx).StartCall(ctx, c.name, "Scan", []interface{}{i0, i1}, opts...); err != nil {
+ if call, err = v23.GetClient(ctx).StartCall(ctx, c.name, "Scan", []interface{}{i0, i1, i2}, opts...); err != nil {
return
}
ocall = &implTableScanClientCall{ClientCall: call}
return
}
-func (c implTableClientStub) GetPermissions(ctx *context.T, i0 string, opts ...rpc.CallOpt) (o0 []PrefixPermissions, err error) {
- err = v23.GetClient(ctx).Call(ctx, c.name, "GetPermissions", []interface{}{i0}, []interface{}{&o0}, opts...)
+func (c implTableClientStub) GetPermissions(ctx *context.T, i0 int32, i1 string, opts ...rpc.CallOpt) (o0 []PrefixPermissions, err error) {
+ err = v23.GetClient(ctx).Call(ctx, c.name, "GetPermissions", []interface{}{i0, i1}, []interface{}{&o0}, opts...)
return
}
-func (c implTableClientStub) SetPermissions(ctx *context.T, i0 string, i1 access.Permissions, opts ...rpc.CallOpt) (err error) {
- err = v23.GetClient(ctx).Call(ctx, c.name, "SetPermissions", []interface{}{i0, i1}, nil, opts...)
+func (c implTableClientStub) SetPermissions(ctx *context.T, i0 int32, i1 string, i2 access.Permissions, opts ...rpc.CallOpt) (err error) {
+ err = v23.GetClient(ctx).Call(ctx, c.name, "SetPermissions", []interface{}{i0, i1, i2}, nil, opts...)
return
}
-func (c implTableClientStub) DeletePermissions(ctx *context.T, i0 string, opts ...rpc.CallOpt) (err error) {
- err = v23.GetClient(ctx).Call(ctx, c.name, "DeletePermissions", []interface{}{i0}, nil, opts...)
+func (c implTableClientStub) DeletePermissions(ctx *context.T, i0 int32, i1 string, opts ...rpc.CallOpt) (err error) {
+ err = v23.GetClient(ctx).Call(ctx, c.name, "DeletePermissions", []interface{}{i0, i1}, nil, opts...)
return
}
@@ -2258,30 +2285,32 @@
//
// Table represents a collection of Rows.
// Table.Glob operates over the primary keys of Rows in the Table.
+// SchemaVersion is the version number that the client expects the database
+// to be at. To disable schema version checking, pass -1.
type TableServerMethods interface {
// Create creates this Table.
// If perms is nil, we inherit (copy) the Database perms.
- Create(ctx *context.T, call rpc.ServerCall, perms access.Permissions) error
+ Create(ctx *context.T, call rpc.ServerCall, schemaVersion int32, perms access.Permissions) error
// Delete deletes this Table.
- Delete(*context.T, rpc.ServerCall) error
+ Delete(ctx *context.T, call rpc.ServerCall, schemaVersion int32) error
// Exists returns true only if this Table exists. Insufficient permissions
// cause Exists to return false instead of an error.
// TODO(ivanpi): Exists may fail with an error if higher levels of hierarchy
// do not exist.
- Exists(*context.T, rpc.ServerCall) (bool, error)
+ Exists(ctx *context.T, call rpc.ServerCall, schemaVersion int32) (bool, error)
// Delete deletes all rows in the given half-open range [start, limit). If
// limit is "", all rows with keys >= start are included.
- DeleteRowRange(ctx *context.T, call rpc.ServerCall, start []byte, limit []byte) error
+ DeleteRowRange(ctx *context.T, call rpc.ServerCall, schemaVersion int32, start []byte, limit []byte) error
// Scan returns all rows in the given half-open range [start, limit). If limit
// is "", all rows with keys >= start are included. Concurrency semantics are
// documented in model.go.
- Scan(ctx *context.T, call TableScanServerCall, start []byte, limit []byte) error
+ Scan(ctx *context.T, call TableScanServerCall, schemaVersion int32, start []byte, limit []byte) error
// GetPermissions returns an array of (prefix, perms) pairs. The array is
// sorted from longest prefix to shortest, so element zero is the one that
// applies to the row with the given key. The last element is always the
// prefix "" which represents the table's permissions -- the array will always
// have at least one element.
- GetPermissions(ctx *context.T, call rpc.ServerCall, key string) ([]PrefixPermissions, error)
+ GetPermissions(ctx *context.T, call rpc.ServerCall, schemaVersion int32, key string) ([]PrefixPermissions, error)
// SetPermissions sets the permissions for all current and future rows with
// the given prefix. If the prefix overlaps with an existing prefix, the
// longest prefix that matches a row applies. For example:
@@ -2289,11 +2318,11 @@
// SetPermissions(ctx, Prefix("a/b/c"), perms2)
// The permissions for row "a/b/1" are perms1, and the permissions for row
// "a/b/c/1" are perms2.
- SetPermissions(ctx *context.T, call rpc.ServerCall, prefix string, perms access.Permissions) error
+ SetPermissions(ctx *context.T, call rpc.ServerCall, schemaVersion int32, prefix string, perms access.Permissions) error
// DeletePermissions deletes the permissions for the specified prefix. Any
// rows covered by this prefix will use the next longest prefix's permissions
// (see the array returned by GetPermissions).
- DeletePermissions(ctx *context.T, call rpc.ServerCall, prefix string) error
+ DeletePermissions(ctx *context.T, call rpc.ServerCall, schemaVersion int32, prefix string) error
}
// TableServerStubMethods is the server interface containing
@@ -2303,27 +2332,27 @@
type TableServerStubMethods interface {
// Create creates this Table.
// If perms is nil, we inherit (copy) the Database perms.
- Create(ctx *context.T, call rpc.ServerCall, perms access.Permissions) error
+ Create(ctx *context.T, call rpc.ServerCall, schemaVersion int32, perms access.Permissions) error
// Delete deletes this Table.
- Delete(*context.T, rpc.ServerCall) error
+ Delete(ctx *context.T, call rpc.ServerCall, schemaVersion int32) error
// Exists returns true only if this Table exists. Insufficient permissions
// cause Exists to return false instead of an error.
// TODO(ivanpi): Exists may fail with an error if higher levels of hierarchy
// do not exist.
- Exists(*context.T, rpc.ServerCall) (bool, error)
+ Exists(ctx *context.T, call rpc.ServerCall, schemaVersion int32) (bool, error)
// Delete deletes all rows in the given half-open range [start, limit). If
// limit is "", all rows with keys >= start are included.
- DeleteRowRange(ctx *context.T, call rpc.ServerCall, start []byte, limit []byte) error
+ DeleteRowRange(ctx *context.T, call rpc.ServerCall, schemaVersion int32, start []byte, limit []byte) error
// Scan returns all rows in the given half-open range [start, limit). If limit
// is "", all rows with keys >= start are included. Concurrency semantics are
// documented in model.go.
- Scan(ctx *context.T, call *TableScanServerCallStub, start []byte, limit []byte) error
+ Scan(ctx *context.T, call *TableScanServerCallStub, schemaVersion int32, start []byte, limit []byte) error
// GetPermissions returns an array of (prefix, perms) pairs. The array is
// sorted from longest prefix to shortest, so element zero is the one that
// applies to the row with the given key. The last element is always the
// prefix "" which represents the table's permissions -- the array will always
// have at least one element.
- GetPermissions(ctx *context.T, call rpc.ServerCall, key string) ([]PrefixPermissions, error)
+ GetPermissions(ctx *context.T, call rpc.ServerCall, schemaVersion int32, key string) ([]PrefixPermissions, error)
// SetPermissions sets the permissions for all current and future rows with
// the given prefix. If the prefix overlaps with an existing prefix, the
// longest prefix that matches a row applies. For example:
@@ -2331,11 +2360,11 @@
// SetPermissions(ctx, Prefix("a/b/c"), perms2)
// The permissions for row "a/b/1" are perms1, and the permissions for row
// "a/b/c/1" are perms2.
- SetPermissions(ctx *context.T, call rpc.ServerCall, prefix string, perms access.Permissions) error
+ SetPermissions(ctx *context.T, call rpc.ServerCall, schemaVersion int32, prefix string, perms access.Permissions) error
// DeletePermissions deletes the permissions for the specified prefix. Any
// rows covered by this prefix will use the next longest prefix's permissions
// (see the array returned by GetPermissions).
- DeletePermissions(ctx *context.T, call rpc.ServerCall, prefix string) error
+ DeletePermissions(ctx *context.T, call rpc.ServerCall, schemaVersion int32, prefix string) error
}
// TableServerStub adds universal methods to TableServerStubMethods.
@@ -2367,36 +2396,36 @@
gs *rpc.GlobState
}
-func (s implTableServerStub) Create(ctx *context.T, call rpc.ServerCall, i0 access.Permissions) error {
- return s.impl.Create(ctx, call, i0)
+func (s implTableServerStub) Create(ctx *context.T, call rpc.ServerCall, i0 int32, i1 access.Permissions) error {
+ return s.impl.Create(ctx, call, i0, i1)
}
-func (s implTableServerStub) Delete(ctx *context.T, call rpc.ServerCall) error {
- return s.impl.Delete(ctx, call)
+func (s implTableServerStub) Delete(ctx *context.T, call rpc.ServerCall, i0 int32) error {
+ return s.impl.Delete(ctx, call, i0)
}
-func (s implTableServerStub) Exists(ctx *context.T, call rpc.ServerCall) (bool, error) {
- return s.impl.Exists(ctx, call)
+func (s implTableServerStub) Exists(ctx *context.T, call rpc.ServerCall, i0 int32) (bool, error) {
+ return s.impl.Exists(ctx, call, i0)
}
-func (s implTableServerStub) DeleteRowRange(ctx *context.T, call rpc.ServerCall, i0 []byte, i1 []byte) error {
- return s.impl.DeleteRowRange(ctx, call, i0, i1)
+func (s implTableServerStub) DeleteRowRange(ctx *context.T, call rpc.ServerCall, i0 int32, i1 []byte, i2 []byte) error {
+ return s.impl.DeleteRowRange(ctx, call, i0, i1, i2)
}
-func (s implTableServerStub) Scan(ctx *context.T, call *TableScanServerCallStub, i0 []byte, i1 []byte) error {
- return s.impl.Scan(ctx, call, i0, i1)
+func (s implTableServerStub) Scan(ctx *context.T, call *TableScanServerCallStub, i0 int32, i1 []byte, i2 []byte) error {
+ return s.impl.Scan(ctx, call, i0, i1, i2)
}
-func (s implTableServerStub) GetPermissions(ctx *context.T, call rpc.ServerCall, i0 string) ([]PrefixPermissions, error) {
- return s.impl.GetPermissions(ctx, call, i0)
+func (s implTableServerStub) GetPermissions(ctx *context.T, call rpc.ServerCall, i0 int32, i1 string) ([]PrefixPermissions, error) {
+ return s.impl.GetPermissions(ctx, call, i0, i1)
}
-func (s implTableServerStub) SetPermissions(ctx *context.T, call rpc.ServerCall, i0 string, i1 access.Permissions) error {
- return s.impl.SetPermissions(ctx, call, i0, i1)
+func (s implTableServerStub) SetPermissions(ctx *context.T, call rpc.ServerCall, i0 int32, i1 string, i2 access.Permissions) error {
+ return s.impl.SetPermissions(ctx, call, i0, i1, i2)
}
-func (s implTableServerStub) DeletePermissions(ctx *context.T, call rpc.ServerCall, i0 string) error {
- return s.impl.DeletePermissions(ctx, call, i0)
+func (s implTableServerStub) DeletePermissions(ctx *context.T, call rpc.ServerCall, i0 int32, i1 string) error {
+ return s.impl.DeletePermissions(ctx, call, i0, i1)
}
func (s implTableServerStub) Globber() *rpc.GlobState {
@@ -2414,24 +2443,31 @@
var descTable = rpc.InterfaceDesc{
Name: "Table",
PkgPath: "v.io/syncbase/v23/services/syncbase/nosql",
- Doc: "// Table represents a collection of Rows.\n// Table.Glob operates over the primary keys of Rows in the Table.",
+ Doc: "// Table represents a collection of Rows.\n// Table.Glob operates over the primary keys of Rows in the Table.\n// SchemaVersion is the version number that the client expects the database\n// to be at. To disable schema version checking, pass -1.",
Methods: []rpc.MethodDesc{
{
Name: "Create",
Doc: "// Create creates this Table.\n// If perms is nil, we inherit (copy) the Database perms.",
InArgs: []rpc.ArgDesc{
- {"perms", ``}, // access.Permissions
+ {"schemaVersion", ``}, // int32
+ {"perms", ``}, // access.Permissions
},
Tags: []*vdl.Value{vdl.ValueOf(access.Tag("Write"))},
},
{
Name: "Delete",
Doc: "// Delete deletes this Table.",
+ InArgs: []rpc.ArgDesc{
+ {"schemaVersion", ``}, // int32
+ },
Tags: []*vdl.Value{vdl.ValueOf(access.Tag("Write"))},
},
{
Name: "Exists",
Doc: "// Exists returns true only if this Table exists. Insufficient permissions\n// cause Exists to return false instead of an error.\n// TODO(ivanpi): Exists may fail with an error if higher levels of hierarchy\n// do not exist.",
+ InArgs: []rpc.ArgDesc{
+ {"schemaVersion", ``}, // int32
+ },
OutArgs: []rpc.ArgDesc{
{"", ``}, // bool
},
@@ -2441,8 +2477,9 @@
Name: "DeleteRowRange",
Doc: "// Delete deletes all rows in the given half-open range [start, limit). If\n// limit is \"\", all rows with keys >= start are included.",
InArgs: []rpc.ArgDesc{
- {"start", ``}, // []byte
- {"limit", ``}, // []byte
+ {"schemaVersion", ``}, // int32
+ {"start", ``}, // []byte
+ {"limit", ``}, // []byte
},
Tags: []*vdl.Value{vdl.ValueOf(access.Tag("Write"))},
},
@@ -2450,8 +2487,9 @@
Name: "Scan",
Doc: "// Scan returns all rows in the given half-open range [start, limit). If limit\n// is \"\", all rows with keys >= start are included. Concurrency semantics are\n// documented in model.go.",
InArgs: []rpc.ArgDesc{
- {"start", ``}, // []byte
- {"limit", ``}, // []byte
+ {"schemaVersion", ``}, // int32
+ {"start", ``}, // []byte
+ {"limit", ``}, // []byte
},
Tags: []*vdl.Value{vdl.ValueOf(access.Tag("Read"))},
},
@@ -2459,7 +2497,8 @@
Name: "GetPermissions",
Doc: "// GetPermissions returns an array of (prefix, perms) pairs. The array is\n// sorted from longest prefix to shortest, so element zero is the one that\n// applies to the row with the given key. The last element is always the\n// prefix \"\" which represents the table's permissions -- the array will always\n// have at least one element.",
InArgs: []rpc.ArgDesc{
- {"key", ``}, // string
+ {"schemaVersion", ``}, // int32
+ {"key", ``}, // string
},
OutArgs: []rpc.ArgDesc{
{"", ``}, // []PrefixPermissions
@@ -2470,8 +2509,9 @@
Name: "SetPermissions",
Doc: "// SetPermissions sets the permissions for all current and future rows with\n// the given prefix. If the prefix overlaps with an existing prefix, the\n// longest prefix that matches a row applies. For example:\n// SetPermissions(ctx, Prefix(\"a/b\"), perms1)\n// SetPermissions(ctx, Prefix(\"a/b/c\"), perms2)\n// The permissions for row \"a/b/1\" are perms1, and the permissions for row\n// \"a/b/c/1\" are perms2.",
InArgs: []rpc.ArgDesc{
- {"prefix", ``}, // string
- {"perms", ``}, // access.Permissions
+ {"schemaVersion", ``}, // int32
+ {"prefix", ``}, // string
+ {"perms", ``}, // access.Permissions
},
Tags: []*vdl.Value{vdl.ValueOf(access.Tag("Admin"))},
},
@@ -2479,7 +2519,8 @@
Name: "DeletePermissions",
Doc: "// DeletePermissions deletes the permissions for the specified prefix. Any\n// rows covered by this prefix will use the next longest prefix's permissions\n// (see the array returned by GetPermissions).",
InArgs: []rpc.ArgDesc{
- {"prefix", ``}, // string
+ {"schemaVersion", ``}, // int32
+ {"prefix", ``}, // string
},
Tags: []*vdl.Value{vdl.ValueOf(access.Tag("Admin"))},
},
@@ -2535,6 +2576,8 @@
// Row represents a single row in a Table.
// All access checks are performed against the most specific matching prefix
// permissions in the Table.
+// SchemaVersion is the version number that the client expects the database
+// to be at. To disable schema version checking, pass -1.
// NOTE(sadovsky): Currently we send []byte values over the wire for Get, Put,
// and Scan. If there's a way to avoid encoding/decoding on the server side, we
// can use vdl.Value everywhere without sacrificing performance.
@@ -2543,13 +2586,13 @@
// cause Exists to return false instead of an error.
// TODO(ivanpi): Exists may fail with an error if higher levels of hierarchy
// do not exist.
- Exists(*context.T, ...rpc.CallOpt) (bool, error)
+ Exists(ctx *context.T, schemaVersion int32, opts ...rpc.CallOpt) (bool, error)
// Get returns the value for this Row.
- Get(*context.T, ...rpc.CallOpt) ([]byte, error)
+ Get(ctx *context.T, schemaVersion int32, opts ...rpc.CallOpt) ([]byte, error)
// Put writes the given value for this Row.
- Put(ctx *context.T, value []byte, opts ...rpc.CallOpt) error
+ Put(ctx *context.T, schemaVersion int32, value []byte, opts ...rpc.CallOpt) error
// Delete deletes this Row.
- Delete(*context.T, ...rpc.CallOpt) error
+ Delete(ctx *context.T, schemaVersion int32, opts ...rpc.CallOpt) error
}
// RowClientStub adds universal methods to RowClientMethods.
@@ -2567,23 +2610,23 @@
name string
}
-func (c implRowClientStub) Exists(ctx *context.T, opts ...rpc.CallOpt) (o0 bool, err error) {
- err = v23.GetClient(ctx).Call(ctx, c.name, "Exists", nil, []interface{}{&o0}, opts...)
+func (c implRowClientStub) Exists(ctx *context.T, i0 int32, opts ...rpc.CallOpt) (o0 bool, err error) {
+ err = v23.GetClient(ctx).Call(ctx, c.name, "Exists", []interface{}{i0}, []interface{}{&o0}, opts...)
return
}
-func (c implRowClientStub) Get(ctx *context.T, opts ...rpc.CallOpt) (o0 []byte, err error) {
- err = v23.GetClient(ctx).Call(ctx, c.name, "Get", nil, []interface{}{&o0}, opts...)
+func (c implRowClientStub) Get(ctx *context.T, i0 int32, opts ...rpc.CallOpt) (o0 []byte, err error) {
+ err = v23.GetClient(ctx).Call(ctx, c.name, "Get", []interface{}{i0}, []interface{}{&o0}, opts...)
return
}
-func (c implRowClientStub) Put(ctx *context.T, i0 []byte, opts ...rpc.CallOpt) (err error) {
- err = v23.GetClient(ctx).Call(ctx, c.name, "Put", []interface{}{i0}, nil, opts...)
+func (c implRowClientStub) Put(ctx *context.T, i0 int32, i1 []byte, opts ...rpc.CallOpt) (err error) {
+ err = v23.GetClient(ctx).Call(ctx, c.name, "Put", []interface{}{i0, i1}, nil, opts...)
return
}
-func (c implRowClientStub) Delete(ctx *context.T, opts ...rpc.CallOpt) (err error) {
- err = v23.GetClient(ctx).Call(ctx, c.name, "Delete", nil, nil, opts...)
+func (c implRowClientStub) Delete(ctx *context.T, i0 int32, opts ...rpc.CallOpt) (err error) {
+ err = v23.GetClient(ctx).Call(ctx, c.name, "Delete", []interface{}{i0}, nil, opts...)
return
}
@@ -2593,6 +2636,8 @@
// Row represents a single row in a Table.
// All access checks are performed against the most specific matching prefix
// permissions in the Table.
+// SchemaVersion is the version number that the client expects the database
+// to be at. To disable schema version checking, pass -1.
// NOTE(sadovsky): Currently we send []byte values over the wire for Get, Put,
// and Scan. If there's a way to avoid encoding/decoding on the server side, we
// can use vdl.Value everywhere without sacrificing performance.
@@ -2601,13 +2646,13 @@
// cause Exists to return false instead of an error.
// TODO(ivanpi): Exists may fail with an error if higher levels of hierarchy
// do not exist.
- Exists(*context.T, rpc.ServerCall) (bool, error)
+ Exists(ctx *context.T, call rpc.ServerCall, schemaVersion int32) (bool, error)
// Get returns the value for this Row.
- Get(*context.T, rpc.ServerCall) ([]byte, error)
+ Get(ctx *context.T, call rpc.ServerCall, schemaVersion int32) ([]byte, error)
// Put writes the given value for this Row.
- Put(ctx *context.T, call rpc.ServerCall, value []byte) error
+ Put(ctx *context.T, call rpc.ServerCall, schemaVersion int32, value []byte) error
// Delete deletes this Row.
- Delete(*context.T, rpc.ServerCall) error
+ Delete(ctx *context.T, call rpc.ServerCall, schemaVersion int32) error
}
// RowServerStubMethods is the server interface containing
@@ -2645,20 +2690,20 @@
gs *rpc.GlobState
}
-func (s implRowServerStub) Exists(ctx *context.T, call rpc.ServerCall) (bool, error) {
- return s.impl.Exists(ctx, call)
+func (s implRowServerStub) Exists(ctx *context.T, call rpc.ServerCall, i0 int32) (bool, error) {
+ return s.impl.Exists(ctx, call, i0)
}
-func (s implRowServerStub) Get(ctx *context.T, call rpc.ServerCall) ([]byte, error) {
- return s.impl.Get(ctx, call)
+func (s implRowServerStub) Get(ctx *context.T, call rpc.ServerCall, i0 int32) ([]byte, error) {
+ return s.impl.Get(ctx, call, i0)
}
-func (s implRowServerStub) Put(ctx *context.T, call rpc.ServerCall, i0 []byte) error {
- return s.impl.Put(ctx, call, i0)
+func (s implRowServerStub) Put(ctx *context.T, call rpc.ServerCall, i0 int32, i1 []byte) error {
+ return s.impl.Put(ctx, call, i0, i1)
}
-func (s implRowServerStub) Delete(ctx *context.T, call rpc.ServerCall) error {
- return s.impl.Delete(ctx, call)
+func (s implRowServerStub) Delete(ctx *context.T, call rpc.ServerCall, i0 int32) error {
+ return s.impl.Delete(ctx, call, i0)
}
func (s implRowServerStub) Globber() *rpc.GlobState {
@@ -2676,11 +2721,14 @@
var descRow = rpc.InterfaceDesc{
Name: "Row",
PkgPath: "v.io/syncbase/v23/services/syncbase/nosql",
- Doc: "// Row represents a single row in a Table.\n// All access checks are performed against the most specific matching prefix\n// permissions in the Table.\n// NOTE(sadovsky): Currently we send []byte values over the wire for Get, Put,\n// and Scan. If there's a way to avoid encoding/decoding on the server side, we\n// can use vdl.Value everywhere without sacrificing performance.",
+ Doc: "// Row represents a single row in a Table.\n// All access checks are performed against the most specific matching prefix\n// permissions in the Table.\n// SchemaVersion is the version number that the client expects the database\n// to be at. To disable schema version checking, pass -1.\n// NOTE(sadovsky): Currently we send []byte values over the wire for Get, Put,\n// and Scan. If there's a way to avoid encoding/decoding on the server side, we\n// can use vdl.Value everywhere without sacrificing performance.",
Methods: []rpc.MethodDesc{
{
Name: "Exists",
Doc: "// Exists returns true only if this Row exists. Insufficient permissions\n// cause Exists to return false instead of an error.\n// TODO(ivanpi): Exists may fail with an error if higher levels of hierarchy\n// do not exist.",
+ InArgs: []rpc.ArgDesc{
+ {"schemaVersion", ``}, // int32
+ },
OutArgs: []rpc.ArgDesc{
{"", ``}, // bool
},
@@ -2689,6 +2737,9 @@
{
Name: "Get",
Doc: "// Get returns the value for this Row.",
+ InArgs: []rpc.ArgDesc{
+ {"schemaVersion", ``}, // int32
+ },
OutArgs: []rpc.ArgDesc{
{"", ``}, // []byte
},
@@ -2698,13 +2749,17 @@
Name: "Put",
Doc: "// Put writes the given value for this Row.",
InArgs: []rpc.ArgDesc{
- {"value", ``}, // []byte
+ {"schemaVersion", ``}, // int32
+ {"value", ``}, // []byte
},
Tags: []*vdl.Value{vdl.ValueOf(access.Tag("Write"))},
},
{
Name: "Delete",
Doc: "// Delete deletes this Row.",
+ InArgs: []rpc.ArgDesc{
+ {"schemaVersion", ``}, // int32
+ },
Tags: []*vdl.Value{vdl.ValueOf(access.Tag("Write"))},
},
},
diff --git a/v23/services/syncbase/nosql/types.vdl b/v23/services/syncbase/nosql/types.vdl
index 23d2d37..37fd383 100644
--- a/v23/services/syncbase/nosql/types.vdl
+++ b/v23/services/syncbase/nosql/types.vdl
@@ -117,7 +117,7 @@
// Non negative Schema version number. Should be increased with every schema change
// (e.g. adding fields to structs) that cannot be handled by previous
// versions of the app.
- Version int64
+ Version int32
Policy CrPolicy
}
diff --git a/v23/services/syncbase/nosql/types.vdl.go b/v23/services/syncbase/nosql/types.vdl.go
index facbd29..ac27aad 100644
--- a/v23/services/syncbase/nosql/types.vdl.go
+++ b/v23/services/syncbase/nosql/types.vdl.go
@@ -172,7 +172,7 @@
// Non negative Schema version number. Should be increased with every schema change
// (e.g. adding fields to structs) that cannot be handled by previous
// versions of the app.
- Version int64
+ Version int32
Policy CrPolicy
}
diff --git a/v23/syncbase/nosql/batch.go b/v23/syncbase/nosql/batch.go
index ea7290e..809a912 100644
--- a/v23/syncbase/nosql/batch.go
+++ b/v23/syncbase/nosql/batch.go
@@ -18,12 +18,12 @@
// Commit implements BatchDatabase.Commit.
func (b *batch) Commit(ctx *context.T) error {
- return b.c.Commit(ctx)
+ return b.c.Commit(ctx, b.schemaVersion())
}
// Abort implements BatchDatabase.Abort.
-func (b *batch) Abort(ctx *context.T) {
- b.c.Abort(ctx)
+func (b *batch) Abort(ctx *context.T) error {
+ return b.c.Abort(ctx, b.schemaVersion())
}
// RunInBatch runs the given fn in a batch, managing retries and commit/abort.
diff --git a/v23/syncbase/nosql/batch_test.go b/v23/syncbase/nosql/batch_test.go
index bbee3af..a1fe15c 100644
--- a/v23/syncbase/nosql/batch_test.go
+++ b/v23/syncbase/nosql/batch_test.go
@@ -497,10 +497,10 @@
// Batch methods on non-batch.
dc := wire.DatabaseClient(d.FullName())
- if err := dc.Commit(ctx); verror.ErrorID(err) != wire.ErrNotBoundToBatch.ID {
+ if err := dc.Commit(ctx, -1); verror.ErrorID(err) != wire.ErrNotBoundToBatch.ID {
t.Fatalf("dc.Commit() should have failed: %v", err)
}
- if err := dc.Abort(ctx); verror.ErrorID(err) != wire.ErrNotBoundToBatch.ID {
+ if err := dc.Abort(ctx, -1); verror.ErrorID(err) != wire.ErrNotBoundToBatch.ID {
t.Fatalf("dc.Abort() should have failed: %v", err)
}
@@ -513,10 +513,10 @@
if err := bc.Create(ctx, nil, nil); err == nil {
t.Fatalf("bc.Create() should have failed: %v", err)
}
- if err := bc.Delete(ctx); verror.ErrorID(err) != wire.ErrBoundToBatch.ID {
+ if err := bc.Delete(ctx, -1); verror.ErrorID(err) != wire.ErrBoundToBatch.ID {
t.Fatalf("bc.Delete() should have failed: %v", err)
}
- if _, err := bc.BeginBatch(ctx, wire.BatchOptions{}); verror.ErrorID(err) != wire.ErrBoundToBatch.ID {
+ if _, err := bc.BeginBatch(ctx, -1, wire.BatchOptions{}); verror.ErrorID(err) != wire.ErrBoundToBatch.ID {
t.Fatalf("bc.BeginBatch() should have failed: %v", err)
}
if _, _, err := bc.GetPermissions(ctx); verror.ErrorID(err) != wire.ErrBoundToBatch.ID {
@@ -532,10 +532,10 @@
// Test that Table.{Create,Delete} fail with ErrBoundToBatch.
tc := wire.TableClient(naming.Join(b.FullName(), "tb"))
- if err := tc.Create(ctx, nil); verror.ErrorID(err) != wire.ErrBoundToBatch.ID {
+ if err := tc.Create(ctx, -1, nil); verror.ErrorID(err) != wire.ErrBoundToBatch.ID {
t.Fatalf("tc.Create() should have failed: %v", err)
}
- if err := tc.Delete(ctx); verror.ErrorID(err) != wire.ErrBoundToBatch.ID {
+ if err := tc.Delete(ctx, -1); verror.ErrorID(err) != wire.ErrBoundToBatch.ID {
t.Fatalf("tc.Delete() should have failed: %v", err)
}
}
diff --git a/v23/syncbase/nosql/database.go b/v23/syncbase/nosql/database.go
index 31aeb19..5a23295 100644
--- a/v23/syncbase/nosql/database.go
+++ b/v23/syncbase/nosql/database.go
@@ -49,12 +49,12 @@
// Exists implements Database.Exists.
func (d *database) Exists(ctx *context.T) (bool, error) {
- return d.c.Exists(ctx)
+ return d.c.Exists(ctx, d.schemaVersion())
}
// Table implements Database.Table.
func (d *database) Table(relativeName string) Table {
- return newTable(d.fullName, relativeName)
+ return newTable(d.fullName, relativeName, d.schemaVersion())
}
// ListTables implements Database.ListTables.
@@ -68,28 +68,28 @@
if d.schema != nil {
schemaMetadata = &d.schema.Metadata
}
- return d.c.Create(ctx, perms, schemaMetadata)
+ return d.c.Create(ctx, schemaMetadata, perms)
}
// Delete implements Database.Delete.
func (d *database) Delete(ctx *context.T) error {
- return d.c.Delete(ctx)
+ return d.c.Delete(ctx, d.schemaVersion())
}
// CreateTable implements Database.CreateTable.
func (d *database) CreateTable(ctx *context.T, relativeName string, perms access.Permissions) error {
- return wire.TableClient(naming.Join(d.fullName, relativeName)).Create(ctx, perms)
+ return wire.TableClient(naming.Join(d.fullName, relativeName)).Create(ctx, d.schemaVersion(), perms)
}
// DeleteTable implements Database.DeleteTable.
func (d *database) DeleteTable(ctx *context.T, relativeName string) error {
- return wire.TableClient(naming.Join(d.fullName, relativeName)).Delete(ctx)
+ return wire.TableClient(naming.Join(d.fullName, relativeName)).Delete(ctx, d.schemaVersion())
}
// Exec implements Database.Exec.
func (d *database) Exec(ctx *context.T, query string) ([]string, ResultStream, error) {
ctx, cancel := context.WithCancel(ctx)
- call, err := d.c.Exec(ctx, query)
+ call, err := d.c.Exec(ctx, d.schemaVersion(), query)
if err != nil {
return nil, nil, err
}
@@ -112,7 +112,7 @@
// BeginBatch implements Database.BeginBatch.
func (d *database) BeginBatch(ctx *context.T, opts wire.BatchOptions) (BatchDatabase, error) {
- relativeName, err := d.c.BeginBatch(ctx, opts)
+ relativeName, err := d.c.BeginBatch(ctx, d.schemaVersion(), opts)
if err != nil {
return nil, err
}
@@ -196,3 +196,10 @@
func (d *database) getSchemaManager() schemaManagerImpl {
return newSchemaManager(d.fullName)
}
+
+func (d *database) schemaVersion() int32 {
+ if d.schema == nil {
+ return -1
+ }
+ return d.schema.Metadata.Version
+}
diff --git a/v23/syncbase/nosql/model.go b/v23/syncbase/nosql/model.go
index 13c3cc6..0d1dacf 100644
--- a/v23/syncbase/nosql/model.go
+++ b/v23/syncbase/nosql/model.go
@@ -129,7 +129,7 @@
// Abort notifies the server that any pending changes can be discarded.
// It is not strictly required, but it may allow the server to release locks
// or other resources sooner than if it was not called.
- Abort(ctx *context.T)
+ Abort(ctx *context.T) error
}
// PrefixPermissions represents a pair of (prefix, perms).
@@ -357,7 +357,7 @@
type SchemaUpgrader interface {
// Takes an instance of database and upgrades data from old
// schema to new schema. This method must be idempotent.
- Run(db Database, oldVersion, newVersion int64) error
+ Run(db Database, oldVersion, newVersion int32) error
}
// Each database has a Schema associated with it which defines the current
diff --git a/v23/syncbase/nosql/row.go b/v23/syncbase/nosql/row.go
index 833f4a6..e5d116e 100644
--- a/v23/syncbase/nosql/row.go
+++ b/v23/syncbase/nosql/row.go
@@ -11,20 +11,22 @@
"v.io/v23/vom"
)
-func newRow(parentFullName, key string) Row {
+func newRow(parentFullName, key string, schemaVersion int32) Row {
// TODO(sadovsky): Escape delimiters in key?
fullName := naming.Join(parentFullName, key)
return &row{
- c: wire.RowClient(fullName),
- fullName: fullName,
- key: key,
+ c: wire.RowClient(fullName),
+ fullName: fullName,
+ key: key,
+ dbSchemaVersion: schemaVersion,
}
}
type row struct {
- c wire.RowClientMethods
- fullName string
- key string
+ c wire.RowClientMethods
+ fullName string
+ key string
+ dbSchemaVersion int32
}
var _ Row = (*row)(nil)
@@ -43,12 +45,12 @@
// Exists implements Row.Exists.
func (r *row) Exists(ctx *context.T) (bool, error) {
- return r.c.Exists(ctx)
+ return r.c.Exists(ctx, r.dbSchemaVersion)
}
// Get implements Row.Get.
func (r *row) Get(ctx *context.T, value interface{}) error {
- bytes, err := r.c.Get(ctx)
+ bytes, err := r.c.Get(ctx, r.dbSchemaVersion)
if err != nil {
return err
}
@@ -61,10 +63,10 @@
if err != nil {
return err
}
- return r.c.Put(ctx, bytes)
+ return r.c.Put(ctx, r.dbSchemaVersion, bytes)
}
// Delete implements Row.Delete.
func (r *row) Delete(ctx *context.T) error {
- return r.c.Delete(ctx)
+ return r.c.Delete(ctx, r.dbSchemaVersion)
}
diff --git a/v23/syncbase/nosql/schema_test.go b/v23/syncbase/nosql/schema_test.go
index bcafcb7..6d7ed0d 100644
--- a/v23/syncbase/nosql/schema_test.go
+++ b/v23/syncbase/nosql/schema_test.go
@@ -9,8 +9,10 @@
wire "v.io/syncbase/v23/services/syncbase/nosql"
"v.io/syncbase/v23/syncbase"
+ "v.io/syncbase/v23/syncbase/nosql"
tu "v.io/syncbase/v23/syncbase/testutil"
"v.io/v23/context"
+ "v.io/v23/verror"
)
// Tests schema checking logic within App.NoSQLDatabase() method.
@@ -26,7 +28,7 @@
ctx, sName, cleanup := tu.SetupOrDie(nil)
defer cleanup()
a := tu.CreateApp(t, ctx, syncbase.NewService(sName), "a")
- schema := tu.DefaultSchema()
+ schema := tu.DefaultSchema(0)
mockUpgrader := schema.Upgrader.(*tu.MockSchemaUpgrader)
db1 := a.NoSQLDatabase("db1", schema)
@@ -99,6 +101,166 @@
}
}
+func TestRPCSchemaCheckError(t *testing.T) {
+ // Setup
+ ctx, sName, cleanup := tu.SetupOrDie(nil)
+ defer cleanup()
+ a := tu.CreateApp(t, ctx, syncbase.NewService(sName), "a")
+ schema := tu.DefaultSchema(0)
+
+ // Create db1 with schema version 0 and add table1 and row1
+ dbHandle1 := a.NoSQLDatabase("db1", schema)
+ if err := dbHandle1.Create(ctx, nil); err != nil {
+ t.Fatalf("db1.Create() failed: %v", err)
+ }
+ if err := dbHandle1.CreateTable(ctx, "table1", nil); err != nil {
+ t.Fatalf("db1.CreateTable() failed: %v", err)
+ }
+ if err := dbHandle1.Table("table1").Put(ctx, "row1", "value1"); err != nil {
+ t.Fatalf("table1.Put() failed: %v", err)
+ }
+
+ // Try writing to database db1 with a db handle with schema version 2
+ schema2 := tu.DefaultSchema(2)
+ dbHandle2 := a.NoSQLDatabase("db1", schema2)
+
+ // verify write rpcs for Database
+ if err := dbHandle2.CreateTable(ctx, "table1", nil); !isVersionMismatchErr(err) {
+ t.Fatal("Expected ErrDatabaseVersionMismatch, found: " + toString(err))
+ }
+ if err := dbHandle2.DeleteTable(ctx, "table1"); !isVersionMismatchErr(err) {
+ t.Fatal("Expected ErrDatabaseVersionMismatch, found: " + toString(err))
+ }
+ if err := dbHandle2.Delete(ctx); !isVersionMismatchErr(err) {
+ t.Fatal("Expected ErrDatabaseVersionMismatch, found: " + toString(err))
+ }
+ if _, err := dbHandle2.Exists(ctx); !isVersionMismatchErr(err) {
+ t.Fatal("Expected ErrDatabaseVersionMismatch, found: " + toString(err))
+ }
+ if _, err := dbHandle2.BeginBatch(ctx, wire.BatchOptions{}); !isVersionMismatchErr(err) {
+ t.Fatal("Expected ErrDatabaseVersionMismatch, found: " + toString(err))
+ }
+ if _, _, err := dbHandle2.Exec(ctx, ""); !isVersionMismatchErr(err) {
+ t.Fatal("Expected ErrDatabaseVersionMismatch, found: " + toString(err))
+ }
+
+ // verify write rpcs for Table
+ table := dbHandle2.Table("table1")
+ if _, err := table.Exists(ctx); !isVersionMismatchErr(err) {
+ t.Fatal("Expected ErrDatabaseVersionMismatch, found: " + toString(err))
+ }
+ if err := table.Delete(ctx, nosql.SingleRow("row1")); !isVersionMismatchErr(err) {
+ t.Fatal("Expected ErrDatabaseVersionMismatch, found: " + toString(err))
+ }
+ stream := table.Scan(ctx, nosql.SingleRow("row1"))
+ if stream.Advance() {
+ t.Fatalf("Stream advanced unexpectedly")
+ }
+ if !isVersionMismatchErr(stream.Err()) {
+ t.Fatal("Expected ErrDatabaseVersionMismatch, found: " + toString(stream.Err()))
+ }
+ if _, err := table.GetPermissions(ctx, "row1"); !isVersionMismatchErr(err) {
+ t.Fatal("Expected ErrDatabaseVersionMismatch, found: " + toString(err))
+ }
+ if err := table.SetPermissions(ctx, nosql.Prefix("row"), nil); !isVersionMismatchErr(err) {
+ t.Fatal("Expected ErrDatabaseVersionMismatch, found: " + toString(err))
+ }
+ if err := table.DeletePermissions(ctx, nosql.Prefix("row")); !isVersionMismatchErr(err) {
+ t.Fatal("Expected ErrDatabaseVersionMismatch, found: " + toString(err))
+ }
+
+ // verify write rpcs for Row
+ row := table.Row("row1")
+ if _, err := row.Exists(ctx); !isVersionMismatchErr(err) {
+ t.Fatal("Expected ErrDatabaseVersionMismatch, found: " + toString(err))
+ }
+ var str string
+ if err := row.Get(ctx, &str); !isVersionMismatchErr(err) {
+ t.Fatal("Expected ErrDatabaseVersionMismatch, found: " + toString(err))
+ }
+ if err := row.Put(ctx, "newValue"); !isVersionMismatchErr(err) {
+ t.Fatal("Expected ErrDatabaseVersionMismatch, found: " + toString(err))
+ }
+ if err := row.Delete(ctx); !isVersionMismatchErr(err) {
+ t.Fatal("Expected ErrDatabaseVersionMismatch, found: " + toString(err))
+ }
+}
+
+func TestRPCSchemaCheckErrorForBatch(t *testing.T) {
+ // Setup
+ ctx, sName, cleanup := tu.SetupOrDie(nil)
+ defer cleanup()
+ a := tu.CreateApp(t, ctx, syncbase.NewService(sName), "a")
+ schema := tu.DefaultSchema(0)
+
+ // Create db1 with schema version 0 and add table1 and row1
+ dbHandle1 := a.NoSQLDatabase("db1", schema)
+ if err := dbHandle1.Create(ctx, nil); err != nil {
+ t.Fatalf("db1.Create() failed: %v", err)
+ }
+ if err := dbHandle1.CreateTable(ctx, "table1", nil); err != nil {
+ t.Fatalf("db1.CreateTable() failed: %v", err)
+ }
+ if err := dbHandle1.Table("table1").Put(ctx, "row1", "value1"); err != nil {
+ t.Fatalf("table1.Put() failed: %v", err)
+ }
+
+ // Create three batches using dbHandle1
+ batch1, batchErr1 := dbHandle1.BeginBatch(ctx, wire.BatchOptions{})
+ if batchErr1 != nil {
+ t.Fatalf("db1.BeginBatch() failed: %v", batchErr1)
+ }
+ batch1.Table("table1").Row("row1").Put(ctx, "newValue1")
+
+ batch2, batchErr2 := dbHandle1.BeginBatch(ctx, wire.BatchOptions{})
+ if batchErr2 != nil {
+ t.Fatalf("db1.BeginBatch() failed: %v", batchErr2)
+ }
+ batch2.Table("table1").Row("row1").Put(ctx, "newValue2")
+
+ batch3, batchErr3 := dbHandle1.BeginBatch(ctx, wire.BatchOptions{})
+ if batchErr3 != nil {
+ t.Fatalf("db1.BeginBatch() failed: %v", batchErr3)
+ }
+
+ // Upgrade schema version for underlying db using a different handle
+ schema2 := tu.DefaultSchema(1)
+ dbHandle2 := a.NoSQLDatabase("db1", schema2)
+ dbHandle2.UpgradeIfOutdated(ctx)
+
+ // Commit batch1, abort batch2, attempt writing a row using batch3.
+ // Each of these operations should fail.
+ if err := batch1.Commit(ctx); !isVersionMismatchErr(err) {
+ t.Fatal("Expected ErrDatabaseVersionMismatch, found: " + toString(err))
+ }
+ if err := batch2.Abort(ctx); !isVersionMismatchErr(err) {
+ t.Fatal("Expected ErrDatabaseVersionMismatch, found: " + toString(err))
+ }
+ if err := batch3.Table("table1").Row("row1").Put(ctx, "newValue3"); !isVersionMismatchErr(err) {
+ t.Fatal("Expected ErrDatabaseVersionMismatch, found: " + toString(err))
+ }
+
+ // Verify that the value of row1 is the original value.
+ var value string
+ if err := dbHandle2.Table("table1").Get(ctx, "row1", &value); err != nil {
+ t.Fatalf("table1.Get() failed: %v", err)
+ }
+}
+
+func toString(err error) string {
+ if err == nil {
+ return "nil"
+ }
+ return string(verror.ErrorID(err)) + ": " + err.Error()
+}
+
+func isVersionMismatchErr(err error) bool {
+ if err == nil {
+ return false
+ }
+ return verror.ErrorID(err) == wire.ErrSchemaVersionMismatch.ID
+}
+
func getSchemaMetadata(ctx *context.T, dbName string) (wire.SchemaMetadata, error) {
return wire.DatabaseClient(dbName).GetSchemaMetadata(ctx)
}
diff --git a/v23/syncbase/nosql/table.go b/v23/syncbase/nosql/table.go
index 759706c..05de44f 100644
--- a/v23/syncbase/nosql/table.go
+++ b/v23/syncbase/nosql/table.go
@@ -11,19 +11,21 @@
"v.io/v23/security/access"
)
-func newTable(parentFullName, relativeName string) Table {
+func newTable(parentFullName, relativeName string, schemaVersion int32) Table {
fullName := naming.Join(parentFullName, relativeName)
return &table{
- c: wire.TableClient(fullName),
- fullName: fullName,
- name: relativeName,
+ c: wire.TableClient(fullName),
+ fullName: fullName,
+ name: relativeName,
+ dbSchemaVersion: schemaVersion,
}
}
type table struct {
- c wire.TableClientMethods
- fullName string
- name string
+ c wire.TableClientMethods
+ fullName string
+ name string
+ dbSchemaVersion int32
}
var _ Table = (*table)(nil)
@@ -42,12 +44,12 @@
// Exists implements Table.Exists.
func (t *table) Exists(ctx *context.T) (bool, error) {
- return t.c.Exists(ctx)
+ return t.c.Exists(ctx, t.dbSchemaVersion)
}
// Row implements Table.Row.
func (t *table) Row(key string) Row {
- return newRow(t.fullName, key)
+ return newRow(t.fullName, key, t.dbSchemaVersion)
}
// Get implements Table.Get.
@@ -62,13 +64,13 @@
// Delete implements Table.Delete.
func (t *table) Delete(ctx *context.T, r RowRange) error {
- return t.c.DeleteRowRange(ctx, []byte(r.Start()), []byte(r.Limit()))
+ return t.c.DeleteRowRange(ctx, t.dbSchemaVersion, []byte(r.Start()), []byte(r.Limit()))
}
// Scan implements Table.Scan.
func (t *table) Scan(ctx *context.T, r RowRange) Stream {
ctx, cancel := context.WithCancel(ctx)
- call, err := t.c.Scan(ctx, []byte(r.Start()), []byte(r.Limit()))
+ call, err := t.c.Scan(ctx, t.dbSchemaVersion, []byte(r.Start()), []byte(r.Limit()))
if err != nil {
return &InvalidStream{Error: err}
}
@@ -77,7 +79,7 @@
// GetPermissions implements Table.GetPermissions.
func (t *table) GetPermissions(ctx *context.T, key string) ([]PrefixPermissions, error) {
- wirePermsList, err := t.c.GetPermissions(ctx, key)
+ wirePermsList, err := t.c.GetPermissions(ctx, t.dbSchemaVersion, key)
permsList := []PrefixPermissions{}
for _, v := range wirePermsList {
permsList = append(permsList, PrefixPermissions{Prefix: Prefix(v.Prefix), Perms: v.Perms})
@@ -87,10 +89,10 @@
// SetPermissions implements Table.SetPermissions.
func (t *table) SetPermissions(ctx *context.T, prefix PrefixRange, perms access.Permissions) error {
- return t.c.SetPermissions(ctx, prefix.Prefix(), perms)
+ return t.c.SetPermissions(ctx, t.dbSchemaVersion, prefix.Prefix(), perms)
}
// DeletePermissions implements Table.DeletePermissions.
func (t *table) DeletePermissions(ctx *context.T, prefix PrefixRange) error {
- return t.c.DeletePermissions(ctx, prefix.Prefix())
+ return t.c.DeletePermissions(ctx, t.dbSchemaVersion, prefix.Prefix())
}
diff --git a/v23/syncbase/testutil/util.go b/v23/syncbase/testutil/util.go
index f7c38b1..ce05711 100644
--- a/v23/syncbase/testutil/util.go
+++ b/v23/syncbase/testutil/util.go
@@ -185,17 +185,17 @@
CallCount int
}
-func (msu *MockSchemaUpgrader) Run(db nosql.Database, oldVersion, newVersion int64) error {
+func (msu *MockSchemaUpgrader) Run(db nosql.Database, oldVersion, newVersion int32) error {
msu.CallCount++
return nil
}
var _ nosql.SchemaUpgrader = (*MockSchemaUpgrader)(nil)
-func DefaultSchema() *nosql.Schema {
+func DefaultSchema(version int32) *nosql.Schema {
return &nosql.Schema{
Metadata: wire.SchemaMetadata{
- Version: 0,
+ Version: version,
},
Upgrader: nosql.SchemaUpgrader(&MockSchemaUpgrader{}),
}
diff --git a/x/ref/services/syncbase/server/nosql/database.go b/x/ref/services/syncbase/server/nosql/database.go
index a9e7ac9..3fe1b96 100644
--- a/x/ref/services/syncbase/server/nosql/database.go
+++ b/x/ref/services/syncbase/server/nosql/database.go
@@ -133,7 +133,7 @@
////////////////////////////////////////
// RPC methods
-func (d *databaseReq) Create(ctx *context.T, call rpc.ServerCall, perms access.Permissions, metadata *wire.SchemaMetadata) error {
+func (d *databaseReq) Create(ctx *context.T, call rpc.ServerCall, metadata *wire.SchemaMetadata, perms access.Permissions) error {
if d.exists {
return verror.New(verror.ErrExist, ctx, d.name)
}
@@ -146,29 +146,39 @@
return d.a.CreateNoSQLDatabase(ctx, call, d.name, perms, metadata)
}
-func (d *databaseReq) Delete(ctx *context.T, call rpc.ServerCall) error {
+func (d *databaseReq) Delete(ctx *context.T, call rpc.ServerCall, schemaVersion int32) error {
if d.batchId != nil {
return wire.NewErrBoundToBatch(ctx)
}
+ if err := d.checkSchemaVersion(ctx, schemaVersion); err != nil {
+ return err
+ }
return d.a.DeleteNoSQLDatabase(ctx, call, d.name)
}
-func (d *databaseReq) Exists(ctx *context.T, call rpc.ServerCall) (bool, error) {
+func (d *databaseReq) Exists(ctx *context.T, call rpc.ServerCall, schemaVersion int32) (bool, error) {
if !d.exists {
return false, nil
}
+ if err := d.checkSchemaVersion(ctx, schemaVersion); err != nil {
+ return false, err
+ }
return util.ErrorToExists(util.GetWithAuth(ctx, call, d.st, d.stKey(), &databaseData{}))
}
var rng *rand.Rand = rand.New(rand.NewSource(time.Now().UTC().UnixNano()))
-func (d *databaseReq) BeginBatch(ctx *context.T, call rpc.ServerCall, bo wire.BatchOptions) (string, error) {
+func (d *databaseReq) BeginBatch(ctx *context.T, call rpc.ServerCall, schemaVersion int32, bo wire.BatchOptions) (string, error) {
if !d.exists {
return "", verror.New(verror.ErrNoExist, ctx, d.name)
}
if d.batchId != nil {
return "", wire.NewErrBoundToBatch(ctx)
}
+ if err := d.checkSchemaVersion(ctx, schemaVersion); err != nil {
+ return "", err
+ }
+
d.mu.Lock()
defer d.mu.Unlock()
var id uint64
@@ -192,7 +202,7 @@
return strings.Join([]string{d.name, batchType, strconv.FormatUint(id, 10)}, util.BatchSep), nil
}
-func (d *databaseReq) Commit(ctx *context.T, call rpc.ServerCall) error {
+func (d *databaseReq) Commit(ctx *context.T, call rpc.ServerCall, schemaVersion int32) error {
if !d.exists {
return verror.New(verror.ErrNoExist, ctx, d.name)
}
@@ -202,6 +212,9 @@
if d.tx == nil {
return wire.NewErrReadOnlyBatch(ctx)
}
+ if err := d.checkSchemaVersion(ctx, schemaVersion); err != nil {
+ return err
+ }
var err error
if err = d.tx.Commit(); err == nil {
d.mu.Lock()
@@ -214,13 +227,16 @@
return err
}
-func (d *databaseReq) Abort(ctx *context.T, call rpc.ServerCall) error {
+func (d *databaseReq) Abort(ctx *context.T, call rpc.ServerCall, schemaVersion int32) error {
if !d.exists {
return verror.New(verror.ErrNoExist, ctx, d.name)
}
if d.batchId == nil {
return wire.NewErrNotBoundToBatch(ctx)
}
+ if err := d.checkSchemaVersion(ctx, schemaVersion); err != nil {
+ return err
+ }
var err error
if d.tx != nil {
if err = d.tx.Abort(); err == nil {
@@ -238,7 +254,10 @@
return err
}
-func (d *databaseReq) Exec(ctx *context.T, call wire.DatabaseExecServerCall, q string) error {
+func (d *databaseReq) Exec(ctx *context.T, call wire.DatabaseExecServerCall, schemaVersion int32, q string) error {
+ if err := d.checkSchemaVersion(ctx, schemaVersion); err != nil {
+ return err
+ }
impl := func(headers []string, rs ResultStream, err error) error {
if err != nil {
return err
@@ -568,3 +587,22 @@
return nil, wire.NewErrReadOnlyBatch(nil)
}
}
+
+// TODO(jlodhia): Schema check should happen within a transaction for each
+// operation in database, table and row. Do schema check along with permissions
+// check when fully-specified permission model is implemented.
+func (d *databaseReq) checkSchemaVersion(ctx *context.T, schemaVersion int32) error {
+ if !d.exists {
+ // database does not exist yet and hence there is no schema to check.
+ // This can happen if delete is called twice on the same database.
+ return nil
+ }
+ schemaMetadata, err := d.getSchemaMetadataWithoutAuth(ctx)
+ if err != nil {
+ return err
+ }
+ if (schemaMetadata == nil) || (schemaMetadata.Version == schemaVersion) {
+ return nil
+ }
+ return wire.NewErrSchemaVersionMismatch(ctx)
+}
diff --git a/x/ref/services/syncbase/server/nosql/database_sm.go b/x/ref/services/syncbase/server/nosql/database_sm.go
index 5f69585..5d1239e 100644
--- a/x/ref/services/syncbase/server/nosql/database_sm.go
+++ b/x/ref/services/syncbase/server/nosql/database_sm.go
@@ -37,7 +37,7 @@
func (d *databaseReq) SetSchemaMetadata(ctx *context.T, call rpc.ServerCall, metadata wire.SchemaMetadata) error {
// Check if database exists
if !d.exists {
- return verror.New(verror.ErrNoExist, ctx, "database: "+d.Name())
+ return verror.New(verror.ErrNoExist, ctx, d.Name())
}
// Check permissions on Database and store schema metadata.
@@ -51,3 +51,14 @@
})
})
}
+
+func (d *databaseReq) getSchemaMetadataWithoutAuth(ctx *context.T) (*wire.SchemaMetadata, error) {
+ if !d.exists {
+ return nil, verror.New(verror.ErrInternal, ctx, "field store in database cannot be nil")
+ }
+ dbData := databaseData{}
+ if err := util.Get(ctx, d.st, d.stKey(), &dbData); err != nil {
+ return nil, err
+ }
+ return dbData.SchemaMetadata, nil
+}
diff --git a/x/ref/services/syncbase/server/nosql/row.go b/x/ref/services/syncbase/server/nosql/row.go
index a562c6d..62a5f3a 100644
--- a/x/ref/services/syncbase/server/nosql/row.go
+++ b/x/ref/services/syncbase/server/nosql/row.go
@@ -26,13 +26,16 @@
////////////////////////////////////////
// RPC methods
-func (r *rowReq) Exists(ctx *context.T, call rpc.ServerCall) (bool, error) {
- _, err := r.Get(ctx, call)
+func (r *rowReq) Exists(ctx *context.T, call rpc.ServerCall, schemaVersion int32) (bool, error) {
+ _, err := r.Get(ctx, call, schemaVersion)
return util.ErrorToExists(err)
}
-func (r *rowReq) Get(ctx *context.T, call rpc.ServerCall) ([]byte, error) {
+func (r *rowReq) Get(ctx *context.T, call rpc.ServerCall, schemaVersion int32) ([]byte, error) {
impl := func(st store.StoreReader) ([]byte, error) {
+ if err := r.t.d.checkSchemaVersion(ctx, schemaVersion); err != nil {
+ return []byte{}, err
+ }
return r.get(ctx, call, st)
}
var st store.StoreReader
@@ -46,8 +49,11 @@
return impl(st)
}
-func (r *rowReq) Put(ctx *context.T, call rpc.ServerCall, value []byte) error {
+func (r *rowReq) Put(ctx *context.T, call rpc.ServerCall, schemaVersion int32, value []byte) error {
impl := func(st store.StoreReadWriter) error {
+ if err := r.t.d.checkSchemaVersion(ctx, schemaVersion); err != nil {
+ return err
+ }
return r.put(ctx, call, st, value)
}
if r.t.d.batchId != nil {
@@ -61,8 +67,11 @@
}
}
-func (r *rowReq) Delete(ctx *context.T, call rpc.ServerCall) error {
+func (r *rowReq) Delete(ctx *context.T, call rpc.ServerCall, schemaVersion int32) error {
impl := func(st store.StoreReadWriter) error {
+ if err := r.t.d.checkSchemaVersion(ctx, schemaVersion); err != nil {
+ return err
+ }
return r.delete(ctx, call, st)
}
if r.t.d.batchId != nil {
diff --git a/x/ref/services/syncbase/server/nosql/table.go b/x/ref/services/syncbase/server/nosql/table.go
index 2eb1f25..7b371e0 100644
--- a/x/ref/services/syncbase/server/nosql/table.go
+++ b/x/ref/services/syncbase/server/nosql/table.go
@@ -30,10 +30,13 @@
////////////////////////////////////////
// RPC methods
-func (t *tableReq) Create(ctx *context.T, call rpc.ServerCall, perms access.Permissions) error {
+func (t *tableReq) Create(ctx *context.T, call rpc.ServerCall, schemaVersion int32, perms access.Permissions) error {
if t.d.batchId != nil {
return wire.NewErrBoundToBatch(ctx)
}
+ if err := t.d.checkSchemaVersion(ctx, schemaVersion); err != nil {
+ return err
+ }
return store.RunInTransaction(t.d.st, func(st store.StoreReadWriter) error {
// Check databaseData perms.
dData := &databaseData{}
@@ -60,10 +63,13 @@
})
}
-func (t *tableReq) Delete(ctx *context.T, call rpc.ServerCall) error {
+func (t *tableReq) Delete(ctx *context.T, call rpc.ServerCall, schemaVersion int32) error {
if t.d.batchId != nil {
return wire.NewErrBoundToBatch(ctx)
}
+ if err := t.d.checkSchemaVersion(ctx, schemaVersion); err != nil {
+ return err
+ }
return store.RunInTransaction(t.d.st, func(st store.StoreReadWriter) error {
// Read-check-delete tableData.
if err := util.GetWithAuth(ctx, call, st, t.stKey(), &tableData{}); err != nil {
@@ -77,16 +83,24 @@
})
}
-func (t *tableReq) Exists(ctx *context.T, call rpc.ServerCall) (bool, error) {
+func (t *tableReq) Exists(ctx *context.T, call rpc.ServerCall, schemaVersion int32) (bool, error) {
+ if err := t.d.checkSchemaVersion(ctx, schemaVersion); err != nil {
+ return false, err
+ }
return util.ErrorToExists(util.GetWithAuth(ctx, call, t.d.st, t.stKey(), &tableData{}))
}
-func (t *tableReq) DeleteRowRange(ctx *context.T, call rpc.ServerCall, start, limit []byte) error {
+func (t *tableReq) DeleteRowRange(ctx *context.T, call rpc.ServerCall, schemaVersion int32, start, limit []byte) error {
impl := func(st store.StoreReadWriter) error {
// Check for table-level access before doing a scan.
if err := t.checkAccess(ctx, call, st, ""); err != nil {
return err
}
+ // Check if the db schema version and the version provided by client
+ // matches.
+ if err := t.d.checkSchemaVersion(ctx, schemaVersion); err != nil {
+ return err
+ }
it := st.Scan(util.ScanRangeArgs(util.JoinKeyParts(util.RowPrefix, t.name), string(start), string(limit)))
key := []byte{}
for it.Advance() {
@@ -121,12 +135,15 @@
}
}
-func (t *tableReq) Scan(ctx *context.T, call wire.TableScanServerCall, start, limit []byte) error {
+func (t *tableReq) Scan(ctx *context.T, call wire.TableScanServerCall, schemaVersion int32, start, limit []byte) error {
impl := func(st store.StoreReader) error {
// Check for table-level access before doing a scan.
if err := t.checkAccess(ctx, call, st, ""); err != nil {
return err
}
+ if err := t.d.checkSchemaVersion(ctx, schemaVersion); err != nil {
+ return err
+ }
it := st.Scan(util.ScanRangeArgs(util.JoinKeyParts(util.RowPrefix, t.name), string(start), string(limit)))
sender := call.SendStream()
key, value := []byte{}, []byte{}
@@ -157,12 +174,15 @@
return impl(st)
}
-func (t *tableReq) GetPermissions(ctx *context.T, call rpc.ServerCall, key string) ([]wire.PrefixPermissions, error) {
+func (t *tableReq) GetPermissions(ctx *context.T, call rpc.ServerCall, schemaVersion int32, key string) ([]wire.PrefixPermissions, error) {
impl := func(st store.StoreReader) ([]wire.PrefixPermissions, error) {
// Check permissions only at table level.
if err := t.checkAccess(ctx, call, st, ""); err != nil {
return nil, err
}
+ if err := t.d.checkSchemaVersion(ctx, schemaVersion); err != nil {
+ return nil, err
+ }
// Get the most specific permissions object.
prefix, prefixPerms, err := t.permsForKey(ctx, st, key)
if err != nil {
@@ -190,11 +210,14 @@
return impl(st)
}
-func (t *tableReq) SetPermissions(ctx *context.T, call rpc.ServerCall, prefix string, perms access.Permissions) error {
+func (t *tableReq) SetPermissions(ctx *context.T, call rpc.ServerCall, schemaVersion int32, prefix string, perms access.Permissions) error {
impl := func(st store.StoreReadWriter) error {
if err := t.checkAccess(ctx, call, st, prefix); err != nil {
return err
}
+ if err := t.d.checkSchemaVersion(ctx, schemaVersion); err != nil {
+ return err
+ }
// Concurrent transactions that touch this table should fail with
// ErrConcurrentTransaction when this transaction commits.
if err := t.lock(ctx, st); err != nil {
@@ -243,7 +266,7 @@
}
}
-func (t *tableReq) DeletePermissions(ctx *context.T, call rpc.ServerCall, prefix string) error {
+func (t *tableReq) DeletePermissions(ctx *context.T, call rpc.ServerCall, schemaVersion int32, prefix string) error {
if prefix == "" {
return verror.New(verror.ErrBadArg, ctx, prefix)
}
@@ -251,6 +274,9 @@
if err := t.checkAccess(ctx, call, st, prefix); err != nil {
return err
}
+ if err := t.d.checkSchemaVersion(ctx, schemaVersion); err != nil {
+ return err
+ }
// Concurrent transactions that touch this table should fail with
// ErrConcurrentTransaction when this transaction commits.
if err := t.lock(ctx, st); err != nil {
