chore(query): add string view dict/freq/onevalue encoding for native format (#17833)

* chore(query): add view encoding

* chore(query): fix bench

* chore(query): fix bench

* chore(query): update

* chore(query): update

* chore(query): update

Author: sundy-li

src/common/column/src/binview/builder.rs

@@ -132,7 +132,6 @@ impl<T: ViewType + ?Sized> BinaryViewColumnBuilder<T> {
         let len = v.length;
         if len <= 12 {
             self.total_bytes_len += len as usize;
-            debug_assert!(self.views.capacity() > self.views.len());
             self.views.push(v)
         } else {
             let data = buffers.get_unchecked(v.buffer_idx as usize);
@@ -143,6 +142,15 @@ impl<T: ViewType + ?Sized> BinaryViewColumnBuilder<T> {
         }
     }
 
+    /// # Safety
+    /// - caller ensure that view is duplicated
+    #[inline]
+    pub(crate) unsafe fn push_duplicated_view_unchecked(&mut self, v: View) {
+        let len = v.length;
+        self.total_bytes_len += len as usize;
+        self.views.push(v);
+    }
+
     pub fn push_value<V: AsRef<T>>(&mut self, value: V) {
         let value = value.as_ref();
         let bytes = value.to_bytes();

src/common/column/src/binview/mod.rs

@@ -18,6 +18,7 @@ pub(crate) mod fmt;
 mod iterator;
 mod view;
 
+use std::collections::HashMap;
 use std::fmt::Debug;
 use std::marker::PhantomData;
 use std::sync::Arc;
@@ -57,7 +58,6 @@ pub trait ViewType: Sealed + 'static + PartialEq + AsRef<Self> {
     /// # Safety
     /// The caller must ensure `index < self.len()`.
     unsafe fn from_bytes_unchecked(slice: &[u8]) -> &Self;
-
     fn to_bytes(&self) -> &[u8];
 
     #[allow(clippy::wrong_self_convention)]
@@ -344,6 +344,44 @@ impl<T: ViewType + ?Sized> BinaryViewColumnGeneric<T> {
         mutable.freeze()
     }
 
+    /// Garbage collect with dict compressed
+    pub fn gc_with_dict(&self, validity: Option<Bitmap>) -> Self {
+        if self.buffers.is_empty() {
+            return self.clone();
+        }
+        let mut map = HashMap::new();
+        let mut mutable = BinaryViewColumnBuilder::with_capacity(self.len());
+        let buffers = self.buffers.as_ref();
+
+        for (idx, (val, view)) in self.iter().zip(self.views.iter()).enumerate() {
+            // if it's null
+            if validity.as_ref().map(|v| !v.get_bit(idx)).unwrap_or(false) {
+                let default_v = View::new_inline(&[]);
+                mutable.views.push(default_v);
+                continue;
+            }
+
+            // did not care about buffer for small values
+            if view.length <= View::MAX_INLINE_SIZE {
+                mutable.total_bytes_len += view.length as usize;
+                mutable.views.push(*view);
+                continue;
+            }
+
+            match map.entry(val.to_bytes()) {
+                std::collections::hash_map::Entry::Occupied(v) => {
+                    unsafe { mutable.push_duplicated_view_unchecked(*v.get()) };
+                }
+                std::collections::hash_map::Entry::Vacant(vacant_entry) => {
+                    unsafe { mutable.push_view_unchecked(*view, buffers) };
+                    let last_view = mutable.views.last().unwrap();
+                    vacant_entry.insert(*last_view);
+                }
+            }
+        }
+        mutable.freeze()
+    }
+
     pub fn is_sliced(&self) -> bool {
         !std::ptr::eq(self.views.as_ptr(), self.views.data_ptr())
     }

src/common/column/src/binview/view.rs

@@ -49,6 +49,11 @@ impl View {
         unsafe { std::mem::transmute(self) }
     }
 
+    #[inline(always)]
+    pub fn as_i128(self) -> i128 {
+        unsafe { std::mem::transmute(self) }
+    }
+
     /// Create a new inline view without verifying the length
     ///
     /// # Safety

src/common/column/tests/it/binview/mod.rs

@@ -15,6 +15,7 @@
 mod builder;
 
 use databend_common_column::binview::BinaryViewColumn;
+use databend_common_column::binview::StringColumn;
 use databend_common_column::binview::Utf8ViewColumn;
 
 #[test]
@@ -97,6 +98,33 @@ fn basics_binary_view() {
     );
 }
 
+#[test]
+fn gc_dict() {
+    let data = (0..100).map(|c| format!("loooooooooooonstr{}", c % 3));
+    let array: StringColumn = data.into_iter().collect();
+    let array2 = array.gc_with_dict(None);
+
+    assert_eq!(array, array2);
+    assert_eq!(array.total_buffer_len(), 18 * 100);
+    assert_eq!(array2.total_buffer_len(), 18 * 3);
+
+    for i in (0..100).step_by(10) {
+        let s = array.clone().sliced(i, 10);
+        let s2 = s.gc_with_dict(None);
+
+        assert_eq!(s, s2);
+        assert_eq!(s2.total_buffer_len(), 18 * 3);
+    }
+}
+
+#[test]
+fn gc_dict_small() {
+    let data = (0..100).map(|c| format!("lo{}", c % 3));
+    let array: StringColumn = data.into_iter().collect();
+    let array2 = array.gc_with_dict(None);
+    assert_eq!(array, array2);
+}
+
 #[test]
 fn from() {
     let array1 = Utf8ViewColumn::from(["hello", " ", ""]);

src/common/native/src/compression/basic.rs

@@ -187,4 +187,30 @@ impl CommonCompression {
         }
         Ok(())
     }
+
+    pub fn decompress_buffer<R: NativeReadBuf>(
+        &self,
+        reader: &mut R,
+        _uncompressed_size: usize,
+        compressed_size: usize,
+        out_slice: &mut [u8],
+        scratch: &mut Vec<u8>,
+    ) -> Result<()> {
+        let mut use_inner = false;
+        reader.fill_buf()?;
+        let input = if reader.buffer_bytes().len() >= compressed_size {
+            use_inner = true;
+            reader.buffer_bytes()
+        } else {
+            scratch.resize(compressed_size, 0);
+            reader.read_exact(scratch.as_mut_slice())?;
+            scratch.as_slice()
+        };
+
+        self.decompress(&input[..compressed_size], out_slice)?;
+        if use_inner {
+            reader.consume(compressed_size);
+        }
+        Ok(())
+    }
 }

src/common/native/src/compression/mod.rs

@@ -18,6 +18,7 @@ pub mod binary;
 pub mod boolean;
 pub mod double;
 pub mod integer;
+pub mod view;
 
 pub use basic::CommonCompression;
 use databend_common_expression::types::Bitmap;

src/common/native/src/compression/view/mod.rs

@@ -0,0 +1,55 @@
+// Copyright 2021 Datafuse Labs
+//
+// Licensed 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 databend_common_column::binview::BinaryViewColumn;
+use databend_common_column::bitmap::Bitmap;
+use databend_common_expression::types::Buffer;
+
+use super::basic::CommonCompression;
+use crate::compression::integer::compress_integer;
+use crate::error::Result;
+use crate::write::WriteOptions;
+
+pub fn compress_view(
+    col: &BinaryViewColumn,
+    validity: Option<Bitmap>,
+    buf: &mut Vec<u8>,
+    write_options: WriteOptions,
+) -> Result<()> {
+    // choose compressor
+    let col = col.gc_with_dict(validity.clone());
+    let view_array: Buffer<i128> = col.views().iter().map(|x| x.as_i128()).collect();
+    compress_integer(&view_array, validity, write_options.clone(), buf)?;
+    compress_buffers(&col, buf, write_options.default_compression)
+}
+
+fn compress_buffers(
+    array: &BinaryViewColumn,
+    buf: &mut Vec<u8>,
+    c: CommonCompression,
+) -> Result<()> {
+    let codec = c.to_compression() as u8;
+    buf.extend_from_slice(&(array.data_buffers().len() as u32).to_le_bytes());
+    // let's encode the buffers
+    for buffer in array.data_buffers().iter() {
+        buf.extend_from_slice(&[codec]);
+        let pos = buf.len();
+        buf.extend_from_slice(&[0u8; 8]);
+
+        let compressed_size = c.compress(buffer.as_slice(), buf)?;
+        buf[pos..pos + 4].copy_from_slice(&(compressed_size as u32).to_le_bytes());
+        buf[pos + 4..pos + 8].copy_from_slice(&(buffer.len() as u32).to_le_bytes());
+    }
+    Ok(())
+}

src/common/native/src/read/array/view.rs

@@ -23,6 +23,7 @@ use databend_common_expression::types::Buffer;
 use databend_common_expression::Column;
 use databend_common_expression::TableDataType;
 
+use crate::compression::integer::decompress_integer;
 use crate::error::Result;
 use crate::nested::InitNested;
 use crate::nested::NestedState;
@@ -112,22 +113,21 @@ fn read_view_col<R: NativeReadBuf>(
     data_type: TableDataType,
     validity: Option<Bitmap>,
 ) -> Result<Column> {
-    let mut scratch = vec![0; 9];
-    let (_c, _compressed_size, _uncompressed_size) = read_compress_header(reader, &mut scratch)?;
+    let mut scratch = Vec::new();
+    let mut views: Vec<i128> = Vec::with_capacity(length);
+    decompress_integer(reader, length, &mut views, &mut scratch)?;
 
-    let mut buffer = vec![View::default(); length];
-    let temp_data =
-        unsafe { std::slice::from_raw_parts_mut(buffer.as_mut_ptr() as *mut u8, length * 16) };
-    reader.read_exact(temp_data)?;
-    let views = Buffer::from(buffer);
+    let views: Buffer<i128> = views.into();
+    let views = unsafe { std::mem::transmute::<Buffer<i128>, Buffer<View>>(views) };
 
     let buffer_len = reader.read_u32::<LittleEndian>()?;
-    let mut buffers = Vec::with_capacity(buffer_len as _);
+    let mut buffers = Vec::with_capacity(buffer_len as usize);
 
     for _ in 0..buffer_len {
         scratch.clear();
         let (compression, compressed_size, uncompressed_size) =
             read_compress_header(reader, &mut scratch)?;
+
         let c = CommonCompression::try_from(&compression)?;
         let mut buffer = vec![];
         c.decompress_common_binary(
@@ -137,6 +137,7 @@ fn read_view_col<R: NativeReadBuf>(
             &mut buffer,
             &mut scratch,
         )?;
+
         buffers.push(Buffer::from(buffer));
     }
 

src/common/native/src/write/serialize.rs

@@ -60,7 +60,9 @@ pub fn write<W: Write>(
             }
         }),
         Column::Boolean(column) => write_bitmap(w, &column, validity, write_options, scratch),
-        Column::String(column) => write_view::<W>(w, &column.to_binview(), write_options, scratch),
+        Column::String(column) => {
+            write_view::<W>(w, &column.to_binview(), validity, write_options, scratch)
+        }
         Column::Timestamp(column) => {
             write_primitive::<i64, W>(w, &column, validity, write_options, scratch)
         }

src/common/native/src/write/view.rs

@@ -15,43 +15,21 @@
 use std::io::Write;
 
 use databend_common_column::binview::BinaryViewColumn;
-use databend_common_column::binview::View;
+use databend_common_column::bitmap::Bitmap;
 
 use super::WriteOptions;
+use crate::compression::view::compress_view;
 use crate::error::Result;
 
 pub(crate) fn write_view<W: Write>(
     w: &mut W,
     array: &BinaryViewColumn,
+    validity: Option<Bitmap>,
     write_options: WriteOptions,
-    buf: &mut Vec<u8>,
+    scratch: &mut Vec<u8>,
 ) -> Result<()> {
-    // TODO: adaptive gc and dict by stats
-    let array = array.clone().gc();
-    let c = write_options.default_compression;
-    let codec = c.to_compression();
-
-    let total_size = array.len() * std::mem::size_of::<View>()
-        + array.data_buffers().iter().map(|x| x.len()).sum::<usize>();
-    w.write_all(&[codec as u8])?;
-    w.write_all(&(total_size as u32).to_le_bytes())?;
-    w.write_all(&(total_size as u32).to_le_bytes())?;
-
-    let input_buf: &[u8] = bytemuck::cast_slice(array.views().as_slice());
-    w.write_all(input_buf)?;
-    w.write_all(&(array.data_buffers().len() as u32).to_le_bytes())?;
-
-    for buffer in array.data_buffers().iter() {
-        buf.clear();
-        let pos = buf.len();
-        w.write_all(&[codec as u8])?;
-        buf.extend_from_slice(&[0u8; 8]);
-
-        let compressed_size = c.compress(buffer.as_slice(), buf)?;
-        buf[pos..pos + 4].copy_from_slice(&(compressed_size as u32).to_le_bytes());
-        buf[pos + 4..pos + 8].copy_from_slice(&(buffer.len() as u32).to_le_bytes());
-        w.write_all(buf.as_slice())?;
-        buf.clear();
-    }
+    scratch.clear();
+    compress_view(array, validity, scratch, write_options)?;
+    w.write_all(scratch.as_slice())?;
     Ok(())
 }

src/common/native/tests/it/native/io.rs

@@ -109,6 +109,25 @@ fn test_freq() {
     test_write_read(chunk);
 }
 
+#[test]
+fn test_dict_string() {
+    let size = WRITE_PAGE * 5;
+
+    let values = vec![
+        (0..size)
+            .map(|s| format!("stringxxxxxxxx{}", s % 4))
+            .collect::<Vec<_>>(),
+        // (0..size)
+        //     .map(|s| format!("abc{}", s % 4))
+        //     .collect::<Vec<_>>(),
+    ];
+    for v in values {
+        let col = StringColumn::from_iter(v.iter());
+        let chunk = vec![Column::String(col)];
+        test_write_read(chunk);
+    }
+}
+
 #[test]
 fn test_bitpacking() {
     let size = WRITE_PAGE * 5;