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: merge_insert.rs
    path of file: ./repos/lance/rust/lance/src/dataset/write
    
    the code of the file until where you have to start completion: // Copyright 2024 Lance Developers.
//
// Licensed 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.

//! The merge insert operation merges a batch of new data into an existing batch of old data.  This can be
//! used to implement a bulk update-or-insert (upsert) or find-or-create operation.  It can also be used to
//! replace a specified region of data with new data (e.g. replace the data for the month of January)
//!
//! The terminology for this operation can be slightly confusing.  We try and stick with the terminology from
//! SQL.  The "target table" is the OLD data that already exists.  The "source table" is the NEW data which is
//! being inserted into the dataset.
//!
//! In order for this operation to work we need to be able to match rows from the source table with rows in the
//! target table.  For example, given a row we need to know if this is a brand new row or matches an existing row.
//!
//! This match condition is currently limited to an key-match.  This means we consider a row to be a match if the
//! key columns are identical in both the source and the target.  This mea
    
    pub async fn execute(self, source: SendableRecordBatchStream) -> Result<Arc<Dataset>> {
        let session_config = SessionConfig::default().with_target_partitions(1);
        let session_ctx = SessionContext::new_with_config(session_config);
        let schema = source.schema();
        self.check_compatible_schema(&schema)?;
        let existing = session_ctx.read_lance(self.dataset.clone(), true)?;
        let new_data = session_ctx.read_one_shot(source)?;
        let merger = Merger::try_new(self.params, schema.clone())?;
        let deleted_rows = merger.deleted_rows.clone();
        let join_cols = merger
            .params
            .on
            .iter()
            .map(|c| c.as_str())
            .collect::<Vec<_>>();
        let joined = existing.join(new_data, JoinType::Full, &join_cols, &join_cols, None)?;
        let joined = joined.execute_stream().await?;

        let stream = joined
            .and_then(move |batch| merger.clone().execute_batch(batch))
            .try_flatten();

        let stream = RecordBatchStreamAdapter::new(schema, stream);

        let new_fragments = write_fragments_internal(
            self.dataset.object_store.clone(),
            &self.dataset.base,
            self.dataset.schema(),
            Box::pin(stream),
            Default::default(),
        )
        .await?;

        // Apply deletions
        let removed_row_ids = Arc::into_inner(deleted_rows).unwrap().into_inner().unwrap();
        let (old_fragments, removed_fragment_ids) =
            Self::apply_deletions(&self.dataset, &removed_row_ids).await?;

        // Commit updated and new fragments
        Self::commit(
            self.dataset,
            removed_fragment_ids,
            old_fragments,
            new_fragments,
        )
        .await
    }