pub async fn create_checkpoint(self, checkpoint_data: CheckpointData) -> Result<()> {
        let object_store = self.catalog.object_store();
        let server_id = self.catalog.server_id();
        let db_name = self.catalog.db_name();

        // sort by key (= path) for deterministic output
        let files = {
            let mut tmp: Vec<_> = checkpoint_data.files.into_iter().collect();
            tmp.sort_by_key(|(path, _metadata)| path.clone());
            tmp
        };

        // create transaction to add parquet files
        let actions: Result<Vec<_>, Error> = files
            .into_iter()
            .map(|(_, info)| {
                let CatalogParquetInfo {
                    file_size_bytes,
                    metadata,
                    path,
                } = info;

                Ok(proto::transaction::Action {
                    action: Some(proto::transaction::action::Action::AddParquet(
                        proto::AddParquet {
                            path: Some(unparse_dirs_and_filename(&path)),
                            file_size_bytes: file_size_bytes as u64,
                            metadata: metadata.to_thrift().context(MetadataEncodingFailed)?,
                        },
                    )),
                })
            })
            .collect();
        let actions = actions?;

        // assemble and store checkpoint protobuf
        let proto = proto::Transaction {
            actions,
            version: TRANSACTION_VERSION,
            uuid: self.tkey.uuid.to_string(),
            revision_counter: self.tkey.revision_counter,
            previous_uuid: self
                .previous_tkey
                .map_or_else(String::new, |tkey| tkey.uuid.to_string()),
            start_timestamp: Some(Utc::now().into()),
            encoding: proto::transaction::Encoding::Full.into(),
        };
        let path = file_path(
            &object_store,
            server_id,
            db_name,
            &self.tkey,
            FileType::Checkpoint,
        );
        store_transaction_proto(&object_store, &path, &proto).await?;

        Ok(())
    }
/// Create a checkpoint for the just-committed transaction.
///
/// Note that `checkpoint_data` must contain the state INCLUDING the just-committed transaction.
///
/// # Error Handling
/// If the checkpoint creation fails, the commit will still be treated as completed since the checkpoint is a mere
/// optimization to speed up transaction replay and allow to prune the history.
Create a checkpoint for the just-committed transaction.
Note that `checkpoint_data` must contain the state INCLUDING the just-committed transaction.
Create a checkpoint for the just - committed transaction . Note that ` checkpoint_data ` must contain the state INCLUDING the just - committed transaction .
pub async fn create_checkpoint(self, checkpoint_data: CheckpointData) -> Result<()> {
        let object_store = self.catalog.object_store();
        let server_id = self.catalog.server_id();
        let db_name = self.catalog.db_name();
        // sort by key (= path) for deterministic output
        let files = {
            let mut tmp: Vec<_> = checkpoint_data.files.into_iter().collect();
            tmp.sort_by_key(|(path, _metadata)| path.clone());
            tmp
        };
        // create transaction to add parquet files
        let actions: Result<Vec<_>, Error> = files
            .into_iter()
            .map(|(_, info)| {
                let CatalogParquetInfo {
                    file_size_bytes,
                    metadata,
                    path,
                } = info;
                Ok(proto::transaction::Action {
                    action: Some(proto::transaction::action::Action::AddParquet(
                        proto::AddParquet {
                            path: Some(unparse_dirs_and_filename(&path)),
                            file_size_bytes: file_size_bytes as u64,
                            metadata: metadata.to_thrift().context(MetadataEncodingFailed)?,
                        },
                    )),
                })
            })
            .collect();
        let actions = actions?;
        // assemble and store checkpoint protobuf
        let proto = proto::Transaction {
            actions,
            version: TRANSACTION_VERSION,
            uuid: self.tkey.uuid.to_string(),
            revision_counter: self.tkey.revision_counter,
            previous_uuid: self
                .previous_tkey
                .map_or_else(String::new, |tkey| tkey.uuid.to_string()),
            start_timestamp: Some(Utc::now().into()),
            encoding: proto::transaction::Encoding::Full.into(),
        };
        let path = file_path(
            &object_store,
            server_id,
            db_name,
            &self.tkey,
            FileType::Checkpoint,
        );
        store_transaction_proto(&object_store, &path, &proto).await?;
        Ok(())
    }
pub async fn create_checkpoint ( self , checkpoint_data : CheckpointData ) -> Result < ( ) > { let object_store = self . catalog . object_store ( ) ; let server_id = self . catalog . server_id ( ) ; let db_name = self . catalog . db_name ( ) ; // sort by key (= path) for deterministic output let files = { let mut tmp : Vec < _ > = checkpoint_data . files . into_iter ( ) . collect ( ) ; tmp . sort_by_key ( | ( path , _metadata ) | path . clone ( ) ) ; tmp } ; // create transaction to add parquet files let actions : Result < Vec < _ > , Error > = files . into_iter ( ) . map ( | ( _ , info ) | { let CatalogParquetInfo { file_size_bytes , metadata , path , } = info ; Ok ( proto :: transaction :: Action { action : Some ( proto :: transaction :: action :: Action :: AddParquet ( proto :: AddParquet { path : Some ( unparse_dirs_and_filename ( & path ) ) , file_size_bytes : file_size_bytes as u64 , metadata : metadata . to_thrift ( ) . context ( MetadataEncodingFailed ) ? , } , ) ) , } ) } ) . collect ( ) ; let actions = actions ? ; // assemble and store checkpoint protobuf let proto = proto :: Transaction { actions , version : TRANSACTION_VERSION , uuid : self . tkey . uuid . to_string ( ) , revision_counter : self . tkey . revision_counter , previous_uuid : self . previous_tkey . map_or_else ( String :: new , | tkey | tkey . uuid . to_string ( ) ) , start_timestamp : Some ( Utc :: now ( ) . into ( ) ) , encoding : proto :: transaction :: Encoding :: Full . into ( ) , } ; let path = file_path ( & object_store , server_id , db_name , & self . tkey , FileType :: Checkpoint , ) ; store_transaction_proto ( & object_store , & path , & proto ) . await ? ; Ok ( ( ) ) }
Create a checkpoint for the just-committed transaction.
Create a checkpoint for the just - committed transaction .
self
checkpoint_data
CheckpointData
self
checkpoint_data
CheckpointData
error handling
If the checkpoint creation fails, the commit will still be treated as completed since the checkpoint is a mere
optimization to speed up transaction replay and allow to prune the history.
If the checkpoint creation fails the commit will still be treated as completed since the checkpoint is a mere optimization to speed up transaction replay and allow to prune the history .