persist cmvs in db

Author: jakeheft

CMV_SUPPORT.md

@@ -82,8 +82,9 @@ row, err := table.ReadRow(ctx, "region-a#account-42#...")
   creation are not reflected in the CMV table.
 - **Backfill**: Data written to the source table before the CMV is registered is not
   retroactively copied.
-- **Persistence**: CMV registrations are in-memory only and do not survive a restart.
-  Re-register via `CreateMaterializedView` on each startup.
+- **Persistence**: CMV registrations are persisted to SQLite alongside table data and are
+  automatically restored on startup. Shadow table row data is also persistent (it is stored
+  as a regular table in `tables_t`).
 
 ## Example: Key Transformation
 

bttest/cmv_test.go

@@ -27,6 +27,7 @@ func newTestServerWithCMV(t *testing.T, configs []CMVConfig) *server {
 		materializedViews: make(map[string]*btapb.MaterializedView),
 		db:                db,
 		tableBackend:      NewSqlTables(db),
+		mvBackend:         NewSqlMaterializedViews(db),
 		cmvs:              newCMVRegistry(),
 	}
 	for _, cfg := range configs {
@@ -287,6 +288,7 @@ func TestCreateMaterializedViewRPC(t *testing.T) {
 		materializedViews: make(map[string]*btapb.MaterializedView),
 		db:                db,
 		tableBackend:      NewSqlTables(db),
+		mvBackend:         NewSqlMaterializedViews(db),
 		cmvs:              newCMVRegistry(),
 	}
 
@@ -431,6 +433,7 @@ func newTestInstanceServer(t *testing.T) (*server, string) {
 		materializedViews: make(map[string]*btapb.MaterializedView),
 		db:                db,
 		tableBackend:      NewSqlTables(db),
+		mvBackend:         NewSqlMaterializedViews(db),
 		cmvs:              newCMVRegistry(),
 	}
 	return s, "projects/test/instances/test"

bttest/inmem.go

@@ -94,6 +94,7 @@ type server struct {
 	gcc          chan int                   // set when gcloop starts, closed when server shuts down
 	db           *sql.DB
 	tableBackend *SqlTables
+	mvBackend    *SqlMaterializedViews
 	cmvs         *cmvRegistry
 
 	materializedViews map[string]*btapb.MaterializedView // keyed by full resource name
@@ -123,6 +124,7 @@ func NewServer(laddr string, db *sql.DB, opt ...grpc.ServerOption) (*Server, err
 			materializedViews: make(map[string]*btapb.MaterializedView),
 			db:                db,
 			tableBackend:      NewSqlTables(db),
+			mvBackend:         NewSqlMaterializedViews(db),
 			cmvs:              newCMVRegistry(),
 		},
 	}
@@ -131,6 +133,7 @@ func NewServer(laddr string, db *sql.DB, opt ...grpc.ServerOption) (*Server, err
 	}
 	longrunningpb.RegisterOperationsServer(s.srv, opsServer)
 	s.s.LoadTables()
+	s.s.LoadMaterializedViews()
 	btapb.RegisterBigtableInstanceAdminServer(s.srv, s.s)
 	btapb.RegisterBigtableTableAdminServer(s.srv, s.s)
 	btpb.RegisterBigtableServer(s.srv, s.s)
@@ -159,6 +162,36 @@ func (s *server) LoadTables() {
 	}
 }
 
+// LoadMaterializedViews restores persisted CMV metadata from SQLite on startup.
+// For each stored view, it re-parses the SQL to reconstruct the CMVConfig and
+// re-registers it so that write-time propagation resumes immediately.
+func (s *server) LoadMaterializedViews() {
+	for _, v := range s.mvBackend.GetAll() {
+		cfg, err := ParseCMVConfigFromSQL(viewIDFromName(v.name), v.query)
+		if err != nil {
+			// Log and skip rather than crashing — the parser may have evolved.
+			log.Printf("WARNING: skipping persisted materialized view %q: could not parse query: %v", v.name, err)
+			continue
+		}
+		s.cmvs.register(*cfg)
+		s.materializedViews[v.name] = &btapb.MaterializedView{
+			Name:               v.name,
+			Query:              v.query,
+			DeletionProtection: v.deletionProtection,
+		}
+	}
+}
+
+// viewIDFromName extracts the view ID from a full resource name of the form
+// projects/.../instances/.../materializedViews/<id>.
+func viewIDFromName(name string) string {
+	const sep = "/materializedViews/"
+	if idx := strings.LastIndex(name, sep); idx >= 0 {
+		return name[idx+len(sep):]
+	}
+	return name
+}
+
 // ensureCMVTable creates the shadow table for a CMV if it doesn't already exist.
 // Copies column families from the source table, filtered by IncludeFamilies.
 // Must be called with s.mu held.
@@ -383,6 +416,7 @@ func (s *server) DeleteTable(ctx context.Context, req *btapb.DeleteTableRequest)
 			}
 			for mvName := range s.materializedViews {
 				if strings.HasSuffix(mvName, "/materializedViews/"+viewID) {
+					s.mvBackend.Delete(mvName)
 					delete(s.materializedViews, mvName)
 					break
 				}

bttest/inmem_test.go

@@ -78,6 +78,7 @@ func newTestServer(t *testing.T) *server {
 		tables:       make(map[string]*table),
 		db:           db,
 		tableBackend: NewSqlTables(db),
+		mvBackend:    NewSqlMaterializedViews(db),
 		cmvs:         newCMVRegistry(),
 	}
 	return s

bttest/instance_server.go

@@ -176,6 +176,7 @@ func (s *server) CreateMaterializedView(ctx context.Context, req *btapb.CreateMa
 		DeletionProtection: mv.DeletionProtection,
 	}
 	s.materializedViews[name] = stored
+	s.mvBackend.Save(name, mv.Query, mv.DeletionProtection)
 
 	respAny, err := anypb.New(stored)
 	if err != nil {
@@ -236,6 +237,7 @@ func (s *server) UpdateMaterializedView(ctx context.Context, req *btapb.UpdateMa
 			return nil, status.Errorf(codes.InvalidArgument, "unsupported update field: %q", path)
 		}
 	}
+	s.mvBackend.Save(stored.Name, stored.Query, stored.DeletionProtection)
 
 	respAny, err := anypb.New(stored)
 	if err != nil {
@@ -271,6 +273,7 @@ func (s *server) DeleteMaterializedView(ctx context.Context, req *btapb.DeleteMa
 			delete(s.tables, fqShadow)
 		}
 	}
+	s.mvBackend.Delete(req.Name)
 	delete(s.materializedViews, req.Name)
 	return new(empty.Empty), nil
 }

bttest/sql_materialized_views.go

@@ -0,0 +1,74 @@
+package bttest
+
+import (
+	"database/sql"
+	"log"
+)
+
+// SqlMaterializedViews persists materialized view metadata to materialized_views_t.
+type SqlMaterializedViews struct {
+	db *sql.DB
+}
+
+// NewSqlMaterializedViews returns a SqlMaterializedViews backed by the given DB.
+func NewSqlMaterializedViews(db *sql.DB) *SqlMaterializedViews {
+	return &SqlMaterializedViews{db: db}
+}
+
+type storedMaterializedView struct {
+	name               string
+	query              string
+	deletionProtection bool
+}
+
+// GetAll returns all persisted materialized views, used to restore state on startup.
+func (m *SqlMaterializedViews) GetAll() []storedMaterializedView {
+	rows, err := m.db.Query("SELECT name, query, deletion_protection FROM materialized_views_t")
+	if err == sql.ErrNoRows {
+		return nil
+	}
+	if err != nil {
+		log.Fatal(err)
+	}
+	defer rows.Close()
+
+	var result []storedMaterializedView
+	for rows.Next() {
+		var v storedMaterializedView
+		var dp int
+		if err := rows.Scan(&v.name, &v.query, &dp); err != nil {
+			log.Fatal(err)
+		}
+		v.deletionProtection = dp != 0
+		result = append(result, v)
+	}
+	if err := rows.Err(); err != nil {
+		log.Fatal(err)
+	}
+	return result
+}
+
+// Save upserts a materialized view record. Called on CreateMaterializedView and
+// UpdateMaterializedView to keep the persisted state in sync with in-memory state.
+func (m *SqlMaterializedViews) Save(name, query string, deletionProtection bool) {
+	dp := 0
+	if deletionProtection {
+		dp = 1
+	}
+	_, err := m.db.Exec(
+		"INSERT INTO materialized_views_t (name, query, deletion_protection) VALUES (?, ?, ?)"+
+			" ON CONFLICT(name) DO UPDATE SET query=excluded.query, deletion_protection=excluded.deletion_protection",
+		name, query, dp,
+	)
+	if err != nil {
+		log.Fatalf("saving materialized view %q: %v", name, err)
+	}
+}
+
+// Delete removes a materialized view record by its full resource name.
+func (m *SqlMaterializedViews) Delete(name string) {
+	_, err := m.db.Exec("DELETE FROM materialized_views_t WHERE name = ?", name)
+	if err != nil {
+		log.Fatal(err)
+	}
+}

bttest/sql_rows.go

@@ -24,6 +24,7 @@ type SqlRows struct {
 	db *sql.DB
 }
 
+// NewSqlRows returns a SqlRows for a specific table within the given parent instance path.
 func NewSqlRows(db *sql.DB, parent, tableId string) *SqlRows {
 	return &SqlRows{
 		parent:  parent,

bttest/sql_schema.go

@@ -5,6 +5,8 @@ import (
 	"database/sql"
 )
 
+// CreateTables initializes the SQLite schema for the emulator, creating all
+// required tables if they do not already exist. Safe to call on an existing DB.
 func CreateTables(ctx context.Context, db *sql.DB) error {
 	query := "CREATE TABLE IF NOT EXISTS rows_t ( \n" +
 		"`parent` TEXT NOT NULL,\n" +
@@ -32,5 +34,15 @@ func CreateTables(ctx context.Context, db *sql.DB) error {
 	if err != nil {
 		return err
 	}
+
+	query = "CREATE TABLE IF NOT EXISTS materialized_views_t (\n" +
+		"`name` TEXT PRIMARY KEY,\n" +
+		"`query` TEXT NOT NULL,\n" +
+		"`deletion_protection` INTEGER NOT NULL DEFAULT 0\n" +
+		")"
+	_, err = db.ExecContext(ctx, query)
+	if err != nil {
+		return err
+	}
 	return nil
 }

bttest/sql_tables.go

@@ -15,6 +15,7 @@ type SqlTables struct {
 	db *sql.DB
 }
 
+// NewSqlTables returns a SqlTables backed by the given DB.
 func NewSqlTables(db *sql.DB) *SqlTables {
 	return &SqlTables{
 		db: db,
@@ -51,6 +52,7 @@ func (t *table) Bytes() ([]byte, error) {
 	return b.Bytes(), err
 }
 
+// Get loads a single table's metadata from the DB. Returns nil if not found.
 func (db *SqlTables) Get(parent, tableId string) *table {
 	tbl := &table{
 		parent:  parent,
@@ -64,6 +66,7 @@ func (db *SqlTables) Get(parent, tableId string) *table {
 	return tbl
 }
 
+// GetAll loads all table metadata from the DB, used to restore state on startup.
 func (db *SqlTables) GetAll() []*table {
 	var tables []*table
 
@@ -89,6 +92,7 @@ func (db *SqlTables) GetAll() []*table {
 	return tables
 }
 
+// Save upserts a table's metadata (column family definitions) to the DB.
 func (db *SqlTables) Save(t *table) {
 	metadata, err := t.Bytes()
 	if err != nil {
@@ -103,6 +107,7 @@ func (db *SqlTables) Save(t *table) {
 	}
 }
 
+// Delete removes a table's metadata record from the DB.
 func (db *SqlTables) Delete(t *table) {
 	_, err := db.db.Exec("DELETE FROM tables_t WHERE parent = ? AND table_id = ? ", t.parent, t.tableId)
 	if err != nil {