ci: validate typed Python DataSource failure head

crates/sail-catalog-iceberg/Cargo.toml

@@ -41,6 +41,7 @@ log = { workspace = true }
 tokio = { workspace = true }
 wiremock = { workspace = true }
 testcontainers = { workspace = true }
+tempfile = { workspace = true }
 
 [build-dependencies]
 sail-build-scripts = { path = "../sail-build-scripts" }

crates/sail-catalog/Cargo.toml

@@ -23,3 +23,6 @@ log = { workspace = true }
 
 sail-common-datafusion = { path = "../sail-common-datafusion" }
 sail-common = { path = "../sail-common" }
+
+[dev-dependencies]
+tempfile = { workspace = true }

crates/sail-catalog/src/credentials.rs

@@ -1,6 +1,7 @@
 use std::fmt::Debug;
+use std::path::PathBuf;
 
-use crate::error::CatalogResult;
+use crate::error::{CatalogError, CatalogResult};
 
 #[async_trait::async_trait]
 pub trait CatalogCredentials: Debug + Send + Sync + 'static {
@@ -34,3 +35,142 @@ impl CatalogCredentials for StaticCatalogCredentials {
         Ok(Some(self.credential.clone()))
     }
 }
+
+/// Credentials backed by a token file on disk, such as a kubelet-projected
+/// service account token. The file is read on every call, so a rotated token
+/// is picked up without restarting the server. The Iceberg REST provider reads
+/// the credential fresh for each request and retries once on a `401`, so a
+/// token that rotates mid-operation is recovered without an in-memory cache.
+#[derive(Debug)]
+pub struct FileCatalogCredentials {
+    path: PathBuf,
+}
+
+impl FileCatalogCredentials {
+    pub fn new(path: impl Into<PathBuf>) -> Self {
+        Self { path: path.into() }
+    }
+}
+
+#[async_trait::async_trait]
+impl CatalogCredentials for FileCatalogCredentials {
+    async fn retrieve(&self) -> CatalogResult<Option<String>> {
+        let credential = tokio::fs::read_to_string(&self.path)
+            .await
+            .map_err(|e| {
+                CatalogError::External(format!(
+                    "failed to read token file {}: {e}",
+                    self.path.display()
+                ))
+            })?
+            .trim()
+            .to_string();
+        if credential.is_empty() {
+            return Err(CatalogError::External(format!(
+                "token file {} is empty",
+                self.path.display()
+            )));
+        }
+        Ok(Some(credential))
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    #![expect(clippy::unwrap_used)]
+
+    use std::fs::File;
+    use std::io::Write;
+    use std::path::Path;
+
+    use tempfile::TempDir;
+
+    use super::*;
+
+    fn write_token(path: &Path, contents: &str) {
+        let mut file = File::create(path).unwrap();
+        file.write_all(contents.as_bytes()).unwrap();
+    }
+
+    #[tokio::test]
+    async fn retrieve_returns_token_from_file() {
+        let dir = TempDir::new().unwrap();
+        let path = dir.path().join("token");
+        write_token(&path, "s3cr3t-token");
+
+        let credentials = FileCatalogCredentials::new(&path);
+        assert_eq!(
+            credentials.retrieve().await.unwrap(),
+            Some("s3cr3t-token".to_string())
+        );
+    }
+
+    #[tokio::test]
+    async fn retrieve_trims_surrounding_whitespace() {
+        let dir = TempDir::new().unwrap();
+        let path = dir.path().join("token");
+        write_token(&path, "  s3cr3t-token\n\n");
+
+        let credentials = FileCatalogCredentials::new(&path);
+        assert_eq!(
+            credentials.retrieve().await.unwrap(),
+            Some("s3cr3t-token".to_string())
+        );
+    }
+
+    #[tokio::test]
+    async fn retrieve_rereads_rotated_token() {
+        let dir = TempDir::new().unwrap();
+        let path = dir.path().join("token");
+        write_token(&path, "first-token");
+
+        let credentials = FileCatalogCredentials::new(&path);
+        assert_eq!(
+            credentials.retrieve().await.unwrap(),
+            Some("first-token".to_string())
+        );
+
+        // Every call reads the file, so a rotated token (for example kubelet
+        // swapping a projected service account token) is picked up on the next
+        // retrieve without restarting the server.
+        write_token(&path, "second-token");
+        assert_eq!(
+            credentials.retrieve().await.unwrap(),
+            Some("second-token".to_string())
+        );
+    }
+
+    #[tokio::test]
+    async fn retrieve_reports_error_for_missing_file() {
+        let dir = TempDir::new().unwrap();
+        let path = dir.path().join("does-not-exist");
+
+        let credentials = FileCatalogCredentials::new(&path);
+        let error = credentials.retrieve().await.unwrap_err();
+        assert!(
+            matches!(error, CatalogError::External(_)),
+            "unexpected error variant: {error:?}"
+        );
+    }
+
+    #[tokio::test]
+    async fn retrieve_reports_error_for_empty_file() {
+        let dir = TempDir::new().unwrap();
+        let path = dir.path().join("token");
+        write_token(&path, "\n  \n");
+
+        let credentials = FileCatalogCredentials::new(&path);
+        let error = credentials.retrieve().await.unwrap_err();
+        assert!(
+            matches!(&error, CatalogError::External(message) if message.contains("empty")),
+            "unexpected error: {error:?}"
+        );
+
+        // Once the file holds a token again, the next retrieve returns it.
+        write_token(&path, "recovered-token");
+        assert_eq!(
+            credentials.retrieve().await.unwrap(),
+            Some("recovered-token".to_string())
+        );
+    }
+}

crates/sail-common-datafusion/src/error.rs

@@ -14,10 +14,37 @@ pub struct RemoteError {
     pub cause: CommonErrorCause,
 }
 
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub enum PythonFailureKind {
+    Terminal,
+    Transient,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Error)]
+pub enum PythonDataSourceFailure {
+    #[error("Python data source reported a terminal failure")]
+    Terminal,
+    #[error("Python data source reported a transient failure")]
+    Transient,
+}
+
+impl PythonDataSourceFailure {
+    pub fn kind(self) -> PythonFailureKind {
+        match self {
+            Self::Terminal => PythonFailureKind::Terminal,
+            Self::Transient => PythonFailureKind::Transient,
+        }
+    }
+}
+
 #[derive(Debug, Clone, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
 pub struct PythonErrorCause {
     pub summary: String,
     pub traceback: Option<Vec<String>>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub failure_kind: Option<PythonFailureKind>,
 }
 
 /// A trait to extract Python error cause from a generic error.
@@ -166,6 +193,14 @@ impl CommonErrorCause {
             };
         }
 
+        if let Some(failure) = error.downcast_ref::<PythonDataSourceFailure>() {
+            return Self::Python(PythonErrorCause {
+                summary: failure.to_string(),
+                traceback: None,
+                failure_kind: Some(failure.kind()),
+            });
+        }
+
         if let Some(cause) = Py::extract(error) {
             return Self::Python(cause);
         }
@@ -187,3 +222,37 @@ impl CommonErrorCause {
         Self::build::<Py>(error, &mut HashSet::new())
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_python_failure_kind_round_trips() -> Result<(), Box<dyn std::error::Error>> {
+        let cause = CommonErrorCause::Python(PythonErrorCause {
+            summary: "Python data source reported a terminal failure".to_string(),
+            traceback: None,
+            failure_kind: Some(PythonFailureKind::Terminal),
+        });
+
+        let encoded = serde_json::to_string(&cause)?;
+        assert!(encoded.contains(r#""failureKind":"terminal""#));
+        let decoded: CommonErrorCause = serde_json::from_str(&encoded)?;
+        let CommonErrorCause::Python(decoded) = decoded else {
+            return Err(std::io::Error::other("expected Python error cause").into());
+        };
+        assert_eq!(decoded.failure_kind, Some(PythonFailureKind::Terminal));
+        Ok(())
+    }
+
+    #[test]
+    fn test_legacy_python_cause_defaults_failure_kind() -> Result<(), Box<dyn std::error::Error>> {
+        let encoded = r#"{"python":{"summary":"legacy","traceback":null}}"#;
+        let decoded: CommonErrorCause = serde_json::from_str(encoded)?;
+        let CommonErrorCause::Python(decoded) = decoded else {
+            return Err(std::io::Error::other("expected Python error cause").into());
+        };
+        assert_eq!(decoded.failure_kind, None);
+        Ok(())
+    }
+}

crates/sail-common/src/config/application.rs

@@ -551,6 +551,14 @@ pub enum CatalogType {
             serialize_with = "serialize_optional_secret"
         )]
         bearer_access_token: Option<SecretString>,
+        /// Path to a file holding the bearer token. When set, the token is
+        /// re-read from this file for every request, so a rotated token (for
+        /// example a kubelet-projected service account token) is picked up
+        /// without restarting the server. Takes precedence over
+        /// `bearer_access_token`. The path is not a secret, so it is kept as a
+        /// plain string.
+        #[serde(skip_serializing_if = "Option::is_none")]
+        bearer_access_token_file: Option<String>,
         #[serde(flatten)]
         cache: CatalogCacheConfig,
     },