perf(dq): SORT rollups by subject for per-vehicle read pruning (scale review)

internal/materializer/ducklake.go

@@ -353,9 +353,90 @@ func (m *DuckLakeMaterializer) ensureSchema(ctx context.Context) error {
 	if eventsPending {
 		m.eventRollupFullRebuild = true
 	}
+	// #4: migrate a pre-existing rollup that was created partitioned-but-unsorted to
+	// SORTED BY (subject). Crash-safe + idempotent (gated on the sort being absent).
+	if err := m.migrateLatestSort(ctx); err != nil {
+		return err
+	}
+	return nil
+}
+
+// migrateLatestSort applies SORTED BY (subject) to a pre-#4 rollup table that was created
+// partitioned-but-unsorted, so per-vehicle reads zone-map-prune within the subject_bucket
+// instead of scanning it whole (scale review #4). Crash-safe and idempotent: the ALTER is
+// issued ONLY when the sort spec is currently absent (re-derived from the catalog each
+// boot), so a normal reboot mints zero schema changes — the same discipline the first-create
+// ALTER gating enforces (re-ALTERing an active spec is the schema-version-churn hazard the
+// gating avoids). Existing data files re-sort lazily as the rollup rewrites/merges them; new
+// writes are sorted immediately.
+//
+// OPERATIONAL NOTE (adversarial review): the ALTER bumps the DuckLake schema_version and
+// renames the inline-data tables, so on the ONE upgrade boot it opens the same "din
+// maintenance flush_inlined_data under a concurrent read → FATAL/instance invalidation"
+// window the layout gating exists to avoid. The materializer self-heals (isDatabaseInvalidated
+// / recoverPoisonedSession → restart), but QUERY pods have no in-process recovery and heal
+// only via K8s liveness — so a brief read blip on signals_latest is possible during the
+// upgrade boot. It is one-time (the gate skips forever after), metadata-only (no data
+// rewrite), and best-effort (a failure stays unsorted, retried next boot). Prefer deploying
+// this materializer release during a low-traffic / low-maintenance window.
+func (m *DuckLakeMaterializer) migrateLatestSort(ctx context.Context) error {
+	// Best-effort: SORTED BY is a read-pruning optimization, not a correctness invariant, so
+	// a transient catalog/metadata hiccup here must NOT block the single writer from booting
+	// (that would halt all decode). Failures are logged and retried on the next boot; the
+	// table just stays unsorted (the pre-#4 status quo) meanwhile. The first-create sort in
+	// setupStatements is still enforced as part of schema creation.
+	for _, table := range []string{"signals_latest", "events_latest"} {
+		if ctx.Err() != nil {
+			return ctx.Err()
+		}
+		exists, err := m.tableExists(ctx, "lake", table)
+		if err != nil {
+			m.log.Warn().Err(err).Str("table", table).Msg("sort migration: table-exists check failed; skipping this boot (#4)")
+			continue
+		}
+		if !exists {
+			continue
+		}
+		sorted, err := m.latestIsSortedBySubject(ctx, table)
+		if err != nil {
+			m.log.Warn().Err(err).Str("table", table).Msg("sort migration: sort-spec check failed; skipping this boot (#4)")
+			continue
+		}
+		if sorted {
+			continue
+		}
+		stmt := fmt.Sprintf(`ALTER TABLE lake.%s SET SORTED BY (subject)`, table)
+		var aerr error
+		for attempt := 0; attempt < 5; attempt++ {
+			if _, aerr = m.db.ExecContext(ctx, stmt); aerr == nil || !isCommitConflict(aerr) {
+				break
+			}
+			select {
+			case <-ctx.Done():
+				return ctx.Err()
+			case <-time.After(time.Duration(attempt+1) * 100 * time.Millisecond):
+			}
+		}
+		if aerr != nil {
+			m.log.Warn().Err(aerr).Str("table", table).Msg("sort migration: SET SORTED BY failed; staying unsorted, will retry next boot (#4)")
+			continue
+		}
+		m.log.Info().Str("table", table).Msg("migrated rollup to SORTED BY (subject) (#4)")
+	}
 	return nil
 }
 
+// latestIsSortedBySubject reports whether table has an ACTIVE SORTED BY spec (a
+// ducklake_sort_info row with end_snapshot IS NULL), mirroring din's isSorted probe.
+func (m *DuckLakeMaterializer) latestIsSortedBySubject(ctx context.Context, table string) (bool, error) {
+	var ok bool
+	err := m.db.QueryRowContext(ctx, `
+		SELECT count(*) > 0 FROM __ducklake_metadata_lake.ducklake_sort_info
+		WHERE end_snapshot IS NULL
+		  AND table_id = (SELECT table_id FROM ducklake_table_info('lake') WHERE table_name = ?)`, table).Scan(&ok)
+	return ok, err
+}
+
 // tableExists reports whether schema.name is already in the attached catalog.
 // Used to decide whether a decoded table needs its first-creation layout applied.
 func (m *DuckLakeMaterializer) tableExists(ctx context.Context, schema, name string) (bool, error) {
@@ -437,6 +518,11 @@ func setupStatements(exists map[string]bool, sigTmp, evTmp string) []string {
 				loc_ts TIMESTAMP WITH TIME ZONE,
 				count BIGINT, first_seen TIMESTAMP WITH TIME ZONE, last_seen TIMESTAMP WITH TIME ZONE)`,
 			`ALTER TABLE lake.signals_latest SET PARTITIONED BY (subject_bucket)`,
+			// SORTED BY (subject) clusters each subject_bucket file by subject so a per-vehicle
+			// latest/dataSummary/location read prunes to a few row groups via parquet zone maps
+			// instead of scanning the whole bucket (~fleet/256 rows) — scale review #4. Existing
+			// catalogs are migrated by migrateLatestSort (below), gated on the sort being absent.
+			`ALTER TABLE lake.signals_latest SET SORTED BY (subject)`,
 		)
 	} else if !exists["signals_latest.loc_ts"] {
 		// Migrate catalogs whose rollup predates loc_ts (H9). Emitted only
@@ -464,6 +550,8 @@ func setupStatements(exists map[string]bool, sigTmp, evTmp string) []string {
 				count BIGINT,
 				first_seen TIMESTAMP WITH TIME ZONE, last_seen TIMESTAMP WITH TIME ZONE)`,
 			`ALTER TABLE lake.events_latest SET PARTITIONED BY (subject_bucket)`,
+			// See signals_latest above (#4); existing catalogs migrated by migrateLatestSort.
+			`ALTER TABLE lake.events_latest SET SORTED BY (subject)`,
 		)
 	}
 	// The remaining statements are genuinely idempotent (CREATE IF NOT EXISTS /

tests/ducklake_latest_sort_test.go

@@ -0,0 +1,67 @@
+// ducklake_latest_sort_test.go pins scale-review #4: signals_latest / events_latest must
+// be SORTED BY (subject) so a per-vehicle latest/dataSummary/location read prunes to a few
+// row groups within its subject_bucket instead of scanning the whole bucket. Verified on a
+// fresh catalog (first-create) and migrated onto an existing unsorted rollup (prod upgrade).
+package tests
+
+import (
+	"context"
+	"database/sql"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// latestSortedBySubject reports whether table currently has an ACTIVE SORTED BY spec
+// (a ducklake_sort_info row with end_snapshot IS NULL), mirroring din's isSorted probe.
+func latestSortedBySubject(t *testing.T, ctx context.Context, db *sql.DB, table string) bool {
+	t.Helper()
+	var ok bool
+	require.NoError(t, db.QueryRowContext(ctx, `
+		SELECT count(*) > 0 FROM __ducklake_metadata_lake.ducklake_sort_info
+		WHERE end_snapshot IS NULL
+		  AND table_id = (SELECT table_id FROM ducklake_table_info('lake') WHERE table_name = ?)`, table).Scan(&ok))
+	return ok
+}
+
+func TestDuckLake_LatestTablesSortedOnCreate(t *testing.T) {
+	ctx := context.Background()
+	svc := newLakeService(t, t.TempDir())
+	db := svc.DB()
+	mustMat(t, ctx, db) // ensureSchema creates the rollups
+	assert.True(t, latestSortedBySubject(t, ctx, db, "signals_latest"), "signals_latest must be SORTED BY (subject) on create")
+	assert.True(t, latestSortedBySubject(t, ctx, db, "events_latest"), "events_latest must be SORTED BY (subject) on create")
+}
+
+func TestDuckLake_LatestTablesSortMigration(t *testing.T) {
+	ctx := context.Background()
+	svc := newLakeService(t, t.TempDir())
+	db := svc.DB()
+	// Simulate a pre-#4 catalog: the rollups exist, partitioned but UNSORTED.
+	for _, ddl := range []string{
+		`CREATE TABLE lake.signals_latest (
+			subject VARCHAR, subject_bucket INTEGER, name VARCHAR, "timestamp" TIMESTAMP WITH TIME ZONE,
+			value_number DOUBLE, value_string VARCHAR, loc_lat DOUBLE, loc_lon DOUBLE, loc_hdop DOUBLE, loc_heading DOUBLE,
+			loc_ts TIMESTAMP WITH TIME ZONE, count BIGINT, first_seen TIMESTAMP WITH TIME ZONE, last_seen TIMESTAMP WITH TIME ZONE)`,
+		`ALTER TABLE lake.signals_latest SET PARTITIONED BY (subject_bucket)`,
+		`CREATE TABLE lake.events_latest (
+			subject VARCHAR, subject_bucket INTEGER, name VARCHAR, count BIGINT,
+			first_seen TIMESTAMP WITH TIME ZONE, last_seen TIMESTAMP WITH TIME ZONE)`,
+		`ALTER TABLE lake.events_latest SET PARTITIONED BY (subject_bucket)`,
+	} {
+		_, err := db.ExecContext(ctx, ddl)
+		require.NoError(t, err, ddl)
+	}
+	require.False(t, latestSortedBySubject(t, ctx, db, "signals_latest"), "precondition: unsorted")
+	require.False(t, latestSortedBySubject(t, ctx, db, "events_latest"), "precondition: unsorted")
+
+	// A fresh materializer boot must migrate the existing unsorted rollups.
+	mustMat(t, ctx, db)
+	assert.True(t, latestSortedBySubject(t, ctx, db, "signals_latest"), "migration must sort an existing unsorted signals_latest")
+	assert.True(t, latestSortedBySubject(t, ctx, db, "events_latest"), "migration must sort an existing unsorted events_latest")
+
+	// Idempotent: a second boot must not error (re-derives from catalog; no re-ALTER churn).
+	mustMat(t, ctx, db)
+	assert.True(t, latestSortedBySubject(t, ctx, db, "signals_latest"))
+}