From 75decc89e355beb5d6d29aa869e7052cda160639 Mon Sep 17 00:00:00 2001 From: yu1005 Date: Sat, 7 Mar 2026 00:57:50 +0800 Subject: [PATCH 1/2] Refactor blocker logic in propagate_target_rows and update tests for date handling --- src/lib.rs | 67 +++++++++++++++++++++++++-------------- test/data/pivot_input.csv | 7 +++- test/test_polars.py | 20 +++++++++++- 3 files changed, 69 insertions(+), 25 deletions(-) diff --git a/src/lib.rs b/src/lib.rs index 867cb0f..7ba6caf 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -170,7 +170,7 @@ fn choose_candidate_by_plan( Ok(best_idx) } -fn has_blocker_on_node_after_target( +fn earliest_blocker_on_node_after_target( py: Python<'_>, rows: &[Py], self_index_all: &HashMap>, @@ -178,11 +178,13 @@ fn has_blocker_on_node_after_target( ref_target: &Bound<'_, PyAny>, target_col: &str, block_key: &HashMap, -) -> PyResult { +) -> PyResult>> { let Some(indices) = self_index_all.get(node_key) else { - return Ok(false); + return Ok(None); }; + let mut earliest: Option> = None; + for idx in indices { let row = rows[*idx].bind(py); if !matches_kv(&row, block_key)? { @@ -196,12 +198,20 @@ fn has_blocker_on_node_after_target( continue; } - if compare_bool(&block_target, ref_target, CompareOp::Gt) { - return Ok(true); + if !compare_bool(&block_target, ref_target, CompareOp::Gt) { + continue; + } + + let should_take = match &earliest { + None => true, + Some(current) => compare_bool(&block_target, current.bind(py), CompareOp::Lt), + }; + if should_take { + earliest = Some(block_target.unbind()); } } - Ok(false) + Ok(earliest) } /// For each row, walk through parent links (single-path DFS over ancestor chain) until a parent @@ -216,12 +226,12 @@ fn has_blocker_on_node_after_target( /// /// block_key: /// - Optional key/value matcher applied at node level. -/// - For each active row, blocker checking starts on the row's own self node first. -/// - If not blocked on self node, blocker checking continues on ancestor parent nodes during -/// traversal. -/// - If any checked node has a blocker event with target_col > current reference target, -/// traversal is terminated. -/// - In that case the original target value remains unchanged. +/// - For each active row, blocker search starts on self node and continues on traversed +/// parent nodes. +/// - Blockers are filtered by target_col > anchor target, then the earliest blocker among +/// visited nodes is tracked. +/// - Before accepting a matched target candidate, compare candidate target_col vs blocker: +/// if target is earlier than blocker, keep target; otherwise break and keep original value. fn propagate_target_rows( py: Python<'_>, rows: &[Py], @@ -264,9 +274,10 @@ fn propagate_target_rows( } let anchor_target = start_target.clone(); let start_self = composite_key(&start_row, self_cols)?; + let mut earliest_blocker: Option> = None; if let Some(kv) = block_key { - if has_blocker_on_node_after_target( + if let Some(blocker) = earliest_blocker_on_node_after_target( py, rows, &self_index_all, @@ -275,7 +286,7 @@ fn propagate_target_rows( target_col, kv, )? { - continue; + earliest_blocker = Some(blocker); } } @@ -290,7 +301,7 @@ fn propagate_target_rows( } if let Some(kv) = block_key { - if has_blocker_on_node_after_target( + if let Some(node_blocker) = earliest_blocker_on_node_after_target( py, rows, &self_index_all, @@ -299,7 +310,15 @@ fn propagate_target_rows( target_col, kv, )? { - break; + let should_take = match &earliest_blocker { + None => true, + Some(current) => { + compare_bool(node_blocker.bind(py), current.bind(py), CompareOp::Lt) + } + }; + if should_take { + earliest_blocker = Some(node_blocker); + } } } @@ -321,12 +340,6 @@ fn propagate_target_rows( break; } - if let Some(kv) = block_key { - if matches_kv(&candidate, kv)? { - break; - } - } - let para_val = candidate .get_item(para_col)? .and_then(|x| x.extract::().ok()); @@ -334,7 +347,15 @@ fn propagate_target_rows( .as_deref() .is_some_and(|val| val.starts_with(para_prefix)) { - replacement = Some(candidate_target.unbind()); + let target_before_block = match &earliest_blocker { + None => true, + Some(blocker_target) => { + compare_bool(&candidate_target, blocker_target.bind(py), CompareOp::Lt) + } + }; + if target_before_block { + replacement = Some(candidate_target.unbind()); + } break; } diff --git a/test/data/pivot_input.csv b/test/data/pivot_input.csv index 41af4c4..be2d40b 100644 --- a/test/data/pivot_input.csv +++ b/test/data/pivot_input.csv @@ -1,6 +1,11 @@ id,parent_id,event,date +parent,root,Remove,2026-01-25 +root,parent0,Remove,2026-01-02 root,parent0,Install,2026-01-01 root2,parent0,Install,2026-01-01 -parent,root,Remove,2025-12-31 +parent,root,Install,2025-12-31 parent,root,Install,2025-12-20 leaf,root2,Install,2025-12-01 +leaf1,parent,Install,2025-12-01 +leaf2,parent,Install,2025-12-01 +leaf3,parent,Install,2025-12-01 diff --git a/test/test_polars.py b/test/test_polars.py index c2db035..c44bc04 100644 --- a/test/test_polars.py +++ b/test/test_polars.py @@ -12,17 +12,34 @@ def _normalize(df: pl.DataFrame) -> pl.DataFrame: def test_propagate_target_from_ancestor_from_local_csv() -> None: csv_path = Path(__file__).parent / "data" / "pivot_input.csv" source_df = pl.read_csv(csv_path, try_parse_dates=False) + + source_df = ( + source_df + .with_columns(pl.col("date").alias("install_date")) + .with_columns(pl.col("date").alias("remove_date")) + ) result_df = hello_rust.propagate_target_from_ancestor( source_df, self_cols=["id"], parent_cols=["parent_id"], - target_col="date", + target_col="install_date", target_key={"event": "Install"}, para={"parent_id": "parent0"}, plan="forward", block_key={"event": "Remove"}, ) + + result_df = hello_rust.propagate_target_from_ancestor( + result_df, + self_cols=["id"], + parent_cols=["parent_id"], + target_col="remove_date", + target_key={"event": "Remove"}, + para={"event": "Remove"}, + plan="backward", + block_key={"event": "Install"}, + ) # expected_df = source_df.with_columns( # pl.when(pl.col("id") == "leaf").then(pl.lit(10)).otherwise(pl.col("ts")).alias("ts") @@ -41,3 +58,4 @@ def test_propagate_target_from_ancestor_from_local_csv() -> None: # print(expected_df.to_dicts()) # compare removed for now; print output only + # test command .\.venv\Scripts\python -m pytest -s -q test/test_polars.py From e185619337d9cef021f4ab7ca09f9f40e66846ae Mon Sep 17 00:00:00 2001 From: yu1005 Date: Sat, 7 Mar 2026 20:30:08 +0800 Subject: [PATCH 2/2] upgrade to 0.1.5 with number 0 --- Cargo.lock | 2 +- Cargo.toml | 2 +- recipe/meta.yaml | 4 ++-- 3 files changed, 4 insertions(+), 4 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 8aad74a..df7b8c3 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -16,7 +16,7 @@ checksum = "2304e00983f87ffb38b55b444b5e3b60a884b5d30c0fca7d82fe33449bbe55ea" [[package]] name = "hello_rust" -version = "0.1.4" +version = "0.1.5" dependencies = [ "pyo3", ] diff --git a/Cargo.toml b/Cargo.toml index 204ea54..e1990a5 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "hello_rust" -version = "0.1.4" +version = "0.1.5" edition = "2024" # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html diff --git a/recipe/meta.yaml b/recipe/meta.yaml index f438a6d..57e8ecb 100644 --- a/recipe/meta.yaml +++ b/recipe/meta.yaml @@ -1,5 +1,5 @@ {% set name = "hello-rust" %} -{% set version = "0.1.4" %} +{% set version = "0.1.5" %} package: name: {{ name|lower }} @@ -9,7 +9,7 @@ source: path: .. build: - number: 1 + number: 0 script: {{ PYTHON }} -m pip install . -vv requirements: