PostgresHook: Add upsert rows support using ON CONFLICT

[SameerMesiah97]

Description

This change adds a new PostgresHook.upsert_rows method that provides native PostgreSQL UPSERT support using INSERT ... ON CONFLICT.

The new method supports configurable conflict targets through conflict_fields and selective updates through update_fields. When update_fields is omitted or empty, conflicting rows are ignored using DO NOTHING.

upsert_rows reuses the existing batching, transaction handling, serialization, and lineage behavior used by insert_rows, while introducing PostgreSQL-specific UPSERT semantics that are not currently exposed through the generic insert abstraction.

This PR is dependent on PR #66893 merging first.

Rationale

DbApiHook.insert_rows currently supports a generic replace=True abstraction delegated through dialect-specific SQL generation. However, PostgreSQL UPSERT semantics require additional concepts that are not representable through the existing API, including explicit conflict targets and selective update columns.

Supporting PostgreSQL-native UPSERT behavior through insert_rows would require introducing PostgreSQL-specific arguments such as conflict_fields and update_fields into the shared public DbApiHook.insert_rows API. Since DbApiHook is inherited broadly across providers, expanding the generic insert abstraction with provider-specific UPSERT semantics would increase API complexity and introduce ambiguous behavior for non-PostgreSQL hooks.

Adding a dedicated PostgresHook.upsert_rows method keeps PostgreSQL ON CONFLICT semantics explicit and self-contained while avoiding backwards compatibility and abstraction concerns in the shared DbApiHook interface.

The implementation uses PostgreSQL-native INSERT ... ON CONFLICT semantics rather than MERGE, since ON CONFLICT is the established and more broadly compatible UPSERT mechanism across supported PostgreSQL versions.

Tests

Added unit tests verifying that:

• Standard UPSERT operations correctly generate ON CONFLICT DO UPDATE SQL.
• UPSERT operations correctly support single and composite conflict fields.
• UPSERT operations correctly support single and multiple update fields.
• DO NOTHING behavior is generated when update_fields is omitted.
• fast_executemany=True uses psycopg2.extras.execute_batch.
• commit_every correctly chunks UPSERT operations across transactions.
• Empty row collections do not generate SQL or emit lineage.
• Empty or invalid target_fields and conflict_fields raise validation errors.

Backwards Compatibility

This change introduces a new provider-specific API and does not modify existing insert_rows behavior or shared DbApiHook interfaces.

[justinpakzad]

Nice PR. Left a couple of comments. I know it's still in draft but figured I'd leave some feedback anyways.

[justinpakzad]

I don't think we need the kwargs here as there is nothing consuming them.

[justinpakzad]

Do we need to pass in the values here? The only thing it's used for is to produce the right number of placeholders but I think that can just be done with len(target_fields).

[justinpakzad]

Related to my comment above - if we don't need to pass values since it's just used for the number of placeholders, I don't think we need to regenerate the same SQL string on every chunk. This can just be done once outside of the loop.

[justinpakzad]

Do we need a guard here since psycopg3 does not support execute_batch? Maybe using the USE_PSYCOPG3 constant that's used in other parts of the code. Either logging a warning and defaulting back to cur.executemany or raising an error.

[justinpakzad]

Also related to my comment above, if we construct the query once outside the loop then this would need to be updated. This could be if nb_rows > 0.

[SameerMesiah97]

Nice PR. Left a couple of comments. I know it's still in draft but figured I'd leave some feedback anyways.

That’s perfectly fine. I will address the feedback once the dependency is merged.