linliu-code commented on code in PR #677:
URL: https://github.com/apache/hudi-rs/pull/677#discussion_r3771133201
##########
crates/core/src/file_group/reader.rs:
##########
@@ -221,6 +222,88 @@ impl FileGroupReader {
.await
}
+ /// Which merge implementation serves this read.
+ ///
+ /// A metadata table is always served by the legacy reader whatever the
+ /// setting says: its base files and log blocks are HFile, which the
+ /// merge-on-read engine has no support for. That is permanent, not
+ /// transitional.
+ ///
+ /// The value is read raw rather than through `get_or_default`, which falls
+ /// back to the default when a value fails to parse. A typo in the engine
+ /// name would then silently read with the other engine, leaving a caller
+ /// convinced they had exercised it — the one outcome this switch must not
+ /// produce.
+ fn file_group_reader_version(&self) -> Result<FileGroupReaderVersion> {
+ if self.is_metadata_table() {
+ return Ok(FileGroupReaderVersion::One);
+ }
+ match self
+ .hudi_configs
+ .as_options()
+ .get(HudiReadConfig::FileGroupReaderVersion.as_ref())
+ {
+ Some(raw) =>
FileGroupReaderVersion::from_str(raw).map_err(CoreError::Config),
+ None => Ok(FileGroupReaderVersion::default()),
+ }
+ }
+
+ /// Why the merge-on-read engine cannot serve this read, if it cannot.
+ ///
+ /// This is a capability check, decided from config before any I/O — never
a
+ /// catch-all on error. A read that fails *inside* the engine propagates:
+ /// retrying it on the legacy reader would make a bug look like a success,
+ /// make results depend on which engine happened to win, and leave the
+ /// differential tests unable to see anything.
+ ///
+ /// Every reason here means the legacy reader serves the read instead, so
+ /// selecting the engine cannot turn a working read into a failing one.
Each
+ /// reason is logged, because a fallback nobody can observe is
+ /// indistinguishable from an engine that is never used.
+ fn version_two_unsupported_reason(
+ &self,
+ options: &ReadOptions,
+ ) -> Result<Option<&'static str>> {
+ // Deliberately an error rather than a fallback: falling back would use
+ // the legacy reader's own merge derivation, which drops deletes on a
+ // commit-time-ordered table. Wrong rows are worse than a refusal.
+ if let Some(mode) = self
+ .hudi_configs
+ .as_options()
+ // Read by raw key: this crate has no typed config for it yet, and
+ // adding one belongs with the reader that acts on it.
+ .get("hoodie.record.merge.mode")
+ && mode.eq_ignore_ascii_case("CUSTOM")
+ {
+ return Err(CoreError::Unsupported(
Review Comment:
Confirmed and fixed in 74ef65d — you are right, and it is worse than a nit
because version 2 is the default, so nobody opted in.
The refusal now takes `base_file_only` and applies only when a merge
actually happens. The reason for refusing is that falling back would merge with
version 1's derivation, which drops deletes — that reasoning simply does not
apply to a read that never consults a merger, so a copy-on-write or
read-optimized read falls back like any other unimplemented capability.
I kept the dispatch where it is rather than moving it below the
`base_file_only` branch, because version 2 is meant to serve base-file-only
reads later; it was the refusal that was mis-scoped, not the dispatch.
##########
crates/core/src/file_group/reader.rs:
##########
@@ -221,6 +222,88 @@ impl FileGroupReader {
.await
}
+ /// Which merge implementation serves this read.
+ ///
+ /// A metadata table is always served by the legacy reader whatever the
+ /// setting says: its base files and log blocks are HFile, which the
+ /// merge-on-read engine has no support for. That is permanent, not
+ /// transitional.
+ ///
+ /// The value is read raw rather than through `get_or_default`, which falls
+ /// back to the default when a value fails to parse. A typo in the engine
+ /// name would then silently read with the other engine, leaving a caller
+ /// convinced they had exercised it — the one outcome this switch must not
+ /// produce.
+ fn file_group_reader_version(&self) -> Result<FileGroupReaderVersion> {
+ if self.is_metadata_table() {
+ return Ok(FileGroupReaderVersion::One);
+ }
+ match self
+ .hudi_configs
+ .as_options()
+ .get(HudiReadConfig::FileGroupReaderVersion.as_ref())
+ {
+ Some(raw) =>
FileGroupReaderVersion::from_str(raw).map_err(CoreError::Config),
+ None => Ok(FileGroupReaderVersion::default()),
+ }
+ }
+
+ /// Why the merge-on-read engine cannot serve this read, if it cannot.
+ ///
+ /// This is a capability check, decided from config before any I/O — never
a
+ /// catch-all on error. A read that fails *inside* the engine propagates:
+ /// retrying it on the legacy reader would make a bug look like a success,
+ /// make results depend on which engine happened to win, and leave the
+ /// differential tests unable to see anything.
+ ///
+ /// Every reason here means the legacy reader serves the read instead, so
+ /// selecting the engine cannot turn a working read into a failing one.
Each
+ /// reason is logged, because a fallback nobody can observe is
+ /// indistinguishable from an engine that is never used.
+ fn version_two_unsupported_reason(
+ &self,
+ options: &ReadOptions,
+ ) -> Result<Option<&'static str>> {
+ // Deliberately an error rather than a fallback: falling back would use
+ // the legacy reader's own merge derivation, which drops deletes on a
+ // commit-time-ordered table. Wrong rows are worse than a refusal.
+ if let Some(mode) = self
+ .hudi_configs
+ .as_options()
+ // Read by raw key: this crate has no typed config for it yet, and
+ // adding one belongs with the reader that acts on it.
+ .get("hoodie.record.merge.mode")
+ && mode.eq_ignore_ascii_case("CUSTOM")
+ {
+ return Err(CoreError::Unsupported(
Review Comment:
Added in 74ef65d — two tests, one either side of the gate:
- `..._custom_merge_mode_without_merge_returns_reason` — a CUSTOM table with
nothing to merge falls back instead of erroring, checked for both the
no-log-files and read-optimized routes.
- `..._custom_merge_mode_with_merge_returns_error` — the same table is still
refused once merging is involved, and the error names why.
Checked the first one is not vacuous: removing the `!base_file_only` gate
makes it fail, and restoring it makes it pass.
##########
crates/core/src/file_group/reader.rs:
##########
@@ -1568,3 +1674,163 @@ mod tests {
Ok(())
}
}
+
+#[cfg(test)]
+mod reader_version_seam_tests {
+ use super::*;
+ use crate::config::util::empty_options;
+ use hudi_test::SampleTable;
+
+ async fn reader_with(
+ options: impl IntoIterator<Item = (&'static str, String)>,
+ ) -> Result<FileGroupReader> {
+ let base_url = SampleTable::V6Nonpartitioned.url_to_mor_parquet();
+ FileGroupReader::new_with_options(base_url.as_ref(), options).await
+ }
+
+ /// The merge-on-read engine is the default, and nothing changes for a
caller
+ /// who sets nothing — because every capability falls back today. Making it
+ /// the default only once it were capable would put the whole behaviour
change
+ /// in one commit; this way each capability carries its own.
+ #[tokio::test]
+ async fn the_default_version_is_two_and_still_falls_back() -> Result<()> {
+ let reader = reader_with(Vec::<(&'static str, String)>::new()).await?;
+ assert_eq!(
+ reader.file_group_reader_version()?,
+ FileGroupReaderVersion::Two
+ );
+ assert!(
+ reader
+ .version_two_unsupported_reason(&ReadOptions::new())?
+ .is_some(),
+ "the default must still be served by the existing reader"
+ );
+ Ok(())
+ }
+
+ /// `legacy` remains reachable, so a caller can opt out of the engine
+ /// entirely rather than relying on it to keep falling back.
+ #[tokio::test]
+ async fn version_one_stays_selectable_as_an_escape_hatch() -> Result<()> {
+ let reader = reader_with([(
+ HudiReadConfig::FileGroupReaderVersion.as_ref(),
+ "1".to_string(),
+ )])
+ .await?;
+ assert_eq!(
+ reader.file_group_reader_version()?,
+ FileGroupReaderVersion::One
+ );
+ Ok(())
+ }
+
+ /// A typo must not read with the other engine. `get_or_default` would have
+ /// swallowed this and left the caller believing they had exercised `v2`.
+ #[tokio::test]
+ async fn an_unrecognised_version_is_an_error() -> Result<()> {
+ let reader = reader_with([(
+ HudiReadConfig::FileGroupReaderVersion.as_ref(),
+ "9".to_string(),
+ )])
+ .await?;
+ let err = reader.file_group_reader_version().unwrap_err();
+ assert!(
+ matches!(err, CoreError::Config(_)),
+ "expected a config error, got {err:?}"
+ );
+ assert!(
+ err.to_string().contains("9"),
+ "the error must name the value"
+ );
+ Ok(())
+ }
+
+ /// Asking for the engine is a request, not a guarantee: every capability
is
+ /// unimplemented so far, so the legacy reader serves the read and says
why.
+ #[tokio::test]
+ async fn asking_for_version_two_falls_back_with_a_reason() -> Result<()> {
+ let reader = reader_with([(
+ HudiReadConfig::FileGroupReaderVersion.as_ref(),
+ "2".to_string(),
+ )])
+ .await?;
+ assert_eq!(
+ reader.file_group_reader_version()?,
+ FileGroupReaderVersion::Two
+ );
+
+ let reason =
reader.version_two_unsupported_reason(&ReadOptions::new())?;
+ assert!(
+ reason.is_some(),
+ "with no engine wired up, every read must fall back"
+ );
+ Ok(())
+ }
+
+ /// The fall back is what makes the default safe: a read works exactly as
it
+ /// did, because the existing reader served it either way.
+ #[tokio::test]
+ async fn selecting_version_two_does_not_change_what_a_read_returns() ->
Result<()> {
+ let base_url = SampleTable::V6Nonpartitioned.url_to_mor_parquet();
+ let table = crate::table::Table::new(base_url.path()).await?;
+ let slices = table.get_file_slices(&ReadOptions::new()).await?;
+ assert!(!slices.is_empty(), "fixture must have a file slice to read");
+
+ let read_with = async |engine: Option<&str>| -> Result<usize> {
+ let options: Vec<(&str, String)> = match engine {
+ Some(e) => vec![(
+ HudiReadConfig::FileGroupReaderVersion.as_ref(),
+ e.to_string(),
+ )],
+ None => Vec::new(),
+ };
+ let reader = FileGroupReader::new_with_options(base_url.as_ref(),
options).await?;
+ let mut rows = 0;
+ for slice in &slices {
+ rows += reader
+ .read_file_slice(slice, &ReadOptions::new())
+ .await?
+ .num_rows();
Review Comment:
Done in 74ef65d. It now renders every cell of every row (and the column
names) and compares those, rather than summing `num_rows()` — a count would
match even if the columns or the values differed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
