feat: separate metadata from parquet's kv_metadata

analytic_engine/src/sst/meta_data/cache.rs

@@ -18,12 +18,20 @@ use std::{
 };
 
 use lru::LruCache;
-use parquet::file::metadata::FileMetaData;
+use object_store::{ObjectStoreRef, Path};
+use parquet::{data_type::AsBytes, file::metadata::FileMetaData, format::KeyValue};
+use parquet_ext::meta_data::ChunkReader;
 use snafu::{ensure, OptionExt, ResultExt};
 
 use crate::sst::{
-    meta_data::{DecodeCustomMetaData, KvMetaDataNotFound, ParquetMetaDataRef, Result},
-    parquet::encoding,
+    meta_data::{
+        DecodeCustomMetaData, FetchAndDecodeSstMeta, KvMetaDataNotFound, KvMetaPathEmpty,
+        KvMetaVersionEmpty, MetaPathVersionWrong, ObjectStoreError, ParquetMetaDataRef, Result,
+    },
+    parquet::{
+        async_reader::ChunkReaderAdapter,
+        encoding::{self, decode_sst_custom_meta_data, META_PATH_VERSION_V1, META_PATH_VERSION_V2},
+    },
 };
 
 pub type MetaCacheRef = Arc<MetaCache>;
@@ -45,37 +53,83 @@ impl MetaData {
     /// contains no extended custom information.
     // TODO: remove it and use the suggested api.
     #[allow(deprecated)]
-    pub fn try_new(
+    pub async fn try_new(
         parquet_meta_data: &parquet_ext::ParquetMetaData,
         ignore_sst_filter: bool,
+        meta_store: ObjectStoreRef,
     ) -> Result<Self> {
         let file_meta_data = parquet_meta_data.file_metadata();
         let kv_metas = file_meta_data
             .key_value_metadata()
             .context(KvMetaDataNotFound)?;
 
         ensure!(!kv_metas.is_empty(), KvMetaDataNotFound);
-        let mut other_kv_metas = Vec::with_capacity(kv_metas.len() - 1);
+
+        let mut meta_path = None;
+
+        let mut other_kv_metas: Vec<KeyValue> = Vec::with_capacity(kv_metas.len() - 1);
         let mut custom_kv_meta = None;
+        let mut meta_path_version = META_PATH_VERSION_V1.to_string();
         for kv_meta in kv_metas {
             // Remove our extended custom meta data from the parquet metadata for small
             // memory consumption in the cache.
             if kv_meta.key == encoding::META_KEY {
                 custom_kv_meta = Some(kv_meta);
+            } else if kv_meta.key == encoding::META_PATH_KEY {
+                meta_path = kv_meta.value.as_ref().map(|path| Path::from(path.as_str()))
+            } else if kv_meta.key == encoding::META_PATH_VERSION_KEY {
+                meta_path_version = kv_meta.value.as_ref().context(KvMetaVersionEmpty)?.clone();
             } else {
                 other_kv_metas.push(kv_meta.clone());
             }
         }
 
-        let custom = {
+        // Must ensure custom metadata only store in one place (V2)
+        ensure!(
+            custom_kv_meta.is_none() || meta_path.is_none(),
+            KvMetaDataNotFound
+        );
+
+        let custom = if meta_path_version == META_PATH_VERSION_V1 {
             let custom_kv_meta = custom_kv_meta.context(KvMetaDataNotFound)?;
             let mut sst_meta =
                 encoding::decode_sst_meta_data(custom_kv_meta).context(DecodeCustomMetaData)?;
             if ignore_sst_filter {
                 sst_meta.parquet_filter = None;
             }
-
             Arc::new(sst_meta)
+        } else if meta_path_version == META_PATH_VERSION_V2 {
+            let decode_custom_metadata = match meta_path {
+                Some(meta_path) => {
+                    let meta_size = meta_store
+                        .head(&meta_path)
+                        .await
+                        .context(ObjectStoreError)?
+                        .size;
+
+                    let meta_chunk_reader_adapter =
+                        ChunkReaderAdapter::new(&meta_path, &meta_store);
+
+                    let metadata = meta_chunk_reader_adapter
+                        .get_bytes(0..meta_size)
+                        .await
+                        .with_context(|| FetchAndDecodeSstMeta {
+                            file_path: meta_path.to_string(),
+                        })?;
+
+                    Some(
+                        decode_sst_custom_meta_data(metadata.as_bytes())
+                            .context(DecodeCustomMetaData)?,
+                    )
+                }
+                None => return KvMetaPathEmpty {}.fail(),
+            };
+            Arc::new(decode_custom_metadata.unwrap())
+        } else {
+            return MetaPathVersionWrong {
+                path_version: meta_path_version,
+            }
+            .fail();
         };
 
         // let's build a new parquet metadata without the extended key value
@@ -155,6 +209,7 @@ mod tests {
         schema::Builder as CustomSchemaBuilder,
         time::{TimeRange, Timestamp},
     };
+    use object_store::LocalFileSystem;
     use parquet::{arrow::ArrowWriter, file::footer};
     use parquet_ext::ParquetMetaData;
 
@@ -245,8 +300,8 @@ mod tests {
         writer.close().unwrap();
     }
 
-    #[test]
-    fn test_arrow_meta_data() {
+    #[tokio::test]
+    async fn test_arrow_meta_data() {
         let temp_dir = tempfile::tempdir().unwrap();
         let parquet_file_path = temp_dir.path().join("test_arrow_meta_data.par");
         let schema = {
@@ -285,7 +340,12 @@ mod tests {
         let parquet_file = File::open(parquet_file_path.as_path()).unwrap();
         let parquet_meta_data = footer::parse_metadata(&parquet_file).unwrap();
 
-        let meta_data = MetaData::try_new(&parquet_meta_data, false).unwrap();
+        let store =
+            Arc::new(LocalFileSystem::new_with_prefix(parquet_file_path.as_path()).unwrap());
+
+        let meta_data = MetaData::try_new(&parquet_meta_data, false, store)
+            .await
+            .unwrap();
 
         assert_eq!(**meta_data.custom(), custom_meta_data);
         check_parquet_meta_data(&parquet_meta_data, meta_data.parquet());

analytic_engine/src/sst/meta_data/mod.rs

@@ -14,10 +14,11 @@
 
 pub mod cache;
 
-use std::sync::Arc;
+use std::{str::Utf8Error, sync::Arc};
 
 use ceresdbproto::sst as sst_pb;
 use common_types::{schema::Schema, time::TimeRange, SequenceNumber};
+use generic_error::GenericError;
 use macros::define_result;
 use snafu::{Backtrace, OptionExt, ResultExt, Snafu};
 use table_engine::table::TableId;
@@ -47,6 +48,25 @@ pub enum Error {
     ))]
     KvMetaDataNotFound { backtrace: Backtrace },
 
+    #[snafu(display(
+        "Key value meta path version in parquet is empty\nBacktrace\n:{}",
+        backtrace
+    ))]
+    KvMetaVersionEmpty { backtrace: Backtrace },
+
+    #[snafu(display("Key value meta path in parquet is empty\nBacktrace\n:{}", backtrace))]
+    KvMetaPathEmpty { backtrace: Backtrace },
+
+    #[snafu(display(
+        "Mata path version {} is wrong\nBacktrace\n:{}",
+        path_version,
+        backtrace
+    ))]
+    MetaPathVersionWrong {
+        path_version: String,
+        backtrace: Backtrace,
+    },
+
     #[snafu(display("Metadata in proto struct is not found.\nBacktrace\n:{}", backtrace))]
     MetaDataNotFound { backtrace: Backtrace },
 
@@ -64,6 +84,26 @@ pub enum Error {
 
     #[snafu(display("Failed to convert parquet meta data, err:{}", source))]
     ConvertParquetMetaData { source: parquet::meta_data::Error },
+
+    #[snafu(display("Meet a object store error, err:{source}\nBacktrace:\n{backtrace}"))]
+    ObjectStoreError {
+        source: object_store::ObjectStoreError,
+        backtrace: Backtrace,
+    },
+    #[snafu(display(
+        "Failed to decode sst meta data, file_path:{file_path}, err:{source}.\nBacktrace:\n{backtrace:?}",
+    ))]
+    FetchAndDecodeSstMeta {
+        file_path: String,
+        source: GenericError,
+        backtrace: Backtrace,
+    },
+
+    #[snafu(display("Meet a object store error, err:{source}\nBacktrace:\n{backtrace}"))]
+    Utf8ErrorWrapper {
+        source: Utf8Error,
+        backtrace: Backtrace,
+    },
 }
 
 define_result!(Error);

analytic_engine/src/sst/parquet/async_reader.rs

@@ -70,7 +70,6 @@ use crate::{
         reader::{error::*, Result, SstReader},
     },
 };
-
 const PRUNE_ROW_GROUPS_METRICS_COLLECTOR_NAME: &str = "prune_row_groups";
 type SendableRecordBatchStream = Pin<Box<dyn Stream<Item = Result<ArrowRecordBatch>> + Send>>;
 type RecordBatchWithKeyStream = Box<dyn Stream<Item = Result<RecordBatchWithKey>> + Send + Unpin>;
@@ -377,7 +376,8 @@ impl<'a> Reader<'a> {
 
         // TODO: Support page index until https://github.com/CeresDB/ceresdb/issues/1040 is fixed.
 
-        MetaData::try_new(&parquet_meta_data, ignore_sst_filter)
+        MetaData::try_new(&parquet_meta_data, ignore_sst_filter, self.store.clone())
+            .await
             .box_err()
             .context(DecodeSstMeta)
     }