Merge remote-tracking branch 'origin' into kafka-integration

src/cli.rs

@@ -27,7 +27,6 @@ use crate::connectors::common::types::ConnectorType;
 use crate::connectors::common::BadData;
 use crate::connectors::kafka::config::{ConsumerConfig, KafkaConfig, SourceOffset};
 use crate::{
-    kafka::SslProtocol,
     oidc::{self, OpenidConfig},
     option::{validation, Compression, Mode},
 };
@@ -51,9 +50,15 @@ pub struct Cli {
     pub domain_address: Option<Url>,
 
     /// The local staging path is used as a temporary landing point
-    /// for incoming events
+    /// for incoming events and local cache
     pub local_staging_path: PathBuf,
 
+    /// The local cache path is used for speeding up query on latest data
+    pub local_cache_path: Option<PathBuf>,
+
+    /// Size for local cache
+    pub local_cache_size: u64,
+
     /// Username for the basic authentication on the server
     pub username: String,
 
@@ -96,6 +101,12 @@ pub struct Cli {
     /// port use by airplane(flight query service)
     pub flight_port: u16,
 
+    /// to query cached data
+    pub query_cache_path: Option<PathBuf>,
+
+    /// Size for local cache
+    pub query_cache_size: u64,
+
     /// CORS behaviour
     pub cors: bool,
 
@@ -125,6 +136,10 @@ impl Cli {
     pub const ADDRESS: &'static str = "address";
     pub const DOMAIN_URI: &'static str = "origin";
     pub const STAGING: &'static str = "local-staging-path";
+    pub const CACHE: &'static str = "cache-path";
+    pub const QUERY_CACHE: &'static str = "query-cache-path";
+    pub const QUERY_CACHE_SIZE: &'static str = "query-cache-size";
+    pub const CACHE_SIZE: &'static str = "cache-size";
     pub const USERNAME: &'static str = "username";
     pub const PASSWORD: &'static str = "password";
     pub const CHECK_UPDATE: &'static str = "check-update";
@@ -891,13 +906,8 @@ impl FromArgMatches for Cli {
         self.trino_schema = m.get_one::<String>(Self::TRINO_SCHEMA).cloned();
         self.trino_username = m.get_one::<String>(Self::TRINO_USER_NAME).cloned();
 
-        self.kafka_topics = m.get_one::<String>(Self::KAFKA_TOPICS).cloned();
-        self.kafka_host = m.get_one::<String>(Self::KAFKA_HOST).cloned();
-        self.kafka_group = m.get_one::<String>(Self::KAFKA_GROUP).cloned();
-        self.kafka_client_id = m.get_one::<String>(Self::KAFKA_CLIENT_ID).cloned();
-        self.kafka_security_protocol = m.get_one::<SslProtocol>(Self::KAFKA_SECURITY_PROTOCOL).cloned();
-        self.kafka_partitions = m.get_one::<String>(Self::KAFKA_PARTITIONS).cloned();
-
+        self.local_cache_path = m.get_one::<PathBuf>(Self::CACHE).cloned();
+        self.query_cache_path = m.get_one::<PathBuf>(Self::QUERY_CACHE).cloned();
         self.tls_cert_path = m.get_one::<PathBuf>(Self::TLS_CERT).cloned();
         self.tls_key_path = m.get_one::<PathBuf>(Self::TLS_KEY).cloned();
         self.trusted_ca_certs_path = m.get_one::<PathBuf>(Self::TRUSTED_CA_CERTS_PATH).cloned();
@@ -917,6 +927,14 @@ impl FromArgMatches for Cli {
             .get_one::<PathBuf>(Self::STAGING)
             .cloned()
             .expect("default value for staging");
+        self.local_cache_size = m
+            .get_one::<u64>(Self::CACHE_SIZE)
+            .cloned()
+            .expect("default value for cache size");
+        self.query_cache_size = m
+            .get_one(Self::QUERY_CACHE_SIZE)
+            .cloned()
+            .expect("default value for query cache size");
         self.username = m
             .get_one::<String>(Self::USERNAME)
             .cloned()
@@ -959,12 +977,20 @@ impl FromArgMatches for Cli {
             .get_one::<usize>(Self::ROW_GROUP_SIZE)
             .cloned()
             .expect("default for row_group size");
-        self.parquet_compression = serde_json::from_str(&format!(
-            "{:?}",
-            m.get_one::<String>(Self::PARQUET_COMPRESSION_ALGO)
-                .expect("default for compression algo")
-        ))
-        .expect("unexpected compression algo");
+        self.parquet_compression = match m
+            .get_one::<String>(Self::PARQUET_COMPRESSION_ALGO)
+            .expect("default for compression algo")
+            .as_str()
+        {
+            "uncompressed" => Compression::UNCOMPRESSED,
+            "snappy" => Compression::SNAPPY,
+            "gzip" => Compression::GZIP,
+            "lzo" => Compression::LZO,
+            "brotli" => Compression::BROTLI,
+            "lz4" => Compression::LZ4,
+            "zstd" => Compression::ZSTD,
+            _ => unreachable!(),
+        };
 
         let openid_client_id = m.get_one::<String>(Self::OPENID_CLIENT_ID).cloned();
         let openid_client_secret = m.get_one::<String>(Self::OPENID_CLIENT_SECRET).cloned();