linliu-code commented on code in PR #660: URL: https://github.com/apache/hudi-rs/pull/660#discussion_r3786608594
########## crates/core/tests/reader_concurrency_tests.rs: ########## @@ -0,0 +1,341 @@ +/* + * 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. + */ +//! Reader-vs-writer concurrency repros. +//! +//! Most of these are `#[ignore]`d, for one of two reasons: they document a +//! divergence from Hudi's Java reader that is not fixed yet, or they are probes +//! that print what the reader does without asserting anything. Neither belongs +//! in the suite — the first would fail it, and the second would pass whatever +//! happened. Run them with +//! `cargo test -p hudi-core --test reader_concurrency_tests -- --ignored --nocapture`. +//! +//! A divergence that gets fixed should leave this file, or become an ordinary +//! test that pins the fixed behavior. An `#[ignore]` reason describing something +//! that no longer happens is worse than no note at all. + +use hudi_core::table::{ReadOptions, Table}; +use std::path::{Path, PathBuf}; + +fn rows(batches: &[arrow_array::RecordBatch]) -> usize { + batches.iter().map(|b| b.num_rows()).sum() +} + +/// Copy an existing base file under a NEW file id at `instant`, leaving no +/// commit for that instant — a writer that wrote its data file and then died. +fn plant_orphan_base_file(table_dir: &Path, instant: &str) -> PathBuf { + let existing = std::fs::read_dir(table_dir) + .unwrap() + .filter_map(|e| e.ok()) + .map(|e| e.path()) + .find(|p| p.extension().is_some_and(|x| x == "parquet")) + .expect("fixture has a base file to copy"); + let orphan = table_dir.join(format!( + "b179bdb3-731c-4894-b855-abfcd6921008-0_0-1-1_{instant}.parquet" + )); + std::fs::copy(&existing, &orphan).unwrap(); + orphan +} + +fn fixture_zip(relative: &str) -> PathBuf { + PathBuf::from(env!("CARGO_MANIFEST_DIR").replace("/core", "/test")).join(relative) +} + +async fn probe(zip: &str, label: &str) { + probe_with_reader_version(zip, label, "1").await; + probe_with_reader_version(zip, label, "2").await; +} + +async fn probe_with_reader_version(zip: &str, label: &str, reader_version: &str) { + use hudi_core::config::read::HudiReadConfig; + let dir = hudi_test::extract_test_table_fresh(&fixture_zip(zip)); + let table_dir = std::fs::read_dir(&dir) + .unwrap() + .filter_map(|e| e.ok()) + .map(|e| e.path()) + .find(|p| p.is_dir() && p.join(".hoodie").exists()) + .expect("extracted fixture holds one table dir"); + + let opts = ReadOptions::new().with_hudi_option( + HudiReadConfig::FileGroupReaderVersion.as_ref(), + reader_version, + ); + let table = Table::new(table_dir.to_str().unwrap()).await.unwrap(); + let before = rows(&table.read(&opts).await.unwrap()); + + // One tick below the latest commit: above the archival boundary (so it reads + // as pending, not as archived) and at or below the snapshot bound (so the + // as-of filter does not exclude it either). That is an in-flight commit. + let latest_ts = table + .get_timeline() + .get_latest_commit_timestamp() + .unwrap() + .to_string(); + let orphan_instant = format!( + "{:0width$}", + latest_ts.parse::<u64>().unwrap() - 1, + width = latest_ts.len() + ); + let orphan_instant = orphan_instant.as_str(); + plant_orphan_base_file(&table_dir, orphan_instant); + + let table = Table::new(table_dir.to_str().unwrap()).await.unwrap(); + let after = rows(&table.read(&opts).await.unwrap()); + println!( + "{label:34} reader version={reader_version} rows before={before} after={after} {}", + if after == before { + "OK (orphan excluded)" + } else { + "*** UNCOMMITTED ROWS VISIBLE ***" + } + ); +} + +#[ignore = "documents a divergence from Java: layout-v1 tables do not filter uncommitted files"] Review Comment: Bisected, and the answer is the first of your two possibilities: **it got fixed, in the very commit that added the probe.** `cd0d5be` ("fix(core): address the merge-on-read port and concurrency reviews") says so itself: > A file was readable whenever a completion timestamp could be found for its commit, which asked the wrong question twice. Timeline layout v1 records no completion timestamps at all, so nothing was filtered there and a snapshot returned rows from commits that never completed — 4 rows became 7 on the v6 fixtures. That commit replaced the layout-v2-only check with `completed_requests: HashSet<String>`, populated for **both** layouts, plus the archival boundary. The probe was written as the reproduction, the fix landed alongside it, and the `#[ignore]` reason was never updated — which is why it now prints "OK (orphan excluded)" for all six combinations. So it is an ordinary test now, exactly as this file's own module doc prescribes: `test_a_base_file_from_an_uncommitted_instant_is_not_readable`, un-ignored, asserting `after == before` instead of printing it, on both timeline layouts and both reader versions. I also corrected the module doc, which claimed the `#[ignore]`s exist for two reasons (unfixed divergence, or non-asserting probe). Only the second applies now — there are no divergence ignores left in this file. -- 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]
