test: add regression coverage for aggregate ordering, DDL correctness, and memory accounting

New test files:
- sql_order_by: GROUP BY + ORDER BY must sort groups in declared order.
- sql_aggregate_functions: aggregate alias casing and count_distinct must not
  return NULL under any column-name mapping.
- sql_subquery_from: FROM (SELECT …) AS t must plan and execute correctly.
- sql_drop_collection: DROP COLLECTION IF EXISTS must not error when the
  target does not exist; DROP TABLE must be accepted as a synonym.
- memory_accounting_balance: over-release counter must be zero after a
  balanced alloc/release cycle; non-zero signals an accounting bug.
- engine_surface_timeseries: continuous aggregate must appear in SHOW after
  CREATE and survive a server restart.
- catalog_integrity_verify: continuous aggregates included in orphan-row
  integrity sweep.

Existing tests updated to compile against the new sort_keys field on
SqlPlan::Aggregate and the expanded pgwire harness helper surface.

Author: farhan-syah

Cargo.lock

@@ -16,6 +16,7 @@ nodedb = { workspace = true }
 nodedb-array = { workspace = true }
 nodedb-bridge = { workspace = true }
 nodedb-cluster = { workspace = true }
+nodedb-mem = { workspace = true }
 nodedb-types = { workspace = true }
 nodedb-wal = { workspace = true }
 

nodedb-test-support/Cargo.toml

@@ -10,6 +10,30 @@ use std::time::Duration;
 
 use nodedb::bridge::dispatch::Dispatcher;
 
+/// Build a `MemoryGovernor` sized for integration tests: 64 MiB per
+/// engine, global ceiling = 2× the sum of per-engine budgets. Returns
+/// `None` only if the underlying `GovernorConfig` validation fails,
+/// matching the original inline behaviour.
+///
+/// Without a wired governor any test that asserts on balanced
+/// acquire/release after a workload trips on `governor.is_none()`
+/// instead of the real accounting bug — every test entry point that
+/// hands out a `SharedState` must install one.
+fn init_test_memory_governor() -> Option<Arc<nodedb_mem::MemoryGovernor>> {
+    let test_budget: usize = 64 * 1024 * 1024; // 64 MiB / engine
+    let mut engine_limits = std::collections::HashMap::new();
+    for id in nodedb_mem::EngineId::ALL {
+        engine_limits.insert(*id, test_budget);
+    }
+    let global_ceiling = test_budget * nodedb_mem::EngineId::ALL.len() * 2;
+    nodedb_mem::MemoryGovernor::new(nodedb_mem::GovernorConfig {
+        global_ceiling,
+        engine_limits,
+    })
+    .ok()
+    .map(Arc::new)
+}
+
 /// Generate a short hex string suitable for unique test name suffixes.
 fn uuid_v4_hex() -> String {
     let id = uuid::Uuid::new_v4();
@@ -166,6 +190,7 @@ impl TestServer {
         // Deterministic 32-byte key — same value every test run.
         if let Some(s) = Arc::get_mut(&mut shared) {
             s.backup_kek = Some(Arc::new([0x42u8; 32]));
+            s.governor = init_test_memory_governor();
         }
         let shared = shared;
 
@@ -428,6 +453,7 @@ impl TestServer {
             SharedState::new_with_credentials(dispatcher, Arc::clone(&wal), credentials);
         if let Some(s) = Arc::get_mut(&mut shared) {
             s.backup_kek = Some(Arc::new([0x42u8; 32]));
+            s.governor = init_test_memory_governor();
         }
         let shared = shared;
         nodedb::bootstrap::credentials::replay_surrogate_wal(&shared, &wal_records);
@@ -504,6 +530,14 @@ impl TestServer {
             .unwrap();
         }
 
+        // Re-register every persisted continuous aggregate on the local
+        // Data Plane manager: the registry is per-core in-memory state
+        // and is otherwise lost across restart.
+        nodedb::control::server::pgwire::ddl::continuous_agg::register_persisted_continuous_aggregates(
+            &shared,
+        )
+        .await;
+
         let watermark_store =
             Arc::new(nodedb::event::watermark::WatermarkStore::open(dir_path).unwrap());
         let trigger_dlq = Arc::new(std::sync::Mutex::new(
@@ -613,6 +647,7 @@ impl TestServer {
             SharedState::new_with_credentials(dispatcher, Arc::clone(&wal), credentials);
         if let Some(s) = Arc::get_mut(&mut shared) {
             s.backup_kek = Some(Arc::new([0x42u8; 32]));
+            s.governor = init_test_memory_governor();
         }
         let shared = shared;
 

nodedb-test-support/src/pgwire_harness.rs

@@ -246,6 +246,7 @@ pub fn columnar_count(collection: &str) -> PhysicalPlan {
         sub_group_by: Vec::new(),
         sub_aggregates: Vec::new(),
         grouping_sets: Vec::new(),
+        sort_keys: Vec::new(),
     })
 }
 

nodedb-test-support/src/tx_batch_helpers.rs

@@ -140,6 +140,12 @@ fn classify(entry: &CatalogEntry) -> VariantClass {
         CatalogEntry::DeleteSchedule { .. } => VariantClass::ParentReplicated,
         CatalogEntry::DeleteChangeStream { .. } => VariantClass::ParentReplicated,
 
+        // Continuous aggregates carry a parent-replicated definition row
+        // plus an owner row (apply path calls `put_parent_owner` /
+        // `delete_parent_owner`).
+        CatalogEntry::PutContinuousAggregate(_) => VariantClass::ParentReplicated,
+        CatalogEntry::DeleteContinuousAggregate { .. } => VariantClass::ParentReplicated,
+
         CatalogEntry::PutOwner(_) => VariantClass::Exempt,
         CatalogEntry::DeleteOwner { .. } => VariantClass::Exempt,
 

nodedb/tests/catalog_integrity_verify.rs

@@ -176,3 +176,127 @@ async fn wal_restart_durability() {
     let expected = 3.14_f64;
     assert!((v - expected).abs() < 0.01);
 }
+
+// ── Continuous-aggregate registration must survive past the in-memory Data ──
+//
+// `CREATE CONTINUOUS AGGREGATE` currently dispatches `RegisterContinuousAggregate`
+// straight to the Data Plane manager and returns success — no catalog row, no
+// target collection, no Raft replication. Every assertion below pins one
+// observable consequence of "no catalog persistence":
+//
+//   1. SHOW CONTINUOUS AGGREGATES must list the registration immediately.
+//   2. The aggregate must still be listed after a restart.
+//   3. The aggregate name must resolve as a queryable relation, not "unknown
+//      table" — that is the whole point of materializing the result.
+//
+// All three are silent-failure spec assertions: if the CA registration ever
+// regresses back to "succeeds locally, vanishes elsewhere", a single test in
+// this group catches it.
+
+/// SHOW CONTINUOUS AGGREGATES must list a freshly-created aggregate. The
+/// `CREATE` path returns success without registering anything observable.
+#[tokio::test]
+async fn continuous_aggregate_visible_in_show_after_create() {
+    let srv = TestServer::start().await;
+    srv.exec(
+        "CREATE COLLECTION ts_cagg_show_src \
+         COLUMNS (id TEXT, ts BIGINT TIME_KEY, value FLOAT) \
+         WITH (engine='timeseries')",
+    )
+    .await
+    .unwrap();
+    srv.exec(
+        "CREATE CONTINUOUS AGGREGATE ts_cagg_show_view \
+         ON ts_cagg_show_src BUCKET '5m' \
+         AGGREGATE SUM(value) AS total_value",
+    )
+    .await
+    .unwrap();
+
+    let rows = srv.query_rows("SHOW CONTINUOUS AGGREGATES").await.unwrap();
+
+    assert!(
+        rows.iter()
+            .any(|r| r.first().map(String::as_str) == Some("ts_cagg_show_view")),
+        "SHOW CONTINUOUS AGGREGATES must list the just-created aggregate; \
+         got {rows:?}. Silent absence means the registration never reached \
+         a durable location and only lives in transient Data Plane state."
+    );
+}
+
+/// A continuous aggregate created before a restart must still be present
+/// after one. The registration is meaningless if it disappears on every
+/// process restart — that is the catalog-persistence gap.
+#[tokio::test]
+async fn continuous_aggregate_survives_restart() {
+    let srv = TestServer::start().await;
+    srv.exec(
+        "CREATE COLLECTION ts_cagg_persist_src \
+         COLUMNS (id TEXT, ts BIGINT TIME_KEY, value FLOAT) \
+         WITH (engine='timeseries')",
+    )
+    .await
+    .unwrap();
+    srv.exec(
+        "CREATE CONTINUOUS AGGREGATE ts_cagg_persist_view \
+         ON ts_cagg_persist_src BUCKET '5m' \
+         AGGREGATE SUM(value) AS total_value",
+    )
+    .await
+    .unwrap();
+
+    let (srv, dir) = srv.take_dir();
+    srv.graceful_shutdown().await;
+
+    let (srv2, _dir) = TestServer::open_on_path(dir).await;
+    let rows = srv2.query_rows("SHOW CONTINUOUS AGGREGATES").await.unwrap();
+
+    assert!(
+        rows.iter()
+            .any(|r| r.first().map(String::as_str) == Some("ts_cagg_persist_view")),
+        "SHOW CONTINUOUS AGGREGATES must still list the aggregate after a \
+         restart; got {rows:?}. Loss across restart is the catalog-persistence \
+         gap: the registration is in-memory Data Plane state only."
+    );
+}
+
+/// The aggregate name must resolve as a queryable relation — `SELECT
+/// … FROM <ca_name>` must not error with "unknown table". The bench's
+/// reported symptom is precisely that: queries against the
+/// materialized name return `42P01` because no target collection ever
+/// existed. Data correctness of the rolled-up rows is a separate
+/// spec (it depends on a refresh path that doesn't ship yet); this
+/// test only pins the "name resolves" half of the bug.
+#[tokio::test]
+async fn continuous_aggregate_name_is_queryable_after_create() {
+    let srv = TestServer::start().await;
+    srv.exec(
+        "CREATE COLLECTION ts_cagg_query_src \
+         COLUMNS (id TEXT, ts BIGINT TIME_KEY, value FLOAT) \
+         WITH (engine='timeseries')",
+    )
+    .await
+    .unwrap();
+
+    srv.exec(
+        "CREATE CONTINUOUS AGGREGATE ts_cagg_query_view \
+         ON ts_cagg_query_src BUCKET '1m' \
+         AGGREGATE SUM(value) AS total_value",
+    )
+    .await
+    .unwrap();
+
+    let _rows = srv
+        .query_rows("SELECT * FROM ts_cagg_query_view")
+        .await
+        .unwrap_or_else(|e| {
+            panic!(
+                "SELECT against the CA name must succeed once the aggregate \
+                 is registered (rolled-up data may be empty until a refresh \
+                 path is wired, but the name must resolve to a real relation); \
+                 got error: {e}. \"unknown table\" here is the bench's \
+                 reported symptom — no target collection was ever \
+                 materialized for the aggregate name."
+            )
+        });
+}

nodedb/tests/engine_surface_timeseries.rs

@@ -77,6 +77,7 @@ fn aggregate_output_uses_user_alias_but_having_reads_canonical_key() {
             sub_group_by: Vec::new(),
             sub_aggregates: Vec::new(),
             grouping_sets: Vec::new(),
+            sort_keys: Vec::new(),
         }),
     );
 

nodedb/tests/executor_tests/test_aggregate_aliases.rs

@@ -216,6 +216,7 @@ fn array_agg_aggregate() {
             sub_group_by: Vec::new(),
             sub_aggregates: Vec::new(),
             grouping_sets: Vec::new(),
+            sort_keys: Vec::new(),
         }),
     );
 

nodedb/tests/executor_tests/test_array_ops.rs

@@ -57,6 +57,7 @@ fn aggregate_count_reads_plain_columnar_engine_rows() {
             sub_group_by: Vec::new(),
             sub_aggregates: Vec::new(),
             grouping_sets: Vec::new(),
+            sort_keys: Vec::new(),
         }),
     );
 
@@ -122,6 +123,7 @@ fn columnar_having_uses_canonical_key_but_output_keeps_user_alias() {
             sub_group_by: Vec::new(),
             sub_aggregates: Vec::new(),
             grouping_sets: Vec::new(),
+            sort_keys: Vec::new(),
         }),
     );
 
@@ -245,6 +247,7 @@ fn aggregate_group_by_does_not_require_full_materialization() {
             sub_group_by: Vec::new(),
             sub_aggregates: Vec::new(),
             grouping_sets: Vec::new(),
+            sort_keys: Vec::new(),
         }),
     );
 

nodedb/tests/executor_tests/test_columnar_aggregate.rs

@@ -85,6 +85,7 @@ fn cross_join_uses_inline_right_scalar_aggregate_for_post_filter() {
                 sub_group_by: Vec::new(),
                 sub_aggregates: Vec::new(),
                 grouping_sets: Vec::new(),
+                sort_keys: Vec::new(),
             }))),
             inline_left_bitmap: None,
             inline_right_bitmap: None,
@@ -181,6 +182,7 @@ fn cross_join_uses_unaliased_scalar_aggregate_key_for_post_filter() {
                 sub_group_by: Vec::new(),
                 sub_aggregates: Vec::new(),
                 grouping_sets: Vec::new(),
+                sort_keys: Vec::new(),
             }))),
             inline_left_bitmap: None,
             inline_right_bitmap: None,
@@ -313,6 +315,7 @@ fn semi_join_uses_nested_scalar_subquery_result_as_inline_right() {
                     sub_group_by: Vec::new(),
                     sub_aggregates: Vec::new(),
                     grouping_sets: Vec::new(),
+                    sort_keys: Vec::new(),
                 }))),
                 inline_left_bitmap: None,
                 inline_right_bitmap: None,