feat(transform): support empty data flow and batch temporary tables

[ChihuaYing]

本次功能与修复

• 修复定时 Batch Transform 状态泄漏：每轮恢复原始 export writer、清空前置 CollectionWriter 缓冲，避免下一轮嵌套旧 PemjaWriter 或混入上一轮数据。
• 移除基于 Mutex 的异步等待，改为同步传播 Python 调用及写入异常，修复空输出时等待不到解锁而死锁、并发文件写入可能丢数据的问题。
• 支持空输入作为正常数据流：SQL、前置 Batch/Stream Transform 的零行结果均会以保留表头的空 BatchData 继续传入用户 Transform。
• 支持合法空输出 [header]：可继续传给下游 Transform；接 SQL 时不创建物理临时列，查询不存在的临时路径得到空结果并正常完成。
• 严格校验 Python 返回值：拒绝 None、空列表、非法/空表头、非列表数据行、列数不一致、非法值类型及列内类型不一致；合法空表保持兼容。
• Python 算子异常和文件写入失败不再只记录日志，而是终止当轮 Job；失败状态可通过 Thrift/Session 查询具体错误信息。
• 支持 Batch Python(outputPrefix) -> SQL，与原 Stream 链路一致写入 IginX 临时表；完善临时表别名的每轮创建、大小写替换、清理及 Quartz 线程复用清理。
• 删除 SQL 最后一条必须以 SELECT/SHOW 开头的校验；无结果流的成功 SQL 会作为空输入继续传递。
• 修正临时表测试断言：临时表查询结果首列是 IginXWriter 生成的非确定时间键，断言其为合法 long，并验证后续计算列 55,55,65。

新增/调整测试

TransformIT 增加并统一命名、标注流水线的测试，覆盖：

• 定时 Batch 多轮隔离。
• Stream/Batch Transform 的空 SQL 输入。
• 用户算子接受、拒绝或基于空输入产生新数据。
• None、空列表、坏表头、坏数据行、类型错误等 7 类非法输出，并断言具体错误信息。
• SQL(empty) -> Transform(empty) -> Transform(empty) 的空数据传播。
• Stream/Batch Transform 空输出接临时表 SQL。
• Batch Transform 写入临时表后由 SQL 消费，并校验聚合结果。

新增测试脚本位于 test/src/test/resources/transform

[Copilot]

Pull request overview

Note

Copilot couldn't run its full agentic review because it didn't start before the timeout. Make sure your repository has a runner available, or add a copilot-code-review.yml file specifying one with the runs-on attribute. See the docs for more details.

This PR enhances the Transform pipeline to correctly support empty inputs/outputs, improve scheduled batch isolation, and propagate Python/write failures as job failures with queryable error details.

Changes:

• Add job failure error message propagation through Thrift/Session APIs and expose root cause text.
• Remove mutex-based async waiting and make Python execution + result writing fail-fast/synchronous; support header-only empty BatchData flow through transforms/SQL.
• Expand integration tests and add Python transformer fixtures to cover empty/invalid outputs, scheduled batch isolation, and temp-table SQL consumption.

Reviewed changes

Copilot reviewed 37 out of 37 changed files in this pull request and generated 3 comments.

Show a summary per file

File	Description
thrift/src/main/thrift/rpc.thrift	Adds optional errorMessage field to transform job status response.
session_py/iginx/iginx_pyclient/thrift/rpc/ttypes.py	Updates generated Python Thrift types to include errorMessage.
session/src/main/java/cn/edu/tsinghua/iginx/session/Session.java	Adds queryTransformJobErrorMessage() API for clients.
core/src/main/java/cn/edu/tsinghua/iginx/transform/utils/Mutex.java	Removes mutex utility no longer used for stage coordination.
core/src/main/java/cn/edu/tsinghua/iginx/transform/pojo/Job.java	Adds batch-python-to-SQL temp table wiring via outputPrefix.
core/src/main/java/cn/edu/tsinghua/iginx/transform/exec/tools/ExecutionMetaManager.java	Adds temp-table meta cleanup and case-insensitive replacement behavior.
core/src/main/java/cn/edu/tsinghua/iginx/transform/exec/TransformJobManager.java	Adds server-side error message extraction for failed jobs.
core/src/main/java/cn/edu/tsinghua/iginx/transform/exec/StreamStageRunner.java	Resets writers per trigger and returns empty RowStream when SQL has no result stream.
core/src/main/java/cn/edu/tsinghua/iginx/transform/exec/ScheduledJob.java	Clears temp-table meta per trigger and stabilizes cleanup logging.
core/src/main/java/cn/edu/tsinghua/iginx/transform/exec/JobValidationChecker.java	Removes “last SQL must be SELECT/SHOW” restriction to allow empty/no-result flows.
core/src/main/java/cn/edu/tsinghua/iginx/transform/exec/BatchStageRunner.java	Removes mutex waiting; resets writers/input buffers between scheduled runs.
core/src/main/java/cn/edu/tsinghua/iginx/transform/driver/PemjaWorker.java	Makes Python processing throw TransformException and validates/propagates invalid output.
core/src/main/java/cn/edu/tsinghua/iginx/transform/driver/IPCWorker.java	Updates exception type to TransformException for writer failures.
core/src/main/java/cn/edu/tsinghua/iginx/transform/data/SplitReader.java	Ensures an empty batch (header-only) is emitted once for empty inputs.
core/src/main/java/cn/edu/tsinghua/iginx/transform/data/RowStreamReader.java	Ensures an empty batch (header-only) is emitted once for empty SQL results.
core/src/main/java/cn/edu/tsinghua/iginx/transform/data/PemjaWriter.java	Updates writer signature to throw TransformException.
core/src/main/java/cn/edu/tsinghua/iginx/transform/data/PemjaReader.java	Strictly validates Python return format, supports header-only empty tables, and enforces column type consistency.
core/src/main/java/cn/edu/tsinghua/iginx/transform/data/LogWriter.java	Skips writes for empty batches to treat header-only output as empty export.
core/src/main/java/cn/edu/tsinghua/iginx/transform/data/IginXWriter.java	Skips writes for empty batches to avoid creating physical temp columns on empty output.
core/src/main/java/cn/edu/tsinghua/iginx/transform/data/FileAppendWriter.java	Makes file export fail-fast on IO errors and batches file writes per batch.
core/src/main/java/cn/edu/tsinghua/iginx/transform/data/Exporter.java	Removes mutex exposure from exporter interface.
core/src/main/java/cn/edu/tsinghua/iginx/transform/data/ExportWriter.java	Removes mutex coordination and propagates TransformException from writes.
core/src/main/java/cn/edu/tsinghua/iginx/transform/data/BatchData.java	Adds isEmpty() helper for header-only empty-batch handling.
core/src/main/java/cn/edu/tsinghua/iginx/transform/api/Writer.java	Changes writeBatch to throw TransformException.
core/src/main/java/cn/edu/tsinghua/iginx/IginxWorker.java	Populates errorMessage in job status response when present.
test/src/test/java/cn/edu/tsinghua/iginx/integration/func/udf/TransformIT.java	Adds extensive IT coverage for empty flows, invalid outputs, temp tables, and scheduled isolation.
test/src/test/resources/transform/transformer_none.py	Adds invalid-output transformer fixture returning None.
test/src/test/resources/transform/transformer_empty_list.py	Adds invalid-output transformer fixture returning [].
test/src/test/resources/transform/transformer_empty_input_pass.py	Adds transformer fixture that accepts header-only input.
test/src/test/resources/transform/transformer_empty_input_fail.py	Adds transformer fixture that errors on header-only input.
test/src/test/resources/transform/transformer_empty_input_emit.py	Adds transformer fixture that emits data on empty input.
test/src/test/resources/transform/transformer_empty_header.py	Adds invalid-output transformer fixture returning [[]].
test/src/test/resources/transform/transformer_empty.py	Adds transformer fixture returning header-only output.
test/src/test/resources/transform/transformer_bad_value.py	Adds invalid-output transformer fixture returning unsupported value type.
test/src/test/resources/transform/transformer_bad_row_type.py	Adds invalid-output transformer fixture returning non-list data row.
test/src/test/resources/transform/transformer_bad_row.py	Adds invalid-output transformer fixture returning wrong row length.
test/src/test/resources/transform/transformer_bad_header.py	Adds invalid-output transformer fixture returning non-string header.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.