fn parse_item(
        &self,
        id: GlobalId,
        create_sql: String,
        pcx: Option<&PlanContext>,
        table_persist_name: Option<String>,
        source_persist_details: Option<SerializedSourcePersistDetails>,
    ) -> Result<CatalogItem, anyhow::Error> {
        let stmt = sql::parse::parse(&create_sql)?.into_element();
        let plan = sql::plan::plan(pcx, &self.for_system_session(), stmt, &Params::empty())?;
        Ok(match plan {
            Plan::CreateTable(CreateTablePlan { table, .. }) => {
                assert!(
                    source_persist_details.is_none(),
                    "got some source_persist_details while we didn't expect them for a table"
                );
                let persist = self.persist.table_details_from_name(table_persist_name)?;

                CatalogItem::Table(Table {
                    create_sql: table.create_sql,
                    desc: table.desc,
                    defaults: table.defaults,
                    conn_id: None,
                    depends_on: table.depends_on,
                    persist,
                })
            }
            Plan::CreateSource(CreateSourcePlan { mut source, .. }) => {
                let mut optimizer = Optimizer::logical_optimizer();
                let optimized_expr = optimizer.optimize(source.expr)?;
                let transformed_desc = RelationDesc::new(optimized_expr.typ(), source.column_names);

                assert!(
                    table_persist_name.is_none(),
                    "got some table_persist_name while we didn't expect them for a source"
                );
                let persist_details = self.persist.source_persist_desc_from_serialized(
                    &source.connector,
                    source_persist_details,
                )?;
                // TODO: I don't like that we're injecting this into the otherwise "pristine"
                // immutable SourceConnector. We should clean this up once we have an
                // ingestd/dataflowd split, where we probably want to send SourceConnector only to
                // ingestd (and always with persistence details) and dataflowd will never see the
                // current style of SourceConnector.
                match &mut source.connector {
                    SourceConnector::External { persist, .. } => {
                        assert!(persist.is_none());
                        *persist = persist_details;
                    }
                    SourceConnector::Local { .. } => unreachable!(),
                }
                CatalogItem::Source(Source {
                    create_sql: source.create_sql,
                    optimized_expr,
                    connector: source.connector,
                    bare_desc: source.bare_desc,
                    desc: transformed_desc,
                })
            }
            Plan::CreateView(CreateViewPlan { view, .. }) => {
                let mut optimizer = Optimizer::logical_optimizer();
                let optimized_expr = optimizer.optimize(view.expr)?;
                let desc = RelationDesc::new(optimized_expr.typ(), view.column_names);
                CatalogItem::View(View {
                    create_sql: view.create_sql,
                    optimized_expr,
                    desc,
                    conn_id: None,
                    depends_on: view.depends_on,
                })
            }
            Plan::CreateIndex(CreateIndexPlan { index, .. }) => CatalogItem::Index(Index {
                create_sql: index.create_sql,
                on: index.on,
                keys: index.keys,
                conn_id: None,
                depends_on: index.depends_on,
                enabled: self.index_enabled_by_default(&id),
            }),
            Plan::CreateSink(CreateSinkPlan {
                sink,
                with_snapshot,
                ..
            }) => CatalogItem::Sink(Sink {
                create_sql: sink.create_sql,
                from: sink.from,
                connector: SinkConnectorState::Pending(sink.connector_builder),
                envelope: sink.envelope,
                with_snapshot,
                depends_on: sink.depends_on,
            }),
            Plan::CreateType(CreateTypePlan { typ, .. }) => CatalogItem::Type(Type {
                create_sql: typ.create_sql,
                inner: typ.inner.into(),
                depends_on: typ.depends_on,
            }),
            _ => bail!("catalog entry generated inappropriate plan"),
        })
    }
// Parses the given SQL string into a `CatalogItem`.
//
// The given `persist_details` are an optional description of the persisted streams that this
// source uses, if it is a persisted source.
Parses the given SQL string into a `CatalogItem`.
The given `persist_details` are an optional description of the persisted streams that this
source uses, if it is a persisted source.
Parses the given SQL string into a ` CatalogItem ` . The given ` persist_details ` are an optional description of the persisted streams that this source uses if it is a persisted source .
fn parse_item(
        &self,
        id: GlobalId,
        create_sql: String,
        pcx: Option<&PlanContext>,
        table_persist_name: Option<String>,
        source_persist_details: Option<SerializedSourcePersistDetails>,
    ) -> Result<CatalogItem, anyhow::Error> {
        let stmt = sql::parse::parse(&create_sql)?.into_element();
        let plan = sql::plan::plan(pcx, &self.for_system_session(), stmt, &Params::empty())?;
        Ok(match plan {
            Plan::CreateTable(CreateTablePlan { table, .. }) => {
                assert!(
                    source_persist_details.is_none(),
                    "got some source_persist_details while we didn't expect them for a table"
                );
                let persist = self.persist.table_details_from_name(table_persist_name)?;
                CatalogItem::Table(Table {
                    create_sql: table.create_sql,
                    desc: table.desc,
                    defaults: table.defaults,
                    conn_id: None,
                    depends_on: table.depends_on,
                    persist,
                })
            }
            Plan::CreateSource(CreateSourcePlan { mut source, .. }) => {
                let mut optimizer = Optimizer::logical_optimizer();
                let optimized_expr = optimizer.optimize(source.expr)?;
                let transformed_desc = RelationDesc::new(optimized_expr.typ(), source.column_names);
                assert!(
                    table_persist_name.is_none(),
                    "got some table_persist_name while we didn't expect them for a source"
                );
                let persist_details = self.persist.source_persist_desc_from_serialized(
                    &source.connector,
                    source_persist_details,
                )?;
                // TODO: I don't like that we're injecting this into the otherwise "pristine"
                // immutable SourceConnector. We should clean this up once we have an
                // ingestd/dataflowd split, where we probably want to send SourceConnector only to
                // ingestd (and always with persistence details) and dataflowd will never see the
                // current style of SourceConnector.
                match &mut source.connector {
                    SourceConnector::External { persist, .. } => {
                        assert!(persist.is_none());
                        *persist = persist_details;
                    }
                    SourceConnector::Local { .. } => unreachable!(),
                }
                CatalogItem::Source(Source {
                    create_sql: source.create_sql,
                    optimized_expr,
                    connector: source.connector,
                    bare_desc: source.bare_desc,
                    desc: transformed_desc,
                })
            }
            Plan::CreateView(CreateViewPlan { view, .. }) => {
                let mut optimizer = Optimizer::logical_optimizer();
                let optimized_expr = optimizer.optimize(view.expr)?;
                let desc = RelationDesc::new(optimized_expr.typ(), view.column_names);
                CatalogItem::View(View {
                    create_sql: view.create_sql,
                    optimized_expr,
                    desc,
                    conn_id: None,
                    depends_on: view.depends_on,
                })
            }
            Plan::CreateIndex(CreateIndexPlan { index, .. }) => CatalogItem::Index(Index {
                create_sql: index.create_sql,
                on: index.on,
                keys: index.keys,
                conn_id: None,
                depends_on: index.depends_on,
                enabled: self.index_enabled_by_default(&id),
            }),
            Plan::CreateSink(CreateSinkPlan {
                sink,
                with_snapshot,
                ..
            }) => CatalogItem::Sink(Sink {
                create_sql: sink.create_sql,
                from: sink.from,
                connector: SinkConnectorState::Pending(sink.connector_builder),
                envelope: sink.envelope,
                with_snapshot,
                depends_on: sink.depends_on,
            }),
            Plan::CreateType(CreateTypePlan { typ, .. }) => CatalogItem::Type(Type {
                create_sql: typ.create_sql,
                inner: typ.inner.into(),
                depends_on: typ.depends_on,
            }),
            _ => bail!("catalog entry generated inappropriate plan"),
        })
    }
fn parse_item ( & self , id : GlobalId , create_sql : String , pcx : Option < & PlanContext > , table_persist_name : Option < String > , source_persist_details : Option < SerializedSourcePersistDetails > , ) -> Result < CatalogItem , anyhow :: Error > { let stmt = sql :: parse :: parse ( & create_sql ) ? . into_element ( ) ; let plan = sql :: plan :: plan ( pcx , & self . for_system_session ( ) , stmt , & Params :: empty ( ) ) ? ; Ok ( match plan { Plan :: CreateTable ( CreateTablePlan { table , .. } ) => { assert ! ( source_persist_details is_none ( ) " " ) ; let persist = self . persist . table_details_from_name ( table_persist_name ) ? ; CatalogItem :: Table ( Table { create_sql : table . create_sql , desc : table . desc , defaults : table . defaults , conn_id : None , depends_on : table . depends_on , persist , } ) } Plan :: CreateSource ( CreateSourcePlan { mut source , .. } ) => { let mut optimizer = Optimizer :: logical_optimizer ( ) ; let optimized_expr = optimizer . optimize ( source . expr ) ? ; let transformed_desc = RelationDesc :: new ( optimized_expr . typ ( ) , source . column_names ) ; assert ! ( table_persist_name is_none ( ) " " ) ; let persist_details = self . persist . source_persist_desc_from_serialized ( & source . connector , source_persist_details , ) ? ; // TODO: I don't like that we're injecting this into the otherwise "pristine" // immutable SourceConnector. We should clean this up once we have an // ingestd/dataflowd split, where we probably want to send SourceConnector only to // ingestd (and always with persistence details) and dataflowd will never see the // current style of SourceConnector. match & mut source . connector { SourceConnector :: External { persist , .. } => { assert ! ( persist is_none ( ) ) ; * persist = persist_details ; } SourceConnector :: Local { .. } => unreachable ! ( ) , } CatalogItem :: Source ( Source { create_sql : source . create_sql , optimized_expr , connector : source . connector , bare_desc : source . bare_desc , desc : transformed_desc , } ) } Plan :: CreateView ( CreateViewPlan { view , .. } ) => { let mut optimizer = Optimizer :: logical_optimizer ( ) ; let optimized_expr = optimizer . optimize ( view . expr ) ? ; let desc = RelationDesc :: new ( optimized_expr . typ ( ) , view . column_names ) ; CatalogItem :: View ( View { create_sql : view . create_sql , optimized_expr , desc , conn_id : None , depends_on : view . depends_on , } ) } Plan :: CreateIndex ( CreateIndexPlan { index , .. } ) => CatalogItem :: Index ( Index { create_sql : index . create_sql , on : index . on , keys : index . keys , conn_id : None , depends_on : index . depends_on , enabled : self . index_enabled_by_default ( & id ) , } ) , Plan :: CreateSink ( CreateSinkPlan { sink , with_snapshot , .. } ) => CatalogItem :: Sink ( Sink { create_sql : sink . create_sql , from : sink . from , connector : SinkConnectorState :: Pending ( sink . connector_builder ) , envelope : sink . envelope , with_snapshot , depends_on : sink . depends_on , } ) , Plan :: CreateType ( CreateTypePlan { typ , .. } ) => CatalogItem :: Type ( Type { create_sql : typ . create_sql , inner : typ . inner . into ( ) , depends_on : typ . depends_on , } ) , _ => bail ! ( " " ) , } ) }
Parses the given SQL string into a `CatalogItem`.
Parses the given SQL string into a ` CatalogItem ` .
&self
id
GlobalId
create_sql
String
pcx
Option<&PlanContext>
table_persist_name
Option<String>
source_persist_details
Option<SerializedSourcePersistDetails>
&self
id
GlobalId
create_sql
String
pcx
Option<&PlanContext>
table_persist_name
Option<String>
source_persist_details
Option<SerializedSourcePersistDetails>