refactor(cubesql): Transport - hide cubeclient with own types (#8582)

Author: ovr

packages/cubejs-backend-native/Cargo.lock

@@ -21,7 +21,6 @@ cubenativeutils = { path = "../../rust/cubenativeutils" }
 async-channel = { version = "2" }
 async-trait = "0.1.36"
 convert_case = "0.6.0"
-cubeclient = { path = "../../rust/cubesql/cubeclient" }
 pin-project = "1.1.5"
 cubesql = { path = "../../rust/cubesql/cubesql" }
 findshlibs = "0.10.2"

packages/cubejs-backend-native/Cargo.toml

@@ -4,10 +4,12 @@ use std::collections::HashMap;
 use std::fmt::Display;
 
 use async_trait::async_trait;
-use cubeclient::models::{V1Error, V1LoadRequestQuery, V1LoadResponse, V1MetaResponse};
 use cubesql::compile::engine::df::scan::{MemberField, SchemaRef};
 use cubesql::compile::engine::df::wrapper::SqlQuery;
-use cubesql::transport::{SpanId, SqlGenerator, SqlResponse};
+use cubesql::transport::{
+    SpanId, SqlGenerator, SqlResponse, TransportError, TransportLoadRequestQuery,
+    TransportLoadResponse, TransportMetaResponse,
+};
 use cubesql::{
     di_service,
     sql::AuthContextRef,
@@ -76,7 +78,7 @@ struct CanSwitchUserForSessionRequest {
 #[derive(Debug, Serialize)]
 struct LoadRequest {
     request: TransportRequest,
-    query: V1LoadRequestQuery,
+    query: TransportLoadRequestQuery,
     #[serde(rename = "sqlQuery", skip_serializing_if = "Option::is_none")]
     sql_query: Option<(String, Vec<Option<String>>)>,
     session: SessionContext,
@@ -129,7 +131,7 @@ impl TransportService for NodeBridgeTransport {
             only_compiler_id: false,
         })?;
 
-        let response = call_js_with_channel_as_callback::<V1MetaResponse>(
+        let response = call_js_with_channel_as_callback::<TransportMetaResponse>(
             self.channel.clone(),
             self.on_meta.clone(),
             Some(extra.clone()),
@@ -229,7 +231,7 @@ impl TransportService for NodeBridgeTransport {
             },
             only_compiler_id: true,
         })?;
-        let response = call_js_with_channel_as_callback::<V1MetaResponse>(
+        let response = call_js_with_channel_as_callback::<TransportMetaResponse>(
             self.channel.clone(),
             self.on_meta.clone(),
             Some(extra.clone()),
@@ -251,7 +253,7 @@ impl TransportService for NodeBridgeTransport {
     async fn sql(
         &self,
         span_id: Option<Arc<SpanId>>,
-        query: V1LoadRequestQuery,
+        query: TransportLoadRequestQuery,
         ctx: AuthContextRef,
         meta: LoadRequestMeta,
         member_to_alias: Option<HashMap<String, String>>,
@@ -328,11 +330,11 @@ impl TransportService for NodeBridgeTransport {
     async fn load(
         &self,
         span_id: Option<Arc<SpanId>>,
-        query: V1LoadRequestQuery,
+        query: TransportLoadRequestQuery,
         sql_query: Option<SqlQuery>,
         ctx: AuthContextRef,
         meta: LoadRequestMeta,
-    ) -> Result<V1LoadResponse, CubeError> {
+    ) -> Result<TransportLoadResponse, CubeError> {
         trace!("[transport] Request ->");
 
         let native_auth = ctx
@@ -381,14 +383,14 @@ impl TransportService for NodeBridgeTransport {
             #[cfg(not(debug_assertions))]
             trace!("[transport] Request <- <hidden>");
 
-            let load_err = match serde_json::from_value::<V1LoadResponse>(response.clone()) {
+            let load_err = match serde_json::from_value::<TransportLoadResponse>(response.clone()) {
                 Ok(r) => {
                     return Ok(r);
                 }
                 Err(err) => err,
             };
 
-            if let Ok(res) = serde_json::from_value::<V1Error>(response) {
+            if let Ok(res) = serde_json::from_value::<TransportError>(response) {
                 if res.error.to_lowercase() == *"continue wait" {
                     debug!(
                         "[transport] load - retrying request (continue wait) requestId: {}",
@@ -413,7 +415,7 @@ impl TransportService for NodeBridgeTransport {
     async fn load_stream(
         &self,
         span_id: Option<Arc<SpanId>>,
-        query: V1LoadRequestQuery,
+        query: TransportLoadRequestQuery,
         sql_query: Option<SqlQuery>,
         ctx: AuthContextRef,
         meta: LoadRequestMeta,

packages/cubejs-backend-native/src/transport.rs

@@ -1,7 +1,6 @@
 use std::{any::Any, collections::HashMap, sync::Arc};
 
 use async_trait::async_trait;
-use cubeclient::models::V1CubeMeta;
 use datafusion::{
     arrow::datatypes::{DataType, Field, Schema, SchemaRef, TimeUnit},
     datasource::{self, TableProvider},
@@ -15,7 +14,7 @@ use datafusion::{
 use crate::{
     compile::{DatabaseProtocolDetails, MetaContext},
     sql::{ColumnType, SessionManager, SessionState},
-    transport::V1CubeMetaExt,
+    transport::{CubeMeta, V1CubeMetaExt},
     CubeError,
 };
 
@@ -92,11 +91,11 @@ pub trait TableName {
 }
 
 pub struct CubeTableProvider {
-    cube: V1CubeMeta,
+    cube: CubeMeta,
 }
 
 impl CubeTableProvider {
-    pub fn new(cube: V1CubeMeta) -> Self {
+    pub fn new(cube: CubeMeta) -> Self {
         Self { cube }
     }
 }

rust/cubesql/cubesql/src/compile/engine/context.rs

@@ -14,12 +14,11 @@ use crate::{
     sql::AuthContextRef,
     transport::{
         AliasedColumn, LoadRequestMeta, MetaContext, SpanId, SqlGenerator, SqlTemplates,
-        TransportService,
+        TransportLoadRequestQuery, TransportService,
     },
     CubeError,
 };
 use chrono::{Days, NaiveDate, SecondsFormat, TimeZone, Utc};
-use cubeclient::models::V1LoadRequestQuery;
 use datafusion::{
     error::{DataFusionError, Result},
     logical_plan::{
@@ -200,7 +199,7 @@ pub struct CubeScanWrapperNode {
     pub meta: Arc<MetaContext>,
     pub auth_context: AuthContextRef,
     pub wrapped_sql: Option<SqlQuery>,
-    pub request: Option<V1LoadRequestQuery>,
+    pub request: Option<TransportLoadRequestQuery>,
     pub member_fields: Option<Vec<MemberField>>,
     pub span_id: Option<Arc<SpanId>>,
     pub config_obj: Arc<dyn ConfigObj>,
@@ -229,7 +228,7 @@ impl CubeScanWrapperNode {
     pub fn with_sql_and_request(
         &self,
         sql: SqlQuery,
-        request: V1LoadRequestQuery,
+        request: TransportLoadRequestQuery,
         member_fields: Vec<MemberField>,
     ) -> Self {
         Self {
@@ -258,7 +257,7 @@ pub struct SqlGenerationResult {
     pub from_alias: Option<String>,
     pub column_remapping: Option<HashMap<Column, Column>>,
     pub sql: SqlQuery,
-    pub request: V1LoadRequestQuery,
+    pub request: TransportLoadRequestQuery,
 }
 
 lazy_static! {
@@ -925,7 +924,7 @@ impl CubeScanWrapperNode {
                     from_alias: None,
                     sql: SqlQuery::new("".to_string(), values.clone()),
                     column_remapping: None,
-                    request: V1LoadRequestQuery::new(),
+                    request: TransportLoadRequestQuery::new(),
                 }),
                 // LogicalPlan::Distinct(_) => {}
                 x => {

rust/cubesql/cubesql/src/compile/engine/df/wrapper.rs

@@ -1,7 +1,6 @@
 use std::{any::Any, sync::Arc};
 
 use async_trait::async_trait;
-use cubeclient::models::V1CubeMeta;
 use datafusion::{
     arrow::{
         array::{Array, ArrayRef, StringBuilder, UInt32Builder},
@@ -14,7 +13,7 @@ use datafusion::{
     physical_plan::{memory::MemoryExec, ExecutionPlan},
 };
 
-use crate::transport::{CubeColumn, V1CubeMetaExt};
+use crate::transport::{CubeColumn, CubeMeta, V1CubeMetaExt};
 
 use super::{
     ext::CubeColumnPostgresExt,
@@ -312,7 +311,7 @@ pub struct InfoSchemaColumnsProvider {
 }
 
 impl InfoSchemaColumnsProvider {
-    pub fn new(db_name: &str, cubes: &Vec<V1CubeMeta>) -> Self {
+    pub fn new(db_name: &str, cubes: &Vec<CubeMeta>) -> Self {
         let mut builder = InformationSchemaColumnsBuilder::new();
 
         for cube in cubes {

rust/cubesql/cubesql/src/compile/engine/information_schema/postgres/columns.rs

@@ -1,7 +1,7 @@
 use std::{any::Any, sync::Arc};
 
+use crate::transport::CubeMeta;
 use async_trait::async_trait;
-use cubeclient::models::V1CubeMeta;
 use datafusion::{
     arrow::{
         array::{Array, ArrayRef, StringBuilder},
@@ -68,7 +68,7 @@ pub struct RedshiftSvvTablesTableProvider {
 }
 
 impl RedshiftSvvTablesTableProvider {
-    pub fn new(db_name: &str, cubes: &Vec<V1CubeMeta>) -> Self {
+    pub fn new(db_name: &str, cubes: &Vec<CubeMeta>) -> Self {
         let mut builder = RedshiftSvvTablesBuilder::new(cubes.len());
 
         for cube in cubes {