Reorg state and cfg some

Author: matthewmturner

Cargo.lock

@@ -17,6 +17,7 @@ version = "0.2.2"
 arrow-flight = { version = "54.1.0", features = [
   "flight-sql-experimental",
 ], optional = true }
+axum = { version = "0.8.1", optional = true }
 clap = { version = "4.5.27", features = ["derive"] }
 color-eyre = "0.6.3"
 crossterm = { version = "0.28.1", features = ["event-stream"] }
@@ -74,6 +75,7 @@ flightsql = [
 ]
 functions-json = ["datafusion-app/functions-json"]
 functions-parquet = ["datafusion-app/functions-parquet"]
+http = ["axum"]
 hudi = ["datafusion-app/hudi"]
 huggingface = ["datafusion-app/huggingface"]
 iceberg = ["datafusion-app/iceberg"]

Cargo.toml

@@ -120,6 +120,12 @@ impl DftArgs {
 
 #[derive(Clone, Debug, Subcommand)]
 pub enum Command {
+    /// Start a HTTP server
+    ServeHttp {
+        #[clap(short, long)]
+        config: Option<String>,
+    },
+
     /// Start a FlightSQL server
     ServeFlightSql {
         #[clap(short, long)]

src/args.rs

@@ -22,6 +22,7 @@ use std::path::PathBuf;
 use datafusion_app::config::ExecutionConfig;
 use directories::{ProjectDirs, UserDirs};
 use lazy_static::lazy_static;
+use log::{debug, error};
 use serde::Deserialize;
 
 #[cfg(feature = "flightsql")]
@@ -188,28 +189,6 @@ fn default_paste() -> bool {
     false
 }
 
-// #[cfg(feature = "flightsql")]
-// #[derive(Clone, Debug, Deserialize)]
-// pub struct FlightSQLConfig {
-//     #[serde(default = "default_connection_url")]
-//     pub connection_url: String,
-//     #[serde(default = "default_benchmark_iterations")]
-//     pub benchmark_iterations: usize,
-//     #[serde(default = "default_server_metrics_port")]
-//     pub server_metrics_port: String,
-// }
-//
-// #[cfg(feature = "flightsql")]
-// impl Default for FlightSQLConfig {
-//     fn default() -> Self {
-//         Self {
-//             connection_url: default_connection_url(),
-//             benchmark_iterations: default_benchmark_iterations(),
-//             server_metrics_port: default_server_metrics_port(),
-//         }
-//     }
-// }
-
 #[cfg(feature = "flightsql")]
 pub fn default_connection_url() -> String {
     "http://localhost:50051".to_string()
@@ -233,3 +212,29 @@ fn default_editor_config() -> EditorConfig {
 fn default_auth_config() -> AuthConfig {
     AuthConfig::default()
 }
+
+pub fn create_config(config_path: PathBuf) -> AppConfig {
+    if config_path.exists() {
+        debug!("Config exists");
+        let maybe_config_contents = std::fs::read_to_string(config_path);
+        if let Ok(config_contents) = maybe_config_contents {
+            let maybe_parsed_config: std::result::Result<AppConfig, toml::de::Error> =
+                toml::from_str(&config_contents);
+            match maybe_parsed_config {
+                Ok(parsed_config) => {
+                    debug!("Parsed config: {:?}", parsed_config);
+                    parsed_config
+                }
+                Err(err) => {
+                    error!("Error parsing config: {:?}", err);
+                    AppConfig::default()
+                }
+            }
+        } else {
+            AppConfig::default()
+        }
+    } else {
+        debug!("No config, using default");
+        AppConfig::default()
+    }
+}

src/config.rs

@@ -195,3 +195,5 @@ impl FlightSqlApp {
         }
     }
 }
+
+// pub fn run_flightsql_server(state: AppState) {}

src/server/mod.rs src/flightsql_server/mod.rssrc/server/mod.rs renamed to src/flightsql_server/mod.rs

@@ -3,7 +3,7 @@ pub mod cli;
 pub mod config;
 pub mod execution;
 #[cfg(feature = "flightsql")]
-pub mod server;
+pub mod flightsql_server;
 pub mod telemetry;
 pub mod test_utils;
 pub mod tui;

src/server/services/flightsql.rs …c/flightsql_server/services/flightsql.rssrc/server/services/flightsql.rs renamed to src/flightsql_server/services/flightsql.rs src/server/services/flightsql.rs renamed to src/flightsql_server/services/flightsql.rs

@@ -20,18 +20,15 @@ use color_eyre::Result;
 use datafusion_app::local::ExecutionContext;
 use datafusion_app::{config::merge_configs, extensions::DftSessionStateBuilder};
 use datafusion_dft::args::Command;
+use datafusion_dft::tui::state::AppState;
 use datafusion_dft::{
-    args::DftArgs,
-    cli::CliApp,
-    execution::AppExecution,
-    telemetry,
-    tui::{state, App},
+    args::DftArgs, cli::CliApp, config::create_config, execution::AppExecution, telemetry, tui::App,
 };
 #[cfg(feature = "flightsql")]
 use {
     datafusion_app::config::{AuthConfig, FlightSQLConfig},
     datafusion_app::flightsql::FlightSQLContext,
-    datafusion_dft::server::FlightSqlApp,
+    datafusion_dft::flightsql_server::FlightSqlApp,
     log::info,
 };
 
@@ -66,44 +63,37 @@ async fn app_entry_point(cli: DftArgs) -> Result<()> {
     if should_init_env_logger(&cli) {
         env_logger::init();
     }
-    let state = state::initialize(cli.config_path());
+    let cfg = create_config(cli.config_path());
     #[cfg(feature = "flightsql")]
     if let Some(Command::ServeFlightSql { .. }) = cli.command {
-        let merged_exec_config = merge_configs(
-            state.config.shared.clone(),
-            state.config.flightsql_server.execution.clone(),
-        );
+        let merged_exec_config =
+            merge_configs(cfg.shared.clone(), cfg.flightsql_server.execution.clone());
         let session_state_builder =
             DftSessionStateBuilder::try_new(Some(merged_exec_config.clone()))?
                 .with_extensions()
                 .await?;
         // FlightSQL Server mode: start a FlightSQL server
         const DEFAULT_SERVER_ADDRESS: &str = "127.0.0.1:50051";
         info!("Starting FlightSQL server on {}", DEFAULT_SERVER_ADDRESS);
-        let session_state = session_state_builder
-            // .with_app_type(AppType::FlightSQLServer)
-            .build()?;
+        let session_state = session_state_builder.build()?;
         let execution_ctx = ExecutionContext::try_new(&merged_exec_config, session_state)?;
         if cli.run_ddl {
             execution_ctx.execute_ddl().await;
         }
         let app_execution = AppExecution::new(execution_ctx);
         let app = FlightSqlApp::try_new(
             app_execution,
-            &state.config,
+            &cfg,
             &cli.flightsql_host
                 .unwrap_or(DEFAULT_SERVER_ADDRESS.to_string()),
-            &state.config.flightsql_server.server_metrics_port,
+            &cfg.flightsql_server.server_metrics_port,
         )
         .await?;
         app.run_app().await;
         return Ok(());
     }
     if !cli.files.is_empty() || !cli.commands.is_empty() {
-        let merged_exec_config = merge_configs(
-            state.config.shared.clone(),
-            state.config.cli.execution.clone(),
-        );
+        let merged_exec_config = merge_configs(cfg.shared.clone(), cfg.cli.execution.clone());
         let session_state_builder =
             DftSessionStateBuilder::try_new(Some(merged_exec_config.clone()))?
                 .with_extensions()
@@ -118,12 +108,12 @@ async fn app_entry_point(cli: DftArgs) -> Result<()> {
         {
             if cli.flightsql {
                 let auth = AuthConfig {
-                    basic_auth: state.config.flightsql_client.auth.basic_auth,
-                    bearer_token: state.config.flightsql_client.auth.bearer_token,
+                    basic_auth: cfg.flightsql_client.auth.basic_auth,
+                    bearer_token: cfg.flightsql_client.auth.bearer_token,
                 };
                 let flightsql_cfg = FlightSQLConfig::new(
-                    state.config.flightsql_client.connection_url,
-                    state.config.flightsql_client.benchmark_iterations,
+                    cfg.flightsql_client.connection_url,
+                    cfg.flightsql_client.benchmark_iterations,
                     auth,
                 );
                 let flightsql_ctx = FlightSQLContext::new(flightsql_cfg);
@@ -136,10 +126,7 @@ async fn app_entry_point(cli: DftArgs) -> Result<()> {
         let app = CliApp::new(app_execution, cli.clone());
         app.execute_files_or_commands().await?;
     } else {
-        let merged_exec_config = merge_configs(
-            state.config.shared.clone(),
-            state.config.tui.execution.clone(),
-        );
+        let merged_exec_config = merge_configs(cfg.shared.clone(), cfg.tui.execution.clone());
         let session_state_builder =
             DftSessionStateBuilder::try_new(Some(merged_exec_config.clone()))?
                 .with_extensions()
@@ -148,7 +135,7 @@ async fn app_entry_point(cli: DftArgs) -> Result<()> {
 
         // TUI mode: running the TUI
         telemetry::initialize_logs()?; // use alternate logging for TUI
-        let state = state::initialize(cli.config_path());
+        let state = AppState::new(cfg);
         let execution_ctx = ExecutionContext::try_new(&merged_exec_config, session_state)?;
         let app_execution = AppExecution::new(execution_ctx);
         let app = App::new(state, cli, app_execution);