This is an automated email from the ASF dual-hosted git repository.
KevinyhZou pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/gluten.git
The following commit(s) were added to refs/heads/main by this push:
new e7d526abc3 [GLUTEN-13450][FLINK][VL] Support
TRIM/LTRIM/RTRIM/LPAD/RPAD to the velox flinksql functions. (#13460)
e7d526abc3 is described below
commit e7d526abc3d9b6356fdff6954fa810bac9636810
Author: 1111_nit <[email protected]>
AuthorDate: Mon Nov 2 11:52:23 2026 +0800
[GLUTEN-13450][FLINK][VL] Support TRIM/LTRIM/RTRIM/LPAD/RPAD to the velox
flinksql functions. (#13460)
* [GLUTEN-13450][VL] Support TRIM/LTRIM/RTRIM/LPAD/RPAD in the Flink planner
---
.github/workflows/flink.yml | 2 +-
gluten-flink/docs/Flink.md | 2 +-
.../rexnode/functions/RexCallConverterFactory.java | 13 ++
.../rexnode/functions/TrimRexCallConverter.java | 100 +++++++++++++
.../functions/TrimSeekRexCallConverter.java | 75 ++++++++++
.../stream/custom/TrimPadFunctionsTest.java | 166 +++++++++++++++++++++
6 files changed, 356 insertions(+), 2 deletions(-)
diff --git a/.github/workflows/flink.yml b/.github/workflows/flink.yml
index 38eedbd926..994e9c4a6a 100644
--- a/.github/workflows/flink.yml
+++ b/.github/workflows/flink.yml
@@ -89,7 +89,7 @@ jobs:
export folly_SOURCE=BUNDLED
git clone -b gluten-20260829 https://github.com/bigo-sg/velox4j.git
cd velox4j
- git reset --hard 2d08448f556bff669fe413874d04dd8cd96f65b2
+ git reset --hard cc8b56d74baa3863ae0fb8387f5745652370fa7d
git apply $GITHUB_WORKSPACE/gluten-flink/patches/fix-velox4j.patch
$GITHUB_WORKSPACE/build/mvn clean install -DskipTests -Dgpg.skip
-Dspotless.skip=true
cd ..
diff --git a/gluten-flink/docs/Flink.md b/gluten-flink/docs/Flink.md
index 6530031f08..cd7b5b4e1d 100644
--- a/gluten-flink/docs/Flink.md
+++ b/gluten-flink/docs/Flink.md
@@ -48,7 +48,7 @@ As some features have not been committed to upstream, you
have to use the follow
## fetch velox4j code
git clone -b gluten-20260829 https://github.com/bigo-sg/velox4j.git
cd velox4j
-git reset --hard 2d08448f556bff669fe413874d04dd8cd96f65b2
+git reset --hard cc8b56d74baa3863ae0fb8387f5745652370fa7d
mvn clean install -DskipTests -Dgpg.skip -Dspotless.skip=true
```
**Get gluten**
diff --git
a/gluten-flink/planner/src/main/java/org/apache/gluten/rexnode/functions/RexCallConverterFactory.java
b/gluten-flink/planner/src/main/java/org/apache/gluten/rexnode/functions/RexCallConverterFactory.java
index c64b425f55..4f0547a144 100644
---
a/gluten-flink/planner/src/main/java/org/apache/gluten/rexnode/functions/RexCallConverterFactory.java
+++
b/gluten-flink/planner/src/main/java/org/apache/gluten/rexnode/functions/RexCallConverterFactory.java
@@ -120,6 +120,19 @@ public class RexCallConverterFactory {
// negative counts from the end) already match Flink.
Map.entry("SUBSTRING", Arrays.asList(() -> new
SubstringRexCallConverter())),
Map.entry("IFNULL", Arrays.asList(() -> new
ScalarOnlyRexCallConverter("ifnull"))),
+ // TRIM is normalized by Calcite to (flag, seek, str); the converter
+ // maps the flag to ltrim/rtrim/trim and reorders to (str, seek).
+ Map.entry("TRIM", Arrays.asList(() -> new TrimRexCallConverter())),
+ // LTRIM/RTRIM keep the Flink argument order (str[, seek]); the
+ // converter routes a two-argument call by the seek constancy:
+ // a constant seek takes the flinksql character-set signatures, a
+ // runtime-sourced seek takes the *_runtime ones (Flink's
+ // isSpaceString degrades a seek starting with a space to
+ // space-only trimming for segment-backed strings).
+ Map.entry("LTRIM", Arrays.asList(() -> new
TrimSeekRexCallConverter("ltrim"))),
+ Map.entry("RTRIM", Arrays.asList(() -> new
TrimSeekRexCallConverter("rtrim"))),
+ Map.entry("LPAD", Arrays.asList(() -> new
DefaultRexCallConverter("lpad"))),
+ Map.entry("RPAD", Arrays.asList(() -> new
DefaultRexCallConverter("rpad"))),
Map.entry("COALESCE", Arrays.asList(() -> new
DefaultRexCallConverter("coalesce"))));
public static RexCallConverter getConverter(RexCall callNode,
RexConversionContext context) {
diff --git
a/gluten-flink/planner/src/main/java/org/apache/gluten/rexnode/functions/TrimRexCallConverter.java
b/gluten-flink/planner/src/main/java/org/apache/gluten/rexnode/functions/TrimRexCallConverter.java
new file mode 100644
index 0000000000..ec92fda6b9
--- /dev/null
+++
b/gluten-flink/planner/src/main/java/org/apache/gluten/rexnode/functions/TrimRexCallConverter.java
@@ -0,0 +1,100 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.gluten.rexnode.functions;
+
+import org.apache.gluten.rexnode.RexConversionContext;
+import org.apache.gluten.rexnode.RexNodeConverter;
+import org.apache.gluten.rexnode.ValidationResult;
+
+import io.github.zhztheplayer.velox4j.expression.CallTypedExpr;
+import io.github.zhztheplayer.velox4j.expression.TypedExpr;
+
+import org.apache.calcite.rex.RexCall;
+import org.apache.calcite.rex.RexLiteral;
+import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.sql.fun.SqlTrimFunction;
+
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * Converts Flink TRIM to the velox flinksql trim family.
+ *
+ * <p>Calcite normalizes every TRIM call to three operands {@code (flag, seek,
str)}: {@code
+ * TRIM(s)} desugars to {@code (BOTH, ' ', s)}, and {@code TRIM(LEADING 'x'
FROM c)} to {@code
+ * (LEADING, 'x', c)}. The flag is a RexLiteral carrying a {@link
SqlTrimFunction.Flag}.
+ *
+ * <p>The velox flinksql signatures take the Flink argument order {@code (str,
seek)}, so the
+ * operands are reordered to {@code (operand[2], operand[1])} and the flag
selects the target
+ * function: LEADING → {@code ltrim}, TRAILING → {@code rtrim}, BOTH → {@code
trim}. All three
+ * accept the optional second argument (a character set, not a prefix).
+ *
+ * <p>The seek picks the flavor by constancy, mirroring Flink's {@code
+ * BinaryStringDataUtil.isSpaceString}: a plan constant keeps its Java object
and takes the
+ * character-set walk ({@code trim}/{@code ltrim}/{@code rtrim}); a
runtime-sourced seek (column or
+ * computed expression) is segment-backed and Flink checks only its first
byte, degrading a seek
+ * that starts with a space to space-only trimming ({@code trim_runtime} et
al.).
+ *
+ * <p>A non-literal flag (a pathological plan) is rejected by {@link
#isSuitable}.
+ */
+public class TrimRexCallConverter extends BaseRexCallConverter {
+
+ public TrimRexCallConverter() {
+ super("trim");
+ }
+
+ @Override
+ public ValidationResult isSuitable(RexCall callNode, RexConversionContext
context) {
+ List<RexNode> operands = callNode.getOperands();
+ if (operands.size() != 3 || !(operands.get(0) instanceof RexLiteral)) {
+ return ValidationResult.failure("TRIM requires a literal trim flag");
+ }
+ if (((RexLiteral) operands.get(0)).getValueAs(SqlTrimFunction.Flag.class)
== null) {
+ return ValidationResult.failure("TRIM requires a literal trim flag");
+ }
+ return ValidationResult.success();
+ }
+
+ @Override
+ public TypedExpr toTypedExpr(RexCall callNode, RexConversionContext context)
{
+ List<RexNode> operands = callNode.getOperands();
+ SqlTrimFunction.Flag flag =
+ ((RexLiteral) operands.get(0)).getValueAs(SqlTrimFunction.Flag.class);
+ String functionName;
+ switch (flag) {
+ case LEADING:
+ functionName = "ltrim";
+ break;
+ case TRAILING:
+ functionName = "rtrim";
+ break;
+ default:
+ functionName = "trim";
+ break;
+ }
+ // A runtime-sourced seek takes the *_runtime flavor (Flink's isSpaceString
+ // degrades a seek starting with a space to space-only trimming).
+ if (!TrimSeekFlavor.isConstantSeek(operands.get(1))) {
+ functionName = functionName + "_runtime";
+ }
+ // Flink/velox argument order: (str, seek) = (operand[2], operand[1]).
+ List<TypedExpr> params = new ArrayList<>();
+ params.add(RexNodeConverter.toTypedExpr(operands.get(2), context));
+ params.add(RexNodeConverter.toTypedExpr(operands.get(1), context));
+ return new CallTypedExpr(getResultType(callNode), params, functionName);
+ }
+}
diff --git
a/gluten-flink/planner/src/main/java/org/apache/gluten/rexnode/functions/TrimSeekRexCallConverter.java
b/gluten-flink/planner/src/main/java/org/apache/gluten/rexnode/functions/TrimSeekRexCallConverter.java
new file mode 100644
index 0000000000..aa0616534e
--- /dev/null
+++
b/gluten-flink/planner/src/main/java/org/apache/gluten/rexnode/functions/TrimSeekRexCallConverter.java
@@ -0,0 +1,75 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.gluten.rexnode.functions;
+
+import org.apache.gluten.rexnode.RexConversionContext;
+import org.apache.gluten.rexnode.ValidationResult;
+
+import io.github.zhztheplayer.velox4j.expression.CallTypedExpr;
+import io.github.zhztheplayer.velox4j.expression.TypedExpr;
+
+import org.apache.calcite.rex.RexCall;
+import org.apache.calcite.rex.RexLiteral;
+import org.apache.calcite.rex.RexNode;
+
+import java.util.List;
+
+/**
+ * Converts the Flink two-argument LTRIM/RTRIM to the velox flinksql
functions, routing by the
+ * constancy of the seek.
+ *
+ * <p>Flink's {@code BinaryStringDataUtil.isSpaceString} keeps the Java object
of a plan constant
+ * (character-set walk: {@code ltrim}/{@code rtrim}) while a runtime-sourced
seek — a column or a
+ * computed expression — is segment-backed and only its first byte is checked,
degrading a seek that
+ * starts with a space to space-only trimming ({@code ltrim_runtime}/{@code
rtrim_runtime}). See
+ * also {@link TrimRexCallConverter}.
+ *
+ * <p>The one-argument form ({@code LTRIM(s)}) maps to the plain function
directly.
+ */
+class TrimSeekRexCallConverter extends BaseRexCallConverter {
+
+ TrimSeekRexCallConverter(String functionName) {
+ super(functionName);
+ }
+
+ @Override
+ public ValidationResult isSuitable(RexCall callNode, RexConversionContext
context) {
+ if (callNode.getOperands().size() > 2) {
+ return ValidationResult.failure("LTRIM/RTRIM take at most two operands");
+ }
+ return ValidationResult.success();
+ }
+
+ @Override
+ public TypedExpr toTypedExpr(RexCall callNode, RexConversionContext context)
{
+ List<RexNode> operands = callNode.getOperands();
+ String name = functionName;
+ if (operands.size() == 2 &&
!TrimSeekFlavor.isConstantSeek(operands.get(1))) {
+ name = name + "_runtime";
+ }
+ return new CallTypedExpr(getResultType(callNode), getParams(callNode,
context), name);
+ }
+}
+
+final class TrimSeekFlavor {
+ private TrimSeekFlavor() {}
+
+ /** A RexLiteral seek maps to a velox constant; anything else is
runtime-sourced. */
+ static boolean isConstantSeek(RexNode seek) {
+ return seek instanceof RexLiteral;
+ }
+}
diff --git
a/gluten-flink/ut/src/test/java/org/apache/gluten/table/runtime/stream/custom/TrimPadFunctionsTest.java
b/gluten-flink/ut/src/test/java/org/apache/gluten/table/runtime/stream/custom/TrimPadFunctionsTest.java
new file mode 100644
index 0000000000..73c2c2cb8c
--- /dev/null
+++
b/gluten-flink/ut/src/test/java/org/apache/gluten/table/runtime/stream/custom/TrimPadFunctionsTest.java
@@ -0,0 +1,166 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.gluten.table.runtime.stream.custom;
+
+import org.apache.gluten.table.runtime.stream.common.GlutenStreamingTestBase;
+
+import org.apache.flink.types.Row;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.util.Arrays;
+import java.util.List;
+
+// End-to-end tests for TRIM/LTRIM/RTRIM/LPAD/RPAD on the velox flinksql
+// implementations. Expected values follow Flink semantics: 1-arg trim removes
+// only 0x20 spaces (tabs preserved), the 2-arg trim seek is a character set,
+// and LPAD/RPAD return NULL for a negative length or an empty pad.
+class TrimPadFunctionsTest extends GlutenStreamingTestBase {
+
+ @Override
+ @BeforeEach
+ public void before() throws Exception {
+ super.before();
+ }
+
+ @Test
+ void testTrimSingleArg() {
+ List<Row> rows =
+ Arrays.asList(
+ Row.of(1, " hello "), Row.of(2, "\t ab \t"), Row.of(3, " "),
Row.of(4, "world"));
+ createSimpleBoundedValuesTable("tblTrimSingle", "a int, s string", rows);
+
+ String query = "select TRIM(s), LTRIM(s), RTRIM(s) from tblTrimSingle";
+ runAndCheck(
+ query,
+ Arrays.asList(
+ "+I[hello, hello , hello]",
+ "+I[\t ab \t, \t ab \t, \t ab \t]",
+ "+I[, , ]",
+ "+I[world, world, world]"));
+ }
+
+ @Test
+ void testTrimFlags() {
+ List<Row> rows = Arrays.asList(Row.of(1, "xxhiyxx"));
+ createSimpleBoundedValuesTable("tblTrimFlags", "a int, s string", rows);
+
+ String query =
+ "select TRIM(LEADING 'x' FROM s), TRIM(TRAILING 'x' FROM s), TRIM(BOTH
'x' FROM s)"
+ + " from tblTrimFlags";
+ runAndCheck(query, Arrays.asList("+I[hiyxx, xxhiy, hiy]"));
+ }
+
+ @Test
+ void testTrimDefaultBothSpaces() {
+ List<Row> rows = Arrays.asList(Row.of(1, " data "));
+ createSimpleBoundedValuesTable("tblTrimBoth", "a int, s string", rows);
+
+ String query = "select TRIM(s) from tblTrimBoth";
+ runAndCheck(query, Arrays.asList("+I[data]"));
+ }
+
+ @Test
+ void testTrimWithColumnSeek() {
+ // The seek column data covers both flavors: seeks not starting with a
+ // space take the character-set walk, while Flink's isSpaceString degrades
+ // a seek whose first byte is a space to space-only trimming for
+ // segment-backed (column-sourced) strings.
+ List<Row> rows =
+ Arrays.asList(
+ Row.of(1, "xxabx", "x"),
+ Row.of(2, "abba", "ab"),
+ Row.of(3, "abc", ""),
+ Row.of(4, "abx ", " x"),
+ Row.of(5, " xab", " x"));
+ createSimpleBoundedValuesTable("tblTrimSeek", "a int, c1 string, c2
string", rows);
+
+ String query = "select TRIM(c2 FROM c1) from tblTrimSeek";
+ runAndCheck(query, Arrays.asList("+I[ab]", "+I[]", "+I[abc]", "+I[abx]",
"+I[xab]"));
+ }
+
+ @Test
+ void testLtrimRtrimWithColumnSeek() {
+ // Column-sourced seek starting with a space: Flink trims spaces only
+ // (BinaryStringDataUtil.isSpaceString checks byteAt(0) for segment-backed
+ // strings), matching the *_runtime velox functions.
+ List<Row> rows = Arrays.asList(Row.of(1, "abx ", " x"));
+ createSimpleBoundedValuesTable("tblLtrimRtrimColSeek", "a int, c1 string,
c2 string", rows);
+
+ String query = "select LTRIM(c1, c2), RTRIM(c1, c2) from
tblLtrimRtrimColSeek";
+ runAndCheck(query, Arrays.asList("+I[abx , abx]"));
+ }
+
+ @Test
+ void testLtrimRtrimWithSeek() {
+ List<Row> rows = Arrays.asList(Row.of(1, "xxabx"));
+ createSimpleBoundedValuesTable("tblLtrimRtrimSeek", "a int, s string",
rows);
+
+ // A literal seek keeps its Java object in Flink and always takes the
+ // character-set walk, even when it starts with a space.
+ String query =
+ "select LTRIM(s, 'x'), RTRIM(s, 'x'), LTRIM(s, ' x'), RTRIM(s, ' x')"
+ + " from tblLtrimRtrimSeek";
+ runAndCheck(query, Arrays.asList("+I[abx, xxab, abx, xxab]"));
+ }
+
+ @Test
+ void testLpadRpad() {
+ List<Row> rows = Arrays.asList(Row.of(1, "text"));
+ createSimpleBoundedValuesTable("tblPad", "a int, s string", rows);
+
+ String query =
+ "select LPAD(s, 5, 'x'), RPAD(s, 5, 'x'), LPAD(s, 3, 'xy'), RPAD(s, 3,
'xy'),"
+ + " LPAD(s, 0, 'e'), LPAD(s, 7, 'xy'), RPAD(s, 7, 'xy') from
tblPad";
+ runAndCheck(query, Arrays.asList("+I[xtext, textx, tex, tex, , xyxtext,
textxyx]"));
+ }
+
+ @Test
+ void testPadInvalidArgs() {
+ List<Row> rows = Arrays.asList(Row.of(1, "abc"));
+ createSimpleBoundedValuesTable("tblPadInvalid", "a int, s string", rows);
+
+ String query =
+ "select LPAD(s, -1, 'e'), RPAD(s, -1, 'e'), LPAD(s, 5, ''), RPAD(s, 5,
'')"
+ + " from tblPadInvalid";
+ runAndCheck(query, Arrays.asList("+I[null, null, null, null]"));
+ }
+
+ @Test
+ void testNullPropagation() {
+ List<Row> rows =
+ Arrays.asList(
+ Row.of(1, null, 5, "x"), Row.of(2, "abc", null, "x"), Row.of(3,
"abc", 5, null));
+ createSimpleBoundedValuesTable("tblTrimPadNull", "a int, s string, b int,
p string", rows);
+
+ String query = "select TRIM(s), LTRIM(s), RTRIM(s), LPAD(s, b, p) from
tblTrimPadNull";
+ runAndCheck(
+ query,
+ Arrays.asList(
+ "+I[null, null, null, null]", "+I[abc, abc, abc, null]", "+I[abc,
abc, abc, null]"));
+ }
+
+ @Test
+ void testNested() {
+ List<Row> rows = Arrays.asList(Row.of(1, " data "));
+ createSimpleBoundedValuesTable("tblNested", "a int, s string", rows);
+
+ String query = "select LPAD(TRIM(s), 9, '*'), RPAD(TRIM(s), 9, '*') from
tblNested";
+ runAndCheck(query, Arrays.asList("+I[*****data, data*****]"));
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
