hudi-core 0.5.0

/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */
//! Hudi Configurations.
use std::any::type_name;
use std::collections::HashMap;
use std::sync::Arc;

use crate::config::error::{ConfigError, Result};
use crate::storage::error::Result as StorageResult;
use crate::storage::util::parse_uri;
use ConfigError::NotFound;
use serde::{Deserialize, Serialize};
use url::Url;

pub mod error;
pub mod internal;
pub mod plan;
pub mod read;
pub mod read_options;
pub mod table;
pub mod util;

pub use read_options::{QueryType, ReadOptions};

pub const HUDI_CONF_DIR: &str = "HUDI_CONF_DIR";

/// An alternative key for a configuration, optionally marked as deprecated.
pub struct ConfigAlias {
    pub key: &'static str,
    pub deprecated: bool,
}

impl ConfigAlias {
    pub const fn new(key: &'static str) -> Self {
        Self {
            key,
            deprecated: false,
        }
    }

    pub const fn deprecated(key: &'static str) -> Self {
        Self {
            key,
            deprecated: true,
        }
    }
}

/// This defines some common APIs for working with configurations in Hudi.
pub trait ConfigParser: AsRef<str> {
    /// Configuration value type.
    type Output;

    /// Supplies the default value of the configuration.
    fn default_value(&self) -> Option<Self::Output>;

    fn key(&self) -> String {
        self.as_ref().to_string()
    }

    /// Returns alternative keys for this configuration.
    fn aliases(&self) -> &[ConfigAlias] {
        &[]
    }

    /// To indicate if the configuration is required or not, this helps in validation.
    fn is_required(&self) -> bool {
        false
    }

    /// Resolve the raw string value from configs, checking the primary key first, then aliases.
    fn resolve_raw_value<'a>(&self, configs: &'a HashMap<String, String>) -> Result<&'a str> {
        if let Some(v) = configs.get(self.as_ref()) {
            return Ok(v.as_str());
        }
        for alias in self.aliases() {
            if let Some(v) = configs.get(alias.key) {
                if alias.deprecated {
                    log_once::warn_once!(
                        "Config '{}' is deprecated; use '{}' instead",
                        alias.key,
                        self.as_ref()
                    );
                }
                return Ok(v.as_str());
            }
        }
        Err(NotFound(self.key()))
    }

    /// Validate the configuration by parsing the given [String] value and check if it is required.
    fn validate(&self, configs: &HashMap<String, String>) -> Result<()> {
        match self.parse_value(configs) {
            Ok(_) => Ok(()),
            Err(e) => {
                if !self.is_required() && matches!(e, NotFound(_)) {
                    Ok(())
                } else {
                    Err(e)
                }
            }
        }
    }

    /// Parse the [String] value to [Self::Output].
    fn parse_value(&self, configs: &HashMap<String, String>) -> Result<Self::Output>;

    /// Parse the [String] value to [Self::Output], or return the default value.
    ///
    /// Panic if the default value is not defined.
    fn parse_value_or_default(&self, configs: &HashMap<String, String>) -> Self::Output {
        self.parse_value(configs).unwrap_or_else(|_| {
            self.default_value()
                .unwrap_or_else(|| panic!("No default value for config '{}'", self.as_ref()))
        })
    }
}

/// All possible data types for Hudi Configuration values.
#[derive(Clone, Debug)]
pub enum HudiConfigValue {
    Boolean(bool),
    Integer(isize),
    UInteger(usize),
    String(String),
    List(Vec<String>),
}

impl HudiConfigValue {
    /// A convenience method to convert [HudiConfigValue] to [Url] when the value is a [String] and is intended to be a URL.
    /// Panic if the value is not a [String].
    pub fn to_url(self) -> StorageResult<Url> {
        match self {
            HudiConfigValue::String(v) => parse_uri(&v),
            _ => panic!(
                "Cannot cast {:?} to {}",
                type_name::<Self>(),
                type_name::<Url>()
            ),
        }
    }
}

impl From<HudiConfigValue> for bool {
    fn from(value: HudiConfigValue) -> Self {
        match value {
            HudiConfigValue::Boolean(v) => v,
            _ => panic!("Cannot cast {:?} to {}", value, type_name::<Self>()),
        }
    }
}

impl From<HudiConfigValue> for isize {
    fn from(value: HudiConfigValue) -> Self {
        match value {
            HudiConfigValue::Integer(v) => v,
            _ => panic!("Cannot cast {:?} to {}", value, type_name::<Self>()),
        }
    }
}

impl From<HudiConfigValue> for usize {
    fn from(value: HudiConfigValue) -> Self {
        match value {
            HudiConfigValue::UInteger(v) => v,
            _ => panic!("Cannot cast {:?} to {}", value, type_name::<Self>()),
        }
    }
}

impl From<HudiConfigValue> for String {
    fn from(value: HudiConfigValue) -> Self {
        match value {
            HudiConfigValue::Boolean(v) => v.to_string(),
            HudiConfigValue::Integer(v) => v.to_string(),
            HudiConfigValue::UInteger(v) => v.to_string(),
            HudiConfigValue::String(v) => v,
            _ => panic!("Cannot cast {:?} to {}", value, type_name::<Self>()),
        }
    }
}

impl From<HudiConfigValue> for Vec<String> {
    fn from(value: HudiConfigValue) -> Self {
        match value {
            HudiConfigValue::List(v) => v,
            _ => panic!("Cannot cast {:?} to {}", value, type_name::<Self>()),
        }
    }
}

/// Hudi configuration container.
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
pub struct HudiConfigs {
    raw_options: Arc<HashMap<String, String>>,
}

impl HudiConfigs {
    /// Create [HudiConfigs] using options in the form of key-value pairs.
    pub fn new<I, K, V>(options: I) -> Self
    where
        I: IntoIterator<Item = (K, V)>,
        K: AsRef<str>,
        V: Into<String>,
    {
        let raw_options = options
            .into_iter()
            .map(|(k, v)| (k.as_ref().into(), v.into()))
            .collect();
        Self {
            raw_options: Arc::new(raw_options),
        }
    }

    /// Create an empty [HudiConfigs].
    pub fn empty() -> Self {
        Self {
            raw_options: Arc::new(HashMap::new()),
        }
    }

    /// Create a deep-copy of the configs as [String] options in the form of key-value pairs.
    pub fn as_options(&self) -> HashMap<String, String> {
        self.raw_options.as_ref().clone()
    }

    /// Validate the associated config using the given parser by execute the [ConfigParser::validate] method.
    pub fn validate(&self, parser: impl ConfigParser<Output = HudiConfigValue>) -> Result<()> {
        parser.validate(&self.raw_options)
    }

    /// Check if the given key exists in the configs.
    pub fn contains(&self, key: impl AsRef<str>) -> bool {
        self.raw_options.contains_key(key.as_ref())
    }

    /// Look up a raw value without copying the option map.
    ///
    /// For a key this crate has no typed config for. Prefer [`Self::try_get`]
    /// where one exists — it parses, and errors rather than silently returning
    /// the default when a value is malformed.
    pub fn get_raw(&self, key: impl AsRef<str>) -> Option<&str> {
        self.raw_options.get(key.as_ref()).map(String::as_str)
    }

    /// Get value for the given config. Return [Result] with the value.
    /// If the config is not found or value was not parsed properly, return [Err].
    pub fn get(
        &self,
        parser: impl ConfigParser<Output = HudiConfigValue>,
    ) -> Result<HudiConfigValue> {
        parser.parse_value(&self.raw_options)
    }

    /// Get value or default value. If the config has no default value, this will panic.
    pub fn get_or_default(
        &self,
        parser: impl ConfigParser<Output = HudiConfigValue>,
    ) -> HudiConfigValue {
        parser.parse_value_or_default(&self.raw_options)
    }

    /// Get value if present, or default if absent. Returns `Err` on parse failures
    /// (e.g. `"yes"` for a bool config) instead of silently falling back to the default.
    pub fn try_get(
        &self,
        parser: impl ConfigParser<Output = HudiConfigValue>,
    ) -> Result<Option<HudiConfigValue>> {
        match parser.parse_value(&self.raw_options) {
            Ok(v) => Ok(Some(v)),
            Err(NotFound(_)) => Ok(parser.default_value()),
            Err(e) => Err(e),
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::config::table::HudiTableConfig;

    #[test]
    fn test_config_alias_constructors() {
        let alias = ConfigAlias::new("key");
        assert_eq!(alias.key, "key");
        assert!(!alias.deprecated);

        let alias = ConfigAlias::deprecated("old_key");
        assert_eq!(alias.key, "old_key");
        assert!(alias.deprecated);
    }

    #[test]
    fn test_aliases_default_returns_empty() {
        assert!(HudiTableConfig::TableName.aliases().is_empty());
    }

    #[test]
    fn test_resolve_raw_value_primary_key() {
        let mut configs = HashMap::new();
        configs.insert("hoodie.table.name".to_string(), "trips".to_string());
        let result = HudiTableConfig::TableName.resolve_raw_value(&configs);
        assert_eq!(result.unwrap(), "trips");
    }

    #[test]
    fn test_resolve_raw_value_deprecated_alias() {
        let mut configs = HashMap::new();
        configs.insert(
            "hoodie.table.precombine.field".to_string(),
            "ts".to_string(),
        );
        let result = HudiTableConfig::OrderingFields.resolve_raw_value(&configs);
        assert_eq!(result.unwrap(), "ts");
    }

    #[test]
    fn test_resolve_raw_value_not_found() {
        let configs = HashMap::new();
        let result = HudiTableConfig::TableName.resolve_raw_value(&configs);
        assert!(matches!(result.unwrap_err(), ConfigError::NotFound(_)));
    }

    #[test]
    fn test_try_get_returns_err_on_parse_failure() {
        let hudi_configs =
            HudiConfigs::new([(HudiTableConfig::PopulatesMetaFields.as_ref(), "not_a_bool")]);
        let result = hudi_configs.try_get(HudiTableConfig::PopulatesMetaFields);
        assert!(result.is_err());
    }

    #[test]
    fn test_new_using_hashmap() {
        let mut options = HashMap::new();
        options.insert("key1".to_string(), "value1".to_string());
        options.insert("key2".to_string(), "value2".to_string());

        let config = HudiConfigs::new(options.clone());

        assert_eq!(*config.raw_options, options);
    }

    #[test]
    fn test_new() {
        let options = vec![("key1", "value1"), ("key2", "value2")];

        let config = HudiConfigs::new(options);

        let expected: HashMap<String, String> = vec![
            ("key1".to_string(), "value1".to_string()),
            ("key2".to_string(), "value2".to_string()),
        ]
        .into_iter()
        .collect();

        assert_eq!(*config.raw_options, expected);
    }

    #[test]
    fn test_empty() {
        let config = HudiConfigs::empty();

        assert!(config.raw_options.is_empty());
    }

    #[test]
    fn test_as_options() {
        let mut options = HashMap::new();
        options.insert("key1".to_string(), "value1".to_string());
        options.insert("key2".to_string(), "value2".to_string());

        let config = HudiConfigs::new(options.clone());

        let result = config.as_options();

        assert_eq!(result, options);
    }
}