perf(cdc-graph): bound cdc_batches scan and add composite index

The CDCGraph query range-joined generate_series ticks against
peerdb_stats.cdc_batches with no global lower bound on start_time, so the
planner could scan every batch for the flow across all history (via the
flow_name hash index) just to render a ~30-point graph.

Rewrite to pull the flow's batches for the window into a bounded CTE
(single scan) and keep the range join over that small set. Semantics are
unchanged, including calendar-correct month stepping.

Add a composite (flow_name, start_time) index so the bounded scan is a
single tight index range scan.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: dtunikov

flow/cmd/mirror_status.go

@@ -253,17 +253,22 @@ func getGraphParams(aggType protos.TimeAggregateType, mode protos.GraphMode) (st
 func (h *FlowRequestHandler) CDCGraph(ctx context.Context, req *protos.GraphRequest) (*protos.GraphResponse, APIError) {
 	truncUnit, tickInterval, numberOfTicks := getGraphParams(req.AggregateType, req.Mode)
 	rows, err := h.pool.Query(ctx, `
-		SELECT tm, COALESCE(SUM(rows_in_batch), 0)
-		FROM generate_series(
-			date_trunc($2, now() - $1::INTERVAL * $3),
-			now(),
-			$1::INTERVAL
-		) tm
-		LEFT JOIN peerdb_stats.cdc_batches
-			ON start_time >= tm AND start_time < tm + $1::INTERVAL
-			AND flow_name = $4
-		GROUP BY 1
-		ORDER BY 1
+		WITH lower_bound AS (
+			SELECT date_trunc($2, now() - $1::INTERVAL * $3) AS lb
+		),
+		batches AS (
+			SELECT start_time, rows_in_batch
+			FROM peerdb_stats.cdc_batches, lower_bound
+			WHERE flow_name = $4
+				AND start_time >= lower_bound.lb
+		)
+		SELECT tm, COALESCE(SUM(b.rows_in_batch), 0)
+		FROM lower_bound,
+			generate_series(lower_bound.lb, now(), $1::INTERVAL) tm
+		LEFT JOIN batches b
+			ON b.start_time >= tm AND b.start_time < tm + $1::INTERVAL
+		GROUP BY tm
+		ORDER BY tm
 	`, tickInterval, truncUnit, numberOfTicks, req.FlowJobName)
 	if err != nil {
 		return nil, NewInternalApiError(fmt.Errorf("failed to query cdc graph: %w", err))

nexus/catalog/migrations/V55__cdc_batches_flow_name_start_time_index.sql

@@ -0,0 +1 @@
+CREATE INDEX IF NOT EXISTS idx_cdc_batches_flow_name_start_time ON peerdb_stats.cdc_batches (flow_name, start_time);