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 read configurations.

use std::collections::HashMap;
use std::fmt::Display;
use std::str::FromStr;

use strum_macros::{AsRefStr, EnumIter, IntoStaticStr};

use crate::config::Result;
use crate::config::error::ConfigError;
use crate::config::error::ConfigError::{InvalidValue, NotFound, ParseBool, ParseInt};
use crate::config::{ConfigParser, HudiConfigValue};

/// Config value for [`HudiReadConfig::QueryType`]. Canonical strings are
/// `snapshot` and `incremental`; [`FromStr`] accepts case-insensitive forms.
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, AsRefStr)]
pub enum QueryType {
    /// Latest table state at one commit (the latest by default; an explicit
    /// `as_of_timestamp` for time-travel).
    #[default]
    #[strum(serialize = "snapshot")]
    Snapshot,
    /// Records changed in the half-open range (`start_timestamp`, `end_timestamp`].
    #[strum(serialize = "incremental")]
    Incremental,
}

impl Display for QueryType {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.write_str(self.as_ref())
    }
}

impl FromStr for QueryType {
    type Err = ConfigError;

    fn from_str(s: &str) -> std::result::Result<Self, Self::Err> {
        match s.to_ascii_lowercase().as_str() {
            "snapshot" => Ok(Self::Snapshot),
            "incremental" => Ok(Self::Incremental),
            v => Err(InvalidValue(v.to_string())),
        }
    }
}

/// Configurations for reading Hudi tables.
///
/// **Example**
///
/// ```rust
/// use hudi_core::config::read::HudiReadConfig;
/// use hudi_core::table::ReadOptions;
///
/// let options = ReadOptions::new()
///     .with_hudi_option(HudiReadConfig::StreamBatchSize.as_ref(), "2048");
/// ```
///

#[derive(Clone, Debug, PartialEq, Eq, Hash, EnumIter, IntoStaticStr)]
pub enum HudiReadConfig {
    /// Selects the read semantic. Accepted values: `snapshot` (default), `incremental`.
    /// See [`crate::table::QueryType`].
    QueryType,

    /// Snapshot/time-travel timestamp. Reads return the table state at this commit.
    AsOfTimestamp,

    /// Start of an incremental window, exclusive.
    ///
    /// # Which timestamp this is
    ///
    /// On timeline layout v2 (table version 8+) this bounds a commit's
    /// **completion** time, matching Hudi 1.x — whose
    /// `IncrementalQueryAnalyzer` names the same bound `startCompletionTime`. A
    /// commit becomes visible when it completes, so that is what a window has to
    /// bound: one requested before the window but completing inside it *is* a
    /// change in that window, and bounding requested times skipped it silently.
    ///
    /// Layout v1 records no completion times, so there it bounds the requested
    /// time, as it always did.
    ///
    /// **Breaking change.** This used to bound requested times on every layout.
    /// Code that passes instant times read off the timeline — `Instant::timestamp`,
    /// or a `{requested}_{completion}` file name's first half — now excludes the
    /// commit it names, because that commit completed strictly later. Pass the
    /// completion half instead (`Instant::completion_timestamp`).
    StartTimestamp,

    /// End of an incremental window, inclusive. Bounds the same timestamp as
    /// [`Self::StartTimestamp`] — see there, including the breaking change.
    ///
    /// Defaults to the greatest completion time among completed commits, i.e.
    /// everything committed so far.
    EndTimestamp,

    /// Number of input partitions to read the data in parallel.
    ///
    /// For processing 100 files, [InputPartitions] being 5 will produce 5 partitions, with each partition having 20 files.
    InputPartitions,

    /// When set to true, only base files will be read for optimized reads.
    /// This is only applicable to Merge-On-Read (MOR) tables.
    UseReadOptimizedMode,
    /// Which implementation of the file group reader serves a read: `2`
    /// (default) or `1`.
    ///
    /// A read version 2 cannot serve is served by version 1 instead, so neither
    /// value can make a working read fail. Set `1` explicitly to opt out of the
    /// newer reader entirely.
    ///
    /// An unrecognised value is an error rather than a fall back to the default
    /// — silently reading with the other implementation would leave a caller
    /// convinced they had exercised the one they asked for.
    FileGroupReaderVersion,

    /// Target number of rows per batch for streaming reads.
    /// This controls the batch size when using streaming APIs.
    StreamBatchSize,

    /// Maximum number of file-slice streams to poll concurrently within one scan partition.
    FileSliceReadConcurrency,

    /// Total memory, in bytes, a whole scan may use for concurrent file-slice
    /// reads. Unset by default, which leaves
    /// [`Self::FileSliceReadConcurrency`] in charge on its own.
    ///
    /// This exists because that knob cannot express the peak. Real peak memory
    /// is the product of the per-merge budget, the slices open inside one engine
    /// partition, and the partitions running at once — three settings whose
    /// product nothing multiplies. Set this instead and the concurrency is
    /// derived from it, shared across partitions rather than granted to each.
    ///
    /// Reaching the limit lowers throughput; it never fails the read.
    ScanMaxMemorySize,

    /// Match a log record to the base row it updates by that row's position in
    /// the base file rather than by record key. Defaults to `false`.
    ///
    /// Only the writer knows whether it recorded positions; a log block that did
    /// not is merged by key regardless of this setting. The two agree except in a
    /// file group holding duplicate record keys, where merging by key cannot say
    /// which of them a log record updates.
    ///
    /// Honored only by file group reader version 2 — see
    /// [`Self::FileGroupReaderVersion`].
    MergeUseRecordPositions,
}

/// Where a [`HudiReadConfig`] may legitimately be set.
///
/// [`Table`](crate::table::Table) holds configuration for its whole lifetime,
/// while a read is a single call, so the two kinds cannot be treated alike: a
/// key that selects *which* read to perform would, baked at table level,
/// silently redirect every subsequent read.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum ReadConfigScope {
    /// Describes *how* to read. Stable for a deployment, so it may be set on the
    /// table (including in `hoodie.properties` or `hudi-defaults.conf`) and
    /// overridden per read.
    TableOrRead,
    /// Selects *which* read to perform. Meaningful only per call — see
    /// [`ReadOptions`](crate::table::ReadOptions).
    ReadOnly,
}

impl HudiReadConfig {
    /// Where this config may be set. See [`ReadConfigScope`].
    ///
    /// Exhaustive on purpose: a new variant has to choose, rather than inheriting
    /// whatever a prefix rule happens to do with its key.
    pub const fn scope(&self) -> ReadConfigScope {
        match self {
            // These name a point or a window in the timeline, or the query shape
            // itself. A table pinned to one of them would answer every later
            // read as at that point, which no caller asked for.
            Self::QueryType | Self::AsOfTimestamp | Self::StartTimestamp | Self::EndTimestamp => {
                ReadConfigScope::ReadOnly
            }
            // These are deployment choices: which file group reader runs, how
            // much parallelism to use, how big a streamed batch is. Setting them
            // once for a table is the natural way to use them.
            Self::InputPartitions
            | Self::UseReadOptimizedMode
            | Self::StreamBatchSize
            | Self::FileSliceReadConcurrency
            | Self::ScanMaxMemorySize
            | Self::FileGroupReaderVersion
            | Self::MergeUseRecordPositions => ReadConfigScope::TableOrRead,
        }
    }

    /// The scope of the read config named by `key`, or `None` when `key` is not
    /// a read config at all.
    pub fn scope_of_key(key: &str) -> Option<ReadConfigScope> {
        use strum::IntoEnumIterator;
        Self::iter()
            .find(|config| config.as_ref() == key)
            .map(|config| config.scope())
    }

    /// `&'static str` form of the config key. `const fn` so callers can use it in
    /// `const` contexts (e.g. building static lookup tables of `hoodie.*` keys
    /// without duplicating the literal strings).
    pub const fn key_str(&self) -> &'static str {
        match self {
            Self::QueryType => "hoodie.read.query.type",
            Self::AsOfTimestamp => "hoodie.read.as.of.timestamp",
            Self::StartTimestamp => "hoodie.read.start.timestamp",
            Self::EndTimestamp => "hoodie.read.end.timestamp",
            Self::InputPartitions => "hoodie.read.input.partitions",
            Self::UseReadOptimizedMode => "hoodie.read.use.read_optimized.mode",
            Self::FileGroupReaderVersion => "hoodie.read.file.group.reader.version",
            Self::StreamBatchSize => "hoodie.read.stream.batch_size",
            Self::FileSliceReadConcurrency => "hoodie.read.file.slice.read.concurrency",
            Self::ScanMaxMemorySize => "hoodie.read.scan.max.memory.size",
            // Hudi's own key, not a `hoodie.read.*` one: a table written with
            // record positions is read with this set, and a reader that invented
            // its own spelling would ignore what the writer was told.
            Self::MergeUseRecordPositions => "hoodie.merge.use.record.positions",
        }
    }
}

impl AsRef<str> for HudiReadConfig {
    fn as_ref(&self) -> &str {
        self.key_str()
    }
}

impl Display for HudiReadConfig {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.write_str(self.as_ref())
    }
}

/// Which implementation of the file group reader serves a read.
///
/// Numbered rather than named after a strategy, because the older one is being
/// retired rather than kept as an alternative: a version says newer supersedes
/// older, where a name like `batch_merge` would imply a permanent choice.
/// Matches how Hudi already versions `hoodie.table.version` and
/// `hoodie.timeline.layout.version`.
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub enum FileGroupReaderVersion {
    /// The reader that has always served reads: whole-batch sort and dedup.
    /// Reachable explicitly, as an escape hatch, and reached by fall back
    /// whenever [`Self::Two`] cannot serve a read.
    One,
    /// The merge-on-read reader being ported in, and the default.
    ///
    /// It does not serve every read yet. Anything it cannot serve is served by
    /// [`Self::One`] instead, which is why it can be the default this early:
    /// what changes a read is the reader gaining a capability, not this setting.
    #[default]
    Two,
}

impl FileGroupReaderVersion {
    /// The integer a caller writes in config.
    pub fn as_usize(&self) -> usize {
        match self {
            Self::One => 1,
            Self::Two => 2,
        }
    }
}

impl TryFrom<usize> for FileGroupReaderVersion {
    type Error = ConfigError;

    fn try_from(value: usize) -> std::result::Result<Self, Self::Error> {
        match value {
            1 => Ok(Self::One),
            2 => Ok(Self::Two),
            v => Err(InvalidValue(v.to_string())),
        }
    }
}

impl FromStr for FileGroupReaderVersion {
    type Err = ConfigError;

    fn from_str(s: &str) -> std::result::Result<Self, Self::Err> {
        match s.trim() {
            "1" => Ok(Self::One),
            "2" => Ok(Self::Two),
            v => Err(InvalidValue(v.to_string())),
        }
    }
}

impl ConfigParser for HudiReadConfig {
    type Output = HudiConfigValue;

    fn default_value(&self) -> Option<HudiConfigValue> {
        match self {
            HudiReadConfig::QueryType => Some(HudiConfigValue::String(
                QueryType::default().as_ref().to_string(),
            )),
            HudiReadConfig::InputPartitions => Some(HudiConfigValue::UInteger(0usize)),
            HudiReadConfig::UseReadOptimizedMode => Some(HudiConfigValue::Boolean(false)),
            HudiReadConfig::FileGroupReaderVersion => Some(HudiConfigValue::UInteger(
                FileGroupReaderVersion::default().as_usize(),
            )),
            HudiReadConfig::MergeUseRecordPositions => Some(HudiConfigValue::Boolean(false)),
            HudiReadConfig::StreamBatchSize => Some(HudiConfigValue::UInteger(1024usize)),
            HudiReadConfig::FileSliceReadConcurrency => Some(HudiConfigValue::UInteger(4usize)),
            _ => None,
        }
    }

    fn parse_value(&self, configs: &HashMap<String, String>) -> Result<Self::Output> {
        let get_result = configs
            .get(self.as_ref())
            .map(|v| v.as_str())
            .ok_or(NotFound(self.key()));

        match self {
            Self::QueryType => get_result
                .and_then(QueryType::from_str)
                .map(|v| HudiConfigValue::String(v.as_ref().to_string())),
            Self::AsOfTimestamp => get_result.map(|v| HudiConfigValue::String(v.to_string())),
            Self::StartTimestamp => get_result.map(|v| HudiConfigValue::String(v.to_string())),
            Self::EndTimestamp => get_result.map(|v| HudiConfigValue::String(v.to_string())),
            Self::InputPartitions => get_result
                .and_then(|v| {
                    usize::from_str(v).map_err(|e| ParseInt(self.key(), v.to_string(), e))
                })
                .map(HudiConfigValue::UInteger),
            Self::FileGroupReaderVersion => get_result
                .and_then(FileGroupReaderVersion::from_str)
                .map(|v| HudiConfigValue::UInteger(v.as_usize())),
            Self::UseReadOptimizedMode | Self::MergeUseRecordPositions => get_result
                .and_then(|v| {
                    bool::from_str(v).map_err(|e| ParseBool(self.key(), v.to_string(), e))
                })
                .map(HudiConfigValue::Boolean),
            Self::StreamBatchSize => get_result
                .and_then(|v| {
                    let key = self.key();
                    let parsed =
                        usize::from_str(v).map_err(|e| ParseInt(key.clone(), v.to_string(), e))?;
                    if parsed == 0 {
                        return Err(InvalidValue(format!("{key}=0 (must be > 0)")));
                    }
                    Ok(parsed)
                })
                .map(HudiConfigValue::UInteger),
            // Rejected at zero rather than silently meaning "unbounded": a
            // caller who sets a memory budget of 0 is asking for a bound, and
            // reading that as "no bound" is the opposite of what they asked.
            Self::ScanMaxMemorySize => get_result
                .and_then(|v| {
                    let key = self.key();
                    let parsed =
                        u64::from_str(v).map_err(|e| ParseInt(key.clone(), v.to_string(), e))?;
                    if parsed == 0 {
                        return Err(InvalidValue(format!("{key}=0 (must be > 0)")));
                    }
                    Ok(parsed as usize)
                })
                .map(HudiConfigValue::UInteger),
            Self::FileSliceReadConcurrency => get_result
                .and_then(|v| {
                    let key = self.key();
                    let parsed =
                        usize::from_str(v).map_err(|e| ParseInt(key.clone(), v.to_string(), e))?;
                    if parsed == 0 {
                        return Err(InvalidValue(format!("{key}=0 (must be > 0)")));
                    }
                    Ok(parsed)
                })
                .map(HudiConfigValue::UInteger),
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::config::read::HudiReadConfig::{
        AsOfTimestamp, EndTimestamp, FileSliceReadConcurrency, InputPartitions,
        QueryType as QueryTypeKey, StartTimestamp, StreamBatchSize, UseReadOptimizedMode,
    };

    #[test]
    fn parse_valid_config_value() {
        let options = HashMap::from([
            (QueryTypeKey.as_ref().to_string(), "Incremental".to_string()),
            (AsOfTimestamp.as_ref().to_string(), "20240101".to_string()),
            (StartTimestamp.as_ref().to_string(), "20240102".to_string()),
            (EndTimestamp.as_ref().to_string(), "20240103".to_string()),
            (InputPartitions.as_ref().to_string(), "100".to_string()),
            (
                UseReadOptimizedMode.as_ref().to_string(),
                "true".to_string(),
            ),
            (StreamBatchSize.as_ref().to_string(), "2048".to_string()),
            (
                FileSliceReadConcurrency.as_ref().to_string(),
                "8".to_string(),
            ),
        ]);
        let actual: String = QueryTypeKey.parse_value(&options).unwrap().into();
        assert_eq!(actual, "incremental");
        let actual: String = AsOfTimestamp.parse_value(&options).unwrap().into();
        assert_eq!(actual, "20240101");
        let actual: String = StartTimestamp.parse_value(&options).unwrap().into();
        assert_eq!(actual, "20240102");
        let actual: String = EndTimestamp.parse_value(&options).unwrap().into();
        assert_eq!(actual, "20240103");
        let actual: usize = InputPartitions.parse_value(&options).unwrap().into();
        assert_eq!(actual, 100);
        let actual: bool = UseReadOptimizedMode.parse_value(&options).unwrap().into();
        assert!(actual);
        let actual: usize = StreamBatchSize.parse_value(&options).unwrap().into();
        assert_eq!(actual, 2048);
        let actual: usize = FileSliceReadConcurrency
            .parse_value(&options)
            .unwrap()
            .into();
        assert_eq!(actual, 8);
    }

    #[test]
    fn parse_invalid_config_value() {
        let options = HashMap::from([
            (QueryTypeKey.as_ref().to_string(), "bogus".to_string()),
            (InputPartitions.as_ref().to_string(), "foo".to_string()),
            (UseReadOptimizedMode.as_ref().to_string(), "1".to_string()),
            (StreamBatchSize.as_ref().to_string(), "abc".to_string()),
            (
                FileSliceReadConcurrency.as_ref().to_string(),
                "abc".to_string(),
            ),
        ]);
        assert!(matches!(
            QueryTypeKey.parse_value(&options).unwrap_err(),
            InvalidValue(_)
        ));
        let actual: String = QueryTypeKey.parse_value_or_default(&options).into();
        assert_eq!(actual, "snapshot");
        assert!(matches!(
            InputPartitions.parse_value(&options).unwrap_err(),
            ParseInt(_, _, _)
        ));
        let actual: usize = InputPartitions.parse_value_or_default(&options).into();
        assert_eq!(actual, 0);
        assert!(matches!(
            UseReadOptimizedMode.parse_value(&options).unwrap_err(),
            ParseBool(_, _, _)
        ));
        let actual: bool = UseReadOptimizedMode.parse_value_or_default(&options).into();
        assert!(!actual);
        assert!(matches!(
            StreamBatchSize.parse_value(&options).unwrap_err(),
            ParseInt(_, _, _)
        ));
        let actual: usize = StreamBatchSize.parse_value_or_default(&options).into();
        assert_eq!(actual, 1024);
        assert!(matches!(
            FileSliceReadConcurrency.parse_value(&options).unwrap_err(),
            ParseInt(_, _, _)
        ));
        let actual: usize = FileSliceReadConcurrency
            .parse_value_or_default(&options)
            .into();
        assert_eq!(actual, 4);

        let zero = HashMap::from([(
            FileSliceReadConcurrency.as_ref().to_string(),
            "0".to_string(),
        )]);
        assert!(matches!(
            FileSliceReadConcurrency.parse_value(&zero).unwrap_err(),
            InvalidValue(_)
        ));
    }

    #[test]
    fn timestamp_keys_have_no_default_value() {
        assert!(AsOfTimestamp.default_value().is_none());
        assert!(StartTimestamp.default_value().is_none());
        assert!(EndTimestamp.default_value().is_none());
    }

    #[test]
    fn file_group_reader_version_try_from_usize_accepts_1_and_2_and_rejects_others() {
        assert_eq!(
            FileGroupReaderVersion::try_from(1).unwrap(),
            FileGroupReaderVersion::One
        );
        assert_eq!(
            FileGroupReaderVersion::try_from(2).unwrap(),
            FileGroupReaderVersion::Two
        );
        assert!(FileGroupReaderVersion::try_from(3).is_err());
    }

    #[test]
    fn query_type_from_str_accepts_case_insensitive_and_rejects_invalid() {
        assert_eq!(
            QueryType::from_str("snapshot").unwrap(),
            QueryType::Snapshot
        );
        assert_eq!(
            QueryType::from_str("SNAPSHOT").unwrap(),
            QueryType::Snapshot
        );
        assert_eq!(
            QueryType::from_str("Incremental").unwrap(),
            QueryType::Incremental
        );
        assert!(matches!(
            QueryType::from_str("bogus").unwrap_err(),
            InvalidValue(_)
        ));
    }

    #[test]
    fn display_impls_match_canonical_keys() {
        assert_eq!(format!("{}", QueryType::Snapshot), "snapshot");
        assert_eq!(format!("{}", QueryType::Incremental), "incremental");
        assert_eq!(
            format!("{}", HudiReadConfig::StreamBatchSize),
            "hoodie.read.stream.batch_size"
        );
        assert_eq!(
            format!("{}", HudiReadConfig::QueryType),
            "hoodie.read.query.type"
        );
    }
}