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: writer.rs
    path of file: ./repos/lance/rust/lance-file/src
    
    the code of the file until where you have to start completion: // Copyright 2023 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.

mod statistics;

use std::collections::HashMap;
use std::marker::PhantomData;

use arrow_array::builder::{ArrayBuilder, PrimitiveBuilder};
use arrow_array::cast::{as_large_list_array, as_list_array, as_struct_array};
use arrow_array::types::{Int32Type, Int64Type};
use arrow_array::{Array, ArrayRef, RecordBatch, StructArray};
use arrow_buffer::ArrowNativeType;
use arrow_schema::DataType;
use async_recursion::async_recursion;
use async_trait::async_trait;
use lance_arrow::*;
use lance_core::datatypes::{Encoding, Field, Schema, SchemaCompareOptions};
use lance_core::{Error, Result};
use lance_io::encodings::{
    binary::BinaryEncoder, dictionary::DictionaryEncoder, plain::PlainEncoder, Encoder,
};
use lance_io::object_store::ObjectStore;
use lance_io::object_writer::ObjectWriter;
use lance_io::traits::{WriteExt, Writer};
use object_store::path::Path;
use snafu::{location, Location};
use tokio::io::AsyncWriteExt;

use crate::format::metadata::{Metadata, StatisticsMetadata};
use crate::format::{MAGIC, MAJOR_VERSION, MINOR_VERSION};
use crate::page_table::{PageInfo, PageTable};

/// The file format currently includes a "manifest" where it stores the schema for
/// self-describing files.  Historically this has been a table format manifest that
/// is empty except for the schema field.
///
/// Since this crate is not aware of the table format we need this to be provided
/// externally.  You should always use lance_table::io::manifest::ManifestDescribing
/// for this today.
#[async_trait]
pub trait ManifestProvider {
    /// Store the schema in the file
    ///
    /// This should just require writing the schema (or a manifest wrapper) as a proto struct
    ///
    /// Note: the dictionaries have already been written by this point and the schema should
    /// be populated with the dictionary lengths/offsets
    async fn store_schema(
        object_writer: &mut ObjectWriter,
        schema: &Schema,
    ) -> Result<Option<usize>>;
}

/// Implementation of ManifestProvider that does not store the schema
pub(crate) struct NotSelfDescribing {}

#[async_trait]
impl ManifestProvider for NotSelfDescribing {
    async fn store_schema(_: &mut ObjectWriter, _: &Schema) -> Result<Option<usize>> {
        Ok(None)
    }
}

/// [FileWriter] writes Arrow [RecordBatch] to one Lance file.
///
/// ```ignored
/// use lance::io::FileWriter;
/// use futures::stream::Stream;
///
/// let mut file_writer = FileWriter::new(object_store, &path, &schema);
/// while let Ok(batch) = stream.next().await {
///     file_writer.write(&batch).unwrap();
/// }
/// // Need to close file writer to flush buffer and footer.
/// file_writer.shutdown();
/// ```
pub struct FileWriter<M: ManifestProvider + Send + Sync> {
    pub object_writer: ObjectWr
    
    async fn test_collect_stats() {
        // Validate:
        // Only collects stats for requested columns
        // Can collect stats in nested structs
        // Won't collect stats for list columns (for now)

        let arrow_schema = ArrowSchema::new(vec![
            ArrowField::new("i", DataType::Int64, true),
            ArrowField::new("i2", DataType::Int64, true),
            ArrowField::new(
                "l",
                DataType::List(Arc::new(ArrowField::new("item", DataType::Int32, true))),
                true,
            ),
            ArrowField::new(
                "s",
                DataType::Struct(ArrowFields::from(vec![
                    ArrowField::new("si", DataType::Int64, true),
                    ArrowField::new("sb", DataType::Utf8, true),
                ])),
                true,
            ),
        ]);

        let schema = Schema::try_from(&arrow_schema).unwrap();

        let store = ObjectStore::memory();
        let path = Path::from("/foo");

        let options = FileWriterOptions {
            collect_stats_for_fields: Some(vec![0, 1, 5, 6]),
        };
        let mut file_writer =
            FileWriter::<NotSelfDescribing>::try_new(&store, &path, schema.clone(), &options)
                .await
                .unwrap();

        let batch1 = RecordBatch::try_new(
            Arc::new(arrow_schema.clone()),
            vec![
                Arc::new(Int64Array::from(vec![1, 2, 3])),
                Arc::new(Int64Array::from(vec![4, 5, 6])),
                Arc::new(ListArray::from_iter_primitive::<Int32Type, _, _>(vec![
                    Some(vec![Some(1i32), Some(2), Some(3)]),
                    Some(vec![Some(4), Some(5)]),
                    Some(vec![]),
                ])),
                Arc::new(StructArray::from(vec![
                    (
                        Arc::new(ArrowField::new("si", DataType::Int64, true)),
                        Arc::new(Int64Array::from(vec![1, 2, 3])) as ArrayRef,
                    ),
                    (
                        Arc::new(ArrowField::new("sb", DataType::Utf8, true)),
                        Arc::new(StringArray::from(vec!["a", "b", "c"])) as ArrayRef,
                    ),
                ])),
            ],
        )
        .unwrap();
        file_writer.write(&[batch1]).await.unwrap();

        let batch2 = RecordBatch::try_new(
            Arc::new(arrow_schema.clone()),
            vec![
                Arc::new(Int64Array::from(vec![5, 6])),
                Arc::new(Int64Array::from(vec![10, 11])),
                Arc::new(ListArray::from_iter_primitive::<Int32Type, _, _>(vec![
                    Some(vec![Some(1i32), Some(2), Some(3)]),
                    Some(vec![]),
                ])),
                Arc::new(StructArray::from(vec![
                    (
                        Arc::new(ArrowField::new("si", DataType::Int64, true)),
                        Arc::new(Int64Array::from(vec![4, 5])) as ArrayRef,
                    ),
                    (
                        Arc::new(ArrowField::new("sb", DataType::Utf8, true)),
                        Arc::new(StringArray::from(vec!["d", "e"])) as ArrayRef,
                    ),
                ])),
            ],
        )
        .unwrap();
        file_writer.write(&[batch2]).await.unwrap();

        file_writer.finish().await.unwrap();

        let reader = FileReader::try_new(&store, &path, schema).await.unwrap();

        let read_stats = reader.read_page_stats(&[0, 1, 5, 6]).await.unwrap();
        assert!(read_stats.is_some());
        let read_stats = read_stats.unwrap();

        let expected_stats_schema = stats_schema([
            (0, DataType::Int64),
            (1, DataType::Int64),
            (5, DataType::Int64),
            (6, DataType::Utf8),
        ]);

        assert_eq!(read_stats.schema().as_ref(), &expected_stats_schema);

        let expected_stats = stats_batch(&[
            Stats {
                field_id: 0,
                null_counts: vec![0, 0],
                min_values: Arc::new(Int64Array::from(vec![1, 5])),
                max_values: Arc::new(Int64Array::from(vec![3, 6])),
            },
            Stats {
                field_id: 1,
                null_counts: vec![0, 0],
                min_values: Arc::new(Int64Array::from(vec![4, 10])),
                max_values: Arc::new(Int64Array::from(vec![6, 11])),
            },
            Stats {
                field_id: 5,
                null_counts: vec![0, 0],
                min_values: Arc::new(Int64Array::from(vec![1, 4])),
                max_values: Arc::new(Int64Array::from(vec![3, 5])),
            },
            // FIXME: these max values shouldn't be incremented
            // https://github.com/lancedb/lance/issues/1517
            Stats {
                field_id: 6,
                null_counts: vec![0, 0],
                min_values: Arc::new(StringArray::from(vec!["a", "d"])),
                max_values: Arc::new(StringArray::from(vec!["c", "e"])),
            },
        ]);

        assert_eq!(read_stats, expected_stats);
    }