This is an automated email from the ASF dual-hosted git repository.
philo-he 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 21ee05ee4c [GLUTEN-13454][FLINK] Add support for CURRENT_DATE and
CURRENT_TIME (#13455)
21ee05ee4c is described below
commit 21ee05ee4ca53b58c5292e1dc80f6446c8eabc7a
Author: wud <[email protected]>
AuthorDate: Fri Nov 6 02:58:44 2026 +0800
[GLUTEN-13454][FLINK] Add support for CURRENT_DATE and CURRENT_TIME (#13455)
---
.github/workflows/flink.yml | 2 +-
gluten-flink/docs/Flink.md | 6 +-
.../apache/gluten/rexnode/RexNodeConverter.java | 5 ++
.../rexnode/functions/RexCallConverterFactory.java | 6 ++
.../table/runtime/config/VeloxQueryConfig.java | 11 ++--
.../apache/gluten/util/LogicalTypeConverter.java | 5 ++
.../gluten/rexnode/RexNodeConverterTest.java | 20 ++++++
.../stream/common/GlutenStreamingTestBase.java | 11 +++-
.../stream/custom/CurrentDateFunctionsTest.java | 50 +++++++++++++++
.../stream/custom/CurrentTimeFunctionsTest.java | 74 ++++++++++++++++++++++
gluten-flink/velox4j-ref.txt | 1 +
11 files changed, 184 insertions(+), 7 deletions(-)
diff --git a/.github/workflows/flink.yml b/.github/workflows/flink.yml
index 1ebcc2e4b6..87df3910a0 100644
--- a/.github/workflows/flink.yml
+++ b/.github/workflows/flink.yml
@@ -91,7 +91,7 @@ jobs:
export folly_SOURCE=BUNDLED
git clone -b gluten-20260829 https://github.com/bigo-sg/velox4j.git
cd velox4j
- git reset --hard cfe63ea5f7838a0221a268ce58d6d223c594bddc
+ 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 cd7b5b4e1d..e11e6e41a1 100644
--- a/gluten-flink/docs/Flink.md
+++ b/gluten-flink/docs/Flink.md
@@ -39,6 +39,10 @@ export PATH=$JAVA_HOME/bin:$PATH
**Get Velox4j**
+The velox4j commit is maintained in the single file
`gluten-flink/velox4j-ref.txt`;
+edit that file to update the pin — the CI workflow and this guide both read the
+commit from there.
+
Gluten for Flink depends on [Velox4j](https://github.com/velox4j/velox4j) to
call velox. This is an experimental feature.
You need to get the Velox4j code, and compile it first.
@@ -48,7 +52,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 cc8b56d74baa3863ae0fb8387f5745652370fa7d
+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/planner/src/main/java/org/apache/gluten/rexnode/RexNodeConverter.java
b/gluten-flink/planner/src/main/java/org/apache/gluten/rexnode/RexNodeConverter.java
index 38228b104d..9e79797be3 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
@@ -119,6 +119,7 @@ public class RexNodeConverter {
case DATE:
case INTERVAL_SECOND:
case SYMBOL:
+ case TIME:
case TIMESTAMP:
return new VarCharValue(null);
default:
@@ -174,6 +175,10 @@ public class RexNodeConverter {
long epochMillis = literal.getValueAs(Long.class);
return 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());
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 4f0547a144..b148fb4877 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,12 @@ 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())),
+ // Offload targets implemented in velox's flinksql namespace;
+ // the names arrive upper-cased from the operator table.
+ Map.entry(
+ "CURRENT_DATE", Arrays.asList(() -> new
DefaultRexCallConverter("current_date"))),
+ Map.entry(
+ "CURRENT_TIME", Arrays.asList(() -> new
DefaultRexCallConverter("current_time"))),
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).
diff --git
a/gluten-flink/runtime/src/main/java/org/apache/gluten/table/runtime/config/VeloxQueryConfig.java
b/gluten-flink/runtime/src/main/java/org/apache/gluten/table/runtime/config/VeloxQueryConfig.java
index a1026cfe49..e66f77f366 100644
---
a/gluten-flink/runtime/src/main/java/org/apache/gluten/table/runtime/config/VeloxQueryConfig.java
+++
b/gluten-flink/runtime/src/main/java/org/apache/gluten/table/runtime/config/VeloxQueryConfig.java
@@ -25,6 +25,7 @@ import org.apache.flink.table.api.config.TableConfigOptions;
import java.util.HashMap;
import java.util.Map;
+import java.util.TimeZone;
public class VeloxQueryConfig {
@@ -42,10 +43,12 @@ public class VeloxQueryConfig {
Map<String, String> configMap = new HashMap<>();
configMap.put(keyVeloxAdjustTimestampToSessionTimeZone, "true");
String localTimeZone = config.get(TableConfigOptions.LOCAL_TIME_ZONE);
- // As flink's default timezone value is `default`, it is not a valid
timezone id, so we should
- // convert it to `UTC` timezone.
- if
(TableConfigOptions.LOCAL_TIME_ZONE.defaultValue().equals(localTimeZone)) {
- configMap.put(keyVeloxSessionTimezone, "UTC");
+ // Flink's `default` value means the JVM default timezone, not UTC; velox
+ // must resolve it the same way so CURRENT_DATE/CURRENT_TIME (and every
+ // session-timezone-sensitive function) match native Flink.
+ if (localTimeZone == null
+ ||
TableConfigOptions.LOCAL_TIME_ZONE.defaultValue().equals(localTimeZone)) {
+ configMap.put(keyVeloxSessionTimezone, TimeZone.getDefault().getID());
} else {
configMap.put(keyVeloxSessionTimezone, localTimeZone);
}
diff --git
a/gluten-flink/runtime/src/main/java/org/apache/gluten/util/LogicalTypeConverter.java
b/gluten-flink/runtime/src/main/java/org/apache/gluten/util/LogicalTypeConverter.java
index 8d684fab52..c5cdc1c84d 100644
---
a/gluten-flink/runtime/src/main/java/org/apache/gluten/util/LogicalTypeConverter.java
+++
b/gluten-flink/runtime/src/main/java/org/apache/gluten/util/LogicalTypeConverter.java
@@ -32,6 +32,7 @@ import org.apache.flink.table.types.logical.LogicalType;
import org.apache.flink.table.types.logical.MapType;
import org.apache.flink.table.types.logical.RowType;
import org.apache.flink.table.types.logical.SymbolType;
+import org.apache.flink.table.types.logical.TimeType;
import org.apache.flink.table.types.logical.TimestampType;
import org.apache.flink.table.types.logical.TinyIntType;
import org.apache.flink.table.types.logical.VarCharType;
@@ -117,6 +118,10 @@ public class LogicalTypeConverter {
Map.entry(
TinyIntType.class,
logicalType -> new
io.github.zhztheplayer.velox4j.type.TinyIntType()),
+ // Flink RowData represents TIME as INT (millisecond-of-day), the
+ // same encoding the flinksql time functions use.
+ Map.entry(
+ TimeType.class, logicalType -> new
io.github.zhztheplayer.velox4j.type.IntegerType()),
Map.entry(
SymbolType.class,
logicalType -> new
io.github.zhztheplayer.velox4j.type.VarCharType()));
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 2acde24b38..8809a15311 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,6 +16,7 @@
*/
package org.apache.gluten.rexnode;
+import io.github.zhztheplayer.velox4j.variant.IntegerValue;
import io.github.zhztheplayer.velox4j.variant.VarCharValue;
import io.github.zhztheplayer.velox4j.variant.Variant;
@@ -30,6 +31,7 @@ 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.apache.calcite.util.TimeString;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
@@ -58,6 +60,24 @@ public class RexNodeConverterTest {
assertThat(((VarCharValue) variant).getValue()).isEqualTo("apple");
}
+ @Test
+ public void testTimeLiteralConvertsToMillisOfDay() {
+ RexLiteral literal = rexBuilder.makeTimeLiteral(new TimeString(12, 34,
56), 0);
+ Variant variant = RexNodeConverter.toVariant(literal);
+ assertThat(variant).isInstanceOf(IntegerValue.class);
+ assertThat(((IntegerValue) variant).getValue())
+ .isEqualTo(12 * 3_600_000 + 34 * 60_000 + 56_000);
+ }
+
+ @Test
+ public void testTimeNullLiteralConvertsToNullCarrier() {
+ RexLiteral literal =
+ (RexLiteral)
rexBuilder.makeNullLiteral(typeFactory.createSqlType(SqlTypeName.TIME));
+ Variant variant = RexNodeConverter.toVariant(literal);
+ assertThat(variant).isInstanceOf(VarCharValue.class);
+ assertThat(((VarCharValue) variant).getValue()).isNull();
+ }
+
@Test
public void testCharLiteralExtractsRawValue() {
RexLiteral literal =
diff --git
a/gluten-flink/ut/src/test/java/org/apache/gluten/table/runtime/stream/common/GlutenStreamingTestBase.java
b/gluten-flink/ut/src/test/java/org/apache/gluten/table/runtime/stream/common/GlutenStreamingTestBase.java
index 393a0064d5..c88ea73509 100644
---
a/gluten-flink/ut/src/test/java/org/apache/gluten/table/runtime/stream/common/GlutenStreamingTestBase.java
+++
b/gluten-flink/ut/src/test/java/org/apache/gluten/table/runtime/stream/common/GlutenStreamingTestBase.java
@@ -131,6 +131,15 @@ public class GlutenStreamingTestBase extends
StreamingTestBase {
}
protected void runAndCheck(String query, List<String> expected) {
+ assertThat(runQuery(query)).isEqualTo(expected);
+ }
+
+ /**
+ * Runs a query through the embedded gluten stack and returns the print rows
— each line is one
+ * changelog row, e.g. "+I[value]". Callers that need tolerance-based checks
on time-varying
+ * expressions compare the returned rows themselves instead of using {@link
#runAndCheck}.
+ */
+ protected List<String> runQuery(String query) {
String printResultDirPath = System.getProperty("user.dir") + "/log/";
new File(printResultDirPath).mkdirs();
String printResultFilePath = printResultDirPath + "taskmanager.out";
@@ -191,7 +200,7 @@ public class GlutenStreamingTestBase extends
StreamingTestBase {
result.add(line);
}
}
- assertThat(result).isEqualTo(expected);
+ return result;
} catch (Exception e) {
throw new FlinkRuntimeException(e);
} finally {
diff --git
a/gluten-flink/ut/src/test/java/org/apache/gluten/table/runtime/stream/custom/CurrentDateFunctionsTest.java
b/gluten-flink/ut/src/test/java/org/apache/gluten/table/runtime/stream/custom/CurrentDateFunctionsTest.java
new file mode 100644
index 0000000000..947ba00ddc
--- /dev/null
+++
b/gluten-flink/ut/src/test/java/org/apache/gluten/table/runtime/stream/custom/CurrentDateFunctionsTest.java
@@ -0,0 +1,50 @@
+/*
+ * 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.time.LocalDate;
+import java.util.Arrays;
+import java.util.List;
+
+class CurrentDateFunctionsTest extends GlutenStreamingTestBase {
+
+ @Override
+ @BeforeEach
+ public void before() throws Exception {
+ super.before();
+ }
+
+ @Test
+ void testCurrentDateIsToday() {
+ List<Row> rows = Arrays.asList(Row.of(1), Row.of(2), Row.of(3));
+ createSimpleBoundedValuesTable("tblCD", "x int", rows);
+ String query = "select x, CAST(CURRENT_DATE AS DATE) from tblCD";
+ // The velox print sink renders DATE as epoch days; assert it equals today.
+ long expected = LocalDate.now().toEpochDay();
+ runAndCheck(
+ query,
+ Arrays.asList(
+ "+I[1, " + expected + "]", "+I[2, " + expected + "]", "+I[3, " +
expected + "]"));
+ }
+}
diff --git
a/gluten-flink/ut/src/test/java/org/apache/gluten/table/runtime/stream/custom/CurrentTimeFunctionsTest.java
b/gluten-flink/ut/src/test/java/org/apache/gluten/table/runtime/stream/custom/CurrentTimeFunctionsTest.java
new file mode 100644
index 0000000000..5a9cb39925
--- /dev/null
+++
b/gluten-flink/ut/src/test/java/org/apache/gluten/table/runtime/stream/custom/CurrentTimeFunctionsTest.java
@@ -0,0 +1,74 @@
+/*
+ * 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.time.LocalTime;
+import java.util.Arrays;
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+class CurrentTimeFunctionsTest extends GlutenStreamingTestBase {
+
+ @Override
+ @BeforeEach
+ public void before() throws Exception {
+ super.before();
+ }
+
+ @Test
+ void testCurrentTimeIsMillisOfDay() {
+ List<Row> rows = Arrays.asList(Row.of(1), Row.of(2), Row.of(3));
+ createSimpleBoundedValuesTable("tblCT", "x int", rows);
+ String query = "select x, CAST(CURRENT_TIME AS TIME) from tblCT";
+ // The velox print sink renders TIME as the underlying INT millis-of-day;
+ // each row must be close to the local wall clock.
+ List<String> result = runQuery(query);
+ assertThat(result).hasSize(3);
+ int nowMillis = (int) (LocalTime.now().toNanoOfDay() / 1_000_000);
+ for (int i = 0; i < result.size(); i++) {
+ String prefix = "+I[" + (i + 1) + ", ";
+ assertThat(result.get(i)).startsWith(prefix).endsWith("]");
+ int millis =
+ Integer.parseInt(result.get(i).substring(prefix.length(),
result.get(i).length() - 1));
+ assertThat(millis).isBetween(0, 86_400_000 - 1);
+ long delta = Math.abs(millis - nowMillis);
+ delta = Math.min(delta, 86_400_000L - delta); // midnight wraparound
+ assertThat(delta).isLessThanOrEqualTo(60_000L);
+ }
+ }
+
+ @Test
+ void testCurrentTimeComparisons() {
+ List<Row> rows = Arrays.asList(Row.of(1), Row.of(2));
+ createSimpleBoundedValuesTable("tblCT2", "x int", rows);
+ // TIME literals become INT millis-of-day constants. Inclusive bounds
+ // covering the entire day: always true at any run time, so the test
+ // cannot flip near midnight (strict bounds did).
+ String query =
+ "select x, CURRENT_TIME >= TIME '00:00:00', CURRENT_TIME <= TIME
'23:59:59.999'"
+ + " from tblCT2";
+ runAndCheck(query, Arrays.asList("+I[1, true, true]", "+I[2, true,
true]"));
+ }
+}
diff --git a/gluten-flink/velox4j-ref.txt b/gluten-flink/velox4j-ref.txt
new file mode 100644
index 0000000000..e443ac0dec
--- /dev/null
+++ b/gluten-flink/velox4j-ref.txt
@@ -0,0 +1 @@
+cfe63ea5f7838a0221a268ce58d6d223c594bddc
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
