The issue comes from defining your source as a temporary materialized view:

@DP.materialized_view(temporary=True)
def source():

Temporary materialized views do not track state between pipeline runs. Because of that, the view is fully refreshed every time, so the pipeline always reprocesses the entire dataset instead of using CDF incrementally.

To fix this, remove temporary=True so the materialized view can maintain state and leverage Change Data Feed properly

@DP.materialized_view
def source():

After making this change, your pipeline should only process incremental changes.