From 1aaa34225f6f02e0c0d872dde6d3af3fe295b25d Mon Sep 17 00:00:00 2001 From: discord9 <55937128+discord9@users.noreply.github.com> Date: Fri, 4 Sep 2026 18:16:13 +0800 Subject: [PATCH] test(mito2): verify BSS write lifecycles Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> --- src/mito2/src/engine/basic_test.rs | 37 ++++++++++++++++++++++++- src/mito2/src/engine/compaction_test.rs | 31 +++++++++++++++++++++ src/mito2/src/engine/flush_test.rs | 37 ++++++++++++++++++++++++- 3 files changed, 103 insertions(+), 2 deletions(-) diff --git a/src/mito2/src/engine/basic_test.rs b/src/mito2/src/engine/basic_test.rs index 6d90f90b784..878215ba06b 100644 --- a/src/mito2/src/engine/basic_test.rs +++ b/src/mito2/src/engine/basic_test.rs @@ -29,6 +29,7 @@ use datatypes::prelude::ConcreteDataType; use datatypes::schema::ColumnSchema; use futures::TryStreamExt; use itertools::Itertools; +use parquet::basic::{Encoding, Type as PhysicalType}; use rstest::rstest; use rstest_reuse::{self, apply}; use store_api::metadata::ColumnMetadata; @@ -739,7 +740,9 @@ async fn test_engine_with_write_cache_with_format(flat_format: bool) { let engine = env.create_engine(mito_config).await; let region_id = RegionId::new(1, 1); - let request = CreateRequestBuilder::new().build(); + let request = CreateRequestBuilder::new() + .insert_option("experimental_sst_float_field_encoding", "byte_stream_split") + .build(); let column_schemas = rows_schema(&request); engine @@ -755,6 +758,38 @@ async fn test_engine_with_write_cache_with_format(flat_format: bool) { flush_region(&engine, region_id, None).await; + let region = engine.get_region(region_id).unwrap(); + let file = region + .version() + .ssts + .levels() + .iter() + .flat_map(|level| level.files.values()) + .next() + .expect("write-cache flushed SST") + .clone(); + let reader = region + .access_layer + .read_sst(file) + .build() + .await + .unwrap() + .expect("write-cache flushed SST reader"); + assert!( + reader + .parquet_metadata() + .row_groups() + .iter() + .flat_map(|row_group| row_group.columns()) + .any(|column| { + column.column_path().string() == "field_0" + && column.column_type() == PhysicalType::DOUBLE + && column + .encodings() + .any(|encoding| encoding == Encoding::BYTE_STREAM_SPLIT) + }) + ); + let request = ScanRequest::default(); let scanner = engine.scanner(region_id, request).await.unwrap(); diff --git a/src/mito2/src/engine/compaction_test.rs b/src/mito2/src/engine/compaction_test.rs index 04c97bdfbba..16f0ceac08a 100644 --- a/src/mito2/src/engine/compaction_test.rs +++ b/src/mito2/src/engine/compaction_test.rs @@ -26,6 +26,7 @@ use common_recordbatch::{RecordBatches, SendableRecordBatchStream}; use common_time::Timestamp; use datatypes::arrow::array::AsArray; use datatypes::arrow::datatypes::TimestampMillisecondType; +use parquet::basic::{Encoding, Type as PhysicalType}; use store_api::region_engine::{RegionEngine, RegionRole}; use store_api::region_request::AlterKind::SetRegionOptions; use store_api::region_request::{ @@ -954,6 +955,7 @@ async fn test_compaction_region_with_format(flat_format: bool) { let request = CreateRequestBuilder::new() .insert_option("compaction.type", "twcs") + .insert_option("experimental_sst_float_field_encoding", "byte_stream_split") .build(); let column_schemas = request @@ -974,6 +976,35 @@ async fn test_compaction_region_with_format(flat_format: bool) { compact(&engine, region_id).await; + let region = engine.get_region(region_id).unwrap(); + let file = region.version().ssts.levels()[1] + .files + .values() + .next() + .expect("compaction output SST") + .clone(); + let reader = region + .access_layer + .read_sst(file) + .build() + .await + .unwrap() + .expect("compaction output SST reader"); + assert!( + reader + .parquet_metadata() + .row_groups() + .iter() + .flat_map(|row_group| row_group.columns()) + .any(|column| { + column.column_path().string() == "field_0" + && column.column_type() == PhysicalType::DOUBLE + && column + .encodings() + .any(|encoding| encoding == Encoding::BYTE_STREAM_SPLIT) + }) + ); + let scanner = engine .scanner(region_id, ScanRequest::default()) .await diff --git a/src/mito2/src/engine/flush_test.rs b/src/mito2/src/engine/flush_test.rs index b71ce9f9fc0..4f680988e9d 100644 --- a/src/mito2/src/engine/flush_test.rs +++ b/src/mito2/src/engine/flush_test.rs @@ -27,6 +27,7 @@ use common_base::readable_size::ReadableSize; use common_recordbatch::RecordBatches; use common_time::util::current_time_millis; use common_wal::options::{KafkaWalOptions, WAL_OPTIONS_KEY, WalOptions}; +use parquet::basic::{Encoding, Type as PhysicalType}; use rstest::rstest; use rstest_reuse::{self, apply}; use store_api::ManifestVersion; @@ -279,7 +280,9 @@ async fn test_manual_flush_with_format(flat_format: bool) { ) .await; - let request = CreateRequestBuilder::new().build(); + let request = CreateRequestBuilder::new() + .insert_option("experimental_sst_float_field_encoding", "byte_stream_split") + .build(); let column_schemas = rows_schema(&request); engine @@ -295,6 +298,38 @@ async fn test_manual_flush_with_format(flat_format: bool) { flush_region(&engine, region_id, None).await; + let region = engine.get_region(region_id).unwrap(); + let file = region + .version() + .ssts + .levels() + .iter() + .flat_map(|level| level.files.values()) + .next() + .expect("flushed SST") + .clone(); + let reader = region + .access_layer + .read_sst(file) + .build() + .await + .unwrap() + .expect("flushed SST reader"); + assert!( + reader + .parquet_metadata() + .row_groups() + .iter() + .flat_map(|row_group| row_group.columns()) + .any(|column| { + column.column_path().string() == "field_0" + && column.column_type() == PhysicalType::DOUBLE + && column + .encodings() + .any(|encoding| encoding == Encoding::BYTE_STREAM_SPLIT) + }) + ); + let request = ScanRequest::default(); let scanner = engine.scanner(region_id, request).await.unwrap(); assert_eq!(0, scanner.num_memtables());