refactor(arrow2): remove parquet2 from daft (#6339)

Remove all parquet2 usage from daft crates, the only remaining usage is
now in arrow2 itself. The arrow-rs reader now handles all parquet
decoding, and the metadata adapter already wraps arrow-rs types, so
parquet2 is fully redundant.

However, we still can't remove `#![allow(deprecated, reason = "arrow2
migration")]` because daft-parquet still uses daft_arrow::array::Array,
daft_arrow::datatypes::Schema, etc. for the pyarrow FFI bridge.

The pyarrow export path (`read_parquet_single_into_arrow`) is rewritten
to call the arrow-rs `read_parquet_single()` and convert the resulting
`RecordBatch` to arrow2 chunks, replacing the parquet2-based
`ParquetReaderBuilder`/`stream_reader` pipeline. For
`StringEncoding::Raw` (reading invalid UTF-8), the parquet metadata is
modified to strip STRING/UTF8 logical types before the arrow-rs reader
sees them, so BYTE_ARRAY columns are decoded as Binary without UTF-8
validation.

---------

Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>

Author: desmondcheongzx

Cargo.lock

@@ -7,9 +7,6 @@ serde_json = {workspace = true}
 thiserror = {workspace = true}
 tokio = {workspace = true}
 
-[dev-dependencies]
-parquet2 = {workspace = true}
-
 [features]
 python = ["dep:pyo3"]
 

src/common/error/Cargo.toml

@@ -127,21 +127,4 @@ mod tests {
             _ => panic!("Expected ByteStreamError"),
         }
     }
-
-    #[test]
-    fn test_parquet_io_error_conversion() {
-        // Ensure that parquet2 IO errors get converted into transient Byte Stream errors.
-        let error_message = "IO error occurred";
-        let parquet_io_error =
-            parquet2::error::Error::IoError(std::io::Error::other(error_message));
-        let arrow_error: daft_arrow::error::Error = parquet_io_error.into();
-        //let arrow_error = daft_arrow::error::Error::from(parquet_io_error);
-        let daft_error: DaftError = arrow_error.into();
-        match daft_error {
-            DaftError::ByteStreamError(e) => {
-                assert_eq!(e.to_string(), format!("Io error: {error_message}"));
-            }
-            _ => panic!("Expected ByteStreamError"),
-        }
-    }
 }

src/common/error/src/error.rs

@@ -15,7 +15,7 @@ use daft_dsl::{AggExpr, Expr, ExprRef};
 use daft_io::{IOClient, IOConfig, IOStatsRef};
 use daft_json::{JsonConvertOptions, JsonParseOptions, JsonReadOptions};
 use daft_parquet::{
-    DaftParquetMetadata, infer_arrow_schema_from_metadata,
+    DaftParquetMetadata,
     read::{ParquetSchemaInferenceOptions, read_parquet_bulk, read_parquet_metadata_bulk},
 };
 use daft_recordbatch::RecordBatch;
@@ -641,11 +641,8 @@ pub fn read_parquet_into_micropartition<T: AsRef<str>>(
         let schemas = metadata
             .iter()
             .map(|m| {
-                let schema = infer_arrow_schema_from_metadata(
-                    m.as_parquet2(),
-                    Some((*schema_infer_options).into()),
-                )?;
-                let daft_schema = Schema::from(schema);
+                let daft_schema =
+                    daft_parquet::infer_schema_from_daft_metadata(m, *schema_infer_options)?;
                 DaftResult::Ok(Arc::new(daft_schema))
             })
             .collect::<DaftResult<Vec<_>>>()?;
@@ -668,11 +665,8 @@ pub fn read_parquet_into_micropartition<T: AsRef<str>>(
         let schemas = metadata
             .iter()
             .map(|m| {
-                let schema = infer_arrow_schema_from_metadata(
-                    m.as_parquet2(),
-                    Some((*schema_infer_options).into()),
-                )?;
-                let daft_schema = schema.into();
+                let daft_schema =
+                    daft_parquet::infer_schema_from_daft_metadata(m, *schema_infer_options)?;
                 DaftResult::Ok(Arc::new(daft_schema))
             })
             .collect::<DaftResult<Vec<_>>>()?;

src/daft-micropartition/src/micropartition.rs

@@ -4,9 +4,7 @@ name = "parquet_read"
 
 [dependencies]
 arrow = {workspace = true}
-async-compat = {workspace = true}
 daft-arrow = {path = "../daft-arrow"}
-async-stream = {workspace = true}
 bytes = {workspace = true}
 common-error = {path = "../common/error", default-features = false}
 common-runtime = {path = "../common/runtime", default-features = false}
@@ -17,17 +15,13 @@ daft-recordbatch = {path = "../daft-recordbatch", default-features = false}
 daft-stats = {path = "../daft-stats", default-features = false}
 futures = {workspace = true}
 indexmap = {workspace = true}
-itertools = {workspace = true}
-log = {workspace = true}
 parquet = {workspace = true, features = ["async"]}
-parquet2 = {workspace = true}
 pyo3 = {workspace = true, optional = true}
 rayon = {workspace = true}
 serde = {workspace = true}
 snafu = {workspace = true}
 tokio = {workspace = true}
 tokio-stream = {workspace = true}
-tokio-util = {workspace = true}
 
 [dev-dependencies]
 bincode = {workspace = true}

src/daft-parquet/Cargo.toml

@@ -1,9 +1,7 @@
-//! Arrow-rs based parquet reader.
+//! Parquet reader built on the arrow-rs `parquet` crate.
 //!
-//! This module provides a parquet reader built on the arrow-rs `parquet` crate,
-//! replacing the parquet2/arrow2 decode pipeline. It uses [`DaftAsyncFileReader`]
-//! as the IO bridge for remote reads, and the sync `ParquetRecordBatchReaderBuilder`
-//! with `std::fs::File` for local reads (avoiding IOClient overhead).
+//! Uses [`DaftAsyncFileReader`] as the IO bridge for remote reads, and the sync
+//! `ParquetRecordBatchReaderBuilder` with `std::fs::File` for local reads.
 
 use std::{
     borrow::Borrow,
@@ -37,10 +35,12 @@ use tokio_stream::wrappers::ReceiverStream;
 
 use crate::{
     async_reader::DaftAsyncFileReader,
-    metadata::apply_field_ids_to_arrowrs_parquet_metadata,
-    read::ParquetSchemaInferenceOptions,
+    metadata::{
+        apply_field_ids_to_arrowrs_parquet_metadata, strip_string_types_from_parquet_metadata,
+    },
+    read::{ParquetSchemaInferenceOptions, StringEncoding},
     schema_inference::{arrow_schema_to_daft_schema, infer_schema_from_parquet_metadata_arrowrs},
-    statistics::arrowrs_row_group_metadata_to_table_stats,
+    statistics::row_group_metadata_to_table_stats,
 };
 
 /// Default batch size for the arrow-rs reader (number of rows per batch).
@@ -89,8 +89,7 @@ fn infer_schemas(
     let arrow_schema = infer_schema_from_parquet_metadata_arrowrs(
         parquet_metadata,
         Some(schema_infer_options.coerce_int96_timestamp_unit),
-        schema_infer_options.string_encoding
-            == daft_arrow::io::parquet::read::schema::StringEncoding::Raw,
+        schema_infer_options.string_encoding == StringEncoding::Raw,
     )
     .map_err(parquet_err)?;
     let daft_schema = arrow_schema_to_daft_schema(&arrow_schema)?;
@@ -325,9 +324,8 @@ fn deletes_to_row_selection(local_deletes: &[usize], total_rows: usize) -> RowSe
     selectors.into()
 }
 
-/// Read a single parquet file into a Daft [`RecordBatch`] using the arrow-rs reader.
+/// Read a single parquet file into a Daft [`RecordBatch`].
 ///
-/// This is the arrow-rs equivalent of the parquet2-based `read_parquet_single`.
 /// When `predicate` and/or `delete_rows` are provided, the reader handles them
 /// internally using arrow-rs `RowFilter` and `RowSelection` for late materialization.
 ///
@@ -339,10 +337,9 @@ fn deletes_to_row_selection(local_deletes: &[usize], total_rows: usize) -> RowSe
 ///   offset (skip file rows) → predicate filter → limit
 ///
 /// Note: `start_offset > 0` is rejected by the micropartition reader and never used
-/// in production (the streaming scan path doesn't even accept the parameter). The
-/// parquet2 reader has latent bugs for this case — both its local and remote paths
-/// produce RecordBatch size mismatches when `start_offset > 0`. Our implementation
-/// follows the intended semantics based on the code structure and the `apply_delete_rows`
+/// in production (the streaming scan path doesn't even accept the parameter). Our
+/// implementation follows the intended semantics based on the code structure and the
+/// `apply_delete_rows`
 /// docstring in `read.rs`, but there is no working reference implementation to compare
 /// against.
 #[allow(clippy::too_many_arguments)]
@@ -370,6 +367,13 @@ pub async fn read_parquet_single_arrowrs(
         parquet_metadata = apply_field_ids_to_arrowrs_parquet_metadata(parquet_metadata, mapping)?;
     }
 
+    // 1c. For StringEncoding::Raw, strip STRING/UTF8 logical types from the parquet
+    // metadata so arrow-rs infers Binary instead of Utf8. This avoids UTF-8
+    // validation during decode, allowing files with invalid UTF-8 to be read.
+    if schema_infer_options.string_encoding == StringEncoding::Raw {
+        parquet_metadata = strip_string_types_from_parquet_metadata(parquet_metadata)?;
+    }
+
     // 2. Infer schema with Daft options (INT96 coercion, string encoding).
     let (arrow_schema, daft_schema) = infer_schemas(&parquet_metadata, &schema_infer_options)?;
 
@@ -672,6 +676,12 @@ pub(crate) fn local_parquet_setup(
         parquet_metadata = apply_field_ids_to_arrowrs_parquet_metadata(parquet_metadata, mapping)?;
     }
 
+    // 1c. For StringEncoding::Raw, strip STRING/UTF8 logical types so arrow-rs
+    // reads BYTE_ARRAY as Binary (no UTF-8 validation).
+    if schema_infer_options.string_encoding == StringEncoding::Raw {
+        parquet_metadata = strip_string_types_from_parquet_metadata(parquet_metadata)?;
+    }
+
     // 2. Infer schema with Daft options.
     let (arrow_schema, daft_schema) = infer_schemas(&parquet_metadata, &schema_infer_options)?;
 
@@ -876,8 +886,7 @@ pub(crate) fn decode_single_rg(
 ///
 /// This avoids the overhead of `DaftAsyncFileReader` + `IOClient` for local files
 /// by using `std::fs::File` directly with `ParquetRecordBatchReaderBuilder`.
-/// Row groups are decoded in parallel using rayon, matching the parquet2 reader's
-/// parallelism strategy. Supports late materialization via `RowFilter` and
+/// Row groups are decoded in parallel using rayon. Supports late materialization via `RowFilter` and
 /// positional delete skipping via `RowSelection`.
 ///
 /// See [`read_parquet_single_arrowrs`] for `start_offset` semantics.
@@ -964,8 +973,8 @@ pub fn local_parquet_read_arrowrs(
 /// Stream a local parquet file as Daft [`RecordBatch`]es using the sync arrow-rs reader,
 /// dispatching per-row-group decode as async tasks on the compute runtime.
 ///
-/// Matches parquet2's `local_parquet_stream` pattern: sync metadata read, then
-/// per-RG tasks on the DAFTCPU pool with semaphore-gated parallelism.
+/// Performs sync metadata read, then per-RG tasks on the DAFTCPU pool with
+/// semaphore-gated parallelism.
 #[allow(clippy::too_many_arguments)]
 pub async fn local_parquet_stream_arrowrs(
     path: &str,
@@ -1009,8 +1018,7 @@ pub async fn local_parquet_stream_arrowrs(
     }
 
     // 2. Semaphore: limit concurrent RG decodes.
-    // Unlike parquet2 (which spawns per-column tasks and divides by num_columns),
-    // arrowrs decodes all columns in a single block_in_place call per RG,
+    // All columns are decoded in a single block_in_place call per RG,
     // so concurrency is limited only by available CPUs.
     let num_cpus = std::thread::available_parallelism()
         .map(|n| n.get())
@@ -1073,7 +1081,6 @@ pub async fn local_parquet_stream_arrowrs(
 
 /// Stream a single parquet file as Daft [`RecordBatch`]es using the arrow-rs reader.
 ///
-/// This is the arrow-rs equivalent of the parquet2-based `stream_parquet_single`.
 /// Supports late materialization via `RowFilter` and positional delete skipping
 /// via `RowSelection`.
 #[allow(clippy::too_many_arguments)]
@@ -1101,6 +1108,12 @@ pub async fn stream_parquet_single_arrowrs(
         parquet_metadata = apply_field_ids_to_arrowrs_parquet_metadata(parquet_metadata, mapping)?;
     }
 
+    // 1c. For StringEncoding::Raw, strip STRING/UTF8 logical types so arrow-rs
+    // reads BYTE_ARRAY as Binary (no UTF-8 validation).
+    if schema_infer_options.string_encoding == StringEncoding::Raw {
+        parquet_metadata = strip_string_types_from_parquet_metadata(parquet_metadata)?;
+    }
+
     // 2. Infer schema with Daft options.
     let (arrow_schema, daft_schema) = infer_schemas(&parquet_metadata, &schema_infer_options)?;
 
@@ -1296,7 +1309,7 @@ fn prune_row_groups(
     let mut result = Vec::with_capacity(candidates.len());
     for rg_idx in candidates {
         let rg_meta = metadata.row_group(rg_idx);
-        match arrowrs_row_group_metadata_to_table_stats(rg_meta, schema) {
+        match row_group_metadata_to_table_stats(rg_meta, schema) {
             Ok(stats) => {
                 let evaled = stats.eval_expression(&bound_pred)?;
                 if evaled.to_truth_value() != TruthValue::False {

src/daft-parquet/src/arrowrs_reader.rs

@@ -11,8 +11,7 @@ use parquet::{
 
 use crate::read_planner::{CoalescePass, ReadPlanner, SplitLargeRequestPass};
 
-// IO coalescing/splitting constants — these match the parquet2 reader in file.rs:384-391
-// so both paths have identical IO behavior.
+// IO coalescing/splitting constants for the read planner.
 
 /// Maximum hole size for the coalesce pass (1 MB).
 /// Two byte ranges within this distance are merged into a single request,