This is an automated email from the ASF dual-hosted git repository.
xushiyan pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/hudi-rs.git
The following commit(s) were added to refs/heads/main by this push:
new f02cabf feat: add `TimelineSelector` to support timeline loading
(#233)
f02cabf is described below
commit f02cabfee4985cf9609690d70c852bb9fb9c0cd6
Author: Shiyan Xu <[email protected]>
AuthorDate: Fri Jan 3 20:01:13 2025 -0600
feat: add `TimelineSelector` to support timeline loading (#233)
---
Cargo.toml | 1 +
crates/core/Cargo.toml | 1 +
crates/core/src/config/table.rs | 59 ++++
crates/core/src/lib.rs | 1 +
crates/core/src/storage/util.rs | 20 --
crates/core/src/table/builder.rs | 2 +-
crates/core/src/table/mod.rs | 3 +-
crates/core/src/timeline/instant.rs | 379 +++++++++++++++++++++
.../src/{table/timeline.rs => timeline/mod.rs} | 148 +++-----
crates/core/src/timeline/selector.rs | 312 +++++++++++++++++
10 files changed, 802 insertions(+), 124 deletions(-)
diff --git a/Cargo.toml b/Cargo.toml
index 971e9eb..b5174d9 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -63,6 +63,7 @@ serde_json = { version = "1" }
# "stdlib"
thiserror = { version = "2.0.3" }
bytes = { version = "1" }
+chrono = { version = "0.4" }
paste = { version = "1.0.15" }
once_cell = { version = "1.19.0" }
strum = { version = "0.26.3", features = ["derive"] }
diff --git a/crates/core/Cargo.toml b/crates/core/Cargo.toml
index d7c9d06..25a5158 100644
--- a/crates/core/Cargo.toml
+++ b/crates/core/Cargo.toml
@@ -50,6 +50,7 @@ serde_json = { workspace = true }
# "stdlib"
thiserror = { workspace = true }
bytes = { workspace = true }
+chrono = { workspace = true }
paste = { workspace = true }
strum = { workspace = true }
strum_macros = { workspace = true }
diff --git a/crates/core/src/config/table.rs b/crates/core/src/config/table.rs
index 5b82b4c..c058582 100644
--- a/crates/core/src/config/table.rs
+++ b/crates/core/src/config/table.rs
@@ -103,6 +103,9 @@ pub enum HudiTableConfig {
/// Version of timeline used, by the table.
TimelineLayoutVersion,
+
+ /// Timezone of the timeline timestamps. Default to UTC.
+ TimelineTimezone,
}
impl AsRef<str> for HudiTableConfig {
@@ -124,6 +127,7 @@ impl AsRef<str> for HudiTableConfig {
Self::TableType => "hoodie.table.type",
Self::TableVersion => "hoodie.table.version",
Self::TimelineLayoutVersion => "hoodie.timeline.layout.version",
+ Self::TimelineTimezone => "hoodie.table.timeline.timezone",
}
}
}
@@ -137,6 +141,9 @@ impl ConfigParser for HudiTableConfig {
Self::DropsPartitionFields =>
Some(HudiConfigValue::Boolean(false)),
Self::PartitionFields => Some(HudiConfigValue::List(vec![])),
Self::PopulatesMetaFields => Some(HudiConfigValue::Boolean(true)),
+ Self::TimelineTimezone => Some(HudiConfigValue::String(
+ TimelineTimezoneValue::UTC.as_ref().to_string(),
+ )),
_ => None,
}
}
@@ -202,6 +209,9 @@ impl ConfigParser for HudiTableConfig {
isize::from_str(v).map_err(|e| ParseInt(self.key(),
v.to_string(), e))
})
.map(HudiConfigValue::Integer),
+ Self::TimelineTimezone => get_result
+ .and_then(TimelineTimezoneValue::from_str)
+ .map(|v| HudiConfigValue::String(v.as_ref().to_string())),
}
}
}
@@ -246,6 +256,27 @@ impl FromStr for BaseFileFormatValue {
}
}
+/// Config value for [HudiTableConfig::TimelineTimezone].
+#[derive(Clone, Debug, PartialEq, AsRefStr)]
+pub enum TimelineTimezoneValue {
+ #[strum(serialize = "utc")]
+ UTC,
+ #[strum(serialize = "local")]
+ Local,
+}
+
+impl FromStr for TimelineTimezoneValue {
+ type Err = ConfigError;
+
+ fn from_str(s: &str) -> Result<Self, Self::Err> {
+ match s.to_ascii_lowercase().as_str() {
+ "utc" => Ok(Self::UTC),
+ "local" => Ok(Self::Local),
+ v => Err(InvalidValue(v.to_string())),
+ }
+ }
+}
+
#[cfg(test)]
mod tests {
use super::*;
@@ -313,4 +344,32 @@ mod tests {
UnsupportedValue(_)
));
}
+
+ #[test]
+ fn create_timeline_timezone() {
+ assert_eq!(
+ TimelineTimezoneValue::from_str("utc").unwrap(),
+ TimelineTimezoneValue::UTC
+ );
+ assert_eq!(
+ TimelineTimezoneValue::from_str("uTc").unwrap(),
+ TimelineTimezoneValue::UTC
+ );
+ assert_eq!(
+ TimelineTimezoneValue::from_str("local").unwrap(),
+ TimelineTimezoneValue::Local
+ );
+ assert_eq!(
+ TimelineTimezoneValue::from_str("LOCAL").unwrap(),
+ TimelineTimezoneValue::Local
+ );
+ assert!(matches!(
+ TimelineTimezoneValue::from_str("").unwrap_err(),
+ InvalidValue(_)
+ ));
+ assert!(matches!(
+ TimelineTimezoneValue::from_str("foo").unwrap_err(),
+ InvalidValue(_)
+ ));
+ }
}
diff --git a/crates/core/src/lib.rs b/crates/core/src/lib.rs
index c9ce815..58f7417 100644
--- a/crates/core/src/lib.rs
+++ b/crates/core/src/lib.rs
@@ -49,6 +49,7 @@ pub mod expr;
pub mod file_group;
pub mod storage;
pub mod table;
+pub mod timeline;
pub mod util;
use error::Result;
diff --git a/crates/core/src/storage/util.rs b/crates/core/src/storage/util.rs
index e3f32b9..b26e865 100644
--- a/crates/core/src/storage/util.rs
+++ b/crates/core/src/storage/util.rs
@@ -18,31 +18,11 @@
*/
//! Utility functions for storage.
-use std::path::Path;
use url::Url;
use crate::storage::error::StorageError::{InvalidPath, UrlParseError};
use crate::storage::Result;
-/// Splits a filename into a stem and an extension.
-pub fn split_filename(filename: &str) -> Result<(String, String)> {
- let path = Path::new(filename);
-
- let stem = path
- .file_stem()
- .and_then(|s| s.to_str())
- .ok_or(InvalidPath(format!("No file stem found in {}", filename)))?
- .to_string();
-
- let extension = path
- .extension()
- .and_then(|e| e.to_str())
- .unwrap_or_default()
- .to_string();
-
- Ok((stem, extension))
-}
-
/// Parses a URI string into a URL.
pub fn parse_uri(uri: &str) -> Result<Url> {
let mut url = match Url::parse(uri) {
diff --git a/crates/core/src/table/builder.rs b/crates/core/src/table/builder.rs
index 4ca2997..411ff09 100644
--- a/crates/core/src/table/builder.rs
+++ b/crates/core/src/table/builder.rs
@@ -38,8 +38,8 @@ use crate::config::{HudiConfigs, HUDI_CONF_DIR};
use crate::error::CoreError;
use crate::storage::Storage;
use crate::table::fs_view::FileSystemView;
-use crate::table::timeline::Timeline;
use crate::table::Table;
+use crate::timeline::Timeline;
use crate::Result;
/// Builder for creating a [Table] instance.
diff --git a/crates/core/src/table/mod.rs b/crates/core/src/table/mod.rs
index f83aa91..3f6b2c2 100644
--- a/crates/core/src/table/mod.rs
+++ b/crates/core/src/table/mod.rs
@@ -87,7 +87,6 @@
pub mod builder;
mod fs_view;
pub mod partition;
-mod timeline;
use crate::config::read::HudiReadConfig::AsOfTimestamp;
use crate::config::table::HudiTableConfig;
@@ -100,7 +99,7 @@ use crate::file_group::FileSlice;
use crate::table::builder::TableBuilder;
use crate::table::fs_view::FileSystemView;
use crate::table::partition::PartitionPruner;
-use crate::table::timeline::Timeline;
+use crate::timeline::Timeline;
use crate::Result;
use arrow::record_batch::RecordBatch;
diff --git a/crates/core/src/timeline/instant.rs
b/crates/core/src/timeline/instant.rs
new file mode 100644
index 0000000..f089e90
--- /dev/null
+++ b/crates/core/src/timeline/instant.rs
@@ -0,0 +1,379 @@
+/*
+ * 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::TimelineTimezoneValue;
+use crate::error::CoreError;
+use crate::storage::error::StorageError;
+use crate::Result;
+use chrono::{DateTime, Local, NaiveDateTime, TimeZone, Timelike, Utc};
+use std::cmp::Ordering;
+use std::path::PathBuf;
+use std::str::FromStr;
+
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub enum Action {
+ Commit,
+ ReplaceCommit,
+}
+
+impl FromStr for Action {
+ type Err = CoreError;
+
+ fn from_str(s: &str) -> Result<Self> {
+ match s {
+ "commit" => Ok(Action::Commit),
+ "replacecommit" => Ok(Action::ReplaceCommit),
+ _ => Err(CoreError::Timeline(format!("Invalid action: {}", s))),
+ }
+ }
+}
+
+impl Action {
+ pub fn as_str(&self) -> &str {
+ match self {
+ Action::Commit => "commit",
+ Action::ReplaceCommit => "replacecommit",
+ }
+ }
+
+ pub fn is_replacecommit(&self) -> bool {
+ self == &Action::ReplaceCommit
+ }
+}
+
+/// The [State] of an [Instant] represents the status of the action performed
on the table.
+#[allow(dead_code)]
+#[derive(Clone, Debug, Eq, PartialEq, Ord, PartialOrd)]
+pub enum State {
+ Requested,
+ Inflight,
+ Completed,
+}
+
+impl FromStr for State {
+ type Err = CoreError;
+
+ fn from_str(suffix: &str) -> Result<Self> {
+ match suffix {
+ "requested" => Ok(State::Requested),
+ "inflight" => Ok(State::Inflight),
+ "" => Ok(State::Completed),
+ _ => Err(CoreError::Timeline(format!(
+ "Invalid state suffix: {}",
+ suffix
+ ))),
+ }
+ }
+}
+
+impl State {
+ pub fn as_str(&self) -> &str {
+ match self {
+ State::Requested => "requested",
+ State::Inflight => "inflight",
+ State::Completed => "",
+ }
+ }
+}
+
+/// An [Instant] represents a point in time when an action was performed on
the table.
+#[allow(dead_code)]
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub struct Instant {
+ pub timestamp: String,
+ pub action: Action,
+ pub state: State,
+ pub epoch_millis: i64,
+}
+
+impl PartialOrd for Instant {
+ fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
+ Some(self.cmp(other))
+ }
+}
+
+impl Ord for Instant {
+ fn cmp(&self, other: &Self) -> Ordering {
+ self.epoch_millis
+ .cmp(&other.epoch_millis)
+ .then_with(|| self.state.cmp(&other.state))
+ }
+}
+
+impl TryFrom<&str> for Instant {
+ type Error = CoreError;
+
+ /// Parse a timeline file name into an [Instant]. Timezone is assumed to
be UTC.
+ fn try_from(file_name: &str) -> Result<Self> {
+ Self::try_from((file_name, "UTC"))
+ }
+}
+
+impl TryFrom<(&str, &str)> for Instant {
+ type Error = CoreError;
+
+ /// Parse a timeline file name and a timezone value into an [Instant].
+ /// See [TimelineTimezoneValue] for valid timezone values.
+ fn try_from(file_name_and_timezone: (&str, &str)) -> Result<Self> {
+ let (file_name, timezone) = file_name_and_timezone;
+ let (timestamp, action_suffix) = file_name
+ .split_once('.')
+ .ok_or_else(|| CoreError::Timeline(format!("Invalid file name:
{}", file_name)))?;
+ Self::validate_timestamp(timestamp)?;
+ let dt = Self::parse_datetime(timestamp, timezone)?;
+ let (action, state) = Self::parse_action_and_state(action_suffix)?;
+
+ Ok(Self {
+ timestamp: timestamp.to_string(),
+ state,
+ action,
+ epoch_millis: dt.timestamp_millis(),
+ })
+ }
+}
+
+impl Instant {
+ fn validate_timestamp(timestamp: &str) -> Result<()> {
+ if !matches!(timestamp.len(), 14 | 17) {
+ return Err(CoreError::Timeline(format!(
+ "Timestamp must be in format yyyyMMddHHmmss or
yyyyMMddHHmmssSSS, but got: {}",
+ timestamp
+ )));
+ }
+ Ok(())
+ }
+
+ /// Convert a timeline timestamp string in UTC to epoch milliseconds.
+ pub fn parse_utc_as_epoch_millis(timestamp: &str) -> Result<i64> {
+ Self::parse_datetime(timestamp, "UTC").map(|dt| dt.timestamp_millis())
+ }
+
+ pub fn parse_datetime(timestamp: &str, timezone: &str) ->
Result<DateTime<Utc>> {
+ let naive_dt = Self::parse_naive_datetime(timestamp)?;
+ Self::convert_to_timezone(naive_dt, timezone)
+ }
+
+ fn parse_naive_datetime(timestamp: &str) -> Result<NaiveDateTime> {
+ let naive_dt = NaiveDateTime::parse_from_str(×tamp[..14],
"%Y%m%d%H%M%S")
+ .map_err(|e| CoreError::Timeline(format!("Failed to parse
timestamp: {}", e)))?;
+
+ if timestamp.len() == 17 {
+ let millis: u32 = timestamp[14..]
+ .parse()
+ .map_err(|e| CoreError::Timeline(format!("Failed to parse
milliseconds: {}", e)))?;
+ naive_dt
+ .with_nanosecond(millis * 1_000_000)
+ .ok_or_else(|| CoreError::Timeline("Invalid
milliseconds".to_string()))
+ } else {
+ Ok(naive_dt)
+ }
+ }
+
+ fn convert_to_timezone(naive_dt: NaiveDateTime, timezone: &str) ->
Result<DateTime<Utc>> {
+ match TimelineTimezoneValue::from_str(timezone)? {
+ TimelineTimezoneValue::UTC =>
Ok(DateTime::from_naive_utc_and_offset(naive_dt, Utc)),
+ TimelineTimezoneValue::Local => Ok(Local
+ .from_local_datetime(&naive_dt)
+ .earliest()
+ .ok_or_else(|| CoreError::Timeline("Invalid local
datetime".to_string()))?
+ .to_utc()),
+ }
+ }
+
+ fn parse_action_and_state(action_suffix: &str) -> Result<(Action, State)> {
+ match action_suffix.split_once('.') {
+ Some((action_str, state_str)) => {
+ Ok((Action::from_str(action_str)?,
State::from_str(state_str)?))
+ }
+ None => match action_suffix {
+ "inflight" => Ok((Action::Commit, State::Inflight)),
+ action_str => Ok((Action::from_str(action_str)?,
State::Completed)),
+ },
+ }
+ }
+
+ pub fn file_name(&self) -> String {
+ match (&self.action, &self.state) {
+ (_, State::Completed) => format!("{}.{}", self.timestamp,
self.action.as_str()),
+ (Action::Commit, State::Inflight) => {
+ format!("{}.{}", self.timestamp, self.state.as_str())
+ }
+ _ => format!(
+ "{}.{}.{}",
+ self.timestamp,
+ self.action.as_str(),
+ self.state.as_str()
+ ),
+ }
+ }
+
+ pub fn relative_path(&self) -> Result<String> {
+ let mut commit_file_path = PathBuf::from(".hoodie");
+ commit_file_path.push(self.file_name());
+ commit_file_path
+ .to_str()
+ .ok_or(StorageError::InvalidPath(format!(
+ "Failed to get file path for {:?}",
+ self
+ )))
+ .map_err(CoreError::Storage)
+ .map(|s| s.to_string())
+ }
+
+ pub fn is_replacecommit(&self) -> bool {
+ self.action.is_replacecommit()
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ #[test]
+ fn test_action_methods() {
+ assert_eq!(Action::Commit.as_str(), "commit");
+ assert_eq!(Action::ReplaceCommit.as_str(), "replacecommit");
+
+ assert!(!Action::Commit.is_replacecommit());
+ assert!(Action::ReplaceCommit.is_replacecommit());
+ }
+
+ #[test]
+ fn test_action_from_str() {
+ assert_eq!(Action::from_str("commit").unwrap(), Action::Commit);
+ assert_eq!(
+ Action::from_str("replacecommit").unwrap(),
+ Action::ReplaceCommit
+ );
+ assert!(Action::from_str("invalid").is_err());
+ }
+
+ #[test]
+ fn test_state_methods() {
+ assert_eq!(State::Requested.as_str(), "requested");
+ assert_eq!(State::Inflight.as_str(), "inflight");
+ assert_eq!(State::Completed.as_str(), "");
+ }
+
+ #[test]
+ fn test_state_from_str() {
+ assert_eq!(State::from_str("requested").unwrap(), State::Requested);
+ assert_eq!(State::from_str("inflight").unwrap(), State::Inflight);
+ assert_eq!(State::from_str("").unwrap(), State::Completed);
+ assert!(State::from_str("invalid").is_err());
+ }
+
+ #[test]
+ fn test_state_ordering() {
+ assert!(State::Requested < State::Inflight);
+ assert!(State::Inflight < State::Completed);
+ assert!(State::Requested < State::Completed);
+
+ assert!(State::Completed > State::Inflight);
+ assert!(State::Inflight > State::Requested);
+ assert!(State::Completed > State::Requested);
+ }
+
+ #[test]
+ fn test_instant_from_file_name() -> Result<()> {
+ // Test completed commit
+ let instant = Instant::try_from("20240103153000.commit")?;
+ assert_eq!(instant.timestamp, "20240103153000");
+ assert_eq!(instant.action, Action::Commit);
+ assert_eq!(instant.state, State::Completed);
+
+ // Test inflight replacecommit with milliseconds
+ let instant =
Instant::try_from("20240103153000123.replacecommit.inflight")?;
+ assert_eq!(instant.timestamp, "20240103153000123");
+ assert_eq!(instant.action, Action::ReplaceCommit);
+ assert_eq!(instant.state, State::Inflight);
+
+ Ok(())
+ }
+
+ #[test]
+ fn test_invalid_file_names() {
+ // Invalid timestamp format
+ assert!(Instant::try_from("2024010315.commit").is_err());
+
+ // Invalid action
+ assert!(Instant::try_from("20240103153000.invalid").is_err());
+
+ // Invalid state
+ assert!(Instant::try_from("20240103153000.commit.invalid").is_err());
+
+ // No dot separator
+ assert!(Instant::try_from("20240103153000commit").is_err());
+ }
+
+ #[test]
+ fn test_file_name_roundtrip() -> Result<()> {
+ let test_cases = vec![
+ "20240103153000.commit",
+ "20240103153000.commit.requested",
+ "20240103153000.replacecommit",
+ "20240103153000123.inflight",
+ ];
+
+ for original_name in test_cases {
+ let instant = Instant::try_from(original_name)?;
+ assert_eq!(instant.file_name(), original_name);
+ }
+
+ Ok(())
+ }
+
+ #[test]
+ fn test_instant_ordering() -> Result<()> {
+ let instant1 = Instant::try_from("20240103153000.commit")?;
+ let instant2 = Instant::try_from("20240103153000001.commit")?;
+ let instant3 =
Instant::try_from("20240103153000999.commit.requested")?;
+ let instant4 = Instant::try_from("20240103153000999.inflight")?;
+ let instant5 = Instant::try_from("20240103153000999.commit")?;
+ let instant6 = Instant::try_from("20240103153001.commit")?;
+
+ assert!(instant1 < instant2);
+ assert!(instant2 < instant3);
+ assert!(instant3 < instant4);
+ assert!(instant4 < instant5);
+ assert!(instant5 < instant6);
+
+ Ok(())
+ }
+
+ #[test]
+ fn test_relative_path() {
+ let instant = Instant::try_from("20240103153000.commit").unwrap();
+ assert_eq!(
+ instant.relative_path().unwrap(),
+ ".hoodie/20240103153000.commit"
+ );
+ }
+
+ #[test]
+ fn test_create_instant_using_local_timezone() {
+ let file_name = "20240103153000.commit";
+ let instant_local = Instant::try_from((file_name, "local")).unwrap();
+ let instant_utc = Instant::try_from((file_name, "utc")).unwrap();
+ let offset_seconds = (instant_local.epoch_millis -
instant_utc.epoch_millis) / 1000;
+ assert_eq!(
+ offset_seconds,
+ Local::now().offset().utc_minus_local() as i64
+ );
+ }
+}
diff --git a/crates/core/src/table/timeline.rs b/crates/core/src/timeline/mod.rs
similarity index 72%
rename from crates/core/src/table/timeline.rs
rename to crates/core/src/timeline/mod.rs
index d48aa55..c414a17 100644
--- a/crates/core/src/table/timeline.rs
+++ b/crates/core/src/timeline/mod.rs
@@ -16,84 +16,23 @@
* specific language governing permissions and limitations
* under the License.
*/
+mod instant;
+mod selector;
-use std::cmp::{Ordering, PartialOrd};
-use std::collections::{HashMap, HashSet};
-use std::fmt::Debug;
-use std::path::PathBuf;
-use std::sync::Arc;
-
-use arrow_schema::Schema;
-use parquet::arrow::parquet_to_arrow_schema;
-use serde_json::{Map, Value};
-
+use crate::config::table::HudiTableConfig;
use crate::config::HudiConfigs;
use crate::error::CoreError;
use crate::file_group::FileGroup;
-use crate::storage::error::StorageError;
-use crate::storage::util::split_filename;
use crate::storage::Storage;
+use crate::timeline::selector::TimelineSelector;
use crate::Result;
-
-/// The [State] of an [Instant] represents the status of the action performed
on the table.
-#[allow(dead_code)]
-#[derive(Clone, Debug, Eq, PartialEq)]
-pub enum State {
- Requested,
- Inflight,
- Completed,
-}
-
-/// An [Instant] represents a point in time when an action was performed on
the table.
-#[derive(Clone, Debug, Eq, PartialEq)]
-pub struct Instant {
- state: State,
- action: String,
- timestamp: String,
-}
-
-impl PartialOrd for Instant {
- fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
- Some(self.timestamp.cmp(&other.timestamp))
- }
-}
-
-impl Ord for Instant {
- fn cmp(&self, other: &Self) -> Ordering {
- self.timestamp.cmp(&other.timestamp)
- }
-}
-
-impl Instant {
- pub fn state_suffix(&self) -> String {
- match self.state {
- State::Requested => ".requested".to_owned(),
- State::Inflight => ".inflight".to_owned(),
- State::Completed => "".to_owned(),
- }
- }
-
- pub fn file_name(&self) -> String {
- format!("{}.{}{}", self.timestamp, self.action, self.state_suffix())
- }
-
- pub fn relative_path(&self) -> Result<String> {
- let mut commit_file_path = PathBuf::from(".hoodie");
- commit_file_path.push(self.file_name());
- commit_file_path
- .to_str()
- .ok_or(StorageError::InvalidPath(format!(
- "Failed to get file path for {:?}",
- self
- )))
- .map_err(CoreError::Storage)
- .map(|s| s.to_string())
- }
-
- pub fn is_replacecommit(&self) -> bool {
- self.action == "replacecommit"
- }
-}
+use arrow_schema::Schema;
+use instant::Instant;
+use parquet::arrow::parquet_to_arrow_schema;
+use serde_json::{Map, Value};
+use std::collections::{HashMap, HashSet};
+use std::fmt::Debug;
+use std::sync::Arc;
/// A [Timeline] contains transaction logs of all actions performed on the
table at different [Instant]s of time.
#[derive(Clone, Debug)]
@@ -110,7 +49,7 @@ impl Timeline {
storage_options: Arc<HashMap<String, String>>,
) -> Result<Self> {
let storage = Storage::new(storage_options.clone(),
hudi_configs.clone())?;
- let instants = Self::load_completed_commits(&storage).await?;
+ let instants = Self::load_active_completed_commits(&storage,
hudi_configs.clone()).await?;
Ok(Self {
storage,
hudi_configs,
@@ -118,16 +57,39 @@ impl Timeline {
})
}
- async fn load_completed_commits(storage: &Storage) -> Result<Vec<Instant>>
{
+ pub async fn from_instants(
+ hudi_configs: Arc<HudiConfigs>,
+ storage_options: Arc<HashMap<String, String>>,
+ instants: Vec<Instant>,
+ ) -> Result<Self> {
+ let storage = Storage::new(storage_options.clone(),
hudi_configs.clone())?;
+ Ok(Self {
+ storage,
+ hudi_configs,
+ instants,
+ })
+ }
+
+ async fn load_active_completed_commits(
+ storage: &Storage,
+ hudi_configs: Arc<HudiConfigs>,
+ ) -> Result<Vec<Instant>> {
let mut completed_commits = Vec::new();
+ let selector = TimelineSelector::active_completed_commits();
+ let timezone = hudi_configs
+ .get_or_default(HudiTableConfig::TimelineTimezone)
+ .to::<String>();
for file_info in storage.list_files(Some(".hoodie")).await? {
- let (file_stem, file_ext) =
split_filename(file_info.name.as_str())?;
- if matches!(file_ext.as_str(), "commit" | "replacecommit") {
- completed_commits.push(Instant {
- state: State::Completed,
- timestamp: file_stem,
- action: file_ext.to_owned(),
- })
+ match Instant::try_from((file_info.name.as_str(),
timezone.as_str())) {
+ Ok(instant) => {
+ if selector.should_select(&instant) {
+ completed_commits.push(instant)
+ }
+ }
+ Err(_) => {
+ // Ignore files like `hoodie.properties` that are not
valid timeline files
+ continue;
+ }
}
}
completed_commits.sort();
@@ -274,16 +236,8 @@ mod tests {
assert_eq!(
timeline.instants,
vec![
- Instant {
- state: State::Completed,
- action: "commit".to_owned(),
- timestamp: "20240402123035233".to_owned(),
- },
- Instant {
- state: State::Completed,
- action: "commit".to_owned(),
- timestamp: "20240402144910683".to_owned(),
- },
+ Instant::try_from("20240402123035233.commit").unwrap(),
+ Instant::try_from("20240402144910683.commit").unwrap(),
]
)
}
@@ -298,11 +252,7 @@ mod tests {
)
.unwrap();
let timeline = create_test_timeline(base_url).await;
- let instant = Instant {
- state: State::Completed,
- action: "commit".to_owned(),
- timestamp: "20240402123035233".to_owned(),
- };
+ let instant = Instant::try_from("20240402123035233.commit").unwrap();
// Test error when reading empty commit metadata file
let result = timeline.get_commit_metadata(&instant).await;
@@ -311,11 +261,7 @@ mod tests {
assert!(matches!(err, CoreError::Timeline(_)));
assert!(err.to_string().contains("Failed to get commit metadata"));
- let instant = Instant {
- state: State::Completed,
- action: "commit".to_owned(),
- timestamp: "20240402144910683".to_owned(),
- };
+ let instant = Instant::try_from("20240402144910683.commit").unwrap();
// Test error when reading a commit metadata file with invalid JSON
let result = timeline.get_commit_metadata(&instant).await;
diff --git a/crates/core/src/timeline/selector.rs
b/crates/core/src/timeline/selector.rs
new file mode 100644
index 0000000..a4c76d1
--- /dev/null
+++ b/crates/core/src/timeline/selector.rs
@@ -0,0 +1,312 @@
+/*
+ * 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::timeline::instant::{Action, Instant, State};
+use crate::timeline::Timeline;
+use crate::Result;
+
+#[allow(dead_code)]
+#[derive(Debug, Clone, PartialEq)]
+pub struct TimelineSelector {
+ states: Vec<State>,
+ actions: Vec<Action>,
+ start_timestamp: Option<i64>,
+ end_timestamp: Option<i64>,
+ include_archived: bool,
+}
+
+impl Default for TimelineSelector {
+ fn default() -> Self {
+ Self::empty()
+ }
+}
+
+#[allow(dead_code)]
+impl TimelineSelector {
+ /// An empty [TimelineSelector] that selects no instants.
+ pub fn empty() -> Self {
+ Self {
+ states: vec![],
+ start_timestamp: None,
+ end_timestamp: None,
+ actions: vec![],
+ include_archived: false,
+ }
+ }
+
+ pub fn active_completed_commits() -> Self {
+ Self {
+ states: vec![State::Completed],
+ start_timestamp: None,
+ end_timestamp: None,
+ actions: vec![Action::Commit, Action::ReplaceCommit],
+ include_archived: false,
+ }
+ }
+
+ /// Check if a single instant should be selected based on the given
criteria.
+ pub fn should_select(&self, instant: &Instant) -> bool {
+ if let Some(start) = self.start_timestamp {
+ if instant.epoch_millis < start {
+ return false;
+ }
+ }
+
+ if let Some(end) = self.end_timestamp {
+ if instant.epoch_millis >= end {
+ return false;
+ }
+ }
+
+ if !self.actions.contains(&instant.action) {
+ return false;
+ }
+
+ if !self.states.contains(&instant.state) {
+ return false;
+ }
+
+ true
+ }
+
+ /// Select loaded instants based on the given criteria.
+ pub fn select(&self, timeline: &Timeline) -> Result<Vec<Instant>> {
+ if self.states.is_empty() || self.actions.is_empty() {
+ return Ok(vec![]);
+ }
+
+ let time_pruned_instants = if let Some(start) = self.start_timestamp {
+ // Find first instant >= start using binary search
+ let start_pos = timeline
+ .instants
+ .partition_point(|instant| instant.epoch_millis < start);
+
+ if let Some(end) = self.end_timestamp {
+ // Find first instant >= end using binary search
+ let end_pos = timeline.instants[start_pos..]
+ .partition_point(|instant| instant.epoch_millis < end);
+ &timeline.instants[start_pos..start_pos + end_pos]
+ } else {
+ &timeline.instants[start_pos..]
+ }
+ } else if let Some(end) = self.end_timestamp {
+ let end_pos = timeline
+ .instants
+ .partition_point(|instant| instant.epoch_millis < end);
+ &timeline.instants[..end_pos]
+ } else {
+ &timeline.instants[..]
+ };
+
+ Ok(time_pruned_instants
+ .iter()
+ .filter(|instant| {
+ self.actions.contains(&instant.action) &&
self.states.contains(&instant.state)
+ })
+ .cloned()
+ .collect())
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use crate::config::table::HudiTableConfig;
+ use crate::config::HudiConfigs;
+ use hudi_tests::assert_not;
+ use std::collections::HashMap;
+ use std::sync::Arc;
+
+ fn create_test_instant(file_name: &str) -> Instant {
+ Instant::try_from(file_name).unwrap()
+ }
+
+ async fn create_test_timeline() -> Timeline {
+ let instants = vec![
+ create_test_instant("20240103153000.commit"),
+ create_test_instant("20240103153010999.commit"),
+ create_test_instant("20240103153020999.commit.requested"),
+ create_test_instant("20240103153020999.inflight"),
+ create_test_instant("20240103153020999.commit"),
+ create_test_instant("20240103153030999.commit"),
+ ];
+ Timeline::from_instants(
+ Arc::new(HudiConfigs::new([(
+ HudiTableConfig::BasePath,
+ "file:///tmp/base",
+ )])),
+ Arc::new(HashMap::new()),
+ instants,
+ )
+ .await
+ .unwrap()
+ }
+
+ #[test]
+ fn test_default() {
+ assert_eq!(TimelineSelector::default(), TimelineSelector::empty());
+ }
+
+ #[test]
+ fn test_should_select() {
+ let instant = create_test_instant("20240103153030999.commit");
+
+ let selector = TimelineSelector {
+ states: vec![State::Completed],
+ actions: vec![Action::Commit],
+ start_timestamp: None,
+ end_timestamp: None,
+ include_archived: false,
+ };
+ assert!(selector.should_select(&instant));
+
+ let selector = TimelineSelector {
+ states: vec![State::Requested],
+ actions: vec![Action::Commit],
+ start_timestamp: None,
+ end_timestamp: None,
+ include_archived: false,
+ };
+ assert_not!(
+ selector.should_select(&instant),
+ "Should not select as state is different"
+ );
+
+ let selector = TimelineSelector {
+ states: vec![State::Completed],
+ actions: vec![Action::ReplaceCommit],
+ start_timestamp: None,
+ end_timestamp: None,
+ include_archived: false,
+ };
+ assert_not!(
+ selector.should_select(&instant),
+ "Should not select as action is different"
+ );
+
+ let selector = TimelineSelector {
+ states: vec![State::Completed],
+ actions: vec![Action::Commit],
+ start_timestamp:
Instant::parse_utc_as_epoch_millis("20240103153031").ok(),
+ end_timestamp: None,
+ include_archived: false,
+ };
+ assert_not!(
+ selector.should_select(&instant),
+ "Should not select as timestamp is before start"
+ );
+
+ let selector = TimelineSelector {
+ states: vec![State::Completed],
+ actions: vec![Action::Commit],
+ start_timestamp: None,
+ end_timestamp:
Instant::parse_utc_as_epoch_millis("20240103153030999").ok(),
+ include_archived: false,
+ };
+ assert_not!(
+ selector.should_select(&instant),
+ "Should not select as timestamp is at the end timestamp
(exclusive)"
+ );
+ }
+
+ #[tokio::test]
+ async fn test_select_no_instants() {
+ let timeline = create_test_timeline().await;
+ assert_not!(timeline.instants.is_empty());
+
+ let selector = TimelineSelector::empty();
+ assert!(selector.select(&timeline).unwrap().is_empty());
+
+ let selector = TimelineSelector {
+ states: vec![],
+ actions: vec![Action::Commit],
+ start_timestamp: None,
+ end_timestamp: None,
+ include_archived: false,
+ };
+ assert!(selector.select(&timeline).unwrap().is_empty());
+
+ let selector = TimelineSelector {
+ states: vec![State::Completed],
+ actions: vec![],
+ start_timestamp: None,
+ end_timestamp: None,
+ include_archived: false,
+ };
+ assert!(selector.select(&timeline).unwrap().is_empty());
+ }
+
+ fn create_test_active_completed_selector(
+ start: Option<&str>,
+ end: Option<&str>,
+ ) -> TimelineSelector {
+ TimelineSelector {
+ states: vec![State::Completed],
+ actions: vec![Action::Commit, Action::ReplaceCommit],
+ start_timestamp: start.map(|s| {
+ Instant::parse_datetime(s, "UTC")
+ .unwrap()
+ .timestamp_millis()
+ }),
+ end_timestamp: end.map(|s| {
+ Instant::parse_datetime(s, "UTC")
+ .unwrap()
+ .timestamp_millis()
+ }),
+ include_archived: false,
+ }
+ }
+
+ #[tokio::test]
+ async fn test_timestamp_filtering() -> Result<()> {
+ let timeline = create_test_timeline().await;
+
+ // starting from the earliest timestamp (inclusive)
+ let selector =
create_test_active_completed_selector(Some("20240103153000000"), None);
+ let selected = selector.select(&timeline)?;
+ assert_eq!(
+ selected.iter().map(|i| &i.timestamp).collect::<Vec<_>>(),
+ &[
+ "20240103153000",
+ "20240103153010999",
+ "20240103153020999",
+ "20240103153030999",
+ ]
+ );
+
+ // ending until the latest timestamp (exclusive)
+ let selector = create_test_active_completed_selector(None,
Some("20240103153030999"));
+ let selected = selector.select(&timeline)?;
+ assert_eq!(
+ selected.iter().map(|i| &i.timestamp).collect::<Vec<_>>(),
+ &["20240103153000", "20240103153010999", "20240103153020999",]
+ );
+
+ // start and end in the middle
+ let selector = create_test_active_completed_selector(
+ Some("20240103153010000"),
+ Some("20240103153030000"),
+ );
+ let selected = selector.select(&timeline)?;
+ assert_eq!(
+ selected.iter().map(|i| &i.timestamp).collect::<Vec<_>>(),
+ &["20240103153010999", "20240103153020999",]
+ );
+ Ok(())
+ }
+}
