use super::support::*;
use crate::{
db::{
data::DecodedDataStoreKey,
predicate::{CoercionId, CompareOp, ComparePredicate, MissingRowPolicy, Predicate},
query::builder::aggregate,
},
error::ErrorClass,
traits::EntityKind,
types::Ulid,
value::Value,
};
fn field_slot_for_test<E>(field: &str) -> crate::db::query::plan::FieldSlot
where
E: EntityKind,
{
crate::db::query::plan::FieldSlot::resolve(E::MODEL, field).unwrap_or_else(|| {
crate::db::query::plan::FieldSlot::from_test_slot(usize::MAX, field.to_string())
})
}
fn remove_pushdown_row_data(id: u128) {
let raw_key = DecodedDataStoreKey::try_new::<PushdownParityEntity>(Ulid::from_u128(id))
.expect("pushdown data key should build")
.to_raw()
.expect("pushdown data key should encode");
TEST_DATA_STORE.with(|store| {
let removed = store.borrow_mut().remove(&raw_key);
assert!(
removed.is_some(),
"expected pushdown row to exist before data-only removal",
);
});
}
fn seed_simple_entities(rows: &[u128]) {
reset_store();
let save = SaveExecutor::<SimpleEntity>::new(DB, false);
for id in rows {
save.insert(SimpleEntity {
id: Ulid::from_u128(*id),
})
.expect("aggregate optimization simple seed save should succeed");
}
}
fn seed_indexed_metrics_rows(rows: &[(u128, u32, &str)]) {
reset_store();
let save = SaveExecutor::<IndexedMetricsEntity>::new(DB, false);
for (id, tag, label) in rows {
save.insert(IndexedMetricsEntity {
id: Ulid::from_u128(*id),
tag: *tag,
label: (*label).to_string(),
})
.expect("aggregate optimization indexed-metrics seed save should succeed");
}
}
fn u32_eq_predicate_strict(field: &str, value: u32) -> Predicate {
Predicate::Compare(ComparePredicate::with_coercion(
field,
CompareOp::Eq,
Value::Nat64(u64::from(value)),
CoercionId::Strict,
))
}
fn u32_in_predicate_strict(field: &str, values: &[u32]) -> Predicate {
Predicate::Compare(ComparePredicate::with_coercion(
field,
CompareOp::In,
Value::List(
values
.iter()
.map(|value| Value::Nat64(u64::from(*value)))
.collect(),
),
CoercionId::Strict,
))
}
#[test]
fn aggregate_optimizations_bytes_by_strict_mode_surfaces_missing_row_corruption() {
reset_store();
let save = SaveExecutor::<PushdownParityEntity>::new(DB, false);
for (id, group, rank) in [(8_961u128, 7u32, 20u32), (8_962, 7, 20), (8_963, 7, 30)] {
save.insert(PushdownParityEntity {
id: Ulid::from_u128(id),
group,
rank,
label: format!("g{group}-r{rank}"),
})
.expect("strict bytes_by seed row save should succeed");
}
remove_pushdown_row_data(8_962);
let load = LoadExecutor::<PushdownParityEntity>::new(DB, false);
let err = load
.bytes_by_slot(
Query::<PushdownParityEntity>::new(MissingRowPolicy::Error)
.filter_predicate(u32_eq_predicate_strict("group", 7))
.order_term(crate::db::asc("rank"))
.plan()
.map(crate::db::executor::PreparedExecutionPlan::from)
.expect("strict bytes_by plan should build"),
field_slot_for_test::<PushdownParityEntity>("rank"),
)
.expect_err("strict bytes_by should fail on missing primary rows");
assert_eq!(
err.class,
ErrorClass::Corruption,
"strict bytes_by must preserve missing-row corruption classification",
);
assert_eq!(
err.diagnostic_code(),
icydb_diagnostic_code::DiagnosticCode::StoreCorruption,
"strict bytes_by must preserve missing-row corruption diagnostics",
);
}
#[test]
fn aggregate_optimizations_index_multi_lookup_count_uses_prefix_cardinality() {
seed_indexed_metrics_rows(&[
(8_711, 10, "a"),
(8_712, 10, "b"),
(8_713, 20, "c"),
(8_714, 30, "d"),
]);
let load = LoadExecutor::<IndexedMetricsEntity>::new(DB, false);
let (count, scanned) = capture_rows_scanned_for_entity(IndexedMetricsEntity::PATH, || {
execute_count_terminal(
&load,
Query::<IndexedMetricsEntity>::new(MissingRowPolicy::Ignore)
.filter_predicate(u32_in_predicate_strict("tag", &[10, 30]))
.plan()
.map(crate::db::executor::PreparedExecutionPlan::from)
.expect("indexed IN COUNT plan should build"),
)
.expect("indexed IN COUNT should succeed")
});
assert_eq!(
count, 3,
"indexed IN COUNT should sum exact prefix cardinalities",
);
assert_eq!(
scanned, 0,
"indexed IN COUNT should not scan rows when prefix cardinality is synchronized",
);
}
#[test]
fn aggregate_optimizations_sparse_index_multi_lookup_count_batches_prefix_cardinality() {
seed_indexed_metrics_rows(&[(8_741, 10, "a"), (8_742, 10, "b"), (8_743, 20, "c")]);
let load = LoadExecutor::<IndexedMetricsEntity>::new(DB, false);
let mut values = (100..180).collect::<Vec<_>>();
values.push(10);
values.extend(180..220);
let lookups_before = IndexStore::current_prefix_cardinality_lookup_count();
let range_scans_before = IndexStore::current_range_scan_call_count();
let (count, scanned) = capture_rows_scanned_for_entity(IndexedMetricsEntity::PATH, || {
execute_count_terminal(
&load,
Query::<IndexedMetricsEntity>::new(MissingRowPolicy::Ignore)
.filter_predicate(u32_in_predicate_strict("tag", values.as_slice()))
.plan()
.map(crate::db::executor::PreparedExecutionPlan::from)
.expect("sparse indexed IN COUNT plan should build"),
)
.expect("sparse indexed IN COUNT should succeed")
});
let lookups =
IndexStore::current_prefix_cardinality_lookup_count().saturating_sub(lookups_before);
let range_scans =
IndexStore::current_range_scan_call_count().saturating_sub(range_scans_before);
assert_eq!(count, 2, "sparse indexed IN COUNT should match rows");
assert_eq!(
scanned, 0,
"sparse indexed IN COUNT should not scan rows with synchronized cardinality",
);
assert_eq!(
range_scans, 0,
"sparse indexed IN COUNT should not open index ranges for missing literals",
);
assert_eq!(
lookups, 1,
"sparse indexed IN COUNT should batch prefix-cardinality metadata probes",
);
}
#[test]
fn aggregate_optimizations_index_multi_lookup_count_limit_stops_prefix_cardinality_after_window() {
seed_indexed_metrics_rows(&[
(8_731, 10, "a"),
(8_732, 10, "b"),
(8_733, 20, "c"),
(8_734, 30, "d"),
]);
let load = LoadExecutor::<IndexedMetricsEntity>::new(DB, false);
let lookups_before = IndexStore::current_prefix_cardinality_lookup_count();
let range_scans_before = IndexStore::current_range_scan_call_count();
let (count, scanned) = capture_rows_scanned_for_entity(IndexedMetricsEntity::PATH, || {
execute_count_terminal(
&load,
Query::<IndexedMetricsEntity>::new(MissingRowPolicy::Ignore)
.filter_predicate(u32_in_predicate_strict("tag", &[10, 20, 30, 40, 50, 60]))
.order_term(crate::db::asc("id"))
.offset(1)
.limit(1)
.plan()
.map(crate::db::executor::PreparedExecutionPlan::from)
.expect("windowed indexed IN COUNT plan should build"),
)
.expect("windowed indexed IN COUNT should succeed")
});
let lookups =
IndexStore::current_prefix_cardinality_lookup_count().saturating_sub(lookups_before);
let range_scans =
IndexStore::current_range_scan_call_count().saturating_sub(range_scans_before);
assert_eq!(count, 1, "windowed indexed IN COUNT should match LIMIT");
assert_eq!(
scanned, 0,
"windowed indexed IN COUNT should not scan rows when prefix cardinality is synchronized",
);
assert_eq!(
range_scans, 0,
"windowed indexed IN COUNT should not open an index range when prefix cardinality is synchronized",
);
assert_eq!(
lookups, 1,
"windowed indexed IN COUNT should stop cardinality probes after offset + limit is proven",
);
}
#[test]
fn aggregate_optimizations_index_multi_lookup_exists_uses_prefix_cardinality() {
seed_indexed_metrics_rows(&[
(8_716, 10, "a"),
(8_717, 10, "b"),
(8_718, 20, "c"),
(8_719, 30, "d"),
]);
let load = LoadExecutor::<IndexedMetricsEntity>::new(DB, false);
let plan = Query::<IndexedMetricsEntity>::new(MissingRowPolicy::Ignore)
.filter_predicate(u32_in_predicate_strict("tag", &[10, 30]))
.plan()
.map(crate::db::executor::PreparedExecutionPlan::from)
.expect("indexed IN EXISTS plan should build");
let lookups_before = IndexStore::current_prefix_cardinality_lookup_count();
let range_scans_before = IndexStore::current_range_scan_call_count();
let (exists, scanned) = capture_rows_scanned_for_entity(IndexedMetricsEntity::PATH, || {
execute_exists_terminal(&load, plan).expect("indexed IN EXISTS should succeed")
});
let lookups =
IndexStore::current_prefix_cardinality_lookup_count().saturating_sub(lookups_before);
let range_scans =
IndexStore::current_range_scan_call_count().saturating_sub(range_scans_before);
assert!(
exists,
"indexed IN EXISTS should answer true from prefix cardinality",
);
assert_eq!(
scanned, 0,
"indexed IN EXISTS should not scan rows when prefix cardinality is synchronized",
);
assert_eq!(
range_scans, 0,
"indexed IN EXISTS should not open an index range when prefix cardinality is synchronized",
);
assert_eq!(
lookups, 1,
"indexed IN EXISTS should stop cardinality probes after the first matching prefix",
);
}
#[test]
fn aggregate_optimizations_empty_index_prefix_exists_uses_prefix_cardinality() {
seed_indexed_metrics_rows(&[(8_721, 10, "a"), (8_722, 10, "b"), (8_723, 20, "c")]);
let load = LoadExecutor::<IndexedMetricsEntity>::new(DB, false);
let plan = Query::<IndexedMetricsEntity>::new(MissingRowPolicy::Ignore)
.filter_predicate(u32_eq_predicate_strict("tag", 250))
.plan()
.map(crate::db::executor::PreparedExecutionPlan::from)
.expect("empty indexed EXISTS plan should build");
let range_scans_before = IndexStore::current_range_scan_call_count();
let (exists, scanned) = capture_rows_scanned_for_entity(IndexedMetricsEntity::PATH, || {
execute_exists_terminal(&load, plan).expect("empty indexed EXISTS should succeed")
});
let range_scans =
IndexStore::current_range_scan_call_count().saturating_sub(range_scans_before);
assert!(
!exists,
"empty indexed EXISTS should return false from prefix cardinality",
);
assert_eq!(
scanned, 0,
"empty indexed EXISTS should not scan rows when prefix cardinality is synchronized",
);
assert_eq!(
range_scans, 0,
"empty indexed EXISTS should not open an index range when prefix cardinality is synchronized",
);
}
#[test]
fn aggregate_optimizations_by_ids_count_dedups_before_windowing() {
seed_simple_entities(&[8_651, 8_652, 8_653, 8_654, 8_655]);
let load = LoadExecutor::<SimpleEntity>::new(DB, false);
let (count, scanned) = capture_rows_scanned_for_entity(SimpleEntity::PATH, || {
execute_count_terminal(
&load,
Query::<SimpleEntity>::new(MissingRowPolicy::Ignore)
.by_ids([
Ulid::from_u128(8_654),
Ulid::from_u128(8_652),
Ulid::from_u128(8_652),
Ulid::from_u128(8_651),
])
.order_term(crate::db::asc("id"))
.offset(1)
.limit(1)
.plan()
.map(crate::db::executor::PreparedExecutionPlan::from)
.expect("by_ids dedup COUNT plan should build"),
)
.expect("by_ids dedup COUNT should succeed")
});
assert_eq!(count, 1, "by_ids dedup COUNT should keep one in-window row");
assert_eq!(
scanned, 2,
"ordered by_ids dedup COUNT should scan only offset + limit rows",
);
}
#[test]
fn aggregate_optimizations_by_ids_count_desc_window_preserves_scan_budget() {
seed_simple_entities(&[8_656, 8_657, 8_658, 8_659, 8_660]);
let load = LoadExecutor::<SimpleEntity>::new(DB, false);
let (count, scanned) = capture_rows_scanned_for_entity(SimpleEntity::PATH, || {
execute_count_terminal(
&load,
Query::<SimpleEntity>::new(MissingRowPolicy::Ignore)
.by_ids([
Ulid::from_u128(8_659),
Ulid::from_u128(8_657),
Ulid::from_u128(8_657),
Ulid::from_u128(8_656),
])
.order_term(crate::db::desc("id"))
.offset(1)
.limit(1)
.plan()
.map(crate::db::executor::PreparedExecutionPlan::from)
.expect("ordered by_ids DESC COUNT plan should build"),
)
.expect("ordered by_ids DESC COUNT should succeed")
});
assert_eq!(
count, 1,
"ordered by_ids DESC COUNT should keep one in-window row",
);
assert_eq!(
scanned, 2,
"ordered by_ids DESC COUNT should scan only offset + limit rows",
);
}
#[test]
fn aggregate_optimizations_unordered_by_ids_count_preserves_canonical_dedup() {
seed_simple_entities(&[8_701, 8_702, 8_703, 8_704]);
let load = LoadExecutor::<SimpleEntity>::new(DB, false);
let count = execute_count_terminal(
&load,
Query::<SimpleEntity>::new(MissingRowPolicy::Ignore)
.by_ids([
Ulid::from_u128(8_704),
Ulid::from_u128(8_702),
Ulid::from_u128(8_702),
Ulid::from_u128(8_701),
])
.plan()
.map(crate::db::executor::PreparedExecutionPlan::from)
.expect("unordered by-ids COUNT plan should build"),
)
.expect("unordered by-ids COUNT should succeed");
assert_eq!(
count, 3,
"unordered by-ids COUNT should preserve canonical dedup semantics",
);
}
#[test]
fn aggregate_optimizations_secondary_aggregate_explain_tracks_covering_projection() {
let covering_exists = Query::<PushdownParityEntity>::new(MissingRowPolicy::Ignore)
.filter_predicate(u32_eq_predicate_strict("group", 7))
.explain_aggregate_terminal(aggregate::exists())
.expect("strict-compatible EXISTS explain should build");
let ordered_exists = Query::<PushdownParityEntity>::new(MissingRowPolicy::Ignore)
.filter_predicate(u32_eq_predicate_strict("group", 7))
.order_term(crate::db::asc("rank"))
.explain_aggregate_terminal(aggregate::exists())
.expect("ordered EXISTS explain should build");
let uncertain_exists = Query::<PushdownParityEntity>::new(MissingRowPolicy::Ignore)
.filter_predicate(Predicate::And(vec![
u32_eq_predicate_strict("group", 7),
Predicate::TextContains {
field: "label".to_string(),
value: Value::Text("keep".to_string()),
},
]))
.explain_aggregate_terminal(aggregate::exists())
.expect("strict-uncertain EXISTS explain should build");
let covering_count = Query::<PushdownParityEntity>::new(MissingRowPolicy::Ignore)
.filter_predicate(u32_eq_predicate_strict("group", 7))
.explain_aggregate_terminal(aggregate::count())
.expect("strict-compatible COUNT explain should build");
let ordered_count = Query::<PushdownParityEntity>::new(MissingRowPolicy::Ignore)
.filter_predicate(u32_eq_predicate_strict("group", 7))
.order_term(crate::db::asc("rank"))
.explain_aggregate_terminal(aggregate::count())
.expect("ordered COUNT explain should build");
assert!(
covering_exists.execution().covering_projection(),
"strict secondary EXISTS explain should mark covering projection",
);
assert!(
!ordered_exists.execution().covering_projection(),
"ordered EXISTS explain should fall back from covering projection",
);
assert!(
!uncertain_exists.execution().covering_projection(),
"strict-uncertain EXISTS explain should fall back from covering projection",
);
assert!(
covering_count.execution().covering_projection(),
"strict secondary COUNT explain should mark covering projection",
);
assert!(
!ordered_count.execution().covering_projection(),
"ordered COUNT explain should fall back from covering projection",
);
}