Skip to content

[SPARK-56882][SDP] Implement SCD1 Batch Processor; Target Column Projection#55991

Open
AnishMahto wants to merge 13 commits into
apache:masterfrom
AnishMahto:SPARK-56882-SCD1-project-target-columns-onto-microbatch
Open

[SPARK-56882][SDP] Implement SCD1 Batch Processor; Target Column Projection#55991
AnishMahto wants to merge 13 commits into
apache:masterfrom
AnishMahto:SPARK-56882-SCD1-project-target-columns-onto-microbatch

Conversation

@AnishMahto
Copy link
Copy Markdown
Contributor

@AnishMahto AnishMahto commented May 19, 2026

Approved AutoCDC SPIP: https://lists.apache.org/thread/j6sj9wo9odgdpgzlxtvhoy7szs0jplf7


This is a stacked PR. Review incremental diff here: AnishMahto/spark@SPARK-56870-extend-microbatch-with-cdc-metadata...SPARK-56882-SCD1-project-target-columns-onto-microbatch

Link to previous PR: #55970


Preamble:

The SCD type 1 flow is a foreachBatch streaming query on an input change-data-feed, and is responsible for reconciling the incoming change data onto some target table that follows SCD1 replication semantics.

SCD1 flows also maintain an "auxiliary" table to keep track of early-arriving out-of-order received events state. Each microbatch will need to reconcile against this auxiliary table as well, and update the auxiliary table's state appropriately for future microbatches.

Target Column Projection:

As per the SPIP and ChangeArgs.columnSelection, users are allowed to specify the set of columns that actually gets persisted in the target table. Any columns not selected should be dropped before target table merge/persistence.

We should project only these selected columns onto the microbatch so that its dataframe is in the correct shape prior to CDC processing and merging into the target table.

Copy link
Copy Markdown
Member

@szehon-ho szehon-ho left a comment

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

LGTM on the incremental projectTargetColumnsOntoMicrobatch change. Left one design question, a comment typo nit, and a few suggested tests inline.

ColumnSelection.applyToSchema(
schemaName = "microbatch",
schema = userColumnsInMicrobatchSchema,
columnSelection = changeArgs.columnSelection,
Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Question: columnSelection can remove key columns (e.g. ExcludeColumns on a key, or a narrow IncludeColumns that omits keys). Will a later merge step still need those columns on this DataFrame?

If keys must remain until after merge, we should validate here (or when constructing ChangeArgs) that changeArgs.keys are not dropped. If merge runs before projection, or keys are re-injected elsewhere, could you add a brief note in the scaladoc on the expected pipeline order?

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yep we do require all keys remain in the column selection.

I added that validation in this PR during flow analysis time (well before flow execution, which is when this would actually be called) - see requireKeysPresentInSelectedSchema.

Flow analysis must always be done before flow execution, so there's no need to do additional user-friendly validation in this internal flow execution step. For unit testing purposes if a test is incorrectly setup, spark will just throw an unresolved column exception.

@AnishMahto AnishMahto force-pushed the SPARK-56882-SCD1-project-target-columns-onto-microbatch branch from 1be3cba to 0660d8d Compare May 21, 2026 19:05
@AnishMahto AnishMahto requested a review from szehon-ho May 21, 2026 19:07
@AnishMahto AnishMahto force-pushed the SPARK-56882-SCD1-project-target-columns-onto-microbatch branch from 0660d8d to b71ec8e Compare May 21, 2026 21:53
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

2 participants