fix(transform): support batch transform temp table and scheduled buffer reset#8
Open
ChihuaYing wants to merge 3 commits into
Open
fix(transform): support batch transform temp table and scheduled buffer reset#8ChihuaYing wants to merge 3 commits into
ChihuaYing wants to merge 3 commits into
Conversation
ChihuaYing
commented
Jul 11, 2026
ChihuaYing
commented
Owner
- 修复 Batch transform 输出 writer 未恢复的 bug
- 修复 Batch transform 读取后未清空缓冲区的 bug
- 支持 Batch transform 输出到临时表
- 新增 Batch transform 定时任务相关测试
2. 修复 Batch transform 读取后未清空缓冲区的 bug 3. 支持 Batch transform 输出到临时表 4. 新增 Batch transform 定时任务相关测试
There was a problem hiding this comment.
Pull request overview
Note
Copilot couldn't run its full agentic review because no GitHub Actions runner was available. Make sure your repository has a runner available to run Copilot's review, or add a copilot-setup-steps.yml file specifying one with the runs-on attribute. See the docs for more details.
This PR fixes Batch transform execution issues (writer restoration and buffer reset), adds support for writing Batch transform outputs into temporary tables (for subsequent SQL tasks), and introduces an integration test for scheduled batch transform behavior.
Changes:
- Fix BatchStageRunner to restore the correct writer and reset the collection buffer after reading.
- Update Job planning to optionally route Python batch outputs into a temp table via
outputPrefixwhen followed by an SQL task. - Add a scheduled batch transform YAML and a new integration test covering the scheduled + temp table flow.
Reviewed changes
Copilot reviewed 5 out of 5 changed files in this pull request and generated 4 comments.
Show a summary per file
| File | Description |
|---|---|
| test/src/test/resources/transform/transformer_row_sum.py | Adjusts transformer output dtype to 64-bit integers. |
| test/src/test/resources/transform/TransformScheduledBatchTransform.yaml | Adds a scheduled batch-transform pipeline that writes to a temp table and then reads via SQL. |
| test/src/test/java/cn/edu/tsinghua/iginx/integration/func/udf/TransformIT.java | Adds an integration test verifying scheduled batch transform with temp table output. |
| core/src/main/java/cn/edu/tsinghua/iginx/transform/pojo/Job.java | Adds temp-table writer selection when a Python batch task with outputPrefix is followed by SQL. |
| core/src/main/java/cn/edu/tsinghua/iginx/transform/exec/BatchStageRunner.java | Restores writer initialization and clears the batch collection buffer after reading. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
Comment on lines
71
to
77
| CollectionWriter collectionWriter = | ||
| (CollectionWriter) batchStage.getBeforeStage().getExportWriter(); | ||
| BatchData batchData = collectionWriter.getCollectedData(); | ||
| collectionWriter.reset(); | ||
|
|
||
| mutex.lock(); | ||
| writer.writeBatch(batchData); |
Comment on lines
+480
to
+482
| for (List<Object> row : queryResult.getValues()) { | ||
| assertEquals(expectedSum, row.get(sumIndex)); | ||
| } |
Comment on lines
+446
to
+448
| try { | ||
| Thread.sleep(5000L); // wait for the scheduled job to run at least twice | ||
| verifyScheduledBatchTransformResult(44L); |
| - taskType: "sql" | ||
| timeout: 10000000 | ||
| sqlList: | ||
| - "SELECT s1, s1, s2 FROM batchsum.scheduleBatch;" |
2. 修复 Batch transform 读取后未清空缓冲区的 bug 3. 支持 Batch transform 输出到临时表 4. 新增 Batch transform 定时任务相关测试
…_20260709 # Conflicts: # test/src/test/java/cn/edu/tsinghua/iginx/integration/func/udf/TransformIT.java
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.