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: update_tests.rs
    path of file: ./repos/LakeSoul/rust/lakesoul-datafusion/src/test
    
    the code of the file until where you have to start completion: // SPDX-FileCopyrightText: 2023 LakeSoul Contributors
//
// SPDX-License-Identifier: Apache-2.0

mod update_tests {
    use std::collections::{HashSet, HashMap};
    use std::env;
    use std::path::PathBuf;
    use std::{sync::Arc, time::SystemTime};

    use arrow::util::pretty::print_batches;
    use datafusion::prelude::DataFrame;
    use datasource::parquet_source::LakeSoulParquetProvider;

    use arrow::datatypes::{SchemaRef, Schema, Field, DataType};
    use datafusion::assert_batches_eq;
    use lakesoul_io::lakesoul_io_config::{LakeSoulIOConfigBuilder, create_session_context};
    use lakesoul_io::lakesoul_reader::{SyncSendableMutableLakeSoulReader, LakeSoulReader};
    use lakesoul_io::lakesoul_writer::SyncSendableMutableLakeSoulWriter;
    use arrow::{record_batch::RecordBatch, array::{Int32Array, ArrayRef}};
    use tokio::runtime::Builder;
    use lakesoul_metadata::{MetaDataClient, MetaDataClientRef};

    use crate::error::Result;
    use proto::proto::entity::{DataCommitInfo, PartitionIn
    
    async fn perform_update(client: MetaDataClientRef, update_condition: Option<&str>) -> Result<()> {
        let mut transaction = Transaction::new();
        transaction.table_info = client.get_table_info_by_table_name("basic_case", "default").await?;

        // get candidateFiles
        let (metadata_predicates, data_predicates) = split_metadata_and_data_predicates()?;
        let candidate_files = transaction.filter_files(client.clone(), &metadata_predicates, &data_predicates)?;
        let path_to_file = candidate_files.iter().map(|f| (f.path.clone(), f.clone())).collect::<HashMap<_, _>>();

        let (add_files, expire_files) : (Vec<DataFileInfo>, Vec<DataFileInfo>) = if candidate_files.is_empty() {
            // Case 1: Do nothing if no row qualifies the partition predicates
            // that are part of Update condition
            (vec![], vec![])
        } else if data_predicates.is_empty() {
            // Case 2: Update all the rows from the files that are in the specified partitions
            // when the data filter is empty
            let files_to_rewrite = candidate_files.iter().map(|file| file.path.clone()).collect::<Vec<_>>();
            let current = SystemTime::now().duration_since(SystemTime::UNIX_EPOCH).unwrap().as_millis() as u64;
            let mut delete_files = candidate_files.iter().map(|file| file.clone()).collect::<Vec<_>>();
            delete_files.iter_mut().for_each(move |file| file.timestamp = current);
            dbg!(&delete_files);
            let rewritten_files = rewrite_files(&mut transaction, files_to_rewrite, path_to_file);
            (rewritten_files, delete_files)
        } else {
            // Case 3: Find all the affected files using the user-specified condition

            // Keep everything from the resolved target except a new LakeSoulFileIndex
            // that only involves the affected files instead of all files.


            // Case 3.1: Do nothing if no row qualifies the UPDATE condition

            // Case 3.2: Delete the old files and generate the new files containing the updated
            // values

            (vec![], vec![])
        };

        if !add_files.is_empty() || !expire_files.is_empty() {
            transaction.commit(add_files, expire_files)?;
        }
        Ok(())
    }