diff --git a/v23/services/syncbase/nosql/service.vdl b/v23/services/syncbase/nosql/service.vdl
index 8803897..c990976 100644
--- a/v23/services/syncbase/nosql/service.vdl
+++ b/v23/services/syncbase/nosql/service.vdl
@@ -67,10 +67,7 @@
 	Delete() error {access.Write}
 
 	// Delete deletes all rows in the given half-open range [start, limit). If
-	// limit is "", all rows with keys >= start are included. If the last row that
-	// is covered by a prefix from SetPermissions is deleted, that (prefix, perms)
-	// pair is removed.
-	// TODO(sadovsky): Automatic GC interacts poorly with sync. Revisit this API.
+	// limit is "", all rows with keys >= start are included.
 	DeleteRowRange(start, limit []byte) error {access.Write}
 
 	// Scan returns all rows in the given half-open range [start, limit). If limit
@@ -78,6 +75,13 @@
 	// from a consistent snapshot taken at the time of the Scan RPC.
 	Scan(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}
+
 	// 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:
@@ -85,18 +89,8 @@
 	//     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 will fail if called with a prefix that does not match any
-	// rows.
 	SetPermissions(prefix string, perms access.Permissions) error {access.Admin}
 
-	// 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}
-
 	// 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).
diff --git a/v23/services/syncbase/nosql/service.vdl.go b/v23/services/syncbase/nosql/service.vdl.go
index 5319420..2af0a60 100644
--- a/v23/services/syncbase/nosql/service.vdl.go
+++ b/v23/services/syncbase/nosql/service.vdl.go
@@ -954,15 +954,18 @@
 	// Delete deletes this Table.
 	Delete(*context.T, ...rpc.CallOpt) error
 	// Delete deletes all rows in the given half-open range [start, limit). If
-	// limit is "", all rows with keys >= start are included. If the last row that
-	// is covered by a prefix from SetPermissions is deleted, that (prefix, perms)
-	// pair is removed.
-	// TODO(sadovsky): Automatic GC interacts poorly with sync. Revisit this API.
+	// limit is "", all rows with keys >= start are included.
 	DeleteRowRange(ctx *context.T, 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. The returned stream reads
 	// from a consistent snapshot taken at the time of the Scan RPC.
 	Scan(ctx *context.T, 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)
 	// 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:
@@ -970,16 +973,7 @@
 	//     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 will fail if called with a prefix that does not match any
-	// rows.
 	SetPermissions(ctx *context.T, prefix string, perms access.Permissions, opts ...rpc.CallOpt) 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)
 	// 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).
@@ -1025,13 +1019,13 @@
 	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) 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...)
 	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) 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...)
 	return
 }
 
@@ -1121,15 +1115,18 @@
 	// Delete deletes this Table.
 	Delete(*context.T, rpc.ServerCall) error
 	// Delete deletes all rows in the given half-open range [start, limit). If
-	// limit is "", all rows with keys >= start are included. If the last row that
-	// is covered by a prefix from SetPermissions is deleted, that (prefix, perms)
-	// pair is removed.
-	// TODO(sadovsky): Automatic GC interacts poorly with sync. Revisit this API.
+	// limit is "", all rows with keys >= start are included.
 	DeleteRowRange(ctx *context.T, call rpc.ServerCall, 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. The returned stream reads
 	// from a consistent snapshot taken at the time of the Scan RPC.
 	Scan(ctx *context.T, call TableScanServerCall, 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)
 	// 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:
@@ -1137,16 +1134,7 @@
 	//     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 will fail if called with a prefix that does not match any
-	// rows.
 	SetPermissions(ctx *context.T, call rpc.ServerCall, prefix string, perms access.Permissions) 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)
 	// 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).
@@ -1164,15 +1152,18 @@
 	// Delete deletes this Table.
 	Delete(*context.T, rpc.ServerCall) error
 	// Delete deletes all rows in the given half-open range [start, limit). If
-	// limit is "", all rows with keys >= start are included. If the last row that
-	// is covered by a prefix from SetPermissions is deleted, that (prefix, perms)
-	// pair is removed.
-	// TODO(sadovsky): Automatic GC interacts poorly with sync. Revisit this API.
+	// limit is "", all rows with keys >= start are included.
 	DeleteRowRange(ctx *context.T, call rpc.ServerCall, 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. The returned stream reads
 	// from a consistent snapshot taken at the time of the Scan RPC.
 	Scan(ctx *context.T, call *TableScanServerCallStub, 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)
 	// 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:
@@ -1180,16 +1171,7 @@
 	//     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 will fail if called with a prefix that does not match any
-	// rows.
 	SetPermissions(ctx *context.T, call rpc.ServerCall, prefix string, perms access.Permissions) 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)
 	// 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).
@@ -1241,14 +1223,14 @@
 	return s.impl.Scan(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) GetPermissions(ctx *context.T, call rpc.ServerCall, i0 string) ([]PrefixPermissions, error) {
 	return s.impl.GetPermissions(ctx, call, i0)
 }
 
+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) DeletePermissions(ctx *context.T, call rpc.ServerCall, i0 string) error {
 	return s.impl.DeletePermissions(ctx, call, i0)
 }
@@ -1285,7 +1267,7 @@
 		},
 		{
 			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. If the last row that\n// is covered by a prefix from SetPermissions is deleted, that (prefix, perms)\n// pair is removed.\n// TODO(sadovsky): Automatic GC interacts poorly with sync. Revisit this API.",
+			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
@@ -1302,15 +1284,6 @@
 			Tags: []*vdl.Value{vdl.ValueOf(access.Tag("Read"))},
 		},
 		{
-			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.\n//\n// SetPermissions will fail if called with a prefix that does not match any\n// rows.",
-			InArgs: []rpc.ArgDesc{
-				{"prefix", ``}, // string
-				{"perms", ``},  // access.Permissions
-			},
-			Tags: []*vdl.Value{vdl.ValueOf(access.Tag("Admin"))},
-		},
-		{
 			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{
@@ -1322,6 +1295,15 @@
 			Tags: []*vdl.Value{vdl.ValueOf(access.Tag("Admin"))},
 		},
 		{
+			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
+			},
+			Tags: []*vdl.Value{vdl.ValueOf(access.Tag("Admin"))},
+		},
+		{
 			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{
diff --git a/v23/syncbase/nosql/client_test.go b/v23/syncbase/nosql/client_test.go
index 435ee50..38f66d2 100644
--- a/v23/syncbase/nosql/client_test.go
+++ b/v23/syncbase/nosql/client_test.go
@@ -19,6 +19,7 @@
 )
 
 // TODO(sadovsky): Finish writing tests.
+// TODO(rogulenko): Test perms checking for Glob and Exec.
 
 // Tests various Name, FullName, and Key methods.
 func TestNameAndKey(t *testing.T) {
@@ -177,14 +178,160 @@
 
 // Tests that Table.{Set,Get,Delete}Permissions methods work as expected.
 func TestTablePerms(t *testing.T) {
-	// TODO(sadovsky): Implement.
+	clientACtx, sName, cleanup, sp, ctx := tu.SetupOrDieCustom("clientA", "server", nil)
+	defer cleanup()
+	clientBCtx := tu.NewClient("clientB", "server", ctx, sp)
+	a := tu.CreateApp(t, clientACtx, syncbase.NewService(sName), "a")
+	d := tu.CreateNoSQLDatabase(t, clientACtx, a, "d")
+	tb := tu.CreateTable(t, clientACtx, d, "tb")
+
+	// Permission objects.
+	aAndB := tu.DefaultPerms("server/clientA", "server/clientB")
+	aOnly := tu.DefaultPerms("server/clientA")
+	bOnly := tu.DefaultPerms("server/clientB")
+
+	// Set initial permissions.
+	if err := tb.SetPermissions(clientACtx, nosql.Prefix(""), aAndB); err != nil {
+		t.Fatalf("tb.SetPermissions() failed: %v", err)
+	}
+	if err := tb.SetPermissions(clientACtx, nosql.Prefix("prefix"), aAndB); err != nil {
+		t.Fatalf("tb.SetPermissions() failed: %v", err)
+	}
+	if err := tb.SetPermissions(clientBCtx, nosql.Prefix("prefix"), aAndB); err != nil {
+		t.Fatalf("tb.SetPermissions() failed: %v", err)
+	}
+	if err := tb.SetPermissions(clientACtx, nosql.Prefix("prefix_a"), aOnly); err != nil {
+		t.Fatalf("tb.SetPermissions() failed: %v", err)
+	}
+	if err := tb.SetPermissions(clientBCtx, nosql.Prefix("prefix_b"), bOnly); err != nil {
+		t.Fatalf("tb.SetPermissions() failed: %v", err)
+	}
+
+	// Checks A has no access to 'prefix_b' and vice versa.
+	if err := tb.SetPermissions(clientACtx, nosql.Prefix("prefix_b"), aOnly); verror.ErrorID(err) != verror.ErrNoAccess.ID {
+		t.Fatalf("tb.SetPermissions() should have failed: %v", err)
+	}
+	if err := tb.SetPermissions(clientACtx, nosql.Prefix("prefix_b_suffix"), aOnly); verror.ErrorID(err) != verror.ErrNoAccess.ID {
+		t.Fatalf("tb.SetPermissions() should have failed: %v", err)
+	}
+	if err := tb.SetPermissions(clientBCtx, nosql.Prefix("prefix_a"), bOnly); verror.ErrorID(err) != verror.ErrNoAccess.ID {
+		t.Fatalf("tb.SetPermissions() should have failed: %v", err)
+	}
+	if err := tb.SetPermissions(clientBCtx, nosql.Prefix("prefix_a_suffix"), bOnly); verror.ErrorID(err) != verror.ErrNoAccess.ID {
+		t.Fatalf("tb.SetPermissions() should have failed: %v", err)
+	}
+
+	// Check GetPermissions.
+	wantPerms := []nosql.PrefixPermissions{
+		nosql.PrefixPermissions{Prefix: nosql.Prefix(""), Perms: aAndB},
+	}
+	if got, _ := tb.GetPermissions(clientACtx, ""); !reflect.DeepEqual(got, wantPerms) {
+		t.Fatalf("Unexpected permissions: got %v, want %v", got, wantPerms)
+	}
+	if got, _ := tb.GetPermissions(clientACtx, "abc"); !reflect.DeepEqual(got, wantPerms) {
+		t.Fatalf("Unexpected permissions: got %v, want %v", got, wantPerms)
+	}
+	wantPerms = []nosql.PrefixPermissions{
+		nosql.PrefixPermissions{Prefix: nosql.Prefix("prefix"), Perms: aAndB},
+		nosql.PrefixPermissions{Prefix: nosql.Prefix(""), Perms: aAndB},
+	}
+	if got, _ := tb.GetPermissions(clientACtx, "prefix"); !reflect.DeepEqual(got, wantPerms) {
+		t.Fatalf("Unexpected permissions: got %v, want %v", got, wantPerms)
+	}
+	if got, _ := tb.GetPermissions(clientACtx, "prefix_c"); !reflect.DeepEqual(got, wantPerms) {
+		t.Fatalf("Unexpected permissions: got %v, want %v", got, wantPerms)
+	}
+	wantPerms = []nosql.PrefixPermissions{
+		nosql.PrefixPermissions{Prefix: nosql.Prefix("prefix_a"), Perms: aOnly},
+		nosql.PrefixPermissions{Prefix: nosql.Prefix("prefix"), Perms: aAndB},
+		nosql.PrefixPermissions{Prefix: nosql.Prefix(""), Perms: aAndB},
+	}
+	if got, _ := tb.GetPermissions(clientACtx, "prefix_a"); !reflect.DeepEqual(got, wantPerms) {
+		t.Fatalf("Unexpected permissions: got %v, want %v", got, wantPerms)
+	}
+	if got, _ := tb.GetPermissions(clientACtx, "prefix_a_suffix"); !reflect.DeepEqual(got, wantPerms) {
+		t.Fatalf("Unexpected permissions: got %v, want %v", got, wantPerms)
+	}
+	wantPerms = []nosql.PrefixPermissions{
+		nosql.PrefixPermissions{Prefix: nosql.Prefix("prefix_b"), Perms: bOnly},
+		nosql.PrefixPermissions{Prefix: nosql.Prefix("prefix"), Perms: aAndB},
+		nosql.PrefixPermissions{Prefix: nosql.Prefix(""), Perms: aAndB},
+	}
+	if got, _ := tb.GetPermissions(clientACtx, "prefix_b"); !reflect.DeepEqual(got, wantPerms) {
+		t.Fatalf("Unexpected permissions: got %v, want %v", got, wantPerms)
+	}
+	if got, _ := tb.GetPermissions(clientACtx, "prefix_b_suffix"); !reflect.DeepEqual(got, wantPerms) {
+		t.Fatalf("Unexpected permissions: got %v, want %v", got, wantPerms)
+	}
+
+	// Delete some prefix permissions and check again.
+	// Check that A can't delete permissions of B.
+	if err := tb.DeletePermissions(clientACtx, nosql.Prefix("prefix_b")); verror.ErrorID(err) != verror.ErrNoAccess.ID {
+		t.Fatalf("tb.DeletePermissions() should have failed: %v", err)
+	}
+	if err := tb.DeletePermissions(clientBCtx, nosql.Prefix("prefix_a")); verror.ErrorID(err) != verror.ErrNoAccess.ID {
+		t.Fatalf("tb.DeletePermissions() should have failed: %v", err)
+	}
+	// Delete 'prefix' and 'prefix_a'
+	if err := tb.DeletePermissions(clientACtx, nosql.Prefix("prefix")); err != nil {
+		t.Fatalf("tb.DeletePermissions() failed: %v", err)
+	}
+	if err := tb.DeletePermissions(clientACtx, nosql.Prefix("prefix_a")); err != nil {
+		t.Fatalf("tb.DeletePermissions() failed: %v", err)
+	}
+	// Check DeletePermissions is idempotent.
+	if err := tb.DeletePermissions(clientACtx, nosql.Prefix("prefix")); err != nil {
+		t.Fatalf("tb.DeletePermissions() failed: %v", err)
+	}
+
+	// Check GetPermissions again.
+	wantPerms = []nosql.PrefixPermissions{
+		nosql.PrefixPermissions{Prefix: nosql.Prefix(""), Perms: aAndB},
+	}
+	if got, _ := tb.GetPermissions(clientACtx, ""); !reflect.DeepEqual(got, wantPerms) {
+		t.Fatalf("Unexpected permissions: got %v, want %v", got, wantPerms)
+	}
+	if got, _ := tb.GetPermissions(clientACtx, "prefix"); !reflect.DeepEqual(got, wantPerms) {
+		t.Fatalf("Unexpected permissions: got %v, want %v", got, wantPerms)
+	}
+	if got, _ := tb.GetPermissions(clientACtx, "prefix_a"); !reflect.DeepEqual(got, wantPerms) {
+		t.Fatalf("Unexpected permissions: got %v, want %v", got, wantPerms)
+	}
+	if got, _ := tb.GetPermissions(clientACtx, "prefix_a_suffix"); !reflect.DeepEqual(got, wantPerms) {
+		t.Fatalf("Unexpected permissions: got %v, want %v", got, wantPerms)
+	}
+	wantPerms = []nosql.PrefixPermissions{
+		nosql.PrefixPermissions{Prefix: nosql.Prefix("prefix_b"), Perms: bOnly},
+		nosql.PrefixPermissions{Prefix: nosql.Prefix(""), Perms: aAndB},
+	}
+	if got, _ := tb.GetPermissions(clientACtx, "prefix_b"); !reflect.DeepEqual(got, wantPerms) {
+		t.Fatalf("Unexpected permissions: got %v, want %v", got, wantPerms)
+	}
+	if got, _ := tb.GetPermissions(clientACtx, "prefix_b_suffix"); !reflect.DeepEqual(got, wantPerms) {
+		t.Fatalf("Unexpected permissions: got %v, want %v", got, wantPerms)
+	}
+
+	// Remove B from table-level permissions and check B has no access.
+	if err := tb.SetPermissions(clientACtx, nosql.Prefix(""), aOnly); err != nil {
+		t.Fatalf("tb.SetPermissions() failed: %v", err)
+	}
+	if _, err := tb.GetPermissions(clientBCtx, ""); verror.ErrorID(err) != verror.ErrNoAccess.ID {
+		t.Fatalf("tb.GetPermissions() should have failed: %v", err)
+	}
+	if _, err := tb.GetPermissions(clientBCtx, "prefix_b"); verror.ErrorID(err) != verror.ErrNoAccess.ID {
+		t.Fatalf("tb.GetPermissions() should have failed: %v", err)
+	}
+	if err := tb.SetPermissions(clientBCtx, nosql.Prefix(""), bOnly); verror.ErrorID(err) != verror.ErrNoAccess.ID {
+		t.Fatalf("tb.SetPermissions() should have failed: %v", err)
+	}
+	if err := tb.SetPermissions(clientBCtx, nosql.Prefix("prefix_b"), bOnly); verror.ErrorID(err) != verror.ErrNoAccess.ID {
+		t.Fatalf("tb.SetPermissions() should have failed: %v", err)
+	}
 }
 
 ////////////////////////////////////////
 // Tests involving rows
 
-// TODO(sadovsky): Test perms-checking in all test functions below.
-
 type Foo struct {
 	I int
 	S string
@@ -365,3 +512,68 @@
 		t.Fatalf("r.Get() should have failed: %v", err)
 	}
 }
+
+// Test permission checking in Row.{Get,Put,Delete} and
+// Table.{Scan, DeleteRowRange}.
+func TestRowPermissions(t *testing.T) {
+	clientACtx, sName, cleanup, sp, ctx := tu.SetupOrDieCustom("clientA", "server", nil)
+	defer cleanup()
+	clientBCtx := tu.NewClient("clientB", "server", ctx, sp)
+	a := tu.CreateApp(t, clientACtx, syncbase.NewService(sName), "a")
+	d := tu.CreateNoSQLDatabase(t, clientACtx, a, "d")
+	tb := tu.CreateTable(t, clientACtx, d, "tb")
+
+	// Permission objects.
+	aAndB := tu.DefaultPerms("server/clientA", "server/clientB")
+	aOnly := tu.DefaultPerms("server/clientA")
+	bOnly := tu.DefaultPerms("server/clientB")
+
+	// Set initial permissions.
+	if err := tb.SetPermissions(clientACtx, nosql.Prefix(""), aAndB); err != nil {
+		t.Fatalf("tb.SetPermissions() failed: %v", err)
+	}
+	if err := tb.SetPermissions(clientACtx, nosql.Prefix("a"), aOnly); err != nil {
+		t.Fatalf("tb.SetPermissions() failed: %v", err)
+	}
+	if err := tb.SetPermissions(clientBCtx, nosql.Prefix("b"), bOnly); err != nil {
+		t.Fatalf("tb.SetPermissions() failed: %v", err)
+	}
+
+	// Add some key-value pairs.
+	ra := tb.Row("afoo")
+	rb := tb.Row("bfoo")
+	if err := ra.Put(clientACtx, Foo{}); err != nil {
+		t.Fatalf("ra.Put() failed: %v", err)
+	}
+	if err := rb.Put(clientBCtx, Foo{}); err != nil {
+		t.Fatalf("rb.Put() failed: %v", err)
+	}
+
+	// Check A doesn't have access to 'b'.
+	if err := rb.Get(clientACtx, &Foo{}); verror.ErrorID(err) != verror.ErrNoAccess.ID {
+		t.Fatalf("rb.Get() should have failed: %v", err)
+	}
+	if err := rb.Put(clientACtx, Foo{}); verror.ErrorID(err) != verror.ErrNoAccess.ID {
+		t.Fatalf("rb.Put() should have failed: %v", err)
+	}
+	if err := rb.Delete(clientACtx); verror.ErrorID(err) != verror.ErrNoAccess.ID {
+		t.Fatalf("rb.Delete() should have failed: %v", err)
+	}
+	// Test Table.Delete and Scan.
+	if err := tb.Delete(clientACtx, nosql.Prefix("")); verror.ErrorID(err) != verror.ErrNoAccess.ID {
+		t.Fatalf("tb.Delete should have failed: %v", err)
+	}
+	s := tb.Scan(clientACtx, nosql.Prefix(""))
+	if !s.Advance() {
+		t.Fatalf("Stream should have advanced: %v", s.Err())
+	}
+	if s.Key() != "afoo" {
+		t.Fatalf("Unexpected key: got %q, want %q", s.Key(), "afoo")
+	}
+	if s.Advance() {
+		t.Fatalf("Stream advanced unexpectedly")
+	}
+	if err := s.Err(); verror.ErrorID(err) != verror.ErrNoAccess.ID {
+		t.Fatalf("Unexpected stream error: %v", err)
+	}
+}
diff --git a/v23/syncbase/nosql/internal/query/demo/db/db.go b/v23/syncbase/nosql/internal/query/demo/db/db.go
index 71434a2..8a96f0b 100644
--- a/v23/syncbase/nosql/internal/query/demo/db/db.go
+++ b/v23/syncbase/nosql/internal/query/demo/db/db.go
@@ -7,7 +7,6 @@
 import (
 	"errors"
 	"fmt"
-	"strings"
 	"time"
 
 	"v.io/syncbase/v23/syncbase/nosql/query_db"
@@ -43,10 +42,10 @@
 }
 
 type keyValueStreamImpl struct {
-	table        table
-	cursor       int
-	prefixes     []string
-	prefixCursor int
+	table       table
+	cursor      int
+	keyRanges   query_db.KeyRanges
+	rangeCursor int
 }
 
 func (kvs *keyValueStreamImpl) Advance() bool {
@@ -55,16 +54,17 @@
 		if kvs.cursor >= len(kvs.table.rows) {
 			return false
 		}
-		for kvs.prefixCursor < len(kvs.prefixes) {
-			// does it match any prefix
-			if kvs.prefixes[kvs.prefixCursor] == "" || strings.HasPrefix(kvs.table.rows[kvs.cursor].key, kvs.prefixes[kvs.prefixCursor]) {
+		for kvs.rangeCursor < len(kvs.keyRanges) {
+			// does it match any keyRange (or is the keyRange the 0-255 wildcard)?
+			if (kvs.keyRanges[kvs.rangeCursor].Start == string([]byte{0}) && kvs.keyRanges[kvs.rangeCursor].Limit == string([]byte{255})) ||
+				(kvs.table.rows[kvs.cursor].key >= kvs.keyRanges[kvs.rangeCursor].Start && kvs.table.rows[kvs.cursor].key <= kvs.keyRanges[kvs.rangeCursor].Limit) {
 				return true
 			}
-			// Keys and prefixes are both sorted low to high, so we can increment
-			// prefixCursor if the prefix is < the key.
-			if kvs.prefixes[kvs.prefixCursor] < kvs.table.rows[kvs.cursor].key {
-				kvs.prefixCursor++
-				if kvs.prefixCursor >= len(kvs.prefixes) {
+			// Keys and keyRanges are both sorted low to high, so we can increment
+			// rangeCursor if the keyRange.Limit is < the key.
+			if kvs.keyRanges[kvs.rangeCursor].Limit < kvs.table.rows[kvs.cursor].key {
+				kvs.rangeCursor++
+				if kvs.rangeCursor >= len(kvs.keyRanges) {
 					return false
 				}
 			} else {
@@ -86,11 +86,11 @@
 func (kvs *keyValueStreamImpl) Cancel() {
 }
 
-func (t table) Scan(prefixes []string) (query_db.KeyValueStream, error) {
+func (t table) Scan(keyRanges query_db.KeyRanges) (query_db.KeyValueStream, error) {
 	var keyValueStreamImpl keyValueStreamImpl
 	keyValueStreamImpl.table = t
 	keyValueStreamImpl.cursor = -1
-	keyValueStreamImpl.prefixes = prefixes
+	keyValueStreamImpl.keyRanges = keyRanges
 	return &keyValueStreamImpl, nil
 }
 
diff --git a/v23/syncbase/nosql/internal/query/demo/demo.go b/v23/syncbase/nosql/internal/query/demo/demo.go
index b19e0cc..e369dc4 100644
--- a/v23/syncbase/nosql/internal/query/demo/demo.go
+++ b/v23/syncbase/nosql/internal/query/demo/demo.go
@@ -24,7 +24,7 @@
 
 var (
 	format = flag.String("format", "table",
-		"Output format.  'table': human-readable table; 'csv': comma-separated values, use -csv-delimiter to control the delimiter.")
+		"Output format.  'table': human-readable table; 'csv': comma-separated values, use -csv-delimiter to control the delimiter; 'json': JSON arrays.")
 	csvDelimiter = flag.String("csv-delimiter", ",", "Delimiter to use when printing data as CSV (e.g. \"\t\", \",\")")
 )
 
@@ -63,10 +63,16 @@
 			if err := writer.WriteTable(os.Stdout, columnNames, rs); err != nil {
 				fmt.Fprintf(os.Stderr, "Error: %s\n", err.Error())
 			}
-		} else {
+		} else if *format == "csv" {
 			if err := writer.WriteCSV(os.Stdout, columnNames, rs, *csvDelimiter); err != nil {
 				fmt.Fprintf(os.Stderr, "Error: %s\n", err.Error())
 			}
+		} else if *format == "json" {
+			if err := writer.WriteJson(os.Stdout, columnNames, rs); err != nil {
+				fmt.Fprintf(os.Stderr, "Error: %s\n", err.Error())
+			}
+		} else {
+			panic(fmt.Sprintf("invalid format flag value: %v", *format))
 		}
 	}
 }
@@ -75,8 +81,8 @@
 	shutdown := db.InitDB()
 	defer shutdown()
 
-	if *format != "table" && *format != "csv" {
-		fmt.Fprintf(os.Stderr, "Unsupported -format %q.  Must be one of 'table' or 'csv'.\n", *format)
+	if *format != "table" && *format != "csv" && *format != "json" {
+		fmt.Fprintf(os.Stderr, "Unsupported -format %q.  Must be one of 'table', 'csv', or 'json'.\n", *format)
 		os.Exit(-1)
 	}
 
@@ -103,10 +109,11 @@
 				break
 			}
 		} else {
-			if q == "" || strings.ToLower(q) == "exit" || strings.ToLower(q) == "quit" {
+			tq := strings.ToLower(strings.TrimSpace(q))
+			if tq == "" || tq == "exit" || tq == "quit" {
 				break
 			}
-			if strings.ToLower(q) == "dump" {
+			if tq == "dump" {
 				dumpDB(d)
 			} else {
 				queryExec(d, q)
diff --git a/v23/syncbase/nosql/internal/query/demo/writer/writer.go b/v23/syncbase/nosql/internal/query/demo/writer/writer.go
index 3e2d997..20d61fe 100644
--- a/v23/syncbase/nosql/internal/query/demo/writer/writer.go
+++ b/v23/syncbase/nosql/internal/query/demo/writer/writer.go
@@ -5,6 +5,7 @@
 package writer
 
 import (
+	"encoding/json"
 	"errors"
 	"fmt"
 	"io"
@@ -138,6 +139,34 @@
 	return str
 }
 
+// WriteJson formats the result as a JSON array of arrays (rows) of values.
+func WriteJson(out io.Writer, columnNames []string, rs query.ResultStream) error {
+	io.WriteString(out, "[\n")
+	io.WriteString(out, "  [")
+	delim := ""
+	for _, cName := range columnNames {
+		str, err := json.Marshal(cName)
+		if err != nil {
+			panic(fmt.Sprintf("JSON marshalling failed for column name: %v", err))
+		}
+		io.WriteString(out, fmt.Sprintf("%s%s", delim, str))
+		delim = ", "
+	}
+	io.WriteString(out, "]\n")
+	for rs.Advance() {
+		io.WriteString(out, fmt.Sprintf(" ,["))
+		delim := ""
+		for _, column := range rs.Result() {
+			str := toJson(column)
+			io.WriteString(out, fmt.Sprintf("%s%s", delim, str))
+			delim = ", "
+		}
+		io.WriteString(out, "]\n")
+	}
+	io.WriteString(out, "]\n")
+	return rs.Err()
+}
+
 // Converts VDL value to readable yet parseable string representation.
 // If nested is not set, strings outside composites are left unquoted.
 // TODO(ivanpi): Handle cycles and improve non-tree DAG handling.
@@ -248,3 +277,129 @@
 	}
 	return begin + strings.Join(elems, sep) + end
 }
+
+// Converts VDL value to JSON representation.
+func toJson(val *vdl.Value) string {
+	jf := toJsonFriendly(val)
+	jOut, err := json.Marshal(jf)
+	if err != nil {
+		panic(fmt.Sprintf("JSON marshalling failed: %v", err))
+	}
+	return string(jOut)
+}
+
+// Converts VDL value to Go type compatible with json.Marshal().
+func toJsonFriendly(val *vdl.Value) interface{} {
+	switch val.Type() {
+	case vdl.TypeOf(vtime.Time{}), vdl.TypeOf(vtime.Duration{}):
+		s, err := toStringNative(val)
+		if err != nil {
+			panic(fmt.Sprintf("toStringNative failed for builtin time type: %v", err))
+		}
+		return s
+	default:
+		// fall through to Kind switch
+	}
+	switch val.Kind() {
+	case vdl.Bool:
+		return val.Bool()
+	case vdl.Byte:
+		return val.Byte()
+	case vdl.Uint16, vdl.Uint32, vdl.Uint64:
+		return val.Uint()
+	case vdl.Int16, vdl.Int32, vdl.Int64:
+		return val.Int()
+	case vdl.Float32, vdl.Float64:
+		return val.Float()
+	case vdl.Complex64, vdl.Complex128:
+		// Go doesn't support marshalling complex values, we need to stringify.
+		c := val.Complex()
+		return fmt.Sprintf("%v+%vi", real(c), imag(c))
+	case vdl.String:
+		return val.RawString()
+	case vdl.Enum:
+		return val.EnumLabel()
+	case vdl.Array, vdl.List:
+		arr := make([]interface{}, val.Len())
+		for i, _ := range arr {
+			arr[i] = toJsonFriendly(val.Index(i))
+		}
+		return arr
+	case vdl.Any, vdl.Optional:
+		if val.IsNil() {
+			return nil
+		}
+		return toJsonFriendly(val.Elem())
+	case vdl.Struct:
+		// TODO(ivanpi): Consider lowercasing field names.
+		return toOrderedMap(val.Type().NumField(), func(i int) (string, interface{}) {
+			return val.Type().Field(i).Name, toJsonFriendly(val.StructField(i))
+		})
+	case vdl.Union:
+		// TODO(ivanpi): Consider lowercasing field name.
+		ui, uv := val.UnionField()
+		return toOrderedMap(1, func(_ int) (string, interface{}) {
+			return val.Type().Field(ui).Name, toJsonFriendly(uv)
+		})
+	case vdl.Set:
+		// TODO(ivanpi): vdl.SortValuesAsString() used for predictable output ordering.
+		// Use a more sensible sort for numbers etc.
+		keys := vdl.SortValuesAsString(val.Keys())
+		return toOrderedMap(len(keys), func(i int) (string, interface{}) {
+			return toString(keys[i], false), true
+		})
+	case vdl.Map:
+		// TODO(ivanpi): vdl.SortValuesAsString() used for predictable output ordering.
+		// Use a more sensible sort for numbers etc.
+		keys := vdl.SortValuesAsString(val.Keys())
+		return toOrderedMap(len(keys), func(i int) (string, interface{}) {
+			return toString(keys[i], false), toJsonFriendly(val.MapIndex(keys[i]))
+		})
+	case vdl.TypeObject:
+		return val.String()
+	default:
+		panic(fmt.Sprintf("unknown Kind %s", val.Kind()))
+	}
+}
+
+// Serializes to JSON object, preserving key order.
+// Native Go map will serialize to JSON object with sorted keys, which is
+// unexpected behaviour for a struct.
+type orderedMap []orderedMapElem
+
+type orderedMapElem struct {
+	Key string
+	Val interface{}
+}
+
+var _ json.Marshaler = (*orderedMap)(nil)
+
+// Builds an orderedMap with n elements, obtaining the key and value of element
+// i using elemToKeyVal(i).
+func toOrderedMap(n int, elemToKeyVal func(i int) (string, interface{})) orderedMap {
+	om := make(orderedMap, n)
+	for i, _ := range om {
+		om[i].Key, om[i].Val = elemToKeyVal(i)
+	}
+	return om
+}
+
+// Serializes orderedMap to JSON object, preserving key order.
+func (om orderedMap) MarshalJSON() (_ []byte, rerr error) {
+	defer func() {
+		if r := recover(); r != nil {
+			rerr = fmt.Errorf("orderedMap: %v", r)
+		}
+	}()
+	return []byte(listToString("{", ",", "}", len(om), func(i int) string {
+		keyJson, err := json.Marshal(om[i].Key)
+		if err != nil {
+			panic(err)
+		}
+		valJson, err := json.Marshal(om[i].Val)
+		if err != nil {
+			panic(err)
+		}
+		return fmt.Sprintf("%s:%s", keyJson, valJson)
+	})), nil
+}
diff --git a/v23/syncbase/nosql/internal/query/demo/writer/writer_test.go b/v23/syncbase/nosql/internal/query/demo/writer/writer_test.go
index 3dd5b95..3c8c642 100644
--- a/v23/syncbase/nosql/internal/query/demo/writer/writer_test.go
+++ b/v23/syncbase/nosql/internal/query/demo/writer/writer_test.go
@@ -6,6 +6,7 @@
 
 import (
 	"bytes"
+	"encoding/json"
 	"testing"
 	"time"
 
@@ -348,3 +349,136 @@
 		}
 	}
 }
+
+func TestWriteJson(t *testing.T) {
+	type testCase struct {
+		columns []string
+		rows    [][]interface{}
+		// To make the test cases easier to read, output should have a leading
+		// newline.
+		output string
+	}
+	tests := []testCase{
+		{ // Basic.
+			[]string{"c\n1", "c鶊2"},
+			[][]interface{}{
+				{5, "foo\nbar"},
+				{6, "bar\tfoo"},
+			},
+			`
+[
+  ["c\n1", "c鶊2"]
+ ,[5, "foo\nbar"]
+ ,[6, "bar\tfoo"]
+]
+`,
+		},
+		{ // Numbers.
+			[]string{"byte", "uint16", "uint32", "uint64", "int16", "int32", "int64",
+				"float32", "float64", "complex64", "complex128"},
+			[][]interface{}{
+				{
+					byte(12), uint16(1234), uint32(5678), uint64(999888777666), int16(9876), int32(876543), int64(128),
+					float32(3.14159), float64(2.71828182846), complex64(123.0 + 7.0i), complex128(456.789 + 10.1112i),
+				},
+				{
+					byte(9), uint16(99), uint32(999), uint64(9999999), int16(9), int32(99), int64(88),
+					float32(1.41421356237), float64(1.73205080757), complex64(9.87 + 7.65i), complex128(4.32 + 1.0i),
+				},
+			},
+			`
+[
+  ["byte", "uint16", "uint32", "uint64", "int16", "int32", "int64", "float32", "float64", "complex64", "complex128"]
+ ,[12, 1234, 5678, 999888777666, 9876, 876543, 128, 3.141590118408203, 2.71828182846, "123+7i", "456.789+10.1112i"]
+ ,[9, 99, 999, 9999999, 9, 99, 88, 1.4142135381698608, 1.73205080757, "9.869999885559082+7.650000095367432i", "4.32+1i"]
+]
+`,
+		},
+		{ // Empty result.
+			[]string{},
+			[][]interface{}{},
+			`
+[
+  []
+]
+`,
+		},
+		{ // Empty values.
+			[]string{"blank", "empty", "nil"},
+			[][]interface{}{
+				{struct{}{}, []string{}, nil},
+				{},
+			},
+			`
+[
+  ["blank", "empty", "nil"]
+ ,[{}, [], null]
+ ,[]
+]
+`,
+		},
+		{
+			[]string{"c1"},
+			[][]interface{}{
+				{db.Customer{"John Smith", 1, true, db.AddressInfo{"1 Main St.", "Palo Alto", "CA", "94303"}, db.CreditReport{Agency: db.CreditAgencyEquifax, Report: db.AgencyReportEquifaxReport{db.EquifaxCreditReport{'A'}}}}},
+				{db.Invoice{1, 1000, 42, db.AddressInfo{"1 Main St.", "Palo Alto", "CA", "94303"}}},
+			},
+			`
+[
+  ["c1"]
+ ,[{"Name":"John Smith","Id":1,"Active":true,"Address":{"Street":"1 Main St.","City":"Palo Alto","State":"CA","Zip":"94303"},"Credit":{"Agency":"Equifax","Report":{"EquifaxReport":{"Rating":65}}}}]
+ ,[{"CustId":1,"InvoiceNum":1000,"Amount":42,"ShipTo":{"Street":"1 Main St.","City":"Palo Alto","State":"CA","Zip":"94303"}}]
+]
+`,
+		},
+		{
+			[]string{"nil", "composite", "typeobj"},
+			[][]interface{}{
+				{
+					nil,
+					db.Composite{db.Array2String{"foo", "bar"}, []int32{1, 2}, map[int32]struct{}{1: struct{}{}, 2: struct{}{}}, map[string]int32{"foo": 1, "bar": 2}},
+					vdl.TypeOf(map[string]struct{}{}),
+				},
+			},
+			`
+[
+  ["nil", "composite", "typeobj"]
+ ,[null, {"Arr":["foo","bar"],"ListInt":[1,2],"MySet":{"1":true,"2":true},"Map":{"bar":2,"foo":1}}, "typeobject(set[string])"]
+]
+`,
+		},
+		{
+			[]string{"c1"},
+			[][]interface{}{
+				{
+					db.Recursive{nil, &db.Times{time.Unix(123456789, 42244224), time.Duration(1337)}, map[db.Array2String]db.Recursive{
+						db.Array2String{"a", "棎鶊鵱"}: db.Recursive{},
+						db.Array2String{"x", "y"}:   db.Recursive{vdl.ValueOf(db.CreditReport{Agency: db.CreditAgencyExperian, Report: db.AgencyReportExperianReport{db.ExperianCreditReport{db.ExperianRatingGood}}}), nil, nil},
+					}},
+				},
+			},
+			`
+[
+  ["c1"]
+ ,[{"Any":null,"Maybe":{"Stamp":"1973-11-29 21:33:09.042244224 +0000 UTC","Interval":"1.337µs"},"Rec":{"[\"a\", \"棎鶊鵱\"]":{"Any":null,"Maybe":null,"Rec":{}},"[\"x\", \"y\"]":{"Any":{"Agency":"Experian","Report":{"ExperianReport":{"Rating":"Good"}}},"Maybe":null,"Rec":{}}}}]
+]
+`,
+		},
+	}
+	for _, test := range tests {
+		var b bytes.Buffer
+		if err := writer.WriteJson(&b, test.columns, newResultStream(test.rows)); err != nil {
+			t.Errorf("Unexpected error: %v", err)
+			continue
+		}
+		var decoded interface{}
+		if err := json.Unmarshal(b.Bytes(), &decoded); err != nil {
+			t.Errorf("Got invalid JSON: %v", err)
+		}
+		// Add a leading newline to the output to match the leading newline
+		// in our test cases.
+		if got, want := "\n"+b.String(), test.output; got != want {
+			t.Errorf("Wrong output:\nGOT: %q\nWANT:%q", got, want)
+		}
+	}
+}
diff --git a/v23/syncbase/nosql/internal/query/query.go b/v23/syncbase/nosql/internal/query/query.go
index da87502..bcbb3cf 100644
--- a/v23/syncbase/nosql/internal/query/query.go
+++ b/v23/syncbase/nosql/internal/query/query.go
@@ -65,18 +65,18 @@
 	return projection
 }
 
-// Given a query (i.e.,, select statement), return the key prefixes needed to satisfy the query.
-// A return of a single empty string ([]string{ "" }) means fetch all keys.
-func CompileKeyPrefixes(w *query_parser.WhereClause) []string {
+// Given a query (i.e.,, select statement), return the key ranges needed to satisfy the query.
+// A return of a single element array of string([]byte{0}), string([]byte{255}) means fetch all.
+func CompileKeyRanges(w *query_parser.WhereClause) query_db.KeyRanges {
 	// First determine if every key needs to be fetched.  To do this, evaluate the
 	// where clause substituting false for every key expression and true for every
 	// other (type for value) expression.  If the where clause evaluates to true,
 	// it is possible for a row to be selected without any dependence on the contents
 	// of the key.  In that case, all keys must be fetched.
 	if w == nil || CheckIfAllKeysMustBeFetched(w.Expr) {
-		return []string{""}
+		return query_db.KeyRanges{query_db.KeyRange{string([]byte{0}), string([]byte{255})}}
 	} else {
-		return query_checker.CompileKeyPrefixes(w)
+		return query_checker.CompileKeyRanges(w)
 	}
 }
 
@@ -176,8 +176,8 @@
 }
 
 func execSelect(db query_db.Database, s *query_parser.SelectStatement) ([]string, ResultStream, error) {
-	prefixes := CompileKeyPrefixes(s.Where)
-	keyValueStream, err := s.From.Table.DBTable.Scan(prefixes)
+	keyRanges := CompileKeyRanges(s.Where)
+	keyValueStream, err := s.From.Table.DBTable.Scan(keyRanges)
 	if err != nil {
 		return nil, nil, syncql.NewErrScanError(db.GetContext(), s.Off, err)
 	}
diff --git a/v23/syncbase/nosql/internal/query/query_checker/query_checker.go b/v23/syncbase/nosql/internal/query/query_checker/query_checker.go
index ce807c0..81bfea3 100644
--- a/v23/syncbase/nosql/internal/query/query_checker/query_checker.go
+++ b/v23/syncbase/nosql/internal/query/query_checker/query_checker.go
@@ -107,10 +107,18 @@
 		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, err := computeRegex(db, e.Operand2.Off, e.Operand2.Str)
+		regex, compRegex, foundWildcard, err := computeRegex(db, e.Operand2.Off, e.Operand2.Str)
 		if err != nil {
 			return err
 		}
+		// Optimization: If like argument contains no wildcards, convert the expression to equals.
+		if !foundWildcard {
+			e.Operator.Type = query_parser.Equal
+			// Since this is no longer a like expression, we need to unescape
+			// any escaped chars (i.e., "\\", "\_" and "\%" become
+			// "\", "_" and "%", respectively).
+			e.Operand2.Str = unescapeLikeExpression(e.Operand2.Str)
+		}
 		e.Operand2.Regex = regex
 		e.Operand2.CompRegex = compRegex
 	}
@@ -227,68 +235,85 @@
 // Convert Like expression to a regex.  That is, convert:
 // % to .*?
 // _ to .
+// Unescape '\\', '\%' and '\_' to '\', '%' and '_', respectively.
 // Escape everything that would be incorrectly interpreted as a regex.
-// Note: \% and \_ are used to escape % and _, respectively.
-func computeRegex(db query_db.Database, off int64, s string) (string, *regexp.Regexp, error) {
-	// Escape everything, this will escape too much as like wildcards can
-	// also be escaped by a backslash (\%, \_, \\).
-	escaped := regexp.QuoteMeta(s)
-	// Change all unescaped '%' chars to ".*?" and all unescaped '_' chars to '.'.
-	var buf bytes.Buffer
-	buf.WriteString("^")
-	backslash_level := 0
-	for _, c := range escaped {
-		switch backslash_level {
-		case 0:
-			switch c {
-			case '%':
-				buf.WriteString(".*?")
-			case '_':
-				buf.WriteString(".")
-			case '\\':
-				backslash_level++
-			default:
-				buf.WriteString(string(c))
+//
+// 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 query_db.Database, off int64, s string) (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
+
+	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
 			}
-		case 1:
-			switch c {
-			case '\\':
-				// backslashes become double backslashes because of the
-				// QuoteMeta above.  Let's see what's next.
-				backslash_level++
-			default:
-				// In this case, QuoteMeta is escaping a regex character.  We
-				// need to honor the escape (e.g., \*, \[, \]).
-				buf.WriteString("\\")
-				buf.WriteString(string(c))
-				backslash_level = 0
+			escapedMode = false
+		case '\\':
+			if escapedMode {
+				toBeEscapedBuf.WriteString(string(c))
 			}
-		case 2:
-			switch c {
-			case '\\':
-				// We've hit a third backslash.
-				// Write out the first \ (escaped as \\).
-				// Set backslash_level to 1 since we've encountered
-				// another backslash and need to see what follows.
-				buf.WriteString("\\\\")
-				backslash_level = 1
-			default:
-				// The user wrote \% or \_.
-				// It was escaped by QuoteMeta to \\% or \\_.
-				// Since the % or _ was escaped, just write it so regex can
-				// treat it like any character.
-				buf.WriteString(string(c))
-				backslash_level = 0
-			}
+			escapedMode = !escapedMode
+		default:
+			toBeEscapedBuf.WriteString(string(c))
 		}
 	}
-	buf.WriteString("$")
+	// Write any remaining chars in toBeEscapedBuf.
+	buf.WriteString(regexp.QuoteMeta(toBeEscapedBuf.String()))
+	buf.WriteString("$") // '^<regex_str>$'
 	regex := buf.String()
 	compRegex, err := regexp.Compile(regex)
 	if err != nil {
-		return "", nil, syncql.NewErrErrorCompilingRegularExpression(db.GetContext(), off, regex, err)
+		return "", nil, false, syncql.NewErrErrorCompilingRegularExpression(db.GetContext(), off, regex, err)
 	}
-	return regex, compRegex, nil
+	return regex, compRegex, foundWildcard, nil
+}
+
+// Unescape '\\', '\%' and '\_' to '\', '%' and '_', respectively.
+func unescapeLikeExpression(s string) string {
+	var buf bytes.Buffer // buffer for returned unescaped string
+
+	escapedMode := false
+
+	for _, c := range s {
+		switch c {
+		case '\\':
+			if escapedMode {
+				buf.WriteString(string(c))
+			}
+			escapedMode = !escapedMode
+		default:
+			buf.WriteString(string(c))
+			escapedMode = false
+		}
+	}
+	return buf.String()
 }
 
 func IsLogicalOperator(o *query_parser.BinaryOperator) bool {
@@ -323,53 +348,97 @@
 	return o.Type == query_parser.TypExpr
 }
 
-// Compile a list of key prefixes to fetch with scan.  Prefixes are returned in sorted order
-// and do not overlap (e.g., prefixes of "ab" and "abc" would be combined as "ab").
-// A single empty string (array len of 1) is returned if all keys are to be fetched.
-// Used by query package.  In query_checker package so prefixes can be tested.
-func CompileKeyPrefixes(where *query_parser.WhereClause) []string {
-	if where == nil {
-		return []string{""}
-	} else {
-		// Collect all key string literal operands.
-		p := collectKeyPrefixes(where.Expr)
-		// Sort
-		sort.Strings(p)
-		// Elminate overlaps
-		var p2 []string
-		for i, s := range p {
-			// Skip over prefixes that are already covered by prior key prefixes.
-			if i != 0 && strings.HasPrefix(s, p2[len(p2)-1]) {
-				continue
-			}
-			p2 = append(p2, s)
+// Function copied from syncbase.
+func computeKeyRangeForPrefix(prefix string) query_db.KeyRange {
+	if prefix == "" {
+		return query_db.KeyRange{string([]byte{0}), string([]byte{255})}
+	}
+	limit := []byte(prefix)
+	for len(limit) > 0 {
+		if limit[len(limit)-1] == 255 {
+			limit = limit[:len(limit)-1] // chop off trailing \x00
+		} else {
+			limit[len(limit)-1] += 1 // add 1
+			break                    // no carry
 		}
-		return p2
+	}
+	return query_db.KeyRange{prefix, string(limit)}
+}
+
+// The limit for a single value range is simply a zero byte appended.
+// In this way, only the single 'start' value will be returned (or nothing if that single
+// value is not present).
+func computeKeyRangeForSingleValue(start string) query_db.KeyRange {
+	limit := []byte(start)
+	limit = append(limit, 0)
+	return query_db.KeyRange{start, string(limit)}
+}
+
+// Compute a list of key ranges to be used by query_db's Table.Scan implementation.
+func CompileKeyRanges(where *query_parser.WhereClause) query_db.KeyRanges {
+	if where == nil {
+		return query_db.KeyRanges{computeKeyRangeForPrefix("")}
+	}
+	var keyRanges query_db.KeyRanges
+	collectKeyRanges(where.Expr, &keyRanges)
+	return keyRanges
+}
+
+func collectKeyRanges(expr *query_parser.Expression, keyRanges *query_db.KeyRanges) {
+	if IsKey(expr.Operand1) {
+		if expr.Operator.Type == query_parser.Like {
+			addKeyRange(computeKeyRangeForPrefix(expr.Operand2.Prefix), keyRanges)
+		} else { // OpEqual
+			addKeyRange(computeKeyRangeForSingleValue(expr.Operand2.Str), keyRanges)
+		}
+	}
+	if IsExpr(expr.Operand1) {
+		collectKeyRanges(expr.Operand1.Expr, keyRanges)
+	}
+	if IsExpr(expr.Operand2) {
+		collectKeyRanges(expr.Operand2.Expr, keyRanges)
 	}
 }
 
-// Collect all operand2 string literals where operand1 of the expression is ident "k".
-func collectKeyPrefixes(expr *query_parser.Expression) []string {
-	var prefixes []string
-	if IsKey(expr.Operand1) {
-		if expr.Operator.Type == query_parser.Like {
-			prefixes = append(prefixes, expr.Operand2.Prefix)
-		} else { // OpEqual
-			prefixes = append(prefixes, expr.Operand2.Str)
-		}
-		return prefixes
-	}
-	if IsExpr(expr.Operand1) {
-		for _, p := range collectKeyPrefixes(expr.Operand1.Expr) {
-			prefixes = append(prefixes, p)
+func addKeyRange(keyRange query_db.KeyRange, keyRanges *query_db.KeyRanges) {
+	handled := false
+	// Is there an overlap with an existing range?
+	for i, r := range *keyRanges {
+		// In the following if,
+		// the first paren expr is true if the start of the range to be added is contained in r
+		// the second paren expr is true if the limit of the range to be added is contained in r
+		// the third paren expr is true if the range to be added entirely contains r
+		if (keyRange.Start >= r.Start && keyRange.Start <= r.Limit) ||
+			(keyRange.Limit >= r.Start && keyRange.Limit <= r.Limit) ||
+			(keyRange.Start <= r.Start && keyRange.Limit >= r.Limit) {
+			// keyRange overlaps with existing range at keyRanges[i]
+			// set newKeyRange to a range that ecompasses both
+			var newKeyRange query_db.KeyRange
+			if keyRange.Start < r.Start {
+				newKeyRange.Start = keyRange.Start
+			} else {
+				newKeyRange.Start = r.Start
+			}
+			if keyRange.Limit > r.Limit {
+				newKeyRange.Limit = keyRange.Limit
+			} else {
+				newKeyRange.Limit = r.Limit
+			}
+			// The new range may overlap with other ranges in keyRanges
+			// delete the current range and call addKeyRange again
+			// This recursion will continue until no ranges overlap.
+			*keyRanges = append((*keyRanges)[:i], (*keyRanges)[i+1:]...)
+			addKeyRange(newKeyRange, keyRanges)
+			handled = true // we don't want to add keyRange below
+			break
 		}
 	}
-	if IsExpr(expr.Operand2) {
-		for _, p := range collectKeyPrefixes(expr.Operand2.Expr) {
-			prefixes = append(prefixes, p)
-		}
+	// no overlap, just add it
+	if !handled {
+		*keyRanges = append(*keyRanges, keyRange)
 	}
-	return prefixes
+	// sort before returning
+	sort.Sort(*keyRanges)
 }
 
 // Check limit clause.  Limit must be >= 1.
diff --git a/v23/syncbase/nosql/internal/query/query_checker/query_checker_test.go b/v23/syncbase/nosql/internal/query/query_checker/query_checker_test.go
index 16dd2d7..2ea1196 100644
--- a/v23/syncbase/nosql/internal/query/query_checker/query_checker_test.go
+++ b/v23/syncbase/nosql/internal/query/query_checker/query_checker_test.go
@@ -41,11 +41,11 @@
 	defer shutdown()
 }
 
-func (t invoiceTable) Scan(prefixes []string) (query_db.KeyValueStream, error) {
+func (t invoiceTable) Scan(keyRanges query_db.KeyRanges) (query_db.KeyValueStream, error) {
 	return nil, errors.New("unimplemented")
 }
 
-func (t customerTable) Scan(prefixes []string) (query_db.KeyValueStream, error) {
+func (t customerTable) Scan(keyRanges query_db.KeyRanges) (query_db.KeyValueStream, error) {
 	return nil, errors.New("unimplemented")
 }
 
@@ -66,9 +66,9 @@
 	query string
 }
 
-type keyPrefixesTest struct {
-	query       string
-	keyPrefixes []string
+type keyRangesTest struct {
+	query     string
+	keyRanges query_db.KeyRanges
 }
 
 type regularExpressionsTest struct {
@@ -121,35 +121,69 @@
 	}
 }
 
-func TestKeyPrefixes(t *testing.T) {
-	basic := []keyPrefixesTest{
+func appendZeroByte(start string) string {
+	limit := []byte(start)
+	limit = append(limit, 0)
+	return string(limit)
+
+}
+
+func TestKeyRanges(t *testing.T) {
+	basic := []keyRangesTest{
 		{
 			"select k, v from Customer",
-			[]string{""},
+			query_db.KeyRanges{
+				query_db.KeyRange{string([]byte{0}), string([]byte{255})},
+			},
 		},
 		{
 			"select k, v from Customer where t = \"Foo.Bar\" and k like \"abc%\" limit 100 offset 200",
-			[]string{"abc"},
+			query_db.KeyRanges{
+				query_db.KeyRange{"abc", "abd"},
+			},
 		},
 		{
 			"select  k,  v from \n  Customer where k like \"002%\" or k like \"001%\" or k like \"%\"",
-			[]string{""},
+			query_db.KeyRanges{
+				query_db.KeyRange{string([]byte{0}), string([]byte{255})},
+			},
 		},
 		{
 			"select k, v from Customer where k = \"Foo.Bar\" and k like \"abc%\" limit 100 offset 200",
-			[]string{"Foo.Bar", "abc"},
+			query_db.KeyRanges{
+				query_db.KeyRange{"Foo.Bar", appendZeroByte("Foo.Bar")},
+				query_db.KeyRange{"abc", "abd"},
+			},
 		},
 		{
+			// 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",
-			[]string{"Foo", "abc"},
+			query_db.KeyRanges{
+				query_db.KeyRange{"Foo", appendZeroByte("Foo")},
+				query_db.KeyRange{"Foo.Bar", appendZeroByte("Foo.Bar")},
+				query_db.KeyRange{"abc", "abd"},
+			},
 		},
 		{
 			"select k, v from Customer where k like \"Foo\\%Bar\" or k like \"abc%\" limit 100 offset 200",
-			[]string{"Foo%Bar", "abc"},
+			query_db.KeyRanges{
+				query_db.KeyRange{"Foo%Bar", appendZeroByte("Foo%Bar")},
+				query_db.KeyRange{"abc", "abd"},
+			},
 		},
 		{
 			"select k, v from Customer where k like \"Foo\\\\%Bar\" or k like \"abc%\" limit 100 offset 200",
-			[]string{"Foo\\", "abc"},
+			query_db.KeyRanges{
+				query_db.KeyRange{"Foo\\", "Foo]"},
+				query_db.KeyRange{"abc", "abd"},
+			},
+		},
+		{
+			"select k, v from Customer where k like \"Foo\\\\\\%Bar\" or k like \"abc%\" limit 100 offset 200",
+			query_db.KeyRanges{
+				query_db.KeyRange{"Foo\\%Bar", appendZeroByte("Foo\\%Bar")},
+				query_db.KeyRange{"abc", "abd"},
+			},
 		},
 	}
 
@@ -164,9 +198,9 @@
 		}
 		switch sel := (*s).(type) {
 		case query_parser.SelectStatement:
-			prefixes := query_checker.CompileKeyPrefixes(sel.Where)
-			if !reflect.DeepEqual(test.keyPrefixes, prefixes) {
-				t.Errorf("query: %s;\nGOT  %v\nWANT %v", test.query, prefixes, test.keyPrefixes)
+			keyRanges := query_checker.CompileKeyRanges(sel.Where)
+			if !reflect.DeepEqual(test.keyRanges, keyRanges) {
+				t.Errorf("query: %s;\nGOT  %v\nWANT %v", test.query, keyRanges, test.keyRanges)
 			}
 		default:
 			t.Errorf("query: %s;\nGOT  %v\nWANT query_parser.SelectStatement", test.query, *s)
diff --git a/v23/syncbase/nosql/internal/query/test/query_test.go b/v23/syncbase/nosql/internal/query/test/query_test.go
index 86e395e..00498a6 100644
--- a/v23/syncbase/nosql/internal/query/test/query_test.go
+++ b/v23/syncbase/nosql/internal/query/test/query_test.go
@@ -8,7 +8,6 @@
 	"errors"
 	"fmt"
 	"reflect"
-	"strings"
 	"testing"
 	"time"
 
@@ -36,10 +35,10 @@
 }
 
 type keyValueStreamImpl struct {
-	table        table
-	cursor       int
-	prefixes     []string
-	prefixCursor int
+	table           table
+	cursor          int
+	keyRanges       query_db.KeyRanges
+	keyRangesCursor int
 }
 
 func (kvs *keyValueStreamImpl) Advance() bool {
@@ -48,16 +47,16 @@
 		if kvs.cursor >= len(kvs.table.rows) {
 			return false
 		}
-		for kvs.prefixCursor < len(kvs.prefixes) {
-			// does it match any prefix
-			if kvs.prefixes[kvs.prefixCursor] == "" || strings.HasPrefix(kvs.table.rows[kvs.cursor].key, kvs.prefixes[kvs.prefixCursor]) {
+		for kvs.keyRangesCursor < len(kvs.keyRanges) {
+			// does it match any keyRange (or is the keyRange the 0-255 wildcard)?
+			if (kvs.keyRanges[kvs.keyRangesCursor].Start == string([]byte{0}) && kvs.keyRanges[kvs.keyRangesCursor].Limit == string([]byte{255})) || (kvs.table.rows[kvs.cursor].key >= kvs.keyRanges[kvs.keyRangesCursor].Start && kvs.table.rows[kvs.cursor].key <= kvs.keyRanges[kvs.keyRangesCursor].Limit) {
 				return true
 			}
-			// Keys and prefixes are both sorted low to high, so we can increment
-			// prefixCursor if the prefix is < the key.
-			if kvs.prefixes[kvs.prefixCursor] < kvs.table.rows[kvs.cursor].key {
-				kvs.prefixCursor++
-				if kvs.prefixCursor >= len(kvs.prefixes) {
+			// Keys and keyRanges are both sorted low to high, so we can increment
+			// keyRangesCursor if the keyRange.Limit is < the key.
+			if kvs.keyRanges[kvs.keyRangesCursor].Limit < kvs.table.rows[kvs.cursor].key {
+				kvs.keyRangesCursor++
+				if kvs.keyRangesCursor >= len(kvs.keyRanges) {
 					return false
 				}
 			} else {
@@ -79,11 +78,11 @@
 func (kvs *keyValueStreamImpl) Cancel() {
 }
 
-func (t table) Scan(prefixes []string) (query_db.KeyValueStream, error) {
+func (t table) Scan(keyRanges query_db.KeyRanges) (query_db.KeyValueStream, error) {
 	var keyValueStreamImpl keyValueStreamImpl
 	keyValueStreamImpl.table = t
 	keyValueStreamImpl.cursor = -1
-	keyValueStreamImpl.prefixes = prefixes
+	keyValueStreamImpl.keyRanges = keyRanges
 	return &keyValueStreamImpl, nil
 }
 
@@ -221,10 +220,10 @@
 	db.tables = append(db.tables, fooTable)
 }
 
-type keyPrefixesTest struct {
-	query       string
-	keyPrefixes []string
-	err         error
+type keyRangesTest struct {
+	query     string
+	keyRanges query_db.KeyRanges
+	err       error
 }
 
 type evalWhereUsingOnlyKeyTest struct {
@@ -983,120 +982,206 @@
 	}
 }
 
-func TestKeyPrefixes(t *testing.T) {
-	basic := []keyPrefixesTest{
+func appendZeroByte(start string) string {
+	limit := []byte(start)
+	limit = append(limit, 0)
+	return string(limit)
+
+}
+
+func plusOne(start string) string {
+	limit := []byte(start)
+	for len(limit) > 0 {
+		if limit[len(limit)-1] == 255 {
+			limit = limit[:len(limit)-1] // chop off trailing \x00
+		} else {
+			limit[len(limit)-1] += 1 // add 1
+			break                    // no carry
+		}
+	}
+	return string(limit)
+}
+
+func TestKeyRanges(t *testing.T) {
+	basic := []keyRangesTest{
 		{
-			// Need all keys (single prefix of "").
+			// Need all keys
 			"select k, v from Customer",
-			[]string{""},
+			query_db.KeyRanges{
+				query_db.KeyRange{string([]byte{0}), string([]byte{255})},
+			},
 			nil,
 		},
 		{
-			// Need all keys (single prefix of "").
-			"   sElEcT  k,  v from \n  Customer WhErE k lIkE \"002%\" oR k LiKe \"001%\" or k lIkE \"%\"",
-			[]string{""},
+			// Keys 001 and 003
+			"   select  k,  v from Customer where k = \"001\" or k = \"003\"",
+			query_db.KeyRanges{
+				query_db.KeyRange{"001", appendZeroByte("001")},
+				query_db.KeyRange{"003", appendZeroByte("003")},
+			},
+			nil,
+		},
+		{
+			// Need all keys
+			"select  k,  v from Customer where k like \"%\" or k like \"001%\" or k like \"002%\"",
+			query_db.KeyRanges{
+				query_db.KeyRange{string([]byte{0}), string([]byte{255})},
+			},
+			nil,
+		},
+		{
+			// Need all keys, likes in where clause in different order
+			"select  k,  v from Customer where k like \"002%\" or k like \"001%\" or k like \"%\"",
+			query_db.KeyRanges{
+				query_db.KeyRange{string([]byte{0}), string([]byte{255})},
+			},
 			nil,
 		},
 		{
 			// All selected rows will have key prefix of "abc".
 			"select k, v from Customer where t = \"Foo.Bar\" and k like \"abc%\"",
-			[]string{"abc"},
+			query_db.KeyRanges{
+				query_db.KeyRange{"abc", plusOne("abc")},
+			},
 			nil,
 		},
 		{
-			// Need all keys (single prefix of "").
+			// Need all keys
 			"select k, v from Customer where t = \"Foo.Bar\" or k like \"abc%\"",
-			[]string{""},
+			query_db.KeyRanges{
+				query_db.KeyRange{string([]byte{0}), string([]byte{255})},
+			},
 			nil,
 		},
 		{
-			// Need all keys (single prefix of "").
+			// Need all keys
 			"select k, v from Customer where k like \"abc%\" or v.zip = \"94303\"",
-			[]string{""},
+			query_db.KeyRanges{
+				query_db.KeyRange{string([]byte{0}), string([]byte{255})},
+			},
 			nil,
 		},
 		{
 			// All selected rows will have key prefix of "foo".
 			"select k, v from Customer where t = \"Foo.Bar\" and k like \"foo_bar\"",
-			[]string{"foo"},
+			query_db.KeyRanges{
+				query_db.KeyRange{"foo", plusOne("foo")},
+			},
 			nil,
 		},
 		{
-			// All selected rows will have key prefix of "baz" or "foo".
+			// All selected rows will have key == "baz" or prefix of "foo".
 			"select k, v from Customer where k like \"foo_bar\" or k = \"baz\"",
-			[]string{"baz", "foo"},
+			query_db.KeyRanges{
+				query_db.KeyRange{"baz", appendZeroByte("baz")},
+				query_db.KeyRange{"foo", plusOne("foo")},
+			},
 			nil,
 		},
 		{
-			// All selected rows will have key prefix of "fo".
+			// All selected rows will have key == "fo" or prefix of "foo".
 			"select k, v from Customer where k like \"foo_bar\" or k = \"fo\"",
-			[]string{"fo"},
+			query_db.KeyRanges{
+				query_db.KeyRange{"fo", appendZeroByte("fo")},
+				query_db.KeyRange{"foo", plusOne("foo")},
+			},
+			nil,
+		},
+		{
+			// All selected rows will have prefix of "fo".
+			// k == foo is a subset of above prefix
+			"select k, v from Customer where k like \"fo_bar\" or k = \"foo\"",
+			query_db.KeyRanges{
+				query_db.KeyRange{"fo", plusOne("fo")},
+			},
 			nil,
 		},
 		{
 			// All selected rows will have key prefix of "foo".
 			"select k, v from Customer where k like \"foo%bar\"",
-			[]string{"foo"},
+			query_db.KeyRanges{
+				query_db.KeyRange{"foo", plusOne("foo")},
+			},
 			nil,
 		},
 		{
-			// All selected rows will have key prefix of "foo\bar".
+			// Select "foo\bar" row.
 			"select k, v from Customer where k like \"foo\\\\bar\"",
-			[]string{"foo\\bar"},
+			query_db.KeyRanges{
+				query_db.KeyRange{"foo\\bar", appendZeroByte("foo\\bar")},
+			},
 			nil,
 		},
 		{
-			// All selected rows will have key prefix of "foo%bar".
+			// Select "foo%bar" row.
 			"select k, v from Customer where k like \"foo\\%bar\"",
-			[]string{"foo%bar"},
+			query_db.KeyRanges{
+				query_db.KeyRange{"foo%bar", appendZeroByte("foo%bar")},
+			},
 			nil,
 		},
 		{
-			// All selected rows will have key prefix of "foo\%bar".
+			// Select "foo\%bar" row.
 			"select k, v from Customer where k like \"foo\\\\\\%bar\"",
-			[]string{"foo\\%bar"},
+			query_db.KeyRanges{
+				query_db.KeyRange{"foo\\%bar", appendZeroByte("foo\\%bar")},
+			},
 			nil,
 		},
 		{
-			// Need all keys (single prefix of "").
+			// Need all keys
 			"select k, v from Customer where k like \"%foo\"",
-			[]string{""},
+			query_db.KeyRanges{
+				query_db.KeyRange{string([]byte{0}), string([]byte{255})},
+			},
 			nil,
 		},
 		{
-			// Need all keys (single prefix of "").
+			// Need all keys
 			"select k, v from Customer where k like \"_foo\"",
-			[]string{""},
+			query_db.KeyRanges{
+				query_db.KeyRange{string([]byte{0}), string([]byte{255})},
+			},
 			nil,
 		},
 		{
-			// All selected rows will have key prefix of "foo_bar".
+			// Select "foo_bar" row.
 			"select k, v from Customer where k like \"foo\\_bar\"",
-			[]string{"foo_bar"},
+			query_db.KeyRanges{
+				query_db.KeyRange{"foo_bar", appendZeroByte("foo_bar")},
+			},
 			nil,
 		},
 		{
-			// All selected rows will have key prefix of "foobar%".
+			// Select "foobar%" row.
 			"select k, v from Customer where k like \"foobar\\%\"",
-			[]string{"foobar%"},
+			query_db.KeyRanges{
+				query_db.KeyRange{"foobar%", appendZeroByte("foobar%")},
+			},
 			nil,
 		},
 		{
-			// All selected rows will have key prefix of "foobar_".
+			// Select "foobar_" row.
 			"select k, v from Customer where k like \"foobar\\_\"",
-			[]string{"foobar_"},
+			query_db.KeyRanges{
+				query_db.KeyRange{"foobar_", appendZeroByte("foobar_")},
+			},
 			nil,
 		},
 		{
-			// All selected rows will have key prefix of "\%_".
+			// Select "\%_" row.
 			"select k, v from Customer where k like \"\\\\\\%\\_\"",
-			[]string{"\\%_"},
+			query_db.KeyRanges{
+				query_db.KeyRange{"\\%_", appendZeroByte("\\%_")},
+			},
 			nil,
 		},
 		{
-			// All selected rows will have key prefix of "%_abc\".
+			// Select "%_abc\" row.
 			"select k, v from Customer where k = \"%_abc\\\"",
-			[]string{"%_abc\\"},
+			query_db.KeyRanges{
+				query_db.KeyRange{"%_abc\\", appendZeroByte("%_abc\\")},
+			},
 			nil,
 		},
 	}
@@ -1114,9 +1199,9 @@
 			if semErr == nil {
 				switch sel := (*s).(type) {
 				case query_parser.SelectStatement:
-					keyPrefixes := query.CompileKeyPrefixes(sel.Where)
-					if !reflect.DeepEqual(test.keyPrefixes, keyPrefixes) {
-						t.Errorf("query: %s;\nGOT  %v\nWANT %v", test.query, keyPrefixes, test.keyPrefixes)
+					keyRanges := query.CompileKeyRanges(sel.Where)
+					if !reflect.DeepEqual(test.keyRanges, keyRanges) {
+						t.Errorf("query: %s;\nGOT  %v\nWANT %v", test.query, keyRanges, test.keyRanges)
 					}
 				default:
 					t.Errorf("query: %s; got %v, want query_parser.SelectStatement", test.query, reflect.TypeOf(*s))
diff --git a/v23/syncbase/nosql/model.go b/v23/syncbase/nosql/model.go
index cf3124e..bd0903b 100644
--- a/v23/syncbase/nosql/model.go
+++ b/v23/syncbase/nosql/model.go
@@ -151,11 +151,8 @@
 	Put(ctx *context.T, key string, value interface{}) error
 
 	// Delete deletes all rows in the given half-open range [start, limit). If
-	// limit is "", all rows with keys >= start are included. If the last row that
-	// is covered by a prefix from SetPermissions is deleted, that (prefix, perms)
-	// pair is removed.
+	// limit is "", all rows with keys >= start are included.
 	// See helpers nosql.Prefix(), nosql.Range(), nosql.SingleRow().
-	// TODO(sadovsky): Automatic GC interacts poorly with sync. Revisit this API.
 	Delete(ctx *context.T, r RowRange) error
 
 	// Scan returns all rows in the given half-open range [start, limit). If limit
@@ -164,6 +161,13 @@
 	// See helpers nosql.Prefix(), nosql.Range(), nosql.SingleRow().
 	Scan(ctx *context.T, r RowRange) Stream
 
+	// 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) ([]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:
@@ -171,18 +175,8 @@
 	//     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 will fail if called with a prefix that does not match any
-	// rows.
 	SetPermissions(ctx *context.T, prefix PrefixRange, perms access.Permissions) 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) ([]PrefixPermissions, 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).
diff --git a/v23/syncbase/nosql/query_db/query_db.go b/v23/syncbase/nosql/query_db/query_db.go
index d6634dc..c1eba0c 100644
--- a/v23/syncbase/nosql/query_db/query_db.go
+++ b/v23/syncbase/nosql/query_db/query_db.go
@@ -19,7 +19,7 @@
 	// of the prefixes arguments.
 	// Note: an empty string prefix (""), matches all keys.
 	// The prefixes argument will be sorted (low to high).
-	Scan(prefixes []string) (KeyValueStream, error)
+	Scan(keyRanges KeyRanges) (KeyValueStream, error)
 }
 
 type KeyValueStream interface {
@@ -50,3 +50,28 @@
 	// return  false.  Cancel does not block.
 	Cancel()
 }
+
+type KeyRange struct {
+	Start string
+	Limit string
+}
+
+type KeyRanges []KeyRange
+
+// Implement sort interface for KeyRanges.
+func (keyRanges KeyRanges) Len() int {
+	return len(keyRanges)
+}
+
+func (keyRanges KeyRanges) Less(i, j int) bool {
+	return keyRanges[i].Start < keyRanges[j].Start
+}
+
+func (keyRanges KeyRanges) Swap(i, j int) {
+	saveStart := keyRanges[i].Start
+	saveLimit := keyRanges[i].Limit
+	keyRanges[i].Start = keyRanges[j].Start
+	keyRanges[i].Limit = keyRanges[j].Limit
+	keyRanges[j].Start = saveStart
+	keyRanges[j].Limit = saveLimit
+}
diff --git a/v23/syncbase/nosql/table.go b/v23/syncbase/nosql/table.go
index c77905c..67aa074 100644
--- a/v23/syncbase/nosql/table.go
+++ b/v23/syncbase/nosql/table.go
@@ -70,11 +70,6 @@
 	return newStream(cancel, call)
 }
 
-// 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)
-}
-
 // GetPermissions implements Table.GetPermissions.
 func (t *table) GetPermissions(ctx *context.T, key string) ([]PrefixPermissions, error) {
 	wirePermsList, err := t.c.GetPermissions(ctx, key)
@@ -85,6 +80,11 @@
 	return permsList, err
 }
 
+// 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)
+}
+
 // DeletePermissions implements Table.DeletePermissions.
 func (t *table) DeletePermissions(ctx *context.T, prefix PrefixRange) error {
 	return t.c.DeletePermissions(ctx, prefix.Prefix())
diff --git a/v23/syncbase/testutil/layer.go b/v23/syncbase/testutil/layer.go
index 14e8497..ccfe62d 100644
--- a/v23/syncbase/testutil/layer.go
+++ b/v23/syncbase/testutil/layer.go
@@ -34,7 +34,7 @@
 	if err := self.Create(ctx, nil); err != nil {
 		t.Fatalf("self.Create() failed: %v", err)
 	}
-	if gotPerms, wantPerms := getPermsOrDie(t, ctx, self), defaultPerms(); !reflect.DeepEqual(gotPerms, wantPerms) {
+	if gotPerms, wantPerms := getPermsOrDie(t, ctx, self), DefaultPerms("server/client"); !reflect.DeepEqual(gotPerms, wantPerms) {
 		t.Errorf("Perms do not match: got %v, want %v", gotPerms, wantPerms)
 	}
 
@@ -60,7 +60,7 @@
 	}
 
 	// Test that create fails if the parent perms disallow access.
-	perms = defaultPerms()
+	perms = DefaultPerms("server/client")
 	perms.Blacklist("server/client", string(access.Write))
 	if err := parent.SetPermissions(ctx, perms, ""); err != nil {
 		t.Fatalf("parent.SetPermissions() failed: %v", err)
@@ -107,7 +107,7 @@
 	if err := self2.Create(ctx, nil); err != nil {
 		t.Fatalf("self2.Create() failed: %v", err)
 	}
-	perms := defaultPerms()
+	perms := DefaultPerms("server/client")
 	perms.Blacklist("server/client", string(access.Write))
 	if err := self2.SetPermissions(ctx, perms, ""); err != nil {
 		t.Fatalf("self2.SetPermissions() failed: %v", err)
@@ -117,7 +117,7 @@
 	}
 
 	// Test that delete succeeds even if the parent perms disallow access.
-	perms = defaultPerms()
+	perms = DefaultPerms("server/client")
 	perms.Blacklist("server/client", string(access.Write))
 	if err := parent.SetPermissions(ctx, perms, ""); err != nil {
 		t.Fatalf("parent.SetPermissions() failed: %v", err)
diff --git a/v23/syncbase/testutil/util.go b/v23/syncbase/testutil/util.go
index 0783e54..c5df00f 100644
--- a/v23/syncbase/testutil/util.go
+++ b/v23/syncbase/testutil/util.go
@@ -6,6 +6,7 @@
 package testutil
 
 import (
+	"fmt"
 	"io/ioutil"
 	"os"
 	"reflect"
@@ -18,13 +19,13 @@
 	"v.io/syncbase/x/ref/services/syncbase/server"
 	"v.io/v23"
 	"v.io/v23/context"
-	"v.io/v23/naming"
-	"v.io/v23/rpc"
 	"v.io/v23/security"
 	"v.io/v23/security/access"
 	"v.io/v23/vdl"
 	"v.io/v23/verror"
 	"v.io/x/lib/vlog"
+	"v.io/x/ref/lib/flags"
+	"v.io/x/ref/lib/xrpc"
 	tsecurity "v.io/x/ref/test/testutil"
 )
 
@@ -67,6 +68,12 @@
 }
 
 func SetupOrDieCustom(client, server string, perms access.Permissions) (clientCtx *context.T, serverName string, cleanup func(), sp security.Principal, ctx *context.T) {
+	// TODO(mattr): Instead of SetDefaultHostPort the arguably more correct thing
+	// would be to call v.io/x/ref/test.Init() from the test packages that import
+	// the profile.  Note you should only call that from the package that imports
+	// the profile, not from libraries like this.  Also, it would be better if
+	// v23.Init was test.V23Init().
+	flags.SetDefaultHostPort("127.0.0.1:0")
 	ctx, shutdown := v23.Init()
 	sp = tsecurity.NewPrincipal(server)
 
@@ -77,7 +84,7 @@
 		vlog.Fatal("v23.WithPrincipal() failed: ", err)
 	}
 
-	serverName, stopServer := newServer(serverCtx, perms)
+	serverName, stopServer := newServer(client, server, serverCtx, perms)
 	cleanup = func() {
 		stopServer()
 		shutdown()
@@ -85,10 +92,12 @@
 	return
 }
 
-func defaultPerms() access.Permissions {
+func DefaultPerms(patterns ...string) access.Permissions {
 	perms := access.Permissions{}
 	for _, tag := range access.AllTypicalTags() {
-		perms.Add(security.BlessingPattern("server/client"), string(tag))
+		for _, pattern := range patterns {
+			perms.Add(security.BlessingPattern(pattern), string(tag))
+		}
 	}
 	return perms
 }
@@ -175,18 +184,9 @@
 	return perms
 }
 
-func newServer(ctx *context.T, perms access.Permissions) (string, func()) {
-	s, err := v23.NewServer(ctx)
-	if err != nil {
-		vlog.Fatal("v23.NewServer() failed: ", err)
-	}
-	eps, err := s.Listen(rpc.ListenSpec{Addrs: rpc.ListenAddrs{{"tcp", "127.0.0.1:0"}}})
-	if err != nil {
-		vlog.Fatal("s.Listen() failed: ", err)
-	}
-
+func newServer(clientName, serverName string, ctx *context.T, perms access.Permissions) (string, func()) {
 	if perms == nil {
-		perms = defaultPerms()
+		perms = DefaultPerms(fmt.Sprintf("%s/%s", serverName, clientName))
 	}
 	rootDir, err := ioutil.TempDir("", "syncbase")
 	if err != nil {
@@ -200,13 +200,11 @@
 	if err != nil {
 		vlog.Fatal("server.NewService() failed: ", err)
 	}
-	d := server.NewDispatcher(service)
-
-	if err := s.ServeDispatcher("", d); err != nil {
-		vlog.Fatal("s.ServeDispatcher() failed: ", err)
+	s, err := xrpc.NewDispatchingServer(ctx, "", server.NewDispatcher(service))
+	if err != nil {
+		vlog.Fatal("xrpc.NewDispatchingServer() failed: ", err)
 	}
-
-	name := naming.JoinAddressName(eps[0].String(), "")
+	name := s.Status().Endpoints[0].Name()
 	return name, func() {
 		s.Stop()
 		os.RemoveAll(rootDir)
diff --git a/x/ref/services/syncbase/localblobstore/fs_cablobstore/fs_cablobstore.go b/x/ref/services/syncbase/localblobstore/fs_cablobstore/fs_cablobstore.go
index 8e173d1..43cc542 100644
--- a/x/ref/services/syncbase/localblobstore/fs_cablobstore/fs_cablobstore.go
+++ b/x/ref/services/syncbase/localblobstore/fs_cablobstore/fs_cablobstore.go
@@ -41,6 +41,7 @@
 import "sync"
 import "time"
 
+import "v.io/syncbase/x/ref/services/syncbase/localblobstore"
 import "v.io/v23/context"
 import "v.io/v23/verror"
 
@@ -247,22 +248,13 @@
 
 // -----------------------------------------------------------
 
-// A BlockOrFile represents a vector of bytes, and contains either a data block
-// (as a []byte), or a (file name, size, offset) triple.
-type BlockOrFile struct {
-	Block    []byte // If FileName is empty, the bytes represented.
-	FileName string // If non-empty, the name of the file containing the bytes.
-	Size     int64  // If FileName is non-empty, the number of bytes (or -1 for "all")
-	Offset   int64  // If FileName is non-empty, the offset of the relevant bytes within the file.
-}
-
 // addFragment() ensures that the store *fscabs contains a fragment comprising
 // the catenation of the byte vectors named by item[..].block and the contents
 // of the files named by item[..].filename.  The block field is ignored if
 // fileName!="".  The fragment is not physically added if already present.
 // The fragment is added to the fragment list of the descriptor *desc.
 func (fscabs *FsCaBlobStore) addFragment(ctx *context.T, extHasher hash.Hash,
-	desc *blobDesc, item ...BlockOrFile) (fileName string, size int64, err error) {
+	desc *blobDesc, item ...localblobstore.BlockOrFile) (fileName string, size int64, err error) {
 
 	hasher := md5.New()
 	var buf []byte
@@ -543,12 +535,12 @@
 // should not be used concurrently by multiple threads.  The returned handle
 // should be closed with either the Close() or CloseWithoutFinalize() method to
 // avoid leaking file handles.
-func (fscabs *FsCaBlobStore) NewBlobWriter(ctx *context.T) (bw *BlobWriter, err error) {
+func (fscabs *FsCaBlobStore) NewBlobWriter(ctx *context.T) (localblobstore.BlobWriter, error) {
+	var bw *BlobWriter
 	newName := newBlobName()
-	var f *file
 	fileName := filepath.Join(fscabs.rootName, newName)
 	os.MkdirAll(filepath.Dir(fileName), dirPermissions)
-	f, err = newFile(os.Create(fileName))
+	f, err := newFile(os.Create(fileName))
 	if err == nil {
 		bw = new(BlobWriter)
 		bw.fscabs = fscabs
@@ -568,8 +560,9 @@
 
 // ResumeBlobWriter() returns a pointer to a newly allocated BlobWriter on an
 // old, but unfinalized blob name.
-func (fscabs *FsCaBlobStore) ResumeBlobWriter(ctx *context.T, blobName string) (bw *BlobWriter, err error) {
-	bw = new(BlobWriter)
+func (fscabs *FsCaBlobStore) ResumeBlobWriter(ctx *context.T, blobName string) (localblobstore.BlobWriter, error) {
+	var err error
+	bw := new(BlobWriter)
 	bw.fscabs = fscabs
 	bw.ctx = ctx
 	bw.desc, err = bw.fscabs.getBlob(ctx, blobName)
@@ -639,7 +632,7 @@
 // AppendFragment() appends a fragment to the blob being written by *bw, where
 // the fragment is composed of the byte vectors described by the elements of
 // item[].  The fragment is copied into the blob store.
-func (bw *BlobWriter) AppendFragment(item ...BlockOrFile) (err error) {
+func (bw *BlobWriter) AppendFragment(item ...localblobstore.BlockOrFile) (err error) {
 	if bw.f == nil {
 		panic("fs_cablobstore.BlobWriter programming error: AppendFragment() after Close()")
 	}
@@ -782,7 +775,7 @@
 // NewBlobReader() returns a pointer to a newly allocated BlobReader on the
 // specified blobName.  BlobReaders should not be used concurrently by multiple
 // threads.  Returned handles should be closed with Close().
-func (fscabs *FsCaBlobStore) NewBlobReader(ctx *context.T, blobName string) (br *BlobReader, err error) {
+func (fscabs *FsCaBlobStore) NewBlobReader(ctx *context.T, blobName string) (br localblobstore.BlobReader, err error) {
 	var desc *blobDesc
 	desc, err = fscabs.getBlob(ctx, blobName)
 	if err == nil {
@@ -954,7 +947,7 @@
 //    if fscabsi.Err() != nil {
 //      // The loop terminated early due to an error.
 //    }
-func (fscabs *FsCaBlobStore) ListBlobIds(ctx *context.T) (fscabsi *FsCasIter) {
+func (fscabs *FsCaBlobStore) ListBlobIds(ctx *context.T) localblobstore.Iter {
 	stack := make([]dirListing, 1)
 	stack[0] = dirListing{pos: -1, nameList: []string{blobDir}}
 	return &FsCasIter{fscabs: fscabs, stack: stack}
@@ -970,7 +963,7 @@
 //    if fscabsi.Err() != nil {
 //      // The loop terminated early due to an error.
 //    }
-func (fscabs *FsCaBlobStore) ListCAIds(ctx *context.T) (fscabsi *FsCasIter) {
+func (fscabs *FsCaBlobStore) ListCAIds(ctx *context.T) localblobstore.Iter {
 	stack := make([]dirListing, 1)
 	stack[0] = dirListing{pos: -1, nameList: []string{casDir}}
 	return &FsCasIter{fscabs: fscabs, stack: stack}
diff --git a/x/ref/services/syncbase/localblobstore/fs_cablobstore/fs_cablobstore_test.go b/x/ref/services/syncbase/localblobstore/fs_cablobstore/fs_cablobstore_test.go
index ded5e3c..886f5c4 100644
--- a/x/ref/services/syncbase/localblobstore/fs_cablobstore/fs_cablobstore_test.go
+++ b/x/ref/services/syncbase/localblobstore/fs_cablobstore/fs_cablobstore_test.go
@@ -5,345 +5,16 @@
 // A test for fs_cablobstore
 package fs_cablobstore_test
 
-import "bytes"
-import "crypto/md5"
-import "fmt"
-import "io"
 import "io/ioutil"
 import "os"
-import "path/filepath"
 import "testing"
+
+import "v.io/syncbase/x/ref/services/syncbase/localblobstore"
 import "v.io/syncbase/x/ref/services/syncbase/localblobstore/fs_cablobstore"
-import "v.io/v23/context"
-import "v.io/v23/verror"
+import "v.io/syncbase/x/ref/services/syncbase/localblobstore/localblobstore_testlib"
 import "v.io/x/ref/test"
 import _ "v.io/x/ref/runtime/factories/generic"
 
-func TestCreate(t *testing.T) {
-	ctx, shutdown := test.V23Init()
-	defer shutdown()
-
-	// Make a temporary directory.
-	var err error
-	var testDirName string
-	testDirName, err = ioutil.TempDir("", "fs_cablobstore_test")
-	if err != nil {
-		t.Fatalf("fs_cablobstore_test: can't make tmp directory: %v\n", err)
-	}
-	defer os.RemoveAll(testDirName)
-
-	// Check that we can create an fs_cablobstore.
-	var fscabs *fs_cablobstore.FsCaBlobStore
-	fscabs, err = fs_cablobstore.Create(ctx, testDirName)
-	if err != nil {
-		t.Errorf("fs_cablobstore.Create failed: %v", err)
-	}
-
-	// Check that there are no files in the newly-created tree.
-	iterator := fscabs.ListBlobIds(ctx)
-	for iterator.Advance() {
-		fileName := iterator.Value()
-		t.Errorf("unexpected file %q\n", fileName)
-	}
-	if iterator.Err() != nil {
-		t.Errorf("fs_cablobstore.ListBlobIds iteration failed: %v", iterator.Err())
-	}
-}
-
-// A blobOrBlockOrFile represents some bytes that may be contained in a named
-// blob, a named file, or in an explicit slice of bytes.
-type blobOrBlockOrFile struct {
-	blob   string // If non-emtpy, the name of the blob containing the bytes.
-	file   string // If non-empty and blob is empty, the name of the file containing the bytes.
-	size   int64  // Size of part of file or blob, or -1 for "everything until EOF".
-	offset int64  // Offset within file or blob.
-	block  []byte // If both blob and file are empty, a slice containing the bytes.
-}
-
-// A testBlob records that some specified content has been stored with a given
-// blob name in the blob store.
-type testBlob struct {
-	content  []byte // content that has been stored.
-	blobName string // the name of the blob.
-}
-
-// removeBlobFromBlobVector() removes the entry named blobName from
-// blobVector[], returning the new vector.
-func removeBlobFromBlobVector(blobVector []testBlob, blobName string) []testBlob {
-	n := len(blobVector)
-	i := 0
-	for i = 0; i != n && blobName != blobVector[i].blobName; i++ {
-	}
-	if i != n {
-		blobVector[i] = blobVector[n-1]
-		blobVector = blobVector[0 : n-1]
-	}
-	return blobVector
-}
-
-// writeBlob() writes a new blob to *fscabs, and returns its name.  The new
-// blob's content is described by the elements of data[].  Any error messages
-// generated include the index of the blob in blobVector and its content; the
-// latter is assumed to be printable.  The expected content of the the blob is
-// "content", so that this routine can check it.  If useResume is true, and data[]
-// has length more than 1, the function artificially uses ResumeBlobWriter(),
-// to test it.
-func writeBlob(t *testing.T, ctx *context.T, fscabs *fs_cablobstore.FsCaBlobStore, blobVector []testBlob,
-	content []byte, useResume bool, data ...blobOrBlockOrFile) []testBlob {
-	var bw *fs_cablobstore.BlobWriter
-	var err error
-	bw, err = fscabs.NewBlobWriter(ctx)
-	if err != nil {
-		t.Fatalf("fs_cablobstore.NewBlobWriter blob %d:%s failed: %v", len(blobVector), string(content), err)
-	}
-	blobName := bw.Name()
-
-	// Construct the blob from the pieces.
-	// There is a loop within the loop to exercise the possibility of
-	// passing multiple fragments to AppendFragment().
-	for i := 0; i != len(data) && err == nil; {
-		if len(data[i].blob) != 0 {
-			err = bw.AppendBlob(data[i].blob, data[i].size, data[i].offset)
-			if err != nil {
-				t.Errorf("fs_cablobstore.AppendBlob %d:%s blob %s failed: %v", len(blobVector), string(content), data[i].blob, err)
-			}
-			i++
-		} else {
-			var pieces []fs_cablobstore.BlockOrFile
-			for ; i != len(data) && len(data[i].blob) == 0; i++ {
-				if len(data[i].file) != 0 {
-					pieces = append(pieces, fs_cablobstore.BlockOrFile{
-						FileName: data[i].file,
-						Size:     data[i].size,
-						Offset:   data[i].offset})
-				} else {
-					pieces = append(pieces, fs_cablobstore.BlockOrFile{Block: data[i].block})
-				}
-			}
-			err = bw.AppendFragment(pieces...)
-			if err != nil {
-				t.Errorf("fs_cablobstore.AppendFragment %d:%s failed on %v: %v", len(blobVector), string(content), pieces, err)
-			}
-		}
-		if useResume && i < len(data)-1 && err == nil {
-			err = bw.CloseWithoutFinalize()
-			if err == nil {
-				bw, err = fscabs.ResumeBlobWriter(ctx, blobName)
-			}
-		}
-	}
-
-	if bw != nil {
-		if bw.Size() != int64(len(content)) {
-			t.Errorf("fs_cablobstore.Size before finalization %d:%s got %d, expected %d", len(blobVector), string(content), bw.Size(), len(content))
-		}
-		if bw.IsFinalized() {
-			t.Errorf("fs_cablobstore.IsFinalized %d:%s got true, expected false", len(blobVector), string(content))
-		}
-		err = bw.Close()
-		if err != nil {
-			t.Errorf("fs_cablobstore.Close %d:%s failed: %v", len(blobVector), string(content), err)
-		}
-		if !bw.IsFinalized() {
-			t.Errorf("fs_cablobstore.IsFinalized %d:%s got true, expected false", len(blobVector), string(content))
-		}
-		if bw.Size() != int64(len(content)) {
-			t.Errorf("fs_cablobstore.Size %d:%s after finalization got %d, expected %d", len(blobVector), string(content), bw.Size(), len(content))
-		}
-		if bw.Name() != blobName {
-			t.Errorf("fs_cablobstore %d:%s name changed when finalized was %s now %s", len(blobVector), string(content), blobName, bw.Name())
-		}
-		hasher := md5.New()
-		hasher.Write(content)
-		if bytes.Compare(bw.Hash(), hasher.Sum(nil)) != 0 {
-			t.Errorf("fs_cablobstore %d:%s BlobWriter.Hash got %v, expected %v", len(blobVector), string(content), bw.Hash(), hasher.Sum(nil))
-		}
-	}
-
-	return append(blobVector,
-		testBlob{
-			content:  content,
-			blobName: blobName,
-		})
-}
-
-// readBlob() returns a substring of the content of the blob named blobName in *fscabs.
-// The return values are:
-// - the "size" bytes from the content, starting at the given "offset",
-//   measured from "whence" (as defined by io.Seeker.Seek).
-// - the position to which BlobBeader seeks to,
-// - the md5 hash of the bytes read, and
-// - the md5 hash of the bytes of the blob, as returned by BlobReader.Hash(),
-// - and error.
-func readBlob(ctx *context.T, fscabs *fs_cablobstore.FsCaBlobStore, blobName string,
-	size int64, offset int64, whence int) (content []byte, pos int64, hash []byte, fullHash []byte, err error) {
-
-	var br *fs_cablobstore.BlobReader
-	hasher := md5.New()
-	br, err = fscabs.NewBlobReader(ctx, blobName)
-	if err == nil {
-		buf := make([]byte, 8192, 8192)
-		fullHash = br.Hash()
-		pos, err = br.Seek(offset, whence)
-		if err == nil {
-			var n int
-			first := true // Read at least once, to test reading zero bytes.
-			for err == nil && (size == -1 || int64(len(content)) < size || first) {
-				// Read just what was asked for.
-				var toRead []byte = buf
-				if size >= 0 && int(size)-len(content) < len(buf) {
-					toRead = buf[0 : int(size)-len(content)]
-				}
-				n, err = br.Read(toRead)
-				hasher.Write(toRead[0:n])
-				if size >= 0 && int64(len(content)+n) > size {
-					n = int(size) - len(content)
-				}
-				content = append(content, toRead[0:n]...)
-				first = false
-			}
-		}
-		br.Close()
-	}
-	return content, pos, hasher.Sum(nil), fullHash, err
-}
-
-// checkWrittenBlobsAreReadable() checks that the blobs in blobVector[] can be
-// read, and that they contain the appropriate data.
-func checkWrittenBlobsAreReadable(t *testing.T, ctx *context.T, fscabs *fs_cablobstore.FsCaBlobStore, blobVector []testBlob) {
-	for i := range blobVector {
-		var size int64
-		data := blobVector[i].content
-		dataLen := int64(len(data))
-		blobName := blobVector[i].blobName
-		for size = -1; size != dataLen+1; size++ {
-			var offset int64
-			for offset = -dataLen - 1; offset != dataLen+1; offset++ {
-				for whence := -1; whence != 4; whence++ {
-					content, pos, hash, fullHash, err := readBlob(ctx, fscabs, blobName, size, offset, whence)
-
-					// Compute expected seek position.
-					expectedPos := offset
-					if whence == 2 {
-						expectedPos += dataLen
-					}
-
-					// Computed expected size.
-					expectedSize := size
-					if expectedSize == -1 || expectedPos+expectedSize > dataLen {
-						expectedSize = dataLen - expectedPos
-					}
-
-					// Check that reads behave as expected.
-					if (whence == -1 || whence == 3) &&
-						verror.ErrorID(err) == "v.io/syncbase/x/ref/services/syncbase/localblobstore/fs_cablobstore.errBadSeekWhence" {
-						// Expected error from bad "whence" value.
-					} else if expectedPos < 0 &&
-						verror.ErrorID(err) == "v.io/syncbase/x/ref/services/syncbase/localblobstore/fs_cablobstore.errNegativeSeekPosition" {
-						// Expected error from negative Seek position.
-					} else if expectedPos > dataLen &&
-						verror.ErrorID(err) == "v.io/syncbase/x/ref/services/syncbase/localblobstore/fs_cablobstore.errIllegalPositionForRead" {
-						// Expected error from too high a Seek position.
-					} else if 0 <= expectedPos && expectedPos+expectedSize <= int64(len(data)) &&
-						bytes.Compare(data[expectedPos:expectedPos+expectedSize], content) == 0 && err == io.EOF &&
-						pos == expectedPos && expectedPos+expectedSize == dataLen {
-						// Expected success with EOF.
-					} else if 0 <= expectedPos && expectedPos+expectedSize <= int64(len(data)) &&
-						bytes.Compare(data[expectedPos:expectedPos+expectedSize], content) == 0 && err == nil &&
-						pos == expectedPos && expectedPos+expectedSize != dataLen {
-						if pos == 0 && size == -1 && bytes.Compare(hash, fullHash) != 0 {
-							t.Errorf("fs_cablobstore read test on %q size %d offset %d whence %d; got hash %v, expected %v  (blob is %q)",
-								string(data), size, offset, whence,
-								hash, fullHash, blobName)
-						} // Else expected success without EOF.
-					} else {
-						t.Errorf("fs_cablobstore read test on %q size %d offset %d whence %d yields %q pos %d %v   (blob is %q)",
-							string(data), size, offset, whence,
-							content, pos, err, blobName)
-					}
-				}
-			}
-		}
-	}
-}
-
-// checkAllBlobs() checks all the blobs in *fscabs to ensure they correspond to
-// those in blobVector[].
-func checkAllBlobs(t *testing.T, ctx *context.T, fscabs *fs_cablobstore.FsCaBlobStore, blobVector []testBlob, testDirName string) {
-	blobCount := 0
-	iterator := fscabs.ListBlobIds(ctx)
-	for iterator.Advance() {
-		fileName := iterator.Value()
-		i := 0
-		for ; i != len(blobVector) && fileName != blobVector[i].blobName; i++ {
-		}
-		if i == len(blobVector) {
-			t.Errorf("fs_cablobstore.ListBlobIds found unexpected file %s", fileName)
-		} else {
-			content, pos, hash, fullHash, err := readBlob(ctx, fscabs, fileName, -1, 0, 0)
-			if err != nil && err != io.EOF {
-				t.Errorf("fs_cablobstore.ListCAIds can't read %q: %v", filepath.Join(testDirName, fileName), err)
-			} else if bytes.Compare(blobVector[i].content, content) != 0 {
-				t.Errorf("fs_cablobstore.ListCAIds found unexpected blob content: %q, contains %q, expected %q",
-					filepath.Join(testDirName, fileName), content, string(blobVector[i].content))
-			} else if pos != 0 {
-				t.Errorf("fs_cablobstore.ListCAIds Seek on %q returned %d instead of 0",
-					filepath.Join(testDirName, fileName), pos)
-			}
-			if bytes.Compare(hash, fullHash) != 0 {
-				t.Errorf("fs_cablobstore.ListCAIds read on %q; got hash %v, expected %v",
-					fileName, hash, fullHash)
-			}
-		}
-		blobCount++
-	}
-	if iterator.Err() != nil {
-		t.Errorf("fs_cablobstore.ListBlobIds iteration failed: %v", iterator.Err())
-	}
-	if blobCount != len(blobVector) {
-		t.Errorf("fs_cablobstore.ListBlobIds iteration expected 4 files, got %d", blobCount)
-	}
-}
-
-// checkFragments() checks all the fragments in *fscabs to ensure they
-// correspond to those fragmentMap[].
-func checkFragments(t *testing.T, ctx *context.T, fscabs *fs_cablobstore.FsCaBlobStore, fragmentMap map[string]bool, testDirName string) {
-	caCount := 0
-	iterator := fscabs.ListCAIds(ctx)
-	for iterator.Advance() {
-		fileName := iterator.Value()
-		content, err := ioutil.ReadFile(filepath.Join(testDirName, fileName))
-		if err != nil && err != io.EOF {
-			t.Errorf("fs_cablobstore.ListCAIds can't read %q: %v", filepath.Join(testDirName, fileName), err)
-		} else if !fragmentMap[string(content)] {
-			t.Errorf("fs_cablobstore.ListCAIds found unexpected fragment entry: %q, contains %q", filepath.Join(testDirName, fileName), content)
-		} else {
-			hasher := md5.New()
-			hasher.Write(content)
-			hash := hasher.Sum(nil)
-			nameFromContent := filepath.Join("cas",
-				fmt.Sprintf("%02x", hash[0]),
-				fmt.Sprintf("%02x", hash[1]),
-				fmt.Sprintf("%02x", hash[2]),
-				fmt.Sprintf("%02x%02x%02x%02x%02x%02x%02x%02x%02x%02x%02x%02x%02x",
-					hash[3],
-					hash[4], hash[5], hash[6], hash[7],
-					hash[8], hash[9], hash[10], hash[11],
-					hash[12], hash[13], hash[14], hash[15]))
-			if nameFromContent != fileName {
-				t.Errorf("fs_cablobstore.ListCAIds hash of fragment: got %q, expected %q (content=%s)", nameFromContent, fileName, string(content))
-			}
-		}
-		caCount++
-	}
-	if iterator.Err() != nil {
-		t.Errorf("fs_cablobstore.ListCAIds iteration failed: %v", iterator.Err())
-	}
-	if caCount != len(fragmentMap) {
-		t.Errorf("fs_cablobstore.ListCAIds iteration expected %d files, got %d", len(fragmentMap), caCount)
-	}
-}
-
 // This test case tests adding files, retrieving them and deleting them.  One
 // can't retrieve or delete something that hasn't been created, so it's all one
 // test case.
@@ -354,231 +25,19 @@
 	// Make a temporary directory.
 	var err error
 	var testDirName string
-	testDirName, err = ioutil.TempDir("", "fs_cablobstore_test")
+	testDirName, err = ioutil.TempDir("", "localblobstore_test")
 	if err != nil {
-		t.Fatalf("fs_cablobstore_test: can't make tmp directory: %v\n", err)
+		t.Fatalf("localblobstore_test: can't make tmp directory: %v\n", err)
 	}
 	defer os.RemoveAll(testDirName)
 
 	// Create an fs_cablobstore.
-	var fscabs *fs_cablobstore.FsCaBlobStore
-	fscabs, err = fs_cablobstore.Create(ctx, testDirName)
+	var bs localblobstore.BlobStore
+	bs, err = fs_cablobstore.Create(ctx, testDirName)
 	if err != nil {
 		t.Fatalf("fs_cablobstore.Create failed: %v", err)
 	}
 
-	// Create the strings:  "wom", "bat", "wombat", "batwom", "atwo", "atwoatwoombatatwo".
-	womData := []byte("wom")
-	batData := []byte("bat")
-	wombatData := []byte("wombat")
-	batwomData := []byte("batwom")
-	atwoData := []byte("atwo")
-	atwoatwoombatatwoData := []byte("atwoatwoombatatwo")
-
-	// fragmentMap will have an entry per content-addressed fragment.
-	fragmentMap := make(map[string]bool)
-
-	// Create the blobs, by various means.
-
-	var blobVector []testBlob // Accumulate the blobs we create here.
-
-	blobVector = writeBlob(t, ctx, fscabs, blobVector,
-		womData, false,
-		blobOrBlockOrFile{block: womData})
-	womName := blobVector[len(blobVector)-1].blobName
-	fragmentMap[string(womData)] = true
-
-	blobVector = writeBlob(t, ctx, fscabs, blobVector,
-		batData, false,
-		blobOrBlockOrFile{block: batData})
-	batName := blobVector[len(blobVector)-1].blobName
-	fragmentMap[string(batData)] = true
-
-	blobVector = writeBlob(t, ctx, fscabs, blobVector,
-		wombatData, false,
-		blobOrBlockOrFile{block: wombatData})
-	firstWombatName := blobVector[len(blobVector)-1].blobName
-	fragmentMap[string(wombatData)] = true
-
-	blobVector = writeBlob(t, ctx, fscabs, blobVector,
-		wombatData, true,
-		blobOrBlockOrFile{block: womData},
-		blobOrBlockOrFile{block: batData})
-
-	blobVector = writeBlob(t, ctx, fscabs, blobVector,
-		wombatData, false,
-		blobOrBlockOrFile{
-			blob:   firstWombatName,
-			size:   -1,
-			offset: 0})
-
-	blobVector = writeBlob(t, ctx, fscabs, blobVector,
-		wombatData, false,
-		blobOrBlockOrFile{
-			blob:   firstWombatName,
-			size:   6,
-			offset: 0})
-
-	blobVector = writeBlob(t, ctx, fscabs, blobVector,
-		batwomData, false,
-		blobOrBlockOrFile{
-			blob:   firstWombatName,
-			size:   3,
-			offset: 3},
-		blobOrBlockOrFile{
-			blob:   firstWombatName,
-			size:   3,
-			offset: 0})
-	batwomName := blobVector[len(blobVector)-1].blobName
-
-	blobVector = writeBlob(t, ctx, fscabs, blobVector,
-		atwoData, false,
-		blobOrBlockOrFile{
-			blob:   batwomName,
-			size:   4,
-			offset: 1})
-	atwoName := blobVector[len(blobVector)-1].blobName
-
-	blobVector = writeBlob(t, ctx, fscabs, blobVector,
-		atwoatwoombatatwoData, true,
-		blobOrBlockOrFile{
-			blob:   atwoName,
-			size:   -1,
-			offset: 0},
-		blobOrBlockOrFile{
-			blob:   atwoName,
-			size:   4,
-			offset: 0},
-		blobOrBlockOrFile{
-			blob:   firstWombatName,
-			size:   -1,
-			offset: 1},
-		blobOrBlockOrFile{
-			blob:   batName,
-			size:   -1,
-			offset: 1},
-		blobOrBlockOrFile{
-			blob:   womName,
-			size:   2,
-			offset: 0})
-	atwoatwoombatatwoName := blobVector[len(blobVector)-1].blobName
-
-	// -------------------------------------------------
-	// Check that the state is as we expect.
-	checkWrittenBlobsAreReadable(t, ctx, fscabs, blobVector)
-	checkAllBlobs(t, ctx, fscabs, blobVector, testDirName)
-	checkFragments(t, ctx, fscabs, fragmentMap, testDirName)
-
-	// -------------------------------------------------
-	// Nothing should change if we garbage collect.
-	fscabs.GC(ctx)
-	checkWrittenBlobsAreReadable(t, ctx, fscabs, blobVector)
-	checkAllBlobs(t, ctx, fscabs, blobVector, testDirName)
-	checkFragments(t, ctx, fscabs, fragmentMap, testDirName)
-
-	// -------------------------------------------------
-	// Ensure that deleting non-existent blobs fails.
-	err = fscabs.DeleteBlob(ctx, "../../../../etc/passwd")
-	if verror.ErrorID(err) != "v.io/syncbase/x/ref/services/syncbase/localblobstore/fs_cablobstore.errInvalidBlobName" {
-		t.Errorf("DeleteBlob attempted to delete a bogus blob name")
-	}
-	err = fscabs.DeleteBlob(ctx, "foo/00/00/00/00000000000000000000000000")
-	if verror.ErrorID(err) != "v.io/syncbase/x/ref/services/syncbase/localblobstore/fs_cablobstore.errInvalidBlobName" {
-		t.Errorf("DeleteBlob attempted to delete a bogus blob name")
-	}
-
-	// -------------------------------------------------
-	// Delete a blob.
-	err = fscabs.DeleteBlob(ctx, batName)
-	if err != nil {
-		t.Errorf("DeleteBlob failed to delete blob %q: %v", batName, err)
-	}
-	blobVector = removeBlobFromBlobVector(blobVector, batName)
-
-	// -------------------------------------------------
-	// Check that the state is as we expect.
-	checkWrittenBlobsAreReadable(t, ctx, fscabs, blobVector)
-	checkAllBlobs(t, ctx, fscabs, blobVector, testDirName)
-	checkFragments(t, ctx, fscabs, fragmentMap, testDirName)
-
-	// -------------------------------------------------
-	// Nothing should change if we garbage collect.
-	fscabs.GC(ctx)
-	checkWrittenBlobsAreReadable(t, ctx, fscabs, blobVector)
-	checkAllBlobs(t, ctx, fscabs, blobVector, testDirName)
-	checkFragments(t, ctx, fscabs, fragmentMap, testDirName)
-
-	// -------------------------------------------------
-	// Open a BlobReader on a blob we're about to delete,
-	// so its fragments won't be garbage collected.
-
-	var br *fs_cablobstore.BlobReader
-	br, err = fscabs.NewBlobReader(ctx, atwoatwoombatatwoName)
-	if err != nil {
-		t.Errorf("NewBlobReader failed in blob %q: %v", atwoatwoombatatwoName, err)
-	}
-
-	// -------------------------------------------------
-	// Delete a blob.  This should be the last on-disc reference to the
-	// content-addressed fragment "bat", but the fragment won't be deleted
-	// until close the reader and garbage collect.
-	err = fscabs.DeleteBlob(ctx, atwoatwoombatatwoName)
-	if err != nil {
-		t.Errorf("DeleteBlob failed to delete blob %q: %v", atwoatwoombatatwoName, err)
-	}
-	blobVector = removeBlobFromBlobVector(blobVector, atwoatwoombatatwoName)
-
-	// -------------------------------------------------
-	// Check that the state is as we expect.
-	checkWrittenBlobsAreReadable(t, ctx, fscabs, blobVector)
-	checkAllBlobs(t, ctx, fscabs, blobVector, testDirName)
-	checkFragments(t, ctx, fscabs, fragmentMap, testDirName)
-
-	// -------------------------------------------------
-	// Garbage collection should change nothing; the fragment involved
-	// is still referenced from the open reader *br.
-	fscabs.GC(ctx)
-	checkWrittenBlobsAreReadable(t, ctx, fscabs, blobVector)
-	checkAllBlobs(t, ctx, fscabs, blobVector, testDirName)
-	checkFragments(t, ctx, fscabs, fragmentMap, testDirName)
-
-	// -------------------------------------------------
-
-	// Close the open BlobReader and garbage collect.
-	err = br.Close()
-	if err != nil {
-		t.Errorf("BlobReader.Close failed on blob %q: %v", atwoatwoombatatwoName, err)
-	}
-	delete(fragmentMap, string(batData))
-
-	fscabs.GC(ctx)
-	checkWrittenBlobsAreReadable(t, ctx, fscabs, blobVector)
-	checkAllBlobs(t, ctx, fscabs, blobVector, testDirName)
-	checkFragments(t, ctx, fscabs, fragmentMap, testDirName)
-
-	// -------------------------------------------------
-	// Delete all blobs.
-	for len(blobVector) != 0 {
-		err = fscabs.DeleteBlob(ctx, blobVector[0].blobName)
-		if err != nil {
-			t.Errorf("DeleteBlob failed to delete blob %q: %v", blobVector[0].blobName, err)
-		}
-		blobVector = removeBlobFromBlobVector(blobVector, blobVector[0].blobName)
-	}
-
-	// -------------------------------------------------
-	// Check that the state is as we expect.
-	checkWrittenBlobsAreReadable(t, ctx, fscabs, blobVector)
-	checkAllBlobs(t, ctx, fscabs, blobVector, testDirName)
-	checkFragments(t, ctx, fscabs, fragmentMap, testDirName)
-
-	// -------------------------------------------------
-	// The remaining fragments should be removed when we garbage collect.
-	for frag := range fragmentMap {
-		delete(fragmentMap, frag)
-	}
-	fscabs.GC(ctx)
-	checkWrittenBlobsAreReadable(t, ctx, fscabs, blobVector)
-	checkAllBlobs(t, ctx, fscabs, blobVector, testDirName)
-	checkFragments(t, ctx, fscabs, fragmentMap, testDirName)
+	// Test it.
+	localblobstore_testlib.AddRetrieveAndDelete(t, ctx, bs, testDirName)
 }
diff --git a/x/ref/services/syncbase/localblobstore/localblobstore_test.go b/x/ref/services/syncbase/localblobstore/localblobstore_test.go
new file mode 100644
index 0000000..973a9c8
--- /dev/null
+++ b/x/ref/services/syncbase/localblobstore/localblobstore_test.go
@@ -0,0 +1,43 @@
+// Copyright 2015 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.
+
+// A test for localblobstore
+package localblobstore_test
+
+import "io/ioutil"
+import "os"
+import "testing"
+
+import "v.io/syncbase/x/ref/services/syncbase/localblobstore"
+import "v.io/syncbase/x/ref/services/syncbase/localblobstore/fs_cablobstore"
+import "v.io/syncbase/x/ref/services/syncbase/localblobstore/localblobstore_testlib"
+import "v.io/x/ref/test"
+import _ "v.io/x/ref/runtime/factories/generic"
+
+// This test case tests adding files, retrieving them and deleting them.  One
+// can't retrieve or delete something that hasn't been created, so it's all one
+// test case.
+func TestAddRetrieveAndDelete(t *testing.T) {
+	ctx, shutdown := test.V23Init()
+	defer shutdown()
+
+	// Make a temporary directory.
+	var err error
+	var testDirName string
+	testDirName, err = ioutil.TempDir("", "localblobstore_test")
+	if err != nil {
+		t.Fatalf("localblobstore_test: can't make tmp directory: %v\n", err)
+	}
+	defer os.RemoveAll(testDirName)
+
+	// Create an fs_cablobstore.
+	var bs localblobstore.BlobStore
+	bs, err = fs_cablobstore.Create(ctx, testDirName)
+	if err != nil {
+		t.Fatalf("fs_cablobstore.Create failed: %v", err)
+	}
+
+	// Test it.
+	localblobstore_testlib.AddRetrieveAndDelete(t, ctx, bs, testDirName)
+}
diff --git a/x/ref/services/syncbase/localblobstore/localblobstore_testlib/localblobstore_testlib.go b/x/ref/services/syncbase/localblobstore/localblobstore_testlib/localblobstore_testlib.go
new file mode 100644
index 0000000..662e964
--- /dev/null
+++ b/x/ref/services/syncbase/localblobstore/localblobstore_testlib/localblobstore_testlib.go
@@ -0,0 +1,548 @@
+// Copyright 2015 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.
+
+// A test library for localblobstores.
+package localblobstore_testlib
+
+import "bytes"
+import "crypto/md5"
+import "fmt"
+import "io"
+import "io/ioutil"
+import "path/filepath"
+import "testing"
+
+import "v.io/syncbase/x/ref/services/syncbase/localblobstore"
+import "v.io/v23/context"
+import "v.io/v23/verror"
+
+// A blobOrBlockOrFile represents some bytes that may be contained in a named
+// blob, a named file, or in an explicit slice of bytes.
+type blobOrBlockOrFile struct {
+	blob   string // If non-empty, the name of the blob containing the bytes.
+	file   string // If non-empty and blob is empty, the name of the file containing the bytes.
+	size   int64  // Size of part of file or blob, or -1 for "everything until EOF".
+	offset int64  // Offset within file or blob.
+	block  []byte // If both blob and file are empty, a slice containing the bytes.
+}
+
+// A testBlob records that some specified content has been stored with a given
+// blob name in the blob store.
+type testBlob struct {
+	content  []byte // content that has been stored.
+	blobName string // the name of the blob.
+}
+
+// removeBlobFromBlobVector() removes the entry named blobName from
+// blobVector[], returning the new vector.
+func removeBlobFromBlobVector(blobVector []testBlob, blobName string) []testBlob {
+	n := len(blobVector)
+	i := 0
+	for i = 0; i != n && blobName != blobVector[i].blobName; i++ {
+	}
+	if i != n {
+		blobVector[i] = blobVector[n-1]
+		blobVector = blobVector[0 : n-1]
+	}
+	return blobVector
+}
+
+// writeBlob() writes a new blob to bs, and returns its name.  The new
+// blob's content is described by the elements of data[].  Any error messages
+// generated include the index of the blob in blobVector and its content; the
+// latter is assumed to be printable.  The expected content of the the blob is
+// "content", so that this routine can check it.  If useResume is true, and data[]
+// has length more than 1, the function artificially uses ResumeBlobWriter(),
+// to test it.
+func writeBlob(t *testing.T, ctx *context.T, bs localblobstore.BlobStore, blobVector []testBlob,
+	content []byte, useResume bool, data ...blobOrBlockOrFile) []testBlob {
+	var bw localblobstore.BlobWriter
+	var err error
+	bw, err = bs.NewBlobWriter(ctx)
+	if err != nil {
+		t.Errorf("localblobstore.NewBlobWriter blob %d:%s failed: %v", len(blobVector), string(content), err)
+	}
+	blobName := bw.Name()
+
+	// Construct the blob from the pieces.
+	// There is a loop within the loop to exercise the possibility of
+	// passing multiple fragments to AppendFragment().
+	for i := 0; i != len(data) && err == nil; {
+		if len(data[i].blob) != 0 {
+			err = bw.AppendBlob(data[i].blob, data[i].size, data[i].offset)
+			if err != nil {
+				t.Errorf("localblobstore.AppendBlob %d:%s blob %s failed: %v", len(blobVector), string(content), data[i].blob, err)
+			}
+			i++
+		} else {
+			var pieces []localblobstore.BlockOrFile
+			for ; i != len(data) && len(data[i].blob) == 0; i++ {
+				if len(data[i].file) != 0 {
+					pieces = append(pieces, localblobstore.BlockOrFile{
+						FileName: data[i].file,
+						Size:     data[i].size,
+						Offset:   data[i].offset})
+				} else {
+					pieces = append(pieces, localblobstore.BlockOrFile{Block: data[i].block})
+				}
+			}
+			err = bw.AppendFragment(pieces...)
+			if err != nil {
+				t.Errorf("localblobstore.AppendFragment %d:%s failed on %v: %v", len(blobVector), string(content), pieces, err)
+			}
+		}
+		if useResume && i < len(data)-1 && err == nil {
+			err = bw.CloseWithoutFinalize()
+			if err == nil {
+				bw, err = bs.ResumeBlobWriter(ctx, blobName)
+			}
+		}
+	}
+
+	if bw != nil {
+		if bw.Size() != int64(len(content)) {
+			t.Errorf("localblobstore.Size before finalization %d:%s got %d, expected %d", len(blobVector), string(content), bw.Size(), len(content))
+		}
+		if bw.IsFinalized() {
+			t.Errorf("localblobstore.IsFinalized %d:%s got true, expected false", len(blobVector), string(content))
+		}
+		err = bw.Close()
+		if err != nil {
+			t.Errorf("localblobstore.Close %d:%s failed: %v", len(blobVector), string(content), err)
+		}
+		if !bw.IsFinalized() {
+			t.Errorf("localblobstore.IsFinalized %d:%s got true, expected false", len(blobVector), string(content))
+		}
+		if bw.Size() != int64(len(content)) {
+			t.Errorf("localblobstore.Size %d:%s after finalization got %d, expected %d", len(blobVector), string(content), bw.Size(), len(content))
+		}
+		if bw.Name() != blobName {
+			t.Errorf("localblobstore %d:%s name changed when finalized was %s now %s", len(blobVector), string(content), blobName, bw.Name())
+		}
+		hasher := md5.New()
+		hasher.Write(content)
+		if bytes.Compare(bw.Hash(), hasher.Sum(nil)) != 0 {
+			t.Errorf("localblobstore %d:%s BlobWriter.Hash got %v, expected %v", len(blobVector), string(content), bw.Hash(), hasher.Sum(nil))
+		}
+	}
+
+	return append(blobVector,
+		testBlob{
+			content:  content,
+			blobName: blobName,
+		})
+}
+
+// readBlob() returns a substring of the content of the blob named blobName in bs.
+// The return values are:
+// - the "size" bytes from the content, starting at the given "offset",
+//   measured from "whence" (as defined by io.Seeker.Seek).
+// - the position to which BlobBeader seeks to,
+// - the md5 hash of the bytes read, and
+// - the md5 hash of the bytes of the blob, as returned by BlobReader.Hash(),
+// - and error.
+func readBlob(ctx *context.T, bs localblobstore.BlobStore, blobName string,
+	size int64, offset int64, whence int) (content []byte, pos int64, hash []byte, fullHash []byte, err error) {
+
+	var br localblobstore.BlobReader
+	hasher := md5.New()
+	br, err = bs.NewBlobReader(ctx, blobName)
+	if err == nil {
+		buf := make([]byte, 8192, 8192)
+		fullHash = br.Hash()
+		pos, err = br.Seek(offset, whence)
+		if err == nil {
+			var n int
+			first := true // Read at least once, to test reading zero bytes.
+			for err == nil && (size == -1 || int64(len(content)) < size || first) {
+				// Read just what was asked for.
+				var toRead []byte = buf
+				if size >= 0 && int(size)-len(content) < len(buf) {
+					toRead = buf[0 : int(size)-len(content)]
+				}
+				n, err = br.Read(toRead)
+				hasher.Write(toRead[0:n])
+				if size >= 0 && int64(len(content)+n) > size {
+					n = int(size) - len(content)
+				}
+				content = append(content, toRead[0:n]...)
+				first = false
+			}
+		}
+		br.Close()
+	}
+	return content, pos, hasher.Sum(nil), fullHash, err
+}
+
+// checkWrittenBlobsAreReadable() checks that the blobs in blobVector[] can be
+// read, and that they contain the appropriate data.
+func checkWrittenBlobsAreReadable(t *testing.T, ctx *context.T, bs localblobstore.BlobStore, blobVector []testBlob) {
+	for i := range blobVector {
+		var size int64
+		data := blobVector[i].content
+		dataLen := int64(len(data))
+		blobName := blobVector[i].blobName
+		for size = -1; size != dataLen+1; size++ {
+			var offset int64
+			for offset = -dataLen - 1; offset != dataLen+1; offset++ {
+				for whence := -1; whence != 4; whence++ {
+					content, pos, hash, fullHash, err := readBlob(ctx, bs, blobName, size, offset, whence)
+
+					// Compute expected seek position.
+					expectedPos := offset
+					if whence == 2 {
+						expectedPos += dataLen
+					}
+
+					// Computed expected size.
+					expectedSize := size
+					if expectedSize == -1 || expectedPos+expectedSize > dataLen {
+						expectedSize = dataLen - expectedPos
+					}
+
+					// Check that reads behave as expected.
+					if (whence == -1 || whence == 3) &&
+						verror.ErrorID(err) == "v.io/syncbase/x/ref/services/syncbase/localblobstore/fs_cablobstore.errBadSeekWhence" {
+						// Expected error from bad "whence" value.
+					} else if expectedPos < 0 &&
+						verror.ErrorID(err) == "v.io/syncbase/x/ref/services/syncbase/localblobstore/fs_cablobstore.errNegativeSeekPosition" {
+						// Expected error from negative Seek position.
+					} else if expectedPos > dataLen &&
+						verror.ErrorID(err) == "v.io/syncbase/x/ref/services/syncbase/localblobstore/fs_cablobstore.errIllegalPositionForRead" {
+						// Expected error from too high a Seek position.
+					} else if 0 <= expectedPos && expectedPos+expectedSize <= int64(len(data)) &&
+						bytes.Compare(data[expectedPos:expectedPos+expectedSize], content) == 0 && err == io.EOF &&
+						pos == expectedPos && expectedPos+expectedSize == dataLen {
+						// Expected success with EOF.
+					} else if 0 <= expectedPos && expectedPos+expectedSize <= int64(len(data)) &&
+						bytes.Compare(data[expectedPos:expectedPos+expectedSize], content) == 0 && err == nil &&
+						pos == expectedPos && expectedPos+expectedSize != dataLen {
+						if pos == 0 && size == -1 && bytes.Compare(hash, fullHash) != 0 {
+							t.Errorf("localblobstore read test on %q size %d offset %d whence %d; got hash %v, expected %v  (blob is %q)",
+								string(data), size, offset, whence,
+								hash, fullHash, blobName)
+						} // Else expected success without EOF.
+					} else {
+						t.Errorf("localblobstore read test on %q size %d offset %d whence %d yields %q pos %d %v   (blob is %q)",
+							string(data), size, offset, whence,
+							content, pos, err, blobName)
+					}
+				}
+			}
+		}
+	}
+}
+
+// checkAllBlobs() checks all the blobs in bs to ensure they correspond to
+// those in blobVector[].
+func checkAllBlobs(t *testing.T, ctx *context.T, bs localblobstore.BlobStore, blobVector []testBlob, testDirName string) {
+	blobCount := 0
+	iterator := bs.ListBlobIds(ctx)
+	for iterator.Advance() {
+		fileName := iterator.Value()
+		i := 0
+		for ; i != len(blobVector) && fileName != blobVector[i].blobName; i++ {
+		}
+		if i == len(blobVector) {
+			t.Errorf("localblobstore.ListBlobIds found unexpected file %s", fileName)
+		} else {
+			content, pos, hash, fullHash, err := readBlob(ctx, bs, fileName, -1, 0, 0)
+			if err != nil && err != io.EOF {
+				t.Errorf("localblobstore.ListCAIds can't read %q: %v", filepath.Join(testDirName, fileName), err)
+			} else if bytes.Compare(blobVector[i].content, content) != 0 {
+				t.Errorf("localblobstore.ListCAIds found unexpected blob content: %q, contains %q, expected %q",
+					filepath.Join(testDirName, fileName), content, string(blobVector[i].content))
+			} else if pos != 0 {
+				t.Errorf("localblobstore.ListCAIds Seek on %q returned %d instead of 0",
+					filepath.Join(testDirName, fileName), pos)
+			}
+			if bytes.Compare(hash, fullHash) != 0 {
+				t.Errorf("localblobstore.ListCAIds read on %q; got hash %v, expected %v",
+					fileName, hash, fullHash)
+			}
+		}
+		blobCount++
+	}
+	if iterator.Err() != nil {
+		t.Errorf("localblobstore.ListBlobIds iteration failed: %v", iterator.Err())
+	}
+	if blobCount != len(blobVector) {
+		t.Errorf("localblobstore.ListBlobIds iteration expected 4 files, got %d", blobCount)
+	}
+}
+
+// checkFragments() checks all the fragments in bs to ensure they
+// correspond to those fragmentMap[], iff testDirName is non-empty.
+func checkFragments(t *testing.T, ctx *context.T, bs localblobstore.BlobStore, fragmentMap map[string]bool, testDirName string) {
+	if testDirName != "" {
+		caCount := 0
+		iterator := bs.ListCAIds(ctx)
+		for iterator.Advance() {
+			fileName := iterator.Value()
+			content, err := ioutil.ReadFile(filepath.Join(testDirName, fileName))
+			if err != nil && err != io.EOF {
+				t.Errorf("localblobstore.ListCAIds can't read %q: %v", filepath.Join(testDirName, fileName), err)
+			} else if !fragmentMap[string(content)] {
+				t.Errorf("localblobstore.ListCAIds found unexpected fragment entry: %q, contains %q", filepath.Join(testDirName, fileName), content)
+			} else {
+				hasher := md5.New()
+				hasher.Write(content)
+				hash := hasher.Sum(nil)
+				nameFromContent := filepath.Join("cas",
+					fmt.Sprintf("%02x", hash[0]),
+					fmt.Sprintf("%02x", hash[1]),
+					fmt.Sprintf("%02x", hash[2]),
+					fmt.Sprintf("%02x%02x%02x%02x%02x%02x%02x%02x%02x%02x%02x%02x%02x",
+						hash[3],
+						hash[4], hash[5], hash[6], hash[7],
+						hash[8], hash[9], hash[10], hash[11],
+						hash[12], hash[13], hash[14], hash[15]))
+				if nameFromContent != fileName {
+					t.Errorf("localblobstore.ListCAIds hash of fragment: got %q, expected %q (content=%s)", nameFromContent, fileName, string(content))
+				}
+			}
+			caCount++
+		}
+		if iterator.Err() != nil {
+			t.Errorf("localblobstore.ListCAIds iteration failed: %v", iterator.Err())
+		}
+		if caCount != len(fragmentMap) {
+			t.Errorf("localblobstore.ListCAIds iteration expected %d files, got %d", len(fragmentMap), caCount)
+		}
+	}
+}
+
+// AddRetrieveAndDelete() tests adding, retrieving, and deleting blobs from a
+// blobstore bs.  One can't retrieve or delete something that hasn't been
+// created, so it's all done in one routine.    If testDirName is non-empty,
+// the blobstore is assumed to be accessible in the file system, and its
+// files are checked.
+func AddRetrieveAndDelete(t *testing.T, ctx *context.T, bs localblobstore.BlobStore, testDirName string) {
+	var err error
+
+	// Check that there are no files in the blobstore we were given.
+	iterator := bs.ListBlobIds(ctx)
+	for iterator.Advance() {
+		fileName := iterator.Value()
+		t.Errorf("unexpected file %q\n", fileName)
+	}
+	if iterator.Err() != nil {
+		t.Errorf("localblobstore.ListBlobIds iteration failed: %v", iterator.Err())
+	}
+
+	// Create the strings:  "wom", "bat", "wombat", "batwom", "atwo", "atwoatwoombatatwo".
+	womData := []byte("wom")
+	batData := []byte("bat")
+	wombatData := []byte("wombat")
+	batwomData := []byte("batwom")
+	atwoData := []byte("atwo")
+	atwoatwoombatatwoData := []byte("atwoatwoombatatwo")
+
+	// fragmentMap will have an entry per content-addressed fragment.
+	fragmentMap := make(map[string]bool)
+
+	// Create the blobs, by various means.
+
+	var blobVector []testBlob // Accumulate the blobs we create here.
+
+	blobVector = writeBlob(t, ctx, bs, blobVector,
+		womData, false,
+		blobOrBlockOrFile{block: womData})
+	womName := blobVector[len(blobVector)-1].blobName
+	fragmentMap[string(womData)] = true
+
+	blobVector = writeBlob(t, ctx, bs, blobVector,
+		batData, false,
+		blobOrBlockOrFile{block: batData})
+	batName := blobVector[len(blobVector)-1].blobName
+	fragmentMap[string(batData)] = true
+
+	blobVector = writeBlob(t, ctx, bs, blobVector,
+		wombatData, false,
+		blobOrBlockOrFile{block: wombatData})
+	firstWombatName := blobVector[len(blobVector)-1].blobName
+	fragmentMap[string(wombatData)] = true
+
+	blobVector = writeBlob(t, ctx, bs, blobVector,
+		wombatData, true,
+		blobOrBlockOrFile{block: womData},
+		blobOrBlockOrFile{block: batData})
+
+	blobVector = writeBlob(t, ctx, bs, blobVector,
+		wombatData, false,
+		blobOrBlockOrFile{
+			blob:   firstWombatName,
+			size:   -1,
+			offset: 0})
+
+	blobVector = writeBlob(t, ctx, bs, blobVector,
+		wombatData, false,
+		blobOrBlockOrFile{
+			blob:   firstWombatName,
+			size:   6,
+			offset: 0})
+
+	blobVector = writeBlob(t, ctx, bs, blobVector,
+		batwomData, false,
+		blobOrBlockOrFile{
+			blob:   firstWombatName,
+			size:   3,
+			offset: 3},
+		blobOrBlockOrFile{
+			blob:   firstWombatName,
+			size:   3,
+			offset: 0})
+	batwomName := blobVector[len(blobVector)-1].blobName
+
+	blobVector = writeBlob(t, ctx, bs, blobVector,
+		atwoData, false,
+		blobOrBlockOrFile{
+			blob:   batwomName,
+			size:   4,
+			offset: 1})
+	atwoName := blobVector[len(blobVector)-1].blobName
+
+	blobVector = writeBlob(t, ctx, bs, blobVector,
+		atwoatwoombatatwoData, true,
+		blobOrBlockOrFile{
+			blob:   atwoName,
+			size:   -1,
+			offset: 0},
+		blobOrBlockOrFile{
+			blob:   atwoName,
+			size:   4,
+			offset: 0},
+		blobOrBlockOrFile{
+			blob:   firstWombatName,
+			size:   -1,
+			offset: 1},
+		blobOrBlockOrFile{
+			blob:   batName,
+			size:   -1,
+			offset: 1},
+		blobOrBlockOrFile{
+			blob:   womName,
+			size:   2,
+			offset: 0})
+	atwoatwoombatatwoName := blobVector[len(blobVector)-1].blobName
+
+	// -------------------------------------------------
+	// Check that the state is as we expect.
+	checkWrittenBlobsAreReadable(t, ctx, bs, blobVector)
+	checkAllBlobs(t, ctx, bs, blobVector, testDirName)
+	checkFragments(t, ctx, bs, fragmentMap, testDirName)
+
+	// -------------------------------------------------
+	// Nothing should change if we garbage collect.
+	bs.GC(ctx)
+	checkWrittenBlobsAreReadable(t, ctx, bs, blobVector)
+	checkAllBlobs(t, ctx, bs, blobVector, testDirName)
+	checkFragments(t, ctx, bs, fragmentMap, testDirName)
+
+	// -------------------------------------------------
+	// Ensure that deleting non-existent blobs fails.
+	err = bs.DeleteBlob(ctx, "../../../../etc/passwd")
+	if verror.ErrorID(err) != "v.io/syncbase/x/ref/services/syncbase/localblobstore/fs_cablobstore.errInvalidBlobName" {
+		t.Errorf("DeleteBlob attempted to delete a bogus blob name")
+	}
+	err = bs.DeleteBlob(ctx, "foo/00/00/00/00000000000000000000000000")
+	if verror.ErrorID(err) != "v.io/syncbase/x/ref/services/syncbase/localblobstore/fs_cablobstore.errInvalidBlobName" {
+		t.Errorf("DeleteBlob attempted to delete a bogus blob name")
+	}
+
+	// -------------------------------------------------
+	// Delete a blob.
+	err = bs.DeleteBlob(ctx, batName)
+	if err != nil {
+		t.Errorf("DeleteBlob failed to delete blob %q: %v", batName, err)
+	}
+	blobVector = removeBlobFromBlobVector(blobVector, batName)
+
+	// -------------------------------------------------
+	// Check that the state is as we expect.
+	checkWrittenBlobsAreReadable(t, ctx, bs, blobVector)
+	checkAllBlobs(t, ctx, bs, blobVector, testDirName)
+	checkFragments(t, ctx, bs, fragmentMap, testDirName)
+
+	// -------------------------------------------------
+	// Nothing should change if we garbage collect.
+	bs.GC(ctx)
+	checkWrittenBlobsAreReadable(t, ctx, bs, blobVector)
+	checkAllBlobs(t, ctx, bs, blobVector, testDirName)
+	checkFragments(t, ctx, bs, fragmentMap, testDirName)
+
+	// -------------------------------------------------
+	// Open a BlobReader on a blob we're about to delete,
+	// so its fragments won't be garbage collected.
+
+	var br localblobstore.BlobReader
+	br, err = bs.NewBlobReader(ctx, atwoatwoombatatwoName)
+	if err != nil {
+		t.Errorf("NewBlobReader failed in blob %q: %v", atwoatwoombatatwoName, err)
+	}
+
+	// -------------------------------------------------
+	// Delete a blob.  This should be the last on-disc reference to the
+	// content-addressed fragment "bat", but the fragment won't be deleted
+	// until close the reader and garbage collect.
+	err = bs.DeleteBlob(ctx, atwoatwoombatatwoName)
+	if err != nil {
+		t.Errorf("DeleteBlob failed to delete blob %q: %v", atwoatwoombatatwoName, err)
+	}
+	blobVector = removeBlobFromBlobVector(blobVector, atwoatwoombatatwoName)
+
+	// -------------------------------------------------
+	// Check that the state is as we expect.
+	checkWrittenBlobsAreReadable(t, ctx, bs, blobVector)
+	checkAllBlobs(t, ctx, bs, blobVector, testDirName)
+	checkFragments(t, ctx, bs, fragmentMap, testDirName)
+
+	// -------------------------------------------------
+	// Garbage collection should change nothing; the fragment involved
+	// is still referenced from the open reader *br.
+	bs.GC(ctx)
+	checkWrittenBlobsAreReadable(t, ctx, bs, blobVector)
+	checkAllBlobs(t, ctx, bs, blobVector, testDirName)
+	checkFragments(t, ctx, bs, fragmentMap, testDirName)
+
+	// -------------------------------------------------
+
+	// Close the open BlobReader and garbage collect.
+	err = br.Close()
+	if err != nil {
+		t.Errorf("BlobReader.Close failed on blob %q: %v", atwoatwoombatatwoName, err)
+	}
+	delete(fragmentMap, string(batData))
+
+	bs.GC(ctx)
+	checkWrittenBlobsAreReadable(t, ctx, bs, blobVector)
+	checkAllBlobs(t, ctx, bs, blobVector, testDirName)
+	checkFragments(t, ctx, bs, fragmentMap, testDirName)
+
+	// -------------------------------------------------
+	// Delete all blobs.
+	for len(blobVector) != 0 {
+		err = bs.DeleteBlob(ctx, blobVector[0].blobName)
+		if err != nil {
+			t.Errorf("DeleteBlob failed to delete blob %q: %v", blobVector[0].blobName, err)
+		}
+		blobVector = removeBlobFromBlobVector(blobVector, blobVector[0].blobName)
+	}
+
+	// -------------------------------------------------
+	// Check that the state is as we expect.
+	checkWrittenBlobsAreReadable(t, ctx, bs, blobVector)
+	checkAllBlobs(t, ctx, bs, blobVector, testDirName)
+	checkFragments(t, ctx, bs, fragmentMap, testDirName)
+
+	// -------------------------------------------------
+	// The remaining fragments should be removed when we garbage collect.
+	for frag := range fragmentMap {
+		delete(fragmentMap, frag)
+	}
+	bs.GC(ctx)
+	checkWrittenBlobsAreReadable(t, ctx, bs, blobVector)
+	checkAllBlobs(t, ctx, bs, blobVector, testDirName)
+	checkFragments(t, ctx, bs, fragmentMap, testDirName)
+}
diff --git a/x/ref/services/syncbase/localblobstore/model.go b/x/ref/services/syncbase/localblobstore/model.go
new file mode 100644
index 0000000..b1a4565
--- /dev/null
+++ b/x/ref/services/syncbase/localblobstore/model.go
@@ -0,0 +1,170 @@
+// Copyright 2015 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 localblobstore is the interface to a local blob store.
+// Implementations include fs_cablobstore.
+package localblobstore
+
+import "v.io/v23/context"
+
+// A BlobStore represents a simple, content-addressable store.
+type BlobStore interface {
+	// NewBlobReader() returns a pointer to a newly allocated BlobReader on
+	// the specified blobName.  BlobReaders should not be used concurrently
+	// by multiple threads.  Returned handles should be closed with
+	// Close().
+	NewBlobReader(ctx *context.T, blobName string) (br BlobReader, err error)
+
+	// NewBlobWriter() returns a pointer to a newly allocated BlobWriter on
+	// a newly created blob name, which can be found using the Name()
+	// method.  BlobWriters should not be used concurrently by multiple
+	// threads.  The returned handle should be closed with either the
+	// Close() or CloseWithoutFinalize() method to avoid leaking file
+	// handles.
+	NewBlobWriter(ctx *context.T) (bw BlobWriter, err error)
+
+	// ResumeBlobWriter() returns a pointer to a newly allocated BlobWriter on
+	// an old, but unfinalized blob name.
+	ResumeBlobWriter(ctx *context.T, blobName string) (bw BlobWriter, err error)
+
+	// DeleteBlob() deletes the named blob from the BlobStore.
+	DeleteBlob(ctx *context.T, blobName string) (err error)
+
+	// GC() removes old temp files and content-addressed blocks that are no
+	// longer referenced by any blob.  It may be called concurrently with
+	// other calls to GC(), and with uses of BlobReaders and BlobWriters.
+	GC(ctx *context.T) error
+
+	// ListBlobIds() returns an iterator that can be used to enumerate the
+	// blobs in a BlobStore.  Expected use is:
+	//
+	//	iter := bs.ListBlobIds(ctx)
+	//	for iter.Advance() {
+	//	  // Process iter.Value() here.
+	//	}
+	//	if iter.Err() != nil {
+	//	  // The loop terminated early due to an error.
+	//	}
+	ListBlobIds(ctx *context.T) (iter Iter)
+
+	// ListCAIds() returns an iterator that can be used to enumerate the
+	// content-addressable fragments in a BlobStore.  Expected use is:
+	//
+	//	iter := bs.ListCAIds(ctx)
+	//	for iter.Advance() {
+	//	  // Process iter.Value() here.
+	//	}
+	//	if iter.Err() != nil {
+	//	  // The loop terminated early due to an error.
+	//	}
+	ListCAIds(ctx *context.T) (iter Iter)
+
+	// Root() returns the name of the root directory where the BlobStore is stored.
+	Root() string
+}
+
+// A BlobReader allows a blob to be read using the standard ReadAt(), Read(),
+// and Seek() calls.  A BlobReader can be created with NewBlobReader(), and
+// should be closed with the Close() method to avoid leaking file handles.
+type BlobReader interface {
+	// ReadAt() fills b[] with up to len(b) bytes of data starting at
+	// position "at" within the blob that the BlobReader indicates, and
+	// returns the number of bytes read.
+	ReadAt(b []byte, at int64) (n int, err error)
+
+	// Read() fills b[] with up to len(b) bytes of data starting at the
+	// current seek position of the BlobReader within the blob that the
+	// BlobReader indicates, and then both returns the number of bytes read
+	// and advances the BlobReader's seek position by that amount.
+	Read(b []byte) (n int, err error)
+
+	// Seek() sets the seek position of the BlobReader to offset if
+	// whence==0, offset+current_seek_position if whence==1, and
+	// offset+end_of_blob if whence==2, and then returns the current seek
+	// position.
+	Seek(offset int64, whence int) (result int64, err error)
+
+	// Close() indicates that the client will perform no further operations
+	// on the BlobReader.  It releases any resources held by the
+	// BlobReader.
+	Close() error
+
+	// Name() returns the BlobReader's name.
+	Name() string
+
+	// Size() returns the BlobReader's size.
+	Size() int64
+
+	// IsFinalized() returns whether the BlobReader has been finalized.
+	IsFinalized() bool
+
+	// Hash() returns the BlobReader's hash.  It may be nil if the blob is
+	// not finalized.
+	Hash() []byte
+}
+
+// A BlockOrFile represents a vector of bytes, and contains either a data
+// block (as a []byte), or a (file name, size, offset) triple.
+type BlockOrFile struct {
+	Block    []byte // If FileName is empty, the bytes represented.
+	FileName string // If non-empty, the name of the file containing the bytes.
+	Size     int64  // If FileName is non-empty, the number of bytes (or -1 for "all")
+	Offset   int64  // If FileName is non-empty, the offset of the relevant bytes within the file.
+}
+
+// A BlobWriter allows a blob to be written.  If a blob has not yet been
+// finalized, it also allows that blob to be extended.  A BlobWriter may be
+// created with NewBlobWriter(), and should be closed with Close() or
+// CloseWithoutFinalize().
+type BlobWriter interface {
+	// AppendBlob() adds a (substring of a) pre-existing blob to the blob
+	// being written by the BlobWriter.  The fragments of the pre-existing
+	// blob are not physically copied; they are referenced by both blobs.
+	AppendBlob(blobName string, size int64, offset int64) (err error)
+
+	// AppendFragment() appends a fragment to the blob being written by the
+	// BlobWriter, where the fragment is composed of the byte vectors
+	// described by the elements of item[].  The fragment is copied into
+	// the blob store.
+	AppendFragment(item ...BlockOrFile) (err error)
+
+	// Close() finalizes the BlobWriter, and indicates that the client will
+	// perform no further append operations on the BlobWriter.  Any
+	// internal open file handles are closed.
+	Close() (err error)
+
+	// CloseWithoutFinalize() indicates that the client will perform no
+	// further append operations on the BlobWriter, but does not finalize
+	// the blob.  Any internal open file handles are closed.  Clients are
+	// expected to need this operation infrequently.
+	CloseWithoutFinalize() (err error)
+
+	// Name() returns the BlobWriter's name.
+	Name() string
+
+	// Size() returns the BlobWriter's size.
+	Size() int64
+
+	// IsFinalized() returns whether the BlobWriter has been finalized.
+	IsFinalized() bool
+
+	// Hash() returns the BlobWriter's hash, reflecting the bytes written so far.
+	Hash() []byte
+}
+
+// A Iter represents an iterator that allows the client to enumerate
+// all the blobs of fragments in a BlobStore.
+type Iter 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.
+	Advance() (advanced bool)
+
+	// Value() returns the item that was staged by Advance.  May panic if
+	// Advance returned false or was not called.  Never blocks.
+	Value() (name string)
+
+	// Err() returns any error encountered by Advance.  Never blocks.
+	Err() error
+}
diff --git a/x/ref/services/syncbase/server/app.go b/x/ref/services/syncbase/server/app.go
index b35afce..827eaf9 100644
--- a/x/ref/services/syncbase/server/app.go
+++ b/x/ref/services/syncbase/server/app.go
@@ -161,7 +161,7 @@
 			return err
 		}
 		// Check for "database already exists".
-		if _, err := a.getDbInfo(ctx, call, st, dbName); verror.ErrorID(err) != verror.ErrNoExist.ID {
+		if _, err := a.getDbInfo(ctx, st, dbName); verror.ErrorID(err) != verror.ErrNoExist.ID {
 			if err != nil {
 				return err
 			}
@@ -172,7 +172,7 @@
 		info := &dbInfo{
 			Name: dbName,
 		}
-		return a.putDbInfo(ctx, call, st, dbName, info)
+		return a.putDbInfo(ctx, st, dbName, info)
 	}); err != nil {
 		return err
 	}
@@ -192,7 +192,7 @@
 
 	// 3. Flip dbInfo.Initialized to true.
 	if err := store.RunInTransaction(a.s.st, func(st store.StoreReadWriter) error {
-		return a.updateDbInfo(ctx, call, st, dbName, func(info *dbInfo) error {
+		return a.updateDbInfo(ctx, st, dbName, func(info *dbInfo) error {
 			info.Initialized = true
 			return nil
 		})
@@ -234,7 +234,7 @@
 
 	// 2. Flip dbInfo.Deleted to true.
 	if err := store.RunInTransaction(a.s.st, func(st store.StoreReadWriter) error {
-		return a.updateDbInfo(ctx, call, st, dbName, func(info *dbInfo) error {
+		return a.updateDbInfo(ctx, st, dbName, func(info *dbInfo) error {
 			info.Deleted = true
 			return nil
 		})
@@ -251,7 +251,7 @@
 	}
 
 	// 4. Delete dbInfo record.
-	if err := a.delDbInfo(ctx, call, a.s.st, dbName); err != nil {
+	if err := a.delDbInfo(ctx, a.s.st, dbName); err != nil {
 		return err
 	}
 
diff --git a/x/ref/services/syncbase/server/db_info.go b/x/ref/services/syncbase/server/db_info.go
index 98216bc..cbfcebd 100644
--- a/x/ref/services/syncbase/server/db_info.go
+++ b/x/ref/services/syncbase/server/db_info.go
@@ -17,7 +17,6 @@
 	"v.io/syncbase/x/ref/services/syncbase/server/util"
 	"v.io/syncbase/x/ref/services/syncbase/store"
 	"v.io/v23/context"
-	"v.io/v23/rpc"
 )
 
 type dbInfoLayer struct {
@@ -49,9 +48,9 @@
 
 // getDbInfo reads data from the storage engine.
 // Returns a VDL-compatible error.
-func (a *app) getDbInfo(ctx *context.T, call rpc.ServerCall, st store.StoreReader, dbName string) (*dbInfo, error) {
+func (a *app) getDbInfo(ctx *context.T, st store.StoreReader, dbName string) (*dbInfo, error) {
 	info := &dbInfo{}
-	if err := util.GetWithoutAuth(ctx, call, st, &dbInfoLayer{dbName, a}, info); err != nil {
+	if err := util.GetWithoutAuth(ctx, st, &dbInfoLayer{dbName, a}, info); err != nil {
 		return nil, err
 	}
 	return info, nil
@@ -59,27 +58,27 @@
 
 // putDbInfo writes data to the storage engine.
 // Returns a VDL-compatible error.
-func (a *app) putDbInfo(ctx *context.T, call rpc.ServerCall, st store.StoreWriter, dbName string, info *dbInfo) error {
-	return util.Put(ctx, call, st, &dbInfoLayer{dbName, a}, info)
+func (a *app) putDbInfo(ctx *context.T, st store.StoreWriter, dbName string, info *dbInfo) error {
+	return util.Put(ctx, st, &dbInfoLayer{dbName, a}, info)
 }
 
 // delDbInfo deletes data from the storage engine.
 // Returns a VDL-compatible error.
-func (a *app) delDbInfo(ctx *context.T, call rpc.ServerCall, st store.StoreWriter, dbName string) error {
-	return util.Delete(ctx, call, st, &dbInfoLayer{dbName, a})
+func (a *app) delDbInfo(ctx *context.T, st store.StoreWriter, dbName string) error {
+	return util.Delete(ctx, st, &dbInfoLayer{dbName, a})
 }
 
 // updateDbInfo performs a read-modify-write.
 // fn should "modify" v, and should return a VDL-compatible error.
 // Returns a VDL-compatible error.
-func (a *app) updateDbInfo(ctx *context.T, call rpc.ServerCall, st store.StoreReadWriter, dbName string, fn func(info *dbInfo) error) error {
+func (a *app) updateDbInfo(ctx *context.T, st store.StoreReadWriter, dbName string, fn func(info *dbInfo) error) error {
 	_ = st.(store.Transaction) // panics on failure, as desired
-	info, err := a.getDbInfo(ctx, call, st, dbName)
+	info, err := a.getDbInfo(ctx, st, dbName)
 	if err != nil {
 		return err
 	}
 	if err := fn(info); err != nil {
 		return err
 	}
-	return a.putDbInfo(ctx, call, st, dbName, info)
+	return a.putDbInfo(ctx, st, dbName, info)
 }
diff --git a/x/ref/services/syncbase/server/nosql/database.go b/x/ref/services/syncbase/server/nosql/database.go
index 0971923..d18ab04 100644
--- a/x/ref/services/syncbase/server/nosql/database.go
+++ b/x/ref/services/syncbase/server/nosql/database.go
@@ -15,7 +15,6 @@
 	wire "v.io/syncbase/v23/services/syncbase/nosql"
 	"v.io/syncbase/v23/syncbase/nosql/query_db"
 	"v.io/syncbase/v23/syncbase/nosql/query_exec"
-	prefixutil "v.io/syncbase/v23/syncbase/util"
 	"v.io/syncbase/x/ref/services/syncbase/server/interfaces"
 	"v.io/syncbase/x/ref/services/syncbase/server/util"
 	"v.io/syncbase/x/ref/services/syncbase/server/watchable"
@@ -103,7 +102,7 @@
 		Name:  d.name,
 		Perms: opts.Perms,
 	}
-	if err := util.Put(ctx, call, d.st, d, data); err != nil {
+	if err := util.Put(ctx, d.st, d, data); err != nil {
 		return nil, err
 	}
 	return d, nil
@@ -404,20 +403,20 @@
 	req *tableReq
 }
 
-func (t *tableDb) Scan(prefixes []string) (query_db.KeyValueStream, error) {
+func (t *tableDb) Scan(keyRanges query_db.KeyRanges) (query_db.KeyValueStream, error) {
 	return &kvs{
-		t:        t,
-		prefixes: prefixes,
-		curr:     -1,
-		validRow: false,
-		it:       nil,
-		err:      nil,
+		t:         t,
+		keyRanges: keyRanges,
+		curr:      -1,
+		validRow:  false,
+		it:        nil,
+		err:       nil,
 	}, nil
 }
 
 type kvs struct {
 	t         *tableDb
-	prefixes  []string
+	keyRanges query_db.KeyRanges
 	curr      int // current index into prefixes, -1 at start
 	validRow  bool
 	currKey   string
@@ -433,16 +432,22 @@
 	if s.curr == -1 {
 		s.curr++
 	}
-	for s.curr < len(s.prefixes) {
+	for s.curr < len(s.keyRanges) {
 		if s.it == nil {
-			start := prefixutil.PrefixRangeStart(s.prefixes[s.curr])
-			limit := prefixutil.PrefixRangeLimit(s.prefixes[s.curr])
-			s.it = s.t.qdb.st.Scan(util.ScanRangeArgs(util.JoinKeyParts(util.RowPrefix, s.t.req.name), string(start), string(limit)))
+			start := s.keyRanges[s.curr].Start
+			limit := s.keyRanges[s.curr].Limit
+			// 0-255 means examine all rows
+			if start == string([]byte{0}) && limit == string([]byte{255}) {
+				start = ""
+				limit = ""
+			}
+			s.it = s.t.qdb.st.Scan(util.ScanRangeArgs(util.JoinKeyParts(util.RowPrefix, s.t.req.name), start, limit))
 		}
 		if s.it.Advance() {
 			// key
 			keyBytes := s.it.Key(nil)
 			parts := util.SplitKeyParts(string(keyBytes))
+			// TODO(rogulenko): Check access for the key.
 			s.currKey = parts[len(parts)-1]
 			// value
 			valueBytes := s.it.Value(nil)
@@ -489,8 +494,8 @@
 		s.it.Cancel()
 		s.it = nil
 	}
-	// set curr to end of prefixes so Advance will return false
-	s.curr = len(s.prefixes)
+	// set curr to end of keyRanges so Advance will return false
+	s.curr = len(s.keyRanges)
 }
 
 ////////////////////////////////////////
diff --git a/x/ref/services/syncbase/server/nosql/row.go b/x/ref/services/syncbase/server/nosql/row.go
index 3397fd1..542224a 100644
--- a/x/ref/services/syncbase/server/nosql/row.go
+++ b/x/ref/services/syncbase/server/nosql/row.go
@@ -91,11 +91,10 @@
 }
 
 // checkAccess checks that this row's table exists in the database, and performs
-// an authorization check (currently against the table perms).
+// an authorization check.
 // Returns a VDL-compatible error.
-// TODO(sadovsky): Use prefix permissions.
 func (r *rowReq) checkAccess(ctx *context.T, call rpc.ServerCall, st store.StoreReader) error {
-	return util.Get(ctx, call, st, r.t, &tableData{})
+	return r.t.checkAccess(ctx, call, st, r.key)
 }
 
 // get reads data from the storage engine.
diff --git a/x/ref/services/syncbase/server/nosql/table.go b/x/ref/services/syncbase/server/nosql/table.go
index 7619fc4..82a0cab 100644
--- a/x/ref/services/syncbase/server/nosql/table.go
+++ b/x/ref/services/syncbase/server/nosql/table.go
@@ -5,6 +5,8 @@
 package nosql
 
 import (
+	"strings"
+
 	wire "v.io/syncbase/v23/services/syncbase/nosql"
 	"v.io/syncbase/x/ref/services/syncbase/server/util"
 	"v.io/syncbase/x/ref/services/syncbase/store"
@@ -12,6 +14,7 @@
 	"v.io/v23/rpc"
 	"v.io/v23/security/access"
 	"v.io/v23/verror"
+	"v.io/v23/vom"
 )
 
 // tableReq is a per-request object that handles Table RPCs.
@@ -39,7 +42,7 @@
 			return err
 		}
 		// Check for "table already exists".
-		if err := util.GetWithoutAuth(ctx, call, st, t, &tableData{}); verror.ErrorID(err) != verror.ErrNoExist.ID {
+		if err := util.GetWithoutAuth(ctx, st, t, &tableData{}); verror.ErrorID(err) != verror.ErrNoExist.ID {
 			if err != nil {
 				return err
 			}
@@ -54,7 +57,7 @@
 			Name:  t.name,
 			Perms: perms,
 		}
-		return util.Put(ctx, call, st, t, data)
+		return util.Put(ctx, st, t, data)
 	})
 }
 
@@ -71,20 +74,30 @@
 			return err
 		}
 		// TODO(sadovsky): Delete all rows in this table.
-		return util.Delete(ctx, call, st, t)
+		return util.Delete(ctx, st, t)
 	})
 }
 
 func (t *tableReq) DeleteRowRange(ctx *context.T, call rpc.ServerCall, start, limit []byte) error {
 	impl := func(st store.StoreReadWriter) error {
-		// Check perms.
-		if err := util.Get(ctx, call, st, t, &tableData{}); err != nil {
+		// Check for table-level access before doing a scan.
+		if err := t.checkAccess(ctx, call, st, ""); err != nil {
 			return err
 		}
 		it := st.Scan(util.ScanRangeArgs(util.JoinKeyParts(util.RowPrefix, t.name), string(start), string(limit)))
 		key := []byte{}
 		for it.Advance() {
 			key = it.Key(key)
+			// Check perms.
+			parts := util.SplitKeyParts(string(key))
+			externalKey := parts[len(parts)-1]
+			if err := t.checkAccess(ctx, call, st, externalKey); err != nil {
+				// TODO(rogulenko): Revisit this behavior. Probably we should
+				// delete all rows that we have access to.
+				it.Cancel()
+				return err
+			}
+			// Delete the key-value pair.
 			if err := st.Delete(key); err != nil {
 				return verror.New(verror.ErrInternal, ctx, err)
 			}
@@ -107,8 +120,8 @@
 
 func (t *tableReq) Scan(ctx *context.T, call wire.TableScanServerCall, start, limit []byte) error {
 	impl := func(st store.StoreReader) error {
-		// Check perms.
-		if err := util.Get(ctx, call, st, t, &tableData{}); err != nil {
+		// Check for table-level access before doing a scan.
+		if err := t.checkAccess(ctx, call, st, ""); err != nil {
 			return err
 		}
 		it := st.Scan(util.ScanRangeArgs(util.JoinKeyParts(util.RowPrefix, t.name), string(start), string(limit)))
@@ -116,8 +129,14 @@
 		key, value := []byte{}, []byte{}
 		for it.Advance() {
 			key, value = it.Key(key), it.Value(value)
+			// Check perms.
 			parts := util.SplitKeyParts(string(key))
-			sender.Send(wire.KeyValue{Key: parts[len(parts)-1], Value: value})
+			externalKey := parts[len(parts)-1]
+			if err := t.checkAccess(ctx, call, st, externalKey); err != nil {
+				it.Cancel()
+				return err
+			}
+			sender.Send(wire.KeyValue{Key: externalKey, Value: value})
 		}
 		if err := it.Err(); err != nil {
 			return verror.New(verror.ErrInternal, ctx, err)
@@ -135,38 +154,27 @@
 	return impl(st)
 }
 
-func (t *tableReq) SetPermissions(ctx *context.T, call rpc.ServerCall, prefix string, perms access.Permissions) error {
-	if prefix != "" {
-		return verror.NewErrNotImplemented(ctx)
-	}
-	impl := func(st store.StoreReadWriter) error {
-		data := &tableData{}
-		return util.Update(ctx, call, st, t, data, func() error {
-			data.Perms = perms
-			return nil
-		})
-	}
-	if t.d.batchId != nil {
-		if st, err := t.d.batchReadWriter(); err != nil {
-			return err
-		} else {
-			return impl(st)
-		}
-	} else {
-		return store.RunInTransaction(t.d.st, impl)
-	}
-}
-
 func (t *tableReq) GetPermissions(ctx *context.T, call rpc.ServerCall, key string) ([]wire.PrefixPermissions, error) {
-	if key != "" {
-		return nil, verror.NewErrNotImplemented(ctx)
-	}
 	impl := func(st store.StoreReader) ([]wire.PrefixPermissions, error) {
-		data := &tableData{}
-		if err := util.Get(ctx, call, t.d.st, t, data); err != nil {
+		// Check permissions only at table level.
+		if err := t.checkAccess(ctx, call, st, ""); err != nil {
 			return nil, err
 		}
-		return []wire.PrefixPermissions{{Prefix: "", Perms: data.Perms}}, nil
+		// Get the most specific permissions object.
+		prefix, prefixPerms, err := t.permsForKey(ctx, st, key)
+		if err != nil {
+			return nil, err
+		}
+		result := []wire.PrefixPermissions{{Prefix: prefix, Perms: prefixPerms.Perms}}
+		// Collect all parent permissions objects all the way up to the table level.
+		for prefix != "" {
+			prefix = prefixPerms.Parent
+			if prefixPerms, err = t.permsForPrefix(ctx, st, prefixPerms.Parent); err != nil {
+				return nil, err
+			}
+			result = append(result, wire.PrefixPermissions{Prefix: prefix, Perms: prefixPerms.Perms})
+		}
+		return result, nil
 	}
 	var st store.StoreReader
 	if t.d.batchId != nil {
@@ -179,17 +187,114 @@
 	return impl(st)
 }
 
+func (t *tableReq) SetPermissions(ctx *context.T, call rpc.ServerCall, prefix string, perms access.Permissions) error {
+	impl := func(st store.StoreReadWriter) error {
+		if err := t.checkAccess(ctx, call, st, prefix); 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 {
+			return err
+		}
+		if prefix == "" {
+			data := &tableData{}
+			return util.Update(ctx, call, st, t, data, func() error {
+				data.Perms = perms
+				return nil
+			})
+		}
+		// Get the most specific permissions object.
+		parent, prefixPerms, err := t.permsForKey(ctx, st, prefix)
+		if err != nil {
+			return err
+		}
+		// In case there is no permissions object for the given prefix, we need
+		// to add a new node to the prefix permissions tree. We do it by updating
+		// parents for all children of the prefix to the node corresponding to
+		// the prefix.
+		if parent != prefix {
+			if err := t.updateParentRefs(ctx, st, prefix, prefix); err != nil {
+				return err
+			}
+		} else {
+			parent = prefixPerms.Parent
+		}
+		stPrefix := t.prefixPermsKey(prefix)
+		stPrefixLimit := stPrefix + util.PrefixRangeLimitSuffix
+		prefixPerms = stPrefixPerms{Parent: parent, Perms: perms}
+		// Put the (prefix, perms) pair to the database.
+		if err := util.PutObject(st, stPrefix, prefixPerms); err != nil {
+			return err
+		}
+		return util.PutObject(st, stPrefixLimit, prefixPerms)
+	}
+	if t.d.batchId != nil {
+		if st, err := t.d.batchReadWriter(); err != nil {
+			return err
+		} else {
+			return impl(st)
+		}
+	} else {
+		return store.RunInTransaction(t.d.st, impl)
+	}
+}
+
 func (t *tableReq) DeletePermissions(ctx *context.T, call rpc.ServerCall, prefix string) error {
-	return verror.NewErrNotImplemented(ctx)
+	if prefix == "" {
+		return verror.New(verror.ErrBadArg, ctx, prefix)
+	}
+	impl := func(st store.StoreReadWriter) error {
+		if err := t.checkAccess(ctx, call, st, prefix); 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 {
+			return err
+		}
+		// Get the most specific permissions object.
+		parent, prefixPerms, err := t.permsForKey(ctx, st, prefix)
+		if err != nil {
+			return err
+		}
+		if parent != prefix {
+			// This can happen only if there is no permissions object for the
+			// given prefix. Since DeletePermissions is idempotent, return nil.
+			return nil
+		}
+		// We need to delete the node corresponding to the prefix from the prefix
+		// permissions tree. We do it by updating parents for all children of the
+		// prefix to the parent of the node corresponding to the prefix.
+		if err := t.updateParentRefs(ctx, st, prefix, prefixPerms.Parent); err != nil {
+			return err
+		}
+		stPrefix := []byte(t.prefixPermsKey(prefix))
+		stPrefixLimit := append(stPrefix, util.PrefixRangeLimitSuffix...)
+		if err := st.Delete(stPrefix); err != nil {
+			return err
+		}
+		return st.Delete(stPrefixLimit)
+	}
+	if t.d.batchId != nil {
+		if st, err := t.d.batchReadWriter(); err != nil {
+			return err
+		} else {
+			return impl(st)
+		}
+	} else {
+		return store.RunInTransaction(t.d.st, impl)
+	}
 }
 
 func (t *tableReq) GlobChildren__(ctx *context.T, call rpc.ServerCall) (<-chan string, error) {
 	impl := func(st store.StoreReader, closeStoreReader func() error) (<-chan string, error) {
 		// Check perms.
-		if err := util.Get(ctx, call, st, t, &tableData{}); err != nil {
+		if err := t.checkAccess(ctx, call, st, ""); err != nil {
 			closeStoreReader()
 			return nil, err
 		}
+		// TODO(rogulenko): Check prefix permissions for children.
 		return util.Glob(ctx, call, "*", st, closeStoreReader, util.JoinKeyParts(util.RowPrefix, t.name))
 	}
 	var st store.StoreReader
@@ -226,3 +331,145 @@
 func (t *tableReq) stKeyPart() string {
 	return t.name
 }
+
+// updateParentRefs updates the parent for all children of the given
+// prefix to newParent.
+func (t *tableReq) updateParentRefs(ctx *context.T, st store.StoreReadWriter, prefix, newParent string) error {
+	stPrefix := []byte(t.prefixPermsKey(prefix))
+	stPrefixStart := append(stPrefix, 0)
+	stPrefixLimit := append(stPrefix, util.PrefixRangeLimitSuffix...)
+	it := st.Scan(stPrefixStart, stPrefixLimit)
+	var key, value []byte
+	for it.Advance() {
+		key, value = it.Key(key), it.Value(value)
+		var prefixPerms stPrefixPerms
+		if err := vom.Decode(value, &prefixPerms); err != nil {
+			it.Cancel()
+			return verror.New(verror.ErrInternal, ctx, err)
+		}
+		prefixPerms.Parent = newParent
+		if err := util.PutObject(st, string(key), prefixPerms); err != nil {
+			it.Cancel()
+			return err
+		}
+	}
+	if err := it.Err(); err != nil {
+		return verror.New(verror.ErrInternal, ctx, err)
+	}
+	return nil
+}
+
+// lock invalidates all concurrent transactions with ErrConcurrentTransaction
+// that have accessed this table.
+// Returns a VDL-compatible error.
+//
+// It is necessary to call lock() every time prefix permissions are updated,
+// so snapshots inside all transactions reflect up-to-date permissions. Since
+// every public function that touches this table has to read the table-level
+// permissions object, it is enough to add the key of table-level permissions
+// to the write set of the current transaction.
+//
+// TODO(rogulenko): Revisit this behavior to provide more granularity.
+// A possible option would be to add prefix and its parent to the write set
+// of the current transaction when permissions object for a prefix is updated.
+func (t *tableReq) lock(ctx *context.T, st store.StoreReadWriter) error {
+	var data tableData
+	if err := util.GetWithoutAuth(ctx, st, t, &data); err != nil {
+		return err
+	}
+	return util.Put(ctx, st, t, data)
+}
+
+// checkAccess checks that this table exists in the database, and performs
+// an authorization check. The access is checked at table level and at the
+// level of the most specific prefix for the given key.
+// Returns a VDL-compatible error.
+// TODO(rogulenko): Revisit this behavior. Eventually we'll want the table-level
+// access check to be a check for "Resolve", i.e. also check access to
+// service, app and database.
+func (t *tableReq) checkAccess(ctx *context.T, call rpc.ServerCall, st store.StoreReader, key string) error {
+	prefix, prefixPerms, err := t.permsForKey(ctx, st, key)
+	if err != nil {
+		return err
+	}
+	if prefix != "" {
+		if err := util.Get(ctx, call, st, t, &tableData{}); err != nil {
+			return err
+		}
+	}
+	auth, _ := access.PermissionsAuthorizer(prefixPerms.Perms, access.TypicalTagType())
+	if err := auth.Authorize(ctx, call.Security()); err != nil {
+		return verror.New(verror.ErrNoAccess, ctx, prefix)
+	}
+	return nil
+}
+
+// permsForKey returns the longest prefix of the given key that has
+// associated permissions with its permissions object.
+// permsForKey doesn't perform an authorization check.
+// Returns a VDL-compatible error.
+//
+// Virtually we represent all prefixes as a forest T, where each vertex maps to
+// a prefix. A parent for a string is the maximum proper prefix of it that
+// belongs to T. Each prefix P from T is represented as a pair of entries with
+// keys P and P~ with values of type stPrefixPerms (parent + perms).
+// High level of how this function works:
+// 1	iter = db.Scan(K, "")
+// 		Here last character of iter.Key() is removed automatically if it is '~'
+// 2	if hasPrefix(K, iter.Key()) return iter.Value()
+// 3	return parent(iter.Key())
+// Short proof:
+// iter returned on line 1 points to one of the following:
+// - a string t that is equal to K;
+// - a string t~: if t is not a prefix of K, then K < t < t~ which
+//   contradicts with property of returned iterator on line 1 => t is prefix of
+//   K; also t is the largest prefix of K, as all larger prefixes of K are
+//   less than t~; in this case line 2 returns correct result;
+// - a string t that doesn't end with '~': it can't be a prefix of K, as all
+//   proper prefixes of K are less than K; parent(t) is a prefix of K, otherwise
+//   K < parent(t) < t; parent(t) is the largest prefix of K, otherwise t is a
+//   prefix of K; in this case line 3 returns correct result.
+func (t *tableReq) permsForKey(ctx *context.T, st store.StoreReader, key string) (string, stPrefixPerms, error) {
+	it := st.Scan(util.ScanRangeArgs(util.JoinKeyParts(util.PermsPrefix, t.name), key, ""))
+	if !it.Advance() {
+		prefixPerms, err := t.permsForPrefix(ctx, st, "")
+		return "", prefixPerms, err
+	}
+	defer it.Cancel()
+	parts := util.SplitKeyParts(string(it.Key(nil)))
+	prefix := strings.TrimSuffix(parts[len(parts)-1], util.PrefixRangeLimitSuffix)
+	value := it.Value(nil)
+	var prefixPerms stPrefixPerms
+	if err := vom.Decode(value, &prefixPerms); err != nil {
+		return "", stPrefixPerms{}, verror.New(verror.ErrInternal, ctx, err)
+	}
+	if strings.HasPrefix(key, prefix) {
+		return prefix, prefixPerms, nil
+	}
+	prefixPerms, err := t.permsForPrefix(ctx, st, prefixPerms.Parent)
+	return prefixPerms.Parent, prefixPerms, err
+}
+
+// permsForPrefix returns the permissions object associated with the
+// provided prefix.
+// Returns a VDL-compatible error.
+func (t *tableReq) permsForPrefix(ctx *context.T, st store.StoreReader, prefix string) (stPrefixPerms, error) {
+	if prefix == "" {
+		var data tableData
+		if err := util.GetWithoutAuth(ctx, st, t, &data); err != nil {
+			return stPrefixPerms{}, err
+		}
+		return stPrefixPerms{Perms: data.Perms}, nil
+	}
+	var prefixPerms stPrefixPerms
+	if err := util.GetObject(st, t.prefixPermsKey(prefix), &prefixPerms); err != nil {
+		return stPrefixPerms{}, verror.New(verror.ErrInternal, ctx, err)
+	}
+	return prefixPerms, nil
+}
+
+// prefixPermsKey returns the key used for storing permissions for the given
+// prefix in the table.
+func (t *tableReq) prefixPermsKey(prefix string) string {
+	return util.JoinKeyParts(util.PermsPrefix, t.name, prefix)
+}
diff --git a/x/ref/services/syncbase/server/nosql/types.vdl b/x/ref/services/syncbase/server/nosql/types.vdl
index 4a8a328..33d3883 100644
--- a/x/ref/services/syncbase/server/nosql/types.vdl
+++ b/x/ref/services/syncbase/server/nosql/types.vdl
@@ -21,3 +21,17 @@
 	Name  string
 	Perms access.Permissions
 }
+
+// stPrefixPerms describes internal representation of prefix permissions
+// in the store.
+//
+// Each pair of (key, perms) is stored as two key-value pairs:
+// "$perms:%table:key"  - stPrefixPerms{parent, perms}
+// "$perms:%table:key~" - stPrefixPerms{parent, perms}
+// where "~" represents a reserved char that's lexicographically greater than
+// all chars allowed by clients, %table is the name of the table and parent is
+// the longest proper prefix of the key that has associated permissions object.
+type stPrefixPerms struct {
+	Parent string
+	Perms  access.Permissions
+}
diff --git a/x/ref/services/syncbase/server/nosql/types.vdl.go b/x/ref/services/syncbase/server/nosql/types.vdl.go
index 313d982..021cef6 100644
--- a/x/ref/services/syncbase/server/nosql/types.vdl.go
+++ b/x/ref/services/syncbase/server/nosql/types.vdl.go
@@ -39,7 +39,27 @@
 }) {
 }
 
+// stPrefixPerms describes internal representation of prefix permissions
+// in the store.
+//
+// Each pair of (key, perms) is stored as two key-value pairs:
+// "$perms:%table:key"  - stPrefixPerms{parent, perms}
+// "$perms:%table:key~" - stPrefixPerms{parent, perms}
+// where "~" represents a reserved char that's lexicographically greater than
+// all chars allowed by clients, %table is the name of the table and parent is
+// the longest proper prefix of the key that has associated permissions object.
+type stPrefixPerms struct {
+	Parent string
+	Perms  access.Permissions
+}
+
+func (stPrefixPerms) __VDLReflect(struct {
+	Name string `vdl:"v.io/syncbase/x/ref/services/syncbase/server/nosql.stPrefixPerms"`
+}) {
+}
+
 func init() {
 	vdl.Register((*databaseData)(nil))
 	vdl.Register((*tableData)(nil))
+	vdl.Register((*stPrefixPerms)(nil))
 }
diff --git a/x/ref/services/syncbase/server/service.go b/x/ref/services/syncbase/server/service.go
index e551431..135ee35 100644
--- a/x/ref/services/syncbase/server/service.go
+++ b/x/ref/services/syncbase/server/service.go
@@ -68,7 +68,7 @@
 	data := &serviceData{
 		Perms: opts.Perms,
 	}
-	if err := util.Put(ctx, call, s.st, s, data); err != nil {
+	if err := util.Put(ctx, s.st, s, data); err != nil {
 		return nil, err
 	}
 	if s.sync, err = vsync.New(ctx, call, s); err != nil {
@@ -178,7 +178,7 @@
 			return err
 		}
 		// Check for "app already exists".
-		if err := util.GetWithoutAuth(ctx, call, st, a, &appData{}); verror.ErrorID(err) != verror.ErrNoExist.ID {
+		if err := util.GetWithoutAuth(ctx, st, a, &appData{}); verror.ErrorID(err) != verror.ErrNoExist.ID {
 			if err != nil {
 				return err
 			}
@@ -192,7 +192,7 @@
 			Name:  appName,
 			Perms: perms,
 		}
-		return util.Put(ctx, call, st, a, data)
+		return util.Put(ctx, st, a, data)
 	}); err != nil {
 		return err
 	}
@@ -218,7 +218,7 @@
 			return err
 		}
 		// TODO(sadovsky): Delete all databases in this app.
-		return util.Delete(ctx, call, st, a)
+		return util.Delete(ctx, st, a)
 	}); err != nil {
 		return err
 	}
diff --git a/x/ref/services/syncbase/server/util/constants.go b/x/ref/services/syncbase/server/util/constants.go
index 44d0ff6..9251375 100644
--- a/x/ref/services/syncbase/server/util/constants.go
+++ b/x/ref/services/syncbase/server/util/constants.go
@@ -12,6 +12,7 @@
 	DatabasePrefix = "$database"
 	DbInfoPrefix   = "$dbInfo"
 	LogPrefix      = "$log"
+	PermsPrefix    = "$perms"
 	RowPrefix      = "$row"
 	ServicePrefix  = "$service"
 	SyncPrefix     = "$sync"
@@ -27,4 +28,8 @@
 	BatchSep = ":"
 	// Separator for parts of storage engine keys.
 	KeyPartSep = ":"
+	// PrefixRangeLimitSuffix is the suffix of a key which indicates the end of
+	// a prefix range. Should be more than any regular key in the store.
+	// TODO(rogulenko): Change this constant to something out of the UTF8 space.
+	PrefixRangeLimitSuffix = "~"
 )
diff --git a/x/ref/services/syncbase/server/util/store_util.go b/x/ref/services/syncbase/server/util/store_util.go
index a495ebd..fb739ca 100644
--- a/x/ref/services/syncbase/server/util/store_util.go
+++ b/x/ref/services/syncbase/server/util/store_util.go
@@ -47,7 +47,7 @@
 
 // GetWithoutAuth does st.Get(l.StKey(), v), populating v.
 // Returns a VDL-compatible error.
-func GetWithoutAuth(ctx *context.T, call rpc.ServerCall, st store.StoreReader, l Layer, v interface{}) error {
+func GetWithoutAuth(ctx *context.T, st store.StoreReader, l Layer, v interface{}) error {
 	if err := GetObject(st, l.StKey(), v); err != nil {
 		if verror.ErrorID(err) == store.ErrUnknownKey.ID {
 			return verror.New(verror.ErrNoExist, ctx, l.Name())
@@ -60,7 +60,7 @@
 // Get does GetWithoutAuth followed by an auth check.
 // Returns a VDL-compatible error.
 func Get(ctx *context.T, call rpc.ServerCall, st store.StoreReader, l Layer, v Permser) error {
-	if err := GetWithoutAuth(ctx, call, st, l, v); err != nil {
+	if err := GetWithoutAuth(ctx, st, l, v); err != nil {
 		return err
 	}
 	auth, _ := access.PermissionsAuthorizer(v.GetPerms(), access.TypicalTagType())
@@ -73,7 +73,7 @@
 // Put does st.Put(l.StKey(), v).
 // Returns a VDL-compatible error.
 // If you need to perform an authorization check, use Update().
-func Put(ctx *context.T, _ rpc.ServerCall, st store.StoreWriter, l Layer, v interface{}) error {
+func Put(ctx *context.T, st store.StoreWriter, l Layer, v interface{}) error {
 	if err := PutObject(st, l.StKey(), v); err != nil {
 		return verror.New(verror.ErrInternal, ctx, err)
 	}
@@ -83,7 +83,7 @@
 // Delete does st.Delete(l.StKey()).
 // Returns a VDL-compatible error.
 // If you need to perform an authorization check, call Get() first.
-func Delete(ctx *context.T, _ rpc.ServerCall, st store.StoreWriter, l Layer) error {
+func Delete(ctx *context.T, st store.StoreWriter, l Layer) error {
 	if err := st.Delete([]byte(l.StKey())); err != nil {
 		return verror.New(verror.ErrInternal, ctx, err)
 	}
@@ -103,7 +103,7 @@
 	if err := fn(); err != nil {
 		return err
 	}
-	return Put(ctx, call, st, l, v)
+	return Put(ctx, st, l, v)
 }
 
 ////////////////////////////////////////////////////////////
diff --git a/x/ref/services/syncbase/syncbased/main.go b/x/ref/services/syncbase/syncbased/main.go
index df96244..a75b471 100644
--- a/x/ref/services/syncbase/syncbased/main.go
+++ b/x/ref/services/syncbase/syncbased/main.go
@@ -19,6 +19,7 @@
 	"v.io/syncbase/x/ref/services/syncbase/server"
 	"v.io/x/ref/lib/security/securityflag"
 	"v.io/x/ref/lib/signals"
+	"v.io/x/ref/lib/xrpc"
 	_ "v.io/x/ref/runtime/factories/generic"
 )
 
@@ -44,26 +45,16 @@
 	ctx, shutdown := v23.Init()
 	defer shutdown()
 
-	s, err := v23.NewServer(ctx)
-	if err != nil {
-		vlog.Fatal("v23.NewServer() failed: ", err)
-	}
-	if _, err := s.Listen(v23.GetListenSpec(ctx)); err != nil {
-		vlog.Fatal("s.Listen() failed: ", err)
-	}
-
 	perms, err := securityflag.PermissionsFromFlag()
 	if err != nil {
 		vlog.Fatal("securityflag.PermissionsFromFlag() failed: ", err)
 	}
-
 	if perms != nil {
 		vlog.Info("Using permissions from command line flag.")
 	} else {
 		vlog.Info("No permissions flag provided. Giving local principal all permissions.")
 		perms = defaultPerms(security.DefaultBlessingPatterns(v23.GetPrincipal(ctx)))
 	}
-
 	service, err := server.NewService(nil, nil, server.ServiceOptions{
 		Perms:   perms,
 		RootDir: *rootDir,
@@ -74,9 +65,8 @@
 	}
 	d := server.NewDispatcher(service)
 
-	// Publish the service in the mount table.
-	if err := s.ServeDispatcher(*name, d); err != nil {
-		vlog.Fatal("s.ServeDispatcher() failed: ", err)
+	if _, err = xrpc.NewDispatchingServer(ctx, *name, d); err != nil {
+		vlog.Fatal("xrpc.NewDispatchingServer() failed: ", err)
 	}
 	vlog.Info("Mounted at: ", *name)