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 9747480c38 feat(gluten-flink): support stateless scalar expressions in
planner (#12666)
9747480c38 is described below
commit 9747480c389cb0d19f894936047521e5e91eab58
Author: GGboom <[email protected]>
AuthorDate: Wed Oct 14 15:14:53 2026 +0800
feat(gluten-flink): support stateless scalar expressions in planner (#12666)
---
.../apache/gluten/rexnode/RexNodeConverter.java | 8 ++
.../rexnode/functions/RexCallConverterFactory.java | 9 ++-
.../functions/SubstringRexCallConverter.java | 67 ++++++++++++++++
.../runtime/stream/custom/ScalarFunctionsTest.java | 93 ++++++++++++++++++++++
4 files changed, 176 insertions(+), 1 deletion(-)
diff --git
a/gluten-flink/planner/src/main/java/org/apache/gluten/rexnode/RexNodeConverter.java
b/gluten-flink/planner/src/main/java/org/apache/gluten/rexnode/RexNodeConverter.java
index 3cc61a2593..297063cf1a 100644
---
a/gluten-flink/planner/src/main/java/org/apache/gluten/rexnode/RexNodeConverter.java
+++
b/gluten-flink/planner/src/main/java/org/apache/gluten/rexnode/RexNodeConverter.java
@@ -32,6 +32,7 @@ import io.github.zhztheplayer.velox4j.variant.DoubleValue;
import io.github.zhztheplayer.velox4j.variant.HugeIntValue;
import io.github.zhztheplayer.velox4j.variant.IntegerValue;
import io.github.zhztheplayer.velox4j.variant.SmallIntValue;
+import io.github.zhztheplayer.velox4j.variant.TimestampValue;
import io.github.zhztheplayer.velox4j.variant.TinyIntValue;
import io.github.zhztheplayer.velox4j.variant.VarBinaryValue;
import io.github.zhztheplayer.velox4j.variant.VarCharValue;
@@ -130,6 +131,13 @@ public class RexNodeConverter {
}
case SYMBOL:
return new VarCharValue(literal.getValue().toString());
+ case TIMESTAMP:
+ // RexLiteral#getValueAs does not support java.sql.Timestamp, so take
the
+ // epoch millis directly. It keeps the wall time as UTC, which is the
same
+ // base as TimestampData#getMillisecond used by the row serializer.
+ long epochMillis = literal.getValueAs(Long.class);
+ return TimestampValue.create(
+ Math.floorDiv(epochMillis, 1000L), Math.floorMod(epochMillis,
1000L) * 1_000_000L);
default:
throw new RuntimeException(
"Unsupported rex node type: " +
literal.getType().getSqlTypeName());
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 cb990204e9..4254e8ff82 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
@@ -112,7 +112,14 @@ public class RexCallConverterFactory {
Map.entry("EXTRACT", Arrays.asList(() -> new
DefaultRexCallConverter("extract"))),
Map.entry("IS TRUE", Arrays.asList(() -> new
IsTrueRexCallConverter())),
Map.entry("CONCAT", Arrays.asList(() -> new
DefaultRexCallConverter("concat"))),
- Map.entry("CONCAT_WS", Arrays.asList(() -> new
DefaultRexCallConverter("concat_ws"))));
+ Map.entry("CONCAT_WS", Arrays.asList(() -> new
DefaultRexCallConverter("concat_ws"))),
+ // A literal negative length is rejected by SubstringRexCallConverter
+ // (stopgap until a Flink-semantics substring in the velox flinksql
+ // function set returns NULL): Flink returns NULL while velox
+ // substring returns an empty string. Start semantics (0 counts as 1,
+ // negative counts from the end) already match Flink.
+ Map.entry("SUBSTRING", Arrays.asList(() -> new
SubstringRexCallConverter())),
+ Map.entry("COALESCE", Arrays.asList(() -> new
DefaultRexCallConverter("coalesce"))));
public static RexCallConverter getConverter(RexCall callNode,
RexConversionContext context) {
String operatorName = callNode.getOperator().getName();
diff --git
a/gluten-flink/planner/src/main/java/org/apache/gluten/rexnode/functions/SubstringRexCallConverter.java
b/gluten-flink/planner/src/main/java/org/apache/gluten/rexnode/functions/SubstringRexCallConverter.java
new file mode 100644
index 0000000000..d134c489bf
--- /dev/null
+++
b/gluten-flink/planner/src/main/java/org/apache/gluten/rexnode/functions/SubstringRexCallConverter.java
@@ -0,0 +1,67 @@
+/*
+ * 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 org.apache.calcite.rex.RexCall;
+import org.apache.calcite.rex.RexLiteral;
+import org.apache.calcite.rex.RexNode;
+
+/**
+ * Converts Flink SUBSTRING to the velox substring function (1-based; the SQL
{@code SUBSTRING(x
+ * FROM n FOR m)} form desugars to the same call).
+ *
+ * <p>Start semantics already match: Flink's runtime ({@code
BinaryStringDataUtil#substringSQL})
+ * treats a start of 0 like 1 and counts a negative start from the string end
— the same rules velox
+ * substring implements — so starts convert as-is. Only a start more negative
than the string length
+ * diverges (Flink returns an empty string, velox returns a prefix from the
head), but whether it
+ * triggers depends on row data and cannot be checked at plan time.
+ *
+ * <p>A literal negative length is rejected: Flink returns NULL while velox
returns an empty string.
+ *
+ * <p>TODO: the rejection is a stopgap. The proper fix is a Flink-semantics
substring function in
+ * the velox flinksql function set that returns NULL for a negative length
(and an empty string for
+ * an over-long negative start). A non-literal length cannot be checked at
plan time and is still
+ * converted as-is; both gaps close once SUBSTRING maps to that function and
this rejection is
+ * lifted.
+ */
+public class SubstringRexCallConverter extends DefaultRexCallConverter {
+
+ public SubstringRexCallConverter() {
+ super("substring");
+ }
+
+ @Override
+ public ValidationResult isSuitable(RexCall callNode, RexConversionContext
context) {
+ if (callNode.getOperands().size() > 2) {
+ RexNode length = callNode.getOperands().get(2);
+ if (length instanceof RexLiteral) {
+ Integer lengthValue = ((RexLiteral) length).getValueAs(Integer.class);
+ if (lengthValue != null && lengthValue < 0) {
+ return ValidationResult.failure(
+ String.format(
+ "SUBSTRING with literal length %d is not supported: Flink
returns NULL for a"
+ + " negative length while velox substring returns an
empty string",
+ lengthValue));
+ }
+ }
+ }
+ return ValidationResult.success();
+ }
+}
diff --git
a/gluten-flink/ut/src/test/java/org/apache/gluten/table/runtime/stream/custom/ScalarFunctionsTest.java
b/gluten-flink/ut/src/test/java/org/apache/gluten/table/runtime/stream/custom/ScalarFunctionsTest.java
index ffc2c9f99f..750d06648f 100644
---
a/gluten-flink/ut/src/test/java/org/apache/gluten/table/runtime/stream/custom/ScalarFunctionsTest.java
+++
b/gluten-flink/ut/src/test/java/org/apache/gluten/table/runtime/stream/custom/ScalarFunctionsTest.java
@@ -32,6 +32,8 @@ import java.time.format.DateTimeFormatter;
import java.util.Arrays;
import java.util.List;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
class ScalarFunctionsTest extends GlutenStreamingTestBase {
@Override
@@ -354,4 +356,95 @@ class ScalarFunctionsTest extends GlutenStreamingTestBase {
String query = "select a from tblIsNull where b is null";
runAndCheck(query, Arrays.asList("+I[2]"));
}
+
+ @Test
+ void testSubstring() {
+ List<Row> rows =
+ Arrays.asList(Row.of("hello world"), Row.of("abcdefghij"),
Row.of((Object) null));
+ createSimpleBoundedValuesTable("tblSubstr", "s varchar NULL", rows);
+ // 3-arg and 2-arg forms; start beyond the string length yields an empty
+ // string; a zero length is legal on both sides and anchors the rejection
+ // threshold at length < 0; NULL input propagates.
+ String query =
+ "select SUBSTRING(s, 1, 5), SUBSTRING(s, 7), SUBSTRING(s, 20),
SUBSTRING(s, 1, 0)"
+ + " from tblSubstr";
+ runAndCheck(
+ query,
+ Arrays.asList("+I[hello, world, , ]", "+I[abcde, ghij, , ]", "+I[null,
null, null, null]"));
+ }
+
+ @Test
+ void testSubstringDynamicStart() {
+ // A non-literal start cannot be checked at plan time; with runtime values
+ // >= 1 the velox results match Flink.
+ List<Row> rows = Arrays.asList(Row.of("hello world", 7),
Row.of("abcdefghij", 2));
+ createSimpleBoundedValuesTable("tblSubstrDyn", "s varchar, p int", rows);
+ String query = "select SUBSTRING(s, p), SUBSTRING(s, p, 3) from
tblSubstrDyn";
+ runAndCheck(query, Arrays.asList("+I[world, wor]", "+I[bcdefghij, bcd]"));
+ }
+
+ @Test
+ void testSubstringNonPositiveStart() {
+ // Flink treats a start of 0 like 1 and counts a negative start from the
+ // string end (BinaryStringDataUtil#substringSQL) — the same rules velox
+ // substring implements — so these forms convert as-is and match the
+ // vanilla Flink results (verified on a local cluster). A start more
+ // negative than the string length diverges but depends on row data; see
+ // SubstringRexCallConverter.
+ List<Row> rows = Arrays.asList(Row.of("hello world"),
Row.of("abcdefghij"));
+ createSimpleBoundedValuesTable("tblSubstrStart", "s varchar", rows);
+ String query =
+ "select SUBSTRING(s, 0, 2), SUBSTRING(s, -3, 5), SUBSTRING(s, -3) from
tblSubstrStart";
+ runAndCheck(query, Arrays.asList("+I[he, rld, rld]", "+I[ab, hij, hij]"));
+ }
+
+ @Test
+ void testSubstringNegativeLengthRejected() {
+ // Flink returns NULL for a negative length while velox returns an empty
+ // string; a literal negative length fails fast instead of silently
+ // returning the velox result. Stopgap until a Flink-semantics substring
+ // in the velox flinksql function set returns NULL.
+ List<Row> rows = Arrays.asList(Row.of("hello world"));
+ createSimpleBoundedValuesTable("tblSubstrLen", "s varchar", rows);
+ tEnv().executeSql("CREATE TABLE printT (s VARCHAR) WITH ('connector' =
'print')");
+ assertThatThrownBy(
+ () ->
+ tEnv()
+ .executeSql("insert into printT select SUBSTRING(s, 1, -3)
from tblSubstrLen"))
+ .hasStackTraceContaining("SUBSTRING with literal length -3 is not
supported");
+ tEnv().executeSql("drop table if exists printT");
+ }
+
+ @Test
+ void testCoalesce() {
+ List<Row> rows = Arrays.asList(Row.of("a", "b"), Row.of(null, "b"),
Row.of("a", null));
+ createSimpleBoundedValuesTable("tblCoalesce", "s1 varchar, s2 varchar",
rows);
+ String query = "select COALESCE(s1, s2) from tblCoalesce";
+ runAndCheck(query, Arrays.asList("+I[a]", "+I[b]", "+I[a]"));
+
+ // Mixed numeric types (common type BIGINT) and an all-NULL row.
+ List<Row> numRows = Arrays.asList(Row.of(1, 10L), Row.of(null, 20L),
Row.of(null, null));
+ createSimpleBoundedValuesTable("tblCoalesceNum", "a int, b bigint",
numRows);
+ String numQuery = "select COALESCE(a, b) from tblCoalesceNum";
+ runAndCheck(numQuery, Arrays.asList("+I[1]", "+I[20]", "+I[null]"));
+ }
+
+ @Test
+ void testTimestampLiteral() {
+ DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd
HH:mm:ss");
+ List<Row> rows =
+ Arrays.asList(
+ Row.of(1, LocalDateTime.parse("2024-01-15 03:04:05", formatter)),
Row.of(2, null));
+ createSimpleBoundedValuesTable("tblTsLiteral", "a int, b Timestamp(3)
NULL", rows);
+ String query =
+ "select DATE_FORMAT(COALESCE(b, TIMESTAMP '2024-03-15 10:15:30'),
'yyyy-MM-dd HH:mm:ss') "
+ + "from tblTsLiteral";
+ runAndCheck(query, Arrays.asList("+I[2024-01-15 03:04:05]", "+I[2024-03-15
10:15:30]"));
+
+ // Project the literal alone: anchors the converted epoch value directly,
+ // without the COALESCE + DATE_FORMAT round trip (formatting sensitive).
+ String literalQuery = "select TIMESTAMP '2024-03-15 10:15:30' from
tblTsLiteral";
+ runAndCheck(
+ literalQuery, Arrays.asList("+I[2024-03-15T10:15:30.000]",
"+I[2024-03-15T10:15:30.000]"));
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
