chore(cubestore): Upgrade DF: Fix usage of MetadataCacheFactory and CubestoreParquetMetadataCache

Author: srh

rust/cubestore/cubestore/src/config/mod.rs

@@ -49,11 +49,7 @@ use crate::util::memory::{MemoryHandler, MemoryHandlerImpl};
 use crate::CubeError;
 use cuberockstore::rocksdb::{Options, DB};
 use datafusion::cube_ext;
-// use datafusion::physical_plan::parquet::BasicMetadataCacheFactory;
-use crate::queryplanner::metadata_cache::{
-    BasicMetadataCacheFactory, LruParquetMetadataCacheFactory, MetadataCacheFactory,
-    NoopParquetMetadataCache,
-};
+use crate::queryplanner::metadata_cache::BasicMetadataCacheFactory;
 use futures::future::join_all;
 use log::Level;
 use log::{debug, error};
@@ -2034,8 +2030,8 @@ impl Config {
                 let metadata_cache_factory: &_ = cubestore_metadata_cache_factory.cache_factory();
                 CubestoreParquetMetadataCacheImpl::new(
                     match c.metadata_cache_max_capacity_bytes() {
-                        0 => NoopParquetMetadataCache::new(),
-                        max_cached_metadata => LruParquetMetadataCacheFactory::new(
+                        0 => metadata_cache_factory.make_noop_cache(),
+                        max_cached_metadata => metadata_cache_factory.make_lru_cache(
                             max_cached_metadata,
                             Duration::from_secs(c.metadata_cache_time_to_idle_secs()),
                         ),
@@ -2093,10 +2089,6 @@ impl Config {
                     i.get_service_typed().await,
                     i.get_service_typed().await,
                     i.get_service_typed().await,
-                    i.get_service_typed::<dyn CubestoreMetadataCacheFactory>()
-                        .await
-                        .cache_factory()
-                        .clone(),
                 )
             })
             .await;

rust/cubestore/cubestore/src/queryplanner/metadata_cache.rs

@@ -8,7 +8,6 @@ use futures_util::future::BoxFuture;
 use futures_util::FutureExt;
 use std::fmt;
 use std::fmt::{Debug, Formatter};
-use std::fs::File;
 use std::ops::Range;
 use std::sync::Arc;
 use std::time::Duration;
@@ -24,20 +23,19 @@ pub trait MetadataCacheFactory: Sync + Send {
         time_to_idle: Duration,
     ) -> Arc<dyn ParquetFileReaderFactory>;
 }
-
 /// Default MetadataCache, does not cache anything
 #[derive(Debug)]
 pub struct NoopParquetMetadataCache {
-    default_factory: Arc<dyn ParquetFileReaderFactory>,
+    default_factory: DefaultParquetFileReaderFactory,
 }
 
 impl NoopParquetMetadataCache {
-    /// Creates a new DefaultMetadataCache
+     /// Creates a new DefaultMetadataCache
     pub fn new() -> Arc<Self> {
         Arc::new(NoopParquetMetadataCache {
-            default_factory: Arc::new(DefaultParquetFileReaderFactory::new(Arc::new(
-                object_store::local::LocalFileSystem::new(),
-            ))),
+            default_factory: DefaultParquetFileReaderFactory::new(Arc::new(
+                 object_store::local::LocalFileSystem::new(),
+            )),
         })
     }
 }
@@ -52,8 +50,9 @@ impl ParquetFileReaderFactory for NoopParquetMetadataCache {
     ) -> datafusion::common::Result<Box<dyn AsyncFileReader + Send>> {
         self.default_factory
             .create_reader(partition_index, file_meta, metadata_size_hint, metrics)
-    }
-}
+     }
+ }
+
 
 /// LruMetadataCache, caches parquet metadata.
 pub struct LruParquetMetadataCacheFactory {
@@ -115,9 +114,7 @@ impl BasicMetadataCacheFactory {
 
 impl MetadataCacheFactory for BasicMetadataCacheFactory {
     fn make_noop_cache(&self) -> Arc<dyn ParquetFileReaderFactory> {
-        Arc::new(DefaultParquetFileReaderFactory::new(Arc::new(
-            object_store::local::LocalFileSystem::new(),
-        )))
+        NoopParquetMetadataCache::new()
     }
 
     fn make_lru_cache(

rust/cubestore/cubestore/src/queryplanner/planning.rs

@@ -50,8 +50,7 @@ use crate::queryplanner::topk::ClusterAggregateTopK;
 use crate::queryplanner::{CubeTableLogical, InfoSchemaTableProvider};
 use crate::table::{cmp_same_types, Row};
 use crate::CubeError;
-// use datafusion::physical_plan::parquet::NoopParquetMetadataCache;
-use crate::queryplanner::metadata_cache::{MetadataCacheFactory, NoopParquetMetadataCache};
+use crate::queryplanner::metadata_cache::NoopParquetMetadataCache;
 use datafusion::common;
 use datafusion::common::DFSchemaRef;
 use datafusion::datasource::DefaultTableSource;

rust/cubestore/cubestore/src/store/compaction.rs

@@ -1464,9 +1464,7 @@ mod tests {
     use crate::metastore::{
         BaseRocksStoreFs, Column, ColumnType, IndexDef, IndexType, RocksMetaStore,
     };
-    use crate::queryplanner::metadata_cache::{
-        BasicMetadataCacheFactory, NoopParquetMetadataCache,
-    };
+    use crate::queryplanner::metadata_cache::BasicMetadataCacheFactory;
     use crate::remotefs::LocalDirRemoteFs;
     use crate::store::MockChunkDataStore;
     use crate::table::data::rows_to_columns;

rust/cubestore/cubestore/src/streaming/kafka.rs

@@ -2,7 +2,6 @@ use crate::config::injection::DIService;
 use crate::config::ConfigObj;
 use crate::metastore::table::StreamOffset;
 use crate::metastore::Column;
-use crate::queryplanner::metadata_cache::MetadataCacheFactory;
 use crate::streaming::kafka_post_processing::{KafkaPostProcessPlan, KafkaPostProcessPlanner};
 use crate::streaming::traffic_sender::TrafficSender;
 use crate::streaming::{parse_json_payload_and_key, StreamingSource};
@@ -12,7 +11,6 @@ use async_std::stream;
 use async_trait::async_trait;
 use datafusion::arrow::array::ArrayRef;
 use datafusion::cube_ext;
-use datafusion::datasource::physical_plan::ParquetFileReaderFactory;
 use futures::Stream;
 use json::object::Object;
 use json::JsonValue;
@@ -61,7 +59,6 @@ impl KafkaStreamingSource {
         kafka_client: Arc<dyn KafkaClientService>,
         use_ssl: bool,
         trace_obj: Option<String>,
-        metadata_cache_factory: Arc<dyn MetadataCacheFactory>,
     ) -> Result<Self, CubeError> {
         let (post_processing_plan, columns, unique_key_columns, seq_column_index) =
             if let Some(select_statement) = select_statement {
@@ -73,7 +70,7 @@ impl KafkaStreamingSource {
                     source_columns,
                 );
                 let plan = planner
-                    .build(select_statement.clone(), metadata_cache_factory)
+                    .build(select_statement.clone())
                     .await?;
                 let columns = plan.source_columns().clone();
                 let seq_column_index = plan.source_seq_column_index();

rust/cubestore/cubestore/src/streaming/kafka_post_processing.rs

@@ -1,5 +1,4 @@
 use crate::metastore::Column;
-use crate::queryplanner::metadata_cache::MetadataCacheFactory;
 use crate::sql::MySqlDialectWithBackTicks;
 use crate::streaming::topic_table_provider::TopicTableProvider;
 use crate::CubeError;
@@ -9,7 +8,6 @@ use datafusion::arrow::datatypes::{Field, Schema, SchemaRef};
 use datafusion::arrow::record_batch::RecordBatch;
 use datafusion::common;
 use datafusion::common::{DFSchema, DFSchemaRef};
-use datafusion::datasource::physical_plan::ParquetFileReaderFactory;
 use datafusion::execution::TaskContext;
 use datafusion::logical_expr::expr::Alias;
 use datafusion::logical_expr::{Expr, Filter, LogicalPlan, Projection};
@@ -138,7 +136,6 @@ impl KafkaPostProcessPlanner {
     pub async fn build(
         &self,
         select_statement: String,
-        metadata_cache_factory: Arc<dyn MetadataCacheFactory>,
     ) -> Result<KafkaPostProcessPlan, CubeError> {
         let target_schema = Arc::new(Schema::new(
             self.columns
@@ -150,7 +147,7 @@ impl KafkaPostProcessPlanner {
         let source_unique_columns = self.extract_source_unique_columns(&logical_plan)?;
 
         let (projection_plan, filter_plan) = self
-            .make_projection_and_filter_physical_plans(&logical_plan, metadata_cache_factory)
+            .make_projection_and_filter_physical_plans(&logical_plan)
             .await?;
         if target_schema != projection_plan.schema() {
             return Err(CubeError::user(format!(
@@ -406,7 +403,6 @@ impl KafkaPostProcessPlanner {
     async fn make_projection_and_filter_physical_plans(
         &self,
         plan: &LogicalPlan,
-        metadata_cache_factory: Arc<dyn MetadataCacheFactory>,
     ) -> Result<(Arc<dyn ExecutionPlan>, Option<Arc<dyn ExecutionPlan>>), CubeError> {
         let source_schema = Arc::new(Schema::new(
             self.source_columns

rust/cubestore/cubestore/src/streaming/mod.rs

@@ -11,7 +11,6 @@ use crate::metastore::replay_handle::{ReplayHandle, SeqPointer, SeqPointerForLoc
 use crate::metastore::source::SourceCredentials;
 use crate::metastore::table::{StreamOffset, Table};
 use crate::metastore::{Column, ColumnType, IdRow, MetaStore};
-use crate::queryplanner::metadata_cache::MetadataCacheFactory;
 use crate::sql::timestamp_from_string;
 use crate::store::ChunkDataStore;
 use crate::streaming::kafka::{KafkaClientService, KafkaStreamingSource};
@@ -24,7 +23,6 @@ use buffered_stream::BufferedStream;
 use chrono::Utc;
 use datafusion::arrow::array::ArrayBuilder;
 use datafusion::arrow::array::ArrayRef;
-use datafusion::datasource::physical_plan::ParquetFileReaderFactory;
 use futures::future::join_all;
 use futures::stream::StreamExt;
 use futures::Stream;
@@ -59,7 +57,6 @@ pub struct StreamingServiceImpl {
     chunk_store: Arc<dyn ChunkDataStore>,
     ksql_client: Arc<dyn KsqlClient>,
     kafka_client: Arc<dyn KafkaClientService>,
-    metadata_cache_factory: Arc<dyn MetadataCacheFactory>,
 }
 
 crate::di_service!(StreamingServiceImpl, [StreamingService]);
@@ -71,15 +68,13 @@ impl StreamingServiceImpl {
         chunk_store: Arc<dyn ChunkDataStore>,
         ksql_client: Arc<dyn KsqlClient>,
         kafka_client: Arc<dyn KafkaClientService>,
-        metadata_cache_factory: Arc<dyn MetadataCacheFactory>,
     ) -> Arc<Self> {
         Arc::new(Self {
             config_obj,
             meta_store,
             chunk_store,
             ksql_client,
             kafka_client,
-            metadata_cache_factory,
         })
     }
 
@@ -170,7 +165,6 @@ impl StreamingServiceImpl {
                 self.kafka_client.clone(),
                 *use_ssl,
                 trace_obj,
-                self.metadata_cache_factory.clone(),
             ).await?)),
         }
     }