feat: add LSM timeline support with timeline layout management for old and new table versions

Cargo.toml

@@ -27,7 +27,7 @@ resolver = "2"
 version = "0.5.0-dev"
 edition = "2021"
 license = "Apache-2.0"
-rust-version = "1.84"
+rust-version = "1.85"
 keywords = ["apachehudi", "hudi", "datalake", "arrow"]
 readme = "README.md"
 description = "The native Rust implementation for Apache Hudi"

crates/core/src/config/internal.rs

@@ -37,18 +37,26 @@ use crate::config::{ConfigParser, HudiConfigValue};
 /// use hudi_core::table::Table as HudiTable;
 ///
 /// let options = [(SkipConfigValidation, "true")];
-/// HudiTable::new_with_options("/tmp/hudi_data", options)
+/// HudiTable::new_with_options_blocking("/tmp/hudi_data", options);
 /// ```
 ///
 #[derive(Clone, Debug, PartialEq, Eq, Hash, EnumIter)]
 pub enum HudiInternalConfig {
     SkipConfigValidation,
+    /// Enable reading archived timeline (v1) and LSM history (v2)
+    TimelineArchivedReadEnabled,
+    /// Behavior when archived timeline is requested but not available/enabled: "error" or "continue"
+    TimelineArchivedUnavailableBehavior,
 }
 
 impl AsRef<str> for HudiInternalConfig {
     fn as_ref(&self) -> &str {
         match self {
             Self::SkipConfigValidation => "hoodie.internal.skip.config.validation",
+            Self::TimelineArchivedReadEnabled => "hoodie.internal.timeline.archived.enabled",
+            Self::TimelineArchivedUnavailableBehavior => {
+                "hoodie.internal.timeline.archived.unavailable.behavior"
+            }
         }
     }
 }
@@ -65,6 +73,11 @@ impl ConfigParser for HudiInternalConfig {
     fn default_value(&self) -> Option<HudiConfigValue> {
         match self {
             Self::SkipConfigValidation => Some(HudiConfigValue::Boolean(false)),
+            Self::TimelineArchivedReadEnabled => Some(HudiConfigValue::Boolean(false)),
+            // default to continue (graceful) when archived is unavailable/disabled
+            Self::TimelineArchivedUnavailableBehavior => {
+                Some(HudiConfigValue::String("continue".to_string()))
+            }
         }
     }
 
@@ -80,6 +93,14 @@ impl ConfigParser for HudiInternalConfig {
                     bool::from_str(v).map_err(|e| ParseBool(self.key(), v.to_string(), e))
                 })
                 .map(HudiConfigValue::Boolean),
+            Self::TimelineArchivedReadEnabled => get_result
+                .and_then(|v| {
+                    bool::from_str(v).map_err(|e| ParseBool(self.key(), v.to_string(), e))
+                })
+                .map(HudiConfigValue::Boolean),
+            Self::TimelineArchivedUnavailableBehavior => {
+                get_result.map(|v| HudiConfigValue::String(v.to_string()))
+            }
         }
     }
 }

crates/core/src/config/read.rs

@@ -37,7 +37,7 @@ use crate::config::{ConfigParser, HudiConfigValue};
 /// use hudi_core::table::Table as HudiTable;
 ///
 /// let options = [(InputPartitions, "2")];
-/// HudiTable::new_with_options("/tmp/hudi_data", options)
+/// HudiTable::new_with_options_blocking("/tmp/hudi_data", options);
 /// ```
 ///
 

crates/core/src/config/table.rs

@@ -117,6 +117,12 @@ pub enum HudiTableConfig {
     ///
     /// - [`TimelineTimezoneValue`] - Possible values for this configuration.
     TimelineTimezone,
+
+    /// Folder for archived timeline files for layout v1 (default: .hoodie/archived)
+    ArchiveLogFolder,
+
+    /// Path for LSM timeline history for layout v2 (default: .hoodie/timeline/history)
+    TimelineHistoryPath,
 }
 
 impl AsRef<str> for HudiTableConfig {
@@ -141,6 +147,8 @@ impl AsRef<str> for HudiTableConfig {
             Self::TableVersion => "hoodie.table.version",
             Self::TimelineLayoutVersion => "hoodie.timeline.layout.version",
             Self::TimelineTimezone => "hoodie.table.timeline.timezone",
+            Self::ArchiveLogFolder => "hoodie.archivelog.folder",
+            Self::TimelineHistoryPath => "hoodie.timeline.history.path",
         }
     }
 }
@@ -166,6 +174,9 @@ impl ConfigParser for HudiTableConfig {
             Self::TimelineTimezone => Some(HudiConfigValue::String(
                 TimelineTimezoneValue::UTC.as_ref().to_string(),
             )),
+            // Fallbacks are handled in callers; no defaults here
+            Self::ArchiveLogFolder => None,
+            Self::TimelineHistoryPath => None,
             _ => None,
         }
     }
@@ -238,6 +249,8 @@ impl ConfigParser for HudiTableConfig {
             Self::TimelineTimezone => get_result
                 .and_then(TimelineTimezoneValue::from_str)
                 .map(|v| HudiConfigValue::String(v.as_ref().to_string())),
+            Self::ArchiveLogFolder => get_result.map(|v| HudiConfigValue::String(v.to_string())),
+            Self::TimelineHistoryPath => get_result.map(|v| HudiConfigValue::String(v.to_string())),
         }
     }
 

crates/core/src/file_group/reader.rs

@@ -75,8 +75,8 @@ impl FileGroupReader {
     /// Creates a new reader with the given base URI and options.
     ///
     /// # Arguments
-    ///     * `base_uri` - The base URI of the file group's residing table.
-    ///     * `options` - Additional options for the reader.
+    /// * `base_uri` - The base URI of the file group's residing table.
+    /// * `options` - Additional options for the reader.
     ///
     /// # Notes
     /// This API uses [`OptionResolver`] that loads table properties from storage to resolve options.
@@ -164,7 +164,7 @@ impl FileGroupReader {
     /// Reads the data from the base file at the given relative path.
     ///
     /// # Arguments
-    ///     * `relative_path` - The relative path to the base file.
+    /// * `relative_path` - The relative path to the base file.
     ///
     /// # Returns
     /// A record batch read from the base file.
@@ -216,7 +216,7 @@ impl FileGroupReader {
     /// Reads the data from the given file slice.
     ///
     /// # Arguments
-    ///     * `file_slice` - The file slice to read.
+    /// * `file_slice` - The file slice to read.
     ///
     /// # Returns
     /// A record batch read from the file slice.

crates/core/src/lib.rs

@@ -23,11 +23,11 @@
 //! **Example**
 //!
 //! ```rust
-//! use hudi_core::config::read::HudiReadConfig::{AsOfTimestamp, InputPartitions};
+//! use hudi_core::config::read::HudiReadConfig::InputPartitions;
 //! use hudi_core::table::Table as HudiTable;
 //!
-//! let options = [(InputPartitions, "2"), (AsOfTimestamp, "20240101010100000")];
-//! HudiTable::new_with_options("/tmp/hudi_data", options);
+//! let options = [(InputPartitions, "2")];
+//! HudiTable::new_with_options_blocking("/tmp/hudi_data", options);
 //! ```
 //!
 //! # The [table] module is responsible for managing Hudi tables.

crates/core/src/timeline/builder.rs

@@ -0,0 +1,101 @@
+/*
+ * 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.
+ */
+
+use crate::config::table::HudiTableConfig::{TableVersion, TimelineLayoutVersion};
+use crate::config::HudiConfigs;
+use crate::error::CoreError;
+use crate::storage::Storage;
+use crate::timeline::loader::TimelineLoader;
+use crate::timeline::Timeline;
+use crate::Result;
+use std::sync::Arc;
+
+pub struct TimelineBuilder {
+    hudi_configs: Arc<HudiConfigs>,
+    storage: Arc<Storage>,
+}
+
+impl TimelineBuilder {
+    pub fn new(hudi_configs: Arc<HudiConfigs>, storage: Arc<Storage>) -> Self {
+        Self {
+            hudi_configs,
+            storage,
+        }
+    }
+
+    pub async fn build(self) -> Result<Timeline> {
+        let (active_loader, archived_loader) = self.resolve_loader_config()?;
+        let timeline = Timeline::new(
+            self.hudi_configs,
+            self.storage,
+            active_loader,
+            archived_loader,
+        );
+        Ok(timeline)
+    }
+
+    fn resolve_loader_config(&self) -> Result<(TimelineLoader, Option<TimelineLoader>)> {
+        let table_version = match self.hudi_configs.get(TableVersion) {
+            Ok(v) => v.to::<isize>(),
+            Err(_) => {
+                return Ok((
+                    TimelineLoader::LayoutOneActive(self.storage.clone()),
+                    Some(TimelineLoader::LayoutOneArchived(self.storage.clone())),
+                ))
+            }
+        };
+
+        let layout_version = self
+            .hudi_configs
+            .try_get(TimelineLayoutVersion)
+            .map(|v| v.to::<isize>())
+            .unwrap_or_else(|| if table_version == 8 { 2 } else { 1 });
+
+        match layout_version {
+            1 => {
+                if !(6..=8).contains(&table_version) {
+                    return Err(CoreError::Unsupported(format!(
+                        "Unsupported table version {} with timeline layout version {}",
+                        table_version, layout_version
+                    )));
+                }
+                Ok((
+                    TimelineLoader::LayoutOneActive(self.storage.clone()),
+                    Some(TimelineLoader::LayoutOneArchived(self.storage.clone())),
+                ))
+            }
+            2 => {
+                if table_version != 8 {
+                    return Err(CoreError::Unsupported(format!(
+                        "Unsupported table version {} with timeline layout version {}",
+                        table_version, layout_version
+                    )));
+                }
+                Ok((
+                    TimelineLoader::LayoutTwoActive(self.storage.clone()),
+                    Some(TimelineLoader::LayoutTwoArchived(self.storage.clone())),
+                ))
+            }
+            _ => Err(CoreError::Unsupported(format!(
+                "Unsupported timeline layout version: {}",
+                layout_version
+            ))),
+        }
+    }
+}