test(mito2): verify BSS write lifecycles

Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>

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();
 

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

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());