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: physical_plan.rs
    path of file: ./repos/arrow-datafusion/datafusion/core/benches
    
    the code of the file until where you have to start completion: // Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements.  See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership.  The ASF licenses this file
// to you under the Apache License, Version 2.0 (the
// "License"); you may not use this file except in compliance
// with the License.  You may obtain a copy of the License at
//
//   http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing,
// software distributed under the License is distributed on an
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
// KIND, either express or implied.  See the License for the
// specific language governing permissions and limitations
// under the License.

#[macro_use]
extern crate criterion;
use criterion::{BatchSize, Criterion};
extern crate arrow;
extern crate datafusion;

use std::{iter::FromIterator, sync::Arc};

use arrow::{
    array::{ArrayRef, Int64Array, StringArray},
    record_batch::RecordBatch,
};
use tokio::runtime::Runtime;

use datafusion::physical_plan::sorts::sort_preserving_merge::SortPreservingMergeExec;
use datafusion::physical_plan::{
    collect,
    expressions::{col, PhysicalSortExpr},
    memory::MemoryExec,
};
use datafusion::prelude::SessionContext;

// Initialise the operator using the provided record batches and the sort key
// as inputs. All record batches must have the same schema.
fn sort_preserving_merge_operator(
    session_ctx: Arc<SessionContext>,
    batches: Vec<RecordBatch>,
    sort: &[&str],
) {
    let schema = batches[0].sch
    
fn criterion_benchmark(c: &mut Criterion) {
    let small_batch = batches(1, 100, 10, 0).remove(0);
    let large_batch = batches(1, 1000, 1, 0).remove(0);

    let benches = vec![
        // Two batches with identical rows. They will need to be merged together
        // with one row from each batch being taken until both batches are
        // drained.
        ("interleave_batches", batches(2, 1000, 10, 1)),
        // Two batches with a small overlapping region of rows for each unique
        // sort key.
        ("merge_batches_some_overlap_small", batches(2, 1000, 10, 5)),
        // Two batches with a large overlapping region of rows for each unique
        // sort key.
        (
            "merge_batches_some_overlap_large",
            batches(2, 1000, 250, 125),
        ),
        // Two batches with no overlapping region of rows for each unique
        // sort key. For a given unique sort key all rows are drained from one
        // batch, then all the rows for the same key from the second batch.
        // This repeats until all rows are drained. There are a small number of
        // rows (10) for each unique sort key.
        ("merge_batches_no_overlap_small", batches(2, 1000, 10, 12)),
        // As above but this time there are a larger number of rows (250) for
        // each unique sort key - still no overlaps.
        ("merge_batches_no_overlap_large", batches(2, 1000, 250, 252)),
        // Merges two batches where one batch is significantly larger than the
        // other.
        (
            "merge_batches_small_into_large",
            vec![large_batch, small_batch],
        ),
    ];

    let ctx = Arc::new(SessionContext::new());
    for (name, input) in benches {
        let ctx_clone = ctx.clone();
        c.bench_function(name, move |b| {
            b.iter_batched(
                || input.clone(),
                |input| {
                    sort_preserving_merge_operator(
                        ctx_clone.clone(),
                        input,
                        &["a", "b", "c", "d"],
                    );
                },
                BatchSize::LargeInput,
            )
        });
    }
}