update breaking changes

Author: XiangpengHao

datafusion/core/src/datasource/physical_plan/arrow_file.rs

@@ -31,6 +31,7 @@ use crate::physical_plan::{
     DisplayAs, DisplayFormatType, ExecutionPlan, Partitioning, SendableRecordBatchStream,
 };
 
+use arrow::buffer::Buffer;
 use arrow_ipc::reader::FileDecoder;
 use arrow_schema::SchemaRef;
 use datafusion_common::config::ConfigOptions;
@@ -296,7 +297,7 @@ impl FileOpener for ArrowOpener {
                     for (dict_block, dict_result) in
                         footer.dictionaries().iter().flatten().zip(dict_results)
                     {
-                        decoder.read_dictionary(dict_block, &dict_result.into())?;
+                        decoder.read_dictionary(dict_block, &Buffer::from_bytes(dict_result.into()))?;
                     }
 
                     // filter recordbatches according to range
@@ -331,7 +332,7 @@ impl FileOpener for ArrowOpener {
                             .into_iter()
                             .zip(recordbatch_results)
                             .filter_map(move |(block, data)| {
-                                match decoder.read_record_batch(&block, &data.into()) {
+                                match decoder.read_record_batch(&block, &Buffer::from_bytes(data.into())) {
                                     Ok(Some(record_batch)) => Some(Ok(record_batch)),
                                     Ok(None) => None,
                                     Err(err) => Some(Err(err)),

datafusion/functions/src/regex/regexpreplace.rs

@@ -402,7 +402,7 @@ fn _regexp_replace_static_pattern_replace<T: OffsetSizeTrait>(
             let string_view_array = as_string_view_array(&args[0])?;
 
             let mut builder = StringViewBuilder::with_capacity(string_view_array.len())
-                .with_block_size(1024 * 1024 * 2);
+                .with_fixed_block_size(1024 * 1024 * 2);
 
             for val in string_view_array.iter() {
                 if let Some(val) = val {

datafusion/physical-expr-common/src/binary_view_map.rs

@@ -149,7 +149,7 @@ where
             output_type,
             map: hashbrown::raw::RawTable::with_capacity(INITIAL_MAP_CAPACITY),
             map_size: 0,
-            builder: GenericByteViewBuilder::new().with_block_size(2 * 1024 * 1024),
+            builder: GenericByteViewBuilder::new().with_fixed_block_size(2 * 1024 * 1024),
             random_state: RandomState::new(),
             hashes_buffer: vec![],
             null: None,

datafusion/physical-plan/src/coalesce_batches.rs

@@ -493,7 +493,7 @@ fn gc_string_view_batch(batch: &RecordBatch) -> RecordBatch {
                 // We set the block size to `ideal_buffer_size` so that the new StringViewArray only has one buffer, which accelerate later concat_batches.
                 // See https://github.com/apache/arrow-rs/issues/6094 for more details.
                 let mut builder = StringViewBuilder::with_capacity(s.len())
-                    .with_block_size(ideal_buffer_size as u32);
+                    .with_fixed_block_size(ideal_buffer_size as u32);
 
                 for v in s.iter() {
                     builder.append_option(v);

datafusion/proto-common/src/from_proto/mod.rs

@@ -408,7 +408,7 @@ impl TryFrom<&protobuf::ScalarValue> for ScalarValue {
                         "Error IPC message while deserializing ScalarValue::List: {e}"
                     ))
                 })?;
-                let buffer = Buffer::from(arrow_data);
+                let buffer = Buffer::from(arrow_data.as_slice());
 
                 let ipc_batch = message.header_as_record_batch().ok_or_else(|| {
                     Error::General(
@@ -423,7 +423,7 @@ impl TryFrom<&protobuf::ScalarValue> for ScalarValue {
                             "Error IPC message while deserializing ScalarValue::List dictionary message: {e}"
                         ))
                     })?;
-                    let buffer = Buffer::from(arrow_data);
+                    let buffer = Buffer::from(arrow_data.as_slice());
 
                     let dict_batch = message.header_as_dictionary_batch().ok_or_else(|| {
                         Error::General(