This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/dev/pr-12528-969a46d29d60ab0e3aa81a22a4cc9b483081bdb3 in repository https://gitbox.apache.org/repos/asf/seatunnel.git
commit 0b22b63495373953183b8e8563f83ca06a49493f Author: Jast <[email protected]> AuthorDate: Thu Oct 22 15:46:18 2026 +0000 [Improve][Transform-V2] Add fast path for plain column projections in Zeta SQL engine (#12528) Co-authored-by: jast <[email protected]> Co-authored-by: Claude Code <[email protected]> Co-authored-by: multica-agent <[email protected]> --- .../transform/sql/zeta/ZetaSQLEngine.java | 58 ++++++++++++++++++++++ .../transform/sql/zeta/ZetaSQLEngineTest.java | 35 +++++++++++++ 2 files changed, 93 insertions(+) diff --git a/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/sql/zeta/ZetaSQLEngine.java b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/sql/zeta/ZetaSQLEngine.java index c998582665..fcf39de014 100644 --- a/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/sql/zeta/ZetaSQLEngine.java +++ b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/sql/zeta/ZetaSQLEngine.java @@ -75,6 +75,12 @@ public class ZetaSQLEngine implements SQLEngine { private Integer allColumnsCount = null; private boolean udfOpened; + /** + * Field indexes for the common {@code SELECT source_column AS target_column} projection. + * Keeping this plan avoids evaluating a full SQL expression tree for every row. + */ + @Nullable private int[] directProjectionIndexes; + public ZetaSQLEngine() {} @Override @@ -148,6 +154,7 @@ public class ZetaSQLEngine implements SQLEngine { // validate SQL statement validateSQL(statement); this.selectBody = (PlainSelect) ((Select) statement).getSelectBody(); + prepareDirectProjection(); } catch (JSQLParserException e) { throw new TransformException( CommonErrorCodeDeprecated.UNSUPPORTED_OPERATION, @@ -155,6 +162,43 @@ public class ZetaSQLEngine implements SQLEngine { } } + /** + * Enables the fast path only for plain column projections. Any expression that needs the + * original evaluator, such as literals, nested fields, functions, or wildcards, leaves this + * disabled. + */ + private void prepareDirectProjection() { + directProjectionIndexes = null; + if (selectBody.getWhere() != null + || selectBody.getHaving() != null + || selectBody.getDistinct() != null) { + return; + } + + List<SelectItem<?>> selectItems = selectBody.getSelectItems(); + int[] projectionIndexes = new int[selectItems.size()]; + for (int i = 0; i < selectItems.size(); i++) { + Expression expression = selectItems.get(i).getExpression(); + if (!(expression instanceof Column)) { + return; + } + + String columnName = ((Column) expression).getColumnName(); + int index = inputRowType.indexOf(columnName, false); + if (index == -1 + && columnName.startsWith(ESCAPE_IDENTIFIER) + && columnName.endsWith(ESCAPE_IDENTIFIER)) { + columnName = columnName.substring(1, columnName.length() - 1); + index = inputRowType.indexOf(columnName, false); + } + if (index == -1) { + return; + } + projectionIndexes[i] = index; + } + directProjectionIndexes = projectionIndexes; + } + private void validateSQL(Statement statement) { try { if (!(statement instanceof Select)) { @@ -320,6 +364,20 @@ public class ZetaSQLEngine implements SQLEngine { } private Object[] project(Object[] inputFields) { + if (directProjectionIndexes != null) { + Object[] fields = new Object[directProjectionIndexes.length]; + List<SelectItem<?>> selectItems = selectBody.getSelectItems(); + for (int i = 0; i < directProjectionIndexes.length; i++) { + try { + fields[i] = inputFields[directProjectionIndexes[i]]; + } catch (Exception e) { + throw TransformCommonError.sqlExpressionError( + selectItems.get(i).getExpression().toString(), e); + } + } + return fields; + } + List<SelectItem<?>> selectItems = selectBody.getSelectItems(); int columnsSize = countColumnsSize(selectItems); diff --git a/seatunnel-transforms-v2/src/test/java/org/apache/seatunnel/transform/sql/zeta/ZetaSQLEngineTest.java b/seatunnel-transforms-v2/src/test/java/org/apache/seatunnel/transform/sql/zeta/ZetaSQLEngineTest.java index ce531acbf5..d508d2d1b6 100644 --- a/seatunnel-transforms-v2/src/test/java/org/apache/seatunnel/transform/sql/zeta/ZetaSQLEngineTest.java +++ b/seatunnel-transforms-v2/src/test/java/org/apache/seatunnel/transform/sql/zeta/ZetaSQLEngineTest.java @@ -152,6 +152,41 @@ public class ZetaSQLEngineTest { Assertions.assertEquals(1, failedUdf.getCloseCount()); } + @Test + public void testDirectColumnProjection() { + SeaTunnelRowType rowType = simpleRowType(); + ZetaSQLEngine engine = new ZetaSQLEngine(); + engine.init("test", "test", rowType, "select id as id2, name as name2 from test"); + + SeaTunnelRowType outType = engine.typeMapping(new ArrayList<>()); + Assertions.assertArrayEquals(new String[] {"id2", "name2"}, outType.getFieldNames()); + + List<SeaTunnelRow> outRows = + engine.transformBySQL(new SeaTunnelRow(new Object[] {1, "Alice", 20}), outType); + + Assertions.assertNotNull(outRows); + Assertions.assertEquals(1, outRows.size()); + Assertions.assertArrayEquals(new Object[] {1, "Alice"}, outRows.get(0).getFields()); + engine.close(); + } + + @Test + public void testDirectColumnProjectionReordersAndQuotesColumns() { + SeaTunnelRowType rowType = simpleRowType(); + ZetaSQLEngine engine = new ZetaSQLEngine(); + engine.init("test", "test", rowType, "select `name`, id from test"); + + SeaTunnelRowType outType = engine.typeMapping(new ArrayList<>()); + + List<SeaTunnelRow> outRows = + engine.transformBySQL(new SeaTunnelRow(new Object[] {1, "Alice", 20}), outType); + + Assertions.assertNotNull(outRows); + Assertions.assertEquals(1, outRows.size()); + Assertions.assertArrayEquals(new Object[] {"Alice", 1}, outRows.get(0).getFields()); + engine.close(); + } + private static final class TestableZetaSQLEngine extends ZetaSQLEngine { private final List<ZetaUDF> testUdfs;
