hamelin_executor 0.21.6

//! Configuration types for Hamelin backends.
//!
//! These types describe catalogs, executors, and dataset sources. They are
//! backend-agnostic — each executor crate resolves the dataset configs it
//! supports and returns a clean error for the rest.

use std::collections::HashMap;
use std::fmt::{Debug, Formatter, Result as FmtResult};
use std::path::PathBuf;

use anyhow::Context;
use hamelin_lib::antlr::parse_type;
use hamelin_lib::catalog::{Column, HamelinType};
use hamelin_lib::tree::ast::dataset_identifier::QualifiedDatasetIdentifier;
use hamelin_lib::types::Type;
use serde::{Deserialize, Serialize, Serializer};

/// Wraps a [`CatalogConfig`] with an optional `default` flag.
///
/// When `default` is `true`, unqualified dataset references (`namespace.table`)
/// resolve against this catalog using the configured default space. The TOML key
/// is the catalog's real name (e.g., `lakekeeper`).
///
/// Uses `#[serde(flatten)]` so that `type`, `uri`, `warehouse`, and `default`
/// all appear as siblings at the same TOML level.
#[derive(Deserialize, Serialize, Clone, Debug, PartialEq, bon::Builder)]
pub struct CatalogEntry {
    #[serde(default)]
    pub default: bool,
    #[serde(flatten)]
    pub config: CatalogConfig,
}

// --- Catalog types ---

#[derive(Deserialize, Serialize, Clone, Debug, PartialEq)]
#[serde(tag = "type", rename_all = "snake_case")]
pub enum CatalogConfig {
    IcebergRest {
        uri: String,
        warehouse: Option<String>,
        #[serde(default, skip_serializing_if = "HashMap::is_empty")]
        extra_properties: HashMap<String, String>,
    },
    Reflection,
}

// --- Dataset types ---

/// File format for file-based datasets.
#[derive(Deserialize, Serialize, Clone, Debug, PartialEq)]
#[serde(rename_all = "lowercase")]
pub enum FileFormat {
    Json,
    CSV,
    Parquet,
    Lines,
}

/// Compression codec for file-based datasets.
#[derive(Deserialize, Serialize, Clone, Debug, PartialEq)]
#[serde(rename_all = "lowercase")]
pub enum Compression {
    Gzip,
    Bzip2,
    Xz,
    Zstd,
}

/// Redacts a secret value when serializing so it is never echoed back out via
/// config serialization (e.g. `--print-config`, logging, or telemetry).
fn redact_secret<S>(_secret: &str, serializer: S) -> Result<S::Ok, S::Error>
where
    S: Serializer,
{
    serializer.serialize_str("[REDACTED]")
}

#[derive(Deserialize, Serialize, Clone, PartialEq, bon::Builder)]
pub struct StaticAwsCredentials {
    pub access_id: String,
    #[serde(serialize_with = "redact_secret")]
    pub secret_key: String,
}

impl Debug for StaticAwsCredentials {
    fn fmt(&self, f: &mut Formatter<'_>) -> FmtResult {
        f.debug_struct("StaticAwsCredentials")
            .field("access_id", &self.access_id)
            .field("secret_key", &"[REDACTED]")
            .finish()
    }
}

#[derive(Deserialize, Serialize, Clone, Debug, PartialEq, bon::Builder)]
pub struct FilesConfig {
    pub paths: Vec<String>,
    #[serde(default)]
    #[builder(into)]
    pub format: Option<FileFormat>,
    #[serde(default)]
    #[builder(into)]
    pub compression: Option<Compression>,
    #[serde(default)]
    #[builder(into)]
    pub file_extension: Option<String>,
    #[serde(default)]
    #[builder(into)]
    pub columns: Option<Vec<ColumnConfig>>,
    #[serde(default)]
    #[builder(into)]
    pub header: Option<bool>,
    #[serde(default)]
    #[builder(into)]
    pub delimiter: Option<String>,
    #[serde(default)]
    #[builder(into)]
    pub region: Option<String>,
    #[serde(default)]
    #[builder(into)]
    pub role_arn: Option<String>,
    #[serde(default)]
    pub static_credentials: Option<StaticAwsCredentials>,
    #[serde(default)]
    #[builder(into)]
    pub endpoint_url: Option<String>,
    #[serde(default)]
    pub force_path_style: Option<bool>,
    #[serde(default)]
    pub allow_http: Option<bool>,
    /// JSONPath to walk at scan time. Optional for `format = "json"`,
    /// rejected for other formats. Accepted subset: `$`, `.<name>`,
    /// `["..."]`, `[*]`. When absent (or just `$`), each top-level
    /// JSON document becomes one Variant row. With a terminal `[*]`,
    /// each array element along the path becomes one Variant row.
    /// With a terminal field, each top-level document's value at that
    /// path becomes one Variant row.
    #[serde(default)]
    #[builder(into)]
    pub json_path: Option<String>,
    /// Working-buffer budget (bytes) for the streaming JSON parser used
    /// by `json`. Defaults to ~1 MiB when `None`. Rejected for
    /// non-`json` formats.
    #[serde(default)]
    pub parser_buffer_size: Option<usize>,
}

#[derive(Deserialize, Serialize, Clone, Debug, PartialEq, bon::Builder)]
pub struct MemConfig {
    pub columns: Vec<ColumnConfig>,
}

#[derive(Deserialize, Serialize, Clone, Debug, PartialEq, bon::Builder)]
pub struct IcebergConfig {
    pub metadata_location: String,
    #[serde(default)]
    pub region: Option<String>,
}

#[derive(Deserialize, Serialize, Clone, Debug, PartialEq, bon::Builder)]
pub struct ViewConfig {
    pub query: String,
}

#[derive(Deserialize, Serialize, Clone, Debug, PartialEq, derive_more::From)]
#[serde(tag = "type", rename_all = "snake_case")]
pub enum DatasetConfig {
    Files(FilesConfig),
    Mem(MemConfig),
    Iceberg(IcebergConfig),
    View(ViewConfig),
}

#[derive(Deserialize, Serialize, Clone, Debug, PartialEq, bon::Builder)]
pub struct ColumnConfig {
    pub name: String,
    #[serde(
        deserialize_with = "deserialize_hamelin_type",
        serialize_with = "serialize_hamelin_type"
    )]
    pub typ: HamelinType,
}

/// Deserialize a `HamelinType` from a Hamelin type expression string
/// (e.g. `"string"`, `"int"`, `"{name: string, age: int}"`).
fn deserialize_hamelin_type<'de, D>(deserializer: D) -> Result<HamelinType, D::Error>
where
    D: serde::Deserializer<'de>,
{
    let s = String::deserialize(deserializer)?;
    let ctx = parse_type(s.clone()).map_err(|e| serde::de::Error::custom(e.contextualize(&s)))?;
    let typ =
        Type::from_parse_tree(&ctx).map_err(|e| serde::de::Error::custom(e.contextualize(&s)))?;
    Ok(typ.into())
}

/// Serialize a `HamelinType` as a Hamelin type expression string,
/// ensuring round-trip compatibility with `deserialize_hamelin_type`.
fn serialize_hamelin_type<S>(typ: &HamelinType, serializer: S) -> Result<S::Ok, S::Error>
where
    S: serde::Serializer,
{
    let t: Type = typ.clone().try_into().map_err(serde::ser::Error::custom)?;
    serializer.serialize_str(&t.to_string())
}

impl From<&Column> for ColumnConfig {
    fn from(col: &Column) -> Self {
        ColumnConfig {
            name: col.name.as_str().to_string(),
            typ: col.typ.clone(),
        }
    }
}

// --- Dataset resolution ---

/// Nested namespace tree for local dataset configuration.
///
/// TOML paths like `datasets.lakekeeper.test.signals.events` deserialize as nested
/// branches ending in a [`DatasetConfig`] leaf.
#[derive(Clone, Debug, Deserialize, Serialize, PartialEq)]
#[serde(untagged)]
pub enum DatasetTree {
    Leaf(DatasetConfig),
    Branch(HashMap<String, DatasetTree>),
}

impl DatasetTree {
    pub fn insert_table(&mut self, namespace: &[&str], table: &str, config: DatasetConfig) {
        self.insert(namespace, table, config);
    }

    fn insert(&mut self, namespace: &[&str], table: &str, config: DatasetConfig) {
        if matches!(self, DatasetTree::Leaf(_)) {
            *self = DatasetTree::Branch(HashMap::new());
        }
        let DatasetTree::Branch(map) = self else {
            return;
        };
        Self::insert_into(map, namespace, table, config);
    }

    fn insert_into(
        map: &mut HashMap<String, DatasetTree>,
        namespace: &[&str],
        table: &str,
        config: DatasetConfig,
    ) {
        if namespace.is_empty() {
            map.insert(table.to_string(), DatasetTree::Leaf(config));
            return;
        }
        map.entry(namespace[0].to_string())
            .or_insert_with(|| DatasetTree::Branch(HashMap::new()))
            .insert(&namespace[1..], table, config);
    }

    fn lookup<'a>(&'a self, namespace: &[&str], table: &str) -> Option<&'a DatasetConfig> {
        if namespace.is_empty() {
            return match self {
                DatasetTree::Branch(map) => match map.get(table)? {
                    DatasetTree::Leaf(config) => Some(config),
                    DatasetTree::Branch(_) => None,
                },
                DatasetTree::Leaf(_) => None,
            };
        }

        match self {
            DatasetTree::Branch(map) => map.get(namespace[0])?.lookup(&namespace[1..], table),
            DatasetTree::Leaf(_) => None,
        }
    }

    fn flatten(
        &self,
        prefix: &str,
        default_catalog: bool,
        catalog: &str,
        out: &mut HashMap<String, DatasetConfig>,
    ) {
        match self {
            DatasetTree::Leaf(config) => {
                let key = if default_catalog {
                    prefix.to_string()
                } else {
                    format!("{catalog}.{prefix}")
                };
                out.insert(key, config.clone());
            }
            DatasetTree::Branch(map) => {
                for (name, child) in map {
                    let next = if prefix.is_empty() {
                        name.clone()
                    } else {
                        format!("{prefix}.{name}")
                    };
                    child.flatten(&next, default_catalog, catalog, out);
                }
            }
        }
    }
}

/// What a table reference resolved to.
#[derive(Debug, Clone)]
pub enum ResolvedDataset {
    /// A locally-configured dataset (local override in the datasets map).
    Local {
        name: String,
        catalog: String,
        namespace: Vec<String>,
        table: String,
        config: DatasetConfig,
    },
    /// A table in a configured catalog — executor handles based on catalog type.
    Catalog {
        catalog_name: String,
        catalog_config: CatalogConfig,
        namespace: Vec<String>,
        table: String,
    },
}

/// Holds configured datasets and catalogs, and resolves Hamelin dataset references.
///
/// Resolution via [`Self::resolve_dataset`]:
/// - A [`QualifiedDatasetIdentifier`] (e.g. `acme:signals.events`) is mapped to nested
///   namespace parts and a table name via [`QualifiedDatasetIdentifier::to_namespace_table`].
/// - The namespace and table are looked up against the catalog marked `default = true`.
///
/// Local dataset overrides always win. If there's no local override, fall back
/// to the catalog's configured resolution mechanism.
#[derive(Debug, Clone)]
pub struct DatasetStore {
    /// catalog → nested namespace tree → table configs
    datasets: HashMap<String, DatasetTree>,
    catalogs: HashMap<String, CatalogConfig>,
    /// The catalog name with `default = true`, used by [`Self::resolve_dataset`].
    default_catalog: Option<String>,
}

impl DatasetStore {
    pub fn new(
        datasets: HashMap<String, DatasetTree>,
        catalog_entries: HashMap<String, CatalogEntry>,
    ) -> Self {
        let default_catalog = catalog_entries
            .iter()
            .find(|(_, entry)| entry.default)
            .map(|(name, _)| name.clone());

        let catalogs = catalog_entries
            .into_iter()
            .map(|(name, entry)| (name, entry.config))
            .collect();

        Self {
            datasets,
            catalogs,
            default_catalog,
        }
    }

    /// The name of the catalog marked as default, if any.
    pub fn default_catalog(&self) -> Option<&str> {
        self.default_catalog.as_deref()
    }

    /// Resolve a space-qualified [`QualifiedDatasetIdentifier`] against the default catalog.
    ///
    /// Maps Hamelin space syntax to nested namespace parts and a table name via
    /// [`QualifiedDatasetIdentifier::to_namespace_table`].
    pub fn resolve_dataset(
        &self,
        id: &QualifiedDatasetIdentifier,
    ) -> Result<ResolvedDataset, DatasetStoreError> {
        let (namespace, table) = id.to_namespace_table();
        let catalog = self.default_catalog.as_deref().ok_or_else(|| {
            DatasetStoreError::NoDefaultCatalog(format_dataset_ref("", &namespace, &table, true))
        })?;
        self.resolve_catalog_table(catalog, &namespace, &table)
    }

    /// Whether `catalog` is the default catalog (used to decide default vs fully-qualified display).
    fn is_default_catalog(&self, catalog: &str) -> bool {
        self.default_catalog
            .as_deref()
            .is_some_and(|dc| dc == catalog)
    }

    fn resolve_catalog_table(
        &self,
        catalog: &str,
        namespace: &[String],
        table: &str,
    ) -> Result<ResolvedDataset, DatasetStoreError> {
        let namespace_str: Vec<&str> = namespace.iter().map(String::as_str).collect();
        if let Some(tree) = self.datasets.get(catalog) {
            if let Some(config) = tree.lookup(&namespace_str, table) {
                let name =
                    format_dataset_ref(catalog, namespace, table, self.is_default_catalog(catalog));
                return Ok(ResolvedDataset::Local {
                    name,
                    catalog: catalog.to_string(),
                    namespace: namespace.to_vec(),
                    table: table.to_string(),
                    config: config.clone(),
                });
            }
        }

        // 2. Check configured catalogs
        if let Some(catalog_config) = self.catalogs.get(catalog) {
            return Ok(ResolvedDataset::Catalog {
                catalog_name: catalog.to_string(),
                catalog_config: catalog_config.clone(),
                namespace: namespace.to_vec(),
                table: table.to_string(),
            });
        }

        // 3. Nothing matched
        Err(DatasetStoreError::UnknownDataset(format_dataset_ref(
            catalog,
            namespace,
            table,
            self.is_default_catalog(catalog),
        )))
    }

    /// Flatten datasets for use by executors that pre-register everything.
    pub fn flat_datasets(&self) -> HashMap<String, DatasetConfig> {
        let mut result = HashMap::new();
        for (catalog, tree) in &self.datasets {
            tree.flatten("", self.is_default_catalog(catalog), catalog, &mut result);
        }
        result
    }

    /// The config for the default catalog, if one is marked `default = true`.
    pub fn default_catalog_config(&self) -> Option<(&str, &CatalogConfig)> {
        let name = self.default_catalog.as_deref()?;
        let config = self.catalogs.get(name)?;
        Some((name, config))
    }
}

fn format_dataset_ref(
    catalog: &str,
    namespace: &[String],
    table: &str,
    default_catalog: bool,
) -> String {
    let path = if namespace.is_empty() {
        table.to_string()
    } else {
        format!("{}.{}", namespace.join("."), table)
    };
    if default_catalog {
        path
    } else {
        format!("{catalog}.{path}")
    }
}

#[derive(Debug, thiserror::Error)]
pub enum DatasetStoreError {
    #[error("Unknown dataset '{0}': no local config or catalog matches")]
    UnknownDataset(String),
    #[error(
        "No default catalog configured; use a space-qualified identifier (space:namespace.table) or set default_space for '{0}'"
    )]
    NoDefaultCatalog(String),
    #[error("Invalid dataset reference: {0}")]
    InvalidDatasetReference(#[from] hamelin_lib::err::TranslationError),
}

// --- Top-level configuration ---

/// Top-level Hamelin configuration loaded from `~/.hamelin.toml` and environment
/// variables. Contains catalog discovery sources and dataset definitions.
///
/// This struct is intentionally backend-agnostic — backend-specific configuration
/// (e.g. Trino connection settings) lives in each executor crate. Unknown TOML
/// keys are silently ignored, so backend-specific sections can coexist in the
/// same config file.
#[derive(Serialize, Deserialize, PartialEq, Debug, Clone)]
#[serde(default)]
pub struct HamelinConfig {
    /// Catalog discovery sources, keyed by catalog name.
    pub catalogs: HashMap<String, CatalogEntry>,
    /// Datasets: `datasets.{catalog}.{namespace...}.{table}` → DatasetConfig.
    pub datasets: HashMap<String, DatasetTree>,
    /// Path to the reflected catalog file for the default catalog.
    /// Used by `--use-catalog-file` / `-e` for offline type-checking.
    pub catalog_filename: String,
    /// Default space for unqualified dataset references (`FROM events` → `space:events`).
    /// CLI `--default-space` overrides this value when set.
    pub default_space: Option<String>,
}

impl Default for HamelinConfig {
    fn default() -> Self {
        Self {
            catalogs: HashMap::new(),
            datasets: HashMap::new(),
            catalog_filename: "~/.hamelin.env".to_string(),
            default_space: None,
        }
    }
}

impl HamelinConfig {
    pub fn catalog_filename_path(&self) -> anyhow::Result<PathBuf> {
        expanduser::expanduser(&self.catalog_filename)
            .context("Failed to expand environment filename")
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    fn store_with_default_catalog() -> DatasetStore {
        let mut catalogs = HashMap::new();
        catalogs.insert(
            "lakekeeper".to_string(),
            CatalogEntry {
                default: true,
                config: CatalogConfig::Reflection,
            },
        );
        DatasetStore::new(HashMap::new(), catalogs)
    }

    #[test]
    fn resolve_dataset_maps_space_to_namespace() {
        let store = store_with_default_catalog();
        let id = QualifiedDatasetIdentifier::from_canonical_str("acme:events").unwrap();
        let resolved = store.resolve_dataset(&id).unwrap();
        match resolved {
            ResolvedDataset::Catalog {
                catalog_name,
                namespace,
                table,
                ..
            } => {
                assert_eq!(catalog_name, "lakekeeper");
                assert_eq!(namespace, vec!["acme".to_string()]);
                assert_eq!(table, "events");
            }
            ResolvedDataset::Local { .. } => panic!("expected catalog resolution"),
        }
    }

    #[test]
    fn resolve_dataset_maps_nested_path_to_namespace_parts() {
        let store = store_with_default_catalog();
        let id = QualifiedDatasetIdentifier::from_canonical_str("acme:signals.events").unwrap();
        let resolved = store.resolve_dataset(&id).unwrap();
        match resolved {
            ResolvedDataset::Catalog {
                namespace, table, ..
            } => {
                assert_eq!(namespace, vec!["acme".to_string(), "signals".to_string()]);
                assert_eq!(table, "events");
            }
            ResolvedDataset::Local { .. } => panic!("expected catalog resolution"),
        }
    }

    #[test]
    fn resolve_dataset_finds_local_override_by_namespace() {
        let mut tree = DatasetTree::Branch(HashMap::new());
        tree.insert_table(
            &["acme", "signals"],
            "events",
            DatasetConfig::Mem(
                MemConfig::builder()
                    .columns(vec![ColumnConfig::builder()
                        .name("id".to_string())
                        .typ(HamelinType::Int)
                        .build()])
                    .build(),
            ),
        );
        let mut datasets = HashMap::new();
        datasets.insert("lakekeeper".to_string(), tree);
        let mut catalogs = HashMap::new();
        catalogs.insert(
            "lakekeeper".to_string(),
            CatalogEntry {
                default: true,
                config: CatalogConfig::Reflection,
            },
        );
        let store = DatasetStore::new(datasets, catalogs);
        let id = QualifiedDatasetIdentifier::from_canonical_str("acme:signals.events").unwrap();
        let resolved = store.resolve_dataset(&id).unwrap();
        match resolved {
            ResolvedDataset::Local {
                namespace, table, ..
            } => {
                assert_eq!(namespace, vec!["acme".to_string(), "signals".to_string()]);
                assert_eq!(table, "events");
            }
            ResolvedDataset::Catalog { .. } => panic!("expected local override"),
        }
    }

    #[test]
    fn static_credentials_debug_redacts_secret_key() {
        let creds = StaticAwsCredentials::builder()
            .access_id("AKIAEXAMPLE".to_string())
            .secret_key("super-secret-value".to_string())
            .build();
        let rendered = format!("{creds:#?}");
        assert!(!rendered.contains("super-secret-value"));
        assert!(rendered.contains("[REDACTED]"));
        assert!(rendered.contains("AKIAEXAMPLE"));
    }

    #[test]
    fn static_credentials_serialize_redacts_secret_key() {
        let creds = StaticAwsCredentials::builder()
            .access_id("AKIAEXAMPLE".to_string())
            .secret_key("super-secret-value".to_string())
            .build();
        let rendered = serde_json::to_string(&creds).unwrap();
        assert!(!rendered.contains("super-secret-value"));
        assert!(rendered.contains("[REDACTED]"));
    }
}