feat(mito2): add opt-in byte-stream-split encoding for float SST fields (#9069)

* feat(mito2): add opt-in byte stream split encoding

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

* fix(mito2): correct float encoding checks

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

* test(compat): cover float SST encoding

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

* fix(mito2): compile float encoding tests

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

* fix(mito2): release parquet test writer

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

* fix(mito2): register float test primary key

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

* test(mito2): verify BSS write lifecycles

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

* test(metric-engine): verify BSS physical SST

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

* test(mito2): verify bulk BSS lifecycle

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

* fix(mito2): compile bulk BSS test

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

* refactor(mito2): narrow bulk encoding constructors

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

* test(compat): accept generated float upgrade output

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

* test(compat): accept generated float downgrade output

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

* refactor(mito2): narrow bulk encoding builder

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

* test(perf): add default versus BSS storage comparison

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

* test(perf): align BSS reader benchmarks with prior study

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

* fix(perf): parse current read benchmark averages

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

* fix(perf): retain default float encoding in direct SST fixtures

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

* test(perf): isolate BSS user SSTs and benchmark every file

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

* docs(perf): record measured BSS storage and reader tradeoffs

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

* docs(perf): expose warm scan variability and evidence limits

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

* docs(perf): clarify BSS baseline and storage measurement scope

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

* test(perf): model bounded mixed integer and fractional metric series

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

* docs(perf): report bounded mixed BSS measurements and query regressions

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

* docs(perf): qualify timings affected by concurrent host builds

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

* test(perf): include float BSS comparison in default regression cases

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

* fix(perf): omit unsupported float encoding option from baseline setup

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

---------

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

.github/runner-scale-sets/query-regression/README.md

@@ -108,7 +108,7 @@ agent, which Aliyun public Ubuntu images include.
 
 An allowlisted repository admin commenting `/query-regression` on the PR is
 **trust admission for that exact PR revision**. `/query-regression` runs the
-six routine default cases; `/query-regression heavy` runs only the
+nine routine default cases; `/query-regression heavy` runs only the
 high-cardinality `prom_remote_write_7913` remote-write case.
 `slash-command-dispatch.yml` (`issue_comment` on the default branch) parses
 the command and `repository_dispatch`es; `query-regression-slash.yml` admits

.github/scripts/query-regression-run.py

@@ -33,6 +33,7 @@ from typing import Any
 
 DEFAULT_CASES = [
     "tests/perf/query_cases/smoke_direct_sst/case.toml",
+    "tests/perf/query_cases/sst_float_bss/case.toml",
     "tests/perf/query_cases/prom_remote_write_seeded_random/case.toml",
     "tests/perf/query_cases/prom_remote_write_run_heavy/case.toml",
     "tests/perf/query_cases/prom_remote_write_mixed_every/case.toml",

src/cmd/src/bin/query_perf_fixture/case.rs

@@ -105,6 +105,10 @@ pub(super) struct PromRemoteWritePlan {
     #[serde(default = "default_visibility_timeout_seconds")]
     pub(super) visibility_timeout_seconds: u64,
     #[serde(default)]
+    pub(super) base_setup_sql: Vec<String>,
+    #[serde(default)]
+    pub(super) candidate_setup_sql: Vec<String>,
+    #[serde(default)]
     pub(super) prom_store: PromStoreConfig,
     #[serde(default)]
     pub(super) value: ValueConfig,
@@ -200,6 +204,7 @@ pub(super) enum ValuePattern {
     QuantizedSignal,
     SignalWithSporadicStalls,
     MixedSignalRepeated,
+    BoundedMixed,
 }
 
 impl std::fmt::Display for ValuePattern {
@@ -436,6 +441,63 @@ pub(super) fn default_parallelism() -> u64 {
     1
 }
 
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn remote_write_setup_sql_defaults_to_empty() {
+        let case: CaseFile = toml::from_str(
+            r#"
+[scenario]
+kind = "prom_remote_write_then_query"
+
+[scenario.remote_write]
+metric = "metric"
+"#,
+        )
+        .unwrap();
+        let Scenario::PromRemoteWriteThenQuery(scenario) = case.scenario else {
+            panic!("expected prom_remote_write_then_query scenario");
+        };
+        assert!(scenario.remote_write.base_setup_sql.is_empty());
+        assert!(scenario.remote_write.candidate_setup_sql.is_empty());
+    }
+
+    #[test]
+    fn remote_write_setup_sql_roundtrips() {
+        let case: CaseFile = toml::from_str(
+            r#"
+[scenario]
+kind = "prom_remote_write_then_query"
+
+[scenario.remote_write]
+metric = "metric"
+base_setup_sql = ["CREATE TABLE base_table"]
+candidate_setup_sql = ["CREATE TABLE candidate_table", "ALTER TABLE candidate_table SET 'x'='y'"]
+"#,
+        )
+        .unwrap();
+
+        let roundtrip: CaseFile =
+            serde_json::from_str(&serde_json::to_string(&case).unwrap()).unwrap();
+        let Scenario::PromRemoteWriteThenQuery(scenario) = roundtrip.scenario else {
+            panic!("expected prom_remote_write_then_query scenario");
+        };
+        assert_eq!(
+            scenario.remote_write.base_setup_sql,
+            ["CREATE TABLE base_table"]
+        );
+        assert_eq!(
+            scenario.remote_write.candidate_setup_sql,
+            [
+                "CREATE TABLE candidate_table",
+                "ALTER TABLE candidate_table SET 'x'='y'"
+            ]
+        );
+    }
+}
+
 impl Scenario {
     pub(super) fn kind(&self) -> &'static str {
         match self {

src/cmd/src/bin/query_perf_fixture/direct_sst.rs

@@ -469,6 +469,7 @@ pub(super) async fn run_direct_sst(args: DirectArgs) {
             write_buffer_size: DEFAULT_WRITE_BUFFER_SIZE,
             row_group_size: scenario.layout.row_group_size,
             max_file_size: None,
+            float_field_encoding: Default::default(),
         };
         let infos = match format {
             FormatType::Flat => writer.write_all_flat(source, Some(sequence), &opts).await,

src/cmd/src/bin/query_perf_fixture/prom_remote_write.rs

@@ -197,8 +197,35 @@ fn deterministic_prom_value(args: &PromRemoteWriteArgs, series_idx: u64, sample_
                 args.value_base + (series_idx % 97) as f64 + local as f64 * args.value_step
             }
         }
+        ValuePattern::BoundedMixed => bounded_mixed_value(args, series_idx, local),
     }
 }
+
+fn bounded_mixed_value(args: &PromRemoteWriteArgs, series_idx: u64, local: u64) -> f64 {
+    const SPANS: [i64; 4] = [10, 1_000, 100_000, 10_000_000];
+    let series = splitmix64(series_idx ^ args.value_seed);
+    let span = SPANS[(series >> 62) as usize] * (1 + ((series >> 32) % 10) as i64);
+    let baseline = (series % args.value_cardinality.max(1)) as f64 * span as f64 / 10.0;
+    let run_length = args.value_run_length.max(1);
+    let group = local / run_length;
+    let phase = local % run_length;
+    let anchor = bounded_mixed_anchor(series, group, span);
+    let next_anchor = bounded_mixed_anchor(series, group + 1, span);
+    let signal = baseline + anchor + (next_anchor - anchor) * phase as f64 / run_length as f64;
+    let value = (args.value_base + args.value_step * signal).round();
+    if series_idx % args.value_mixed_every.max(1) == args.value_mixed_every.max(1) - 1 {
+        let numerator = splitmix64(series ^ local) % 999 + 1;
+        value + numerator as f64 / 1_000.0
+    } else {
+        value
+    }
+}
+
+fn bounded_mixed_anchor(series: u64, group: u64, span: i64) -> f64 {
+    let limit = span - 1;
+    (splitmix64(series ^ group) % (2 * limit as u64 + 1)) as i64 as f64 - limit as f64
+}
+
 fn splitmix64(mut value: u64) -> u64 {
     value = value.wrapping_add(0x9e3779b97f4a7c15);
     let mut mixed = value;
@@ -307,7 +334,7 @@ mod tests {
             for chunk_idx in 0..10 {
                 let mut chunk = args(ValuePattern::SeededRandom);
                 chunk.samples_per_series = 1_440;
-                chunk.value_sample_offset = chunk_idx * 1_440;
+                chunk.value_sample_offset = chunk_idx * 1_439;
                 chunk.value_total_samples_per_series = Some(14_400);
                 for sample_idx in 0..chunk.samples_per_series {
                     assert_eq!(
@@ -319,6 +346,118 @@ mod tests {
         }
     }
 
+    #[test]
+    fn bounded_mixed_has_bounded_95_5_per_series_values() {
+        let mut args = args(ValuePattern::BoundedMixed);
+        args.series_count = 1_000;
+        args.samples_per_series = 4_320;
+        args.value_cardinality = 1_000;
+        args.value_seed = 42;
+        args.value_run_length = 60;
+        args.value_mixed_every = 20;
+
+        let mut integral_series = 0;
+        let mut fractional_series = 0;
+        let mut spans = HashSet::new();
+        let mut ranges = HashSet::new();
+        for series_idx in 0..args.series_count {
+            let series = splitmix64(series_idx ^ args.value_seed);
+            let span = [10_i64, 1_000, 100_000, 10_000_000][(series >> 62) as usize]
+                * (1 + ((series >> 32) % 10) as i64);
+            let baseline = (series % args.value_cardinality.max(1)) as f64 * span as f64 / 10.0;
+            let values = (0..args.samples_per_series)
+                .map(|sample_idx| deterministic_prom_value(&args, series_idx, sample_idx))
+                .collect::<Vec<_>>();
+
+            assert!(values.iter().all(|value| value.is_finite()));
+            assert!(
+                values
+                    .iter()
+                    .all(|value| *value >= baseline - span as f64 - 1.0
+                        && *value <= baseline + span as f64 + 1.0)
+            );
+            assert!(
+                values.windows(2).any(|pair| pair[1] > pair[0]),
+                "series {series_idx} never rises"
+            );
+            assert!(
+                values.windows(2).any(|pair| pair[1] < pair[0]),
+                "series {series_idx} never falls"
+            );
+            ranges.insert((
+                values
+                    .iter()
+                    .copied()
+                    .fold(f64::INFINITY, f64::min)
+                    .to_bits(),
+                values
+                    .iter()
+                    .copied()
+                    .fold(f64::NEG_INFINITY, f64::max)
+                    .to_bits(),
+            ));
+            spans.insert(span);
+            if series_idx % args.value_mixed_every == args.value_mixed_every - 1 {
+                fractional_series += 1;
+                assert!(values.iter().all(|value| value.fract() != 0.0));
+            } else {
+                integral_series += 1;
+                assert!(values.iter().all(|value| value.fract() == 0.0));
+            }
+        }
+
+        assert_eq!((integral_series, fractional_series), (950, 50));
+        assert!(spans.len() > 4);
+        assert!(ranges.len() > 1);
+    }
+
+    #[test]
+    fn bounded_mixed_is_seeded_and_chunk_bit_identical() {
+        let mut monolithic = args(ValuePattern::BoundedMixed);
+        monolithic.samples_per_series = 14_400;
+        monolithic.value_total_samples_per_series = Some(14_400);
+        monolithic.value_cardinality = 1_000;
+        monolithic.value_seed = 42;
+        monolithic.value_run_length = 60;
+        monolithic.value_mixed_every = 20;
+
+        let original = (0..600)
+            .map(|sample_idx| deterministic_prom_value(&monolithic, 19, sample_idx))
+            .collect::<Vec<_>>();
+        monolithic.value_seed = 43;
+        assert_ne!(
+            original,
+            (0..600)
+                .map(|sample_idx| deterministic_prom_value(&monolithic, 19, sample_idx))
+                .collect::<Vec<_>>()
+        );
+
+        monolithic.value_seed = 42;
+        for series_idx in [0, 19] {
+            for chunk_idx in 0..10 {
+                let mut chunk = args(ValuePattern::BoundedMixed);
+                chunk.samples_per_series = 1_440;
+                chunk.value_sample_offset = chunk_idx * 1_439;
+                chunk.value_total_samples_per_series = Some(14_400);
+                chunk.value_cardinality = 1_000;
+                chunk.value_seed = 42;
+                chunk.value_run_length = 60;
+                chunk.value_mixed_every = 20;
+                for sample_idx in 0..chunk.samples_per_series {
+                    assert_eq!(
+                        deterministic_prom_value(&chunk, series_idx, sample_idx).to_bits(),
+                        deterministic_prom_value(
+                            &monolithic,
+                            series_idx,
+                            chunk.value_sample_offset + sample_idx
+                        )
+                        .to_bits()
+                    );
+                }
+            }
+        }
+    }
+
     #[test]
     fn values_continue_across_series_boundaries() {
         let mut args = args(ValuePattern::Unique);

src/cmd/src/bin/query_regression_runner/model.rs

@@ -116,6 +116,10 @@ pub(super) struct RemoteWrite {
     pub(super) sample_chunk_size: Option<u64>,
     pub(super) flush_every_sample_chunks: u64,
     pub(super) visibility_timeout_seconds: u64,
+    #[serde(default)]
+    pub(super) base_setup_sql: Vec<String>,
+    #[serde(default)]
+    pub(super) candidate_setup_sql: Vec<String>,
     pub(super) prom_store: PromStore,
     pub(super) value: RemoteValue,
     pub(super) storage: Option<StorageConfig>,

src/cmd/src/bin/query_regression_runner/remote/ingest.rs

@@ -57,6 +57,7 @@ pub(super) async fn run_prepare_remote(args: PrepareRemoteArgs) -> Result<()> {
         args.base_http_port,
         &args.fixture_generator,
         &remote,
+        &remote.base_setup_sql,
         &client,
     )
     .await?;
@@ -65,6 +66,7 @@ pub(super) async fn run_prepare_remote(args: PrepareRemoteArgs) -> Result<()> {
         args.candidate_http_port,
         &args.fixture_generator,
         &remote,
+        &remote.candidate_setup_sql,
         &client,
     )
     .await?;
@@ -88,6 +90,7 @@ async fn prepare_remote_target(
     port: u16,
     generator: &Path,
     remote: &RemoteWrite,
+    setup_sql: &[String],
     client: &Client,
 ) -> Result<Value> {
     let create_database = http_post_sql(
@@ -107,6 +110,7 @@ async fn prepare_remote_target(
         )
         .into());
     }
+    let setup = run_setup_sql(client, port, &remote.database, name, setup_sql).await?;
     let (remote_write, flushes) = ingest_remote_write(generator, port, remote, client).await?;
     let expected_rows = remote
         .series_count
@@ -124,6 +128,7 @@ async fn prepare_remote_target(
     Ok(json!({
         "name": name,
         "create_database": create_database,
+        "setup_sql": setup,
         "remote_write": remote_write,
         "flushes": flushes,
         "visibility": visibility,
@@ -131,6 +136,28 @@ async fn prepare_remote_target(
     }))
 }
 
+async fn run_setup_sql(
+    client: &Client,
+    port: u16,
+    database: &str,
+    target: &str,
+    setup_sql: &[String],
+) -> Result<Vec<Value>> {
+    let mut results = Vec::with_capacity(setup_sql.len());
+    for (index, statement) in setup_sql.iter().enumerate() {
+        let result = http_post_sql(client, port, statement, database).await;
+        if !result["ok"].as_bool().unwrap_or(false) {
+            return Err(format!(
+                "setup SQL statement {} failed for {target}: {result}",
+                index + 1
+            )
+            .into());
+        }
+        results.push(result);
+    }
+    Ok(results)
+}
+
 #[derive(Clone, Debug, PartialEq, Eq)]
 struct SampleChunk {
     index: u64,
@@ -434,9 +461,110 @@ fn json_number(value: f64) -> String {
 
 #[cfg(test)]
 mod tests {
+    use tokio::io::{AsyncReadExt, AsyncWriteExt};
+    use tokio::net::{TcpListener, TcpStream};
+
     use super::*;
     use crate::query_regression_runner::model::{PromStore, RemoteValue};
 
+    async fn read_http_request(stream: &mut TcpStream) -> Vec<u8> {
+        let mut request = Vec::new();
+        let mut chunk = [0; 1024];
+        loop {
+            let read = stream.read(&mut chunk).await.unwrap();
+            assert_ne!(
+                read, 0,
+                "HTTP client closed request before sending its body"
+            );
+            request.extend_from_slice(&chunk[..read]);
+            let Some(header_end) = request.windows(4).position(|window| window == b"\r\n\r\n")
+            else {
+                continue;
+            };
+            let headers = std::str::from_utf8(&request[..header_end]).unwrap();
+            let content_length = headers
+                .lines()
+                .find_map(|line| line.strip_prefix("content-length: "))
+                .or_else(|| {
+                    headers
+                        .lines()
+                        .find_map(|line| line.strip_prefix("Content-Length: "))
+                })
+                .unwrap()
+                .parse::<usize>()
+                .unwrap();
+            if request.len() >= header_end + 4 + content_length {
+                return request;
+            }
+        }
+    }
+
+    async fn spawn_sql_server(
+        responses: Vec<String>,
+    ) -> (u16, tokio::task::JoinHandle<Vec<String>>) {
+        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let port = listener.local_addr().unwrap().port();
+        let server = tokio::spawn(async move {
+            let mut requests = Vec::with_capacity(responses.len());
+            for body in responses {
+                let (mut stream, _) = listener.accept().await.unwrap();
+                let request = read_http_request(&mut stream).await;
+                requests.push(String::from_utf8(request).unwrap());
+                let response = format!(
+                    "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\n\
+                     Connection: close\r\nContent-Length: {}\r\n\r\n{body}",
+                    body.len()
+                );
+                stream.write_all(response.as_bytes()).await.unwrap();
+            }
+            requests
+        });
+        (port, server)
+    }
+
+    #[tokio::test]
+    async fn setup_sql_runs_in_order_with_configured_database() {
+        let (port, server) = spawn_sql_server(vec!["{\"code\":0}".to_string(); 2]).await;
+        let client = Client::new();
+        let setup = vec![
+            "CREATE TABLE physical (ts TIMESTAMP TIME INDEX) ENGINE=metric".to_string(),
+            "ALTER TABLE physical SET 'compaction.twcs.trigger_file_num'='100'".to_string(),
+        ];
+
+        let results = run_setup_sql(&client, port, "perf", "candidate", &setup)
+            .await
+            .unwrap();
+
+        assert_eq!(results.len(), 2);
+        let requests = server.await.unwrap();
+        assert_eq!(requests.len(), 2);
+        assert!(requests[0].contains("db=perf"));
+        assert!(requests[0].contains("sql=CREATE"));
+        assert!(requests[1].contains("db=perf"));
+        assert!(requests[1].contains("sql=ALTER"));
+    }
+
+    #[tokio::test]
+    async fn setup_sql_stops_at_the_failing_statement() {
+        let (port, server) =
+            spawn_sql_server(vec!["{\"code\":0}".to_string(), "{\"code\":1}".to_string()]).await;
+        let client = Client::new();
+        let setup = vec![
+            "CREATE TABLE physical (ts TIMESTAMP TIME INDEX) ENGINE=metric".to_string(),
+            "invalid SQL".to_string(),
+            "ALTER TABLE physical SET 'compaction.twcs.trigger_file_num'='100'".to_string(),
+        ];
+
+        let error = run_setup_sql(&client, port, "perf", "base", &setup)
+            .await
+            .unwrap_err();
+
+        assert!(error.to_string().contains("statement 2 failed for base"));
+        let requests = server.await.unwrap();
+        assert_eq!(requests.len(), 2);
+        assert!(requests[1].contains("sql=invalid+SQL"));
+    }
+
     #[test]
     fn schedules_remote_sample_chunks_and_flushes() {
         let remote = RemoteWrite {
@@ -452,6 +580,8 @@ mod tests {
             sample_chunk_size: Some(2),
             flush_every_sample_chunks: 2,
             visibility_timeout_seconds: 30,
+            base_setup_sql: Vec::new(),
+            candidate_setup_sql: Vec::new(),
             prom_store: PromStore {
                 pending_rows_flush_interval: "1s".to_string(),
                 max_batch_rows: 1,

src/cmd/src/bin/query_regression_runner/remote/read_bench.rs

@@ -306,13 +306,25 @@ fn run_bench_command(command: Vec<String>, mut run: Value) -> Result<Value> {
 }
 
 fn parse_average_duration(stdout: &str) -> Option<f64> {
-    Regex::new(r"(?i)Average duration[^0-9]*([0-9.]+)\s*ms")
-        .ok()?
-        .captures(stdout)?
-        .get(1)?
-        .as_str()
-        .parse()
-        .ok()
+    let captures = Regex::new(
+        r"(?i)Average:\s+(?:\x1b\[[0-9;]*m)*\d+(?:\x1b\[[0-9;]*m)*\s+rows(?:,\s+(?:\x1b\[[0-9;]*m)*\d+(?:\x1b\[[0-9;]*m)*\s+record batches)?\s+in\s+([0-9]+(?:\.[0-9]+)?)\s*(ns|µs|us|ms|s)\s+over\s+\d+\s+iterations|Average duration[^0-9]*([0-9]+(?:\.[0-9]+)?)\s*ms",
+    )
+    .ok()?
+    .captures(stdout)?;
+    let (value, unit) = match (captures.get(1), captures.get(2), captures.get(3)) {
+        (Some(value), Some(unit), _) => (value.as_str(), unit.as_str()),
+        (_, _, Some(value)) => (value.as_str(), "ms"),
+        _ => return None,
+    };
+    let value = value.parse::<f64>().ok()?;
+    let milliseconds = match unit.to_ascii_lowercase().as_str() {
+        "ns" => value / 1_000_000.0,
+        "µs" | "us" => value / 1_000.0,
+        "ms" => value,
+        "s" => value * 1_000.0,
+        _ => return None,
+    };
+    milliseconds.is_finite().then_some(milliseconds)
 }
 
 fn bench_median(runs: &[Value]) -> Option<f64> {
@@ -328,12 +340,51 @@ mod tests {
     use super::*;
 
     #[test]
-    fn parses_average_duration_and_groups_scan_paths() {
-        assert_eq!(
-            parse_average_duration("work\nAverage duration: 12.5 ms\n"),
-            Some(12.5)
-        );
-        assert_eq!(parse_average_duration("completed"), None);
+    fn parses_average_durations() {
+        let cases = [
+            (
+                "parquetbench nanoseconds",
+                "ℹ Average: 42 rows, 2 record batches in 500ns over 4 iterations",
+                Some(0.0005),
+            ),
+            (
+                "scanbench colored microseconds",
+                "\x1b[1;32mSummary\x1b[0m Average: \x1b[36m12\x1b[0m rows in 250µs over 2 iterations",
+                Some(0.25),
+            ),
+            (
+                "scanbench seconds",
+                "Summary Average: 12 rows in 2.5s over 2 iterations",
+                Some(2500.0),
+            ),
+            (
+                "legacy milliseconds",
+                "Average duration: 12.5 ms",
+                Some(12.5),
+            ),
+            ("absent", "Benchmark completed!", None),
+            (
+                "malformed",
+                "Summary Average: 12 rows in unknown over 2 iterations",
+                None,
+            ),
+        ];
+
+        for (name, stdout, expected) in cases {
+            let actual = parse_average_duration(stdout);
+            match (actual, expected) {
+                (Some(actual), Some(expected)) => assert!(
+                    (actual - expected).abs() < f64::EPSILON,
+                    "{name}: expected {expected}, got {actual}"
+                ),
+                (None, None) => {}
+                (actual, expected) => panic!("{name}: expected {expected:?}, got {actual:?}"),
+            }
+        }
+    }
+
+    #[test]
+    fn groups_scan_paths() {
         let targets = vec![
             BenchTarget {
                 relative_path: "data/a.parquet".to_string(),

src/metric-engine/src/engine/bulk_insert.rs

@@ -242,18 +242,22 @@ mod tests {
     use common_error::ext::ErrorExt;
     use common_query::prelude::{greptime_timestamp, greptime_value};
     use common_recordbatch::RecordBatches;
+    use datafusion::parquet::basic::Encoding;
     use datatypes::arrow::array::{Float64Array, StringArray, TimestampMillisecondArray};
     use datatypes::arrow::datatypes::{DataType, Field, Schema as ArrowSchema, TimeUnit};
     use datatypes::arrow::record_batch::RecordBatch;
     use mito2::config::MitoConfig;
+    use mito2::sst::parquet::metadata::MetadataLoader;
+    use mito2::sst::parquet::reader::MetadataCacheMetrics;
     use store_api::metric_engine_consts::{
         METRIC_ENGINE_NAME, PHYSICAL_TABLE_METADATA_KEY, PRIMARY_KEY_ENCODING,
     };
+    use store_api::mito_engine_options::EXPERIMENTAL_SST_FLOAT_FIELD_ENCODING;
     use store_api::path_utils::table_dir;
     use store_api::region_engine::RegionEngine;
     use store_api::region_request::{
-        PathType, RegionBulkInsertsRequest, RegionCloseRequest, RegionOpenRequest,
-        RegionPutRequest, RegionRequest,
+        PathType, RegionBulkInsertsRequest, RegionCloseRequest, RegionFlushRequest,
+        RegionOpenRequest, RegionPutRequest, RegionRequest,
     };
     use store_api::storage::{RegionId, ScanRequest};
 
@@ -611,16 +615,68 @@ mod tests {
     #[tokio::test]
     async fn test_bulk_insert_sparse_encoding() {
         let env = TestEnv::new().await;
-        env.init_metric_region().await;
+        let physical_region_id = env.default_physical_region_id();
+        env.create_physical_region(
+            physical_region_id,
+            &TestEnv::default_table_dir(),
+            vec![(
+                EXPERIMENTAL_SST_FLOAT_FIELD_ENCODING.to_string(),
+                "byte_stream_split".to_string(),
+            )],
+        )
+        .await;
         let logical_region_id = env.default_logical_region_id();
+        env.create_logical_region(physical_region_id, logical_region_id)
+            .await;
 
-        let request = build_bulk_request(logical_region_id, build_logical_batch(0, 4), false);
+        let rows = 4;
+        let request = build_bulk_request(logical_region_id, build_logical_batch(0, rows), false);
         let response = env
             .metric()
             .handle_request(logical_region_id, request)
             .await
             .unwrap();
-        assert_eq!(response.affected_rows, 4);
+        assert_eq!(response.affected_rows, rows);
+
+        let data_region_id = crate::utils::to_data_region_id(physical_region_id);
+        env.mito()
+            .handle_request(
+                data_region_id,
+                RegionRequest::Flush(RegionFlushRequest::default()),
+            )
+            .await
+            .unwrap();
+
+        let region = env.mito().find_region(data_region_id).unwrap();
+        let entry = region
+            .manifest_sst_entries()
+            .await
+            .into_iter()
+            .find(|entry| entry.visible && entry.file_path.ends_with(".parquet"))
+            .unwrap();
+        let mut cache_metrics = MetadataCacheMetrics::default();
+        let footer = MetadataLoader::new(
+            region.access_layer().object_store().clone(),
+            &entry.file_path,
+            entry.file_size,
+        )
+        .load(&mut cache_metrics)
+        .await
+        .unwrap();
+        let field_column_index = footer
+            .file_metadata()
+            .schema_descr()
+            .columns()
+            .iter()
+            .position(|column| column.name() == greptime_value())
+            .unwrap();
+        assert!(!footer.row_groups().is_empty());
+        assert!(footer.row_groups().iter().all(|row_group| {
+            row_group
+                .column(field_column_index)
+                .encodings()
+                .any(|encoding| encoding == Encoding::BYTE_STREAM_SPLIT)
+        }));
 
         let stream = env
             .metric()
@@ -628,7 +684,7 @@ mod tests {
             .await
             .unwrap();
         let batches = RecordBatches::try_collect(stream).await.unwrap();
-        assert_eq!(batches.iter().map(|b| b.num_rows()).sum::<usize>(), 4);
+        assert_eq!(batches.iter().map(|b| b.num_rows()).sum::<usize>(), rows);
     }
 
     #[tokio::test]

src/mito2/src/compaction/compactor.rs

@@ -655,6 +655,7 @@ where
                 write_buffer_size: compaction_region.engine_config.sst_write_buffer_size,
                 max_file_size: picker_output.max_file_size,
                 row_group_size: compaction_region.region_options.row_group_size(),
+                float_field_encoding: compaction_region.region_options.float_field_encoding,
             };
             let merger = self.merger.clone();
             let compaction_region = compaction_region.clone();

src/mito2/src/compaction/window.rs

@@ -337,6 +337,7 @@ mod tests {
                 primary_key_encoding: None,
                 write_buffer_size: None,
                 preserve_row_sequence: false,
+                float_field_encoding: Default::default(),
             },
             compaction_time_window: None,
         }

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;
@@ -826,7 +827,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
@@ -842,6 +845,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());

src/mito2/src/flush.rs

@@ -451,6 +451,7 @@ impl RegionFlushTask {
 
         let mut write_opts = WriteOptions {
             write_buffer_size: self.engine_config.sst_write_buffer_size,
+            float_field_encoding: version.options.float_field_encoding,
             ..Default::default()
         };
         if let Some(row_group_size) = self.row_group_size {

src/mito2/src/memtable.rs

@@ -491,6 +491,7 @@ impl MemtableBuilderProvider {
                 BulkMemtableBuilder::new(self.write_buffer_manager.clone(), !dedup, merge_mode)
                     .with_config(config.clone())
                     .with_row_group_size(options.row_group_size())
+                    .with_float_field_encoding(options.float_field_encoding)
                     .with_compact_dispatcher(self.compact_dispatcher.clone()),
             ),
             Some(MemtableOptions::TimeSeries) => Arc::new(TimeSeriesMemtableBuilder::new(
@@ -515,6 +516,7 @@ impl MemtableBuilderProvider {
         )
         .with_config(self.default_bulk_memtable_config.clone())
         .with_row_group_size(options.row_group_size())
+        .with_float_field_encoding(options.float_field_encoding)
         .with_compact_dispatcher(self.compact_dispatcher.clone());
 
         if let Some(MemtableOptions::Bulk(config)) = &options.memtable {

src/mito2/src/memtable/bulk.rs

@@ -40,6 +40,7 @@ use rayon::prelude::*;
 use serde::{Deserialize, Serialize};
 use serde_with::{DisplayFromStr, serde_as};
 use store_api::metadata::RegionMetadataRef;
+use store_api::mito_engine_options::FloatFieldEncoding;
 use store_api::storage::{ColumnId, FileId, RegionId, SequenceRange};
 use tokio::sync::Semaphore;
 
@@ -435,6 +436,7 @@ pub struct BulkMemtable {
     merge_mode: MergeMode,
     /// Max number of rows in a parquet row group for encoded parts.
     row_group_size: usize,
+    float_field_encoding: FloatFieldEncoding,
 }
 
 impl std::fmt::Debug for BulkMemtable {
@@ -665,27 +667,7 @@ impl Memtable for BulkMemtable {
     }
 
     fn fork(&self, id: MemtableId, metadata: &RegionMetadataRef) -> MemtableRef {
-        Arc::new(Self {
-            id,
-            config: self.config.clone(),
-            parts: Arc::new(RwLock::new(BulkParts::default())),
-            metadata: metadata.clone(),
-            alloc_tracker: AllocTracker::new(self.alloc_tracker.write_buffer_manager()),
-            max_timestamp: AtomicI64::new(i64::MIN),
-            min_timestamp: AtomicI64::new(i64::MAX),
-            max_sequence: AtomicU64::new(0),
-            num_rows: AtomicUsize::new(0),
-            compactor: Arc::new(Mutex::new(MemtableCompactor::new(
-                metadata.region_id,
-                id,
-                self.config.clone(),
-                self.row_group_size,
-            ))),
-            compact_dispatcher: self.compact_dispatcher.clone(),
-            append_mode: self.append_mode,
-            merge_mode: self.merge_mode,
-            row_group_size: self.row_group_size,
-        })
+        self.fork_inner(id, metadata)
     }
 
     fn compact(&self, for_flush: bool) -> Result<()> {
@@ -714,6 +696,32 @@ impl Memtable for BulkMemtable {
 }
 
 impl BulkMemtable {
+    fn fork_inner(&self, id: MemtableId, metadata: &RegionMetadataRef) -> Arc<BulkMemtable> {
+        Arc::new(Self {
+            id,
+            config: self.config.clone(),
+            parts: Arc::new(RwLock::new(BulkParts::default())),
+            metadata: metadata.clone(),
+            alloc_tracker: AllocTracker::new(self.alloc_tracker.write_buffer_manager()),
+            max_timestamp: AtomicI64::new(i64::MIN),
+            min_timestamp: AtomicI64::new(i64::MAX),
+            max_sequence: AtomicU64::new(0),
+            num_rows: AtomicUsize::new(0),
+            compactor: Arc::new(Mutex::new(MemtableCompactor::new(
+                metadata.region_id,
+                id,
+                self.config.clone(),
+                self.row_group_size,
+                self.float_field_encoding,
+            ))),
+            compact_dispatcher: self.compact_dispatcher.clone(),
+            append_mode: self.append_mode,
+            merge_mode: self.merge_mode,
+            row_group_size: self.row_group_size,
+            float_field_encoding: self.float_field_encoding,
+        })
+    }
+
     /// Creates a new BulkMemtable with the default row group size.
     pub fn new(
         id: MemtableId,
@@ -747,6 +755,32 @@ impl BulkMemtable {
         append_mode: bool,
         merge_mode: MergeMode,
         row_group_size: usize,
+    ) -> Self {
+        Self::new_with_row_group_size_and_encoding(
+            id,
+            config,
+            metadata,
+            write_buffer_manager,
+            compact_dispatcher,
+            append_mode,
+            merge_mode,
+            row_group_size,
+            FloatFieldEncoding::default(),
+        )
+    }
+
+    /// Creates a new BulkMemtable with the given row group size and float encoding.
+    #[allow(clippy::too_many_arguments)]
+    fn new_with_row_group_size_and_encoding(
+        id: MemtableId,
+        config: BulkMemtableConfig,
+        metadata: RegionMetadataRef,
+        write_buffer_manager: Option<WriteBufferManagerRef>,
+        compact_dispatcher: Option<Arc<CompactDispatcher>>,
+        append_mode: bool,
+        merge_mode: MergeMode,
+        row_group_size: usize,
+        float_field_encoding: FloatFieldEncoding,
     ) -> Self {
         let config = config.sanitize();
         let region_id = metadata.region_id;
@@ -765,11 +799,13 @@ impl BulkMemtable {
                 id,
                 config,
                 row_group_size,
+                float_field_encoding,
             ))),
             compact_dispatcher,
             append_mode,
             merge_mode,
             row_group_size,
+            float_field_encoding,
         }
     }
 
@@ -1179,6 +1215,7 @@ struct MemtableCompactor {
     config: BulkMemtableConfig,
     /// Max number of rows in a parquet row group for encoded parts.
     row_group_size: usize,
+    float_field_encoding: FloatFieldEncoding,
 }
 
 impl MemtableCompactor {
@@ -1188,12 +1225,14 @@ impl MemtableCompactor {
         memtable_id: MemtableId,
         config: BulkMemtableConfig,
         row_group_size: usize,
+        float_field_encoding: FloatFieldEncoding,
     ) -> Self {
         Self {
             region_id,
             memtable_id,
             config,
             row_group_size,
+            float_field_encoding,
         }
     }
 
@@ -1233,6 +1272,7 @@ impl MemtableCompactor {
         let encode_row_threshold = self.config.encode_row_threshold;
         let encode_bytes_threshold = self.config.encode_bytes_threshold;
         let row_group_size = self.row_group_size;
+        let float_field_encoding = self.float_field_encoding;
 
         // Merge all groups in parallel
         let merged_parts = collected
@@ -1247,6 +1287,7 @@ impl MemtableCompactor {
                     encode_row_threshold,
                     encode_bytes_threshold,
                     row_group_size,
+                    float_field_encoding,
                 )
             })
             .collect::<Result<Vec<Option<MergedPart>>>>()?;
@@ -1282,6 +1323,7 @@ impl MemtableCompactor {
         encode_row_threshold: usize,
         encode_bytes_threshold: usize,
         row_group_size: usize,
+        float_field_encoding: FloatFieldEncoding,
     ) -> Result<Option<MergedPart>> {
         if parts_to_merge.is_empty() {
             return Ok(None);
@@ -1382,7 +1424,11 @@ impl MemtableCompactor {
         if estimated_total_rows > encode_row_threshold
             || estimated_total_bytes > encode_bytes_threshold
         {
-            let encoder = BulkPartEncoder::new(metadata.clone(), row_group_size)?;
+            let encoder = BulkPartEncoder::new_with_float_field_encoding(
+                metadata.clone(),
+                row_group_size,
+                float_field_encoding,
+            )?;
             let mut metrics = BulkPartEncodeMetrics::default();
             let encoded_part = encoder.encode_record_batch_iter(
                 boxed_iter,
@@ -1508,6 +1554,7 @@ pub struct BulkMemtableBuilder {
     merge_mode: MergeMode,
     /// Max number of rows in a parquet row group for encoded parts.
     row_group_size: usize,
+    float_field_encoding: FloatFieldEncoding,
 }
 
 impl Default for BulkMemtableBuilder {
@@ -1519,6 +1566,7 @@ impl Default for BulkMemtableBuilder {
             append_mode: false,
             merge_mode: MergeMode::default(),
             row_group_size: DEFAULT_ROW_GROUP_SIZE,
+            float_field_encoding: FloatFieldEncoding::default(),
         }
     }
 }
@@ -1550,6 +1598,11 @@ impl BulkMemtableBuilder {
         self
     }
 
+    pub(super) fn with_float_field_encoding(mut self, encoding: FloatFieldEncoding) -> Self {
+        self.float_field_encoding = encoding;
+        self
+    }
+
     /// Sets the compact dispatcher.
     pub fn with_compact_dispatcher(mut self, compact_dispatcher: Arc<CompactDispatcher>) -> Self {
         self.compact_dispatcher = Some(compact_dispatcher);
@@ -1564,7 +1617,7 @@ impl BulkMemtableBuilder {
 
 impl MemtableBuilder for BulkMemtableBuilder {
     fn build(&self, id: MemtableId, metadata: &RegionMetadataRef) -> MemtableRef {
-        Arc::new(BulkMemtable::new_with_row_group_size(
+        Arc::new(BulkMemtable::new_with_row_group_size_and_encoding(
             id,
             self.config.clone(),
             metadata.clone(),
@@ -1573,6 +1626,7 @@ impl MemtableBuilder for BulkMemtableBuilder {
             self.append_mode,
             self.merge_mode,
             self.row_group_size,
+            self.float_field_encoding,
         ))
     }
 
@@ -1597,6 +1651,7 @@ mod tests {
     use datatypes::schema::ColumnSchema;
     use datatypes::types::json_type::{JsonNativeType, JsonObjectType};
     use mito_codec::row_converter::build_primary_key_codec;
+    use parquet::basic::{Encoding, Type as PhysicalType};
     use serde_json::json;
     use store_api::metadata::{ColumnMetadata, RegionMetadataBuilder, RegionMetadataRef};
 
@@ -1829,6 +1884,7 @@ mod tests {
             usize::MAX,
             usize::MAX,
             DEFAULT_ROW_GROUP_SIZE,
+            FloatFieldEncoding::default(),
         )?
         .unwrap();
         let MergedPart::Multi(part) = merged else {
@@ -2143,7 +2199,7 @@ mod tests {
     #[test]
     fn test_bulk_memtable_fork() {
         let metadata = metadata_for_test();
-        let original_memtable = BulkMemtable::new(
+        let original_memtable = BulkMemtable::new_with_row_group_size_and_encoding(
             333,
             BulkMemtableConfig::default(),
             metadata.clone(),
@@ -2151,6 +2207,8 @@ mod tests {
             None,
             false,
             MergeMode::LastRow,
+            DEFAULT_ROW_GROUP_SIZE,
+            FloatFieldEncoding::ByteStreamSplit,
         );
 
         let bulk_part =
@@ -2159,9 +2217,25 @@ mod tests {
 
         original_memtable.write_bulk(bulk_part).unwrap();
 
-        let forked_memtable = original_memtable.fork(444, &metadata);
+        let forked_memtable = original_memtable.fork_inner(444, &metadata);
 
         assert_eq!(forked_memtable.id(), 444);
+        assert_eq!(
+            FloatFieldEncoding::ByteStreamSplit,
+            original_memtable.float_field_encoding
+        );
+        assert_eq!(
+            original_memtable.float_field_encoding,
+            forked_memtable.float_field_encoding
+        );
+        assert_eq!(
+            original_memtable.float_field_encoding,
+            forked_memtable
+                .compactor
+                .lock()
+                .unwrap()
+                .float_field_encoding
+        );
         assert!(forked_memtable.is_empty());
         assert_eq!(0, forked_memtable.stats().num_rows);
 
@@ -2780,6 +2854,42 @@ mod tests {
         assert!(!BulkParts::is_merge_candidate_by_size(true, 9, 8));
     }
 
+    #[test]
+    fn test_bulk_part_encoder_preserves_byte_stream_split() -> WhateverResult<()> {
+        let metadata = metadata_for_test();
+        let bulk_part = create_bulk_part_with_converter(
+            "byte_stream_split",
+            0,
+            vec![1000, 2000],
+            vec![Some(10.0), Some(20.0)],
+            100,
+        )?;
+        let encoder = BulkPartEncoder::new_with_float_field_encoding(
+            metadata,
+            DEFAULT_ROW_GROUP_SIZE,
+            FloatFieldEncoding::ByteStreamSplit,
+        )?;
+        let encoded_part = encoder.encode_part(&bulk_part)?.unwrap();
+        let field_column = encoded_part
+            .metadata()
+            .parquet_metadata
+            .row_groups()
+            .iter()
+            .flat_map(|row_group| row_group.columns())
+            .find(|column| {
+                column.column_path().string() == "v1"
+                    && column.column_type() == PhysicalType::DOUBLE
+            })
+            .expect("Float64 field should be present in encoded bulk part");
+
+        assert!(
+            field_column
+                .encodings()
+                .any(|encoding| encoding == Encoding::BYTE_STREAM_SPLIT)
+        );
+        Ok(())
+    }
+
     #[test]
     fn test_encoded_part_batch_size_uses_largest_uncompressed_row_group() {
         const NUM_ROWS: usize = 14;

src/mito2/src/memtable/bulk/part.rs

@@ -53,6 +53,7 @@ use smallvec::SmallVec;
 use snafu::{OptionExt, ResultExt};
 use store_api::codec::PrimaryKeyEncoding;
 use store_api::metadata::{RegionMetadata, RegionMetadataRef};
+use store_api::mito_engine_options::FloatFieldEncoding;
 use store_api::storage::consts::PRIMARY_KEY_COLUMN_NAME;
 use store_api::storage::{ColumnId, FileId, SequenceNumber, SequenceRange};
 
@@ -70,7 +71,7 @@ use crate::sst::SeriesEstimator;
 use crate::sst::index::IndexOutput;
 use crate::sst::parquet::flat_format::primary_key_column_index;
 use crate::sst::parquet::format::{PrimaryKeyArray, PrimaryKeyArrayBuilder};
-use crate::sst::parquet::{PARQUET_METADATA_KEY, SstInfo};
+use crate::sst::parquet::{PARQUET_METADATA_KEY, SstInfo, apply_float_field_encoding};
 
 const INIT_DICT_VALUE_CAPACITY: usize = 8;
 
@@ -1306,22 +1307,29 @@ pub struct BulkPartEncoder {
 
 impl BulkPartEncoder {
     pub fn new(metadata: RegionMetadataRef, row_group_size: usize) -> Result<BulkPartEncoder> {
+        Self::new_with_float_field_encoding(metadata, row_group_size, FloatFieldEncoding::default())
+    }
+
+    pub(super) fn new_with_float_field_encoding(
+        metadata: RegionMetadataRef,
+        row_group_size: usize,
+        float_field_encoding: FloatFieldEncoding,
+    ) -> Result<BulkPartEncoder> {
         // TODO(yingwen): Skip arrow schema if needed.
         let json = metadata.to_json().context(InvalidMetadataSnafu)?;
         let key_value_meta =
             parquet::file::metadata::KeyValue::new(PARQUET_METADATA_KEY.to_string(), json);
 
         // TODO(yingwen): Do we need compression?
-        let writer_props = Some(
-            WriterProperties::builder()
-                .set_key_value_metadata(Some(vec![key_value_meta]))
-                .set_write_batch_size(row_group_size)
-                .set_max_row_group_row_count(Some(row_group_size))
-                .set_compression(Compression::ZSTD(ZstdLevel::default()))
-                .set_column_index_truncate_length(None)
-                .set_statistics_truncate_length(None)
-                .build(),
-        );
+        let mut props = WriterProperties::builder()
+            .set_key_value_metadata(Some(vec![key_value_meta]))
+            .set_write_batch_size(row_group_size)
+            .set_max_row_group_row_count(Some(row_group_size))
+            .set_compression(Compression::ZSTD(ZstdLevel::default()))
+            .set_column_index_truncate_length(None)
+            .set_statistics_truncate_length(None);
+        props = apply_float_field_encoding(props, &metadata, float_field_encoding);
+        let writer_props = Some(props.build());
 
         Ok(Self {
             metadata,

src/mito2/src/region/options.rs

@@ -32,7 +32,9 @@ use store_api::codec::PrimaryKeyEncoding;
 use store_api::metric_engine_consts::{
     MEMTABLE_PARTITION_TREE_PRIMARY_KEY_ENCODING, PRIMARY_KEY_ENCODING,
 };
-use store_api::mito_engine_options::{COMPACTION_OVERRIDE, MAX_ROW_GROUP_ROW_COUNT_LIMIT};
+use store_api::mito_engine_options::{
+    COMPACTION_OVERRIDE, FloatFieldEncoding, MAX_ROW_GROUP_ROW_COUNT_LIMIT,
+};
 use store_api::storage::{ColumnId, RegionId};
 use strum::EnumString;
 
@@ -128,6 +130,9 @@ pub struct RegionOptions {
     /// sequence metadata.
     #[serde(default, skip_serializing_if = "is_false")]
     pub preserve_row_sequence: bool,
+    /// Encoding for direct floating-point field columns in Parquet SSTs.
+    #[serde(default)]
+    pub float_field_encoding: FloatFieldEncoding,
 }
 
 fn is_false(value: &bool) -> bool {
@@ -287,6 +292,8 @@ impl RegionOptions {
             sst_format = Some(FormatType::Flat);
         }
 
+        let float_field_encoding = options.float_field_encoding.unwrap_or_default();
+
         let compaction_override_flag = options_map
             .get(COMPACTION_OVERRIDE)
             .map(|v| matches!(v.to_lowercase().as_str(), "true" | "1"))
@@ -322,6 +329,7 @@ impl RegionOptions {
             primary_key_encoding,
             write_buffer_size: options.write_buffer_size,
             preserve_row_sequence: options.preserve_row_sequence,
+            float_field_encoding,
         };
         opts.validate()?;
 
@@ -469,6 +477,9 @@ struct RegionOptionsWithoutEnum {
     max_row_group_row_count: Option<usize>,
     #[serde_as(as = "DisplayFromStr")]
     preserve_row_sequence: bool,
+    #[serde(rename = "experimental_sst_float_field_encoding")]
+    #[serde_as(as = "NoneAsEmptyString")]
+    float_field_encoding: Option<FloatFieldEncoding>,
 }
 
 impl Default for RegionOptionsWithoutEnum {
@@ -485,6 +496,7 @@ impl Default for RegionOptionsWithoutEnum {
             sst_format: options.sst_format,
             max_row_group_row_count: options.max_row_group_row_count,
             preserve_row_sequence: options.preserve_row_sequence,
+            float_field_encoding: Some(options.float_field_encoding),
         }
     }
 }
@@ -628,7 +640,9 @@ mod tests {
     use common_error::ext::ErrorExt;
     use common_error::status_code::StatusCode;
     use common_wal::options::KafkaWalOptions;
-    use store_api::mito_engine_options::{SKIP_WAL_KEY, WRITE_BUFFER_SIZE_KEY};
+    use store_api::mito_engine_options::{
+        EXPERIMENTAL_SST_FLOAT_FIELD_ENCODING, SKIP_WAL_KEY, WRITE_BUFFER_SIZE_KEY,
+    };
 
     use super::*;
 
@@ -646,6 +660,26 @@ mod tests {
         assert_eq!(RegionOptions::default(), options);
     }
 
+    #[test]
+    fn test_float_field_encoding_defaults_and_parses() {
+        let options = RegionOptions::try_from_options(RegionId::new(0, 0), &make_map(&[])).unwrap();
+        assert_eq!(FloatFieldEncoding::Default, options.float_field_encoding);
+
+        let map = make_map(&[(EXPERIMENTAL_SST_FLOAT_FIELD_ENCODING, "default")]);
+        let options = RegionOptions::try_from_options(RegionId::new(0, 0), &map).unwrap();
+        assert_eq!(FloatFieldEncoding::Default, options.float_field_encoding);
+
+        let map = make_map(&[(EXPERIMENTAL_SST_FLOAT_FIELD_ENCODING, "byte_stream_split")]);
+        let options = RegionOptions::try_from_options(RegionId::new(0, 0), &map).unwrap();
+        assert_eq!(
+            FloatFieldEncoding::ByteStreamSplit,
+            options.float_field_encoding
+        );
+
+        let map = make_map(&[(EXPERIMENTAL_SST_FLOAT_FIELD_ENCODING, "invalid")]);
+        assert!(RegionOptions::try_from_options(RegionId::new(0, 0), &map).is_err());
+    }
+
     #[test]
     fn test_with_ttl() {
         let map = make_map(&[("ttl", "7d")]);
@@ -1103,6 +1137,7 @@ mod tests {
             primary_key_encoding: None,
             write_buffer_size: None,
             preserve_row_sequence: false,
+            float_field_encoding: FloatFieldEncoding::default(),
         };
         assert_eq!(expect, options);
     }
@@ -1161,6 +1196,7 @@ mod tests {
             primary_key_encoding: None,
             write_buffer_size: Some(ReadableSize::mb(128)),
             preserve_row_sequence: true,
+            float_field_encoding: FloatFieldEncoding::default(),
         };
         let region_options_json_str = serde_json::to_string(&options).unwrap();
         assert!(region_options_json_str.contains("preserve_row_sequence"));
@@ -1172,6 +1208,7 @@ mod tests {
         let got: RegionOptions = serde_json::from_str(old_region_options_json_str).unwrap();
         assert_eq!(None, got.write_buffer_size);
         assert!(!got.preserve_row_sequence);
+        assert_eq!(FloatFieldEncoding::Default, got.float_field_encoding);
         let CompactionOptions::Twcs(twcs) = got.compaction;
         assert_eq!(16, twcs.active_window_l1_merge_trigger);
         assert_eq!(8, twcs.inactive_window_l1_merge_trigger);
@@ -1238,6 +1275,7 @@ mod tests {
             primary_key_encoding: None,
             write_buffer_size: None,
             preserve_row_sequence: false,
+            float_field_encoding: FloatFieldEncoding::default(),
         };
         assert_eq!(options, got);
     }

src/mito2/src/sst/parquet.rs

@@ -17,9 +17,14 @@
 use std::collections::BTreeMap;
 use std::sync::Arc;
 
+use api::v1::SemanticType;
 use common_base::readable_size::ReadableSize;
 use datatypes::json::JsonSettings;
 use parquet::file::metadata::ParquetMetaData;
+use parquet::file::properties::WriterPropertiesBuilder;
+use parquet::schema::types::ColumnPath;
+use store_api::metadata::RegionMetadataRef;
+use store_api::mito_engine_options::FloatFieldEncoding;
 use store_api::storage::{ColumnId, FileId};
 
 use crate::sst::DEFAULT_WRITE_BUFFER_SIZE;
@@ -72,6 +77,27 @@ pub(crate) struct Json2TargetLayout {
 /// batching without changing the row group layout of newly written SSTs.
 pub const DEFAULT_ROW_GROUP_SIZE: usize = 100 * 1024;
 
+/// Applies the configured encoding to direct floating-point field columns.
+pub(crate) fn apply_float_field_encoding(
+    mut builder: WriterPropertiesBuilder,
+    metadata: &RegionMetadataRef,
+    encoding: FloatFieldEncoding,
+) -> WriterPropertiesBuilder {
+    if encoding == FloatFieldEncoding::ByteStreamSplit {
+        for column in &metadata.column_metadatas {
+            if column.semantic_type == SemanticType::Field
+                && column.column_schema.data_type.is_float()
+            {
+                let path = ColumnPath::new(vec![column.column_schema.name.clone()]);
+                builder = builder
+                    .set_column_encoding(path.clone(), parquet::basic::Encoding::BYTE_STREAM_SPLIT)
+                    .set_column_dictionary_enabled(path, false);
+            }
+        }
+    }
+    builder
+}
+
 /// Parquet write options.
 #[derive(Debug, Clone)]
 pub struct WriteOptions {
@@ -83,6 +109,8 @@ pub struct WriteOptions {
     /// Note: This is not a hard limit as we can only observe the file size when
     /// ArrowWrite writes to underlying writers.
     pub max_file_size: Option<usize>,
+    /// Encoding policy for direct floating-point field columns.
+    pub float_field_encoding: FloatFieldEncoding,
 }
 
 impl Default for WriteOptions {
@@ -91,6 +119,7 @@ impl Default for WriteOptions {
             write_buffer_size: DEFAULT_WRITE_BUFFER_SIZE,
             row_group_size: DEFAULT_ROW_GROUP_SIZE,
             max_file_size: None,
+            float_field_encoding: FloatFieldEncoding::default(),
         }
     }
 }
@@ -125,6 +154,7 @@ mod tests {
     use std::sync::Arc;
 
     use api::v1::{OpType, SemanticType};
+    use bytes::Bytes;
     use common_function::function::FunctionRef;
     use common_function::function_factory::ScalarFunctionFactory;
     use common_function::scalars::matches::MatchesFunction;
@@ -135,20 +165,24 @@ mod tests {
     use datafusion_expr::{BinaryExpr, Expr, Literal, Operator, col, lit};
     use datatypes::arrow;
     use datatypes::arrow::array::{
-        ArrayRef, AsArray, BinaryDictionaryBuilder, RecordBatch, StringArray,
-        StringDictionaryBuilder, TimestampMillisecondArray, UInt8Array, UInt64Array,
+        Array, ArrayRef, AsArray, BinaryDictionaryBuilder, Float32Array, Float64Array, Int32Array,
+        RecordBatch, StringArray, StringDictionaryBuilder, TimestampMillisecondArray, UInt8Array,
+        UInt64Array,
     };
-    use datatypes::arrow::datatypes::{DataType, Field, Schema, UInt32Type};
+    use datatypes::arrow::datatypes::{DataType, Field, Schema, TimeUnit, UInt32Type};
     use datatypes::arrow::util::pretty::pretty_format_batches;
     use datatypes::prelude::ConcreteDataType;
     use datatypes::schema::{FulltextAnalyzer, FulltextBackend, FulltextOptions};
     use object_store::ObjectStore;
-    use parquet::arrow::AsyncArrowWriter;
+    use parquet::arrow::arrow_reader::ParquetRecordBatchReaderBuilder;
+    use parquet::arrow::{ArrowWriter, AsyncArrowWriter};
     use parquet::basic::{Compression, Encoding, ZstdLevel};
     use parquet::file::metadata::{KeyValue, PageIndexPolicy};
     use parquet::file::properties::WriterProperties;
+    use parquet::schema::types::ColumnPath;
     use store_api::codec::PrimaryKeyEncoding;
     use store_api::metadata::{ColumnMetadata, RegionMetadata, RegionMetadataBuilder};
+    use store_api::mito_engine_options::FloatFieldEncoding;
     use store_api::region_request::PathType;
     use store_api::storage::{ColumnSchema, RegionId};
     use table::predicate::Predicate;
@@ -187,6 +221,176 @@ mod tests {
     const FILE_DIR: &str = "/";
     const REGION_ID: RegionId = RegionId::new(0, 0);
 
+    #[test]
+    fn test_float_field_encoding_properties_and_roundtrip() {
+        let mut metadata_builder = RegionMetadataBuilder::new(REGION_ID);
+        metadata_builder
+            .push_column_metadata(ColumnMetadata {
+                column_schema: ColumnSchema::new("f32", ConcreteDataType::float32_datatype(), true),
+                semantic_type: SemanticType::Field,
+                column_id: 0,
+            })
+            .push_column_metadata(ColumnMetadata {
+                column_schema: ColumnSchema::new("f64", ConcreteDataType::float64_datatype(), true),
+                semantic_type: SemanticType::Field,
+                column_id: 1,
+            })
+            .push_column_metadata(ColumnMetadata {
+                column_schema: ColumnSchema::new("tag", ConcreteDataType::float32_datatype(), true),
+                semantic_type: SemanticType::Tag,
+                column_id: 2,
+            })
+            .push_column_metadata(ColumnMetadata {
+                column_schema: ColumnSchema::new("i32", ConcreteDataType::int32_datatype(), true),
+                semantic_type: SemanticType::Field,
+                column_id: 3,
+            })
+            .push_column_metadata(ColumnMetadata {
+                column_schema: ColumnSchema::new(
+                    "ts",
+                    ConcreteDataType::timestamp_millisecond_datatype(),
+                    false,
+                ),
+                semantic_type: SemanticType::Timestamp,
+                column_id: 4,
+            });
+        metadata_builder.primary_key(vec![2]);
+        let metadata = Arc::new(metadata_builder.build().unwrap());
+
+        let f32_values = [
+            Some(1.5_f32),
+            Some(2.0),
+            Some(0.0),
+            Some(-0.0),
+            Some(f32::INFINITY),
+            Some(f32::from_bits(0x7fc0_1234)),
+            None,
+        ];
+        let f64_values = [
+            Some(1.5_f64),
+            Some(2.0),
+            Some(0.0),
+            Some(-0.0),
+            Some(f64::NEG_INFINITY),
+            Some(f64::from_bits(0x7ff8_0000_0000_1234)),
+            None,
+        ];
+        let schema = Arc::new(Schema::new(vec![
+            Field::new("f32", DataType::Float32, true),
+            Field::new("f64", DataType::Float64, true),
+            Field::new("tag", DataType::Float32, true),
+            Field::new("i32", DataType::Int32, true),
+            Field::new(
+                "ts",
+                DataType::Timestamp(TimeUnit::Millisecond, None),
+                false,
+            ),
+        ]));
+        let batch = RecordBatch::try_new(
+            schema.clone(),
+            vec![
+                Arc::new(Float32Array::from(f32_values.to_vec())) as ArrayRef,
+                Arc::new(Float64Array::from(f64_values.to_vec())) as ArrayRef,
+                Arc::new(Float32Array::from(vec![
+                    Some(1.0),
+                    None,
+                    Some(-0.0),
+                    Some(2.0),
+                    Some(3.0),
+                    Some(4.0),
+                    None,
+                ])),
+                Arc::new(Int32Array::from(vec![
+                    Some(1),
+                    None,
+                    Some(3),
+                    Some(4),
+                    Some(5),
+                    Some(6),
+                    None,
+                ])),
+                Arc::new(TimestampMillisecondArray::from_iter_values(0..7)),
+            ],
+        )
+        .unwrap();
+
+        let path = |name: &str| ColumnPath::new(vec![name.to_string()]);
+        let bss = apply_float_field_encoding(
+            WriterProperties::builder(),
+            &metadata,
+            FloatFieldEncoding::ByteStreamSplit,
+        );
+        assert_eq!(
+            Some(Encoding::BYTE_STREAM_SPLIT),
+            bss.clone().build().encoding(&path("f32"))
+        );
+        assert_eq!(
+            Some(Encoding::BYTE_STREAM_SPLIT),
+            bss.clone().build().encoding(&path("f64"))
+        );
+        assert!(!bss.clone().build().dictionary_enabled(&path("f32")));
+        assert!(!bss.clone().build().dictionary_enabled(&path("f64")));
+        assert_eq!(None, bss.clone().build().encoding(&path("i32")));
+        assert!(bss.clone().build().dictionary_enabled(&path("tag")));
+
+        let default = apply_float_field_encoding(
+            WriterProperties::builder().set_encoding(Encoding::PLAIN),
+            &metadata,
+            FloatFieldEncoding::Default,
+        )
+        .build();
+        assert_eq!(Some(Encoding::PLAIN), default.encoding(&path("f32")));
+        assert!(default.dictionary_enabled(&path("f32")));
+        assert!(default.dictionary_enabled(&path("f64")));
+
+        let mut bytes = Vec::new();
+        let mut writer = ArrowWriter::try_new(&mut bytes, schema, Some(bss.build())).unwrap();
+        writer.write(&batch).unwrap();
+        let footer = writer.finish().unwrap();
+        drop(writer);
+        for name in ["f32", "f64"] {
+            let column = footer.row_groups()[0]
+                .columns()
+                .iter()
+                .find(|column| column.column_path().string() == name)
+                .unwrap();
+            assert!(
+                column
+                    .encodings()
+                    .any(|encoding| encoding == Encoding::BYTE_STREAM_SPLIT)
+            );
+        }
+        let mut reader = ParquetRecordBatchReaderBuilder::try_new(Bytes::from(bytes))
+            .unwrap()
+            .build()
+            .unwrap();
+        let actual = reader.next().unwrap().unwrap();
+        let actual_f32 = actual
+            .column(0)
+            .as_any()
+            .downcast_ref::<Float32Array>()
+            .unwrap();
+        let actual_f64 = actual
+            .column(1)
+            .as_any()
+            .downcast_ref::<Float64Array>()
+            .unwrap();
+        for (index, value) in f32_values.into_iter().enumerate() {
+            assert_eq!(value.is_none(), actual_f32.is_null(index));
+            if let Some(value) = value {
+                assert_eq!(value.to_bits(), actual_f32.value(index).to_bits());
+            }
+        }
+        for (index, value) in f64_values.into_iter().enumerate() {
+            assert_eq!(value.is_none(), actual_f64.is_null(index));
+            if let Some(value) = value {
+                assert_eq!(value.to_bits(), actual_f64.value(index).to_bits());
+            }
+        }
+        assert!(actual.column(2).is_null(1));
+        assert!(actual.column(3).is_null(1));
+    }
+
     #[derive(Clone)]
     struct FixedPathProvider {
         region_file_id: RegionFileId,

src/mito2/src/sst/parquet/writer.rs

@@ -38,7 +38,7 @@ use object_store::{FuturesAsyncWriter, ObjectStore};
 use parquet::arrow::AsyncArrowWriter;
 use parquet::basic::{Compression, Encoding, ZstdLevel};
 use parquet::file::metadata::KeyValue;
-use parquet::file::properties::{WriterProperties, WriterPropertiesBuilder};
+use parquet::file::properties::WriterProperties;
 use parquet::schema::types::ColumnPath;
 use smallvec::smallvec;
 use snafu::{OptionExt, ResultExt};
@@ -61,7 +61,9 @@ use crate::sst::parquet::flat_format::{
     FlatWriteFormat, primary_key_column_index, time_index_column_index,
 };
 use crate::sst::parquet::format::{PrimaryKeyArray, PrimaryKeyWriteFormat};
-use crate::sst::parquet::{PARQUET_METADATA_KEY, SstInfo, WriteOptions};
+use crate::sst::parquet::{
+    PARQUET_METADATA_KEY, SstInfo, WriteOptions, apply_float_field_encoding,
+};
 use crate::sst::{
     DEFAULT_WRITE_BUFFER_SIZE, DEFAULT_WRITE_CONCURRENCY, FlatSchemaOptions, SeriesEstimator,
     maybe_wrap_schema,
@@ -473,29 +475,6 @@ where
         Ok(results)
     }
 
-    /// Customizes per-column config according to schema and maybe column cardinality.
-    fn customize_column_config(
-        builder: WriterPropertiesBuilder,
-        region_metadata: &RegionMetadataRef,
-    ) -> WriterPropertiesBuilder {
-        let ts_col = ColumnPath::new(vec![
-            region_metadata
-                .time_index_column()
-                .column_schema
-                .name
-                .clone(),
-        ]);
-        let seq_col = ColumnPath::new(vec![SEQUENCE_COLUMN_NAME.to_string()]);
-        let op_type_col = ColumnPath::new(vec![OP_TYPE_COLUMN_NAME.to_string()]);
-
-        builder
-            .set_column_encoding(seq_col.clone(), Encoding::DELTA_BINARY_PACKED)
-            .set_column_dictionary_enabled(seq_col, false)
-            .set_column_encoding(ts_col.clone(), Encoding::DELTA_BINARY_PACKED)
-            .set_column_dictionary_enabled(ts_col, false)
-            .set_column_compression(op_type_col, Compression::UNCOMPRESSED)
-    }
-
     async fn append_flat_batch(
         &mut self,
         batch: &RecordBatch,
@@ -564,8 +543,22 @@ where
                 .set_max_row_group_row_count(Some(opts.row_group_size))
                 .set_column_index_truncate_length(None)
                 .set_statistics_truncate_length(None);
-
-            let props_builder = Self::customize_column_config(props_builder, &self.metadata);
+            let ts_col = ColumnPath::new(vec![
+                self.metadata.time_index_column().column_schema.name.clone(),
+            ]);
+            let seq_col = ColumnPath::new(vec![SEQUENCE_COLUMN_NAME.to_string()]);
+            let op_type_col = ColumnPath::new(vec![OP_TYPE_COLUMN_NAME.to_string()]);
+            let props_builder = props_builder
+                .set_column_encoding(seq_col.clone(), Encoding::DELTA_BINARY_PACKED)
+                .set_column_dictionary_enabled(seq_col, false)
+                .set_column_encoding(ts_col.clone(), Encoding::DELTA_BINARY_PACKED)
+                .set_column_dictionary_enabled(ts_col, false)
+                .set_column_compression(op_type_col, Compression::UNCOMPRESSED);
+            let props_builder = apply_float_field_encoding(
+                props_builder,
+                &self.metadata,
+                opts.float_field_encoding,
+            );
             let writer_props = props_builder.build();
 
             let sst_file_path = self.path_provider.build_sst_file_path(RegionFileId::new(

src/store-api/src/mito_engine_options.rs

@@ -88,6 +88,31 @@ pub const MAX_ROW_GROUP_ROW_COUNT: &str = "max_row_group_row_count";
 pub const MAX_ROW_GROUP_ROW_COUNT_LIMIT: usize = 10 * 1024 * 1024;
 /// Option key for preserving per-row sequence numbers through flush and compaction.
 pub const PRESERVE_ROW_SEQUENCE: &str = "preserve_row_sequence";
+/// Option key for experimental Parquet float field encoding.
+pub const EXPERIMENTAL_SST_FLOAT_FIELD_ENCODING: &str = "experimental_sst_float_field_encoding";
+
+/// Encoding policy for direct floating-point field columns in Parquet SSTs.
+#[derive(
+    Debug,
+    Default,
+    Clone,
+    Copy,
+    PartialEq,
+    Eq,
+    Hash,
+    serde::Serialize,
+    serde::Deserialize,
+    strum::EnumString,
+)]
+#[serde(rename_all = "snake_case")]
+#[strum(serialize_all = "snake_case")]
+pub enum FloatFieldEncoding {
+    /// The existing Parquet writer behavior.
+    #[default]
+    Default,
+    /// Parquet byte-stream-split encoding.
+    ByteStreamSplit,
+}
 // Note: Adding new options here should also check if this option should be removed in [metric_engine::engine::create::region_options_for_metadata_region].
 
 /// Conflicting values supplied through the legacy and canonical TWCS trigger options.
@@ -155,6 +180,7 @@ pub fn is_mito_engine_option_key(key: &str) -> bool {
         SST_FORMAT_KEY,
         MAX_ROW_GROUP_ROW_COUNT,
         PRESERVE_ROW_SEQUENCE,
+        EXPERIMENTAL_SST_FLOAT_FIELD_ENCODING,
     ]
     .contains(&key)
 }
@@ -217,9 +243,29 @@ mod tests {
         assert!(is_mito_engine_option_key("append_mode"));
         assert!(is_mito_engine_option_key("max_row_group_row_count"));
         assert!(is_mito_engine_option_key("preserve_row_sequence"));
+        assert!(is_mito_engine_option_key(
+            EXPERIMENTAL_SST_FLOAT_FIELD_ENCODING
+        ));
         assert!(!is_mito_engine_option_key("foo"));
     }
 
+    #[test]
+    fn test_float_field_encoding_values() {
+        assert_eq!(
+            "default".parse::<FloatFieldEncoding>(),
+            Ok(FloatFieldEncoding::Default)
+        );
+        assert_eq!(
+            "byte_stream_split".parse::<FloatFieldEncoding>(),
+            Ok(FloatFieldEncoding::ByteStreamSplit)
+        );
+        assert_eq!(
+            serde_json::from_str::<FloatFieldEncoding>("\"byte_stream_split\"").unwrap(),
+            FloatFieldEncoding::ByteStreamSplit
+        );
+        assert!(serde_json::from_str::<FloatFieldEncoding>("\"unknown\"").is_err());
+    }
+
     #[test]
     fn test_normalize_twcs_trigger_aliases_to_legacy_key() {
         let expected = HashMap::from([(TWCS_TRIGGER_FILE_NUM.to_string(), "4".to_string())]);

src/table/src/requests.rs

@@ -36,10 +36,10 @@ use store_api::metric_engine_consts::{
     LOGICAL_TABLE_METADATA_KEY, PHYSICAL_TABLE_METADATA_KEY, is_metric_engine_option_key,
 };
 use store_api::mito_engine_options::{
-    APPEND_MODE_KEY, COMPACTION_TYPE, MEMTABLE_BULK_ENCODE_BYTES_THRESHOLD,
-    MEMTABLE_BULK_ENCODE_ROW_THRESHOLD, MEMTABLE_BULK_MAX_MERGE_GROUPS,
-    MEMTABLE_BULK_MERGE_THRESHOLD, MEMTABLE_TYPE, MERGE_MODE_KEY, SST_FORMAT_KEY,
-    TWCS_ACTIVE_WINDOW_L1_MERGE_TRIGGER, TWCS_ACTIVE_WINDOW_TRIGGER_FILE_NUM,
+    APPEND_MODE_KEY, COMPACTION_TYPE, EXPERIMENTAL_SST_FLOAT_FIELD_ENCODING, FloatFieldEncoding,
+    MEMTABLE_BULK_ENCODE_BYTES_THRESHOLD, MEMTABLE_BULK_ENCODE_ROW_THRESHOLD,
+    MEMTABLE_BULK_MAX_MERGE_GROUPS, MEMTABLE_BULK_MERGE_THRESHOLD, MEMTABLE_TYPE, MERGE_MODE_KEY,
+    SST_FORMAT_KEY, TWCS_ACTIVE_WINDOW_L1_MERGE_TRIGGER, TWCS_ACTIVE_WINDOW_TRIGGER_FILE_NUM,
     TWCS_FALLBACK_TO_LOCAL, TWCS_INACTIVE_WINDOW_L1_MERGE_TRIGGER,
     TWCS_INACTIVE_WINDOW_TRIGGER_FILE_NUM, TWCS_MAX_OUTPUT_FILE_SIZE, TWCS_TIME_WINDOW,
     TWCS_TRIGGER_FILE_NUM, is_mito_engine_option_key, normalize_twcs_trigger_options,
@@ -256,6 +256,16 @@ impl TableOptions {
             options.ttl = Some(ttl_value);
         }
 
+        if let Some(encoding) = kvs.get(EXPERIMENTAL_SST_FLOAT_FIELD_ENCODING) {
+            encoding.parse::<FloatFieldEncoding>().map_err(|_| {
+                ParseTableOptionSnafu {
+                    key: EXPERIMENTAL_SST_FLOAT_FIELD_ENCODING,
+                    value: encoding,
+                }
+                .build()
+            })?;
+        }
+
         if let Some(skip_wal) = kvs.get(SKIP_WAL_KEY) {
             options.skip_wal = skip_wal.parse().map_err(|_| {
                 ParseTableOptionSnafu {
@@ -885,6 +895,7 @@ mod tests {
         assert!(validate_table_option(WRITE_BUFFER_SIZE_KEY));
         assert!(validate_table_option(STORAGE_KEY));
         assert!(validate_table_option(MEMTABLE_BULK_MERGE_THRESHOLD));
+        assert!(validate_table_option(EXPERIMENTAL_SST_FLOAT_FIELD_ENCODING));
         assert!(validate_table_option(REPARTITION_COLUMN_HINT_KEY));
         assert!(validate_table_option(REPARTITION_PARTITION_NUM_HINT_KEY));
         assert_eq!(AnnotationFamily::of_key("repartition.unknown.hint"), None);
@@ -923,6 +934,25 @@ mod tests {
         assert!(!validate_database_option("foo"));
     }
 
+    #[test]
+    fn test_parse_float_field_encoding_option() {
+        let options = TableOptions::try_from_iter([(
+            EXPERIMENTAL_SST_FLOAT_FIELD_ENCODING,
+            "byte_stream_split",
+        )])
+        .unwrap();
+        assert_eq!(
+            options
+                .extra_options
+                .get(EXPERIMENTAL_SST_FLOAT_FIELD_ENCODING),
+            Some(&"byte_stream_split".to_string())
+        );
+        assert!(
+            TableOptions::try_from_iter([(EXPERIMENTAL_SST_FLOAT_FIELD_ENCODING, "invalid",)])
+                .is_err()
+        );
+    }
+
     #[test]
     fn test_database_trigger_value_boundaries() {
         let maximum = usize::MAX.to_string();

tests/compatibility/cases/downgrade_compatibility/case.toml

@@ -1,8 +1,8 @@
 name = "downgrade_compatibility"
-reason = "Verify v1.1.4 can reopen a table whose region WAL options were written by the current binary, and can read all rows of an append-only table flushed and compacted with preserve_row_sequence enabled."
+reason = "Verify v1.1.4 can reopen tables whose region WAL options or byte-stream-split (BSS) float SSTs were written by the current binary, and can read all rows of an append-only table flushed and compacted with preserve_row_sequence enabled."
 introduced_by = "fix: preserve legacy region WAL options format; preserve_row_sequence"
 topologies = ["distributed", "standalone"]
 from_range = [">=v1.2.0"]
 to_range = ["=v1.1.4"]
-features = ["table", "wal", "downgrade", "append", "preserve_row_sequence"]
+features = ["table", "wal", "downgrade", "append", "preserve_row_sequence", "sst", "float", "byte_stream_split"]
 owner = "metasrv"

tests/compatibility/cases/downgrade_compatibility/setup.sql

@@ -32,3 +32,18 @@ INSERT INTO t_preserve_sequence_downgrade VALUES
 ADMIN FLUSH_TABLE('t_preserve_sequence_downgrade');
 
 ADMIN COMPACT_TABLE('t_preserve_sequence_downgrade');
+
+CREATE TABLE t_sst_float_field_encoding_downgrade(
+    ts TIMESTAMP TIME INDEX,
+    host STRING PRIMARY KEY,
+    f FLOAT,
+    d DOUBLE
+) ENGINE=mito
+WITH('experimental_sst_float_field_encoding'='byte_stream_split');
+
+INSERT INTO t_sst_float_field_encoding_downgrade VALUES
+('2024-02-10 00:00:00+0000', 'host_a', 1.25, 10.5),
+('2024-02-10 00:01:00+0000', 'host_b', -2.5, 20.25),
+('2024-02-10 00:02:00+0000', 'host_c', NULL, NULL);
+
+ADMIN FLUSH_TABLE('t_sst_float_field_encoding_downgrade');

tests/compatibility/cases/downgrade_compatibility/verify.result

@@ -17,3 +17,15 @@ SELECT ts, host, val FROM t_preserve_sequence_downgrade ORDER BY ts, host;
 | 2024-02-09T00:02:00 | host_a | 3   |
 | 2024-02-09T00:03:00 | host_c | 4   |
 +---------------------+--------+-----+
+
+SELECT ts, host, f, d
+FROM t_sst_float_field_encoding_downgrade
+ORDER BY ts, host;
+
++---------------------+--------+------+-------+
+| ts                  | host   | f    | d     |
++---------------------+--------+------+-------+
+| 2024-02-10T00:00:00 | host_a | 1.25 | 10.5  |
+| 2024-02-10T00:01:00 | host_b | -2.5 | 20.25 |
+| 2024-02-10T00:02:00 | host_c |      |       |
++---------------------+--------+------+-------+

tests/compatibility/cases/downgrade_compatibility/verify.sql

@@ -1,3 +1,7 @@
 SELECT ts, host, val FROM t_downgrade_compatibility ORDER BY ts, host;
 
 SELECT ts, host, val FROM t_preserve_sequence_downgrade ORDER BY ts, host;
+
+SELECT ts, host, f, d
+FROM t_sst_float_field_encoding_downgrade
+ORDER BY ts, host;

tests/compatibility/cases/sst_float_field_encoding_upgrade/case.toml

@@ -0,0 +1,8 @@
+name = "sst_float_field_encoding_upgrade"
+reason = "Verify newer binaries can read default-encoded Mito SSTs containing FLOAT and DOUBLE fields written by older binaries."
+introduced_by = "experimental_sst_float_field_encoding"
+topologies = ["distributed", "standalone"]
+from_range = ["*"]
+to_range = ["*"]
+features = ["table", "sst", "float", "upgrade"]
+owner = "storage"

tests/compatibility/cases/sst_float_field_encoding_upgrade/setup.sql

@@ -0,0 +1,13 @@
+CREATE TABLE t_sst_float_field_encoding_upgrade(
+    ts TIMESTAMP TIME INDEX,
+    host STRING PRIMARY KEY,
+    f FLOAT,
+    d DOUBLE
+) ENGINE=mito;
+
+INSERT INTO t_sst_float_field_encoding_upgrade VALUES
+('2024-02-10 00:00:00+0000', 'host_a', 1.25, 10.5),
+('2024-02-10 00:01:00+0000', 'host_b', -2.5, 20.25),
+('2024-02-10 00:02:00+0000', 'host_c', NULL, NULL);
+
+ADMIN FLUSH_TABLE('t_sst_float_field_encoding_upgrade');

tests/compatibility/cases/sst_float_field_encoding_upgrade/verify.result

@@ -0,0 +1,11 @@
+SELECT ts, host, f, d
+FROM t_sst_float_field_encoding_upgrade
+ORDER BY ts, host;
+
++---------------------+--------+------+-------+
+| ts                  | host   | f    | d     |
++---------------------+--------+------+-------+
+| 2024-02-10T00:00:00 | host_a | 1.25 | 10.5  |
+| 2024-02-10T00:01:00 | host_b | -2.5 | 20.25 |
+| 2024-02-10T00:02:00 | host_c |      |       |
++---------------------+--------+------+-------+

tests/compatibility/cases/sst_float_field_encoding_upgrade/verify.sql

@@ -0,0 +1,3 @@
+SELECT ts, host, f, d
+FROM t_sst_float_field_encoding_upgrade
+ORDER BY ts, host;

tests/perf/README.md

@@ -86,15 +86,19 @@ sample values without changing label cardinality or the runner lifecycle:
 [scenario.remote_write.value]
 pattern = "quantized_signal" # linear, constant, modulo, unique, seeded_random,
                               # run_length, quantized_signal,
-                              # signal_with_sporadic_stalls, mixed_signal_repeated
+                              # signal_with_sporadic_stalls, mixed_signal_repeated,
+                              # bounded_mixed
 base = 0.0
 step = 0.125
-cardinality = 4096           # buckets for modulo/seeded_random/quantized_signal/run_length
-seed = 12345                 # deterministic seeded_random input
-run_length = 8               # adjacent samples per bucket for run_length/quantized_signal
+cardinality = 4096           # buckets for modulo/seeded_random/quantized_signal/run_length;
+                              # baseline buckets for bounded_mixed
+seed = 12345                 # deterministic seeded_random input and bounded_mixed series hash
+run_length = 8               # adjacent samples per bucket for run_length/quantized_signal;
+                              # interpolation interval for bounded_mixed
 stall_every = 100            # interval for signal_with_sporadic_stalls
 stall_length = 16            # held samples inside each stall interval
-mixed_every = 5              # every Nth sample becomes the repeated base value
+mixed_every = 5              # every Nth sample becomes the repeated base value;
+                              # one fractional bounded_mixed series per N series
 ```
 
 The default `linear` pattern preserves the helper's historical formula. Use
@@ -103,12 +107,29 @@ data shapes, `run_length` for run-heavy low-cardinality series, `quantized_signa
 for signal-like values collapsed into a finite bucket set, `signal_with_sporadic_stalls`
 for mostly continuous signals with periodic flat spots, and `mixed_signal_repeated`
 for signal-plus-periodic-default mixtures. `unique` or high-cardinality buckets
-still work for broad sample-value distributions. This is a generic sample-value
-control for query/ingestion cases; it does not inspect or assert storage
-encoding, Parquet footers, or storage policy choices. For chunked remote-write
-ingestion, the runner passes the sample offset and total sample count to the
-helper so non-linear value patterns use a stable global/per-series ordinal across
-chunks.
+still work for broad sample-value distributions.
+
+`bounded_mixed` is synthetic rather than an empirical workload. For series `s`,
+`h = splitmix64(s ^ seed)` selects a span from `[10, 1_000, 100_000, 10_000_000]`
+using its top two bits, multiplied by `1 + ((h >> 32) % 10)`. Its baseline is
+`(h % max(cardinality, 1)) * span / 10`. Ranges can overlap. At local sample
+`l = sample_offset + sample_idx`, it linearly interpolates hash-derived anchors
+in `[-span + 1, span - 1]`, changing anchors every `max(run_length, 1)` samples.
+It applies `base + step * signal`, then rounds to an integer. Every
+`max(mixed_every, 1)`-th series adds a hash-derived fraction from `1/1000` to
+`999/1000`. The 95/5 finite, nonintegral guarantee applies to the case's bounded
+parameters, not arbitrary huge or nonfinite base/step inputs. With `base = 0`
+and `step = 1`, output lies within `baseline ± span`, allowing one unit for
+rounding and the fractional addition. The explicit case uses 1,000 equal-length
+series and `mixed_every = 20`, giving exactly 95% integer-valued and 5%
+fractional Float64 samples. Its ranges and temporal shape are synthetic design
+parameters, not measured properties from the survey.
+
+This is a generic sample-value control for query/ingestion cases; it does not
+inspect or assert storage encoding, Parquet footers, or storage policy choices.
+For chunked remote-write ingestion, the runner passes the sample offset and total
+sample count to the helper so non-linear value patterns use a stable global/per-series
+ordinal across chunks.
 
 Case schema, value distribution defaults, storage defaults, and read-bench
 defaults are owned by Rust. The outer CI driver calls
@@ -162,12 +183,15 @@ measures region scan cost, and query measurements still exercise the SQL/TQL
 frontend path. Treat all performance conclusions as release-only; debug builds
 are suitable only for command wiring and correctness checks.
 
-`prepare-remote` creates the configured database if needed. The outer driver writes a per-target
-frontend config enabling `[prom_store]` with metric engine storage and a non-zero
-`pending_rows_flush_interval`, and validates that the logical metric table reaches
-`series_count * samples_per_series` rows before trusting the query measurements.
-Use `--fixture-generator /path/to/query_perf_fixture` to provide the Rust helper
-to the outer driver.
+`prepare-remote` creates the configured database if needed. A remote-write case
+can provide `base_setup_sql` and `candidate_setup_sql` lists; each target runs its
+own complete statements in order after database creation and before ingestion.
+The outer driver writes a per-target frontend config enabling `[prom_store]` with
+metric engine storage and a non-zero `pending_rows_flush_interval`, and validates
+that the logical metric table reaches `series_count * samples_per_series` rows
+before trusting the query measurements. Use
+`--fixture-generator /path/to/query_perf_fixture` to provide the Rust helper to
+the outer driver.
 
 Large manual remote-write cases can set `sample_chunk_size` to split ingestion by
 time. For each chunk, `prepare-remote` invokes `query_perf_fixture prom-remote-write` with the
@@ -204,7 +228,9 @@ case for issue #7913. It writes 8192 series × 20160 samples through remote-writ
 in 1440-sample daily time chunks, flushing after each chunk before running 1d/7d/14d
 TQL selectors. It is not included in the default `all` case set because ingestion
 cost dominates routine CI validation. Commenting `/query-regression heavy` runs
-only this case; `/query-regression` runs the eight routine default cases. Manual
+only this case; `/query-regression` runs the nine routine default cases, including
+`sst_float_bss`, `promql_instant_last_row_9034`, and
+`mito_prefilter_all_match`. Manual
 workflow dispatch accepts the `heavy` token to select this case.
 
 The routine default set also includes
@@ -371,6 +397,89 @@ uv run --no-project python .github/scripts/query-regression-run.py \
   --work-dir /tmp/query-regression-work
 ```
 
+### SST float BSS comparison
+
+`tests/perf/query_cases/sst_float_bss/case.toml` is included in the routine
+`all` default case group and compares a default empty physical metric table with
+a candidate byte-stream-split (BSS) physical table. It writes 1,000 series × 4,320 samples
+(4,320,000 rows) using synthetic `bounded_mixed` values: 95% integral series and
+5% nonintegral series, with bounded per-series fluctuation rather than globally
+unique values. This deliberately matches a 95/5 design; it is not an empirical
+claim about any production population. A separate survey sample found 94.79%
+integral values, but that observation does not make its values globally unique.
+Current mixed-data evidence, including SST inspection, exact-bit row verification,
+warmed endpoint SQL, and both warm-reader projections, is recorded in
+[`query_cases/sst_float_bss/RESULTS.md`](query_cases/sst_float_bss/RESULTS.md).
+Timing observations were collected while other builds saturated the shared host;
+they are not performance acceptance evidence. Before rerunning timings, ensure
+the host is idle (not merely this agent), record load throughout, and avoid
+concurrent builds. CPU affinity alone does not isolate memory or I/O contention.
+The historical unique-integer workload is preserved only in Git history and does
+not apply to this case.
+Both targets must use the exact same release `greptime` binary; only the
+per-target table setup SQL differs. Run the existing driver from a checkout
+containing that binary, with absolute paths and fresh data directories for every
+run:
+
+```bash
+REPO="$(pwd -P)"
+RELEASE_GREPTIME="/absolute/path/to/release/greptime"
+FIXTURE_GENERATOR="/absolute/path/to/release/query_perf_fixture"
+RUNNER="/absolute/path/to/release/query_regression_runner"
+WORK_DIR="/absolute/path/to/fresh/sst-float-bss-run-1"
+cd "$REPO"
+uv run --no-project python "$REPO/.github/scripts/query-regression-run.py" \
+  --cases "$REPO/tests/perf/query_cases/sst_float_bss/case.toml" \
+  --base-src "$REPO" \
+  --candidate-src . \
+  --base-bin "$RELEASE_GREPTIME" \
+  --candidate-bin "$RELEASE_GREPTIME" \
+  --fixture-generator "$FIXTURE_GENERATOR" \
+  --runner "$RUNNER" \
+  --work-dir "$WORK_DIR" \
+  --summary-script "$REPO/.github/scripts/query-regression-summary.py"
+```
+
+Repeat the command three times with a different fresh absolute `WORK_DIR` each
+run. In each `query-regression-report.json`, compare base and candidate
+`targets[].storage_inspection.summary.summary.total_file_size`, and each query's
+`targets[].measurements[].latency_ms_median`. Storage percentage is
+`(candidate_total_file_size - base_total_file_size) / base_total_file_size * 100`;
+query latency percentage is
+`(candidate_latency_ms_median - base_latency_ms_median) / base_latency_ms_median * 100`.
+The configured three warmups occur after the initial query validation and before
+that query's 15 measured endpoint requests, so query latency is a warmed
+frontend/cache measurement. The case restricts storage inspection to
+`data/greptime/public`, excluding `greptime_private` SSTs. After the datanodes
+stop, it also runs seven iterations of value-only `parquetbench` for all
+inspected SST files and sequential `scanbench` over the corresponding regions,
+with parallelism one. The three explicit flushes yielded six SSTs in the observed
+mixed runs—three 1,105,920-row files and three 334,080-row files—because storage split each
+flush; this is an observation, not a guaranteed file layout. Their per-run
+output and aggregate `parquetbench_median_average_ms` and
+`scanbench_median_average_ms` are under
+`targets[].read_bench`; they are quiescent local-file read/scan diagnostics, not
+warmed frontend-query latency measurements. Bench averages include iteration one;
+no OS cache is dropped, and the driver runs base before candidate.
+
+Historical [PR #8548](https://github.com/GreptimeTeam/greptimedb/pull/8548)
+reported storage savings alongside warm-read slowdowns for a different mixed
+counter/gauge study. It used value-only and all-column projections, discarded the
+first iteration, and alternated target order. This case reuses that reader
+measurement approach, not its dataset or results. For the supplementary warm
+comparison, reuse recorded commands against stopped data directories with eight
+iterations, discard iteration one, and alternate target order for eight rounds.
+For parquetbench, sum all per-file warm medians before taking the outer median;
+for scanbench, take the outer median of whole-region warm medians. Keep
+post-flush and post-compaction measurements separate.
+
+The case's `-5.0` storage target and `25` query-latency guardrail are experimental
+acceptance targets, not observed-benefit claims; do not relax them if a run
+fails. Its three periodic flushes and shared high TWCS trigger avoid the normal
+four-file compaction trigger from confounding the layout. Check footer encodings
+(BSS on candidate and no BSS on base) and data equality from the artifacts rather
+than through a new harness framework.
+
 For a focused manual reproduction of the Mito prefilter all-match optimization, use the existing lifecycle command above with `--cases tests/perf/query_cases/mito_prefilter_all_match/case.toml`. Its four count probes require explicit result inspection rather than automatic validation: expect `262144`, `131072`, `16896`, and `16896` in query order. The default lifecycle retains caches, so the optimization remains exercised but its timing includes warm-cache effects; use an explicitly configured cold environment when a cold comparison is required.
 
 The Rust runner subcommands are also useful for focused diagnostics:
@@ -400,8 +509,8 @@ parquetbench/scanbench` as the read-bench tool against each target's data direct
 
 The workflow runs when an allowlisted repository admin comments
 `/query-regression` on a non-draft PR. It does not rerun on pushes,
-ready-for-review, or reopen events. `/query-regression` runs the eight routine
-default cases, including `promql_instant_last_row_9034` and
+ready-for-review, or reopen events. `/query-regression` runs the nine routine
+default cases, including `sst_float_bss`, `promql_instant_last_row_9034`, and
 `mito_prefilter_all_match`; `/query-regression heavy` runs only the
 high-cardinality remote-write #7913 case. PR runs build base/candidate once and
 use `--allow-large-fixture`. Manual `workflow_dispatch` runs can pass `all`,

tests/perf/query_cases/sst_float_bss/RESULTS.md

@@ -0,0 +1,145 @@
+# SST float BSS: mixed-data post-flush evidence
+
+This report replaces the earlier unique-integer experiment as the primary evidence
+for `sst_float_bss`. The detailed historical numbers remain in Git history; they
+apply only to that superseded workload and must not be used for the current mixed
+case.
+
+> **Timing measurements are confounded.** After these runs, the user reported
+> that other agents' builds saturated all 20 logical CPUs during measurement.
+> No host-wide idle/load gate was enforced. The latency tables below are retained
+> as raw observations only: neither improvements nor threshold failures establish
+> BSS performance. SST byte counts and cross-target stored-row checks remain
+> valid for the inspected dataset. Stable-host timing verification is pending.
+
+## Case and data shape
+
+- Generator/case revision: `149b49de5f3`.
+- Base and candidate used the same release `greptime` binary. The only physical
+  table difference was `experimental_sst_float_field_encoding`: `default` versus
+  `byte_stream_split`; base used dictionary encoding and candidate used BSS
+  without dictionaries. This is not a dictionary-policy-held-constant comparison.
+- Each of three fresh post-flush runs wrote 1,000 series × 4,320 samples at
+  60-second intervals: 4,320,000 rows per target. Three explicit flushes produced
+  six SSTs per target: three × 1,105,920 rows and three × 334,080 rows, with 45
+  `greptime_value` row groups total. This is observed layout, not a layout
+  guarantee.
+- `bounded_mixed` is a synthetic, uncalibrated design: a series hash selects its
+  baseline and span band, and 60-sample anchor interpolation produces bounded
+  fluctuation. It does not represent a measured production distribution.
+- Exactly 950 equal-length series are integer-only and 50 are fractional-only,
+  yielding 4,104,000 integer rows and 216,000 fractional rows per target. The
+  equal-length 1,000-series design deliberately makes both the series and row
+  shares 95%/5%.
+- The 95% design is not an empirical claim. A [local preview survey project](http://192.168.50.85:8765/view/prometheus-metrics-186-project-preliminary)
+  observed 94.79% integral **samples**; it does not establish a global rate,
+  global uniqueness, or a 95% series distribution.
+
+## Initial SST inspection and value verification
+
+Inspection is limited to `data/greptime/public`; bytes exclude WAL, indexes,
+metadata, and internal/private-table storage. Each of the three runs had the
+same totals:
+
+| Measure | Default | BSS | BSS change |
+| --- | ---: | ---: | ---: |
+| Selected SST bytes | 20,443,131 | 8,884,743 | -56.5392% |
+| `greptime_value` compressed bytes | 20,201,188 | 8,643,964 | -57.2106% |
+| SST files / rows / value row groups | 6 / 4,320,000 / 45 | 6 / 4,320,000 / 45 | — |
+
+Footer inspection found `PLAIN`/`RLE`/`RLE_DICTIONARY` for the default value
+column and `RLE`/`BYTE_STREAM_SPLIT` for BSS (both ZSTD level 1). The configured
+storage threshold remains a 5% reduction; this repeated SST-size result passes
+that threshold. It is post-flush evidence only, not a compaction result or a
+general float-compression claim.
+
+For every base/candidate group in all three runs, rows were compared by the same
+primary key and timestamp and their Float64 bits were exact matches: 950
+integer-only and 50 fractional-only series, 4,104,000 and 216,000 rows
+respectively. The six canonical sorted-row SHA-256 results are identical:
+
+```
+48f87d7d01bc8a0ce9a9cdb3033f322e787784fcae61f1417ac25b05c4a40375
+```
+
+This verifies exact cross-target stored rows, not independent reconstruction of
+the generator output. All 1,000 series had both rises and falls, and every one
+of the 36 inspected SSTs contained both integer-valued and fractional values;
+these were not stored as two disjoint file sets.
+
+## Warmed endpoint SQL
+
+Each query used three warmups followed by 15 measured endpoint requests. Values
+are median/p95 milliseconds; delta is `(BSS median - default median) / default
+median × 100%`. The guardrail remains a maximum 25% candidate median regression and
+was not changed for these runs.
+
+| Run | Query | Default median / p95 | BSS median / p95 | Delta | Guardrail |
+| --- | --- | ---: | ---: | ---: | --- |
+| mixed-1 | `sum_all_values` | 27.13 / 37.38 | 25.45 / 38.99 | -6.21% | pass |
+| mixed-1 | `hourly_sum_values` | 118.78 / 136.63 | 130.93 / 185.26 | +10.23% | pass |
+| mixed-2 | `sum_all_values` | 24.02 / 35.02 | 36.34 / 56.01 | +51.31% | **fail** |
+| mixed-2 | `hourly_sum_values` | 95.58 / 120.40 | 85.45 / 148.03 | -10.60% | pass |
+| mixed-3 | `sum_all_values` | 38.25 / 55.41 | 19.68 / 29.49 | -48.55% | pass |
+| mixed-3 | `hourly_sum_values` | 56.77 / 148.29 | 121.87 / 143.16 | +114.66% | **fail** |
+
+The two recorded threshold failures remain in the artifacts, but the saturated
+shared host prevents attributing them to BSS. Passing entries likewise do not
+establish performance acceptance. No CPU isolation or cache flush was used.
+
+## Stopped-directory warm readers
+
+For each of the three datasets, eight outer rounds alternate base/candidate order.
+Each command runs eight iterations, discards the first, and takes the median of
+iterations 2–8. Parquetbench reads every SST: sum the six per-file medians, then
+take the median of the eight round totals. Scanbench scans the whole region and
+reports the outer median of its eight warm medians. These are distinct metrics,
+not interchangeable whole-query latencies. Deltas are ratios of separately
+aggregated medians. Value projection selects `greptime_value`; all-column
+parquetbench returns five physical columns, while scanbench uses `{}` (all
+region columns). Projection order is fixed; target order alternates.
+
+| Dataset | Projection | Reader | Base ms | BSS ms | Change |
+| --- | --- | --- | ---: | ---: | ---: |
+| mixed-1 | value | parquetbench | 95.957 | 68.442 | -28.67% |
+| mixed-1 | value | scanbench | 144.293 | 115.421 | -20.01% |
+| mixed-1 | allcolumns | parquetbench | 176.620 | 103.761 | -41.25% |
+| mixed-1 | allcolumns | scanbench | 200.881 | 202.660 | +0.89% |
+| mixed-2 | value | parquetbench | 112.151 | 74.275 | -33.77% |
+| mixed-2 | value | scanbench | 109.824 | 78.189 | -28.80% |
+| mixed-2 | allcolumns | parquetbench | 141.944 | 102.681 | -27.66% |
+| mixed-2 | allcolumns | scanbench | 211.414 | 163.261 | -22.78% |
+| mixed-3 | value | parquetbench | 121.827 | 67.565 | -44.54% |
+| mixed-3 | value | scanbench | 100.968 | 79.878 | -20.89% |
+| mixed-3 | allcolumns | parquetbench | 113.984 | 94.361 | -17.22% |
+| mixed-3 | allcolumns | scanbench | 193.101 | 164.175 | -14.98% |
+
+All 672 commands / 5,376 raw iterations returned the expected rows; timing,
+projection configurations, parquet schemas and order metadata were checked.
+Run 1 all-column scanbench differed by **+0.89%**, negligible relative to the
+observed variation, not evidence of a regression. Its round medians ranged from
+167.197–663.465 ms (base) and 133.339–557.352 ms (BSS). There is substantial
+variability; these measurements do not establish a stable gain on every read
+path. The reported concurrent CPU saturation affects apparent improvements as
+well as slowdowns; no causal performance conclusion is drawn from these runs.
+
+## Scope and recorded checks
+
+This case compares default and BSS post-flush data only. It adds no benchmark
+framework and no public artifact upload. Historical unique-integer results and
+other mixed counter/gauge studies have different data and methods; they are
+context only and do not apply to this workload.
+
+Recorded checks for this change set: 37 targeted Rust tests, 20 Python tooling
+tests, and 15 plans passed. Full-workspace tests were not run.
+
+Local artifacts are under
+`/mnt/nvme_rust/rust-targets/metric-bss-perf/experiments/mixed-{1,2,3}/`,
+including `value-verification.json`, `verification.txt`, and each run's
+`sst_float_bss/query-regression-report.json`.
+
+Warm artifacts are in sibling `warm-mixed-{1,2,3}/` directories;
+`mixed-warm-comparison.json` contains the table and per-round ranges.
+`mixed-file-composition.json` records integer/fractional counts for each SST.
+The shared release `greptime` SHA-256 is
+`2ce0ab1670cd4bdb0f60c87af3e1f494ee66c3c6ab83bce943a073638514eb2b`.

tests/perf/query_cases/sst_float_bss/case.toml

@@ -0,0 +1,83 @@
+# Default all-group default-versus-BSS storage and warmed-query comparison.
+
+[case]
+name = "sst_float_bss"
+description = "Compare default and byte-stream-split floating-point SST bytes and warmed value-scanning query latency with bounded 95/5 per-series DOUBLE values"
+
+[scenario]
+kind = "prom_remote_write_then_query"
+
+[scenario.remote_write]
+database = "public"
+metric = "sst_float_bss"
+physical_table = "sst_float_bss_physical"
+series_count = 1000
+samples_per_series = 4320
+sample_chunk_size = 1440
+flush_every_sample_chunks = 1
+start_unix_millis = 1_704_067_200_000 # 2024-01-01T00:00:00Z
+step_millis = 60000
+chunk_series_count = 128
+timeout_seconds = 600
+visibility_timeout_seconds = 300
+
+# Omit the default encoding from base setup because older binaries do not recognize it.
+# The high shared TWCS trigger avoids compaction changing the three-flush layout.
+base_setup_sql = [
+  "CREATE TABLE sst_float_bss_physical (greptime_timestamp TIMESTAMP TIME INDEX, greptime_value DOUBLE) ENGINE=metric WITH ('physical_metric_table'='', 'compaction.twcs.trigger_file_num'='100')",
+]
+candidate_setup_sql = [
+  "CREATE TABLE sst_float_bss_physical (greptime_timestamp TIMESTAMP TIME INDEX, greptime_value DOUBLE) ENGINE=metric WITH ('physical_metric_table'='', 'experimental_sst_float_field_encoding'='byte_stream_split', 'compaction.twcs.trigger_file_num'='100')",
+]
+
+[scenario.remote_write.value]
+# Synthetic, not empirical: 95% integral and 5% nonintegral series vary
+# smoothly within deterministic per-series ranges; series are not a global monotonic counter.
+pattern = "bounded_mixed"
+base = 0
+step = 1
+cardinality = 1000
+seed = 42
+mixed_every = 20
+run_length = 60
+
+[scenario.remote_write.prom_store]
+pending_rows_flush_interval = "1s"
+max_batch_rows = 100000
+
+[scenario.remote_write.storage]
+root_suffix = "data/greptime/public"
+column = "greptime_value"
+min_files = 1
+min_files_with_column = 1
+max_candidate_total_file_size_regression_pct = -5.0
+
+[scenario.remote_write.read_bench]
+enabled = true
+parquetbench = true
+scanbench = true
+iterations = 7
+projection = ["greptime_value"]
+parquet_reader = "direct"
+scan_scanner = "seq"
+parallelism = 1
+
+[[scenario.queries]]
+name = "sum_all_values"
+kind = "sql"
+query = "SELECT sum(greptime_value) FROM sst_float_bss"
+warmup = 3
+iterations = 15
+
+[scenario.queries.thresholds]
+max_candidate_latency_regression_pct = 25
+
+[[scenario.queries]]
+name = "hourly_sum_values"
+kind = "sql"
+query = "SELECT date_bin(INTERVAL '1 hour', greptime_timestamp) AS time_window, sum(greptime_value) FROM sst_float_bss WHERE greptime_timestamp >= TIMESTAMP '2024-01-01 00:00:00' AND greptime_timestamp < TIMESTAMP '2024-01-04 00:00:00' GROUP BY time_window ORDER BY time_window"
+warmup = 3
+iterations = 15
+
+[scenario.queries.thresholds]
+max_candidate_latency_regression_pct = 25

tests/perf/test_query_regression_case_selection.py

@@ -16,6 +16,7 @@
 """Coverage for query-regression case group selection."""
 
 import importlib.util
+import re
 import sys
 import unittest
 from pathlib import Path
@@ -35,6 +36,7 @@ class QueryRegressionCaseSelectionTest(unittest.TestCase):
             runner.split_cases(["all"]),
             [
                 "tests/perf/query_cases/smoke_direct_sst/case.toml",
+                "tests/perf/query_cases/sst_float_bss/case.toml",
                 "tests/perf/query_cases/prom_remote_write_seeded_random/case.toml",
                 "tests/perf/query_cases/prom_remote_write_run_heavy/case.toml",
                 "tests/perf/query_cases/prom_remote_write_mixed_every/case.toml",
@@ -45,8 +47,8 @@ class QueryRegressionCaseSelectionTest(unittest.TestCase):
             ],
         )
 
-    def test_implicit_selection_has_eight_routine_cases(self) -> None:
-        self.assertEqual(len(runner.DEFAULT_CASES), 8)
+    def test_implicit_selection_has_nine_routine_cases(self) -> None:
+        self.assertEqual(len(runner.DEFAULT_CASES), 9)
         self.assertEqual(runner.split_cases([]), runner.DEFAULT_CASES)
 
     def test_heavy_selects_only_remote_write_7913(self) -> None:
@@ -59,6 +61,23 @@ class QueryRegressionCaseSelectionTest(unittest.TestCase):
         case = "tests/perf/query_cases/sql_topk_order_by/case.toml"
         self.assertEqual(runner.split_cases([case]), [case])
 
+    def test_sst_float_bss_omits_base_encoding_for_older_binaries(self) -> None:
+        case_path = Path(__file__).parent / "query_cases/sst_float_bss/case.toml"
+        setup_sql = dict(re.findall(
+            r'(base|candidate)_setup_sql = \[\s*"([^"\n]+)"',
+            case_path.read_text(),
+        ))
+        base_setup = setup_sql["base"]
+        candidate_setup = setup_sql["candidate"]
+        candidate_option = "'experimental_sst_float_field_encoding'='byte_stream_split'"
+
+        self.assertNotIn("experimental_sst_float_field_encoding", base_setup)
+        self.assertIn(candidate_option, candidate_setup)
+        self.assertEqual(
+            base_setup,
+            candidate_setup.replace(f", {candidate_option}", ""),
+        )
+
 
 if __name__ == "__main__":
     unittest.main()