From 483cce818165e63833538b8e4c316d5896413fbe Mon Sep 17 00:00:00 2001 From: Sergey Olontsev Date: Mon, 8 Jun 2026 22:26:00 +0100 Subject: [PATCH] unmaintained aquamarine dependency removal --- Cargo.toml | 1 - src/materialized/dependencies.rs | 115 +++++++++++++++++++++++-------- 2 files changed, 88 insertions(+), 28 deletions(-) diff --git a/Cargo.toml b/Cargo.toml index 041add6..270e668 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -28,7 +28,6 @@ keywords = ["arrow", "arrow-rs", "datafusion"] rust-version = "1.85.1" [dependencies] -aquamarine = "0.6.0" arrow = "57.1.0" arrow-schema = "57.1.0" async-trait = "0.1.89" diff --git a/src/materialized/dependencies.rs b/src/materialized/dependencies.rs index 7295b36..f1f3468 100644 --- a/src/materialized/dependencies.rs +++ b/src/materialized/dependencies.rs @@ -276,7 +276,6 @@ fn get_table_name(args: &[Expr]) -> Result<&String> { } } -#[cfg_attr(doc, aquamarine::aquamarine)] /// Returns a logical plan that, when executed, lists expected build targets /// for this materialized view, together with the dependencies for each target. /// @@ -304,26 +303,65 @@ fn get_table_name(args: &[Expr]) -> Result<&String> { /// Assume that both tables are partitioned by `date` only. We desired a materialized view partitioned by `date` and stored at `s3://daily_close/`. /// This query gives us the following logical plan: /// -/// ```mermaid -/// %%{init: { 'flowchart': { 'wrappingWidth': 1000 }}}%% -/// graph TD -/// A["Projection:
ticker, LAST_VALUE(trades.price) AS close, LAST_VALUE(daily_statistics.settlement_price) AS settlement_price, trades.date AS date"] -/// A --> B["Aggregate:
expr=[LAST_VALUE(trades.price), LAST_VALUE(daily_statistics.settlement_price)]
groupby=[ticker, trades.date]"] -/// B --> C["Inner Join:
trades.ticker = daily_statistics.ticker AND
trades.date = daily_statistics.reference_date AND
daily_statistics.date BETWEEN trades.date AND trades.date + INTERVAL 2 WEEKS"] -/// C --> D["TableScan: trades
projection=[ticker, price, date]"] -/// C --> E["TableScan: daily_statistics
projection=[ticker, settlement_price, reference_date, date]"] +/// ```text +/// +--------------------------------------------------------------------+ +/// | Projection: | +/// | ticker, LAST_VALUE(trades.price) AS close, | +/// | LAST_VALUE(daily_statistics.settlement_price) AS settlement_price, | +/// | **trades.date AS date** | +/// +--------------------------------------------------------------------+ +/// | +/// v +/// +------------------------------------------------------+ +/// | Aggregate: | +/// | expr=[LAST_VALUE(trades.price), | +/// | LAST_VALUE(daily_statistics.settlement_price)] | +/// | groupby=[ticker, **trades.date**] | +/// +------------------------------------------------------+ +/// | +/// v +/// +---------------------------------------------------+ +/// | Inner Join: | +/// | trades.ticker = daily_statistics.ticker AND | +/// | trades.date = daily_statistics.reference_date AND | +/// | **daily_statistics.date BETWEEN trades.date AND | +/// | trades.date + INTERVAL 2 WEEKS** | +/// +---------------------------------------------------+ +/// | +/// +---------------------+--------------+ +/// v v +/// +-----------------------------+ +---------------------------------------+ +/// | TableScan: trades | | TableScan: daily_statistics | +/// | projection= | | projection=[ticker, settlement_price, | +/// | [ticker, price, **date**] | | reference_date, **date**] | +/// +-----------------------------+ +---------------------------------------+ /// ``` /// -/// All partition-column-derived expressions are marked in yellow. We now proceed with **Inexact Projection Pushdown**, and prune all unmarked expressions, resulting in the following plan: +/// All partition-column-derived expressions are marked with double asterisks (`**`). We now proceed with **Inexact Projection Pushdown**, and prune all unmarked expressions, resulting in the following plan: /// -/// ```mermaid -/// %%{init: { 'flowchart': { 'wrappingWidth': 1000 }}}%% -/// graph TD -/// A["Projection: trades.date AS date"] -/// A --> B["Projection: trades.date"] -/// B --> C["Inner Join:
daily_statistics.date BETWEEN trades.date AND trades.date + INTERVAL 2 WEEKS"] -/// C --> D["TableScan: trades (projection=[date])"] -/// C --> E["TableScan: daily_statistics (projection=[date])"] +/// ```text +/// +---------------------------------+ +/// | Projection: trades.date AS date | +/// +---------------------------------+ +/// | +/// v +/// +-------------------------+ +/// | Projection: trades.date | +/// +-------------------------+ +/// | +/// v +/// +-------------------------------------------+ +/// | Inner Join: | +/// | daily_statistics.date BETWEEN trades.date | +/// | AND trades.date + INTERVAL 2 WEEKS | +/// +-------------------------------------------+ +/// | +/// +----------------+-----------+ +/// v v +/// +---------------------+ +-----------------------------+ +/// | TableScan: trades | | TableScan: daily_statistics | +/// | (projection=[date]) | | (projection=[date]) | +/// +---------------------+ +-----------------------------+ /// ``` /// /// Note that the `Aggregate` node was converted into a projection. This is valid because we do not need to preserve duplicate rows. However, it does imply that @@ -332,15 +370,38 @@ fn get_table_name(args: &[Expr]) -> Result<&String> { /// Now we substitute all scans with equivalent row metadata scans (up to addition or removal of duplicates), and push up the row metadata to the root of the plan, /// together with the target path constructed from the (static) partition columns. This gives us the following plan: /// -/// ```mermaid -/// %%{init: { 'flowchart': { 'wrappingWidth': 1000 }}}%% -/// graph TD -/// A["Projection: concat('s3://daily_close/date=', date::string, '/') AS target, __meta"] -/// A --> B["Projection: __meta, trades.date AS date"] -/// B --> C["Projection:
concat(trades_meta.__meta, daily_statistics_meta.__meta) AS __meta, date"] -/// C --> D["Inner Join:
daily_statistics_meta.date BETWEEN trades_meta.date AND trades_meta.date + INTERVAL 2 WEEKS"] -/// D --> E["TableScan: trades_meta (projection=[__meta, date])"] -/// D --> F["TableScan: daily_statistics_meta (projection=[__meta, date])"] +/// ```text +/// +----------------------------------------------------------------+ +/// | Projection: | +/// | concat('s3://daily_close/date=', date::string, '/') AS target, | +/// | __meta | +/// +----------------------------------------------------------------+ +/// | +/// v +/// +-----------------------------------------+ +/// | Projection: __meta, trades.date AS date | +/// +-----------------------------------------+ +/// | +/// v +/// +----------------------------------------------------------+ +/// | Projection: | +/// | concat(trades_meta.__meta, daily_statistics_meta.__meta) | +/// | AS __meta, date | +/// +----------------------------------------------------------+ +/// | +/// v +/// +-----------------------------------------------------+ +/// | Inner Join: | +/// | daily_statistics_meta.date BETWEEN trades_meta.date | +/// | AND trades_meta.date + INTERVAL 2 WEEKS | +/// +-----------------------------------------------------+ +/// | +/// +------------------+----------------+ +/// v v +/// +-----------------------------+ +----------------------------------+ +/// | TableScan: trades_meta | | TableScan: daily_statistics_meta | +/// | (projection=[__meta, date]) | | (projection=[__meta, date]) | +/// +-----------------------------+ +----------------------------------+ /// ``` /// /// Here, `__meta` is a column containing a list of structs with the row metadata for each source file. The final query has this struct column