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 dc64bba138 [GLUTEN-13369][FLINK] Support function IFNULL (#13375)
dc64bba138 is described below
commit dc64bba1388871691c12faba0c36eb6cacbcac96
Author: wud <[email protected]>
AuthorDate: Thu Oct 29 15:49:49 2026 +0800
[GLUTEN-13369][FLINK] Support function IFNULL (#13375)
* Support IFNULL offload to the velox flink sql function
---
.github/workflows/flink.yml | 2 +-
gluten-flink/docs/Flink.md | 2 +-
.../apache/gluten/rexnode/RexNodeConverter.java | 36 +++++++++++
.../java/org/apache/gluten/rexnode/TypeUtils.java | 14 +++++
.../rexnode/functions/RexCallConverterFactory.java | 1 +
.../functions/ScalarOnlyRexCallConverter.java | 63 ++++++++++++++++++++
.../gluten/rexnode/RexNodeConverterTest.java | 44 ++++++++++++++
.../runtime/stream/custom/IfNullFunctionsTest.java | 69 ++++++++++++++++++++++
8 files changed, 229 insertions(+), 2 deletions(-)
diff --git a/.github/workflows/flink.yml b/.github/workflows/flink.yml
index 37a6acbf88..38eedbd926 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 26c7715278e6f6795084f6334998cb2ce382f7aa
+ git reset --hard 2d08448f556bff669fe413874d04dd8cd96f65b2
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 aa163daeca..6530031f08 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 26c7715278e6f6795084f6334998cb2ce382f7aa
+git reset --hard 2d08448f556bff669fe413874d04dd8cd96f65b2
mvn clean install -DskipTests -Dgpg.skip -Dspotless.skip=true
```
**Get gluten**
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 297063cf1a..38228b104d 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
@@ -97,7 +97,43 @@ public class RexNodeConverter {
return
LogicalTypeConverter.toVLType(FlinkTypeFactory.toLogicalType(relDataType));
}
+ private static Variant toNullVariant(SqlTypeName sqlTypeName) {
+ // A typed null needs no per-type carrier: the velox-side
+ // ConstantTypedExpr renders a null variant as a null constant of the
+ // EXPRESSION type (the variant kind is irrelevant on the null path).
+ // The velox serde accepts both an explicit "value": null (what velox4j
+ // emits today via @JsonInclude(ALWAYS)) and an absent "value" member as
+ // the typed null. VarCharValue is the natural null-accepting carrier.
+ switch (sqlTypeName) {
+ case BOOLEAN:
+ case TINYINT:
+ case SMALLINT:
+ case INTEGER:
+ case BIGINT:
+ case DOUBLE:
+ case FLOAT:
+ case VARCHAR:
+ case CHAR:
+ case BINARY:
+ case DECIMAL:
+ case DATE:
+ case INTERVAL_SECOND:
+ case SYMBOL:
+ case TIMESTAMP:
+ return new VarCharValue(null);
+ default:
+ throw new RuntimeException("Unsupported null rex node type: " +
sqlTypeName);
+ }
+ }
+
public static Variant toVariant(RexLiteral literal) {
+ if (literal.getValue() == null || literal.isNull()) {
+ // A typed null literal. velox4j serializes null carriers with an
+ // explicit "value": null (@JsonInclude(ALWAYS) on the Variant base
+ // class); the velox-side serde accepts both an explicit null and an
+ // absent "value" member as the typed null.
+ return toNullVariant(literal.getType().getSqlTypeName());
+ }
switch (literal.getType().getSqlTypeName()) {
case BOOLEAN:
return new BooleanValue((boolean) literal.getValue());
diff --git
a/gluten-flink/planner/src/main/java/org/apache/gluten/rexnode/TypeUtils.java
b/gluten-flink/planner/src/main/java/org/apache/gluten/rexnode/TypeUtils.java
index c16d563d67..388a8f4084 100644
---
a/gluten-flink/planner/src/main/java/org/apache/gluten/rexnode/TypeUtils.java
+++
b/gluten-flink/planner/src/main/java/org/apache/gluten/rexnode/TypeUtils.java
@@ -53,6 +53,20 @@ public class TypeUtils {
return type instanceof VarCharType;
}
+ // Matches the scalar signatures registered under the velox flinksql
+ // namespace: booleans, numerics, strings, dates, timestamps and decimals.
+ // BINARY, ARRAY, MAP and ROW have no flinksql registration and are
+ // rejected at the converter level so conversion fails at planning time
+ // instead of failing at runtime.
+ public static boolean isScalarType(Type type) {
+ return type instanceof BooleanType
+ || isNumericType(type)
+ || isStringType(type)
+ || type instanceof DateType
+ || type instanceof TimestampType
+ || type instanceof DecimalType;
+ }
+
public static boolean isIntegerType(Type type) {
return type instanceof IntegerType;
}
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 4254e8ff82..c64b425f55 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
@@ -119,6 +119,7 @@ public class RexCallConverterFactory {
// 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("IFNULL", Arrays.asList(() -> new
ScalarOnlyRexCallConverter("ifnull"))),
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/ScalarOnlyRexCallConverter.java
b/gluten-flink/planner/src/main/java/org/apache/gluten/rexnode/functions/ScalarOnlyRexCallConverter.java
new file mode 100644
index 0000000000..4b496a6fde
--- /dev/null
+++
b/gluten-flink/planner/src/main/java/org/apache/gluten/rexnode/functions/ScalarOnlyRexCallConverter.java
@@ -0,0 +1,63 @@
+/*
+ * 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.TypeUtils;
+import org.apache.gluten.rexnode.ValidationResult;
+
+import io.github.zhztheplayer.velox4j.expression.CallTypedExpr;
+import io.github.zhztheplayer.velox4j.expression.TypedExpr;
+import io.github.zhztheplayer.velox4j.type.Type;
+
+import org.apache.calcite.rex.RexCall;
+
+import java.util.List;
+
+/**
+ * Converter for functions whose velox flinksql registrations cover scalar
types only. Operands
+ * (BINARY, ARRAY, MAP, ROW) are rejected here so conversion fails loudly at
planning time instead
+ * of failing at runtime with an unresolved velox signature.
+ */
+class ScalarOnlyRexCallConverter extends BaseRexCallConverter {
+
+ public ScalarOnlyRexCallConverter(String functionName) {
+ super(functionName);
+ }
+
+ @Override
+ public ValidationResult isSuitable(RexCall callNode, RexConversionContext
context) {
+ boolean allScalar =
+ callNode.getOperands().stream()
+ .allMatch(arg ->
TypeUtils.isScalarType(RexNodeConverter.toType(arg.getType())));
+ if (!allScalar) {
+ return ValidationResult.failure(
+ String.format(
+ "Scalar-only operation requires all operands to be scalar types,
but found: %s",
+ getFunctionProtoTypeName(callNode)));
+ }
+ return ValidationResult.success();
+ }
+
+ @Override
+ public TypedExpr toTypedExpr(RexCall callNode, RexConversionContext context)
{
+ List<TypedExpr> params = getParams(callNode, context);
+ Type resultType = getResultType(callNode);
+ return new CallTypedExpr(resultType, params, functionName);
+ }
+}
diff --git
a/gluten-flink/ut/src/test/java/org/apache/gluten/rexnode/RexNodeConverterTest.java
b/gluten-flink/ut/src/test/java/org/apache/gluten/rexnode/RexNodeConverterTest.java
index 0cae77d7b6..2acde24b38 100644
---
a/gluten-flink/ut/src/test/java/org/apache/gluten/rexnode/RexNodeConverterTest.java
+++
b/gluten-flink/ut/src/test/java/org/apache/gluten/rexnode/RexNodeConverterTest.java
@@ -23,8 +23,12 @@ import
org.apache.flink.table.planner.calcite.FlinkRexBuilder;
import org.apache.flink.table.planner.calcite.FlinkTypeFactory;
import org.apache.flink.table.planner.calcite.FlinkTypeSystem;
+import org.apache.calcite.avatica.util.TimeUnit;
+import org.apache.calcite.rel.type.RelDataType;
import org.apache.calcite.rex.RexBuilder;
import org.apache.calcite.rex.RexLiteral;
+import org.apache.calcite.sql.SqlIntervalQualifier;
+import org.apache.calcite.sql.parser.SqlParserPos;
import org.apache.calcite.sql.type.SqlTypeName;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
@@ -63,4 +67,44 @@ public class RexNodeConverterTest {
assertThat(variant).isInstanceOf(VarCharValue.class);
assertThat(((VarCharValue) variant).getValue()).isEqualTo("banana");
}
+
+ @Test
+ public void testNullLiteralConvertsToNullCarrier() {
+ RexLiteral intervalLiteral =
+ (RexLiteral)
+ rexBuilder.makeNullLiteral(
+ typeFactory.createSqlIntervalType(
+ new SqlIntervalQualifier(
+ TimeUnit.SECOND,
+ RelDataType.PRECISION_NOT_SPECIFIED,
+ TimeUnit.SECOND,
+ RelDataType.PRECISION_NOT_SPECIFIED,
+ SqlParserPos.ZERO)));
+ assertThat(((VarCharValue)
RexNodeConverter.toVariant(intervalLiteral)).getValue()).isNull();
+ SqlTypeName[] names = {
+ SqlTypeName.VARCHAR,
+ SqlTypeName.INTEGER,
+ SqlTypeName.BIGINT,
+ SqlTypeName.DOUBLE,
+ SqlTypeName.TIMESTAMP,
+ SqlTypeName.DATE,
+ SqlTypeName.SYMBOL,
+ };
+ for (SqlTypeName name : names) {
+ RexLiteral literal = (RexLiteral)
rexBuilder.makeNullLiteral(typeFactory.createSqlType(name));
+ Variant variant = RexNodeConverter.toVariant(literal);
+ assertThat(variant).isInstanceOf(VarCharValue.class);
+ assertThat(((VarCharValue) variant).getValue()).isNull();
+ }
+ }
+
+ @Test
+ public void testDecimalNullLiteralConvertsToNullCarrier() {
+ RexLiteral literal =
+ (RexLiteral)
+
rexBuilder.makeNullLiteral(typeFactory.createSqlType(SqlTypeName.DECIMAL, 10,
2));
+ Variant variant = RexNodeConverter.toVariant(literal);
+ assertThat(variant).isInstanceOf(VarCharValue.class);
+ assertThat(((VarCharValue) variant).getValue()).isNull();
+ }
}
diff --git
a/gluten-flink/ut/src/test/java/org/apache/gluten/table/runtime/stream/custom/IfNullFunctionsTest.java
b/gluten-flink/ut/src/test/java/org/apache/gluten/table/runtime/stream/custom/IfNullFunctionsTest.java
new file mode 100644
index 0000000000..375d44e8e0
--- /dev/null
+++
b/gluten-flink/ut/src/test/java/org/apache/gluten/table/runtime/stream/custom/IfNullFunctionsTest.java
@@ -0,0 +1,69 @@
+/*
+ * 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.math.BigDecimal;
+import java.util.Arrays;
+import java.util.List;
+
+class IfNullFunctionsTest extends GlutenStreamingTestBase {
+
+ @Override
+ @BeforeEach
+ public void before() throws Exception {
+ super.before();
+ }
+
+ @Test
+ void testIfNullWithNullableColumns() {
+ List<Row> rows =
+ Arrays.asList(
+ Row.of("a", 1L, 1.5d, new BigDecimal("1.50")),
+ Row.of(null, 7L, null, new BigDecimal("12.34")),
+ Row.of(null, null, null, null));
+ createSimpleBoundedValuesTable(
+ "tblIfNull", "s string, b bigint, d double, p decimal(10, 2)", rows);
+ String query =
+ "select IFNULL(s, 'N/A'), IFNULL(b, -1), IFNULL(d, 0.5), IFNULL(p,
9.99) from tblIfNull";
+ runAndCheck(
+ query,
+ Arrays.asList("+I[a, 1, 1.5, 1.50]", "+I[N/A, 7, 0.5, 12.34]",
"+I[N/A, -1, 0.5, 9.99]"));
+ }
+
+ @Test
+ void testIfNullWithNullLiteralFirstArg() {
+ List<Row> rows = Arrays.asList(Row.of("x"), Row.of("y"));
+ createSimpleBoundedValuesTable("tblIfNullLit", "s string", rows);
+ String query = "select IFNULL(CAST(NULL AS STRING), s) from tblIfNullLit";
+ runAndCheck(query, Arrays.asList("+I[x]", "+I[y]"));
+ }
+
+ @Test
+ void testIfNullBothNull() {
+ List<Row> rows = Arrays.asList(Row.of((Object) null), Row.of("y"));
+ createSimpleBoundedValuesTable("tblIfNullBoth", "s string", rows);
+ String query = "select IFNULL(s, CAST(NULL AS STRING)) from tblIfNullBoth";
+ runAndCheck(query, Arrays.asList("+I[null]", "+I[y]"));
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
