Function plan_create_sink

src/sql/src/plan/statement/ddl.rs:3193–3212

(
    scx: &StatementContext,
    stmt: CreateSinkStatement<Aug>,
)

);

pub fn plan_create_sink(
    scx: &StatementContext,
    stmt: CreateSinkStatement<Aug>,
) -> Result<Plan, PlanError> {
    // Check for an object in the catalog with this same name
    let Some(name) = stmt.name.clone() else {
        return Err(PlanError::MissingName(CatalogItemType::Sink));
    };
    let name = scx.allocate_qualified_name(normalize::unresolved_item_name(name)?)?;
    let full_name = scx.catalog.resolve_full_name(&name);
    let partial_name = PartialItemName::from(full_name.clone());
    if let (false, Ok(item)) = (stmt.if_not_exists, scx.catalog.resolve_item(&partial_name)) {
        return Err(PlanError::ItemAlreadyExists {
            name: full_name.to_string(),
            item_type: item.item_type(),
        });
    }

    plan_sink(scx, stmt)
}

/// This function will plan a sink as if it does not exist in the catalog. This is so the planning
/// logic is reused by both CREATE SINK and ALTER SINK planning. It is the responsibility of the