feat: ignore row group filter when certain column type (#958)

## Rationale
Part of #955

## Detailed Changes
- Ignore filter for null, double, float, varbinary, boolean 
- Fix parse meta in  `meta_from_sst`

## Test Plan
- Before: max_seq:132734153, size:348.675M, metadata:40.142M,
kv:38.175M, filter:28.525M, row_num:7038949
- After: max_seq:132734153, size:334.045M, metadata:25.512M, kv:23.545M,
filter:17.562M, row_num:7038949

Author: jiacai2050

analytic_engine/src/sst/parquet/meta_data.rs

@@ -6,7 +6,12 @@ use std::{fmt, ops::Index, sync::Arc};
 
 use bytes::Bytes;
 use ceresdbproto::{schema as schema_pb, sst as sst_pb};
-use common_types::{schema::Schema, time::TimeRange, SequenceNumber};
+use common_types::{
+    datum::DatumKind,
+    schema::{RecordSchemaWithKey, Schema},
+    time::TimeRange,
+    SequenceNumber,
+};
 use common_util::define_result;
 use snafu::{Backtrace, OptionExt, ResultExt, Snafu};
 use xorfilter::{Xor8, Xor8Builder};
@@ -120,14 +125,38 @@ pub struct RowGroupFilterBuilder {
 }
 
 impl RowGroupFilterBuilder {
-    pub(crate) fn with_num_columns(num_col: usize) -> Self {
-        Self {
-            builders: vec![None; num_col],
-        }
+    pub(crate) fn new(record_schema: &RecordSchemaWithKey) -> Self {
+        let builders = record_schema
+            .columns()
+            .iter()
+            .enumerate()
+            .map(|(i, col)| {
+                if record_schema.is_primary_key_index(i) {
+                    return None;
+                }
+
+                if matches!(
+                    col.data_type,
+                    DatumKind::Null
+                        | DatumKind::Double
+                        | DatumKind::Float
+                        | DatumKind::Varbinary
+                        | DatumKind::Boolean
+                ) {
+                    return None;
+                }
+
+                Some(Xor8Builder::default())
+            })
+            .collect();
+
+        Self { builders }
     }
 
     pub(crate) fn add_key(&mut self, col_idx: usize, key: &[u8]) {
-        self.builders[col_idx].get_or_insert_default().insert(key)
+        if let Some(b) = self.builders[col_idx].as_mut() {
+            b.insert(key)
+        }
     }
 
     pub(crate) fn build(self) -> Result<RowGroupFilter> {
@@ -403,6 +432,8 @@ impl TryFrom<sst_pb::ParquetMetaData> for ParquetMetaData {
 
 #[cfg(test)]
 mod tests {
+    use common_types::tests::build_schema;
+
     use super::*;
 
     #[test]
@@ -447,16 +478,22 @@ mod tests {
 
     #[test]
     fn test_row_group_filter_builder() {
-        let mut builders = RowGroupFilterBuilder::with_num_columns(1);
+        // (key1(varbinary), key2(timestamp), field1(double), field2(string))
+        let schema = build_schema();
+        let record_schema = schema.to_record_schema_with_key();
+        let mut builders = RowGroupFilterBuilder::new(&record_schema);
         for key in ["host-123", "host-456", "host-789"] {
-            builders.add_key(0, key.as_bytes());
+            builders.add_key(3, key.as_bytes());
         }
         let row_group_filter = builders.build().unwrap();
+        for i in 0..3 {
+            assert!(row_group_filter.column_filters[i].is_none());
+        }
 
         let testcase = [("host-123", true), ("host-321", false)];
         for (key, expected) in testcase {
             let actual = row_group_filter
-                .contains_column_data(0, key.as_bytes())
+                .contains_column_data(3, key.as_bytes())
                 .unwrap();
 
             assert_eq!(expected, actual);

analytic_engine/src/sst/parquet/writer.rs

@@ -147,7 +147,7 @@ impl RecordBatchGroupWriter {
         &self,
         row_group_batch: &[RecordBatchWithKey],
     ) -> Result<RowGroupFilter> {
-        let mut builder = RowGroupFilterBuilder::with_num_columns(row_group_batch[0].num_columns());
+        let mut builder = RowGroupFilterBuilder::new(row_group_batch[0].schema_with_key());
 
         for partial_batch in row_group_batch {
             for (col_idx, column) in partial_batch.columns().iter().enumerate() {

common_types/src/schema.rs

@@ -514,6 +514,10 @@ impl RecordSchemaWithKey {
         &self.primary_key_indexes
     }
 
+    pub fn is_primary_key_index(&self, idx: usize) -> bool {
+        self.primary_key_indexes.contains(&idx)
+    }
+
     pub fn index_of(&self, name: &str) -> Option<usize> {
         self.record_schema.index_of(name)
     }

tools/src/sst_util.rs

@@ -9,8 +9,14 @@ pub async fn meta_from_sst(store: &ObjectStoreRef, sst_path: &Path) -> MetaData
     let get_result = store.get(sst_path).await.unwrap();
     let chunk_reader = get_result.bytes().await.unwrap();
     let metadata = footer::parse_metadata(&chunk_reader).unwrap();
-    let kv_metas = metadata.file_metadata().key_value_metadata().unwrap();
 
-    let parquet_meta_data = encoding::decode_sst_meta_data(&kv_metas[0]).unwrap();
+    let file_meta_data = metadata.file_metadata();
+    let kv_metas = file_meta_data.key_value_metadata().unwrap();
+    let kv_meta = kv_metas
+        .iter()
+        .find(|kv| kv.key == encoding::META_KEY)
+        .unwrap();
+
+    let parquet_meta_data = encoding::decode_sst_meta_data(kv_meta).unwrap();
     MetaData::from(parquet_meta_data)
 }