Skip to content

refactor: narrow asap-types test-only fields and frontend visibility (#1054 step 8b, part 2) #1099

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
zzylol merged 1 commit into from
Oct 30, 2026
Merged

refactor: narrow asap-types test-only fields and frontend visibility (#1054 step 8b, part 2) #1099

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions Cargo.lock
View file
Open in desktop

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

2 changes: 1 addition & 1 deletion crates/devtools/src/lib.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,7 +32,7 @@ pub mod rqe_atomic_costs;
pub use asap_frontend_promql::{
lower_promql_workload, lower_promql_workload_with_histograms, PromqlError,
};
pub use asap_frontend_sql::{lower_sql, lower_sql_batch, SqlCatalog, SqlError, SqlLowerer};
pub use asap_frontend_sql::{lower_sql, lower_sql_batch, SqlCatalog, SqlError};

/// Lower one developer-supplied PromQL query whose metrics are each sampled
/// every `interval_ms`, declared on each metric's source facts. Tools
Expand Down
4 changes: 2 additions & 2 deletions crates/devtools/tests/viewer_contract.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -277,8 +277,8 @@ fn viewer_categorizes_exactly_the_exported_node_kinds() {
let asap = asap_samples();
let expected: BTreeSet<&str> = non_asap
.iter()
.map(NonASAPOp::kind_name)
.chain(asap.iter().map(ASAPOp::kind_name))
.map(|op| Operator::NonASAP(op.clone()).kind_name())
.chain(asap.iter().map(|op| Operator::ASAP(op.clone()).kind_name()))
.collect();
assert_eq!(
expected.len(),
Expand Down
6 changes: 4 additions & 2 deletions crates/executor/Cargo.toml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,8 +19,10 @@ regex = "1"

[features]
# Executor internals the Planner's own tests use (the PromQL row encoding,
# the timing frontier); a deployment runtime does not need them.
test-support = []
# the timing frontier, binding views in a compiled DAG); a deployment
# runtime does not need them. The view binding reads `asap-types`' test-only
# DAG accessors.
test-support = ["planner-types/test-support"]

[dev-dependencies]
asap-executor = { path = ".", features = ["test-support"] }
Expand Down
5 changes: 3 additions & 2 deletions crates/executor/src/physical_planner/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1810,7 +1810,8 @@ pub mod test_support {
.nodes
.iter()
.filter(|node| {
nodes.contains(&(node.id as NodeId)) && !node.materialization.is_materialized_view()
nodes.contains(&(node.id as NodeId))
&& !planner_types::test_support::is_materialized_view(node.materialization)
})
.map(|node| node.id as NodeId)
.collect();
Expand All @@ -1832,7 +1833,7 @@ pub mod test_support {
}
for edge in &mut dag.edges {
if at_ingestion.contains(&(edge.producer as NodeId)) {
edge.data_state.timing = ExecutionTiming::IngestionTime;
edge.data_state_mut().timing = ExecutionTiming::IngestionTime;
}
}
let views = dag
Expand Down
53 changes: 28 additions & 25 deletions crates/executor/tests/physical_dag.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -574,15 +574,15 @@ fn bind_physical_asap_dag_before_execution() {
}),
),
],
edges: vec![PhysicalASAPDAGEdge {
producer: 0,
consumer: 1,
role: EdgeRole::Input,
intermediate_schema: (*schema).clone(),
data_state: ExecutionDataState::QUERY_ROWS,
grouping: GroupingEdgeCompatibility::NotApplicable,
window: WindowEdgeCompatibility::NotApplicable,
}],
edges: vec![PhysicalASAPDAGEdge::new(
0,
1,
EdgeRole::Input,
(*schema).clone(),
ExecutionDataState::QUERY_ROWS,
GroupingEdgeCompatibility::NotApplicable,
WindowEdgeCompatibility::NotApplicable,
)],
roots: vec![1],
scalar_roots: vec![],
};
Expand Down Expand Up @@ -704,16 +704,17 @@ fn planner_semijoin_sort_limit_contract_at_both_phases() {
guarantee: None,
retention_ms: None,
};
let edge =
|producer, consumer, role, schema: &asap_executor::values::SchemaRef| PhysicalASAPDAGEdge {
let edge = |producer, consumer, role, schema: &asap_executor::values::SchemaRef| {
PhysicalASAPDAGEdge::new(
producer,
consumer,
role,
intermediate_schema: (**schema).clone(),
data_state: ExecutionDataState::QUERY_ROWS,
grouping: GroupingEdgeCompatibility::NotApplicable,
window: WindowEdgeCompatibility::NotApplicable,
};
(**schema).clone(),
ExecutionDataState::QUERY_ROWS,
GroupingEdgeCompatibility::NotApplicable,
WindowEdgeCompatibility::NotApplicable,
)
};
let groups = GroupKeys::by(vec![0]);
let dag = PhysicalASAPDAG {
nodes: vec![
Expand Down Expand Up @@ -785,7 +786,7 @@ fn planner_semijoin_sort_limit_contract_at_both_phases() {
node.output_state.timing = phase;
}
for edge in &mut dag.edges {
edge.data_state.timing = phase;
edge.data_state().timing = phase;
}
let store = with_table(
Default::default(),
Expand Down Expand Up @@ -1140,14 +1141,16 @@ fn grouped_temporal_schema_compiles_and_executes_topk() {
guarantee: None,
retention_ms: None,
};
let edge = |producer, consumer| PhysicalASAPDAGEdge {
producer,
consumer,
role: EdgeRole::Input,
intermediate_schema: (*input).clone(),
data_state: ExecutionDataState::QUERY_ROWS,
grouping: GroupingEdgeCompatibility::NotApplicable,
window: WindowEdgeCompatibility::NotApplicable,
let edge = |producer, consumer| {
PhysicalASAPDAGEdge::new(
producer,
consumer,
EdgeRole::Input,
(*input).clone(),
ExecutionDataState::QUERY_ROWS,
GroupingEdgeCompatibility::NotApplicable,
WindowEdgeCompatibility::NotApplicable,
)
};
let dag = PhysicalASAPDAG {
nodes: vec![
Expand Down
18 changes: 9 additions & 9 deletions crates/executor/tests/plan_properties.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -77,15 +77,15 @@ fn sorted_scan(schema: &Schema, identity: &Source) -> PhysicalASAPDAG {
}),
),
],
edges: vec![PhysicalASAPDAGEdge {
producer: 0,
consumer: 1,
role: EdgeRole::Input,
intermediate_schema: schema.clone(),
data_state: ExecutionDataState::QUERY_ROWS,
grouping: GroupingEdgeCompatibility::NotApplicable,
window: WindowEdgeCompatibility::NotApplicable,
}],
edges: vec![PhysicalASAPDAGEdge::new(
0,
1,
EdgeRole::Input,
schema.clone(),
ExecutionDataState::QUERY_ROWS,
GroupingEdgeCompatibility::NotApplicable,
WindowEdgeCompatibility::NotApplicable,
)],
roots: vec![1],
scalar_roots: vec![],
}
Expand Down
12 changes: 6 additions & 6 deletions crates/executor/tests/precompute_population.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -102,15 +102,15 @@ fn edge(
consumer: usize,
role: EdgeRole,
) -> PhysicalASAPDAGEdge {
PhysicalASAPDAGEdge {
PhysicalASAPDAGEdge::new(
producer,
consumer,
role,
intermediate_schema: nodes[producer].output_schema.clone(),
data_state: nodes[producer].output_state,
grouping: GroupingEdgeCompatibility::NotApplicable,
window: WindowEdgeCompatibility::NotApplicable,
}
nodes[producer].output_schema.clone(),
nodes[producer].output_state,
GroupingEdgeCompatibility::NotApplicable,
WindowEdgeCompatibility::NotApplicable,
)
}
fn state_dag(
family: FieldDataType,
Expand Down
18 changes: 10 additions & 8 deletions crates/executor/tests/promql_binary.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -436,14 +436,16 @@ fn stored_series_evaluations_support_filters_and_sets() {
(3, 4, EdgeRole::Right),
]
.into_iter()
.map(|(producer, consumer, role)| PhysicalASAPDAGEdge {
producer,
consumer,
role,
intermediate_schema: nodes[producer].output_schema.clone(),
data_state: nodes[producer].output_state,
grouping: GroupingEdgeCompatibility::NotApplicable,
window: WindowEdgeCompatibility::NotApplicable,
.map(|(producer, consumer, role)| {
PhysicalASAPDAGEdge::new(
producer,
consumer,
role,
nodes[producer].output_schema.clone(),
nodes[producer].output_state,
GroupingEdgeCompatibility::NotApplicable,
WindowEdgeCompatibility::NotApplicable,
)
})
.collect();
let mut dag = PhysicalASAPDAG {
Expand Down
66 changes: 38 additions & 28 deletions crates/executor/tests/raw_scan.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -64,6 +64,17 @@ fn fixture() -> (planner_types::ir::NonASAPOp, SchemaRef, Vec<Batch>) {
];
(scan, output, batches)
}
/// The scan with its (absent) children as DAG node ids.
fn scan_payload(scan: planner_types::ir::NonASAPOp) -> planner_types::ir::NonASAPOp<usize> {
use planner_types::ir::Operator;
let Operator::NonASAP(scan) =
Operator::NonASAP(scan).map_children(|_| -> usize { panic!("no plan refs") })
else {
unreachable!("mapping children keeps the category")
};
scan
}

fn plan(
scan: planner_types::ir::NonASAPOp,
schema: &SchemaRef,
Expand All @@ -78,23 +89,20 @@ fn plan(
guarantee: None,
retention_ms: None,
};
let edge = |producer, consumer| PhysicalASAPDAGEdge {
producer,
consumer,
role: EdgeRole::Input,
intermediate_schema: (**schema).clone(),
data_state: state,
grouping: GroupingEdgeCompatibility::NotApplicable,
window: WindowEdgeCompatibility::NotApplicable,
let edge = |producer, consumer| {
PhysicalASAPDAGEdge::new(
producer,
consumer,
EdgeRole::Input,
(**schema).clone(),
state,
GroupingEdgeCompatibility::NotApplicable,
WindowEdgeCompatibility::NotApplicable,
)
};
PhysicalASAPDAG {
nodes: vec![
node(
0,
PhysicalASAPOperatorPayload::NonASAP(
scan.map_children(|_| -> usize { panic!("no plan refs") }),
),
),
node(0, PhysicalASAPOperatorPayload::NonASAP(scan_payload(scan))),
node(
1,
PhysicalASAPOperatorPayload::NonASAP(NonASAPOp::Sort {
Expand Down Expand Up @@ -501,14 +509,16 @@ fn windowed_plan(shift_ms: Option<i64>) -> PhysicalASAPDAG {
guarantee: None,
retention_ms: None,
};
let edge = |producer, consumer| PhysicalASAPDAGEdge {
producer,
consumer,
role: EdgeRole::Input,
intermediate_schema: schema.clone(),
data_state: state,
grouping: GroupingEdgeCompatibility::NotApplicable,
window: WindowEdgeCompatibility::NotApplicable,
let edge = |producer, consumer| {
PhysicalASAPDAGEdge::new(
producer,
consumer,
EdgeRole::Input,
schema.clone(),
state,
GroupingEdgeCompatibility::NotApplicable,
WindowEdgeCompatibility::NotApplicable,
)
};
let status_is_500 = Predicate(planner_types::ir::ScalarExpr::Compare {
semantics: planner_types::ir::ExprSemantics::Sql,
Expand Down Expand Up @@ -857,15 +867,15 @@ fn windowed_count_plan(shift_ms: Option<i64>) -> PhysicalASAPDAG {
ExecutionDataState::QUERY_ROWS,
));
let edge = |producer, consumer, schema: &planner_types::ir::schema::Schema, data_state| {
PhysicalASAPDAGEdge {
PhysicalASAPDAGEdge::new(
producer,
consumer,
role: EdgeRole::Input,
intermediate_schema: schema.clone(),
EdgeRole::Input,
schema.clone(),
data_state,
grouping: GroupingEdgeCompatibility::NotApplicable,
window: WindowEdgeCompatibility::NotApplicable,
}
GroupingEdgeCompatibility::NotApplicable,
WindowEdgeCompatibility::NotApplicable,
)
};
dag.edges.extend([
edge(2, 3, &requests_schema(), ExecutionDataState::QUERY_ROWS),
Expand Down
18 changes: 10 additions & 8 deletions crates/executor/tests/stored_pane_memory_limit.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -374,14 +374,16 @@ fn merge_over_a_stored_pane(
];
let edges = [(2, 3), (3, 0), (0, 1)]
.into_iter()
.map(|(producer, consumer): (usize, usize)| PhysicalASAPDAGEdge {
producer,
consumer,
role: EdgeRole::Input,
intermediate_schema: nodes[producer].output_schema.clone(),
data_state: nodes[producer].output_state,
grouping: GroupingEdgeCompatibility::NotApplicable,
window: WindowEdgeCompatibility::NotApplicable,
.map(|(producer, consumer): (usize, usize)| {
PhysicalASAPDAGEdge::new(
producer,
consumer,
EdgeRole::Input,
nodes[producer].output_schema.clone(),
nodes[producer].output_state,
GroupingEdgeCompatibility::NotApplicable,
WindowEdgeCompatibility::NotApplicable,
)
})
.collect();
PhysicalASAPDAG {
Expand Down
32 changes: 16 additions & 16 deletions crates/executor/tests/summary_projection.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -60,15 +60,15 @@ fn raw_node(id: usize, payload: NonASAPOp<usize>) -> PhysicalASAPDAGNode {

/// An edge carrying raw samples.
fn raw_edge(producer: usize, consumer: usize) -> PhysicalASAPDAGEdge {
PhysicalASAPDAGEdge {
PhysicalASAPDAGEdge::new(
producer,
consumer,
role: EdgeRole::Input,
intermediate_schema: sample_schema(),
data_state: ExecutionDataState::QUERY_ROWS,
grouping: GroupingEdgeCompatibility::NotApplicable,
window: WindowEdgeCompatibility::NotApplicable,
}
EdgeRole::Input,
sample_schema(),
ExecutionDataState::QUERY_ROWS,
GroupingEdgeCompatibility::NotApplicable,
WindowEdgeCompatibility::NotApplicable,
)
}

// A projection over a summary may reorder/rename its columns; recovery must retain
Expand Down Expand Up @@ -159,15 +159,15 @@ fn summary_projection_survives_recovery() {
edges: vec![
raw_edge(2, 3),
raw_edge(3, 0),
PhysicalASAPDAGEdge {
producer: 0,
consumer: 1,
role: EdgeRole::Input,
intermediate_schema: (*schema).clone(),
data_state: ExecutionDataState::INGESTION_SUMMARY,
grouping: GroupingEdgeCompatibility::NotApplicable,
window: WindowEdgeCompatibility::NotApplicable,
},
PhysicalASAPDAGEdge::new(
0,
1,
EdgeRole::Input,
(*schema).clone(),
ExecutionDataState::INGESTION_SUMMARY,
GroupingEdgeCompatibility::NotApplicable,
WindowEdgeCompatibility::NotApplicable,
),
],
roots: vec![1],
scalar_roots: vec![],
Expand Down
Loading
Loading