You are an expert in writing code in many different languages. Your goal is to perform code completion for the following code, keeping in mind the rest of the code and the file meta data.
    Metadata details:
    filename: alter_table.rs
    path of file: ./repos/gluesql/storages/sled-storage/src
    
    the code of the file until where you have to start completion: use {
    super::{
        error::err_into,
        fetch_schema, key,
        lock::{self, LockAcquired},
        transaction::TxPayload,
        SledStorage, Snapshot,
    },
    async_io::block_on,
    async_trait::async_trait,
    gluesql_core::{
        ast::ColumnDef,
        data::{schema::Schema, Value},
        error::{AlterTableError, Error, Result},
        executor::evaluate_stateless,
        store::{AlterTable, DataRow},
    },
    sled::transaction::ConflictableTransactionError,
    std::{iter::once, str},
    utils::Vector,
};

#[async_trait(?Send)]
impl AlterTable for SledStorage {
    async fn rename_schema(&mut self, table_name: &str, new_table_name: &str) -> Result<()> {
        let prefix = format!("data/{}/", table_name);
        let items = self
            .tree
            .scan_prefix(prefix.as_bytes())
            .map(|item| item.map_err(err_into))
            .collect::<Result<Vec<_>>>()?;

        let state = &self.state;
        let tx_timeout = self.tx_timeout;
        let tx_result = self.tree.transaction(move |tree| {
            let (txid, autocommit) = match lock::acquire(tree, state, tx_timeout)? {
                LockAcquired::Success { txid, autocommit } => (txid, autocommit),
                LockAcquired::RollbackAndRetry { lock_txid } => {
                    return Ok(TxPayload::RollbackAndRetry(lock_txid));
                }
            };

            let (old_schema_key, schema_snapshot) = fetch_schema(tree, table_name)?;
            let schema_snapshot = schema_snapshot
                .ok_or_else(|| AlterTableError::TableNotFound(table_name.to_owned()).into())
                .map_err(ConflictableTransactionError::Abort)?;

            // remove existing schema
            let (old_snapshot, old_schema) = schema_snapshot.delete(txid);
            let Schema {
                column_defs,
                indexes,
                engine,
                ..
            } = old_schema
                .ok_or_else(|| AlterTableError::TableNotFound(table_name.to_owned()).into())
                .map_err(ConflictableTransactionError::Abort)?;

            let new_schema = Schema {
                table_name: new_table_name.to_owned(),
                column_defs,
                indexes,
                engine,
            };

            bincode::serialize(&old_snapshot)
                .map_err(err_into)
                .map_err(ConflictableTransactionError::Abort)
                .map(|snapshot| tree.insert(old_schema_key.as_bytes(), snapshot))??;

            // insert new schema
            let new_snapshot = Snapshot::<Schema>::new(txid, new_schema);
            let value = bincode::serialize(&new_snapshot)
                .map_err(err_into)
                .map_err(ConflictableTransactionError::Abort)?;
            let new_schema_key = format!("schema/{}", new_table_name);
            tree.insert(new_schema_key.as_bytes(), value)?;

            // replace data
            for (old_key, value) in items.iter() {
                let new_key = str::from_utf8(old_key.as_ref())
                    .map_err(err_into)
                    .map_err(ConflictableTransactionError::Abort)?;
                let new_key = new_key.replace(table_name, new_table_name);

                let old_row_snapshot: Snapshot<DataRow> = bincode::deserialize(value)
                    .map_err(err_into)
                    .map_err(ConflictableTransactionError::Abort)?;

                let (old_row_snapshot, row) = old_row_snapshot.delete(txid);
                let row = match row {
                    Some(row) => row,
                    None => {
                        continue;
                    }
                };

                let old_row_snapshot = bincode::serialize(&old_row_snapshot)
                    .map_err(err_into)
                    .map_err(ConflictableTransactionError::Abort)?;

                let new_row_snapshot = Snapshot::<DataRow>::new(txid, row);
                let new_row_snapshot = bincode::serialize(&new_row_snapshot)
                    .map_err(err_into)
                    .map_err(ConflictableTransactionError::Abort)?;

                tree.insert(old_key, old_row_snapshot)?;
                tree.insert(new_key.as_bytes(), new_row_snapshot)?;

                if !autocommit {
                    let temp_o
    
    async fn rename_schema(&mut self, table_name: &str, new_table_name: &str) -> Result<()> {
        let prefix = format!("data/{}/", table_name);
        let items = self
            .tree
            .scan_prefix(prefix.as_bytes())
            .map(|item| item.map_err(err_into))
            .collect::<Result<Vec<_>>>()?;

        let state = &self.state;
        let tx_timeout = self.tx_timeout;
        let tx_result = self.tree.transaction(move |tree| {
            let (txid, autocommit) = match lock::acquire(tree, state, tx_timeout)? {
                LockAcquired::Success { txid, autocommit } => (txid, autocommit),
                LockAcquired::RollbackAndRetry { lock_txid } => {
                    return Ok(TxPayload::RollbackAndRetry(lock_txid));
                }
            };

            let (old_schema_key, schema_snapshot) = fetch_schema(tree, table_name)?;
            let schema_snapshot = schema_snapshot
                .ok_or_else(|| AlterTableError::TableNotFound(table_name.to_owned()).into())
                .map_err(ConflictableTransactionError::Abort)?;

            // remove existing schema
            let (old_snapshot, old_schema) = schema_snapshot.delete(txid);
            let Schema {
                column_defs,
                indexes,
                engine,
                ..
            } = old_schema
                .ok_or_else(|| AlterTableError::TableNotFound(table_name.to_owned()).into())
                .map_err(ConflictableTransactionError::Abort)?;

            let new_schema = Schema {
                table_name: new_table_name.to_owned(),
                column_defs,
                indexes,
                engine,
            };

            bincode::serialize(&old_snapshot)
                .map_err(err_into)
                .map_err(ConflictableTransactionError::Abort)
                .map(|snapshot| tree.insert(old_schema_key.as_bytes(), snapshot))??;

            // insert new schema
            let new_snapshot = Snapshot::<Schema>::new(txid, new_schema);
            let value = bincode::serialize(&new_snapshot)
                .map_err(err_into)
                .map_err(ConflictableTransactionError::Abort)?;
            let new_schema_key = format!("schema/{}", new_table_name);
            tree.insert(new_schema_key.as_bytes(), value)?;

            // replace data
            for (old_key, value) in items.iter() {
                let new_key = str::from_utf8(old_key.as_ref())
                    .map_err(err_into)
                    .map_err(ConflictableTransactionError::Abort)?;
                let new_key = new_key.replace(table_name, new_table_name);

                let old_row_snapshot: Snapshot<DataRow> = bincode::deserialize(value)
                    .map_err(err_into)
                    .map_err(ConflictableTransactionError::Abort)?;

                let (old_row_snapshot, row) = old_row_snapshot.delete(txid);
                let row = match row {
                    Some(row) => row,
                    None => {
                        continue;
                    }
                };

                let old_row_snapshot = bincode::serialize(&old_row_snapshot)
                    .map_err(err_into)
                    .map_err(ConflictableTransactionError::Abort)?;

                let new_row_snapshot = Snapshot::<DataRow>::new(txid, row);
                let new_row_snapshot = bincode::serialize(&new_row_snapshot)
                    .map_err(err_into)
                    .map_err(ConflictableTransactionError::Abort)?;

                tree.insert(old_key, old_row_snapshot)?;
                tree.insert(new_key.as_bytes(), new_row_snapshot)?;

                if !autocommit {
                    let temp_old_key = key::temp_data(txid, old_key);
                    let temp_new_key = key::temp_data_str(txid, &new_key);

                    tree.insert(temp_old_key, old_key)?;
                    tree.insert(temp_new_key, new_key.as_bytes())?;
                }
            }

            if !autocommit {
                let temp_old_key = key::temp_schema(txid, table_name);
                let temp_new_key = key::temp_schema(txid, new_table_name);

                tree.insert(temp_old_key, old_schema_key.as_bytes())?;
                tree.insert(temp_new_key, new_schema_key.as_bytes())?;
            }

            Ok(TxPayload::Success)
        });

        if self.check_retry(tx_result)? {
            self.rename_schema(table_name, new_table_name).await?;
        }

        Ok(())
    }