fix(storage): read legacy bincode v4 segments (#20191)

* fix(storage): read legacy bincode v4 segments

Decode bincode v4 segment block metas and summaries through v3 frozen metadata so old segment stats use the legacy bincode layout.

Expose encoding, compression, and block_meta_size in fuse_segment for low-cost online diagnosis of segment metadata encoding.

Add regression coverage for legacy bincode segments, current MessagePack segments, and the new fuse_segment diagnostic columns.

* fix(storage): adapt segment tests to parquet metadata

Author: dantengsky

src/query/expression/src/converts/meta/bincode.rs

@@ -44,14 +44,15 @@ pub enum LegacyScalar {
     Decimal(DecimalScalar),
     Timestamp(i64),
     Date(i32),
-    Interval(months_days_micros),
     Boolean(bool),
     String(Vec<u8>),
     Array(LegacyColumn),
     Map(LegacyColumn),
     Bitmap(Vec<u8>),
     Tuple(Vec<Scalar>),
     Variant(Vec<u8>),
+    // Interval must be at end: old bincode data did not have this variant
+    Interval(months_days_micros),
 }
 
 #[allow(unused, dead_code)]

src/query/expression/tests/it/meta_scalar.rs

@@ -17,12 +17,61 @@ use databend_common_expression::Scalar;
 use databend_common_expression::converts::meta::IndexScalar;
 use databend_common_expression::converts::meta::LegacyColumn;
 use databend_common_expression::converts::meta::LegacyScalar;
+use databend_common_expression::types::DecimalScalar;
+use databend_common_expression::types::NumberScalar;
 use databend_common_io::prelude::bincode_deserialize_from_slice;
 use databend_common_io::prelude::bincode_serialize_into_buf;
 
 use crate::DataTypeFilter;
 use crate::rand_block_for_all_types;
 
+#[allow(dead_code)]
+#[derive(serde::Serialize)]
+enum OldLegacyScalar {
+    Null,
+    EmptyArray,
+    EmptyMap,
+    Number(NumberScalar),
+    Decimal(DecimalScalar),
+    Timestamp(i64),
+    Date(i32),
+    Boolean(bool),
+    String(Vec<u8>),
+    Array(()),
+    Map(()),
+    Bitmap(Vec<u8>),
+    Tuple(Vec<Scalar>),
+    Variant(Vec<u8>),
+}
+
+#[test]
+pub fn test_legacy_scalar_reads_old_bincode_layout() -> databend_common_exception::Result<()> {
+    let old_scalars = vec![
+        OldLegacyScalar::Boolean(true),
+        OldLegacyScalar::String(b"abc".to_vec()),
+        OldLegacyScalar::Tuple(vec![Scalar::String("tuple".to_string())]),
+        OldLegacyScalar::Variant(vec![1, 2, 3]),
+    ];
+
+    let mut data = vec![];
+    bincode_serialize_into_buf(&mut data, &old_scalars)?;
+    let new_scalars: Vec<LegacyScalar> = bincode_deserialize_from_slice(&data)?;
+
+    let decoded = new_scalars
+        .into_iter()
+        .map(Scalar::from)
+        .collect::<Vec<_>>();
+    assert_eq!(decoded[0], Scalar::Boolean(true));
+    assert_eq!(decoded[1], Scalar::String("abc".to_string()));
+    assert_eq!(
+        decoded[2],
+        Scalar::Tuple(vec![Scalar::String("tuple".to_string())])
+    );
+    assert!(matches!(&decoded[3], Scalar::Variant(bytes) if bytes == &[1, 2, 3]));
+
+    Ok(())
+}
+
 #[test]
 pub fn test_legacy_converts() -> databend_common_exception::Result<()> {
     use rand::Rng;

src/query/storages/common/table_meta/src/meta/v3/mod.rs

@@ -14,7 +14,7 @@
 
 // we use "frozen" types of table meta, to make sure that the type we used for
 // bincode deserialization is compatible with the type we used for bincode serialization.
-mod frozen;
+pub(crate) mod frozen;
 mod segment;
 mod snapshot;
 mod table_snapshot_statistics;