(
scx: &StatementContext,
stmt: SubscribeStatement<Aug>,
)
| 1529 | generate_extracted_config!(SubscribeOption, (Snapshot, bool), (Progress, bool)); |
| 1530 | |
| 1531 | pub fn describe_subscribe( |
| 1532 | scx: &StatementContext, |
| 1533 | stmt: SubscribeStatement<Aug>, |
| 1534 | ) -> Result<StatementDesc, PlanError> { |
| 1535 | let relation_desc = match stmt.relation { |
| 1536 | SubscribeRelation::Name(name) => { |
| 1537 | let item = scx.get_item_by_resolved_name(&name)?; |
| 1538 | match item.relation_desc() { |
| 1539 | Some(desc) => desc.into_owned(), |
| 1540 | None => sql_bail!( |
| 1541 | "'{}' cannot be subscribed to because it is a {}", |
| 1542 | name.full_name_str(), |
| 1543 | item.item_type(), |
| 1544 | ), |
| 1545 | } |
| 1546 | } |
| 1547 | SubscribeRelation::Query(query) => { |
| 1548 | let query::PlannedRootQuery { desc, .. } = |
| 1549 | query::plan_root_query(scx, query, QueryLifetime::Subscribe)?; |
| 1550 | desc |
| 1551 | } |
| 1552 | }; |
| 1553 | let SubscribeOptionExtracted { progress, .. } = stmt.options.try_into()?; |
| 1554 | let progress = progress.unwrap_or(false); |
| 1555 | let mut desc = RelationDesc::builder().with_column( |
| 1556 | "mz_timestamp", |
| 1557 | SqlScalarType::Numeric { |
| 1558 | max_scale: Some(NumericMaxScale::ZERO), |
| 1559 | } |
| 1560 | .nullable(false), |
| 1561 | ); |
| 1562 | if progress { |
| 1563 | desc = desc.with_column("mz_progressed", SqlScalarType::Bool.nullable(false)); |
| 1564 | } |
| 1565 | |
| 1566 | let debezium = matches!(stmt.output, SubscribeOutput::EnvelopeDebezium { .. }); |
| 1567 | match stmt.output { |
| 1568 | SubscribeOutput::Diffs | SubscribeOutput::WithinTimestampOrderBy { .. } => { |
| 1569 | desc = desc.with_column("mz_diff", SqlScalarType::Int64.nullable(true)); |
| 1570 | for (name, mut ty) in relation_desc.into_iter() { |
| 1571 | if progress { |
| 1572 | ty.nullable = true; |
| 1573 | } |
| 1574 | desc = desc.with_column(name, ty); |
| 1575 | } |
| 1576 | } |
| 1577 | SubscribeOutput::EnvelopeUpsert { key_columns } |
| 1578 | | SubscribeOutput::EnvelopeDebezium { key_columns } => { |
| 1579 | desc = desc.with_column("mz_state", SqlScalarType::String.nullable(true)); |
| 1580 | let key_columns = key_columns |
| 1581 | .into_iter() |
| 1582 | .map(normalize::column_name) |
| 1583 | .collect_vec(); |
| 1584 | let mut before_values_desc = RelationDesc::builder(); |
| 1585 | let mut after_values_desc = RelationDesc::builder(); |
| 1586 | |
| 1587 | // Add the key columns in the order that they're specified. |
| 1588 | for column_name in &key_columns { |
no test coverage detected