diff --git a/protos/file.proto b/protos/file.proto
index 358d53bf3a..9b28f8e837 100644
--- a/protos/file.proto
+++ b/protos/file.proto
@@ -16,6 +16,22 @@ syntax = "proto3";
 
 package lance.file;
   
+// A file descriptor that describes the contents of a Lance file
+message FileDescriptor {
+  // The schema of the file
+  Schema schema = 1;
+  // The number of rows in the file
+  uint64 length = 2;
+}
+
+// A schema which describes the data type of each of the columns
+message Schema {
+    // All fields in this file, including the nested fields.
+    repeated lance.file.Field fields = 1;
+    // Schema metadata.
+    map<string, bytes> metadata = 5;
+}
+
 // Metadata of one Lance file.
 message Metadata {
   // 4 was used for StatisticsMetadata in the past, but has been moved to prevent
diff --git a/rust/lance-file/Cargo.toml b/rust/lance-file/Cargo.toml
index 85cf917a03..3f1ae03c37 100644
--- a/rust/lance-file/Cargo.toml
+++ b/rust/lance-file/Cargo.toml
@@ -23,9 +23,11 @@ arrow-schema.workspace = true
 arrow-select.workspace = true
 async-recursion.workspace = true
 async-trait.workspace = true
+byteorder.workspace = true
 bytes.workspace = true
 datafusion-common.workspace = true
 futures.workspace = true
+lance-datagen.workspace = true
 num_cpus.workspace = true
 num-traits.workspace = true
 object_store.workspace = true
diff --git a/rust/lance-file/src/format.rs b/rust/lance-file/src/format.rs
index 366f44d2aa..ed09a5f9cf 100644
--- a/rust/lance-file/src/format.rs
+++ b/rust/lance-file/src/format.rs
@@ -29,4 +29,5 @@ pub mod metadata;
 /// These version/magic values are written at the end of Lance files (e.g. versions/1.version)
 pub const MAJOR_VERSION: i16 = 0;
 pub const MINOR_VERSION: i16 = 2;
+pub const MINOR_VERSION_TWO: u16 = 3;
 pub const MAGIC: &[u8; 4] = b"LANC";
diff --git a/rust/lance-file/src/v2.rs b/rust/lance-file/src/v2.rs
index 364d2ccf3b..f6d28e32e5 100644
--- a/rust/lance-file/src/v2.rs
+++ b/rust/lance-file/src/v2.rs
@@ -1 +1,3 @@
 pub(crate) mod io;
+pub mod reader;
+pub mod writer;
diff --git a/rust/lance-file/src/v2/reader.rs b/rust/lance-file/src/v2/reader.rs
new file mode 100644
index 0000000000..90bf44ee6b
--- /dev/null
+++ b/rust/lance-file/src/v2/reader.rs
@@ -0,0 +1,515 @@
+use std::{io::Cursor, ops::Range, sync::Arc};
+
+use arrow_array::RecordBatch;
+use arrow_schema::Schema;
+use byteorder::{ByteOrder, LittleEndian, ReadBytesExt};
+use bytes::{Bytes, BytesMut};
+use futures::{stream::BoxStream, StreamExt};
+use lance_encoding::{
+    decoder::{BatchDecodeStream, ColumnInfo, DecodeBatchScheduler, PageInfo},
+    EncodingsIo,
+};
+use prost::Message;
+use snafu::{location, Location};
+
+use lance_core::{Error, Result};
+use lance_encoding::format::pb as pbenc;
+use lance_io::{scheduler::FileScheduler, ReadBatchParams};
+use tokio::{sync::mpsc, task::JoinHandle};
+
+use crate::{
+    datatypes::{Fields, FieldsWithMeta},
+    format::{pb, pbfile, MAGIC, MAJOR_VERSION, MINOR_VERSION_TWO},
+};
+
+use super::io::LanceEncodingsIo;
+
+// TODO: Caching
+pub struct CachedFileMetadata {
+    /// The schema of the file
+    pub file_schema: Schema,
+    /// The column metadatas
+    pub column_metadatas: Vec<pbfile::ColumnMetadata>,
+    /// The number of rows in the file
+    pub num_rows: u64,
+}
+
+pub struct FileReader {
+    scheduler: Arc<LanceEncodingsIo>,
+    file_schema: Schema,
+    column_infos: Vec<ColumnInfo>,
+    num_rows: u64,
+}
+
+struct Footer {
+    column_meta_start: u64,
+    // We don't use this today because we get here by subtracting backward from global_buf_start
+    #[allow(dead_code)]
+    column_meta_offsets_start: u64,
+    global_buff_start: u64,
+    global_buff_offsets_start: u64,
+    num_global_buffers: u32,
+    num_columns: u32,
+}
+
+const FOOTER_LEN: usize = 48;
+
+impl FileReader {
+    async fn read_tail(scheduler: &FileScheduler) -> Result<(Bytes, u64)> {
+        let file_size = scheduler.reader().size().await? as u64;
+        let begin = if file_size < scheduler.reader().block_size() as u64 {
+            0
+        } else {
+            file_size - scheduler.reader().block_size() as u64
+        };
+        let tail_bytes = scheduler.submit_single(begin..file_size).await?;
+        Ok((tail_bytes, file_size))
+    }
+
+    // Checks to make sure the footer is written correctly and returns the
+    // position of the file descriptor (which comes from the footer)
+    fn decode_footer(footer_bytes: &Bytes) -> Result<Footer> {
+        let len = footer_bytes.len();
+        if len < FOOTER_LEN {
+            return Err(Error::IO {
+                message: format!(
+                    "does not have sufficient data, len: {}, bytes: {:?}",
+                    len, footer_bytes
+                ),
+                location: location!(),
+            });
+        }
+        let mut cursor = Cursor::new(footer_bytes.slice(len - FOOTER_LEN..));
+
+        let column_meta_start = cursor.read_u64::<LittleEndian>()?;
+        let column_meta_offsets_start = cursor.read_u64::<LittleEndian>()?;
+        let global_buff_start = cursor.read_u64::<LittleEndian>()?;
+        let global_buff_offsets_start = cursor.read_u64::<LittleEndian>()?;
+        let num_global_buffers = cursor.read_u32::<LittleEndian>()?;
+        let num_columns = cursor.read_u32::<LittleEndian>()?;
+        let major_version = cursor.read_u16::<LittleEndian>()?;
+        let minor_version = cursor.read_u16::<LittleEndian>()?;
+
+        if major_version != MAJOR_VERSION as u16 || minor_version != MINOR_VERSION_TWO {
+            return Err(Error::IO {
+                message: format!(
+                    "Attempt to use the lance v0.2 reader to read a file with version {}.{}",
+                    major_version, minor_version
+                ),
+                location: location!(),
+            });
+        }
+
+        let magic_bytes = footer_bytes.slice(len - 4..);
+        if magic_bytes.as_ref() != MAGIC {
+            return Err(Error::IO {
+                message: format!(
+                    "file does not appear to be a Lance file (invalid magic: {:?})",
+                    MAGIC
+                ),
+                location: location!(),
+            });
+        }
+        Ok(Footer {
+            column_meta_start,
+            column_meta_offsets_start,
+            global_buff_start,
+            global_buff_offsets_start,
+            num_global_buffers,
+            num_columns,
+        })
+    }
+
+    // TODO: Once we have coalesced I/O we should only read the column metadatas that we need
+    async fn read_all_column_metadata(
+        scheduler: &FileScheduler,
+        footer: &Footer,
+    ) -> Result<Vec<pbfile::ColumnMetadata>> {
+        let column_metadata_start = footer.column_meta_start;
+        // This range includes both the offsets table and all of the column metadata
+        // We can't just grab col_meta_start..cmo_table_start because there may be padding
+        // between the last column and the start of the cmo table.
+        let column_metadata_range = column_metadata_start..footer.global_buff_start;
+        let column_metadata_bytes = scheduler.submit_single(column_metadata_range).await?;
+
+        // cmo == column_metadata_offsets
+        let cmo_table_size = 16 * footer.num_columns as usize;
+        let cmo_table = column_metadata_bytes.slice(column_metadata_bytes.len() - cmo_table_size..);
+
+        (0..footer.num_columns)
+            .map(|col_idx| {
+                let offset = (col_idx * 16) as usize;
+                let position = LittleEndian::read_u64(&cmo_table[offset..offset + 8]);
+                let length = LittleEndian::read_u64(&cmo_table[offset + 8..offset + 16]);
+                let normalized_position = (position - column_metadata_start) as usize;
+                let normalized_end = normalized_position + (length as usize);
+                Ok(pbfile::ColumnMetadata::decode(
+                    &column_metadata_bytes[normalized_position..normalized_end],
+                )?)
+            })
+            .collect::<Result<Vec<_>>>()
+    }
+
+    async fn get_all_meta_bytes(
+        tail_bytes: Bytes,
+        file_len: u64,
+        scheduler: &FileScheduler,
+        footer: &Footer,
+    ) -> Result<Bytes> {
+        let num_bytes_needed = (file_len - footer.column_meta_start) as usize;
+        if tail_bytes.len() >= num_bytes_needed {
+            Ok(tail_bytes.slice(tail_bytes.len() - num_bytes_needed..))
+        } else {
+            let num_bytes_missing = (num_bytes_needed - tail_bytes.len()) as u64;
+            let missing_bytes = scheduler
+                .submit_single(
+                    footer.column_meta_start..footer.column_meta_start + num_bytes_missing,
+                )
+                .await;
+            let mut combined =
+                BytesMut::with_capacity(tail_bytes.len() + num_bytes_missing as usize);
+            combined.extend(missing_bytes);
+            combined.extend(tail_bytes);
+            Ok(combined.freeze())
+        }
+    }
+
+    fn decode_schema(schema_bytes: Bytes) -> Result<(u64, lance_core::datatypes::Schema)> {
+        let file_descriptor = pb::FileDescriptor::decode(schema_bytes)?;
+        let pb_schema = file_descriptor.schema.unwrap();
+        let num_rows = file_descriptor.length;
+        let fields_with_meta = FieldsWithMeta {
+            fields: Fields(pb_schema.fields),
+            metadata: pb_schema.metadata,
+        };
+        let schema = lance_core::datatypes::Schema::from(fields_with_meta);
+        Ok((num_rows, schema))
+    }
+
+    // TODO: Support projection
+    async fn read_all_metadata(
+        scheduler: &FileScheduler,
+        _projection: &Schema,
+    ) -> Result<CachedFileMetadata> {
+        // 1. read the footer
+        let (tail_bytes, file_len) = Self::read_tail(scheduler).await?;
+        let footer = Self::decode_footer(&tail_bytes)?;
+
+        let all_metadata_bytes =
+            Self::get_all_meta_bytes(tail_bytes, file_len, scheduler, &footer).await?;
+        let meta_offset = footer.column_meta_start;
+
+        // 2. read any global buffers (just the schema right now)
+        let global_bufs_table_nbytes = footer.num_global_buffers as usize * 16;
+        let global_bufs_table_start = (footer.global_buff_offsets_start - meta_offset) as usize;
+        let global_bufs_table_end = global_bufs_table_start + global_bufs_table_nbytes;
+        let global_bufs_table =
+            all_metadata_bytes.slice(global_bufs_table_start..global_bufs_table_end);
+        let mut global_bufs_cursor = Cursor::new(&global_bufs_table);
+        let schema_pos = global_bufs_cursor.read_u64::<LittleEndian>()? - meta_offset;
+        let schema_size = global_bufs_cursor.read_u64::<LittleEndian>()?;
+        let schema_end = schema_pos + schema_size;
+        let schema_bytes = all_metadata_bytes.slice(schema_pos as usize..schema_end as usize);
+        let (num_rows, schema) = Self::decode_schema(schema_bytes)?;
+
+        // Next, read the metadata for the columns
+        let column_metadatas = Self::read_all_column_metadata(scheduler, &footer).await?;
+
+        Ok(CachedFileMetadata {
+            file_schema: Schema::from(&schema),
+            column_metadatas,
+            num_rows,
+        })
+    }
+
+    pub async fn print_all_metadata(scheduler: &FileScheduler) -> Result<()> {
+        // 1. read and print the footer
+        let (tail_bytes, file_len) = Self::read_tail(scheduler).await?;
+        let footer = Self::decode_footer(&tail_bytes)?;
+
+        println!("# Footer");
+        println!();
+        println!(
+            "File version           : {}.{}",
+            MAJOR_VERSION, MINOR_VERSION_TWO
+        );
+        println!("Data bytes             : {}", footer.column_meta_start);
+        println!(
+            "Col. meta size (padded): {}",
+            footer.column_meta_offsets_start - footer.column_meta_start
+        );
+        println!(
+            "Glo. buff size (padded): {}",
+            footer.global_buff_offsets_start - footer.global_buff_start
+        );
+
+        let all_metadata_bytes =
+            Self::get_all_meta_bytes(tail_bytes, file_len, scheduler, &footer).await?;
+        let meta_offset = footer.column_meta_start;
+
+        // 2. print the global buffers
+        let global_bufs_table_nbytes = footer.num_global_buffers as usize * 16;
+        let global_bufs_table_start = (footer.global_buff_offsets_start - meta_offset) as usize;
+        let global_bufs_table_end = global_bufs_table_start + global_bufs_table_nbytes;
+        let global_bufs_table =
+            all_metadata_bytes.slice(global_bufs_table_start..global_bufs_table_end);
+        let mut global_bufs_cursor = Cursor::new(&global_bufs_table);
+        println!("Global buffers:");
+        for _ in 0..footer.num_global_buffers {
+            let buf_pos = global_bufs_cursor.read_u64::<LittleEndian>()? - meta_offset;
+            let buf_size = global_bufs_cursor.read_u64::<LittleEndian>()?;
+            println!(" * {}..{}", buf_pos, buf_pos + buf_size);
+        }
+
+        let col_meta = Self::read_all_column_metadata(scheduler, &footer).await?;
+        println!("Columns:");
+        for (idx, col) in col_meta.iter().enumerate() {
+            println!(" * Column {}", idx);
+            println!();
+            println!("   Buffers:");
+            for idx in 0..col.buffer_offsets.len() {
+                println!(
+                    "    * {}..{}",
+                    col.buffer_offsets[idx],
+                    col.buffer_offsets[idx] + col.buffer_sizes[idx]
+                );
+            }
+            println!("   Pages:");
+            println!();
+            for (page_idx, page) in col.pages.iter().enumerate() {
+                println!("   * Page {}", page_idx);
+                println!();
+                println!("     Buffers:");
+                for buf_idx in 0..page.buffer_offsets.len() {
+                    println!(
+                        "      * {}..{}",
+                        page.buffer_offsets[buf_idx],
+                        page.buffer_offsets[buf_idx] + page.buffer_sizes[buf_idx]
+                    );
+                }
+                let encoding = page.encoding.as_ref().unwrap();
+                let encoding = Self::fetch_encoding(encoding);
+                println!("     Encoding:");
+                println!();
+                let encoding_dbg = format!("{:#?}", encoding);
+                for line in encoding_dbg.lines() {
+                    println!("       {}", line);
+                }
+                println!();
+            }
+        }
+
+        Ok(())
+    }
+
+    fn fetch_encoding(encoding: &pbfile::Encoding) -> pbenc::ArrayEncoding {
+        match &encoding.style {
+            Some(pbfile::encoding::Style::Deferred(_)) => todo!(),
+            Some(pbfile::encoding::Style::Direct(encoding)) => encoding
+                .encoding
+                .as_ref()
+                .unwrap()
+                .to_msg::<pbenc::ArrayEncoding>()
+                .unwrap(),
+            None => panic!(),
+        }
+    }
+
+    fn meta_to_col_infos(meta: &CachedFileMetadata) -> Vec<ColumnInfo> {
+        meta.column_metadatas
+            .iter()
+            .map(|col_meta| {
+                let page_infos = col_meta
+                    .pages
+                    .iter()
+                    .map(|page| {
+                        let num_rows = page.length;
+                        let buffer_offsets = Arc::new(page.buffer_offsets.clone());
+                        let encoding = Self::fetch_encoding(page.encoding.as_ref().unwrap());
+                        Arc::new(PageInfo {
+                            buffer_offsets,
+                            encoding,
+                            num_rows,
+                        })
+                    })
+                    .collect::<Vec<_>>();
+                ColumnInfo {
+                    page_infos,
+                    buffer_offsets: vec![],
+                }
+            })
+            .collect::<Vec<_>>()
+    }
+
+    /// Opens a new file reader without any pre-existing knowledge
+    ///
+    /// This will read the file schema and desired column metadata from the
+    pub async fn try_open(scheduler: FileScheduler, projection: Schema) -> Result<Self> {
+        let file_metadata = Arc::new(Self::read_all_metadata(&scheduler, &projection).await?);
+        let column_infos = Self::meta_to_col_infos(&file_metadata);
+        let num_rows = file_metadata.num_rows;
+        Ok(Self {
+            scheduler: Arc::new(LanceEncodingsIo(scheduler)),
+            column_infos,
+            file_schema: file_metadata.file_schema.clone(),
+            num_rows,
+        })
+    }
+
+    async fn read_range(
+        &self,
+        range: Range<u64>,
+        batch_size: u32,
+    ) -> BoxStream<'static, JoinHandle<Result<RecordBatch>>> {
+        let mut decode_scheduler =
+            DecodeBatchScheduler::new(&self.file_schema, &self.column_infos, &vec![]);
+
+        let (tx, rx) = mpsc::unbounded_channel();
+
+        let scheduler = self.scheduler.clone() as Arc<dyn EncodingsIo>;
+        // FIXME: spawn this, change this method to sync
+        decode_scheduler
+            .schedule_range(range, tx, &scheduler)
+            .await
+            .unwrap();
+
+        BatchDecodeStream::new(rx, batch_size, self.num_rows).into_stream()
+    }
+
+    // TODO: change output to sendable record batch stream
+
+    pub async fn read_stream(
+        &self,
+        params: ReadBatchParams,
+        batch_size: u32,
+    ) -> BoxStream<'static, Result<RecordBatch>> {
+        let futures_stream = match params {
+            ReadBatchParams::Indices(_) => todo!(),
+            ReadBatchParams::Range(range) => {
+                // TODO: Make err
+                assert!((range.end as u64) < self.num_rows);
+                self.read_range(range.start as u64..range.end as u64, batch_size)
+                    .await
+            }
+            ReadBatchParams::RangeFrom(range) => {
+                self.read_range(range.start as u64..self.num_rows, batch_size)
+                    .await
+            }
+            ReadBatchParams::RangeTo(range) => {
+                // TODO: Make err
+                assert!((range.end as u64) < self.num_rows);
+                self.read_range(0..range.end as u64, batch_size).await
+            }
+            ReadBatchParams::RangeFull => self.read_range(0..self.num_rows, batch_size).await,
+        };
+        futures_stream
+            .buffered(16)
+            // JoinHandle returns Result<Result<...>> where the outer Result means the thread
+            // task panic'd so we propagate that panic here.
+            .map(|res_res| res_res.unwrap())
+            .boxed()
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use std::sync::Arc;
+
+    use arrow_array::{types::Float64Type, RecordBatchReader};
+    use arrow_schema::ArrowError;
+    use futures::StreamExt;
+    use lance_datagen::{array, gen, BatchCount, RowCount};
+    use lance_io::{object_store::ObjectStore, scheduler::StoreScheduler};
+    use object_store::path::Path;
+
+    use crate::v2::{
+        reader::FileReader,
+        writer::{FileWriter, FileWriterOptions},
+    };
+
+    #[tokio::test]
+    async fn test_round_trip() {
+        let tmp_dir = tempfile::tempdir().unwrap();
+        let tmp_path: String = tmp_dir.path().to_str().unwrap().to_owned();
+        let tmp_path = Path::parse(tmp_path).unwrap();
+        let tmp_path = tmp_path.child("some_file.lance");
+        let obj_store = Arc::new(ObjectStore::local());
+        let scheduler = StoreScheduler::new(obj_store.clone(), 8);
+
+        let reader = gen()
+            .col(Some("score".to_string()), array::rand::<Float64Type>())
+            .into_reader_rows(RowCount::from(1000), BatchCount::from(100));
+
+        let writer = obj_store.create(&tmp_path).await.unwrap();
+
+        let mut file_writer = FileWriter::try_new(
+            writer,
+            (*reader.schema()).clone(),
+            FileWriterOptions::default(),
+        )
+        .unwrap();
+
+        let schema = reader.schema();
+        let data = reader
+            .collect::<std::result::Result<Vec<_>, ArrowError>>()
+            .unwrap();
+
+        for batch in &data {
+            file_writer.write_batch(batch).await.unwrap();
+        }
+        file_writer.finish().await.unwrap();
+
+        let file_scheduler = scheduler.open_file(&tmp_path).await.unwrap();
+        FileReader::print_all_metadata(&file_scheduler)
+            .await
+            .unwrap();
+
+        for read_size in [32, 1024, 1024 * 1024] {
+            let file_scheduler = scheduler.open_file(&tmp_path).await.unwrap();
+            let file_reader = FileReader::try_open(file_scheduler, (*schema).clone())
+                .await
+                .unwrap();
+
+            let mut batch_stream = file_reader
+                .read_stream(lance_io::ReadBatchParams::RangeFull, read_size)
+                .await;
+
+            let mut total_remaining = 1000 * 100;
+            let mut expected_iter = data.iter();
+            let mut next_expected = expected_iter.next().unwrap().clone();
+            while let Some(actual) = batch_stream.next().await {
+                let mut actual = actual.unwrap();
+                let mut rows_to_verify = actual.num_rows() as u32;
+                let expected_length = total_remaining.min(read_size);
+                assert_eq!(expected_length, rows_to_verify);
+
+                while rows_to_verify > 0 {
+                    let next_slice_len = (next_expected.num_rows() as u32).min(rows_to_verify);
+                    assert_eq!(
+                        next_expected.slice(0, next_slice_len as usize),
+                        actual.slice(0, next_slice_len as usize)
+                    );
+                    total_remaining -= next_slice_len;
+                    rows_to_verify -= next_slice_len;
+                    if total_remaining > 0 {
+                        if next_slice_len == next_expected.num_rows() as u32 {
+                            next_expected = expected_iter.next().unwrap().clone();
+                        } else {
+                            next_expected = next_expected.slice(
+                                next_slice_len as usize,
+                                next_expected.num_rows() - next_slice_len as usize,
+                            );
+                        }
+                    }
+                    if rows_to_verify > 0 {
+                        actual = actual.slice(
+                            next_slice_len as usize,
+                            actual.num_rows() - next_slice_len as usize,
+                        );
+                    }
+                }
+            }
+            debug_assert_eq!(total_remaining, 0);
+        }
+    }
+}
diff --git a/rust/lance-file/src/v2/writer.rs b/rust/lance-file/src/v2/writer.rs
new file mode 100644
index 0000000000..a197984a70
--- /dev/null
+++ b/rust/lance-file/src/v2/writer.rs
@@ -0,0 +1,312 @@
+use arrow_array::RecordBatch;
+
+use arrow_schema::Schema;
+use futures::future::BoxFuture;
+use futures::stream::FuturesUnordered;
+use futures::StreamExt;
+use lance_core::{Error, Result};
+use lance_encoding::encoder::{BatchEncoder, EncodedPage, FieldEncoder};
+use lance_io::object_writer::ObjectWriter;
+use lance_io::traits::Writer;
+use prost::Message;
+use prost_types::Any;
+use snafu::{location, Location};
+use tokio::io::AsyncWriteExt;
+
+use crate::datatypes::FieldsWithMeta;
+use crate::format::pb;
+use crate::format::pbfile;
+use crate::format::pbfile::DirectEncoding;
+use crate::format::MAGIC;
+use crate::format::MAJOR_VERSION;
+use crate::format::MINOR_VERSION_TWO;
+
+#[derive(Debug, Clone, Default)]
+pub struct FileWriterOptions {
+    /// The field ids to collect statistics for.
+    ///
+    /// If None, will collect for all fields in the schema (that support stats).
+    /// If an empty vector, will not collect any statistics.
+    ///
+    // TODO: Not yet implemented
+    pub collect_stats_for_fields: Option<Vec<i32>>,
+    /// How many bytes to use for buffering column data
+    ///
+    /// When data comes in small batches the writer will buffer column data so that
+    /// larger pages can be created.  This value will be divided evenly across all of the
+    /// columns.  Generally you want this to be at least large enough to match your
+    /// filesystem's ideal read size per column.
+    ///
+    /// In some cases you might want this value to be even larger if you have highly
+    /// compressible data.  However, if this is too large, then the writer could require
+    /// a lot of memory and write performance may suffer if the CPU-expensive encoding
+    /// falls behind and can't be interleaved with the I/O expensive flushing.
+    ///
+    /// The default will use 8MiB per column which should be reasonable for most cases.
+    // TODO: Do we need to be able to set this on a per-column basis?
+    pub data_cache_bytes: Option<u64>,
+    /// How many bytes to use for buffering data to write
+    ///
+    /// After data has been encoded it is placed into an I/O cache while we wait for it
+    /// to be written to disk.  If data is arriving quickly and your I/O is very jittery
+    /// then you might want to increase this value.  However, the default of 128MiB should
+    /// be sufficient for most cases.
+    ///
+    /// If data is arriving slowly or write speed is not important you can turn this down
+    /// all the way to zero to eliminate I/O caching entirely and free up some RAM.  In this
+    /// mode each write will block until its data has been flushed.  This prevents any
+    /// interleaving of encoding and writing.
+    pub io_cache_bytes: Option<u64>,
+}
+
+pub struct FileWriter {
+    writer: ObjectWriter,
+    schema: Schema,
+    column_writers: Vec<Box<dyn FieldEncoder>>,
+    column_metadata: Vec<pbfile::ColumnMetadata>,
+    num_columns: u32,
+    rows_written: u64,
+}
+
+impl FileWriter {
+    /// Create a new FileWriter
+    pub fn try_new(
+        object_writer: ObjectWriter,
+        schema: Schema,
+        options: FileWriterOptions,
+    ) -> Result<Self> {
+        let cache_bytes_per_column = if let Some(data_cache_bytes) = options.data_cache_bytes {
+            data_cache_bytes / schema.fields.len() as u64
+        } else {
+            8 * 1024 * 1024
+        };
+
+        let encoder = BatchEncoder::try_new(&schema, cache_bytes_per_column)?;
+        let num_columns = encoder.num_columns();
+
+        let column_writers = encoder.field_encoders;
+        let column_metadata = vec![pbfile::ColumnMetadata::default(); num_columns as usize];
+
+        Ok(Self {
+            writer: object_writer,
+            schema,
+            column_writers,
+            column_metadata,
+            num_columns,
+            rows_written: 0,
+        })
+    }
+
+    async fn write_page(&mut self, encoded_page: EncodedPage) -> Result<()> {
+        let mut buffer_offsets = Vec::with_capacity(encoded_page.buffers.len());
+        let mut buffer_sizes = Vec::with_capacity(encoded_page.buffers.len());
+        for buffer in encoded_page.buffers {
+            buffer_offsets.push(self.writer.tell().await? as u64);
+            buffer_sizes.push(
+                buffer
+                    .parts
+                    .iter()
+                    .map(|part| part.len() as u64)
+                    .sum::<u64>(),
+            );
+            // Note: could potentially use write_vectored here but there is no
+            // write_vectored_all and object_store doesn't support it anyways and
+            // buffers won't normally be in *too* many parts so its unlikely to
+            // have much benefit in most cases.
+            for part in &buffer.parts {
+                self.writer.write_all(part).await?;
+            }
+        }
+        let encoded_encoding = Any::from_msg(&encoded_page.encoding)?;
+        let page = pbfile::column_metadata::Page {
+            buffer_offsets,
+            buffer_sizes,
+            encoding: Some(pbfile::Encoding {
+                style: Some(pbfile::encoding::Style::Direct(DirectEncoding {
+                    encoding: Some(encoded_encoding),
+                })),
+            }),
+            length: encoded_page.num_rows,
+        };
+        self.column_metadata[encoded_page.column_idx as usize]
+            .pages
+            .push(page);
+        Ok(())
+    }
+
+    async fn write_pages(
+        &mut self,
+        mut encoding_tasks: FuturesUnordered<BoxFuture<'static, Result<EncodedPage>>>,
+    ) -> Result<()> {
+        // As soon as an encoding task is done we write it.  There is no parallelism
+        // needed here because "writing" is really just submitting the buffer to the
+        // underlying write scheduler (either the OS or object_store's scheduler for
+        // cloud writes).  The only time we might truly await on write_page is if the
+        // scheduler's write queue is full.
+        //
+        // Also, there is no point in trying to make write_page parallel anyways
+        // because we wouldn't want buffers getting mixed up across pages.
+        while let Some(encoding_task) = encoding_tasks.next().await {
+            let encoded_page = encoding_task?;
+            self.write_page(encoded_page).await?;
+        }
+        Ok(())
+    }
+
+    /// Schedule a batch of data to be written to the file
+    ///
+    /// Note: the future returned by this method may complete before the data has been fully
+    /// flushed to the file (some data may be in the data cache or the I/O cache)
+    pub async fn write_batch(&mut self, batch: &RecordBatch) -> Result<()> {
+        let num_rows = batch.num_rows() as u64;
+        if num_rows == 0 {
+            return Ok(());
+        }
+        if num_rows > u32::MAX as u64 {
+            return Err(Error::InvalidInput {
+                source: "cannot write Lance files with more than 2^32 rows".into(),
+                location: location!(),
+            });
+        }
+        self.rows_written = match self.rows_written.checked_add(batch.num_rows() as u64) {
+            Some(rows_written) => rows_written,
+            None => {
+                return Err(Error::InvalidInput { source: format!("cannot write batch with {} rows because {} rows have already been written and Lance files cannot contain more than 2^32 rows", num_rows, self.rows_written).into(), location: location!() });
+            }
+        };
+        // First we push each array into its column writer.  This may or may not generate enough
+        // data to trigger an encoding task.  We collect any encoding tasks into a queue.
+        let encoding_tasks = self
+            .schema
+            .fields
+            .iter()
+            .zip(self.column_writers.iter_mut())
+            .map(|(field, column_writer)| {
+                let array = batch
+                    .column_by_name(field.name())
+                    .ok_or(Error::InvalidInput {
+                        source: format!(
+                            "Cannot write batch.  The batch was missing the column `{}`",
+                            field.name()
+                        )
+                        .into(),
+                        location: location!(),
+                    })?;
+                column_writer.maybe_encode(array.clone())
+            })
+            .collect::<Result<Vec<_>>>()?;
+        let encoding_tasks = encoding_tasks
+            .into_iter()
+            .flatten()
+            .collect::<FuturesUnordered<_>>();
+
+        self.write_pages(encoding_tasks).await?;
+
+        Ok(())
+    }
+
+    async fn write_column_metadata(
+        &mut self,
+        metadata: pbfile::ColumnMetadata,
+    ) -> Result<(u64, u64)> {
+        let metadata_bytes = metadata.encode_to_vec();
+        let position = self.writer.tell().await? as u64;
+        let len = metadata_bytes.len() as u64;
+        self.writer.write_all(&metadata_bytes).await?;
+        Ok((position, len))
+    }
+
+    async fn write_column_metadatas(&mut self) -> Result<Vec<(u64, u64)>> {
+        let mut metadatas = Vec::new();
+        std::mem::swap(&mut self.column_metadata, &mut metadatas);
+        let mut metadata_positions = Vec::with_capacity(metadatas.len());
+        for metadata in metadatas {
+            metadata_positions.push(self.write_column_metadata(metadata).await?);
+        }
+        Ok(metadata_positions)
+    }
+
+    fn make_file_descriptor(&self) -> Result<pb::FileDescriptor> {
+        let lance_schema = lance_core::datatypes::Schema::try_from(&self.schema)?;
+        let fields_with_meta = FieldsWithMeta::from(&lance_schema);
+        Ok(pb::FileDescriptor {
+            schema: Some(pb::Schema {
+                fields: fields_with_meta.fields.0,
+                metadata: fields_with_meta.metadata,
+            }),
+            length: self.rows_written,
+        })
+    }
+
+    async fn write_global_buffers(&mut self) -> Result<Vec<(u64, u64)>> {
+        let file_descriptor = self.make_file_descriptor()?;
+        let file_descriptor_bytes = file_descriptor.encode_to_vec();
+        let file_descriptor_len = file_descriptor_bytes.len() as u64;
+        let file_descriptor_position = self.writer.tell().await? as u64;
+        self.writer.write_all(&file_descriptor_bytes).await?;
+        Ok(vec![(file_descriptor_position, file_descriptor_len)])
+    }
+
+    /// Finishes writing the file
+    ///
+    /// This method will wait until all data has been flushed to the file.  Then it
+    /// will write the file metadata and the footer.  It will not return until all
+    /// data has been flushed and the file has been closed.
+    pub async fn finish(&mut self) -> Result<()> {
+        // 1. flush any remaining data and write out those pages
+        let encoding_tasks = self
+            .column_writers
+            .iter_mut()
+            .map(|writer| writer.flush())
+            .collect::<Result<Vec<_>>>()?;
+        let encoding_tasks = encoding_tasks
+            .into_iter()
+            .flatten()
+            .collect::<FuturesUnordered<_>>();
+        self.write_pages(encoding_tasks).await?;
+
+        // No data, so don't create a file
+        if self.rows_written == 0 {
+            self.writer.shutdown().await?;
+            return Ok(());
+        }
+
+        // 2. write the column metadatas
+        let column_metadata_start = self.writer.tell().await? as u64;
+        let metadata_positions = self.write_column_metadatas().await?;
+
+        // 3. write the column metadata position table
+        let cmo_table_start = self.writer.tell().await? as u64;
+        for (meta_pos, meta_len) in metadata_positions {
+            self.writer.write_u64_le(meta_pos).await?;
+            self.writer.write_u64_le(meta_len).await?;
+        }
+
+        // 3. write global buffers (we write the schema here)
+        let global_buffers_start = self.writer.tell().await? as u64;
+        let global_buffer_offsets = self.write_global_buffers().await?;
+        let num_global_buffers = global_buffer_offsets.len() as u32;
+
+        // write global buffers offset table
+        let gbo_table_start = self.writer.tell().await? as u64;
+        for (gbo_pos, gbo_len) in global_buffer_offsets {
+            self.writer.write_u64_le(gbo_pos).await?;
+            self.writer.write_u64_le(gbo_len).await?;
+        }
+
+        // 6. write the footer
+        self.writer.write_u64_le(column_metadata_start).await?;
+        self.writer.write_u64_le(cmo_table_start).await?;
+        self.writer.write_u64_le(global_buffers_start).await?;
+        self.writer.write_u64_le(gbo_table_start).await?;
+        self.writer.write_u32_le(num_global_buffers).await?;
+        self.writer.write_u32_le(self.num_columns).await?;
+        self.writer.write_u16_le(MAJOR_VERSION as u16).await?;
+        self.writer.write_u16_le(MINOR_VERSION_TWO).await?;
+        self.writer.write_all(MAGIC).await?;
+
+        // 7. close the writer
+        self.writer.shutdown().await?;
+        Ok(())
+    }
+}