file upload interfaces

[dentiny]

Summary

Briefly explain what this PR does.

Related Issues

Closes # or links to related issues.

Changes

Checklist

• Code builds correctly
• Tests have been added or updated
• Documentation updated if necessary
• I have reviewed my own changes

[gemini-code-assist]

Summary of Changes

Hello @dentiny, 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 introduces new interfaces and event types to support file upload operations within the moonlink_connectors REST ingestion system. It refactors the existing event handling to differentiate between row-based and file-based events, laying the groundwork for processing file uploads alongside traditional row inserts, updates, and deletes.

Highlights

• Event Type Segregation: The existing EventRequest and EventOperation types have been refactored into distinct RowEventRequest, RowEventOperation for row-based events, and new FileEventRequest, FileEventOperation for file-based events.
• Unified Event Handling: The top-level EventRequest is now an enum that can encapsulate either a RowEventRequest or a FileEventRequest, allowing for a unified entry point for different event types.
• File Event Integration: A new FileEvent variant has been added to the RestEvent enum, and a placeholder process_file_event function has been introduced in RestSink to handle future file upload logic.
• Codebase Adaptation: Imports and function calls across replication_connection.rs, rest_ingest.rs, and moonlink_rest_sink.rs have been updated to reflect the new event type definitions.

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 in your pull request via creating an issue comment (i.e. comment on the pull request page) using either /gemini <command> or @gemini-code-assist <command>. Below is a summary of the supported commands.

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 issue 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 lays the groundwork for handling file upload interfaces by refactoring existing event structures and introducing new ones for file-based operations. While the new interfaces and type definitions are a good start, the core implementation for processing file uploads is missing. Key functions in both rest_source.rs and moonlink_rest_sink.rs are currently stubs. My review focuses on these missing implementations, which are critical for the feature to be functional, and also points out a design inconsistency in the new event structure.

[gemini-code-assist]

The FileEvent arm is not implemented. It currently ignores src_table_id and table_events, meaning any file upload events will be dropped. This should be implemented to handle the file events, for example by calling process_file_event.

Please note that process_file_event's signature expects an operation which is not present in RestEvent::FileEvent. This design inconsistency will need to be addressed.

            RestEvent::FileEvent {
                src_table_id,
                table_events,
            } => {
                // TODO: Implement file event processing.
                // This will likely involve calling `self.process_file_event(...)`.
                // Note that `RestEvent::FileEvent` will need to be updated to include
                // the `FileEventOperation` for this to work.
                Ok(())
            }

[gemini-code-assist]

The process_file_event function is a stub and does not implement any logic for processing file events. For the file upload feature to work, this function must be implemented to consume the table_events stream and apply the changes. The operation and src_table_id parameters are also currently unused.

[gemini-code-assist]

The process_file_request function is a stub and returns an empty vector of events. This is the entry point for creating file-based RestEvents and needs to be implemented. The implementation should likely create a channel, spawn a background task to read the specified files and send TableEvents, and then return a RestEvent::FileEvent containing the receiver end of the channel.

[gemini-code-assist]

RestEvent::FileEvent is missing the operation field. The corresponding FileEventRequest contains an operation, but this information is lost when creating a RestEvent. The sink (moonlink_rest_sink.rs) has a process_file_event function that expects a FileEventOperation. To maintain consistency and enable correct processing in the sink, the operation should be included here.

    FileEvent {
        src_table_id: SrcTableId,
        operation: FileEventOperation,
        table_events: Arc<tokio::sync::mpsc::UnboundedReceiver<TableEvent>>,
    },