fix: gate matview read path on shape check to prevent "column does not exist" during rolling deploys

framework/logstore/matviews.go

@@ -78,7 +78,12 @@ ON mv_logs_hourly (hour, provider, model, status, object_type, selected_key_id,
 
 // mvLogsHourlyRequiredColumns is the canonical column set used by
 // repairMatViewShapes to detect old-shape mv_logs_hourly views from prior
-// schema versions and drop them so they get rebuilt on startup.
+// schema versions and drop them so they get rebuilt on startup, and by
+// matViewShapesReady to gate the matview read path.
+//
+// Must mirror every output column of mvLogsHourlyDDL. A partial list would let
+// a view missing an unlisted column pass both checks, and readers selecting it
+// would fail with "column does not exist" — see TestMvLogsHourlyRequiredColumnsMatchDDL.
 var mvLogsHourlyRequiredColumns = []string{
 	"hour",
 	"provider",
@@ -94,10 +99,22 @@ var mvLogsHourlyRequiredColumns = []string{
 	"business_unit_id",
 	"alias",
 	"canonical_model_name",
+	"count",
+	"success_count",
+	"error_count",
 	"cancelled_count",
+	"avg_latency",
+	"p90_latency",
+	"p95_latency",
+	"p99_latency",
+	"total_prompt_tokens",
+	"total_completion_tokens",
 	"throughput_completion_tokens",
 	"throughput_latency_ms",
 	"throughput_request_count",
+	"total_tokens",
+	"total_cached_read_tokens",
+	"total_cost",
 }
 
 // legacyMatViewNames are matviews from previous schema versions that no longer
@@ -429,39 +446,43 @@ var matviewRequiredColumns = func() map[string][]string {
 // transaction. Multi-replica deployments serialize on the advisory lock so
 // only one instance does the work. It shares the same advisory lock as
 // refreshMatViews so startup create/repair cannot overlap a periodic refresh.
-func ensureMatViews(ctx context.Context, db *gorm.DB) error {
+//
+// Returns false (with a nil error) when another replica holds the lock: nothing
+// about the view shape is established then, so callers must verify with
+// matViewShapesReady before enabling the matview read path.
+func ensureMatViews(ctx context.Context, db *gorm.DB) (bool, error) {
 	if db.Dialector.Name() != "postgres" {
-		return nil
+		return false, nil
 	}
 
 	sqlDB, err := db.DB()
 	if err != nil {
-		return fmt.Errorf("failed to get sql.DB for matview creation: %w", err)
+		return false, fmt.Errorf("failed to get sql.DB for matview creation: %w", err)
 	}
 
 	conn, err := sqlDB.Conn(ctx)
 	if err != nil {
-		return fmt.Errorf("failed to get dedicated connection for matview creation: %w", err)
+		return false, fmt.Errorf("failed to get dedicated connection for matview creation: %w", err)
 	}
 	defer conn.Close()
 
 	var acquired bool
 	if err := conn.QueryRowContext(ctx, "SELECT pg_try_advisory_lock($1)", matviewRefreshAdvisoryLockKey).Scan(&acquired); err != nil {
-		return fmt.Errorf("failed to try advisory lock for matview creation: %w", err)
+		return false, fmt.Errorf("failed to try advisory lock for matview creation: %w", err)
 	}
 	if !acquired {
 		// Another replica is doing the work — nothing to do here.
-		return nil
+		return false, nil
 	}
 	defer func() {
 		_, _ = conn.ExecContext(ctx, "SELECT pg_advisory_unlock($1)", matviewRefreshAdvisoryLockKey)
 	}()
 
 	if err := dropLegacyMatViews(ctx, conn); err != nil {
-		return err
+		return false, err
 	}
 	if err := repairMatViewShapes(ctx, conn); err != nil {
-		return err
+		return false, err
 	}
 
 	ddls := []string{mvLogsHourlyDDL}
@@ -470,15 +491,80 @@ func ensureMatViews(ctx context.Context, db *gorm.DB) error {
 	}
 	for _, ddl := range ddls {
 		if _, err := conn.ExecContext(ctx, ddl); err != nil {
-			return fmt.Errorf("failed to create materialized view: %w", err)
+			return false, fmt.Errorf("failed to create materialized view: %w", err)
 		}
 	}
 
 	if err := ensureMatViewUniqueIndexes(ctx, conn); err != nil {
-		return err
+		return false, err
 	}
 
-	return nil
+	return true, nil
+}
+
+// matViewShapesReady reports whether every managed materialized view exists and
+// carries the full column set this build reads. One read-only pg_catalog query,
+// no lock, no DDL — safe to repeat on refresher ticks.
+//
+// Gates matViewsReady during a rolling deploy: without it a replica that skipped
+// the repair reads an old-shape view and gets "column does not exist".
+func matViewShapesReady(ctx context.Context, db *gorm.DB) (bool, error) {
+	if db.Dialector.Name() != "postgres" {
+		return false, nil
+	}
+
+	sqlDB, err := db.DB()
+	if err != nil {
+		return false, fmt.Errorf("failed to get sql.DB for matview shape check: %w", err)
+	}
+
+	conn, err := sqlDB.Conn(ctx)
+	if err != nil {
+		return false, fmt.Errorf("failed to get connection for matview shape check: %w", err)
+	}
+	defer conn.Close()
+
+	// Flattened into parallel arrays so the check is one round trip. A pair is
+	// unsatisfied when the view is absent, shadowed in the search path, or
+	// missing the column — all three mean "not ready".
+	views := make([]string, 0, len(matviewRequiredColumns)*len(mvLogsHourlyRequiredColumns))
+	columns := make([]string, 0, cap(views))
+	for view, required := range matviewRequiredColumns {
+		for _, column := range required {
+			views = append(views, view)
+			columns = append(columns, column)
+		}
+	}
+
+	var missing int
+	if err := conn.QueryRowContext(ctx, `
+		SELECT COUNT(*)
+		FROM unnest($1::text[], $2::text[]) AS required(view_name, column_name)
+		WHERE NOT EXISTS (
+			SELECT 1
+			FROM pg_class c
+			JOIN pg_attribute a ON a.attrelid = c.oid
+			WHERE c.relkind = 'm'
+			  AND c.relname = required.view_name
+			  AND pg_catalog.pg_table_is_visible(c.oid)
+			  AND a.attnum > 0
+			  AND NOT a.attisdropped
+			  AND a.attname = required.column_name
+		)
+	`, pqTextArray(views), pqTextArray(columns)).Scan(&missing); err != nil {
+		return false, fmt.Errorf("failed to inspect matview shapes: %w", err)
+	}
+	return missing == 0, nil
+}
+
+// pqTextArray renders a []string as a Postgres text[] literal, avoiding a
+// driver-specific array type for the one place that needs it.
+func pqTextArray(values []string) string {
+	quoted := make([]string, 0, len(values))
+	for _, v := range values {
+		quoted = append(quoted, `"`+strings.NewReplacer(`\`, `\\`, `"`, `\"`).Replace(v)+`"`)
+	}
+	return "{" + strings.Join(quoted, ",") + "}"
 }
 
 // dropLegacyMatViews removes matviews from prior schema versions that no
@@ -513,9 +599,10 @@ func matViewNeedsRebuild(ctx context.Context, conn *sql.Conn, view string, requi
 	if err := conn.QueryRowContext(ctx, `
 		SELECT EXISTS (
 			SELECT 1
-			FROM pg_class
-			WHERE relkind = 'm'
-			  AND relname = $1
+			FROM pg_class c
+			WHERE c.relkind = 'm'
+			  AND c.relname = $1
+			  AND pg_catalog.pg_table_is_visible(c.oid)
 		)
 	`, view).Scan(&exists); err != nil {
 		return false, fmt.Errorf("failed to check matview %s existence: %w", view, err)
@@ -530,6 +617,7 @@ func matViewNeedsRebuild(ctx context.Context, conn *sql.Conn, view string, requi
 		JOIN pg_attribute a ON a.attrelid = c.oid
 		WHERE c.relkind = 'm'
 		  AND c.relname = $1
+		  AND pg_catalog.pg_table_is_visible(c.oid)
 		  AND a.attnum > 0
 		  AND NOT a.attisdropped
 	`, view)
@@ -763,8 +851,16 @@ func startMatViewRefresher(ctx context.Context, db *gorm.DB, interval time.Durat
 				if err := refreshMatViews(ctx, db); err != nil {
 					logger.Warn(fmt.Sprintf("logstore: matview refresh failed: %s", err))
 				} else if readyFlag != nil && !readyFlag.Load() {
-					logger.Info("logstore: materialized views are ready (recovered)")
-					readyFlag.Store(true)
+					// A successful refresh is not evidence of the right shape:
+					// REFRESH works fine on an old-shape view, and refreshMatViews
+					// also returns nil when it skipped. Check the catalog.
+					shapesOK, err := matViewShapesReady(ctx, db)
+					if err != nil {
+						logger.Warn(fmt.Sprintf("logstore: matview shape check failed: %s (dashboard queries will use raw tables)", err))
+					} else if shapesOK {
+						logger.Info("logstore: materialized views are ready (recovered)")
+						readyFlag.Store(true)
+					}
 				}
 			case <-ctx.Done():
 				return

framework/logstore/matviews_lock_test.go

@@ -63,11 +63,15 @@ func TestEnsureMatViewsSharesRefreshAdvisoryLock(t *testing.T) {
 	require.False(t, testMatViewExists(t, db, "mv_filter_users"))
 
 	holder := acquireTestAdvisoryLock(t, db, matviewRefreshAdvisoryLockKey)
-	require.NoError(t, ensureMatViews(ctx, db))
+	maintained, err := ensureMatViews(ctx, db)
+	require.NoError(t, err)
+	require.False(t, maintained, "ensureMatViews should report it did not maintain the views while the lock is held elsewhere")
 	require.False(t, testMatViewExists(t, db, "mv_filter_users"), "ensureMatViews should skip while refresh lock is held elsewhere")
 
 	releaseTestAdvisoryLock(t, holder, matviewRefreshAdvisoryLockKey)
-	require.NoError(t, ensureMatViews(ctx, db))
+	maintained, err = ensureMatViews(ctx, db)
+	require.NoError(t, err)
+	require.True(t, maintained, "ensureMatViews should report it maintained the views once the lock is free")
 	require.True(t, testMatViewExists(t, db, "mv_filter_users"))
 }
 

framework/logstore/matviews_shape_test.go

@@ -0,0 +1,78 @@
+package logstore
+
+import (
+	"regexp"
+	"strings"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+var (
+	// aliasedSelectItemRe matches a select item that names its output column,
+	// e.g. `COUNT(*) AS count,` -> count.
+	aliasedSelectItemRe = regexp.MustCompile(`(?i)\bAS[ \t]+([a-z0-9_]+)$`)
+	// bareSelectItemRe matches a select item that is just a column reference,
+	// e.g. `provider,` -> provider. Those carry their own name as the output.
+	bareSelectItemRe = regexp.MustCompile(`^[a-z0-9_]+$`)
+)
+
+// mvLogsHourlyOutputColumns returns the output column of every select item in
+// mvLogsHourlyDDL. Each item occupies one line, so the select list is parsed
+// line-wise between SELECT and FROM. Anything unrecognized fails the test
+// rather than being skipped — a silently-dropped item would weaken the check
+// this test exists to enforce.
+func mvLogsHourlyOutputColumns(t *testing.T) map[string]struct{} {
+	t.Helper()
+
+	_, body, found := strings.Cut(mvLogsHourlyDDL, "SELECT")
+	require.True(t, found, "DDL should contain SELECT")
+	body, _, found = strings.Cut(body, "FROM logs")
+	require.True(t, found, "DDL should contain FROM logs")
+
+	columns := make(map[string]struct{})
+	for _, line := range strings.Split(body, "\n") {
+		item := strings.TrimSpace(line)
+		item = strings.TrimSuffix(item, ",")
+		item = strings.TrimSpace(item)
+		if item == "" || strings.HasPrefix(item, "--") {
+			continue
+		}
+		if m := aliasedSelectItemRe.FindStringSubmatch(item); m != nil {
+			columns[m[1]] = struct{}{}
+			continue
+		}
+		if bareSelectItemRe.MatchString(item) {
+			columns[item] = struct{}{}
+			continue
+		}
+		t.Fatalf("could not determine the output column of select item %q — update this parser", item)
+	}
+	return columns
+}
+
+// TestMvLogsHourlyRequiredColumnsMatchDDL pins mvLogsHourlyRequiredColumns to
+// the DDL. The list gates two things — whether repairMatViewShapes rebuilds a
+// drifted view, and whether matViewShapesReady lets readers onto the matview
+// path — so a column in the DDL but not the list means a view lacking it passes
+// both checks and readers fail with "column does not exist". Adding a column to
+// the view without adding it here should fail loudly.
+func TestMvLogsHourlyRequiredColumnsMatchDDL(t *testing.T) {
+	inDDL := mvLogsHourlyOutputColumns(t)
+	require.NotEmpty(t, inDDL)
+
+	required := make(map[string]struct{}, len(mvLogsHourlyRequiredColumns))
+	for _, c := range mvLogsHourlyRequiredColumns {
+		required[c] = struct{}{}
+	}
+
+	for column := range inDDL {
+		assert.Containsf(t, required, column,
+			"mv_logs_hourly selects %q but mvLogsHourlyRequiredColumns omits it: a view missing this column would pass the shape gate and readers would fail with \"column does not exist\"", column)
+	}
+	for column := range required {
+		assert.Containsf(t, inDDL, column,
+			"mvLogsHourlyRequiredColumns lists %q but mv_logs_hourly does not select it: the shape gate would never be satisfied", column)
+	}
+}

framework/logstore/migrations_scale_test.go

@@ -80,7 +80,8 @@ func TestScalePostgresLogstoreMigrations(t *testing.T) {
 	indexElapsed := time.Since(start)
 
 	start = time.Now()
-	require.NoError(t, ensureMatViews(ctx, db), "matviews should be maintained at scale")
+	_, err = ensureMatViews(ctx, db)
+	require.NoError(t, err, "matviews should be maintained at scale")
 	matviewElapsed := time.Since(start)
 
 	stopMonitor()

framework/logstore/postgres.go

@@ -199,10 +199,26 @@ func newPostgresLogStore(ctx context.Context, config *PostgresConfig, logger sch
 		if db.Dialector.Name() != "postgres" {
 			return
 		}
-		if err := ensureMatViews(context.Background(), db); err != nil {
+		maintained, err := ensureMatViews(context.Background(), db)
+		if err != nil {
 			logger.Warn(fmt.Sprintf("logstore: matview creation failed: %s (dashboard queries will use raw tables)", err))
 			return
 		}
+		if !maintained {
+			// Another replica owns the create/repair and its rebuild may not have
+			// landed, so the views we see can still be an older schema version.
+			// Confirm the shape before enabling the read path; if it isn't current
+			// the refresher flips the flag once it is.
+			shapesOK, err := matViewShapesReady(context.Background(), db)
+			if err != nil {
+				logger.Warn(fmt.Sprintf("logstore: matview shape check failed: %s (dashboard queries will use raw tables)", err))
+			}
+			if err != nil || !shapesOK {
+				logger.Info("logstore: matview maintenance is owned by another replica and views are not current yet (dashboard queries will use raw tables until they are)")
+				startMatViewRefresher(context.Background(), db, resolveMatViewRefreshInterval(config.MatViewRefreshInterval, logger), logger, &d.matViewsReady)
+				return
+			}
+		}
 		if err := refreshMatViews(context.Background(), db); err != nil {
 			logger.Warn(fmt.Sprintf("logstore: initial matview refresh failed: %s", err))
 		} else {

framework/logstore/rdb_postgres_perf_test.go

@@ -39,7 +39,7 @@ func setupPerfTestDB(t *testing.T) (*RDBLogStore, *gorm.DB) {
 	err := triggerMigrations(ctx, db, testLogger{})
 	require.NoError(t, err, "migrations should succeed")
 
-	err = ensureMatViews(ctx, db)
+	_, err = ensureMatViews(ctx, db)
 	require.NoError(t, err, "matview creation should succeed")
 
 	store := &RDBLogStore{db: db}
@@ -105,7 +105,8 @@ func TestEnsureMatViewsRebuildsBadSameNameIndex(t *testing.T) {
 	require.NoError(t, db.Exec("CREATE INDEX mv_logs_hourly_uniq ON mv_logs_hourly(hour)").Error)
 	require.False(t, matviewIndexReady(t, db, "mv_logs_hourly", "mv_logs_hourly_uniq"), "same-name non-unique index should not be considered ready")
 
-	require.NoError(t, ensureMatViews(ctx, db))
+	_, err := ensureMatViews(ctx, db)
+	require.NoError(t, err)
 
 	assert.True(t, matviewIndexReady(t, db, "mv_logs_hourly", "mv_logs_hourly_uniq"), "ensureMatViews should rebuild the required unique index")
 	for _, v := range filterMatViews {