Update pg_walstream to version 0.4.1 and refactor transaction handling for improved SQL execution

[isdaniel]

No description provided.

[gemini-code-assist]

Summary of Changes

Hello @isdaniel, I'm Gemini Code Assist¹! I'm currently reviewing this pull request and will post my feedback shortly. In the meantime, here's a summary to help you and other reviewers quickly get up to speed!

This pull request primarily focuses on enhancing the PostgreSQL replication stream integration and streamlining SQL execution across different database destinations. By upgrading the pg_walstream dependency and directly using its LogicalReplicationStream, the project benefits from upstream improvements and a more direct interaction with the replication mechanism. Additionally, the consolidation of SQL transaction handling into a common utility function improves code maintainability and ensures consistent atomic operations for MySQL and SQLite destinations.

Highlights

• Dependency Update: The pg_walstream dependency has been updated from version 0.4.0 to 0.4.1, bringing in the latest improvements and fixes from the upstream library.
• PostgreSQL Replication Stream Refactoring: The internal pg_replication module has been removed, and the pg2any-lib now directly utilizes the pg_walstream::LogicalReplicationStream. This simplifies the codebase by removing an abstraction layer and directly leveraging the capabilities of the updated library.
• Consolidated SQL Execution Logic: A new shared utility function, execute_sqlx_batch_with_hook, has been introduced in pg2any-lib/src/destinations/common.rs. This function centralizes the transaction handling and SQL execution logic for sqlx-based destinations (MySQL and SQLite), reducing code duplication and improving consistency.
• Transaction Handling Improvements: Minor fixes and refinements have been applied to the TransactionManager to improve the calculation of SQL bytes and segment processing logic, enhancing the robustness of transaction file handling.
• Code Cleanup: Unused methods new_batch and new_streaming were removed from the Transaction struct, and the LSN tracker's directory creation logic was simplified.

🧠 New Feature in Public Preview: You can now enable Memory to help Gemini Code Assist learn from your team's feedback. This makes future code reviews more consistent and personalized to your project's style. Click here to enable Memory in your admin console.

Changelog

• Cargo.lock
  • Updated pg_walstream version from 0.4.0 to 0.4.1.
  • Updated the checksum for the pg_walstream package.
• Cargo.toml
  • Updated pg_walstream dependency version from 0.4.0 to 0.4.1.
• pg2any-lib/src/client.rs
  • Removed imports related to the internal pg_replication module.
  • Updated the replication_stream field type in CdcClient from ReplicationStream to LogicalReplicationStream.
  • Refactored the stream creation logic to directly use LogicalReplicationStream::new and ReplicationStreamConfig::from.
  • Modified the start method to pass start_xlog (raw LSN value) to the underlying pg_walstream method.
  • Changed access to shared_lsn_feedback from a method call to a direct field access.
  • Updated the run_producer function signature to accept LogicalReplicationStream.
  • Modified the next_event call to use next_event_with_retry from LogicalReplicationStream.
  • Updated the conditional check for streaming_txs.remove to use is_some().
  • Improved the shutdown sequence in stop by adding send_feedback() and more detailed logging, and wrapping pg_walstream errors in CdcError.
• pg2any-lib/src/destinations/common.rs
  • Added a new public utility function execute_sqlx_batch_with_hook to encapsulate SQL transaction execution logic for sqlx databases (MySQL, SQLite), including pre-commit hooks and error handling.
• pg2any-lib/src/destinations/mysql.rs
  • Replaced the duplicated SQL transaction execution logic with a call to the new super::common::execute_sqlx_batch_with_hook function.
  • Removed the test_map_schema unit test as it was redundant and moved to common.rs.
• pg2any-lib/src/destinations/sqlite.rs
  • Replaced the duplicated SQL transaction execution logic with a call to the new super::common::execute_sqlx_batch_with_hook function.
• pg2any-lib/src/lib.rs
  • Removed the pg_replication module declaration.
  • Removed re-exports related to pg_replication (e.g., PgReplicationConnection, ReplicationConnectionRetry, RetryConfig, PgResult).
• pg2any-lib/src/lsn_tracker.rs
  • Simplified the conditional logic for creating parent directories for LSN files.
• pg2any-lib/src/pg_replication.rs
  • Removed the entire pg_replication.rs file, as its functionality is now directly handled by pg_walstream.
• pg2any-lib/src/transaction_manager.rs
  • Corrected the calculation of sql_bytes to use sql.len() instead of sql.as_bytes().len() for string length.
  • Refined the conditional statement for skipping segments in replay_transaction_file to improve clarity and correctness.
• pg2any-lib/src/types.rs
  • Removed the new_batch and new_streaming associated functions from the Transaction struct, as they were redundant.
• pg2any-lib/tests/integration_tests.rs
  • Updated the use statement to import format_lsn and parse_lsn directly from pg2any_lib instead of pg2any_lib::pg_replication.

Activity

• No specific activity (comments, reviews, progress) has been recorded for this pull request yet.

Using Gemini Code Assist

The full guide for Gemini Code Assist can be found on our documentation page, here are some quick tips.

Invoking Gemini

You can request assistance from Gemini at any point by creating a comment using either /gemini <command> or @gemini-code-assist <command>. Below is a summary of the supported commands on the current page.

Feature	Command	Description
Code Review	/gemini review	Performs a code review for the current pull request in its current state.
Pull Request Summary	/gemini summary	Provides a summary of the current pull request in its current state.
Comment	@gemini-code-assist	Responds in comments when explicitly tagged, both in pull request comments and review comments.
Help	/gemini help	Displays a list of available commands.

Customization

To customize Gemini Code Assist for GitHub experience, repository maintainers can create a configuration file and/or provide a custom code review style guide (such as PEP-8 for Python) by creating and adding files to a .gemini/ folder in the base of the repository. Detailed instructions can be found here.

Limitations & Feedback

Gemini Code Assist may make mistakes. Please leave feedback on any instances where its feedback is incorrect or counter productive. You can react with 👍 and 👎 on @gemini-code-assist comments. If you're interested in giving your feedback about your experience with Gemini Code Assist for Github and other Google products, sign up here.

You can also get AI-powered code generation, chat, as well as code reviews directly in the IDE at no cost with the Gemini Code Assist IDE Extension.

Footnotes

1. Review the Privacy Notices, Generative AI Prohibited Use Policy, Terms of Service, and learn how to configure Gemini Code Assist in GitHub here. Gemini can make mistakes, so double check it and use code with caution. ↩

[gemini-code-assist]

Code Review

This pull request updates the pg_walstream dependency to version 0.4.1 and refactors the replication client to use LogicalReplicationStream directly, removing the local ReplicationStream wrapper. It also consolidates SQL batch execution logic into a shared helper for MySQL and SQLite destinations, improving maintainability and atomicity. The changes include idiomatic Rust improvements, such as simplified LSN handling and cleaner conditional logic in the transaction manager.