Merge branch 'main' into reduce-copy

Author: XiangpengHao

.github/workflows/dev.yml

@@ -51,7 +51,7 @@ jobs:
     runs-on: ubuntu-latest
     steps:
       - uses: actions/checkout@v5
-      - uses: actions/setup-node@v5
+      - uses: actions/setup-node@v6
         with:
           node-version: "14"
       - name: Prettier check

.github/workflows/docs.yml

@@ -71,7 +71,7 @@ jobs:
     steps:
       - uses: actions/checkout@v5
       - name: Download crate docs
-        uses: actions/download-artifact@v5
+        uses: actions/download-artifact@v6
         with:
           name: crate-docs
           path: website/build

CHANGELOG-old.md

@@ -68,7 +68,7 @@ exclude = [
 ]
 
 [workspace.package]
-version = "56.2.0"
+version = "57.0.0"
 homepage = "https://github.com/apache/arrow-rs"
 repository = "https://github.com/apache/arrow-rs"
 authors = ["Apache Arrow <dev@arrow.apache.org>"]
@@ -85,22 +85,22 @@ edition = "2024"
 rust-version = "1.85"
 
 [workspace.dependencies]
-arrow = { version = "56.2.0", path = "./arrow", default-features = false }
-arrow-arith = { version = "56.2.0", path = "./arrow-arith" }
-arrow-array = { version = "56.2.0", path = "./arrow-array" }
-arrow-buffer = { version = "56.2.0", path = "./arrow-buffer" }
-arrow-cast = { version = "56.2.0", path = "./arrow-cast" }
-arrow-csv = { version = "56.2.0", path = "./arrow-csv" }
-arrow-data = { version = "56.2.0", path = "./arrow-data" }
-arrow-ipc = { version = "56.2.0", path = "./arrow-ipc" }
-arrow-json = { version = "56.2.0", path = "./arrow-json" }
-arrow-ord = { version = "56.2.0", path = "./arrow-ord" }
-arrow-pyarrow = { version = "56.2.0", path = "./arrow-pyarrow" }
-arrow-row = { version = "56.2.0", path = "./arrow-row" }
-arrow-schema = { version = "56.2.0", path = "./arrow-schema" }
-arrow-select = { version = "56.2.0", path = "./arrow-select" }
-arrow-string = { version = "56.2.0", path = "./arrow-string" }
-parquet = { version = "56.2.0", path = "./parquet", default-features = false }
+arrow = { version = "57.0.0", path = "./arrow", default-features = false }
+arrow-arith = { version = "57.0.0", path = "./arrow-arith" }
+arrow-array = { version = "57.0.0", path = "./arrow-array" }
+arrow-buffer = { version = "57.0.0", path = "./arrow-buffer" }
+arrow-cast = { version = "57.0.0", path = "./arrow-cast" }
+arrow-csv = { version = "57.0.0", path = "./arrow-csv" }
+arrow-data = { version = "57.0.0", path = "./arrow-data" }
+arrow-ipc = { version = "57.0.0", path = "./arrow-ipc" }
+arrow-json = { version = "57.0.0", path = "./arrow-json" }
+arrow-ord = { version = "57.0.0", path = "./arrow-ord" }
+arrow-pyarrow = { version = "57.0.0", path = "./arrow-pyarrow" }
+arrow-row = { version = "57.0.0", path = "./arrow-row" }
+arrow-schema = { version = "57.0.0", path = "./arrow-schema" }
+arrow-select = { version = "57.0.0", path = "./arrow-select" }
+arrow-string = { version = "57.0.0", path = "./arrow-string" }
+parquet = { version = "57.0.0", path = "./parquet", default-features = false }
 
 # These crates have not yet been released and thus do not use the workspace version
 parquet-geospatial = { version = "0.1.0", path = "./parquet-geospatial" }

CHANGELOG.md

@@ -80,3 +80,7 @@ harness = false
 [[bench]]
 name = "union_array"
 harness = false
+
+[[bench]]
+name = "record_batch"
+harness = false

Cargo.toml

@@ -0,0 +1,92 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you 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.
+
+use arrow_array::{ArrayRef, Int64Array, RecordBatch, RecordBatchOptions};
+use arrow_schema::{DataType, Field, Schema, SchemaRef};
+use criterion::*;
+use num_integer::Integer;
+use std::sync::Arc;
+
+fn make_record_batch(column_count: usize, row_count: usize) -> RecordBatch {
+    let fields = (0..column_count)
+        .map(|i| Field::new(format!("col_{}", i), DataType::Int64, i.is_even()))
+        .collect::<Vec<_>>();
+
+    let columns = fields
+        .iter()
+        .map(|_| {
+            let array_ref: ArrayRef = Arc::new(Int64Array::from_value(0, row_count));
+            array_ref
+        })
+        .collect::<Vec<_>>();
+
+    let schema = Schema::new(fields);
+
+    let mut options = RecordBatchOptions::new();
+    options.row_count = Some(row_count);
+
+    RecordBatch::try_new_with_options(SchemaRef::new(schema), columns, &options).unwrap()
+}
+
+fn project_benchmark(
+    c: &mut Criterion,
+    column_count: usize,
+    row_count: usize,
+    projection_size: usize,
+) {
+    let input = make_input(column_count, row_count, projection_size);
+
+    c.bench_with_input(
+        BenchmarkId::new(
+            "project",
+            format!(
+                "{:?}x{:?} -> {:?}x{:?}",
+                input.0.num_columns(),
+                input.0.num_rows(),
+                input.1.len(),
+                input.0.num_rows()
+            ),
+        ),
+        &input,
+        |b, (rb, projection)| {
+            b.iter(|| black_box(rb.project(projection).unwrap()));
+        },
+    );
+}
+
+fn make_input(
+    column_count: usize,
+    row_count: usize,
+    projection_size: usize,
+) -> (RecordBatch, Vec<usize>) {
+    let rb = make_record_batch(column_count, row_count);
+    let projection = (0..projection_size).collect::<Vec<_>>();
+    (rb, projection)
+}
+
+fn criterion_benchmark(c: &mut Criterion) {
+    [10, 100, 1000].iter().for_each(|&column_count| {
+        [1, column_count / 2, column_count - 1]
+            .iter()
+            .for_each(|&projection_size| {
+                project_benchmark(c, column_count, 8192, projection_size);
+            })
+    });
+}
+
+criterion_group!(benches, criterion_benchmark);
+criterion_main!(benches);

arrow-array/Cargo.toml

@@ -466,7 +466,7 @@ impl<OffsetSize: OffsetSizeTrait> From<FixedSizeListArray> for GenericListArray<
             _ => unreachable!(),
         };
 
-        let offsets = OffsetBuffer::from_lengths(std::iter::repeat_n(size, value.len()));
+        let offsets = OffsetBuffer::from_repeated_length(size, value.len());
 
         Self {
             data_type: Self::DATA_TYPE_CONSTRUCTOR(field.clone()),

arrow-array/benches/record_batch.rs

@@ -306,15 +306,30 @@ impl<T: ByteViewType + ?Sized> GenericByteViewBuilder<T> {
     /// - String length exceeds `u32::MAX`
     #[inline]
     pub fn append_value(&mut self, value: impl AsRef<T::Native>) {
+        self.try_append_value(value).unwrap()
+    }
+
+    /// Appends a value into the builder
+    ///
+    /// # Errors
+    ///
+    /// Returns an error if:
+    /// - String buffer count exceeds `u32::MAX`
+    /// - String length exceeds `u32::MAX`
+    #[inline]
+    pub fn try_append_value(&mut self, value: impl AsRef<T::Native>) -> Result<(), ArrowError> {
         let v: &[u8] = value.as_ref().as_ref();
-        let length: u32 = v.len().try_into().unwrap();
+        let length: u32 = v.len().try_into().map_err(|_| {
+            ArrowError::InvalidArgumentError(format!("String length {} exceeds u32::MAX", v.len()))
+        })?;
+
         if length <= MAX_INLINE_VIEW_LEN {
             let mut view_buffer = [0; 16];
             view_buffer[0..4].copy_from_slice(&length.to_le_bytes());
             view_buffer[4..4 + v.len()].copy_from_slice(v);
             self.views_buffer.push(u128::from_le_bytes(view_buffer));
             self.null_buffer_builder.append_non_null();
-            return;
+            return Ok(());
         }
 
         // Deduplication if:
@@ -339,7 +354,7 @@ impl<T: ByteViewType + ?Sized> GenericByteViewBuilder<T> {
                     self.views_buffer.push(self.views_buffer[*idx]);
                     self.null_buffer_builder.append_non_null();
                     self.string_tracker = Some((ht, hasher));
-                    return;
+                    return Ok(());
                 }
                 Entry::Vacant(vacant) => {
                     // o.w. we insert the (string hash -> view index)
@@ -356,17 +371,28 @@ impl<T: ByteViewType + ?Sized> GenericByteViewBuilder<T> {
             let to_reserve = v.len().max(self.block_size.next_size() as usize);
             self.in_progress.reserve(to_reserve);
         };
+
         let offset = self.in_progress.len() as u32;
         self.in_progress.extend_from_slice(v);
 
+        let buffer_index: u32 = self.completed.len().try_into().map_err(|_| {
+            ArrowError::InvalidArgumentError(format!(
+                "Buffer count {} exceeds u32::MAX",
+                self.completed.len()
+            ))
+        })?;
+
         let view = ByteView {
             length,
+            // This won't panic as we checked the length of prefix earlier.
             prefix: u32::from_le_bytes(v[0..4].try_into().unwrap()),
-            buffer_index: self.completed.len() as u32,
+            buffer_index,
             offset,
         };
         self.views_buffer.push(view.into());
         self.null_buffer_builder.append_non_null();
+
+        Ok(())
     }
 
     /// Append an `Option` value into the builder
@@ -581,7 +607,6 @@ mod tests {
     use core::str;
 
     use super::*;
-    use crate::Array;
 
     #[test]
     fn test_string_view_deduplicate() {

arrow-array/src/array/list_array.rs

@@ -445,14 +445,16 @@ impl RecordBatch {
             })
             .collect::<Result<Vec<_>, _>>()?;
 
-        RecordBatch::try_new_with_options(
-            SchemaRef::new(projected_schema),
-            batch_fields,
-            &RecordBatchOptions {
-                match_field_names: true,
-                row_count: Some(self.row_count),
-            },
-        )
+        unsafe {
+            // Since we're starting from a valid RecordBatch and project
+            // creates a strict subset of the original, there's no need to
+            // redo the validation checks in `try_new_with_options`.
+            Ok(RecordBatch::new_unchecked(
+                SchemaRef::new(projected_schema),
+                batch_fields,
+                self.row_count,
+            ))
+        }
     }
 
     /// Normalize a semi-structured [`RecordBatch`] into a flat table.