feat: implement StringColumn using StringViewArray

src/common/arrow/src/arrow/array/binview/from.rs

@@ -12,13 +12,46 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use arrow_data::ArrayData;
+use arrow_data::ArrayDataBuilder;
+use arrow_schema::DataType;
+
+use crate::arrow::array::Arrow2Arrow;
 use crate::arrow::array::BinaryViewArrayGeneric;
 use crate::arrow::array::MutableBinaryViewArray;
 use crate::arrow::array::ViewType;
+use crate::arrow::bitmap::Bitmap;
 
 impl<T: ViewType + ?Sized, P: AsRef<T>> FromIterator<Option<P>> for BinaryViewArrayGeneric<T> {
     #[inline]
     fn from_iter<I: IntoIterator<Item = Option<P>>>(iter: I) -> Self {
         MutableBinaryViewArray::<T>::from_iter(iter).into()
     }
 }
+
+impl<T: ViewType + ?Sized> Arrow2Arrow for BinaryViewArrayGeneric<T> {
+    fn to_data(&self) -> ArrayData {
+        let builder = ArrayDataBuilder::new(DataType::BinaryView)
+            .len(self.len())
+            .add_buffer(self.views.clone().into())
+            .add_buffers(self.buffers.iter().map(|x| x.clone().into()).collect())
+            .nulls(self.validity.clone().map(Into::into));
+        unsafe { builder.build_unchecked() }
+    }
+
+    fn from_data(data: &ArrayData) -> Self {
+        let views = crate::arrow::buffer::Buffer::from(data.buffers()[0].clone());
+        let buffers = data.buffers()[1..]
+            .iter()
+            .map(|x| crate::arrow::buffer::Buffer::from(x.clone()))
+            .collect();
+        let validity = data.nulls().map(|x| Bitmap::from_null_buffer(x.clone()));
+        Self::try_new(
+            crate::arrow::datatypes::DataType::BinaryView,
+            views,
+            buffers,
+            validity,
+        )
+        .unwrap()
+    }
+}

src/common/arrow/src/arrow/array/binview/mutable.rs

@@ -41,9 +41,9 @@ pub struct MutableBinaryViewArray<T: ViewType + ?Sized> {
     pub(super) validity: Option<MutableBitmap>,
     pub(super) phantom: std::marker::PhantomData<T>,
     /// Total bytes length if we would concatenate them all.
-    pub(super) total_bytes_len: usize,
+    pub total_bytes_len: usize,
     /// Total bytes in the buffer (excluding remaining capacity)
-    pub(super) total_buffer_len: usize,
+    pub total_buffer_len: usize,
 }
 
 impl<T: ViewType + ?Sized> Clone for MutableBinaryViewArray<T> {

src/common/arrow/src/arrow/array/mod.rs

@@ -496,7 +496,8 @@ pub fn to_data(array: &dyn Array) -> arrow_data::ArrayData {
             })
         }
         Map => to_data_dyn!(array, MapArray),
-        BinaryView | Utf8View => unimplemented!(),
+        BinaryView => to_data_dyn!(array, BinaryViewArray),
+        Utf8View => unimplemented!(),
     }
 }
 
@@ -527,7 +528,7 @@ pub fn from_data(data: &arrow_data::ArrayData) -> Box<dyn Array> {
             })
         }
         Map => Box::new(MapArray::from_data(data)),
-        BinaryView | Utf8View => unimplemented!(),
+        BinaryView | Utf8View => Box::new(BinaryViewArray::from_data(data)),
     }
 }
 

src/common/arrow/src/arrow/datatypes/mod.rs

@@ -237,7 +237,8 @@ impl From<DataType> for arrow_schema::DataType {
             DataType::Decimal(precision, scale) => Self::Decimal128(precision as _, scale as _),
             DataType::Decimal256(precision, scale) => Self::Decimal256(precision as _, scale as _),
             DataType::Extension(_, d, _) => (*d).into(),
-            DataType::BinaryView | DataType::Utf8View => {
+            DataType::BinaryView => Self::BinaryView,
+            DataType::Utf8View => {
                 panic!("view datatypes are not supported by arrow-rs")
             }
         }

src/query/expression/src/converts/arrow2/to.rs

@@ -92,8 +92,8 @@ fn table_type_to_arrow_type(ty: &TableDataType) -> ArrowDataType {
             None,
         ),
         TableDataType::Boolean => ArrowDataType::Boolean,
-        TableDataType::Binary => ArrowDataType::LargeBinary,
-        TableDataType::String => ArrowDataType::LargeUtf8,
+        TableDataType::Binary => ArrowDataType::BinaryView,
+        TableDataType::String => ArrowDataType::Utf8View,
         TableDataType::Number(ty) => with_number_type!(|TYPE| match ty {
             NumberDataType::TYPE => ArrowDataType::TYPE,
         }),
@@ -135,7 +135,7 @@ fn table_type_to_arrow_type(ty: &TableDataType) -> ArrowDataType {
         }
         TableDataType::Bitmap => ArrowDataType::Extension(
             ARROW_EXT_TYPE_BITMAP.to_string(),
-            Box::new(ArrowDataType::LargeBinary),
+            Box::new(ArrowDataType::BinaryView),
             None,
         ),
         TableDataType::Tuple {
@@ -157,17 +157,17 @@ fn table_type_to_arrow_type(ty: &TableDataType) -> ArrowDataType {
         }
         TableDataType::Variant => ArrowDataType::Extension(
             ARROW_EXT_TYPE_VARIANT.to_string(),
-            Box::new(ArrowDataType::LargeBinary),
+            Box::new(ArrowDataType::BinaryView),
             None,
         ),
         TableDataType::Geometry => ArrowDataType::Extension(
             ARROW_EXT_TYPE_GEOMETRY.to_string(),
-            Box::new(ArrowDataType::LargeBinary),
+            Box::new(ArrowDataType::BinaryView),
             None,
         ),
         TableDataType::Geography => ArrowDataType::Extension(
             ARROW_EXT_TYPE_GEOGRAPHY.to_string(),
-            Box::new(ArrowDataType::LargeBinary),
+            Box::new(ArrowDataType::BinaryView),
             None,
         ),
     }
@@ -304,32 +304,10 @@ impl Column {
                 )
                 .unwrap(),
             ),
-            Column::Binary(col) => {
-                let offsets: Buffer<i64> =
-                    col.offsets().iter().map(|offset| *offset as i64).collect();
-                Box::new(
-                    databend_common_arrow::arrow::array::BinaryArray::<i64>::try_new(
-                        arrow_type,
-                        unsafe { OffsetsBuffer::new_unchecked(offsets) },
-                        col.data().clone(),
-                        None,
-                    )
-                    .unwrap(),
-                )
-            }
-            Column::String(col) => {
-                let offsets: Buffer<i64> =
-                    col.offsets().iter().map(|offset| *offset as i64).collect();
-                Box::new(
-                    databend_common_arrow::arrow::array::Utf8Array::<i64>::try_new(
-                        arrow_type,
-                        unsafe { OffsetsBuffer::new_unchecked(offsets) },
-                        col.data().clone(),
-                        None,
-                    )
-                    .unwrap(),
-                )
-            }
+            Column::Binary(col) => Box::new(col.clone().into_inner()),
+            Column::String(col) => unsafe {
+                Box::new(col.clone().into_inner().to_utf8view_unchecked())
+            },
             Column::Timestamp(col) => Box::new(
                 databend_common_arrow::arrow::array::PrimitiveArray::<i64>::try_new(
                     arrow_type,
@@ -401,19 +379,7 @@ impl Column {
             Column::Bitmap(col)
             | Column::Variant(col)
             | Column::Geometry(col)
-            | Column::Geography(GeographyColumn(col)) => {
-                let offsets: Buffer<i64> =
-                    col.offsets().iter().map(|offset| *offset as i64).collect();
-                Box::new(
-                    databend_common_arrow::arrow::array::BinaryArray::<i64>::try_new(
-                        arrow_type,
-                        unsafe { OffsetsBuffer::new_unchecked(offsets) },
-                        col.data().clone(),
-                        None,
-                    )
-                    .unwrap(),
-                )
-            }
+            | Column::Geography(GeographyColumn(col)) => Box::new(col.clone().into_inner()),
         }
     }
 }

src/query/expression/src/filter/filter_executor.rs

@@ -125,7 +125,7 @@ impl FilterExecutor {
             if self.keep_order && self.has_or {
                 self.true_selection[0..result_count].sort();
             }
-            data_block.take(&self.true_selection[0..result_count], &mut None)
+            data_block.take(&self.true_selection[0..result_count])
         }
     }
 

src/query/expression/src/filter/select_value/select_column_scalar.rs

@@ -239,48 +239,19 @@ impl<'a> Selector<'a> {
                     Some(validity) => {
                         // search the whole string buffer
                         if let LikePattern::SurroundByPercent(searcher) = like_pattern {
-                            let needle = searcher.needle();
-                            let needle_byte_len = needle.len();
-                            let data = column.data().as_slice();
-                            let offsets = column.offsets().as_slice();
-                            let mut idx = 0;
-                            let mut pos = (*offsets.first().unwrap()) as usize;
-                            let end = (*offsets.last().unwrap()) as usize;
-
-                            while pos < end && idx < count {
-                                if let Some(p) = searcher.search(&data[pos..end]) {
-                                    while offsets[idx + 1] as usize <= pos + p {
-                                        let ret = NOT && validity.get_bit_unchecked(idx);
-                                        update_index(
-                                            ret,
-                                            idx as u32,
-                                            true_selection,
-                                            false_selection,
-                                        );
-                                        idx += 1;
-                                    }
-
-                                    // check if the substring is in bound
-                                    let ret =
-                                        pos + p + needle_byte_len <= offsets[idx + 1] as usize;
-
-                                    let ret = if NOT {
-                                        validity.get_bit_unchecked(idx) && !ret
-                                    } else {
-                                        validity.get_bit_unchecked(idx) && ret
-                                    };
-                                    update_index(ret, idx as u32, true_selection, false_selection);
-
-                                    pos = offsets[idx + 1] as usize;
-                                    idx += 1;
+                            for idx in 0u32..count as u32 {
+                                let ret = if NOT {
+                                    validity.get_bit_unchecked(idx as usize)
+                                        && searcher
+                                            .search(column.index_unchecked_bytes(idx as usize))
+                                            .is_none()
                                 } else {
-                                    break;
-                                }
-                            }
-                            while idx < count {
-                                let ret = NOT && validity.get_bit_unchecked(idx);
-                                update_index(ret, idx as u32, true_selection, false_selection);
-                                idx += 1;
+                                    validity.get_bit_unchecked(idx as usize)
+                                        && searcher
+                                            .search(column.index_unchecked_bytes(idx as usize))
+                                            .is_some()
+                                };
+                                update_index(ret, idx, true_selection, false_selection);
                             }
                         } else {
                             for idx in 0u32..count as u32 {
@@ -300,40 +271,17 @@ impl<'a> Selector<'a> {
                     None => {
                         // search the whole string buffer
                         if let LikePattern::SurroundByPercent(searcher) = like_pattern {
-                            let needle = searcher.needle();
-                            let needle_byte_len = needle.len();
-                            let data = column.data().as_slice();
-                            let offsets = column.offsets().as_slice();
-                            let mut idx = 0;
-                            let mut pos = (*offsets.first().unwrap()) as usize;
-                            let end = (*offsets.last().unwrap()) as usize;
-
-                            while pos < end && idx < count {
-                                if let Some(p) = searcher.search(&data[pos..end]) {
-                                    while offsets[idx + 1] as usize <= pos + p {
-                                        update_index(
-                                            NOT,
-                                            idx as u32,
-                                            true_selection,
-                                            false_selection,
-                                        );
-                                        idx += 1;
-                                    }
-                                    // check if the substring is in bound
-                                    let ret =
-                                        pos + p + needle_byte_len <= offsets[idx + 1] as usize;
-                                    let ret = if NOT { !ret } else { ret };
-                                    update_index(ret, idx as u32, true_selection, false_selection);
-
-                                    pos = offsets[idx + 1] as usize;
-                                    idx += 1;
+                            for idx in 0u32..count as u32 {
+                                let ret = if NOT {
+                                    searcher
+                                        .search(column.index_unchecked_bytes(idx as usize))
+                                        .is_none()
                                 } else {
-                                    break;
-                                }
-                            }
-                            while idx < count {
-                                update_index(NOT, idx as u32, true_selection, false_selection);
-                                idx += 1;
+                                    searcher
+                                        .search(column.index_unchecked_bytes(idx as usize))
+                                        .is_some()
+                                };
+                                update_index(ret, idx, true_selection, false_selection);
                             }
                         } else {
                             for idx in 0u32..count as u32 {

src/query/expression/src/kernels/concat.rs

@@ -24,11 +24,11 @@ use itertools::Itertools;
 
 use crate::copy_continuous_bits;
 use crate::kernels::take::BIT_MASK;
-use crate::kernels::utils::copy_advance_aligned;
 use crate::kernels::utils::set_vec_len_by_ptr;
 use crate::store_advance_aligned;
 use crate::types::array::ArrayColumnBuilder;
 use crate::types::binary::BinaryColumn;
+use crate::types::binary::BinaryColumnBuilder;
 use crate::types::decimal::Decimal;
 use crate::types::decimal::DecimalColumn;
 use crate::types::geography::GeographyColumn;
@@ -275,37 +275,11 @@ impl Column {
         cols: impl Iterator<Item = BinaryColumn> + Clone,
         num_rows: usize,
     ) -> BinaryColumn {
-        // [`BinaryColumn`] consists of [`data`] and [`offset`], we build [`data`] and [`offset`] respectively,
-        // and then call `BinaryColumn::new(data.into(), offsets.into())` to create [`BinaryColumn`].
-        let mut offsets: Vec<u64> = Vec::with_capacity(num_rows + 1);
-        let mut data_size = 0;
-
-        // Build [`offset`] and calculate `data_size` required by [`data`].
-        offsets.push(0);
-        for col in cols.clone() {
-            let mut start = col.offsets()[0];
-            for end in col.offsets()[1..].iter() {
-                data_size += end - start;
-                start = *end;
-                offsets.push(data_size);
-            }
-        }
-
-        // Build [`data`].
-        let mut data: Vec<u8> = Vec::with_capacity(data_size as usize);
-        let mut data_ptr = data.as_mut_ptr();
-
-        unsafe {
-            for col in cols {
-                let offsets = col.offsets();
-                let col_data = &(col.data().as_slice())
-                    [offsets[0] as usize..offsets[offsets.len() - 1] as usize];
-                copy_advance_aligned(col_data.as_ptr(), &mut data_ptr, col_data.len());
-            }
-            set_vec_len_by_ptr(&mut data, data_ptr);
+        let mut builder = BinaryColumnBuilder::with_capacity(num_rows, 0);
+        for col in cols {
+            builder.append_column(&col);
         }
-
-        BinaryColumn::new(data.into(), offsets.into())
+        builder.build()
     }
 
     pub fn concat_string_types(