[
https://issues.apache.org/jira/browse/SPARK-58937?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Anish Mahto updated SPARK-58937:
--------------------------------
Description:
In order for SCD2 to reconcile a microbatch, it finds the existing rows in the
auxiliary and target tables that could be affected by said microbatch, and
pulls them in for reconciliation.
The current implementation finds the affected rows in the auxiliary table and
target table independently, by finding the first rows in each table that
independently precede the minimum sequence (per key) in the microbatch.
This is incorrect behavior as it can lead to pulling in additional rows from
the auxiliary table that are not actually affected by the microbatch. Per
reconciliation there can be at most one actual row per key that immediately
precedes the microbatch's events and therefore should be considered the
starting point for all rows that need to be pulled in. By calculating an
affected row anchor for the aux and target tables independently, up to two
anchor rows could be pulled in instead.
For any given microbatch its possible the aux table's anchor row is the one
that is actually redundant, and will be pulled in unnecessarily. Pulling in
these additional rows can actually be consequential for correctness, because
the auxiliary table holds hidden no-op upsert run rows. In several scenarios,
such upsert run rows can be then further be incorrectly promoted to run-tails
and moved to the target table.
The fix is to consider a unified or global timeline of all existing rows across
the auxiliary and target tables, before finding the single and truthful anchor
row (if one exists). That is, union the two tables before finding the first row
that immediately precedes the microbatch, rather than doing so independently on
each table.
was:
In order for SCD2 to reconcile a microbatch, it finds the existing rows in the
auxiliary and target tables that could be affected by said microbatch, and
pulls them in for reconciliation.
The current implementation finds the affected rows in the auxiliary table and
target table independently, by finding the first rows in each table that
independently precede the minimum sequence (per key) in the microbatch.
This is incorrect behavior as it can lead to pulling in additional rows from
the auxiliary table that are not actually affected by the microbatch. Per
reconciliation there can be at most one actual row per key that immediately
precedes the microbatch's events and therefore should be considered the
starting point for all rows that need to be pulled in. By calculating an
affected row anchor for the aux and target tables independently, up to two
anchor rows could be pulled in instead.
For any given microbatch its possible the aux table's anchor row is the one
that is actually redundant, and will be pulled in unnecessarily. Pulling in
these additional rows can actually be consequential for correctness, because
the auxiliary table holds hidden no-op upsert run rows.
In several scenarios, such upsert run rows can be then further be incorrectly
promoted to run-tails and moved to the target table.
The fix is to consider a unified or global timeline of all existing rows across
the auxiliary and target tables, before finding the single and truthful anchor
row (if one exists). That is, union the two tables before finding the first row
that immediately precedes the microbatch, rather than doing so independently on
each table.
> Consider unified timeline when determining affected rows for SCD2
> -----------------------------------------------------------------
>
> Key: SPARK-58937
> URL: https://issues.apache.org/jira/browse/SPARK-58937
> Project: Spark
> Issue Type: Sub-task
> Components: Declarative Pipelines
> Affects Versions: 4.3.0
> Reporter: Anish Mahto
> Priority: Blocker
>
> In order for SCD2 to reconcile a microbatch, it finds the existing rows in
> the auxiliary and target tables that could be affected by said microbatch,
> and pulls them in for reconciliation.
> The current implementation finds the affected rows in the auxiliary table and
> target table independently, by finding the first rows in each table that
> independently precede the minimum sequence (per key) in the microbatch.
> This is incorrect behavior as it can lead to pulling in additional rows from
> the auxiliary table that are not actually affected by the microbatch. Per
> reconciliation there can be at most one actual row per key that immediately
> precedes the microbatch's events and therefore should be considered the
> starting point for all rows that need to be pulled in. By calculating an
> affected row anchor for the aux and target tables independently, up to two
> anchor rows could be pulled in instead.
> For any given microbatch its possible the aux table's anchor row is the one
> that is actually redundant, and will be pulled in unnecessarily. Pulling in
> these additional rows can actually be consequential for correctness, because
> the auxiliary table holds hidden no-op upsert run rows. In several scenarios,
> such upsert run rows can be then further be incorrectly promoted to run-tails
> and moved to the target table.
> The fix is to consider a unified or global timeline of all existing rows
> across the auxiliary and target tables, before finding the single and
> truthful anchor row (if one exists). That is, union the two tables before
> finding the first row that immediately precedes the microbatch, rather than
> doing so independently on each table.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
