pub fn handle_storage_command(&mut self, cmd: StorageCommand) {
        match cmd {
            StorageCommand::CreateSources(sources) => {
                for source in sources {
                    match &source.desc.connector {
                        SourceConnector::Local { .. } | SourceConnector::Log => {
                            self.storage_state.table_state.insert(
                                source.id,
                                TableState {
                                    since: Antichain::from_elem(Timestamp::minimum()),
                                    upper: Timestamp::minimum(),
                                    data: vec![],
                                    last_consolidated_size: 0,
                                },
                            );

                            // Initialize shared frontier tracking.
                            self.storage_state.source_uppers.insert(
                                source.id,
                                Rc::new(RefCell::new(Antichain::from_elem(
                                    mz_repr::Timestamp::minimum(),
                                ))),
                            );
                        }
                        SourceConnector::External {
                            connector: ExternalSourceConnector::Persist(persist_connector),
                            ..
                        } => {
                            let location = PersistLocation {
                                blob_uri: persist_connector.blob_uri.clone(),
                                consensus_uri: persist_connector.consensus_uri.clone(),
                            };

                            // TODO: Make these parts async aware?
                            let persist_client =
                                futures_executor::block_on(location.open()).unwrap();

                            let write = futures_executor::block_on(
                                persist_client
                                    .open_writer::<Row, Row, mz_repr::Timestamp, mz_repr::Diff>(
                                        persist_connector.shard_id,
                                    ),
                            )
                            .unwrap();

                            self.storage_state.persist_handles.insert(source.id, write);
                        }
                        SourceConnector::External { .. } => {
                            // Initialize shared frontier tracking.
                            self.storage_state.source_uppers.insert(
                                source.id,
                                Rc::new(RefCell::new(Antichain::from_elem(
                                    mz_repr::Timestamp::minimum(),
                                ))),
                            );

                            crate::render::build_storage_dataflow(
                                self.timely_worker,
                                self.storage_state,
                                &source.id.to_string(),
                                source.clone(),
                            );
                        }
                    }

                    self.storage_state
                        .source_descriptions
                        .insert(source.id, source.desc);

                    use timely::progress::Timestamp;
                    self.storage_state.reported_frontiers.insert(
                        source.id,
                        Antichain::from_elem(mz_repr::Timestamp::minimum()),
                    );

                    self.storage_state
                        .collection_metadata
                        .insert(source.id, source.storage_metadata);
                }
            }
            StorageCommand::AllowCompaction(list) => {
                for (id, frontier) in list {
                    if frontier.is_empty() {
                        // Indicates that we may drop `id`, as there are no more valid times to read.

                        // Drop table-related state.
                        self.storage_state.table_state.remove(&id);
                        // Clean up per-source state.
                        self.storage_state.source_descriptions.remove(&id);
                        self.storage_state.source_uppers.remove(&id);
                        self.storage_state.reported_frontiers.remove(&id);
                        self.storage_state.source_tokens.remove(&id);
                        self.storage_state.ts_source_mapping.remove(&id);
                        self.storage_state.persist_handles.remove(&id);
                    } else if let Some(table_state) = self.storage_state.table_state.get_mut(&id) {
                        table_state.since = frontier.clone();
                    }
                }
            }
        }
    }
/// Entry point for applying a storage command.
Entry point for applying a storage command.
Entry point for applying a storage command .
pub fn handle_storage_command(&mut self, cmd: StorageCommand) {
        match cmd {
            StorageCommand::CreateSources(sources) => {
                for source in sources {
                    match &source.desc.connector {
                        SourceConnector::Local { .. } | SourceConnector::Log => {
                            self.storage_state.table_state.insert(
                                source.id,
                                TableState {
                                    since: Antichain::from_elem(Timestamp::minimum()),
                                    upper: Timestamp::minimum(),
                                    data: vec![],
                                    last_consolidated_size: 0,
                                },
                            );
                            // Initialize shared frontier tracking.
                            self.storage_state.source_uppers.insert(
                                source.id,
                                Rc::new(RefCell::new(Antichain::from_elem(
                                    mz_repr::Timestamp::minimum(),
                                ))),
                            );
                        }
                        SourceConnector::External {
                            connector: ExternalSourceConnector::Persist(persist_connector),
                            ..
                        } => {
                            let location = PersistLocation {
                                blob_uri: persist_connector.blob_uri.clone(),
                                consensus_uri: persist_connector.consensus_uri.clone(),
                            };
                            // TODO: Make these parts async aware?
                            let persist_client =
                                futures_executor::block_on(location.open()).unwrap();
                            let write = futures_executor::block_on(
                                persist_client
                                    .open_writer::<Row, Row, mz_repr::Timestamp, mz_repr::Diff>(
                                        persist_connector.shard_id,
                                    ),
                            )
                            .unwrap();
                            self.storage_state.persist_handles.insert(source.id, write);
                        }
                        SourceConnector::External { .. } => {
                            // Initialize shared frontier tracking.
                            self.storage_state.source_uppers.insert(
                                source.id,
                                Rc::new(RefCell::new(Antichain::from_elem(
                                    mz_repr::Timestamp::minimum(),
                                ))),
                            );
                            crate::render::build_storage_dataflow(
                                self.timely_worker,
                                self.storage_state,
                                &source.id.to_string(),
                                source.clone(),
                            );
                        }
                    }
                    self.storage_state
                        .source_descriptions
                        .insert(source.id, source.desc);
                    use timely::progress::Timestamp;
                    self.storage_state.reported_frontiers.insert(
                        source.id,
                        Antichain::from_elem(mz_repr::Timestamp::minimum()),
                    );
                    self.storage_state
                        .collection_metadata
                        .insert(source.id, source.storage_metadata);
                }
            }
            StorageCommand::AllowCompaction(list) => {
                for (id, frontier) in list {
                    if frontier.is_empty() {
                        // Indicates that we may drop `id`, as there are no more valid times to read.
                        // Drop table-related state.
                        self.storage_state.table_state.remove(&id);
                        // Clean up per-source state.
                        self.storage_state.source_descriptions.remove(&id);
                        self.storage_state.source_uppers.remove(&id);
                        self.storage_state.reported_frontiers.remove(&id);
                        self.storage_state.source_tokens.remove(&id);
                        self.storage_state.ts_source_mapping.remove(&id);
                        self.storage_state.persist_handles.remove(&id);
                    } else if let Some(table_state) = self.storage_state.table_state.get_mut(&id) {
                        table_state.since = frontier.clone();
                    }
                }
            }
        }
    }
pub fn handle_storage_command ( & mut self , cmd : StorageCommand ) { match cmd { StorageCommand :: CreateSources ( sources ) => { for source in sources { match & source . desc . connector { SourceConnector :: Local { .. } | SourceConnector :: Log => { self . storage_state . table_state . insert ( source . id , TableState { since : Antichain :: from_elem ( Timestamp :: minimum ( ) ) , upper : Timestamp :: minimum ( ) , data : vec ! [ ] , last_consolidated_size : 0 , } , ) ; // Initialize shared frontier tracking. self . storage_state . source_uppers . insert ( source . id , Rc :: new ( RefCell :: new ( Antichain :: from_elem ( mz_repr :: Timestamp :: minimum ( ) , ) ) ) , ) ; } SourceConnector :: External { connector : ExternalSourceConnector :: Persist ( persist_connector ) , .. } => { let location = PersistLocation { blob_uri : persist_connector . blob_uri . clone ( ) , consensus_uri : persist_connector . consensus_uri . clone ( ) , } ; // TODO: Make these parts async aware? let persist_client = futures_executor :: block_on ( location . open ( ) ) . unwrap ( ) ; let write = futures_executor :: block_on ( persist_client . open_writer :: < Row , Row , mz_repr :: Timestamp , mz_repr :: Diff > ( persist_connector . shard_id , ) , ) . unwrap ( ) ; self . storage_state . persist_handles . insert ( source . id , write ) ; } SourceConnector :: External { .. } => { // Initialize shared frontier tracking. self . storage_state . source_uppers . insert ( source . id , Rc :: new ( RefCell :: new ( Antichain :: from_elem ( mz_repr :: Timestamp :: minimum ( ) , ) ) ) , ) ; crate :: render :: build_storage_dataflow ( self . timely_worker , self . storage_state , & source . id . to_string ( ) , source . clone ( ) , ) ; } } self . storage_state . source_descriptions . insert ( source . id , source . desc ) ; use timely :: progress :: Timestamp ; self . storage_state . reported_frontiers . insert ( source . id , Antichain :: from_elem ( mz_repr :: Timestamp :: minimum ( ) ) , ) ; self . storage_state . collection_metadata . insert ( source . id , source . storage_metadata ) ; } } StorageCommand :: AllowCompaction ( list ) => { for ( id , frontier ) in list { if frontier . is_empty ( ) { // Indicates that we may drop `id`, as there are no more valid times to read. // Drop table-related state. self . storage_state . table_state . remove ( & id ) ; // Clean up per-source state. self . storage_state . source_descriptions . remove ( & id ) ; self . storage_state . source_uppers . remove ( & id ) ; self . storage_state . reported_frontiers . remove ( & id ) ; self . storage_state . source_tokens . remove ( & id ) ; self . storage_state . ts_source_mapping . remove ( & id ) ; self . storage_state . persist_handles . remove ( & id ) ; } else if let Some ( table_state ) = self . storage_state . table_state . get_mut ( & id ) { table_state . since = frontier . clone ( ) ; } } } } }
Entry point for applying a storage command.
Entry point for applying a storage command .
&mut self
cmd
StorageCommand
&mut self
cmd
StorageCommand