From 433ad0234ae8b7b3606e26417877ff691f2fb55e Mon Sep 17 00:00:00 2001 From: MrAlders0n <55921894+MrAlders0n@users.noreply.github.com> Date: Tue, 20 Oct 2026 18:47:08 -0400 Subject: [PATCH] Observer comparison: read each observer's receptions on their own (#260) * Read each observer's receptions on their own in the observer comparison * Tighten the observer comparison plan checks --- .github/workflows/ci.yml | 2 +- ...server_comparison_plan_integration_test.go | 226 ++++++++++++++++++ .../fixture/obscmp.busy_24h_all.custom.json | 13 + .../fixture/obscmp.busy_24h_site.custom.json | 13 + .../fixture/obscmp.busy_7d_all.custom.json | 13 + .../obscmp.busy_7d_empty_region.custom.json | 13 + .../fixture/obscmp.busy_7d_region.custom.json | 13 + .../fixture/obscmp.mixed_7d_all.custom.json | 13 + .../fixture/obscmp.quiet_7d_all.custom.json | 13 + .../fixture/obscmp.unknown_7d.custom.json | 13 + .../local/obscmp.busy_24h_all.custom.json | 13 + .../local/obscmp.busy_24h_site.custom.json | 13 + .../local/obscmp.busy_7d_all.custom.json | 13 + .../obscmp.busy_7d_empty_region.custom.json | 13 + .../local/obscmp.busy_7d_region.custom.json | 13 + .../local/obscmp.mixed_7d_all.custom.json | 13 + .../local/obscmp.quiet_7d_all.custom.json | 13 + .../local/obscmp.unknown_7d.custom.json | 13 + db/planguard/corpus.yaml | 88 +++++++ db/queries/observer_comparison.sql | 36 ++- db/sqlc/observer_comparison.sql.go | 34 ++- db/sqlc/querier.go | 9 + db/statement_capture_test.go | 7 + 23 files changed, 592 insertions(+), 18 deletions(-) create mode 100644 db/observer_comparison_plan_integration_test.go create mode 100644 db/planguard/baseline/fixture/obscmp.busy_24h_all.custom.json create mode 100644 db/planguard/baseline/fixture/obscmp.busy_24h_site.custom.json create mode 100644 db/planguard/baseline/fixture/obscmp.busy_7d_all.custom.json create mode 100644 db/planguard/baseline/fixture/obscmp.busy_7d_empty_region.custom.json create mode 100644 db/planguard/baseline/fixture/obscmp.busy_7d_region.custom.json create mode 100644 db/planguard/baseline/fixture/obscmp.mixed_7d_all.custom.json create mode 100644 db/planguard/baseline/fixture/obscmp.quiet_7d_all.custom.json create mode 100644 db/planguard/baseline/fixture/obscmp.unknown_7d.custom.json create mode 100644 db/planguard/baseline/local/obscmp.busy_24h_all.custom.json create mode 100644 db/planguard/baseline/local/obscmp.busy_24h_site.custom.json create mode 100644 db/planguard/baseline/local/obscmp.busy_7d_all.custom.json create mode 100644 db/planguard/baseline/local/obscmp.busy_7d_empty_region.custom.json create mode 100644 db/planguard/baseline/local/obscmp.busy_7d_region.custom.json create mode 100644 db/planguard/baseline/local/obscmp.mixed_7d_all.custom.json create mode 100644 db/planguard/baseline/local/obscmp.quiet_7d_all.custom.json create mode 100644 db/planguard/baseline/local/obscmp.unknown_7d.custom.json diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index b5e712b..08ad585 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -69,7 +69,7 @@ jobs: - name: Verify stats and endpoint queries against PostgreSQL 16 env: BEACON_TEST_POSTGRES_DSN: postgres://postgres:backup-ci-only@127.0.0.1:5432/postgres?sslmode=disable - run: go test ./db -run '^Test(Signal|Paths|PacketSummaries|PacketEndpointsResolveLive|ObserverMetrics|RouteEvidence|RouteEvidenceIndex|AnalyticsRetention|AnalyticsRetentionConcurrent|DeleteOldPacketsBatches|MeshMapperCatalogue|ChannelMessageScopes|NodeLocationReset|RunMigrations[A-Za-z]*|KnownRoutesIndexSet|PacketsDomainChecks|Reconfirm[A-Za-z]*|BigintCursors|RouteSearchHopOrder|StatsNodeTypesSplit|ListKnownRoutes|ListKnownRoutesMatchesLegacy|KnownRoutesCursorTies|KnownRoutesUnanalyzedRegion|KnownRoutesHopStatementsStayCustom|KnownRoutePaginationTies|RouteRetentionCutoffsExact|DeleteExpiredRoutes|DeleteGraceRoutesSlice|GraceWatermarkHeldByLockedRow|OpenPoolsSettings|PlanRouterClassCUnnamed|APIStoreHopStatementsUnnamed|PgStatStatementsMigrationGuarded|DBStatements|DBStatementsNoExtension|PgStatStatementsMigrationNoCreatableSchema|IndexHealthBaselineOnReindex|UnusedIndexes|PgActivity|RollHourSetsWorkMem|SmallTableIndexSet|NodeReadersAfterIndexDrops|ObserverChannelReadersAfterIndexDrops|UpsertChannelWriteThrough|UpsertChannelHashOnlyWriteThrough|ReadOnlyPoolsRejectWrites|ListPacketsGlobalMatchesLegacy|PacketsGlobalCursorTies|ListPacketsWindowInvertedRows|PacketsLatestObserverTie|NodesLegacyCursorMatchesToday|NodesPageCursorTies|ObserversLegacyCursorMatchesToday|ObserversNextCursorFromLastSeen|ObserverAdvertsPartialIndex|AdvertsWaitForPayloadTypeBackfill|StatsProbeMatchesToday|TraceTagsKeysetMatchesToday|TelemetrySinceMatchesCatchAll|TraceTagPagination|TraceTagPaginationTies|TraceSummarySurvivesMissedHearing|CmiMigrationsIndexSet|InsertChannelMessageSeed[A-Za-z]*|AddChannelMessageIATA|CmiSeedAndAddRace|CmiInitAndBackfill|CmiMessageRangeReconcile|CmiObservationRangeReconcile|CmiBackfillRetries23503|WaitForPreBootWriters|ChannelMessagesByIATAsMatchesLegacy|ChannelMessagesZeroHitRegion|CmiTrustLifecycle|CmiFirstVerification|PacketSitesMigrationsIndexSet|InsertObservationMaintainsPacketSites|InsertObservationResendWritesNothing|PacketSitesRaceMaxMerge|InsertObservationSiteFallback|PacketSitesInitAndBackfill|FillPacketSitesSliceIdempotent|FillPacketSitesSliceRetries23503|PacketSitesReconcile|PacketSitesAudit|PacketSitesAuditTakesNewestHearing|PacketsBySitesMatchesTruth|PacketsBySitesCursorTies|PacketsBySitesManyBogusSites|PacketsBySitesSaturatedPagesFull|PacketsBySitesMatchesLegacy|PacketSitesTrustLifecycle|PacketsBySitesBounded|PacketsBySitesFallbackKeepsPageCursor|SweepPacketSitesFrontsPlan|IndexHealthPacketSitesFreshBaseline|PacketSitesFirstVerification|PacedBackfillSlices|FillSessionsPlanCustom|FillAnalyzeFailureIsIgnored|FillAnalyzeOnlyNeverAnalyzed|HygieneIndexSet|HygieneReadersAfterIndexDrops|KnownRoutesColumns|SequenceUse|SlowPoolIsolation|SlowPoolSize|SequenceUseMixedCaseSchema|ObservationsIndexSet|ReadersAfterIndexDrops)Postgres$' -count=1 -v + run: go test ./db -run '^Test(Signal|Paths|PacketSummaries|PacketEndpointsResolveLive|ObserverMetrics|RouteEvidence|RouteEvidenceIndex|AnalyticsRetention|AnalyticsRetentionConcurrent|DeleteOldPacketsBatches|MeshMapperCatalogue|ChannelMessageScopes|NodeLocationReset|RunMigrations[A-Za-z]*|KnownRoutesIndexSet|PacketsDomainChecks|Reconfirm[A-Za-z]*|BigintCursors|RouteSearchHopOrder|StatsNodeTypesSplit|ListKnownRoutes|ListKnownRoutesMatchesLegacy|KnownRoutesCursorTies|KnownRoutesUnanalyzedRegion|KnownRoutesHopStatementsStayCustom|KnownRoutePaginationTies|RouteRetentionCutoffsExact|DeleteExpiredRoutes|DeleteGraceRoutesSlice|GraceWatermarkHeldByLockedRow|OpenPoolsSettings|PlanRouterClassCUnnamed|APIStoreHopStatementsUnnamed|PgStatStatementsMigrationGuarded|DBStatements|DBStatementsNoExtension|PgStatStatementsMigrationNoCreatableSchema|IndexHealthBaselineOnReindex|UnusedIndexes|PgActivity|RollHourSetsWorkMem|SmallTableIndexSet|NodeReadersAfterIndexDrops|ObserverChannelReadersAfterIndexDrops|UpsertChannelWriteThrough|UpsertChannelHashOnlyWriteThrough|ReadOnlyPoolsRejectWrites|ListPacketsGlobalMatchesLegacy|PacketsGlobalCursorTies|ListPacketsWindowInvertedRows|PacketsLatestObserverTie|NodesLegacyCursorMatchesToday|NodesPageCursorTies|ObserversLegacyCursorMatchesToday|ObserversNextCursorFromLastSeen|ObserverAdvertsPartialIndex|ObserverComparisonMatchesToday|AdvertsWaitForPayloadTypeBackfill|StatsProbeMatchesToday|TraceTagsKeysetMatchesToday|TelemetrySinceMatchesCatchAll|TraceTagPagination|TraceTagPaginationTies|TraceSummarySurvivesMissedHearing|CmiMigrationsIndexSet|InsertChannelMessageSeed[A-Za-z]*|AddChannelMessageIATA|CmiSeedAndAddRace|CmiInitAndBackfill|CmiMessageRangeReconcile|CmiObservationRangeReconcile|CmiBackfillRetries23503|WaitForPreBootWriters|ChannelMessagesByIATAsMatchesLegacy|ChannelMessagesZeroHitRegion|CmiTrustLifecycle|CmiFirstVerification|PacketSitesMigrationsIndexSet|InsertObservationMaintainsPacketSites|InsertObservationResendWritesNothing|PacketSitesRaceMaxMerge|InsertObservationSiteFallback|PacketSitesInitAndBackfill|FillPacketSitesSliceIdempotent|FillPacketSitesSliceRetries23503|PacketSitesReconcile|PacketSitesAudit|PacketSitesAuditTakesNewestHearing|PacketsBySitesMatchesTruth|PacketsBySitesCursorTies|PacketsBySitesManyBogusSites|PacketsBySitesSaturatedPagesFull|PacketsBySitesMatchesLegacy|PacketSitesTrustLifecycle|PacketsBySitesBounded|PacketsBySitesFallbackKeepsPageCursor|SweepPacketSitesFrontsPlan|IndexHealthPacketSitesFreshBaseline|PacketSitesFirstVerification|PacedBackfillSlices|FillSessionsPlanCustom|FillAnalyzeFailureIsIgnored|FillAnalyzeOnlyNeverAnalyzed|HygieneIndexSet|HygieneReadersAfterIndexDrops|KnownRoutesColumns|SequenceUse|SlowPoolIsolation|SlowPoolSize|SequenceUseMixedCaseSchema|ObservationsIndexSet|ReadersAfterIndexDrops)Postgres$' -count=1 -v - name: Verify the channel message site rule against PostgreSQL 16 env: diff --git a/db/observer_comparison_plan_integration_test.go b/db/observer_comparison_plan_integration_test.go new file mode 100644 index 0000000..eb1176f --- /dev/null +++ b/db/observer_comparison_plan_integration_test.go @@ -0,0 +1,226 @@ +// Copyright 2026 Beacon Contributors +// SPDX-License-Identifier: AGPL-3.0-or-later + +package db + +import ( + "encoding/json" + "fmt" + "strings" + "testing" + "time" + + sqlc "github.com/MeshCore-Beacon/beacon-server/db/sqlc" + "github.com/google/uuid" + "github.com/jackc/pgx/v5/pgtype" +) + +// legacyGetObserverComparison is the comparison before each observer's receptions were read on their own, +// verbatim from the generated db/sqlc/observer_comparison.sql.go. The statement changed in place. +const legacyGetObserverComparison = `-- name: GetObserverComparison :one +WITH heard AS ( + SELECT po.packet_hash, + bool_or(po.observer_id = $1::uuid) AS seen_a, + bool_or(po.observer_id = $2::uuid) AS seen_b + FROM packet_observations po + JOIN packets p ON p.packet_hash = po.packet_hash + WHERE po.observer_id IN ($1::uuid, $2::uuid) + AND po.heard_at >= $3::timestamptz + AND po.heard_at < $4::timestamptz + AND p.route_type IN (0, 1) + AND (COALESCE(cardinality($5::text[]), 0) = 0 OR po.iata = ANY($5::text[])) + GROUP BY po.packet_hash +) +SELECT count(*) AS total_packets, + count(*) FILTER (WHERE seen_a AND NOT seen_b) AS only_a, + count(*) FILTER (WHERE seen_b AND NOT seen_a) AS only_b, + count(*) FILTER (WHERE seen_a AND seen_b) AS both, + EXISTS (SELECT 1 FROM observers WHERE id = $1::uuid) AS known_a, + EXISTS (SELECT 1 FROM observers WHERE id = $2::uuid) AS known_b +FROM heard +` + +var comparisonBase = time.Date(2026, 10, 1, 0, 0, 0, 0, time.UTC) + +// comparisonFixture: 2,000 packets of every route type, one every five minutes from comparisonBase (about seven +// days), heard by eight observers with skewed rates (90 % down to 2 %) at a per-observer offset of whole seconds, so +// many receptions share an instant. 100,000 older packets that nobody heard make reading all of packets cost what it +// does at size. Observer 1 hears at YVR and sometimes YYJ, observer 4 alternates YVR and YYZ, +// observers 5-8 hear at YOW. Observer 9 is known and never hears anything. +var comparisonFixture = []string{` +INSERT INTO observers (id, public_key) +SELECT ('00000000-0000-0000-0000-' || lpad(i::text, 12, '0'))::uuid, decode(lpad(to_hex(i), 64, '0'), 'hex') +FROM generate_series(1, 9) i WHERE $1::timestamptz IS NOT NULL`, ` +INSERT INTO packets (packet_hash, payload_type, payload_version, route_type, raw_payload, raw_header, first_heard_at, last_heard_at) +SELECT decode(md5('cmp-' || i), 'hex'), i % 12, 0, (ARRAY[0, 1, 1, 0, 2, 1, 3, 0])[i % 8 + 1], '\x00', '\x00', + $1::timestamptz + i * interval '5 minutes', $1::timestamptz + i * interval '5 minutes' +FROM generate_series(1, 2000) i`, ` +INSERT INTO packets (packet_hash, payload_type, payload_version, route_type, raw_payload, raw_header, first_heard_at, last_heard_at) +SELECT decode(md5('cmp-unheard-' || i), 'hex'), i % 12, 0, i % 4, '\x00', '\x00', + $1::timestamptz - interval '30 days' + i * interval '1 second', $1::timestamptz - interval '30 days' + i * interval '1 second' +FROM generate_series(1, 100000) i`, ` +INSERT INTO packet_observations (packet_hash, observer_id, iata, heard_at, path_length_byte, hash_size, hop_count, source_broker) +SELECT decode(md5('cmp-' || i), 'hex'), + ('00000000-0000-0000-0000-' || lpad(o::text, 12, '0'))::uuid, + CASE WHEN o = 1 AND i % 7 = 0 THEN 'YYJ' WHEN o IN (1, 2) THEN 'YVR' WHEN o = 3 THEN 'YYJ' + WHEN o = 4 THEN (ARRAY['YVR', 'YYZ'])[i % 2 + 1] ELSE 'YOW' END, + $1::timestamptz + i * interval '5 minutes' + (o % 3) * interval '1 second', + 0, 1, 0, 'a' +FROM generate_series(1, 2000) i +CROSS JOIN generate_series(1, 8) o +WHERE ('x' || substr(md5(i || '-' || o), 1, 8))::bit(32)::bigint % 100 < (ARRAY[90, 70, 40, 10, 2, 2, 2, 2])[o]`} + +// The comparison reads each observer's receptions from idx_observations_observer and returns exactly what the +// statement it replaced returned, in custom and generic plans, also with sequential scans off. +func TestObserverComparisonMatchesTodayPostgres(t *testing.T) { + ctx, tx := retentionTx(t) + applyBaseline(t, ctx, tx) + // Migration 046 drops it; nothing may lean on it. + mustExec(t, ctx, tx, "DROP INDEX idx_observations_observer_id") + for _, sql := range comparisonFixture { + if _, err := tx.Exec(ctx, sql, comparisonBase); err != nil { + t.Fatal(err) + } + } + mustExec(t, ctx, tx, "ANALYZE observers; ANALYZE packets; ANALYZE packet_observations") + mustExec(t, ctx, tx, "SET LOCAL jit = off") + + obs := func(i int) uuid.UUID { return uuid.MustParse(fmt.Sprintf("00000000-0000-0000-0000-%012d", i)) } + unknown := uuid.MustParse("00000000-0000-0000-0000-0000000000ff") + pairs := [][2]uuid.UUID{ + {obs(1), obs(2)}, {obs(2), obs(1)}, {obs(1), obs(4)}, {obs(3), obs(4)}, {obs(4), obs(5)}, + {obs(5), obs(6)}, {obs(1), obs(9)}, {obs(9), obs(5)}, {obs(1), obs(1)}, {obs(2), unknown}, + {unknown, obs(1)}, + } + // One reception instant, to put window edges exactly on receptions. + edge := comparisonBase.Add(500*5*time.Minute + time.Second) + ts := func(at time.Time) pgtype.Timestamptz { return pgtype.Timestamptz{Time: at, Valid: true} } + windows := [][2]time.Time{ + {comparisonBase.Add(-time.Hour), comparisonBase.Add(8 * 24 * time.Hour)}, + {comparisonBase.Add(24 * time.Hour), comparisonBase.Add(48 * time.Hour)}, + {edge, edge.Add(time.Hour)}, + {edge.Add(-time.Hour), edge}, + {edge, edge.Add(time.Nanosecond * 1000)}, + {comparisonBase.Add(30 * 24 * time.Hour), comparisonBase.Add(31 * 24 * time.Hour)}, + } + iataSets := [][]string{nil, {}, {"YVR"}, {"YYJ", "YVR", "YYJ"}, {"YOW"}, {"ZZZ"}, {""}, {"yvr"}} + + q := sqlc.New(tx) + settings := []string{"RESET enable_seqscan", "SET LOCAL enable_seqscan = off"} + compared, nonEmpty := 0, 0 + for _, setting := range settings { + for _, mode := range []string{"force_custom_plan", "force_generic_plan"} { + mustExec(t, ctx, tx, setting) + mustExec(t, ctx, tx, "SET LOCAL plan_cache_mode = "+mode) + // Cached plans keep the settings they were made under. + if err := tx.Conn().DeallocateAll(ctx); err != nil { + t.Fatal(err) + } + for _, pair := range pairs { + for _, w := range windows { + for _, iatas := range iataSets { + p := sqlc.GetObserverComparisonParams{ + ObserverA: pair[0], ObserverB: pair[1], Since: ts(w[0]), Until: ts(w[1]), Iatas: iatas, + } + got, err := q.GetObserverComparison(ctx, p) + if err != nil { + t.Fatal(err) + } + want := legacyRows[sqlc.GetObserverComparisonRow](t, ctx, tx, legacyGetObserverComparison, + p.ObserverA, p.ObserverB, p.Since, p.Until, p.Iatas) + if len(want) != 1 || got != want[0] { + t.Fatalf("%s, %s: GetObserverComparison(%+v) = %+v, today %+v", setting, mode, p, got, want) + } + compared++ + if got.TotalPackets > 0 { + nonEmpty++ + } + } + } + } + } + } + t.Logf("%d parameter sets compared, %d non-empty", compared, nonEmpty) + if nonEmpty < compared/4 { + t.Fatalf("only %d of %d parameter sets counted packets; the grid is too sparse to compare", nonEmpty, compared) + } + + // Plans: every observation scan is one observer's range of idx_observations_observer, never a walk of the window, + // and packets is only probed through packets_pkey once per heard packet, never scanned whole. + capture := &statementCapture{Tx: tx, name: "GetObserverComparison"} + if _, err := sqlc.New(capture).GetObserverComparison(ctx, sqlc.GetObserverComparisonParams{ObserverA: obs(1), ObserverB: obs(2)}); err != nil { + t.Fatal(err) + } + if capture.statement == "" { + t.Fatal("GetObserverComparison statement not captured") + } + mustExec(t, ctx, tx, "SET LOCAL enable_seqscan = off") + mustExec(t, ctx, tx, "PREPARE observer_comparison AS "+capture.statement) + for _, mode := range []string{"force_custom_plan", "force_generic_plan"} { + mustExec(t, ctx, tx, "SET LOCAL plan_cache_mode = "+mode) + for _, pair := range [][2]int{{1, 2}, {5, 6}, {1, 9}} { + execute := fmt.Sprintf("EXPLAIN (FORMAT JSON) EXECUTE observer_comparison('%s', '%s', '%s', '%s', NULL)", + obs(pair[0]), obs(pair[1]), comparisonBase.Format(time.RFC3339), comparisonBase.Add(7*24*time.Hour).Format(time.RFC3339)) + var raw []byte + if err := tx.QueryRow(ctx, execute).Scan(&raw); err != nil { + t.Fatal(err) + } + var plans []struct{ Plan map[string]any } + if err := json.Unmarshal(raw, &plans); err != nil || len(plans) != 1 { + t.Fatalf("EXPLAIN JSON did not parse (%v): %s", err, raw) + } + var scans, packetScans []map[string]any + var packetParents []any + var walk func(n, parent map[string]any) + walk = func(n, parent map[string]any) { + index, _ := n["Index Name"].(string) + bitmap := n["Node Type"] == "Bitmap Index Scan" + switch { + case n["Relation Name"] == "packet_observations" || + (bitmap && (strings.HasPrefix(index, "idx_observations") || strings.HasPrefix(index, "packet_observations"))): + scans = append(scans, n) + case n["Relation Name"] == "packets" || (bitmap && (strings.HasPrefix(index, "idx_packets") || index == "packets_pkey")): + packetScans = append(packetScans, n) + packetParents = append(packetParents, parent["Node Type"]) + } + if children, ok := n["Plans"].([]any); ok { + for _, c := range children { + walk(c.(map[string]any), n) + } + } + } + walk(plans[0].Plan, nil) + indexScans := 0 + for _, s := range scans { + switch s["Node Type"] { + case "Bitmap Heap Scan": + continue // its Bitmap Index Scan child names the index + case "Index Scan", "Index Only Scan", "Bitmap Index Scan": + cond, _ := s["Index Cond"].(string) + if s["Index Name"] != "idx_observations_observer" || !strings.Contains(cond, "observer_id =") { + t.Errorf("%s %v: observations read through %v (%s), not one observer's idx_observations_observer range: %s", + mode, pair, s["Index Name"], cond, raw) + } + indexScans++ + default: + t.Errorf("%s %v: observations read by %v: %s", mode, pair, s["Node Type"], raw) + } + } + if indexScans != 2 { + t.Errorf("%s %v: %d observation index scans, want one per observer: %s", mode, pair, indexScans, raw) + } + if len(packetScans) != 1 { + t.Errorf("%s %v: %d scans of packets, want one probe: %s", mode, pair, len(packetScans), raw) + } + for i, s := range packetScans { + cond, _ := s["Index Cond"].(string) + if s["Node Type"] != "Index Scan" || s["Index Name"] != "packets_pkey" || + !strings.Contains(cond, "packet_hash = COALESCE") || packetParents[i] != "Nested Loop" { + t.Errorf("%s %v: packets read by %v on %v (%s) under %v, not a packets_pkey probe per heard packet: %s", + mode, pair, s["Node Type"], s["Index Name"], cond, packetParents[i], raw) + } + } + } + } + mustExec(t, ctx, tx, "DEALLOCATE observer_comparison") +} diff --git a/db/planguard/baseline/fixture/obscmp.busy_24h_all.custom.json b/db/planguard/baseline/fixture/obscmp.busy_24h_all.custom.json new file mode 100644 index 0000000..46d9e31 --- /dev/null +++ b/db/planguard/baseline/fixture/obscmp.busy_24h_all.custom.json @@ -0,0 +1,13 @@ +[ + "Aggregate partial=Simple", + " Index Only Scan on observers using observers_pkey scan=Forward [InitPlan]", + " Index Only Scan on observers observers_1 using observers_pkey scan=Forward [InitPlan]", + " Result [InitPlan]", + " Result [InitPlan]", + " Nested Loop", + " Hash Join (Full)", + " Index Scan on packet_observations po using idx_observations_observer scan=Forward", + " Hash", + " Index Scan on packet_observations po_1 using idx_observations_observer scan=Forward", + " Index Scan on packets p using packets_pkey scan=Forward" +] diff --git a/db/planguard/baseline/fixture/obscmp.busy_24h_site.custom.json b/db/planguard/baseline/fixture/obscmp.busy_24h_site.custom.json new file mode 100644 index 0000000..46d9e31 --- /dev/null +++ b/db/planguard/baseline/fixture/obscmp.busy_24h_site.custom.json @@ -0,0 +1,13 @@ +[ + "Aggregate partial=Simple", + " Index Only Scan on observers using observers_pkey scan=Forward [InitPlan]", + " Index Only Scan on observers observers_1 using observers_pkey scan=Forward [InitPlan]", + " Result [InitPlan]", + " Result [InitPlan]", + " Nested Loop", + " Hash Join (Full)", + " Index Scan on packet_observations po using idx_observations_observer scan=Forward", + " Hash", + " Index Scan on packet_observations po_1 using idx_observations_observer scan=Forward", + " Index Scan on packets p using packets_pkey scan=Forward" +] diff --git a/db/planguard/baseline/fixture/obscmp.busy_7d_all.custom.json b/db/planguard/baseline/fixture/obscmp.busy_7d_all.custom.json new file mode 100644 index 0000000..46d9e31 --- /dev/null +++ b/db/planguard/baseline/fixture/obscmp.busy_7d_all.custom.json @@ -0,0 +1,13 @@ +[ + "Aggregate partial=Simple", + " Index Only Scan on observers using observers_pkey scan=Forward [InitPlan]", + " Index Only Scan on observers observers_1 using observers_pkey scan=Forward [InitPlan]", + " Result [InitPlan]", + " Result [InitPlan]", + " Nested Loop", + " Hash Join (Full)", + " Index Scan on packet_observations po using idx_observations_observer scan=Forward", + " Hash", + " Index Scan on packet_observations po_1 using idx_observations_observer scan=Forward", + " Index Scan on packets p using packets_pkey scan=Forward" +] diff --git a/db/planguard/baseline/fixture/obscmp.busy_7d_empty_region.custom.json b/db/planguard/baseline/fixture/obscmp.busy_7d_empty_region.custom.json new file mode 100644 index 0000000..46d9e31 --- /dev/null +++ b/db/planguard/baseline/fixture/obscmp.busy_7d_empty_region.custom.json @@ -0,0 +1,13 @@ +[ + "Aggregate partial=Simple", + " Index Only Scan on observers using observers_pkey scan=Forward [InitPlan]", + " Index Only Scan on observers observers_1 using observers_pkey scan=Forward [InitPlan]", + " Result [InitPlan]", + " Result [InitPlan]", + " Nested Loop", + " Hash Join (Full)", + " Index Scan on packet_observations po using idx_observations_observer scan=Forward", + " Hash", + " Index Scan on packet_observations po_1 using idx_observations_observer scan=Forward", + " Index Scan on packets p using packets_pkey scan=Forward" +] diff --git a/db/planguard/baseline/fixture/obscmp.busy_7d_region.custom.json b/db/planguard/baseline/fixture/obscmp.busy_7d_region.custom.json new file mode 100644 index 0000000..46d9e31 --- /dev/null +++ b/db/planguard/baseline/fixture/obscmp.busy_7d_region.custom.json @@ -0,0 +1,13 @@ +[ + "Aggregate partial=Simple", + " Index Only Scan on observers using observers_pkey scan=Forward [InitPlan]", + " Index Only Scan on observers observers_1 using observers_pkey scan=Forward [InitPlan]", + " Result [InitPlan]", + " Result [InitPlan]", + " Nested Loop", + " Hash Join (Full)", + " Index Scan on packet_observations po using idx_observations_observer scan=Forward", + " Hash", + " Index Scan on packet_observations po_1 using idx_observations_observer scan=Forward", + " Index Scan on packets p using packets_pkey scan=Forward" +] diff --git a/db/planguard/baseline/fixture/obscmp.mixed_7d_all.custom.json b/db/planguard/baseline/fixture/obscmp.mixed_7d_all.custom.json new file mode 100644 index 0000000..46d9e31 --- /dev/null +++ b/db/planguard/baseline/fixture/obscmp.mixed_7d_all.custom.json @@ -0,0 +1,13 @@ +[ + "Aggregate partial=Simple", + " Index Only Scan on observers using observers_pkey scan=Forward [InitPlan]", + " Index Only Scan on observers observers_1 using observers_pkey scan=Forward [InitPlan]", + " Result [InitPlan]", + " Result [InitPlan]", + " Nested Loop", + " Hash Join (Full)", + " Index Scan on packet_observations po using idx_observations_observer scan=Forward", + " Hash", + " Index Scan on packet_observations po_1 using idx_observations_observer scan=Forward", + " Index Scan on packets p using packets_pkey scan=Forward" +] diff --git a/db/planguard/baseline/fixture/obscmp.quiet_7d_all.custom.json b/db/planguard/baseline/fixture/obscmp.quiet_7d_all.custom.json new file mode 100644 index 0000000..46d9e31 --- /dev/null +++ b/db/planguard/baseline/fixture/obscmp.quiet_7d_all.custom.json @@ -0,0 +1,13 @@ +[ + "Aggregate partial=Simple", + " Index Only Scan on observers using observers_pkey scan=Forward [InitPlan]", + " Index Only Scan on observers observers_1 using observers_pkey scan=Forward [InitPlan]", + " Result [InitPlan]", + " Result [InitPlan]", + " Nested Loop", + " Hash Join (Full)", + " Index Scan on packet_observations po using idx_observations_observer scan=Forward", + " Hash", + " Index Scan on packet_observations po_1 using idx_observations_observer scan=Forward", + " Index Scan on packets p using packets_pkey scan=Forward" +] diff --git a/db/planguard/baseline/fixture/obscmp.unknown_7d.custom.json b/db/planguard/baseline/fixture/obscmp.unknown_7d.custom.json new file mode 100644 index 0000000..46d9e31 --- /dev/null +++ b/db/planguard/baseline/fixture/obscmp.unknown_7d.custom.json @@ -0,0 +1,13 @@ +[ + "Aggregate partial=Simple", + " Index Only Scan on observers using observers_pkey scan=Forward [InitPlan]", + " Index Only Scan on observers observers_1 using observers_pkey scan=Forward [InitPlan]", + " Result [InitPlan]", + " Result [InitPlan]", + " Nested Loop", + " Hash Join (Full)", + " Index Scan on packet_observations po using idx_observations_observer scan=Forward", + " Hash", + " Index Scan on packet_observations po_1 using idx_observations_observer scan=Forward", + " Index Scan on packets p using packets_pkey scan=Forward" +] diff --git a/db/planguard/baseline/local/obscmp.busy_24h_all.custom.json b/db/planguard/baseline/local/obscmp.busy_24h_all.custom.json new file mode 100644 index 0000000..46d9e31 --- /dev/null +++ b/db/planguard/baseline/local/obscmp.busy_24h_all.custom.json @@ -0,0 +1,13 @@ +[ + "Aggregate partial=Simple", + " Index Only Scan on observers using observers_pkey scan=Forward [InitPlan]", + " Index Only Scan on observers observers_1 using observers_pkey scan=Forward [InitPlan]", + " Result [InitPlan]", + " Result [InitPlan]", + " Nested Loop", + " Hash Join (Full)", + " Index Scan on packet_observations po using idx_observations_observer scan=Forward", + " Hash", + " Index Scan on packet_observations po_1 using idx_observations_observer scan=Forward", + " Index Scan on packets p using packets_pkey scan=Forward" +] diff --git a/db/planguard/baseline/local/obscmp.busy_24h_site.custom.json b/db/planguard/baseline/local/obscmp.busy_24h_site.custom.json new file mode 100644 index 0000000..46d9e31 --- /dev/null +++ b/db/planguard/baseline/local/obscmp.busy_24h_site.custom.json @@ -0,0 +1,13 @@ +[ + "Aggregate partial=Simple", + " Index Only Scan on observers using observers_pkey scan=Forward [InitPlan]", + " Index Only Scan on observers observers_1 using observers_pkey scan=Forward [InitPlan]", + " Result [InitPlan]", + " Result [InitPlan]", + " Nested Loop", + " Hash Join (Full)", + " Index Scan on packet_observations po using idx_observations_observer scan=Forward", + " Hash", + " Index Scan on packet_observations po_1 using idx_observations_observer scan=Forward", + " Index Scan on packets p using packets_pkey scan=Forward" +] diff --git a/db/planguard/baseline/local/obscmp.busy_7d_all.custom.json b/db/planguard/baseline/local/obscmp.busy_7d_all.custom.json new file mode 100644 index 0000000..46d9e31 --- /dev/null +++ b/db/planguard/baseline/local/obscmp.busy_7d_all.custom.json @@ -0,0 +1,13 @@ +[ + "Aggregate partial=Simple", + " Index Only Scan on observers using observers_pkey scan=Forward [InitPlan]", + " Index Only Scan on observers observers_1 using observers_pkey scan=Forward [InitPlan]", + " Result [InitPlan]", + " Result [InitPlan]", + " Nested Loop", + " Hash Join (Full)", + " Index Scan on packet_observations po using idx_observations_observer scan=Forward", + " Hash", + " Index Scan on packet_observations po_1 using idx_observations_observer scan=Forward", + " Index Scan on packets p using packets_pkey scan=Forward" +] diff --git a/db/planguard/baseline/local/obscmp.busy_7d_empty_region.custom.json b/db/planguard/baseline/local/obscmp.busy_7d_empty_region.custom.json new file mode 100644 index 0000000..46d9e31 --- /dev/null +++ b/db/planguard/baseline/local/obscmp.busy_7d_empty_region.custom.json @@ -0,0 +1,13 @@ +[ + "Aggregate partial=Simple", + " Index Only Scan on observers using observers_pkey scan=Forward [InitPlan]", + " Index Only Scan on observers observers_1 using observers_pkey scan=Forward [InitPlan]", + " Result [InitPlan]", + " Result [InitPlan]", + " Nested Loop", + " Hash Join (Full)", + " Index Scan on packet_observations po using idx_observations_observer scan=Forward", + " Hash", + " Index Scan on packet_observations po_1 using idx_observations_observer scan=Forward", + " Index Scan on packets p using packets_pkey scan=Forward" +] diff --git a/db/planguard/baseline/local/obscmp.busy_7d_region.custom.json b/db/planguard/baseline/local/obscmp.busy_7d_region.custom.json new file mode 100644 index 0000000..46d9e31 --- /dev/null +++ b/db/planguard/baseline/local/obscmp.busy_7d_region.custom.json @@ -0,0 +1,13 @@ +[ + "Aggregate partial=Simple", + " Index Only Scan on observers using observers_pkey scan=Forward [InitPlan]", + " Index Only Scan on observers observers_1 using observers_pkey scan=Forward [InitPlan]", + " Result [InitPlan]", + " Result [InitPlan]", + " Nested Loop", + " Hash Join (Full)", + " Index Scan on packet_observations po using idx_observations_observer scan=Forward", + " Hash", + " Index Scan on packet_observations po_1 using idx_observations_observer scan=Forward", + " Index Scan on packets p using packets_pkey scan=Forward" +] diff --git a/db/planguard/baseline/local/obscmp.mixed_7d_all.custom.json b/db/planguard/baseline/local/obscmp.mixed_7d_all.custom.json new file mode 100644 index 0000000..46d9e31 --- /dev/null +++ b/db/planguard/baseline/local/obscmp.mixed_7d_all.custom.json @@ -0,0 +1,13 @@ +[ + "Aggregate partial=Simple", + " Index Only Scan on observers using observers_pkey scan=Forward [InitPlan]", + " Index Only Scan on observers observers_1 using observers_pkey scan=Forward [InitPlan]", + " Result [InitPlan]", + " Result [InitPlan]", + " Nested Loop", + " Hash Join (Full)", + " Index Scan on packet_observations po using idx_observations_observer scan=Forward", + " Hash", + " Index Scan on packet_observations po_1 using idx_observations_observer scan=Forward", + " Index Scan on packets p using packets_pkey scan=Forward" +] diff --git a/db/planguard/baseline/local/obscmp.quiet_7d_all.custom.json b/db/planguard/baseline/local/obscmp.quiet_7d_all.custom.json new file mode 100644 index 0000000..46d9e31 --- /dev/null +++ b/db/planguard/baseline/local/obscmp.quiet_7d_all.custom.json @@ -0,0 +1,13 @@ +[ + "Aggregate partial=Simple", + " Index Only Scan on observers using observers_pkey scan=Forward [InitPlan]", + " Index Only Scan on observers observers_1 using observers_pkey scan=Forward [InitPlan]", + " Result [InitPlan]", + " Result [InitPlan]", + " Nested Loop", + " Hash Join (Full)", + " Index Scan on packet_observations po using idx_observations_observer scan=Forward", + " Hash", + " Index Scan on packet_observations po_1 using idx_observations_observer scan=Forward", + " Index Scan on packets p using packets_pkey scan=Forward" +] diff --git a/db/planguard/baseline/local/obscmp.unknown_7d.custom.json b/db/planguard/baseline/local/obscmp.unknown_7d.custom.json new file mode 100644 index 0000000..46d9e31 --- /dev/null +++ b/db/planguard/baseline/local/obscmp.unknown_7d.custom.json @@ -0,0 +1,13 @@ +[ + "Aggregate partial=Simple", + " Index Only Scan on observers using observers_pkey scan=Forward [InitPlan]", + " Index Only Scan on observers observers_1 using observers_pkey scan=Forward [InitPlan]", + " Result [InitPlan]", + " Result [InitPlan]", + " Nested Loop", + " Hash Join (Full)", + " Index Scan on packet_observations po using idx_observations_observer scan=Forward", + " Hash", + " Index Scan on packet_observations po_1 using idx_observations_observer scan=Forward", + " Index Scan on packets p using packets_pkey scan=Forward" +] diff --git a/db/planguard/corpus.yaml b/db/planguard/corpus.yaml index f2a1742..c7bae60 100644 --- a/db/planguard/corpus.yaml +++ b/db/planguard/corpus.yaml @@ -108,6 +108,7 @@ statements: GetStatsTopAdvertisers: {sqlc: GetStatsTopAdvertisers, plan: custom} GetPathStats: {sqlc: GetPathStats, plan: custom} GetStatsSeries: {sqlc: GetStatsSeries, group: stats_adverts, plan: custom} + GetObserverComparison: {sqlc: GetObserverComparison, plan: custom} ListTraceTags: {sqlc: ListTraceTags, group: stats_adverts, plan: custom} # One observer's adverts and telemetry pages. ListObserverAdverts: {sqlc: ListObserverAdverts, group: stats_adverts, plan: cached, admission: observer_page} @@ -244,6 +245,19 @@ resolvers: ps_sparse_site: "SELECT iata FROM (SELECT po.iata, count(*) n FROM packet_observations po WHERE po.id > (SELECT max(id) - 200000 FROM packet_observations) GROUP BY po.iata ORDER BY n DESC, po.iata) r ORDER BY n, iata LIMIT 1" ps_stale_site: "SELECT iata FROM packet_sites WHERE iata = 'NEW' LIMIT 1" ps_zero_scope: "SELECT ARRAY[ts.name] FROM transport_scopes ts WHERE NOT EXISTS (SELECT 1 FROM packet_sites ps WHERE ps.iata = (SELECT iata FROM (SELECT po.iata, count(*) n FROM packet_observations po WHERE po.id > (SELECT max(id) - 200000 FROM packet_observations) GROUP BY po.iata ORDER BY n DESC, po.iata LIMIT 1) r) AND ps.scope_id = ts.id) ORDER BY ts.name LIMIT 1" + # Observer comparison roles, picked among the observers in the newest 200,000 observation ids (a pkey range): the + # two busiest, the busiest one's latest site, and the two that heard least (but something) in the 7 d window, each + # counted on idx_observations_observer up to 1,000. cmp_unknown is a fixed non-zero id that matches no observer. + # Windows end at the newest reception (a max on idx_observations_heard). + cmp_busy1: "SELECT observer_id FROM packet_observations WHERE id > (SELECT max(id) FROM packet_observations) - 200000 GROUP BY 1 ORDER BY count(*) DESC, observer_id LIMIT 1" + cmp_busy2: "SELECT observer_id FROM packet_observations WHERE id > (SELECT max(id) FROM packet_observations) - 200000 GROUP BY 1 ORDER BY count(*) DESC, observer_id OFFSET 1 LIMIT 1" + cmp_quiet1_7d: "WITH t AS (SELECT max(heard_at) hi FROM packet_observations), c AS (SELECT DISTINCT observer_id FROM packet_observations WHERE id > (SELECT max(id) FROM packet_observations) - 200000) SELECT c.observer_id FROM c CROSS JOIN t CROSS JOIN LATERAL (SELECT count(*) n FROM (SELECT 1 FROM packet_observations x WHERE x.observer_id = c.observer_id AND x.heard_at >= t.hi - interval '7 days' AND x.heard_at < t.hi LIMIT 1000) s) k WHERE k.n > 0 ORDER BY k.n, c.observer_id LIMIT 1" + cmp_quiet2_7d: "WITH t AS (SELECT max(heard_at) hi FROM packet_observations), c AS (SELECT DISTINCT observer_id FROM packet_observations WHERE id > (SELECT max(id) FROM packet_observations) - 200000) SELECT c.observer_id FROM c CROSS JOIN t CROSS JOIN LATERAL (SELECT count(*) n FROM (SELECT 1 FROM packet_observations x WHERE x.observer_id = c.observer_id AND x.heard_at >= t.hi - interval '7 days' AND x.heard_at < t.hi LIMIT 1000) s) k WHERE k.n > 0 ORDER BY k.n, c.observer_id OFFSET 1 LIMIT 1" + cmp_busy1_site: "SELECT iata FROM packet_observations WHERE id > (SELECT max(id) FROM packet_observations) - 200000 AND observer_id = (SELECT observer_id FROM packet_observations WHERE id > (SELECT max(id) FROM packet_observations) - 200000 GROUP BY 1 ORDER BY count(*) DESC, observer_id LIMIT 1) ORDER BY id DESC LIMIT 1" + cmp_unknown: "SELECT md5('observer comparison: no such observer')::uuid WHERE NOT EXISTS (SELECT 1 FROM observers WHERE id = md5('observer comparison: no such observer')::uuid)" + cmp_until: "SELECT max(heard_at) FROM packet_observations" + cmp_since_1d: "SELECT max(heard_at) - interval '1 day' FROM packet_observations" + cmp_since_7d: "SELECT max(heard_at) - interval '7 days' FROM packet_observations" # The adverts and telemetry roles are picked among the newest 5,000,000 observation ids (200,000 telemetry ids, about # 10K rows), a pkey range: advert counts are aggregate filters, never a WHERE on payload_type, so the partial adverts # index cannot replace the window. They stay far below the prod tier's 20 s cap; adv_zero and adv_quiet then probe @@ -2404,6 +2418,80 @@ entries: budget: {buffers: 520, planning_ms: 10} note: "measured: 3 / 3 B (custom / generic); budget: gate <= 10 B/row + 10" + # Observer comparison: each observer's receptions in the window from idx_observations_observer, then one packets_pkey + # probe per heard packet; never a walk of the window through idx_observations_heard, never a scan of all packets. + - id: obscmp.busy_24h_all + stmt: GetObserverComparison + shape: "the two busiest observers, 24 h, every site" + params: [{r: cmp_busy1}, {r: cmp_busy2}, {r: cmp_since_1d}, {r: cmp_until}, null] + expect: + driver: [{relation: packet_observations, index: idx_observations_observer, nodes: ["Bitmap Heap Scan", "Index Scan"], index_cond: ["observer_id ="]}, + {relation: packets, index: packets_pkey, nodes: ["Index Scan"], index_cond: ["packet_hash = COALESCE"]}] + budget: {buffers: 196341, planning_ms: 10} + note: "measured: 130894 B on the local copy (t_end, custom; the statement it replaced: 130894 B); budget: 1.5x measured" + - id: obscmp.busy_7d_all + stmt: GetObserverComparison + shape: "the two busiest observers, 7 d, every site" + params: [{r: cmp_busy1}, {r: cmp_busy2}, {r: cmp_since_7d}, {r: cmp_until}, null] + expect: + driver: [{relation: packet_observations, index: idx_observations_observer, nodes: ["Bitmap Heap Scan", "Index Scan"], index_cond: ["observer_id ="]}, + {relation: packets, index: packets_pkey, nodes: ["Index Scan"], index_cond: ["packet_hash = COALESCE"]}] + budget: {buffers: 558295, planning_ms: 10} + note: "measured: 372197 B on the local copy (t_end, custom; the statement it replaced: 370407 B); budget: 1.5x measured" + - id: obscmp.busy_24h_site + stmt: GetObserverComparison + shape: "the two busiest observers, 24 h, the busiest one's site" + params: [{r: cmp_busy1}, {r: cmp_busy2}, {r: cmp_since_1d}, {r: cmp_until}, [{r: cmp_busy1_site}]] + expect: + driver: [{relation: packet_observations, index: idx_observations_observer, nodes: ["Bitmap Heap Scan", "Index Scan"], index_cond: ["observer_id ="]}, + {relation: packets, index: packets_pkey, nodes: ["Index Scan"], index_cond: ["packet_hash = COALESCE"]}] + budget: {buffers: 91743, planning_ms: 10} + note: "measured: 61162 B on the local copy (t_end, custom; the statement it replaced: 61162 B); budget: 1.5x measured" + - id: obscmp.busy_7d_region + stmt: GetObserverComparison + shape: "the two busiest observers, 7 d, the seven busiest sites" + params: [{r: cmp_busy1}, {r: cmp_busy2}, {r: cmp_since_7d}, {r: cmp_until}, {r: ps_busy_sites7}] + expect: + driver: [{relation: packet_observations, index: idx_observations_observer, nodes: ["Bitmap Heap Scan", "Index Scan"], index_cond: ["observer_id ="]}, + {relation: packets, index: packets_pkey, nodes: ["Index Scan"], index_cond: ["packet_hash = COALESCE"]}] + budget: {buffers: 258049, planning_ms: 10} + note: "measured: 172033 B on the local copy (t_end, custom; the statement it replaced: 170243 B); budget: 1.5x measured" + - id: obscmp.busy_7d_empty_region + stmt: GetObserverComparison + shape: "the two busiest observers, 7 d, a region with no sites" + params: [{r: cmp_busy1}, {r: cmp_busy2}, {r: cmp_since_7d}, {r: cmp_until}, [""]] + expect: + driver: [{relation: packet_observations, index: idx_observations_observer, nodes: ["Bitmap Heap Scan", "Index Scan"], index_cond: ["observer_id ="]}, + {relation: packets, index: packets_pkey, nodes: ["Index Scan"], index_cond: ["packet_hash = COALESCE"]}] + budget: {buffers: 101311, planning_ms: 10} + note: "measured: 67541 B on the local copy (t_end, custom; the statement it replaced: 65751 B); budget: 1.5x measured" + - id: obscmp.quiet_7d_all + stmt: GetObserverComparison + shape: "the two quietest observers, 7 d, every site" + params: [{r: cmp_quiet1_7d}, {r: cmp_quiet2_7d}, {r: cmp_since_7d}, {r: cmp_until}, null] + expect: + driver: [{relation: packet_observations, index: idx_observations_observer, nodes: ["Bitmap Heap Scan", "Index Scan"], index_cond: ["observer_id ="]}, + {relation: packets, index: packets_pkey, nodes: ["Index Scan"], index_cond: ["packet_hash = COALESCE"]}] + budget: {buffers: 261, planning_ms: 10} + note: "measured: 174 B on the local copy (t_end, custom; the statement it replaced: 174 B); budget: 1.5x measured" + - id: obscmp.mixed_7d_all + stmt: GetObserverComparison + shape: "the busiest and the quietest observer, 7 d, every site" + params: [{r: cmp_busy1}, {r: cmp_quiet1_7d}, {r: cmp_since_7d}, {r: cmp_until}, null] + expect: + driver: [{relation: packet_observations, index: idx_observations_observer, nodes: ["Bitmap Heap Scan", "Index Scan"], index_cond: ["observer_id ="]}, + {relation: packets, index: packets_pkey, nodes: ["Index Scan"], index_cond: ["packet_hash = COALESCE"]}] + budget: {buffers: 195237, planning_ms: 10} + note: "measured: 130158 B on the local copy (t_end, custom; the statement it replaced: 130158 B); budget: 1.5x measured" + - id: obscmp.unknown_7d + stmt: GetObserverComparison + shape: "the busiest observer and an unknown id, 7 d" + params: [{r: cmp_busy1}, {r: cmp_unknown}, {r: cmp_since_7d}, {r: cmp_until}, null] + expect: + driver: [{relation: packet_observations, index: idx_observations_observer, nodes: ["Bitmap Heap Scan", "Index Scan"], index_cond: ["observer_id ="]}, + {relation: packets, index: packets_pkey, nodes: ["Index Scan"], index_cond: ["packet_hash = COALESCE"]}] + budget: {buffers: 195123, planning_ms: 10} + note: "measured: 130082 B on the local copy (t_end, custom; the statement it replaced: 130082 B); budget: 1.5x measured" # ---- small tables: nodes, observers, channels, node types, clock drift ---- - id: sm.nodes.global stmt: ListNodes diff --git a/db/queries/observer_comparison.sql b/db/queries/observer_comparison.sql index dab05da..cfb7e09 100644 --- a/db/queries/observer_comparison.sql +++ b/db/queries/observer_comparison.sql @@ -1,18 +1,34 @@ -- name: GetObserverComparison :one -- Count packet identities, not reception rows: a broker repeat or another -- reception by the same observer must not inflate the comparison. -WITH heard AS ( - SELECT po.packet_hash, - bool_or(po.observer_id = @observer_a::uuid) AS seen_a, - bool_or(po.observer_id = @observer_b::uuid) AS seen_b +-- +-- Each observer's receptions are read on their own from idx_observations_observer +-- (observer_id, heard_at): (packet_hash, observer_id) is unique, so each side +-- holds a packet at most once, and the full join yields each heard packet once. +-- The route type is checked once per packet after the join. +-- The observer ids are scalar subqueries so the planner costs every observer +-- alike: given a busy observer's own row count, a custom plan may instead walk +-- every reception in the window through idx_observations_heard, which reads +-- several times the buffers. +WITH a AS ( + SELECT po.packet_hash FROM packet_observations po - JOIN packets p ON p.packet_hash = po.packet_hash - WHERE po.observer_id IN (@observer_a::uuid, @observer_b::uuid) + WHERE po.observer_id = (SELECT @observer_a::uuid) AND po.heard_at >= @since::timestamptz AND po.heard_at < @until::timestamptz - AND p.route_type IN (0, 1) AND (COALESCE(cardinality(@iatas::text[]), 0) = 0 OR po.iata = ANY(@iatas::text[])) - GROUP BY po.packet_hash +), b AS ( + SELECT po.packet_hash + FROM packet_observations po + WHERE po.observer_id = (SELECT @observer_b::uuid) + AND po.heard_at >= @since::timestamptz + AND po.heard_at < @until::timestamptz + AND (COALESCE(cardinality(@iatas::text[]), 0) = 0 OR po.iata = ANY(@iatas::text[])) +), heard AS ( + SELECT COALESCE(a.packet_hash, b.packet_hash) AS packet_hash, + a.packet_hash IS NOT NULL AS seen_a, + b.packet_hash IS NOT NULL AS seen_b + FROM a FULL JOIN b ON b.packet_hash = a.packet_hash ) SELECT count(*) AS total_packets, count(*) FILTER (WHERE seen_a AND NOT seen_b) AS only_a, @@ -20,4 +36,6 @@ SELECT count(*) AS total_packets, count(*) FILTER (WHERE seen_a AND seen_b) AS both, EXISTS (SELECT 1 FROM observers WHERE id = @observer_a::uuid) AS known_a, EXISTS (SELECT 1 FROM observers WHERE id = @observer_b::uuid) AS known_b -FROM heard; +FROM heard +JOIN packets p ON p.packet_hash = heard.packet_hash +WHERE p.route_type IN (0, 1); diff --git a/db/sqlc/observer_comparison.sql.go b/db/sqlc/observer_comparison.sql.go index 00c0769..56b04b3 100644 --- a/db/sqlc/observer_comparison.sql.go +++ b/db/sqlc/observer_comparison.sql.go @@ -13,18 +13,25 @@ import ( ) const getObserverComparison = `-- name: GetObserverComparison :one -WITH heard AS ( - SELECT po.packet_hash, - bool_or(po.observer_id = $1::uuid) AS seen_a, - bool_or(po.observer_id = $2::uuid) AS seen_b +WITH a AS ( + SELECT po.packet_hash FROM packet_observations po - JOIN packets p ON p.packet_hash = po.packet_hash - WHERE po.observer_id IN ($1::uuid, $2::uuid) + WHERE po.observer_id = (SELECT $1::uuid) AND po.heard_at >= $3::timestamptz AND po.heard_at < $4::timestamptz - AND p.route_type IN (0, 1) AND (COALESCE(cardinality($5::text[]), 0) = 0 OR po.iata = ANY($5::text[])) - GROUP BY po.packet_hash +), b AS ( + SELECT po.packet_hash + FROM packet_observations po + WHERE po.observer_id = (SELECT $2::uuid) + AND po.heard_at >= $3::timestamptz + AND po.heard_at < $4::timestamptz + AND (COALESCE(cardinality($5::text[]), 0) = 0 OR po.iata = ANY($5::text[])) +), heard AS ( + SELECT COALESCE(a.packet_hash, b.packet_hash) AS packet_hash, + a.packet_hash IS NOT NULL AS seen_a, + b.packet_hash IS NOT NULL AS seen_b + FROM a FULL JOIN b ON b.packet_hash = a.packet_hash ) SELECT count(*) AS total_packets, count(*) FILTER (WHERE seen_a AND NOT seen_b) AS only_a, @@ -33,6 +40,8 @@ SELECT count(*) AS total_packets, EXISTS (SELECT 1 FROM observers WHERE id = $1::uuid) AS known_a, EXISTS (SELECT 1 FROM observers WHERE id = $2::uuid) AS known_b FROM heard +JOIN packets p ON p.packet_hash = heard.packet_hash +WHERE p.route_type IN (0, 1) ` type GetObserverComparisonParams struct { @@ -54,6 +63,15 @@ type GetObserverComparisonRow struct { // Count packet identities, not reception rows: a broker repeat or another // reception by the same observer must not inflate the comparison. +// +// Each observer's receptions are read on their own from idx_observations_observer +// (observer_id, heard_at): (packet_hash, observer_id) is unique, so each side +// holds a packet at most once, and the full join yields each heard packet once. +// The route type is checked once per packet after the join. +// The observer ids are scalar subqueries so the planner costs every observer +// alike: given a busy observer's own row count, a custom plan may instead walk +// every reception in the window through idx_observations_heard, which reads +// several times the buffers. func (q *Queries) GetObserverComparison(ctx context.Context, arg GetObserverComparisonParams) (GetObserverComparisonRow, error) { row := q.db.QueryRow(ctx, getObserverComparison, arg.ObserverA, diff --git a/db/sqlc/querier.go b/db/sqlc/querier.go index fdcf222..a8157f1 100644 --- a/db/sqlc/querier.go +++ b/db/sqlc/querier.go @@ -148,6 +148,15 @@ type Querier interface { GetObserverByPubkey(ctx context.Context, publicKey []byte) (Observer, error) // Count packet identities, not reception rows: a broker repeat or another // reception by the same observer must not inflate the comparison. + // + // Each observer's receptions are read on their own from idx_observations_observer + // (observer_id, heard_at): (packet_hash, observer_id) is unique, so each side + // holds a packet at most once, and the full join yields each heard packet once. + // The route type is checked once per packet after the join. + // The observer ids are scalar subqueries so the planner costs every observer + // alike: given a busy observer's own row count, a custom plan may instead walk + // every reception in the window through idx_observations_heard, which reads + // several times the buffers. GetObserverComparison(ctx context.Context, arg GetObserverComparisonParams) (GetObserverComparisonRow, error) GetObserverLastIATA(ctx context.Context, id uuid.UUID) (string, error) GetObserverRadio(ctx context.Context, id uuid.UUID) (GetObserverRadioRow, error) diff --git a/db/statement_capture_test.go b/db/statement_capture_test.go index 22b68f0..4a5caed 100644 --- a/db/statement_capture_test.go +++ b/db/statement_capture_test.go @@ -24,3 +24,10 @@ func (c *statementCapture) Query(ctx context.Context, sql string, args ...any) ( } return c.Tx.Query(ctx, sql, args...) } + +func (c *statementCapture) QueryRow(ctx context.Context, sql string, args ...any) pgx.Row { + if strings.Contains(sql, "-- name: "+c.name) { + c.statement, c.args = sql, args + } + return c.Tx.QueryRow(ctx, sql, args...) +}