This is an automated email from the ASF dual-hosted git repository.
nzw921rx pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/seatunnel.git
The following commit(s) were added to refs/heads/dev by this push:
new d01059082a [Improve][Transform-V2] Add fast path for plain column
projections in Zeta SQL engine (#12528)
d01059082a is described below
commit d01059082a91045512cc8c654f9e45272d4baaac
Author: Jast <[email protected]>
AuthorDate: Fri Oct 23 00:46:51 2026 +0800
[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;
