feat: block stream write (#17744)

* add block serializer

* fix test

* add settings enable_block_stream_write

* fix test

* Optimize handling of tail fragments and block serialization

* make lint

Author: zhyass

Cargo.lock

@@ -3,13 +3,11 @@
 set -e
 
 ARCH=x86_64
-glibc=`getconf GNU_LIBC_VERSION|awk '{print $2}'`
+glibc=$(getconf GNU_LIBC_VERSION | awk '{print $2}')
 TARGET=${ARCH}-linux-glib$glibc
 MUSL_RUST_TARGET=${ARCH}-unknown-linux-musl
 BENSQL_VERSION=0.26.1
 
-
-
 RUSTFLAGS="-C target-feature=+sse4.2 -C link-arg=-Wl,--compress-debug-sections=zlib"
 
 echo "==> https_proxy=${https_proxy}"
@@ -21,14 +19,13 @@ mkdir -p distro/{pkg,bin,configs,systemd,scripts}
 BENSQL_PKG=bendsql-${MUSL_RUST_TARGET}.tar.gz
 
 if [ ! -f distro/pkg/${BENSQL_PKG} ]; then
-  echo "==> downloading ${BENSQL_PKG} ..."
-  curl -L https://github.com/databendlabs/bendsql/releases/download/v${BENSQL_VERSION}/${BENSQL_PKG} -o distro/pkg/${BENSQL_PKG}
+	echo "==> downloading ${BENSQL_PKG} ..."
+	curl -L https://github.com/databendlabs/bendsql/releases/download/v${BENSQL_VERSION}/${BENSQL_PKG} -o distro/pkg/${BENSQL_PKG}
 fi
 
 tar -xzvf distro/pkg/${BENSQL_PKG} -C distro/bin
 distro/bin/bendsql --version
 
-
 rustup show
 
 DATABEND_VERSION=$(git describe --tags --abbrev=0)
@@ -38,9 +35,9 @@ echo "==> building databend ${DATABEND_VERSION} for ${TARGET} ..."
 #cargo clean
 
 cargo build --release \
-  --bin=databend-query \
-  --bin=databend-meta \
-  --bin=databend-metactl
+	--bin=databend-query \
+	--bin=databend-meta \
+	--bin=databend-metactl
 
 #  --bin=table-meta-inspector
 

scripts/build/build-pkg.sh

@@ -76,8 +76,8 @@ impl<W: Write> NativeWriter<W> {
     }
 
     /// Consumes itself into the inner writer
-    pub fn into_inner(self) -> W {
-        self.writer.w
+    pub fn inner_mut(&mut self) -> &mut W {
+        &mut self.writer.w
     }
 
     /// Writes the header and first (schema) message to the file.

src/common/native/src/write/writer.rs

@@ -12,7 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use std::cmp::Ordering;
 use std::collections::HashMap;
 
 use databend_common_exception::ErrorCode;
@@ -132,15 +131,6 @@ pub fn u64_literal(val: u64) -> Scalar {
     Scalar::Number(NumberScalar::UInt64(val))
 }
 
-pub fn cmp_with_null(v1: &Scalar, v2: &Scalar) -> Ordering {
-    match (v1.is_null(), v2.is_null()) {
-        (true, true) => Ordering::Equal,
-        (true, false) => Ordering::Greater,
-        (false, true) => Ordering::Less,
-        (false, false) => v1.cmp(v2),
-    }
-}
-
 pub fn parse_sequence_args(table_args: &TableArgs, func_name: &str) -> Result<String> {
     let args = table_args.expect_all_positioned(func_name, Some(1))?;
     let sequence = string_value(&args[0])?;

src/query/catalog/src/table_args.rs

@@ -36,7 +36,7 @@ impl Default for BlockThresholds {
         BlockThresholds {
             max_rows_per_block: DEFAULT_BLOCK_ROW_COUNT,
             min_rows_per_block: (DEFAULT_BLOCK_ROW_COUNT * 4).div_ceil(5),
-            max_bytes_per_block: DEFAULT_BLOCK_BUFFER_SIZE,
+            max_bytes_per_block: DEFAULT_BLOCK_BUFFER_SIZE * 2,
             min_bytes_per_block: (DEFAULT_BLOCK_BUFFER_SIZE * 4).div_ceil(5),
             max_compressed_per_block: DEFAULT_BLOCK_COMPRESSED_SIZE,
             min_compressed_per_block: (DEFAULT_BLOCK_COMPRESSED_SIZE * 4).div_ceil(5),
@@ -48,15 +48,15 @@ impl Default for BlockThresholds {
 impl BlockThresholds {
     pub fn new(
         max_rows_per_block: usize,
-        max_bytes_per_block: usize,
+        bytes_per_block: usize,
         max_compressed_per_block: usize,
         block_per_segment: usize,
     ) -> Self {
         BlockThresholds {
             max_rows_per_block,
             min_rows_per_block: (max_rows_per_block * 4).div_ceil(5),
-            max_bytes_per_block,
-            min_bytes_per_block: (max_bytes_per_block * 4).div_ceil(5),
+            max_bytes_per_block: bytes_per_block * 2,
+            min_bytes_per_block: (bytes_per_block * 4).div_ceil(5),
             max_compressed_per_block,
             min_compressed_per_block: (max_compressed_per_block * 4).div_ceil(5),
             block_per_segment,

src/query/expression/src/utils/block_thresholds.rs

@@ -159,7 +159,8 @@ impl PipelineBuilder {
                     SortPipelineBuilder::create(self.ctx.clone(), schema, sort_descs.into())?
                         .with_block_size_hit(sort_block_size)
                         .remove_order_col_at_last();
-                sort_pipeline_builder.build_merge_sort_pipeline(&mut self.main_pipeline, false)?;
+                // Todo(zhyass): Recluster will no longer perform sort in the near future.
+                sort_pipeline_builder.build_full_sort_pipeline(&mut self.main_pipeline)?;
 
                 // Compact after merge sort.
                 let max_threads = self.ctx.get_settings().get_max_threads()? as usize;

src/query/service/src/pipelines/builders/builder_recluster.rs

@@ -26,6 +26,7 @@ use databend_common_storages_fuse::io::WriteSettings;
 use databend_common_storages_fuse::FuseStorageFormat;
 use databend_storages_common_blocks::blocks_to_parquet;
 use databend_storages_common_index::BloomIndex;
+use databend_storages_common_index::BloomIndexBuilder;
 use databend_storages_common_table_meta::meta::BlockMeta;
 use databend_storages_common_table_meta::meta::ClusterStatistics;
 use databend_storages_common_table_meta::meta::Compression;
@@ -128,12 +129,9 @@ impl<'a> BlockWriter<'a> {
         let bloom_index_cols = BloomIndexColumns::All;
         let bloom_columns_map =
             bloom_index_cols.bloom_index_fields(schema.clone(), BloomIndex::supported_type)?;
-        let maybe_bloom_index = BloomIndex::try_create(
-            FunctionContext::default(),
-            location.1,
-            block,
-            bloom_columns_map,
-        )?;
+        let mut builder = BloomIndexBuilder::create(FunctionContext::default(), bloom_columns_map);
+        builder.add_block(block)?;
+        let maybe_bloom_index = builder.finalize()?;
         if let Some(bloom_index) = maybe_bloom_index {
             let index_block = bloom_index.serialize_to_data_block()?;
             let filter_schema = bloom_index.filter_schema;

src/query/service/src/test_kits/block_writer.rs

@@ -84,8 +84,8 @@ fn test_ft_stats_block_stats() -> databend_common_exception::Result<()> {
 #[test]
 fn test_ft_stats_block_stats_with_column_distinct_count() -> databend_common_exception::Result<()> {
     let schema = Arc::new(TableSchema::new(vec![
-        TableField::new("a", TableDataType::Number(NumberDataType::Int32)),
-        TableField::new("b", TableDataType::String),
+        TableField::new_from_column_id("a", TableDataType::Number(NumberDataType::Int32), 0),
+        TableField::new_from_column_id("b", TableDataType::String, 1),
     ]));
 
     let block = DataBlock::new_from_columns(vec![

src/query/service/tests/it/storages/fuse/statistics.rs

@@ -1250,14 +1250,20 @@ impl DefaultSettings {
                     range: Some(SettingRange::Numeric(0..=1)),
                 }),
                 // Add this to the HashMap in DefaultSettings::instance()
-   ("optimizer_skip_list", DefaultSettingValue {
-                value: UserSettingValue::String(String::new()),
-                desc: "Comma-separated(,) list of optimizer names to skip during query optimization",
-                mode: SettingMode::Both,
-                scope: SettingScope::Both,
-                range: None,
-            }),
-
+                ("optimizer_skip_list", DefaultSettingValue {
+                    value: UserSettingValue::String(String::new()),
+                    desc: "Comma-separated(,) list of optimizer names to skip during query optimization",
+                    mode: SettingMode::Both,
+                    scope: SettingScope::Both,
+                    range: None,
+                }),
+                ("enable_block_stream_write", DefaultSettingValue {
+                    value: UserSettingValue::UInt64(0),
+                    desc: "Enables block stream write",
+                    mode: SettingMode::Both,
+                    scope: SettingScope::Both,
+                    range: Some(SettingRange::Numeric(0..=1)),
+                }),
             ]);
 
             Ok(Arc::new(DefaultSettings {

src/query/settings/src/settings_default.rs

@@ -932,4 +932,8 @@ impl Settings {
     pub fn get_optimizer_skip_list(&self) -> Result<String> {
         self.try_get_string("optimizer_skip_list")
     }
+
+    pub fn get_enable_block_stream_write(&self) -> Result<bool> {
+        Ok(self.try_get_u64("enable_block_stream_write")? == 1)
+    }
 }