refactor: extract establish_session, djb2_oid, json_str helpers

Three small dedup passes that came out of the human review:

- `startup.rs`: the auth=true and auth=false branches both ran the
  same post-Trino-client session setup (allocate pid+secret, register
  cancel, register connection, finish_authentication). Pull the
  shared tail into `establish_session(client, trino_client, user)`.
  The one branch difference (logging "client connected" vs "client
  connected (no auth)") is captured by the `Option<&str>` user
  parameter.
- `catalog/pg_class.rs`: `table_oid` and `namespace_oid` both used
  djb2 inline. Extract `djb2_oid(parts: &[&str], base: u32) -> u32`
  and call from both. Doc-comment notes why we use djb2 rather than
  `std::hash::DefaultHasher` (DefaultHasher's algorithm is allowed
  to change across Rust releases; OIDs need to be stable across
  reconnects).
- `catalog/mod.rs`: introduce `json_str(values, idx, default)`
  helper, used from both `pg_class::respond_pg_class` and
  `pg_attribute::respond_pg_attribute` to keep row-decode boilerplate
  to one line per column.

No behavioural change. All 191 unit tests still pass; clippy clean
under `--all-targets -- -D warnings`.

License headers migrated to inline SPDX along the way.

Author: lfrancke

src/catalog/pg_attribute.rs

@@ -1,6 +1,5 @@
-// Copyright 2026 Stackable GmbH
-// Licensed under the Open Software License version 3.0 (OSL-3.0).
-// See LICENSE file in the project root for full license text.
+// SPDX-FileCopyrightText: 2026 Stackable GmbH
+// SPDX-License-Identifier: OSL-3.0
 use std::sync::Arc;
 
 use pgwire::api::Type;
@@ -9,7 +8,7 @@ use pgwire::error::{PgWireError, PgWireResult};
 use trino_rust_client::{Client, Row};
 
 use super::pg_class::table_oid;
-use super::{build_response, text_field};
+use super::{build_response, json_str, text_field};
 use crate::types::trino_type_to_pg;
 
 /// Query Trino's information_schema.columns and return a pg_attribute-compatible response.
@@ -41,18 +40,21 @@ pub async fn respond_pg_attribute(client: &Arc<Client>) -> PgWireResult<Vec<Resp
         let values = trino_row.into_json();
         // columns: table_schema (0), table_name (1), column_name (2),
         //          ordinal_position (3), is_nullable (4), data_type (5)
-        let schema_name = values.first().and_then(|v| v.as_str()).unwrap_or("public");
-        let tbl_name = values.get(1).and_then(|v| v.as_str()).unwrap_or("");
-        let col_name = values.get(2).and_then(|v| v.as_str()).unwrap_or("");
+        let schema_name = json_str(&values, 0, "public");
+        let tbl_name = json_str(&values, 1, "");
+        let col_name = json_str(&values, 2, "");
+        // Trino reports ordinal_position as a JSON number, but some
+        // information_schema bridges (e.g. legacy connectors) stringify it.
+        // Accept either form.
         let ordinal = values
             .get(3)
             .and_then(|v| {
                 v.as_i64()
                     .or_else(|| v.as_str().and_then(|s| s.parse().ok()))
             })
             .unwrap_or(0);
-        let is_nullable = values.get(4).and_then(|v| v.as_str()).unwrap_or("YES");
-        let data_type = values.get(5).and_then(|v| v.as_str()).unwrap_or("varchar");
+        let is_nullable = json_str(&values, 4, "YES");
+        let data_type = json_str(&values, 5, "varchar");
 
         let attrelid = table_oid(schema_name, tbl_name);
         let pg_type = trino_type_to_pg(data_type);

src/catalog/pg_class.rs

@@ -1,14 +1,13 @@
-// Copyright 2026 Stackable GmbH
-// Licensed under the Open Software License version 3.0 (OSL-3.0).
-// See LICENSE file in the project root for full license text.
+// SPDX-FileCopyrightText: 2026 Stackable GmbH
+// SPDX-License-Identifier: OSL-3.0
 use std::sync::Arc;
 
 use pgwire::api::Type;
 use pgwire::api::results::Response;
 use pgwire::error::{PgWireError, PgWireResult};
 use trino_rust_client::{Client, Row};
 
-use super::{build_response, text_field};
+use super::{build_response, json_str, text_field};
 
 /// Starting OID for synthetic table entries (above PostgreSQL's system range).
 const BASE_TABLE_OID: u32 = 16384;
@@ -19,37 +18,47 @@ const PUBLIC_NAMESPACE_OID: u32 = 2200;
 /// Starting OID for synthetic namespace entries (schemas other than "public").
 const BASE_NAMESPACE_OID: u32 = 20000;
 
+/// djb2 hash, then map into `[base, i32::MAX)`. Used to derive synthetic
+/// OIDs from string names.
+///
+/// We need a deterministic hash that produces the same OID every time the
+/// gateway sees the same name — clients that reconnect must see stable
+/// OIDs in pg_class so they can resolve them in pg_attribute. Rust's
+/// `std::collections::hash_map::DefaultHasher` is *not* stability-
+/// guaranteed across compiler releases (the standard library is allowed
+/// to swap the algorithm), so we use the well-known djb2 hash which is
+/// fixed and trivial.
+fn djb2_oid(parts: &[&str], base: u32) -> u32 {
+    let mut h: u32 = 5381;
+    for (i, part) in parts.iter().enumerate() {
+        if i > 0 {
+            // Separator byte to avoid collisions between e.g. "abctable"
+            // and "ab" + "ctable".
+            h = h.wrapping_mul(33).wrapping_add(0xFF);
+        }
+        for b in part.bytes() {
+            h = h.wrapping_mul(33).wrapping_add(u32::from(b));
+        }
+    }
+    base + (h % (i32::MAX as u32 - base))
+}
+
 /// Compute a deterministic OID for a table given its schema and table name.
 ///
-/// Uses the same algorithm as `table_oid`, so pg_attribute can reference the
-/// same OID for attrelid.
+/// Uses the same hash as `namespace_oid`, so pg_attribute can reference
+/// the same OID for `attrelid`.
 pub fn table_oid(schema_name: &str, table_name: &str) -> u32 {
-    // Simple deterministic hash: we want stable OIDs across pg_class and
-    // pg_attribute calls within the same session.  Using a basic hash that
-    // fits in u32 and is offset from BASE_TABLE_OID.
-    let mut h: u32 = 5381;
-    for b in schema_name.bytes() {
-        h = h.wrapping_mul(33).wrapping_add(u32::from(b));
-    }
-    // Separator to avoid collisions between "abctable" and "ab" + "ctable"
-    h = h.wrapping_mul(33).wrapping_add(0xFF);
-    for b in table_name.bytes() {
-        h = h.wrapping_mul(33).wrapping_add(u32::from(b));
-    }
-    // Ensure it's at least BASE_TABLE_OID and positive (i32 range)
-    BASE_TABLE_OID + (h % (i32::MAX as u32 - BASE_TABLE_OID))
+    djb2_oid(&[schema_name, table_name], BASE_TABLE_OID)
 }
 
-/// Compute a deterministic namespace OID for a schema name.
+/// Compute a deterministic namespace OID for a schema name. The "public"
+/// schema gets the well-known PostgreSQL OID (2200); everything else is
+/// hashed.
 pub fn namespace_oid(schema_name: &str) -> u32 {
     if schema_name == "public" {
         return PUBLIC_NAMESPACE_OID;
     }
-    let mut h: u32 = 5381;
-    for b in schema_name.bytes() {
-        h = h.wrapping_mul(33).wrapping_add(u32::from(b));
-    }
-    BASE_NAMESPACE_OID + (h % (i32::MAX as u32 - BASE_NAMESPACE_OID))
+    djb2_oid(&[schema_name], BASE_NAMESPACE_OID)
 }
 
 /// Map Trino's table_type string to PostgreSQL's relkind char.
@@ -85,12 +94,9 @@ pub async fn respond_pg_class(client: &Arc<Client>) -> PgWireResult<Vec<Response
     for trino_row in dataset.into_vec() {
         let values = trino_row.into_json();
         // columns: table_schema (0), table_name (1), table_type (2)
-        let schema_name = values.first().and_then(|v| v.as_str()).unwrap_or("public");
-        let tbl_name = values.get(1).and_then(|v| v.as_str()).unwrap_or("");
-        let tbl_type = values
-            .get(2)
-            .and_then(|v| v.as_str())
-            .unwrap_or("BASE TABLE");
+        let schema_name = json_str(&values, 0, "public");
+        let tbl_name = json_str(&values, 1, "");
+        let tbl_type = json_str(&values, 2, "BASE TABLE");
 
         let oid = table_oid(schema_name, tbl_name);
         let ns_oid = namespace_oid(schema_name);

src/startup.rs

@@ -1,6 +1,5 @@
-// Copyright 2026 Stackable GmbH
-// Licensed under the Open Software License version 3.0 (OSL-3.0).
-// See LICENSE file in the project root for full license text.
+// SPDX-FileCopyrightText: 2026 Stackable GmbH
+// SPDX-License-Identifier: OSL-3.0
 use std::collections::HashMap;
 use std::fmt::Debug;
 use std::sync::Arc;
@@ -16,7 +15,6 @@ use pgwire::api::{ClientInfo, PgWireConnectionState};
 use pgwire::error::{ErrorInfo, PgWireError, PgWireResult};
 use pgwire::messages::startup::{Authentication, SecretKey};
 use pgwire::messages::{PgWireBackendMessage, PgWireFrontendMessage};
-use rand::Rng;
 use trino_rust_client::auth::Auth;
 
 use crate::config::Config;
@@ -32,7 +30,7 @@ static CONNECTION_PID: AtomicI32 = AtomicI32::new(1);
 /// authorisation token for somebody else's connection.
 fn new_pid_and_secret_key() -> (i32, SecretKey) {
     let pid = CONNECTION_PID.fetch_add(1, Ordering::Relaxed);
-    let secret = rand::thread_rng().r#gen::<i32>();
+    let secret = rand::random::<i32>();
     (pid, SecretKey::I32(secret))
 }
 
@@ -142,31 +140,9 @@ impl StartupHandler for GatewayStartupHandler {
                         ))
                         .await?;
                 } else {
-                    // No auth — create Trino client immediately
+                    // No auth — create Trino client immediately and finish.
                     let trino_client = Arc::new(self.build_trino_client(None, None)?);
-                    let (pid, secret_key) = new_pid_and_secret_key();
-                    client.set_pid_and_secret_key(pid, secret_key.clone());
-                    let active_query_id = session::register_cancel(
-                        pid,
-                        secret_key.clone(),
-                        Arc::clone(&trino_client),
-                    );
-                    let conn_id = format!("{}_{}", client.socket_addr(), pid);
-                    client
-                        .metadata_mut()
-                        .insert(session::CONNECTION_ID_KEY.to_owned(), conn_id.clone());
-                    session::register_connection(
-                        conn_id,
-                        ConnectionState {
-                            trino_client,
-                            config: self.config.clone(),
-                            portals: Default::default(),
-                            active_query_id,
-                            cancel_key: (pid, secret_key),
-                        },
-                    );
-                    finish_authentication(client, &GatewayParameterProvider).await?;
-                    tracing::info!(addr = %client.socket_addr(), "client connected (no auth)");
+                    self.establish_session(client, trino_client, None).await?;
                 }
             }
             PgWireFrontendMessage::PasswordMessageFamily(pwd) => {
@@ -177,50 +153,35 @@ impl StartupHandler for GatewayStartupHandler {
                 let password = pwd.into_password()?;
                 let user = client.metadata().get("user").cloned().unwrap_or_default();
 
-                // Build Trino client with the PG client's credentials
+                // Build Trino client with the PG client's credentials.
                 let trino_client =
                     self.build_trino_client(Some(&user), Some(&password.password))?;
 
-                // Validate credentials by running a lightweight query against Trino.
-                // If Trino rejects the auth, we reject the PG connection immediately
-                // rather than letting the first real query fail with a confusing error.
+                // Validate credentials by running a lightweight query against
+                // Trino. We reject the PG connection here on any failure
+                // (auth rejection, network error, malformed reply, ...) so
+                // the client gets a clear failure during startup rather than
+                // a confusing error on its first real query. The error
+                // message is logged at the gateway with the underlying
+                // cause; the client gets the standard `InvalidPassword`
+                // response — distinguishing auth-failure from other failures
+                // requires inspecting `trino-rust-client`'s error enum and
+                // is tracked separately.
                 if let Err(e) = trino_client
                     .get::<trino_rust_client::Row>("SELECT 1".to_owned())
                     .await
                 {
-                    let msg = e.to_string();
                     tracing::warn!(
                         addr = %client.socket_addr(),
                         user = %user,
-                        "Trino authentication failed: {msg}"
+                        error = %e,
+                        "rejecting PG connection — Trino credential check failed (auth rejection, network error, or unreachable Trino)"
                     );
                     return Err(PgWireError::InvalidPassword(user));
                 }
 
-                let trino_client = Arc::new(trino_client);
-                let (pid, secret_key) = new_pid_and_secret_key();
-                client.set_pid_and_secret_key(pid, secret_key.clone());
-                let active_query_id = session::register_cancel(
-                    pid,
-                    secret_key.clone(),
-                    Arc::clone(&trino_client),
-                );
-                let conn_id = format!("{}_{}", client.socket_addr(), pid);
-                client
-                    .metadata_mut()
-                    .insert(session::CONNECTION_ID_KEY.to_owned(), conn_id.clone());
-                session::register_connection(
-                    conn_id,
-                    ConnectionState {
-                        trino_client,
-                        config: self.config.clone(),
-                        portals: Default::default(),
-                        active_query_id,
-                        cancel_key: (pid, secret_key),
-                    },
-                );
-                finish_authentication(client, &GatewayParameterProvider).await?;
-                tracing::info!(addr = %client.socket_addr(), user = %user, "client connected");
+                self.establish_session(client, Arc::new(trino_client), Some(user.as_str()))
+                    .await?;
             }
             _ => {
                 return Err(PgWireError::ApiError(
@@ -241,6 +202,51 @@ impl GatewayStartupHandler {
         self.config.tls_cert.is_some() && self.config.tls_key.is_some()
     }
 
+    /// Common per-connection setup once the Trino client has been built
+    /// and (for `--auth=true`) credentials have been verified. Allocates a
+    /// fresh `(pid, secret_key)`, registers the connection in the cancel
+    /// registry and the per-connection state map, and finishes the PG
+    /// startup handshake. `user` is `Some` only on the auth=true path,
+    /// where it's logged for operator visibility.
+    async fn establish_session<C>(
+        &self,
+        client: &mut C,
+        trino_client: Arc<trino_rust_client::Client>,
+        user: Option<&str>,
+    ) -> PgWireResult<()>
+    where
+        C: ClientInfo + Sink<PgWireBackendMessage> + Unpin + Send + Sync,
+        C::Error: Debug,
+        PgWireError: From<<C as Sink<PgWireBackendMessage>>::Error>,
+    {
+        let (pid, secret_key) = new_pid_and_secret_key();
+        client.set_pid_and_secret_key(pid, secret_key.clone());
+        let active_query_id =
+            session::register_cancel(pid, secret_key.clone(), Arc::clone(&trino_client));
+        let conn_id = format!("{}_{}", client.socket_addr(), pid);
+        client
+            .metadata_mut()
+            .insert(session::CONNECTION_ID_KEY.to_owned(), conn_id.clone());
+        session::register_connection(
+            conn_id,
+            ConnectionState {
+                trino_client,
+                config: self.config.clone(),
+                portals: Default::default(),
+                active_query_id,
+                cancel_key: (pid, secret_key),
+            },
+        );
+        finish_authentication(client, &GatewayParameterProvider).await?;
+        match user {
+            Some(u) => {
+                tracing::info!(addr = %client.socket_addr(), user = %u, "client connected")
+            }
+            None => tracing::info!(addr = %client.socket_addr(), "client connected (no auth)"),
+        }
+        Ok(())
+    }
+
     /// Build a Trino client, optionally with Basic auth credentials from the PG client.
     fn build_trino_client(
         &self,