From 7fd526f057ea6bca8e48a239ef5106cb2215b4c7 Mon Sep 17 00:00:00 2001 From: TianJialong <1076317258@qq.com> Date: Mon, 26 Oct 2026 21:04:36 +0800 Subject: [PATCH 1/7] [FLINK] Support functions TO_DATE/TO_TIMESTAMP/TO_TIMESTAMP_LTZ --- .../gluten/rexnode/RexNodeConverter.java | 78 ++++- .../functions/RexCallConverterFactory.java | 12 + .../gluten/rexnode/RexNodeConverterTest.java | 270 ++++++++++++++++++ 3 files changed, 354 insertions(+), 6 deletions(-) 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 297063cf1ae..c2f71cf3132 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 @@ -43,11 +43,13 @@ import org.apache.flink.util.Preconditions; import org.apache.calcite.rel.type.RelDataType; +import org.apache.calcite.rel.type.RelDataTypeField; import org.apache.calcite.rex.RexCall; import org.apache.calcite.rex.RexFieldAccess; import org.apache.calcite.rex.RexInputRef; import org.apache.calcite.rex.RexLiteral; import org.apache.calcite.rex.RexNode; +import org.apache.calcite.sql.SqlKind; import org.apache.calcite.sql.type.SqlTypeName; import org.apache.calcite.util.NlsString; @@ -67,6 +69,14 @@ public static TypedExpr toTypedExpr(RexNode rexNode, RexConversionContext contex return new ConstantTypedExpr(toType(literal.getType()), toVariant(literal), null); } else if (rexNode instanceof RexCall) { RexCall rexCall = (RexCall) rexNode; + if (rexCall.getOperator().getKind() == SqlKind.CAST && rexCall.getOperands().size() == 1) { + RexNode operand = rexCall.getOperands().get(0); + // An identity cast (types equal ignoring only nullability, e.g. around folded + // literals) carries no semantics; velox rejects same-type casts, so pass through. + if (isSameTypeIgnoringNullability(rexCall.getType(), operand.getType())) { + return toTypedExpr(operand, context); + } + } RexCallConverter converter = RexCallConverterFactory.getConverter(rexCall, context); return converter.toTypedExpr(rexCall, context); } else if (rexNode instanceof RexInputRef) { @@ -97,6 +107,48 @@ public static Type toType(RelDataType relDataType) { return LogicalTypeConverter.toVLType(FlinkTypeFactory.toLogicalType(relDataType)); } + /** + * Structural equality ignoring nullability only: name, precision, scale, and every component type + * recursively (field names included). ANY is opaque and never qualifies. + */ + private static boolean isSameTypeIgnoringNullability(RelDataType a, RelDataType b) { + if (a == b) { + return true; + } + SqlTypeName kind = a.getSqlTypeName(); + if (kind != b.getSqlTypeName() + || a.getPrecision() != b.getPrecision() + || a.getScale() != b.getScale()) { + return false; + } + if (kind == SqlTypeName.ARRAY || kind == SqlTypeName.MULTISET) { + return isSameTypeIgnoringNullability(a.getComponentType(), b.getComponentType()); + } + if (kind == SqlTypeName.MAP) { + return isSameTypeIgnoringNullability(a.getKeyType(), b.getKeyType()) + && isSameTypeIgnoringNullability(a.getValueType(), b.getValueType()); + } + if (kind == SqlTypeName.ANY) { + return false; + } + if (a.isStruct()) { + if (a.getFieldCount() != b.getFieldCount()) { + return false; + } + for (int i = 0; i < a.getFieldCount(); i++) { + RelDataTypeField fieldA = a.getFieldList().get(i); + RelDataTypeField fieldB = b.getFieldList().get(i); + if (!fieldA.getName().equals(fieldB.getName()) + || !isSameTypeIgnoringNullability(fieldA.getType(), fieldB.getType())) { + return false; + } + } + return true; + } + // Scalars: name + precision + scale carry the identity (charset/collation are factory-wide). + return true; + } + public static Variant toVariant(RexLiteral literal) { switch (literal.getType().getSqlTypeName()) { case BOOLEAN: @@ -131,13 +183,27 @@ public static Variant toVariant(RexLiteral literal) { } case SYMBOL: return new VarCharValue(literal.getValue().toString()); + case DATE: + // Calcite keeps a DATE literal as int epoch days — the velox DATE representation; NULL + // maps to a null variant (folded NULLs of any type still fail velox4j serialization). + Integer dateValue = literal.getValueAs(Integer.class); + return dateValue == null ? new VarCharValue(null) : new IntegerValue(dateValue); 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); + case TIMESTAMP_WITH_LOCAL_TIME_ZONE: + // Calcite 1.32 resolves both names through the same branch + // (TimestampString.getMillisSinceEpoch()); Flink's folding round-trip + // stores an LTZ literal as a GMT-frame TimestampString and reads it + // back atOffset(UTC) — the same epoch-millis base as 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. A NULL literal maps to a null variant (planner no + // longer NPEs; velox4j serialization of inner nulls is a known gap). + Long epochMillis = literal.getValueAs(Long.class); + return epochMillis == null + ? new VarCharValue(null) + : 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 4254e8ff82e..dec2be23dc8 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 @@ -107,6 +107,18 @@ public class RexCallConverterFactory { Map.entry("IS NULL", Arrays.asList(() -> new DefaultRexCallConverter("isnull"))), Map.entry( "REGEXP_EXTRACT", Arrays.asList(() -> new DefaultRexCallConverter("regexp_extract"))), + // Direct name mapping: both arities are covered by the flinksql to_date signatures. + Map.entry("TO_DATE", Arrays.asList(() -> new DefaultRexCallConverter("to_date"))), + // Direct name mapping: both arities are covered by the flinksql to_timestamp + // signatures; the returned TIMESTAMP(3) maps to the velox TimestampType. + Map.entry( + "TO_TIMESTAMP", Arrays.asList(() -> new DefaultRexCallConverter("to_timestamp"))), + // Direct name mapping: the only SQL form is (numeric, precision) — the first + // operand's type is routed by the velox signature set, the returned + // TIMESTAMP_LTZ(3) maps to the velox TimestampType. + Map.entry( + "TO_TIMESTAMP_LTZ", + Arrays.asList(() -> new DefaultRexCallConverter("to_timestamp_ltz"))), Map.entry("LOWER", Arrays.asList(() -> new DefaultRexCallConverter("lower"))), Map.entry("count_char", Arrays.asList(() -> new DefaultRexCallConverter("count_char"))), Map.entry("EXTRACT", Arrays.asList(() -> new DefaultRexCallConverter("extract"))), 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 0cae77d7b68..f4d414a0f6d 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 @@ -16,16 +16,31 @@ */ package org.apache.gluten.rexnode; +import org.apache.gluten.rexnode.functions.RexCallConverterFactory; + +import io.github.zhztheplayer.velox4j.expression.CallTypedExpr; +import io.github.zhztheplayer.velox4j.expression.ConstantTypedExpr; +import io.github.zhztheplayer.velox4j.expression.FieldAccessTypedExpr; +import io.github.zhztheplayer.velox4j.expression.TypedExpr; +import io.github.zhztheplayer.velox4j.type.DateType; +import io.github.zhztheplayer.velox4j.type.TimestampType; +import io.github.zhztheplayer.velox4j.variant.IntegerValue; +import io.github.zhztheplayer.velox4j.variant.TimestampValue; import io.github.zhztheplayer.velox4j.variant.VarCharValue; import io.github.zhztheplayer.velox4j.variant.Variant; 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.flink.table.planner.functions.sql.FlinkSqlOperatorTable; +import org.apache.calcite.rel.type.RelDataType; import org.apache.calcite.rex.RexBuilder; +import org.apache.calcite.rex.RexCall; +import org.apache.calcite.rex.RexInputRef; import org.apache.calcite.rex.RexLiteral; import org.apache.calcite.sql.type.SqlTypeName; +import org.apache.calcite.util.TimestampString; import org.junit.jupiter.api.BeforeAll; import org.junit.jupiter.api.Test; @@ -63,4 +78,259 @@ public void testCharLiteralExtractsRawValue() { assertThat(variant).isInstanceOf(VarCharValue.class); assertThat(((VarCharValue) variant).getValue()).isEqualTo("banana"); } + + @Test + public void testDateLiteralExtractsRawValue() { + // Calcite keeps a DATE literal as int epoch days — the velox DATE representation. + RexLiteral literal = + (RexLiteral) + rexBuilder.makeLiteral(20707, typeFactory.createSqlType(SqlTypeName.DATE), false); + Variant variant = RexNodeConverter.toVariant(literal); + assertThat(variant).isInstanceOf(IntegerValue.class); + assertThat(((IntegerValue) variant).getValue()).isEqualTo(20707); + } + + @Test + public void testNullDateLiteralConvertsToNullVariant() { + // TO_DATE(NULL) folds to a NULL DATE literal; it must produce a null variant. + RexLiteral literal = rexBuilder.makeNullLiteral(typeFactory.createSqlType(SqlTypeName.DATE)); + Variant variant = RexNodeConverter.toVariant(literal); + assertThat(variant).isInstanceOf(VarCharValue.class); + assertThat(((VarCharValue) variant).getValue()).isNull(); + } + + @Test + public void testIdentityCastAroundDateLiteralIsRemoved() { + // The planner wraps folded constants in identity casts; velox rejects same-type casts. + RexLiteral literal = + (RexLiteral) + rexBuilder.makeLiteral(20707, typeFactory.createSqlType(SqlTypeName.DATE), false); + RexCall cast = + (RexCall) + rexBuilder.makeAbstractCast( + typeFactory.createTypeWithNullability( + typeFactory.createSqlType(SqlTypeName.DATE), true), + literal); + TypedExpr expr = + RexNodeConverter.toTypedExpr( + cast, new RexConversionContext(java.util.Collections.emptyList())); + assertThat(expr).isInstanceOf(ConstantTypedExpr.class); + ConstantTypedExpr constant = (ConstantTypedExpr) expr; + assertThat(constant.getReturnType()).isInstanceOf(DateType.class); + assertThat(constant.getValue()).isInstanceOf(IntegerValue.class); + assertThat(((IntegerValue) constant.getValue()).getValue()).isEqualTo(20707); + } + + @Test + public void testToDateCallMapsToVeloxName() { + RexInputRef stringFieldRef = + rexBuilder.makeInputRef(typeFactory.createSqlType(SqlTypeName.VARCHAR), 1); + RexInputRef formatFieldRef = + rexBuilder.makeInputRef(typeFactory.createSqlType(SqlTypeName.VARCHAR), 2); + RexConversionContext context = + new RexConversionContext(java.util.Arrays.asList("ignored", "s", "f")); + + RexCall oneArg = (RexCall) rexBuilder.makeCall(FlinkSqlOperatorTable.TO_DATE, stringFieldRef); + CallTypedExpr oneArgExpr = + (CallTypedExpr) + RexCallConverterFactory.getConverter(oneArg, context).toTypedExpr(oneArg, context); + assertThat(oneArgExpr.getFunctionName()).isEqualTo("to_date"); + assertThat(oneArgExpr.getReturnType()).isInstanceOf(DateType.class); + assertThat(oneArgExpr.getInputs()).hasSize(1); + + RexCall twoArgs = + (RexCall) + rexBuilder.makeCall(FlinkSqlOperatorTable.TO_DATE, stringFieldRef, formatFieldRef); + CallTypedExpr twoArgExpr = + (CallTypedExpr) + RexCallConverterFactory.getConverter(twoArgs, context).toTypedExpr(twoArgs, context); + assertThat(twoArgExpr.getFunctionName()).isEqualTo("to_date"); + assertThat(twoArgExpr.getReturnType()).isInstanceOf(DateType.class); + assertThat(twoArgExpr.getInputs()).hasSize(2); + } + + @Test + public void testIdentityCastOnCompositeTypeIsRemoved() { + // Nullability-only casts on composites are identity too (all component types match). + RelDataType intType = typeFactory.createSqlType(SqlTypeName.INTEGER); + RexConversionContext context = + new RexConversionContext(java.util.Collections.singletonList("f0")); + + assertIdentityCastRemoved(typeFactory.createArrayType(intType, -1), context); + + assertIdentityCastRemoved( + typeFactory.createStructType( + java.util.Collections.singletonList(intType), java.util.Collections.singletonList("f")), + context); + + assertIdentityCastRemoved(typeFactory.createMapType(intType, intType), context); + } + + @Test + public void testCompositeCastWithDifferentComponentTypeIsKept() { + // Component type differences are not identity; the cast must stay a cast. + RelDataType intType = typeFactory.createSqlType(SqlTypeName.INTEGER); + RelDataType bigType = typeFactory.createSqlType(SqlTypeName.BIGINT); + RexConversionContext context = + new RexConversionContext(java.util.Collections.singletonList("f0")); + + assertCastKept( + typeFactory.createArrayType(intType, -1), + typeFactory.createArrayType(bigType, -1), + context); + + assertCastKept( + typeFactory.createStructType( + java.util.Collections.singletonList(intType), java.util.Collections.singletonList("f")), + typeFactory.createStructType( + java.util.Collections.singletonList(bigType), java.util.Collections.singletonList("f")), + context); + + assertCastKept( + typeFactory.createMapType(intType, intType), + typeFactory.createMapType(bigType, intType), + context); + + assertCastKept( + typeFactory.createSqlType(SqlTypeName.VARCHAR, 5), + typeFactory.createSqlType(SqlTypeName.VARCHAR, 10), + context); + + assertCastKept(intType, bigType, context); + } + + @Test + public void testTimestampLiteralExtractsRawValue() { + // The folded TO_TIMESTAMP literal keeps the wall time as UTC epoch millis. + RexLiteral literal = + (RexLiteral) rexBuilder.makeTimestampLiteral(new TimestampString("2026-09-11 10:00:00"), 3); + Variant variant = RexNodeConverter.toVariant(literal); + assertThat(variant).isInstanceOf(TimestampValue.class); + assertThat(((TimestampValue) variant).getSeconds()).isEqualTo(1789120800L); + assertThat(((TimestampValue) variant).getNanos()).isEqualTo(0L); + } + + @Test + public void testTimestampWithLocalTimeZoneLiteralExtractsRawValue() { + // The folded TO_TIMESTAMP_LTZ literal shares the TIMESTAMP conversion: + // the epoch-millis base (Calcite resolves both names through + // TimestampString#getMillisSinceEpoch; the Flink round-trip is atOffset(UTC)). + RexLiteral literal = + (RexLiteral) + rexBuilder.makeTimestampWithLocalTimeZoneLiteral( + new TimestampString("2026-09-11 10:00:00"), 3); + Variant variant = RexNodeConverter.toVariant(literal); + assertThat(variant).isInstanceOf(TimestampValue.class); + assertThat(((TimestampValue) variant).getSeconds()).isEqualTo(1789120800L); + assertThat(((TimestampValue) variant).getNanos()).isEqualTo(0L); + } + + @Test + public void testNullLtzLiteralConvertsToNullVariant() { + // TO_TIMESTAMP_LTZ never folds to NULL (the TableException aborts + // folding), but a NULL LTZ literal from any path must not NPE the planner. + RexLiteral literal = + rexBuilder.makeNullLiteral( + typeFactory.createSqlType(SqlTypeName.TIMESTAMP_WITH_LOCAL_TIME_ZONE, 3)); + Variant variant = RexNodeConverter.toVariant(literal); + assertThat(variant).isInstanceOf(VarCharValue.class); + assertThat(((VarCharValue) variant).getValue()).isNull(); + } + + @Test + public void testToTimestampCallMapsToVeloxName() { + RexInputRef stringFieldRef = + rexBuilder.makeInputRef(typeFactory.createSqlType(SqlTypeName.VARCHAR), 1); + RexInputRef formatFieldRef = + rexBuilder.makeInputRef(typeFactory.createSqlType(SqlTypeName.VARCHAR), 2); + RexConversionContext context = + new RexConversionContext(java.util.Arrays.asList("ignored", "s", "f")); + + RexCall oneArg = + (RexCall) rexBuilder.makeCall(FlinkSqlOperatorTable.TO_TIMESTAMP, stringFieldRef); + CallTypedExpr oneArgExpr = + (CallTypedExpr) + RexCallConverterFactory.getConverter(oneArg, context).toTypedExpr(oneArg, context); + assertThat(oneArgExpr.getFunctionName()).isEqualTo("to_timestamp"); + assertThat(oneArgExpr.getReturnType()).isInstanceOf(TimestampType.class); + assertThat(oneArgExpr.getInputs()).hasSize(1); + + RexCall twoArgs = + (RexCall) + rexBuilder.makeCall(FlinkSqlOperatorTable.TO_TIMESTAMP, stringFieldRef, formatFieldRef); + CallTypedExpr twoArgExpr = + (CallTypedExpr) + RexCallConverterFactory.getConverter(twoArgs, context).toTypedExpr(twoArgs, context); + assertThat(twoArgExpr.getFunctionName()).isEqualTo("to_timestamp"); + assertThat(twoArgExpr.getReturnType()).isInstanceOf(TimestampType.class); + assertThat(twoArgExpr.getInputs()).hasSize(2); + } + + @Test + public void testToTimestampLtzCallMapsToVeloxName() { + RexInputRef bigIntFieldRef = + rexBuilder.makeInputRef(typeFactory.createSqlType(SqlTypeName.BIGINT), 1); + RexLiteral precisionLiteral = + (RexLiteral) + rexBuilder.makeLiteral(3, typeFactory.createSqlType(SqlTypeName.INTEGER), false); + RexConversionContext context = + new RexConversionContext(java.util.Arrays.asList("ignored", "x", "p")); + + RexCall call = + (RexCall) + rexBuilder.makeCall( + FlinkSqlOperatorTable.TO_TIMESTAMP_LTZ, bigIntFieldRef, precisionLiteral); + CallTypedExpr expr = + (CallTypedExpr) + RexCallConverterFactory.getConverter(call, context).toTypedExpr(call, context); + assertThat(expr.getFunctionName()).isEqualTo("to_timestamp_ltz"); + assertThat(expr.getReturnType()).isInstanceOf(TimestampType.class); + assertThat(expr.getInputs()).hasSize(2); + } + + @Test + public void testIdentityCastAroundLtzLiteralIsRemoved() { + // The planner wraps folded constants in identity casts (e.g. when the sink + // declares TIMESTAMP_LTZ(3)); nullability-only casts are identity and pass + // through, keeping the folding path reachable. + RexLiteral literal = + (RexLiteral) + rexBuilder.makeTimestampWithLocalTimeZoneLiteral( + new TimestampString("2026-09-11 10:00:00"), 3); + RexCall cast = + (RexCall) + rexBuilder.makeAbstractCast( + typeFactory.createTypeWithNullability( + typeFactory.createSqlType(SqlTypeName.TIMESTAMP_WITH_LOCAL_TIME_ZONE, 3), true), + literal); + TypedExpr expr = + RexNodeConverter.toTypedExpr( + cast, new RexConversionContext(java.util.Collections.emptyList())); + assertThat(expr).isInstanceOf(ConstantTypedExpr.class); + ConstantTypedExpr constant = (ConstantTypedExpr) expr; + assertThat(constant.getReturnType()).isInstanceOf(TimestampType.class); + assertThat(constant.getValue()).isInstanceOf(TimestampValue.class); + assertThat(((TimestampValue) constant.getValue()).getSeconds()).isEqualTo(1789120800L); + assertThat(((TimestampValue) constant.getValue()).getNanos()).isEqualTo(0L); + } + + private void assertIdentityCastRemoved(RelDataType fromType, RexConversionContext context) { + RexInputRef ref = rexBuilder.makeInputRef(fromType, 0); + RexCall cast = + (RexCall) + rexBuilder.makeAbstractCast(typeFactory.createTypeWithNullability(fromType, true), ref); + TypedExpr expr = RexNodeConverter.toTypedExpr(cast, context); + assertThat(expr).isInstanceOf(FieldAccessTypedExpr.class); + assertThat(((FieldAccessTypedExpr) expr).getFieldName()).isEqualTo("f0"); + } + + private void assertCastKept( + RelDataType fromType, RelDataType toType, RexConversionContext context) { + RexInputRef ref = rexBuilder.makeInputRef(fromType, 0); + RexCall cast = (RexCall) rexBuilder.makeAbstractCast(toType, ref); + TypedExpr expr = RexNodeConverter.toTypedExpr(cast, context); + assertThat(expr).isInstanceOf(CallTypedExpr.class); + assertThat(((CallTypedExpr) expr).getFunctionName()).isEqualTo("cast"); + assertThat(((CallTypedExpr) expr).getInputs()).hasSize(1); + } } From 157cb39589b485cb04c6668862b2f4e8d49b3d96 Mon Sep 17 00:00:00 2001 From: TianJialong <1076317258@qq.com> Date: Thu, 29 Oct 2026 10:57:10 +0800 Subject: [PATCH 2/7] [FLINK] functions TO_DATE/TO_TIMESTAMP/TO_TIMESTAMP_LTZ add e2e tests --- .github/workflows/flink.yml | 4 +- gluten-flink/docs/Flink.md | 4 +- .../stream/custom/DateTimeFunctionsTest.java | 195 ++++++++++++++++++ 3 files changed, 199 insertions(+), 4 deletions(-) create mode 100644 gluten-flink/ut/src/test/java/org/apache/gluten/table/runtime/stream/custom/DateTimeFunctionsTest.java diff --git a/.github/workflows/flink.yml b/.github/workflows/flink.yml index 37a6acbf889..2c07bf053ee 100644 --- a/.github/workflows/flink.yml +++ b/.github/workflows/flink.yml @@ -87,9 +87,9 @@ jobs: export VELOX_DEPENDENCY_SOURCE=BUNDLED export fmt_SOURCE=BUNDLED export folly_SOURCE=BUNDLED - git clone -b gluten-20260829 https://github.com/bigo-sg/velox4j.git + git clone -b gluten-20260829 https://github.com/TianJialong/velox4j.git cd velox4j - git reset --hard 26c7715278e6f6795084f6334998cb2ce382f7aa + # use latest commit 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 aa163daecac..098bb344a95 100644 --- a/gluten-flink/docs/Flink.md +++ b/gluten-flink/docs/Flink.md @@ -46,9 +46,9 @@ As some features have not been committed to upstream, you have to use the follow ```bash ## fetch velox4j code -git clone -b gluten-20260829 https://github.com/bigo-sg/velox4j.git +git clone -b gluten-20260829 https://github.com/TianJialong/velox4j.git cd velox4j -git reset --hard 26c7715278e6f6795084f6334998cb2ce382f7aa +# use latest commit mvn clean install -DskipTests -Dgpg.skip -Dspotless.skip=true ``` **Get gluten** diff --git a/gluten-flink/ut/src/test/java/org/apache/gluten/table/runtime/stream/custom/DateTimeFunctionsTest.java b/gluten-flink/ut/src/test/java/org/apache/gluten/table/runtime/stream/custom/DateTimeFunctionsTest.java new file mode 100644 index 00000000000..384c6fcfaf1 --- /dev/null +++ b/gluten-flink/ut/src/test/java/org/apache/gluten/table/runtime/stream/custom/DateTimeFunctionsTest.java @@ -0,0 +1,195 @@ +/* + * 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 DateTimeFunctionsTest extends GlutenStreamingTestBase { + + @Override + @BeforeEach + public void before() throws Exception { + super.before(); + } + + @Test + void testToDate() { + List rows = + Arrays.asList( + Row.of("2026-09-11"), + Row.of("2026-9-1"), + Row.of("2026-09-11 10:00:00"), + Row.of("9999-12-31"), + Row.of("2026-13-01"), + Row.of("abc"), + Row.of((Object) null)); + createSimpleBoundedValuesTable("tblToDate", "s string", rows); + runAndCheck( + "select to_date(s) as d from tblToDate", + Arrays.asList( + "+I[20707]", + "+I[20697]", + "+I[20707]", + "+I[2932896]", + "+I[null]", + "+I[null]", + "+I[null]")); + } + + @Test + void testToDateWithColumnFormat() { + List rows = + Arrays.asList( + Row.of("2026/09/11", "yyyy/MM/dd"), + Row.of("20260911", "yyyyMMdd"), + Row.of("2026-09-11", "yyyy/MM/dd"), + Row.of("26-09-11", "yy-MM-dd"), + Row.of("2026-09-11", null), + Row.of(null, "yyyy-MM-dd")); + createSimpleBoundedValuesTable("tblToDateFmt", "s string, f string", rows); + runAndCheck( + "select to_date(s, f) as d from tblToDateFmt", + Arrays.asList("+I[20707]", "+I[20707]", "+I[null]", "+I[20707]", "+I[null]", "+I[null]")); + } + + @Test + void testToDateWithConstantFormat() { + List rows = Arrays.asList(Row.of("2026/09/11"), Row.of("2026/9/1"), Row.of("garbage")); + createSimpleBoundedValuesTable("tblToDateConst", "s string", rows); + runAndCheck( + "select to_date(s, 'yyyy/MM/dd') as d from tblToDateConst", + Arrays.asList("+I[20707]", "+I[20697]", "+I[null]")); + // Constant folding: the planner folds this into a DATE literal, which exercises + // the DATE literal channel end to end. + runAndCheck( + "select to_date('2026/09/11', 'yyyy/MM/dd') as d from tblToDateConst", + Arrays.asList("+I[20707]", "+I[20707]", "+I[20707]")); + } + + @Test + void testToTimestamp() { + List rows = + Arrays.asList( + Row.of("2026-09-11 10:00:00"), + Row.of("2026-02-30 10:00:00"), + Row.of("2026-09-11 10:00:00.123456789"), + Row.of("2026-09-11 24:00:00"), + Row.of("2026-13-01 10:00:00"), + Row.of((Object) null)); + createSimpleBoundedValuesTable("tblToTimestamp", "s string", rows); + runAndCheck( + "select to_timestamp(s) as ts from tblToTimestamp", + Arrays.asList( + "+I[2026-09-11T10:00:00.000]", + "+I[2026-02-28T10:00:00.000]", + "+I[2026-09-11T10:00:00.123]", + "+I[2026-09-12T00:00:00.000]", + "+I[null]", + "+I[null]")); + } + + @Test + void testToTimestampWithFormat() { + List rows = + Arrays.asList( + Row.of("2026/09/11 10:00:00", "yyyy/MM/dd HH:mm:ss"), + Row.of("2026-09-11 10:00:00", "yyyy-MM-dd HH:mm:ss"), + Row.of("2026-09-11 10:00:00", "QQ"), + Row.of(null, "yyyy-MM-dd HH:mm:ss"), + Row.of("2026-09-11 10:00:00", null)); + createSimpleBoundedValuesTable("tblToTimestampFmt", "s string, f string", rows); + runAndCheck( + "select to_timestamp(s, f) as ts from tblToTimestampFmt", + Arrays.asList( + "+I[2026-09-11T10:00:00.000]", + "+I[2026-09-11T10:00:00.000]", + "+I[null]", + "+I[null]", + "+I[null]")); + } + + @Test + void testToTimestampLtz() { + // Precision changes the unit of the same number: seconds vs milliseconds. + List rows = + Arrays.asList( + Row.of(1789120800L, 0), + Row.of(1789120800L, 3), + Row.of(1789120800000L, 3), + Row.of(253402300800000L, 3), + Row.of((Object) null, (Object) null)); + createSimpleBoundedValuesTable("tblToTimestampLtz", "b bigint, p int", rows); + runAndCheck( + "select to_timestamp_ltz(b, p) as ts from tblToTimestampLtz", + Arrays.asList( + "+I[2026-09-11T10:00:00.000]", + "+I[1970-01-21T16:58:40.800]", + "+I[2026-09-11T10:00:00.000]", + "+I[null]", + "+I[null]")); + } + + @Test + void testToTimestampLtzDoubleAndDecimal() { + List secRows = + Arrays.asList( + Row.of(1789120800.9d, new BigDecimal("1789120800.90")), + Row.of(253402300800d, new BigDecimal("253402300800.00"))); + createSimpleBoundedValuesTable("tblLtzSec", "d double, p decimal(20,2)", secRows); + // Seconds: the double keeps its fractional part through *1000, the decimal + // truncates toward zero; the second row exceeds the epoch domain on both. + runAndCheck( + "select to_timestamp_ltz(d, 0) as ts1, to_timestamp_ltz(p, 0) as ts2 from tblLtzSec", + Arrays.asList("+I[2026-09-11T10:00:00.900, 2026-09-11T10:00:00.000]", "+I[null, null]")); + + List milliRows = + Arrays.asList( + Row.of(1789120800900.9d, new BigDecimal("1789120800900.90")), + Row.of(253402300800000.9d, new BigDecimal("253402300800000.90"))); + createSimpleBoundedValuesTable("tblLtzMilli", "d double, p decimal(20,2)", milliRows); + runAndCheck( + "select to_timestamp_ltz(d, 3) as ts1, to_timestamp_ltz(p, 3) as ts2 from tblLtzMilli", + Arrays.asList("+I[2026-09-11T10:00:00.900, 2026-09-11T10:00:00.900]", "+I[null, null]")); + } + + @Test + void testTemporalFunctionsInCaseAndFilter() { + List rows = + Arrays.asList( + Row.of("2026-09-11"), + Row.of("2026-09-12"), + Row.of("2026-13-01"), + Row.of((Object) null)); + createSimpleBoundedValuesTable("tblTemporalCombo", "s string", rows); + runAndCheck( + "select case when cast(to_date(s) as string) = '2026-09-11' then 'hit' else 'miss' end" + + " as r from tblTemporalCombo", + Arrays.asList("+I[hit]", "+I[miss]", "+I[miss]", "+I[miss]")); + runAndCheck( + "select s from tblTemporalCombo where to_date(s) is not null", + Arrays.asList("+I[2026-09-11]", "+I[2026-09-12]")); + } +} From 74bc67a45e7166199a7688c830b2be613d85d310 Mon Sep 17 00:00:00 2001 From: TianJialong <1076317258@qq.com> Date: Thu, 29 Oct 2026 19:52:44 +0800 Subject: [PATCH 3/7] ignore .out files(such as ut/log/taskmanager.out) --- .gitignore | 1 + 1 file changed, 1 insertion(+) diff --git a/.gitignore b/.gitignore index c6aeadc0b18..73dec73462f 100644 --- a/.gitignore +++ b/.gitignore @@ -3,6 +3,7 @@ *.iml .DS_Store *.log +*.out **/core$ hs_err_pid* *.cache From 6e50e9758e438b92f15783268929b6a9ee26e5d4 Mon Sep 17 00:00:00 2001 From: TianJialong <1076317258@qq.com> Date: Mon, 9 Nov 2026 09:13:47 +0800 Subject: [PATCH 4/7] fix branch merge bug --- .../main/java/org/apache/gluten/rexnode/RexNodeConverter.java | 4 ++++ 1 file changed, 4 insertions(+) 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 7eb202ac47c..8c962a3fb7f 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 @@ -242,6 +242,10 @@ public static Variant toVariant(RexLiteral literal) { ? new VarCharValue(null) : TimestampValue.create( Math.floorDiv(epochMillis, 1000L), Math.floorMod(epochMillis, 1000L) * 1_000_000L); + case TIME: + // Calcite stores TIME literals as the millisecond-of-day Integer — + // the same encoding the velox side uses for TIME. + return new IntegerValue(literal.getValueAs(Integer.class)); default: throw new RuntimeException( "Unsupported rex node type: " + literal.getType().getSqlTypeName()); From 4252e34b4fa1ec03c692b1c7fddb43208ec7d84f Mon Sep 17 00:00:00 2001 From: TianJialong <1076317258@qq.com> Date: Wed, 11 Nov 2026 10:25:46 +0800 Subject: [PATCH 5/7] Add default missing years to 1970 testcases --- .../stream/custom/DateTimeFunctionsTest.java | 20 +++++++++++++++++++ 1 file changed, 20 insertions(+) diff --git a/gluten-flink/ut/src/test/java/org/apache/gluten/table/runtime/stream/custom/DateTimeFunctionsTest.java b/gluten-flink/ut/src/test/java/org/apache/gluten/table/runtime/stream/custom/DateTimeFunctionsTest.java index 384c6fcfaf1..b3574e9587d 100644 --- a/gluten-flink/ut/src/test/java/org/apache/gluten/table/runtime/stream/custom/DateTimeFunctionsTest.java +++ b/gluten-flink/ut/src/test/java/org/apache/gluten/table/runtime/stream/custom/DateTimeFunctionsTest.java @@ -175,6 +175,26 @@ void testToTimestampLtzDoubleAndDecimal() { Arrays.asList("+I[2026-09-11T10:00:00.900, 2026-09-11T10:00:00.900]", "+I[null, null]")); } + @Test + void testDefaultedYear() { + // Flink defaults a missing year to 1970: to_date rolls Feb 29 to Mar 1 + // (SimpleDateFormat lenient), to_timestamp yields NULL for Feb 29 + // (java.time strict recomposition). + List rows = Arrays.asList(Row.of("09-11"), Row.of("02-29")); + createSimpleBoundedValuesTable("tblDefaultedYear", "s string", rows); + runAndCheck( + "select to_date(s, 'MM-dd') as d from tblDefaultedYear", + Arrays.asList("+I[253]", "+I[59]")); + runAndCheck( + "select to_timestamp(s, 'MM-dd') as ts from tblDefaultedYear", + Arrays.asList("+I[1970-09-11T00:00:00.000]", "+I[null]")); + List timeRows = Arrays.asList(Row.of("10:00:00")); + createSimpleBoundedValuesTable("tblDefaultedTime", "s string", timeRows); + runAndCheck( + "select to_timestamp(s, 'HH:mm:ss') as ts from tblDefaultedTime", + Arrays.asList("+I[1970-01-01T10:00:00.000]")); + } + @Test void testTemporalFunctionsInCaseAndFilter() { List rows = From 1456c9ad8f9f4a8096e7a56296a060a9d0098986 Mon Sep 17 00:00:00 2001 From: TianJialong <1076317258@qq.com> Date: Wed, 11 Nov 2026 12:51:10 +0800 Subject: [PATCH 6/7] Update velox4j reference --- .github/workflows/flink.yml | 4 ++-- gluten-flink/docs/Flink.md | 4 ++-- gluten-flink/velox4j-ref.txt | 2 +- 3 files changed, 5 insertions(+), 5 deletions(-) diff --git a/.github/workflows/flink.yml b/.github/workflows/flink.yml index 67868d349b3..87df3910a0d 100644 --- a/.github/workflows/flink.yml +++ b/.github/workflows/flink.yml @@ -89,9 +89,9 @@ jobs: export VELOX_DEPENDENCY_SOURCE=BUNDLED export fmt_SOURCE=BUNDLED export folly_SOURCE=BUNDLED - git clone -b gluten-20260829 https://github.com/TianJialong/velox4j.git + git clone -b gluten-20260829 https://github.com/bigo-sg/velox4j.git cd velox4j - # use latest commit + git reset --hard $(cat $GITHUB_WORKSPACE/gluten-flink/velox4j-ref.txt) 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 2b13662e189..e11e6e41a16 100644 --- a/gluten-flink/docs/Flink.md +++ b/gluten-flink/docs/Flink.md @@ -50,9 +50,9 @@ As some features have not been committed to upstream, you have to use the follow ```bash ## fetch velox4j code -git clone -b gluten-20260829 https://github.com/TianJialong/velox4j.git +git clone -b gluten-20260829 https://github.com/bigo-sg/velox4j.git cd velox4j -# use latest commit +git reset --hard $(cat $GLUTEN_FLINK_HOME/velox4j-ref.txt) mvn clean install -DskipTests -Dgpg.skip -Dspotless.skip=true ``` **Get gluten** diff --git a/gluten-flink/velox4j-ref.txt b/gluten-flink/velox4j-ref.txt index e443ac0decb..75c1b021fe9 100644 --- a/gluten-flink/velox4j-ref.txt +++ b/gluten-flink/velox4j-ref.txt @@ -1 +1 @@ -cfe63ea5f7838a0221a268ce58d6d223c594bddc +301ddb5849e6a6fb95d6b4f016ddda804c0a1876 From 957fa900014198fa1da39d72bd1261abc391ab45 Mon Sep 17 00:00:00 2001 From: TianJialong <1076317258@qq.com> Date: Wed, 11 Nov 2026 14:46:02 +0800 Subject: [PATCH 7/7] Revert gitignore --- .gitignore | 1 - 1 file changed, 1 deletion(-) diff --git a/.gitignore b/.gitignore index 73dec73462f..c6aeadc0b18 100644 --- a/.gitignore +++ b/.gitignore @@ -3,7 +3,6 @@ *.iml .DS_Store *.log -*.out **/core$ hs_err_pid* *.cache