chore: remove hybrid related logic

analytic_engine/src/compaction/picker.rs

@@ -724,7 +724,6 @@ mod tests {
             max_sequence: 200,
             schema: build_schema(),
             parquet_filter: Default::default(),
-            collapsible_cols_idx: Vec::new(),
         };
 
         SstMetaData::Parquet(Arc::new(parquet_meta_data))

analytic_engine/src/sst/factory.rs

@@ -171,7 +171,7 @@ impl Factory for FactoryImpl {
         };
 
         match storage_format {
-            StorageFormat::Columnar | StorageFormat::Hybrid => {
+            StorageFormat::Columnar => {
                 let reader = AsyncParquetReader::new(
                     path,
                     options,
@@ -197,16 +197,9 @@ impl Factory for FactoryImpl {
         store_picker: &'a ObjectStorePickerRef,
         level: Level,
     ) -> Result<Box<dyn SstWriter + Send + 'a>> {
-        let hybrid_encoding = match options.storage_format_hint {
-            StorageFormatHint::Specific(format) => matches!(format, StorageFormat::Hybrid),
-            // `Auto` is mapped to columnar parquet format now, may change in future.
-            StorageFormatHint::Auto => false,
-        };
-
         Ok(Box::new(ParquetSstWriter::new(
             path,
             level,
-            hybrid_encoding,
             store_picker,
             options,
         )))

analytic_engine/src/sst/meta_data/cache.rs

@@ -288,7 +288,6 @@ mod tests {
             max_sequence: 1001,
             schema,
             parquet_filter: None,
-            collapsible_cols_idx: vec![],
         };
         write_parquet_file_with_metadata(parquet_file_path.as_path(), &custom_meta_data);
 

analytic_engine/src/sst/parquet/async_reader.rs

@@ -63,9 +63,7 @@ use crate::{
             SstMetaData,
         },
         parquet::{
-            encoding::ParquetDecoder,
-            meta_data::{ParquetFilter, ParquetMetaDataRef},
-            row_group_pruner::RowGroupPruner,
+            encoding::ParquetDecoder, meta_data::ParquetFilter, row_group_pruner::RowGroupPruner,
         },
         reader::{error::*, Result, SstReader},
     },
@@ -157,20 +155,12 @@ impl<'a> Reader<'a> {
             ArrowRecordBatchProjector::from(row_projector)
         };
 
-        let sst_meta_data = self
-            .meta_data
-            .as_ref()
-            // metadata must be inited after `init_if_necessary`.
-            .unwrap()
-            .custom();
-
         let streams: Vec<_> = streams
             .into_iter()
             .map(|stream| {
                 Box::new(RecordBatchProjector::new(
                     stream,
                     row_projector.clone(),
-                    sst_meta_data.clone(),
                     self.metrics.metrics_collector.clone(),
                 )) as _
             })
@@ -474,14 +464,12 @@ struct RecordBatchProjector {
 
     metrics: ProjectorMetrics,
     start_time: Instant,
-    sst_meta: ParquetMetaDataRef,
 }
 
 impl RecordBatchProjector {
     fn new(
         stream: SendableRecordBatchStream,
         row_projector: ArrowRecordBatchProjector,
-        sst_meta: ParquetMetaDataRef,
         metrics_collector: Option<MetricsCollector>,
     ) -> Self {
         let metrics = ProjectorMetrics {
@@ -494,7 +482,6 @@ impl RecordBatchProjector {
             row_projector,
             metrics,
             start_time: Instant::now(),
-            sst_meta,
         }
     }
 }
@@ -510,8 +497,7 @@ impl Stream for RecordBatchProjector {
                 match record_batch.box_err().context(DecodeRecordBatch {}) {
                     Err(e) => Poll::Ready(Some(Err(e))),
                     Ok(record_batch) => {
-                        let parquet_decoder =
-                            ParquetDecoder::new(&projector.sst_meta.collapsible_cols_idx);
+                        let parquet_decoder = ParquetDecoder::new();
                         let record_batch = parquet_decoder
                             .decode_record_batch(record_batch)
                             .box_err()