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: upsert_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 upsert_with_io_config_tests {
    use chrono::naive::NaiveDate;
    use std::env;
    use std::path::PathBuf;
    use std::sync::Arc;
    use std::time::SystemTime;

    use arrow::array::{ArrayRef, Int32Array, StringArray, TimestampMicrosecondArray};
    use arrow::datatypes::{Field, Schema, SchemaRef, TimeUnit};
    use arrow::record_batch::RecordBatch;
    use datafusion::assert_batches_eq;
    use lakesoul_io::lakesoul_io_config::LakeSoulIOConfigBuilder;
    use lakesoul_io::lakesoul_reader::{LakeSoulReader, SyncSendableMutableLakeSoulReader};
    use lakesoul_io::lakesoul_writer::SyncSendableMutableLakeSoulWriter;
    use tokio::runtime::Builder;

    enum StrOrI32 {
        V1(&'static str),
        V2(i32),
    }

    fn init_table(batch: RecordBatch, table_name: &str, pks: Vec<String>) -> LakeSoulIOConfigBuilder {
        let builder = LakeSoulIOConfigBuilder::new()
            .with_schema(batch.schema())
            .with_primary_keys(pks);
       
    
    fn test_filter_requested_columns_upsert_3_times_i32() {
        let table_name = "filter_requested_columns_upsert_3_times";
        let builder = init_table(
            create_batch_i32(
                vec!["range", "hash", "value", "name", "age"],
                vec![
                    &[20201101, 20201101, 20201101, 20201101],
                    &[1, 2, 3, 4],
                    &[1, 2, 3, 4],
                    &[1, 2, 3, 4],
                    &[1, 2, 3, 4],
                ],
            ),
            table_name,
            vec!["range".to_string(), "hash".to_string()],
        );

        let builder = execute_upsert(
            create_batch_i32(
                vec!["range", "hash", "value", "name"],
                vec![
                    &[20201102, 20201102, 20201102],
                    &[1, 2, 3],
                    &[111, 222, 333],
                    &[11, 22, 33],
                ],
            ),
            table_name,
            builder,
        );

        let builder = execute_upsert(
            create_batch_i32(
                vec!["range", "hash", "value"],
                vec![&[20201102, 20201102, 20201102], &[1, 3, 4], &[11, 33, 44]],
            ),
            table_name,
            builder,
        );

        /***
            !!! Error all below conditions are satisfied
            1. entire 'value' column is null
            2. filter exists
            3. filter pushed down into TableProvider with TableProviderFilterPushDown::Inexact
            4. SessionConfig.execution.parquet.pruning = true （equivalent SessionConfig.with_parquet_pruning(true)）
            5. SessionConfig.execution.parquet.enable_page_index = true
            6.
        */
        check_upsert(
            create_batch_optional_i32(
                vec!["range", "hash", "age", "name", "value"],
                vec![
                    &[Some(20201102), Some(20201102)],
                    &[Some(1), Some(3)],
                    &[Some(111), Some(333)],
                    &[Some(11), Some(33)],
                    &[None, Some(3333)],
                ],
            ),
            // &[None, None]]),
            table_name,
            vec!["range", "hash", "value", "name", "age"],
            Some("and(noteq(range, null), eq(range, 20201102))".to_string()),
            // None,
            builder.clone(),
            &[
                "+----------+------+-------+------+-----+",
                "| range    | hash | value | name | age |",
                "+----------+------+-------+------+-----+",
                "| 20201102 | 1    |       | 11   | 111 |",
                "| 20201102 | 2    | 222   | 22   |     |",
                "| 20201102 | 3    | 3333  | 33   | 333 |",
                // "| 20201102 | 3    |       | 33   | 333 |",
                "| 20201102 | 4    | 44    |      |     |",
                "+----------+------+-------+------+-----+",
            ],
        );
    }