pub async fn init(
        &self,
        catalog_id: Uuid,
        build: BuildInfo,
        reg: &MetricsRegistry,
    ) -> Result<PersisterWithConfig, Error> {
        let persister = if self.user_table_enabled
            || self.system_table_enabled
            || self.kafka_upsert_source_enabled
        {
            let lock_reentrance_id = catalog_id.to_string();
            let lock_info = LockInfo::new(lock_reentrance_id, self.lock_info.clone())?;
            let log = ErrorLog;
            let persister = match &self.storage {
                PersistStorage::File(s) => {
                    let mut blob = FileBlob::open_exclusive((&s.blob_path).into(), lock_info)?;
                    persist::storage::check_meta_version_maybe_delete_data(&mut blob)?;
                    runtime::start(
                        RuntimeConfig::with_min_step_interval(self.min_step_interval),
                        log,
                        blob,
                        build,
                        reg,
                        self.runtime.clone(),
                    )
                }
                PersistStorage::S3(s) => {
                    let config =
                        S3BlobConfig::new(s.bucket.clone(), s.prefix.clone(), s.role_arn.clone())
                            .await?;
                    let mut blob = S3Blob::open_exclusive(config, lock_info)?;
                    persist::storage::check_meta_version_maybe_delete_data(&mut blob)?;
                    runtime::start(
                        RuntimeConfig::with_min_step_interval(self.min_step_interval),
                        log,
                        blob,
                        build,
                        reg,
                        self.runtime.clone(),
                    )
                }
            }?;
            Some(persister)
        } else {
            None
        };
        Ok(PersisterWithConfig {
            persister,
            config: self.clone(),
        })
    }
/// Initializes the persistence runtime and returns a clone-able handle for
/// interacting with it. Returns None and does not start the runtime if all
/// persistence features are disabled.
Initializes the persistence runtime and returns a clone-able handle for
interacting with it. Returns None and does not start the runtime if all
persistence features are disabled.
Initializes the persistence runtime and returns a clone - able handle for interacting with it . Returns None and does not start the runtime if all persistence features are disabled .
pub async fn init(
        &self,
        catalog_id: Uuid,
        build: BuildInfo,
        reg: &MetricsRegistry,
    ) -> Result<PersisterWithConfig, Error> {
        let persister = if self.user_table_enabled
            || self.system_table_enabled
            || self.kafka_upsert_source_enabled
        {
            let lock_reentrance_id = catalog_id.to_string();
            let lock_info = LockInfo::new(lock_reentrance_id, self.lock_info.clone())?;
            let log = ErrorLog;
            let persister = match &self.storage {
                PersistStorage::File(s) => {
                    let mut blob = FileBlob::open_exclusive((&s.blob_path).into(), lock_info)?;
                    persist::storage::check_meta_version_maybe_delete_data(&mut blob)?;
                    runtime::start(
                        RuntimeConfig::with_min_step_interval(self.min_step_interval),
                        log,
                        blob,
                        build,
                        reg,
                        self.runtime.clone(),
                    )
                }
                PersistStorage::S3(s) => {
                    let config =
                        S3BlobConfig::new(s.bucket.clone(), s.prefix.clone(), s.role_arn.clone())
                            .await?;
                    let mut blob = S3Blob::open_exclusive(config, lock_info)?;
                    persist::storage::check_meta_version_maybe_delete_data(&mut blob)?;
                    runtime::start(
                        RuntimeConfig::with_min_step_interval(self.min_step_interval),
                        log,
                        blob,
                        build,
                        reg,
                        self.runtime.clone(),
                    )
                }
            }?;
            Some(persister)
        } else {
            None
        };
        Ok(PersisterWithConfig {
            persister,
            config: self.clone(),
        })
    }
pub async fn init ( & self , catalog_id : Uuid , build : BuildInfo , reg : & MetricsRegistry , ) -> Result < PersisterWithConfig , Error > { let persister = if self . user_table_enabled || self . system_table_enabled || self . kafka_upsert_source_enabled { let lock_reentrance_id = catalog_id . to_string ( ) ; let lock_info = LockInfo :: new ( lock_reentrance_id , self . lock_info . clone ( ) ) ? ; let log = ErrorLog ; let persister = match & self . storage { PersistStorage :: File ( s ) => { let mut blob = FileBlob :: open_exclusive ( ( & s . blob_path ) . into ( ) , lock_info ) ? ; persist :: storage :: check_meta_version_maybe_delete_data ( & mut blob ) ? ; runtime :: start ( RuntimeConfig :: with_min_step_interval ( self . min_step_interval ) , log , blob , build , reg , self . runtime . clone ( ) , ) } PersistStorage :: S3 ( s ) => { let config = S3BlobConfig :: new ( s . bucket . clone ( ) , s . prefix . clone ( ) , s . role_arn . clone ( ) ) . await ? ; let mut blob = S3Blob :: open_exclusive ( config , lock_info ) ? ; persist :: storage :: check_meta_version_maybe_delete_data ( & mut blob ) ? ; runtime :: start ( RuntimeConfig :: with_min_step_interval ( self . min_step_interval ) , log , blob , build , reg , self . runtime . clone ( ) , ) } } ? ; Some ( persister ) } else { None } ; Ok ( PersisterWithConfig { persister , config : self . clone ( ) , } ) }
Initializes the persistence runtime and returns a clone-able handle for
interacting with it.
Initializes the persistence runtime and returns a clone - able handle for interacting with it .
&self
catalog_id
Uuid
build
BuildInfo
reg
&MetricsRegistry
&self
catalog_id
Uuid
build
BuildInfo
reg
&MetricsRegistry