Implement configurable PgWire authentication

- Add PGWIRE_USER and PGWIRE_PASSWORD config options with defaults
- Replace NoopStartupHandler with CleartextPasswordAuthStartupHandler
- Remove unused auth_manager parameters from handler constructors
- Extract classify_query() and sanitize_query() helpers to reduce duplication
- Update all .env files with postgres/postgres credentials for backward compatibility

Author: tonyalaribe

.env.example

@@ -11,6 +11,8 @@ AWS_S3_BUCKET=
 AWS_ACCESS_KEY_ID=
 AWS_SECRET_ACCESS_KEY=
 PGWIRE_PORT=5432
+PGWIRE_USER=postgres
+PGWIRE_PASSWORD=postgres
 TIMEFUSION_TABLE_PREFIX=timefusion
 
 # Delta Lake DynamoDB Locking Configuration (optional but recommended for multi-writer scenarios)

.env.minio

@@ -8,6 +8,8 @@ AWS_ALLOW_HTTP=true
 AWS_ACCESS_KEY_ID=minioadmin
 AWS_SECRET_ACCESS_KEY=minioadmin
 PGWIRE_PORT=12345
+PGWIRE_USER=postgres
+PGWIRE_PASSWORD=postgres
 PORT=80
 
 TIMEFUSION_TABLE_PREFIX=timefusion-minio-test

.env.test

@@ -13,6 +13,8 @@ AWS_SDK_LOAD_CONFIG=false
 
 # PostgreSQL Wire Protocol Configuration
 PGWIRE_PORT=12345
+PGWIRE_USER=postgres
+PGWIRE_PASSWORD=postgres
 TIMEFUSION_TABLE_PREFIX=test
 
 # No DynamoDB locking for tests (uses local file-based locking)

src/config.rs

@@ -93,6 +93,7 @@ const_default!(d_wal_dir: PathBuf = "/var/lib/timefusion/wal");
 const_default!(d_pgwire_port: u16 = 5432);
 const_default!(d_table_prefix: String = "timefusion");
 const_default!(d_batch_queue_capacity: usize = 100_000_000);
+const_default!(d_pgwire_user: String = "postgres");
 const_default!(d_flush_interval: u64 = 600);
 const_default!(d_retention_mins: u64 = 70);
 const_default!(d_eviction_interval: u64 = 60);
@@ -230,6 +231,10 @@ pub struct CoreConfig {
     pub enable_batch_queue: bool,
     #[serde(default = "d_batch_queue_capacity")]
     pub timefusion_batch_queue_capacity: usize,
+    #[serde(default = "d_pgwire_user")]
+    pub pgwire_user: String,
+    #[serde(default)]
+    pub pgwire_password: Option<String>,
 }
 
 #[derive(Debug, Clone, Deserialize)]

src/main.rs

@@ -1,7 +1,7 @@
 // main.rs
 #![recursion_limit = "512"]
 
-use datafusion_postgres::{ServerOptions, auth::AuthManager};
+use datafusion_postgres::ServerOptions;
 use dotenv::dotenv;
 use std::sync::Arc;
 use timefusion::buffered_write_layer::BufferedWriteLayer;
@@ -85,12 +85,15 @@ async fn async_main(cfg: &'static AppConfig) -> anyhow::Result<()> {
     let pg_port = cfg.core.pgwire_port;
     info!("Starting PGWire server on port: {}", pg_port);
 
+    let auth_config = timefusion::pgwire_handlers::AuthConfig {
+        username: cfg.core.pgwire_user.clone(),
+        password: cfg.core.pgwire_password.clone(),
+    };
+
     let pg_task = tokio::spawn(async move {
         let opts = ServerOptions::new().with_port(pg_port).with_host("0.0.0.0".to_string());
-        let auth_manager = Arc::new(AuthManager::new());
 
-        // Use our custom handlers that log UPDATE queries
-        if let Err(e) = timefusion::pgwire_handlers::serve_with_logging(Arc::new(session_context), &opts, auth_manager).await {
+        if let Err(e) = timefusion::pgwire_handlers::serve_with_logging(Arc::new(session_context), &opts, auth_config).await {
             error!("PGWire server error: {}", e);
         }
     });