fix: fix ray redundant execution

[ChenZiHong-Gavin]

This PR addresses redundant execution issues within the Ray data processing pipeline by changing how intermediate results are handled. Instead of materializing datasets (occupying large memory), the system now uses disk cache by writing node outputs to JSON files and then lazily re-reading them.

When handling pipelines like below:

we can avoid ray's default redundant task execution.

[gemini-code-assist]

Summary of Changes

Hello @ChenZiHong-Gavin, 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 addresses redundant execution issues within the Ray data processing pipeline by changing how intermediate results are handled. Instead of eagerly materializing datasets, the system now writes node outputs to JSON files and then lazily re-reads them. This approach, coupled with custom filename generation and consistent JSON serialization/deserialization of graph data, aims to improve efficiency and reduce unnecessary computations in the Ray-based graph generation process.

Highlights

• Refactored Ray Dataset Materialization: The Engine.execute method no longer explicitly calls materialize() on Ray datasets. Instead, it writes intermediate node outputs to JSON files and then lazily re-reads them, which can improve efficiency by avoiding redundant computations.
• Custom Filename Provider for Ray Data: A new NodeFilenameProvider class is introduced to allow custom naming conventions for output files when writing Ray datasets to JSON, ensuring files are named based on the node ID and other task-specific details.
• Data Serialization/Deserialization for Node/Edge Data: The partition_service now serializes node and edge data into JSON strings before passing them in a Pandas DataFrame, and the generate_service deserializes these JSON strings back into Python objects. This ensures data consistency when writing to and reading from JSON files.
• Centralized Output Handling: The logic for saving node outputs, previously handled in graphgen/run.py, has been moved into the Engine.execute method, simplifying the main execution flow and making it more self-contained.

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 effectively addresses potential redundant executions in the Ray-based data processing pipeline by introducing checkpointing. Intermediate datasets are now saved to disk and read back lazily, which breaks the computation graph and prevents re-computation. The refactoring moves the saving logic into the Engine class, improving encapsulation. The data serialization strategy has been updated to use JSON strings, ensuring compatibility with the write_json method. My review includes a couple of minor suggestions to improve code clarity and remove obsolete code.

[gemini-code-assist]

This line of code is commented out. It's good practice to remove dead or commented-out code to improve readability and maintainability.