feat(go-collector): migrate with CREATE TABLE IF NOT EXISTS (SQLR-11)

examples/go-collector/README.md

@@ -275,8 +275,8 @@ against the engine, each with a code comment pointing here:
 | Constraint | Where it bites | How the collector handles it |
 |---|---|---|
 | **No parameter binding** in the Go driver | every write | All values are inlined through `internal/store/sqlquote.go`, the single chokepoint that escapes text (doubled `''`) and validates JSON. |
-| **`CREATE TABLE IF NOT EXISTS` not honored**; `sqlrite_master` not queryable | reopening a populated DB | `migrate()` probes for the `events` table with a `SELECT` and only runs DDL on a fresh database. |
-| **`CREATE INDEX` rejected under `journal_mode = mvcc`** | the optional index | All DDL runs in WAL mode *before* the MVCC switch; the index choice is fixed at DB-creation time. |
+| **`CREATE INDEX` has no `IF NOT EXISTS` under MVCC** | reopening a populated DB | Tables use `CREATE TABLE IF NOT EXISTS` on every startup (SQLR-10). The optional `idx_events_device` is created only when `sqlrite_master` does not already list it, in WAL mode, before the MVCC switch. |
+| **`CREATE INDEX` rejected under `journal_mode = mvcc`** | the optional index | Issued only when `sqlrite_master` does not already list `idx_events_device`, before the MVCC switch. Reopening with a different `-indexed` flag does not drop an existing index. |
 | **`BEGIN CONCURRENT` commit batch capped at 4 KiB** (the encoded row image, not just the SQL) | a large checkpoint, and any single oversized row | Two guards: event payloads are bounded at ingest (`maxPayloadBytes`, returns `400`) so any one row commits; and `CommitUpload` marks rows in adaptively-sized chunks that halve on a cap error down to one-per-commit (`writeAdaptive`). Relaxes the checkpoint from atomic to incremental → at-least-once delivery. |
 | **`AUTOINCREMENT` rowids collide under MVCC** | concurrent inserts | Event ids are assigned application-side from an atomic counter seeded off `MAX(id)` at open. |
 | **`IS NULL` never uses an index** | the backlog scan | `WHERE uploaded_at IS NULL` is a full scan by design — fine for a bounded edge buffer; the optional index is on `device_id` instead. |

examples/go-collector/internal/store/store.go

@@ -20,17 +20,14 @@
 //
 //   - No parameter binding in the Go SDK → values are inlined via the
 //     helpers in sqlquote.go.
-//   - migrate() probes for the events table with a SELECT and only runs
-//     DDL on a fresh database. NOTE: as of SQLR-10 the engine now honors
-//     `CREATE TABLE IF NOT EXISTS` and exposes a queryable `sqlrite_master`
-//     (and `PRAGMA table_list`), so the table-existence probe is no longer
-//     strictly required for table creation. We keep the fresh/reopen
-//     distinction because the `CREATE INDEX` below must NOT be re-issued on
-//     reopen (it's rejected once `journal_mode = mvcc`); the probe also
-//     keeps this example working against pre-SQLR-10 engine builds.
-//   - `CREATE INDEX` is rejected once `journal_mode = mvcc` → all DDL,
-//     including the optional secondary index, runs at migrate time
-//     before MVCC is switched on.
+//   - migrate() runs `CREATE TABLE IF NOT EXISTS` unconditionally (SQLR-10
+//     made that a no-op on reopen). The optional `CREATE INDEX` is still
+//     gated: it is rejected once `journal_mode = mvcc`, including on reopen
+//     of a DB that already switched, so we only issue it when
+//     `sqlrite_master` does not already list `idx_events_device`, and only
+//     before the MVCC switch below.
+//   - `CREATE INDEX` is rejected once `journal_mode = mvcc` → the optional
+//     secondary index runs at migrate time before MVCC is switched on.
 //   - A single BEGIN CONCURRENT commit batch is capped at 4 KiB → event
 //     payloads are bounded at ingest (see maxPayloadBytes) so any one
 //     row commits, and the uploader's checkpoint marks rows in
@@ -141,10 +138,10 @@ type Store struct {
 	devs  map[string]int64
 }
 
-// Open creates/opens the database, applies the schema on first use, and
-// (in Concurrent mode) switches the database into MVCC. DDL runs before
-// the MVCC switch because CREATE INDEX is rejected once
-// journal_mode = mvcc; see migrate for the fresh-vs-reopen detection.
+// Open creates/opens the database, applies the schema (idempotent
+// CREATE TABLE IF NOT EXISTS), and (in Concurrent mode) switches the
+// database into MVCC. The optional CREATE INDEX runs before the MVCC
+// switch and only when sqlrite_master does not already list it.
 func Open(ctx context.Context, opts Options) (*Store, error) {
 	if opts.MaxOpenConns <= 0 {
 		opts.MaxOpenConns = 8
@@ -195,63 +192,56 @@ func (s *Store) Close() error {
 	return s.db.Close()
 }
 
-// migrate creates the schema on a fresh database and is a no-op on
-// reopen. What shapes this:
+// migrate applies the schema on every open. What shapes this:
 //
-//   - We detect a fresh database by probing for the events table with a
-//     cheap SELECT and only run DDL when it's absent. As of SQLR-10 the
-//     engine honors `CREATE TABLE IF NOT EXISTS` and exposes a queryable
-//     `sqlrite_master`, so the tables alone wouldn't need the probe — but
-//     see the next point.
-//   - `CREATE INDEX` is rejected once `journal_mode = mvcc`. All DDL
-//     (tables + the optional index) therefore runs on the fresh path,
-//     in WAL mode, *before* the MVCC switch. On reopen the index already
-//     exists, so we never re-issue it — which is why the fresh/reopen
-//     probe stays even though IF NOT EXISTS would cover the tables.
+//   - Tables use `CREATE TABLE IF NOT EXISTS` (SQLR-10). Running them
+//     on reopen is a no-op; there is no SELECT-to-probe workaround.
+//   - `CREATE INDEX` is rejected once `journal_mode = mvcc`, including
+//     on reopen of a DB that already switched. The optional index is
+//     therefore issued only when `sqlrite_master` does not already list
+//     `idx_events_device`, and only *before* the MVCC switch below.
 func (s *Store) migrate(ctx context.Context) error {
-	fresh := !s.tableExists(ctx, "events")
-
-	if fresh {
-		ddl := []string{
-			`CREATE TABLE events (
-				id           INTEGER PRIMARY KEY,
-				device_id    TEXT NOT NULL,
-				kind         TEXT NOT NULL,
-				payload_json JSON,
-				ts           INTEGER NOT NULL,
-				uploaded_at  INTEGER
-			)`,
-			`CREATE TABLE devices (
-				id           INTEGER PRIMARY KEY,
-				device_key   TEXT NOT NULL,
-				label        TEXT,
-				last_seen_at INTEGER
-			)`,
-			`CREATE TABLE upload_runs (
-				id          INTEGER PRIMARY KEY,
-				started_at  INTEGER NOT NULL,
-				finished_at INTEGER,
-				event_count INTEGER NOT NULL,
-				status      TEXT NOT NULL,
-				error       TEXT
-			)`,
+	tables := []string{
+		`CREATE TABLE IF NOT EXISTS events (
+			id           INTEGER PRIMARY KEY,
+			device_id    TEXT NOT NULL,
+			kind         TEXT NOT NULL,
+			payload_json JSON,
+			ts           INTEGER NOT NULL,
+			uploaded_at  INTEGER
+		)`,
+		`CREATE TABLE IF NOT EXISTS devices (
+			id           INTEGER PRIMARY KEY,
+			device_key   TEXT NOT NULL,
+			label        TEXT,
+			last_seen_at INTEGER
+		)`,
+		`CREATE TABLE IF NOT EXISTS upload_runs (
+			id          INTEGER PRIMARY KEY,
+			started_at  INTEGER NOT NULL,
+			finished_at INTEGER,
+			event_count INTEGER NOT NULL,
+			status      TEXT NOT NULL,
+			error       TEXT
+		)`,
+	}
+	for _, q := range tables {
+		if _, err := s.db.ExecContext(ctx, q); err != nil {
+			return fmt.Errorf("migrate: %w", err)
 		}
-		if s.opts.Indexed {
-			// Single-column B-tree index (composite indexes are
-			// unsupported). Accelerates per-device diagnostic queries
-			// (`WHERE device_id = '...'`); the trade is extra index
-			// maintenance on every concurrent write, which the loadgen
-			// measures. The index choice is fixed at DB-creation time —
-			// reopening with a different -indexed flag does not add or
-			// drop it (we'd have to CREATE INDEX under MVCC, which the
-			// engine rejects).
-			ddl = append(ddl,
-				`CREATE INDEX idx_events_device ON events (device_id)`)
-		}
-		for _, q := range ddl {
-			if _, err := s.db.ExecContext(ctx, q); err != nil {
-				return fmt.Errorf("migrate: %w", err)
-			}
+	}
+
+	if s.opts.Indexed && !s.indexExists(ctx, "idx_events_device") {
+		// Single-column B-tree index (composite indexes are
+		// unsupported). Accelerates per-device diagnostic queries
+		// (`WHERE device_id = '...'`); the trade is extra index
+		// maintenance on every concurrent write, which the loadgen
+		// measures. Do not re-issue this on reopen: CREATE INDEX is
+		// rejected under MVCC, and SQLRite has no IF NOT EXISTS
+		// escape for indexes.
+		q := `CREATE INDEX idx_events_device ON events (device_id)`
+		if _, err := s.db.ExecContext(ctx, q); err != nil {
+			return fmt.Errorf("migrate index: %w", err)
 		}
 	}
 
@@ -265,19 +255,18 @@ func (s *Store) migrate(ctx context.Context) error {
 	return nil
 }
 
-// tableExists probes for a table with a zero-row SELECT. This is how we
-// tell a fresh database from a reopened one so the MVCC-incompatible
-// `CREATE INDEX` only runs once. A query error (the engine returns
-// "Table '<name>' not found") means absent. (As of SQLR-10 the engine
-// also exposes `sqlrite_master` and `PRAGMA table_list` for catalog
-// introspection — either could back this probe on a current engine.)
-func (s *Store) tableExists(ctx context.Context, name string) bool {
-	rows, err := s.db.QueryContext(ctx, fmt.Sprintf("SELECT id FROM %s LIMIT 1", name))
-	if err != nil {
+// indexExists reports whether sqlrite_master already lists the named
+// index. Used to skip CREATE INDEX on reopen (SQLR-10 catalog).
+func (s *Store) indexExists(ctx context.Context, name string) bool {
+	q := fmt.Sprintf(
+		"SELECT name FROM sqlrite_master WHERE type = 'index' AND name = %s LIMIT 1",
+		quoteText(name),
+	)
+	var found string
+	if err := s.db.QueryRowContext(ctx, q).Scan(&found); err != nil {
 		return false
 	}
-	_ = rows.Close()
-	return true
+	return found != ""
 }
 
 // seed primes the atomic id counters and the backlog gauge from

examples/go-collector/internal/store/store_test.go

@@ -321,6 +321,63 @@ func TestSerializedModeWrites(t *testing.T) {
 	}
 }
 
+// TestMigrateIdempotentFreshAndReopen is the SQLR-11 regression: CREATE
+// TABLE IF NOT EXISTS must be a no-op on reopen, and CREATE INDEX must
+// not be re-issued (it is rejected under persisted MVCC).
+func TestMigrateIdempotentFreshAndReopen(t *testing.T) {
+	ctx := context.Background()
+	dir := t.TempDir()
+
+	for _, tc := range []struct {
+		name    string
+		mode    WriteMode
+		indexed bool
+	}{
+		{"concurrent-plain", Concurrent, false},
+		{"concurrent-indexed", Concurrent, true},
+		{"serialized-plain", Serialized, false},
+		{"serialized-indexed", Serialized, true},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			path := filepath.Join(dir, tc.name+".sqlrite")
+			st1, err := Open(ctx, Options{Path: path, Mode: tc.mode, Indexed: tc.indexed, MaxOpenConns: 4})
+			if err != nil {
+				t.Fatalf("fresh open: %v", err)
+			}
+			if _, err := st1.InsertEvent(ctx, ev("d", "k", 1)); err != nil {
+				st1.Close()
+				t.Fatalf("insert on fresh: %v", err)
+			}
+			if got := st1.indexExists(ctx, "idx_events_device"); got != tc.indexed {
+				st1.Close()
+				t.Fatalf("fresh indexExists = %v, want %v", got, tc.indexed)
+			}
+			if n := scalar(t, st1, "SELECT COUNT(*) FROM sqlrite_master WHERE type = 'table' AND name = 'events'"); n != 1 {
+				st1.Close()
+				t.Fatalf("events catalog count = %d, want 1", n)
+			}
+			if err := st1.Close(); err != nil {
+				t.Fatalf("close fresh: %v", err)
+			}
+
+			st2, err := Open(ctx, Options{Path: path, Mode: tc.mode, Indexed: tc.indexed, MaxOpenConns: 4})
+			if err != nil {
+				t.Fatalf("reopen: %v", err)
+			}
+			defer st2.Close()
+			if _, err := st2.InsertEvent(ctx, ev("d", "k", 2)); err != nil {
+				t.Fatalf("insert on reopen: %v", err)
+			}
+			if got, err := st2.CountEvents(ctx); err != nil || got != 2 {
+				t.Fatalf("CountEvents after reopen = %d, err=%v, want 2", got, err)
+			}
+			if got := st2.indexExists(ctx, "idx_events_device"); got != tc.indexed {
+				t.Fatalf("reopen indexExists = %v, want %v", got, tc.indexed)
+			}
+		})
+	}
+}
+
 // --- helpers ---
 
 func scalar(t *testing.T, st *Store, q string) int64 {