(
scx: &StatementContext,
SubscribeStatement {
relation,
options,
as_of,
up_to,
output,
}: SubscribeStatement<Aug>,
params: &Params,
copy_to
| 1625 | } |
| 1626 | |
| 1627 | pub fn plan_subscribe( |
| 1628 | scx: &StatementContext, |
| 1629 | SubscribeStatement { |
| 1630 | relation, |
| 1631 | options, |
| 1632 | as_of, |
| 1633 | up_to, |
| 1634 | output, |
| 1635 | }: SubscribeStatement<Aug>, |
| 1636 | params: &Params, |
| 1637 | copy_to: Option<CopyFormat>, |
| 1638 | ) -> Result<Plan, PlanError> { |
| 1639 | let (from, desc, scope) = match relation { |
| 1640 | SubscribeRelation::Name(name) => { |
| 1641 | let item = scx.get_item_by_resolved_name(&name)?; |
| 1642 | let Some(desc) = item.relation_desc() else { |
| 1643 | sql_bail!( |
| 1644 | "'{}' cannot be subscribed to because it is a {}", |
| 1645 | name.full_name_str(), |
| 1646 | item.item_type(), |
| 1647 | ); |
| 1648 | }; |
| 1649 | let item_name = match name { |
| 1650 | ResolvedItemName::Item { full_name, .. } => Some(full_name.into()), |
| 1651 | _ => None, |
| 1652 | }; |
| 1653 | let scope = Scope::from_source(item_name, desc.iter().map(|(name, _type)| name)); |
| 1654 | ( |
| 1655 | SubscribeFrom::Id(item.global_id()), |
| 1656 | desc.into_owned(), |
| 1657 | scope, |
| 1658 | ) |
| 1659 | } |
| 1660 | SubscribeRelation::Query(query) => { |
| 1661 | #[allow(deprecated)] // TODO(aalexandrov): Use HirRelationExpr in Subscribe |
| 1662 | let query::PlannedRootQuery { |
| 1663 | mut expr, |
| 1664 | desc, |
| 1665 | finishing, |
| 1666 | scope, |
| 1667 | } = query::plan_root_query(scx, query, QueryLifetime::Subscribe)?; |
| 1668 | expr.bind_parameters_and_simplify_offset(scx, QueryLifetime::Subscribe, params)?; |
| 1669 | let query = query::PlannedRootQuery { |
| 1670 | expr, |
| 1671 | desc, |
| 1672 | finishing, |
| 1673 | scope, |
| 1674 | }; |
| 1675 | // There's no way to apply finishing operations to a `SUBSCRIBE` directly, so the |
| 1676 | // finishing should have already been turned into a `TopK` by |
| 1677 | // `plan_query` / `plan_root_query`, upon seeing the `QueryLifetime::Subscribe`. |
| 1678 | assert!(HirRelationExpr::is_trivial_row_set_finishing_hir( |
| 1679 | &query.finishing, |
| 1680 | query.desc.arity() |
| 1681 | )); |
| 1682 | let desc = query.desc.clone(); |
| 1683 | ( |
| 1684 | SubscribeFrom::Query { |
no test coverage detected