feat: Add support for reading/writing vortex in iceberg

Cargo.toml

@@ -152,5 +152,6 @@ url = "2.5.7"
 uuid = { version = "1.18", features = ["v7"] }
 volo = "0.10.6"
 volo-thrift = "0.10.8"
+vortex = { version = "0.78", features = ["tokio"] }
 zeroize = "1.7"
 zstd = "0.13.3"

bindings/python/src/data_file.rs

@@ -71,6 +71,7 @@ impl PyDataFile {
             DataFileFormat::Orc => "orc",
             DataFileFormat::Parquet => "parquet",
             DataFileFormat::Puffin => "puffin",
+            DataFileFormat::Vortex => "vortex",
         }
     }
 

crates/iceberg/Cargo.toml

@@ -80,6 +80,7 @@ typed-builder = { workspace = true }
 typetag = { workspace = true }
 url = { workspace = true }
 uuid = { workspace = true }
+vortex = { workspace = true }
 zeroize = { workspace = true }
 zstd = { workspace = true }
 

crates/iceberg/public-api.txt

@@ -1370,6 +1370,7 @@ pub iceberg::spec::DataFileFormat::Avro
 pub iceberg::spec::DataFileFormat::Orc
 pub iceberg::spec::DataFileFormat::Parquet
 pub iceberg::spec::DataFileFormat::Puffin
+pub iceberg::spec::DataFileFormat::Vortex
 impl core::clone::Clone for iceberg::spec::DataFileFormat
 pub fn iceberg::spec::DataFileFormat::clone(&self) -> iceberg::spec::DataFileFormat
 impl core::cmp::Eq for iceberg::spec::DataFileFormat
@@ -3265,18 +3266,42 @@ pub fn iceberg::writer::file_writer::ParquetWriterBuilder::fmt(&self, f: &mut co
 impl iceberg::writer::file_writer::FileWriterBuilder for iceberg::writer::file_writer::ParquetWriterBuilder
 pub type iceberg::writer::file_writer::ParquetWriterBuilder::R = iceberg::writer::file_writer::ParquetWriter
 pub async fn iceberg::writer::file_writer::ParquetWriterBuilder::build(&self, output_file: iceberg::io::OutputFile) -> iceberg::Result<Self::R>
+pub struct iceberg::writer::file_writer::VortexWriter
+impl iceberg::writer::CurrentFileStatus for iceberg::writer::file_writer::VortexWriter
+pub fn iceberg::writer::file_writer::VortexWriter::current_file_path(&self) -> alloc::string::String
+pub fn iceberg::writer::file_writer::VortexWriter::current_row_num(&self) -> usize
+pub fn iceberg::writer::file_writer::VortexWriter::current_written_size(&self) -> usize
+impl iceberg::writer::file_writer::FileWriter for iceberg::writer::file_writer::VortexWriter
+pub async fn iceberg::writer::file_writer::VortexWriter::close(self) -> iceberg::Result<alloc::vec::Vec<iceberg::spec::DataFileBuilder>>
+pub async fn iceberg::writer::file_writer::VortexWriter::write(&mut self, batch: &arrow_array::record_batch::RecordBatch) -> iceberg::Result<()>
+pub struct iceberg::writer::file_writer::VortexWriterBuilder
+impl iceberg::writer::file_writer::VortexWriterBuilder
+pub fn iceberg::writer::file_writer::VortexWriterBuilder::new(schema: iceberg::spec::SchemaRef, session: vortex_session::session::VortexSession) -> Self
+impl core::clone::Clone for iceberg::writer::file_writer::VortexWriterBuilder
+pub fn iceberg::writer::file_writer::VortexWriterBuilder::clone(&self) -> iceberg::writer::file_writer::VortexWriterBuilder
+impl core::fmt::Debug for iceberg::writer::file_writer::VortexWriterBuilder
+pub fn iceberg::writer::file_writer::VortexWriterBuilder::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result
+impl iceberg::writer::file_writer::FileWriterBuilder for iceberg::writer::file_writer::VortexWriterBuilder
+pub type iceberg::writer::file_writer::VortexWriterBuilder::R = iceberg::writer::file_writer::VortexWriter
+pub async fn iceberg::writer::file_writer::VortexWriterBuilder::build(&self, output_file: iceberg::io::OutputFile) -> iceberg::Result<Self::R>
 pub trait iceberg::writer::file_writer::FileWriter<O>: core::marker::Send + iceberg::writer::CurrentFileStatus + 'static
 pub fn iceberg::writer::file_writer::FileWriter::close(self) -> impl core::future::future::Future<Output = iceberg::Result<O>> + core::marker::Send
 pub fn iceberg::writer::file_writer::FileWriter::write(&mut self, batch: &arrow_array::record_batch::RecordBatch) -> impl core::future::future::Future<Output = iceberg::Result<()>> + core::marker::Send
 impl iceberg::writer::file_writer::FileWriter for iceberg::writer::file_writer::ParquetWriter
 pub async fn iceberg::writer::file_writer::ParquetWriter::close(self) -> iceberg::Result<alloc::vec::Vec<iceberg::spec::DataFileBuilder>>
 pub async fn iceberg::writer::file_writer::ParquetWriter::write(&mut self, batch: &arrow_array::record_batch::RecordBatch) -> iceberg::Result<()>
+impl iceberg::writer::file_writer::FileWriter for iceberg::writer::file_writer::VortexWriter
+pub async fn iceberg::writer::file_writer::VortexWriter::close(self) -> iceberg::Result<alloc::vec::Vec<iceberg::spec::DataFileBuilder>>
+pub async fn iceberg::writer::file_writer::VortexWriter::write(&mut self, batch: &arrow_array::record_batch::RecordBatch) -> iceberg::Result<()>
 pub trait iceberg::writer::file_writer::FileWriterBuilder<O>: core::clone::Clone + core::marker::Send + core::marker::Sync + 'static
 pub type iceberg::writer::file_writer::FileWriterBuilder::R: iceberg::writer::file_writer::FileWriter<O>
 pub fn iceberg::writer::file_writer::FileWriterBuilder::build(&self, output_file: iceberg::io::OutputFile) -> impl core::future::future::Future<Output = iceberg::Result<Self::R>> + core::marker::Send
 impl iceberg::writer::file_writer::FileWriterBuilder for iceberg::writer::file_writer::ParquetWriterBuilder
 pub type iceberg::writer::file_writer::ParquetWriterBuilder::R = iceberg::writer::file_writer::ParquetWriter
 pub async fn iceberg::writer::file_writer::ParquetWriterBuilder::build(&self, output_file: iceberg::io::OutputFile) -> iceberg::Result<Self::R>
+impl iceberg::writer::file_writer::FileWriterBuilder for iceberg::writer::file_writer::VortexWriterBuilder
+pub type iceberg::writer::file_writer::VortexWriterBuilder::R = iceberg::writer::file_writer::VortexWriter
+pub async fn iceberg::writer::file_writer::VortexWriterBuilder::build(&self, output_file: iceberg::io::OutputFile) -> iceberg::Result<Self::R>
 pub mod iceberg::writer::partitioning
 pub mod iceberg::writer::partitioning::clustered_writer
 pub struct iceberg::writer::partitioning::clustered_writer::ClusteredWriter<B, I, O> where B: iceberg::writer::IcebergWriterBuilder<I, O>, O: core::iter::traits::collect::IntoIterator + core::iter::traits::collect::FromIterator<<O as core::iter::traits::collect::IntoIterator>::Item>, <O as core::iter::traits::collect::IntoIterator>::Item: core::clone::Clone
@@ -3315,6 +3340,10 @@ impl iceberg::writer::CurrentFileStatus for iceberg::writer::file_writer::Parque
 pub fn iceberg::writer::file_writer::ParquetWriter::current_file_path(&self) -> alloc::string::String
 pub fn iceberg::writer::file_writer::ParquetWriter::current_row_num(&self) -> usize
 pub fn iceberg::writer::file_writer::ParquetWriter::current_written_size(&self) -> usize
+impl iceberg::writer::CurrentFileStatus for iceberg::writer::file_writer::VortexWriter
+pub fn iceberg::writer::file_writer::VortexWriter::current_file_path(&self) -> alloc::string::String
+pub fn iceberg::writer::file_writer::VortexWriter::current_row_num(&self) -> usize
+pub fn iceberg::writer::file_writer::VortexWriter::current_written_size(&self) -> usize
 impl<B, L, F> iceberg::writer::CurrentFileStatus for iceberg::writer::base_writer::data_file_writer::DataFileWriter<B, L, F> where B: iceberg::writer::file_writer::FileWriterBuilder, L: iceberg::writer::file_writer::location_generator::LocationGenerator, F: iceberg::writer::file_writer::location_generator::FileNameGenerator
 pub fn iceberg::writer::base_writer::data_file_writer::DataFileWriter<B, L, F>::current_file_path(&self) -> alloc::string::String
 pub fn iceberg::writer::base_writer::data_file_writer::DataFileWriter<B, L, F>::current_row_num(&self) -> usize

crates/iceberg/src/arrow/mod.rs

@@ -29,6 +29,8 @@ pub(crate) mod delete_filter;
 
 mod int96;
 mod reader;
+mod vortex_util;
+pub(crate) use vortex_util::*;
 /// RecordBatch projection utilities
 pub mod record_batch_projector;
 pub(crate) mod record_batch_transformer;

crates/iceberg/src/arrow/reader/mod.rs

@@ -41,6 +41,7 @@ mod positional_deletes;
 mod predicate_visitor;
 mod projection;
 mod row_filter;
+mod vortex;
 pub use file_reader::ArrowFileReader;
 pub(crate) use options::ParquetReadOptions;
 use predicate_visitor::{CollectFieldIdVisitor, PredicateConverter};

crates/iceberg/src/arrow/reader/pipeline.rs

@@ -43,7 +43,7 @@ use crate::metadata_columns::{
     RESERVED_FIELD_ID_FILE, RESERVED_FIELD_ID_SPEC_ID, is_metadata_field,
 };
 use crate::scan::{ArrowRecordBatchStream, FileScanTask, FileScanTaskStream};
-use crate::spec::Datum;
+use crate::spec::{DataFileFormat, Datum};
 use crate::{Error, ErrorKind};
 
 impl ArrowReader {
@@ -108,6 +108,26 @@ struct FileScanTaskReader {
 
 impl FileScanTaskReader {
     async fn process(self, task: FileScanTask) -> Result<ArrowRecordBatchStream> {
+        match task.data_file_format {
+            DataFileFormat::Parquet => self.process_parquet(task).await,
+            DataFileFormat::Vortex => {
+                super::vortex::read_vortex_task(
+                    task,
+                    &self.file_io,
+                    self.batch_size,
+                    &self.delete_file_loader,
+                    self.scan_metrics.bytes_read_counter().clone(),
+                )
+                .await
+            }
+            format => Err(Error::new(
+                ErrorKind::FeatureUnsupported,
+                format!("Unsupported data file format: {format}"),
+            )),
+        }
+    }
+
+    async fn process_parquet(self, task: FileScanTask) -> Result<ArrowRecordBatchStream> {
         let should_load_page_index =
             (self.row_selection_enabled && task.predicate.is_some()) || !task.deletes.is_empty();
         let mut parquet_read_options = self.parquet_read_options;