You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by tw...@apache.org on 2021/10/25 13:19:10 UTC
[flink] branch master updated: [FLINK-24500][table] Move
SqlDateTimeUtils to table-common as DateTimeUtils
This is an automated email from the ASF dual-hosted git repository.
twalthr pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 055c8c8 [FLINK-24500][table] Move SqlDateTimeUtils to table-common as DateTimeUtils
055c8c8 is described below
commit 055c8c808a6c1115ba5dd1526c3769e654850500
Author: slinkydeveloper <fr...@gmail.com>
AuthorDate: Mon Oct 11 09:52:19 2021 +0200
[FLINK-24500][table] Move SqlDateTimeUtils to table-common as DateTimeUtils
Signed-off-by: slinkydeveloper <fr...@gmail.com>
This closes #17454.
---
.../flink/connector/hbase1/util/HBaseTestBase.java | 6 +-
.../flink/connector/hbase2/util/HBaseTestBase.java | 6 +-
.../orc/nohive/vector/AbstractOrcNoHiveVector.java | 2 +-
.../flink/orc/vector/AbstractOrcColumnVector.java | 2 +-
.../flink/orc/OrcColumnarRowSplitReaderTest.java | 2 +-
.../parquet/vector/ParquetSplitReaderUtil.java | 2 +-
.../parquet/ParquetColumnarRowInputFormatTest.java | 4 +-
.../vector/ParquetColumnarRowSplitReaderTest.java | 5 +-
.../table/runtime/typeutils/PythonTypeUtils.java | 8 +-
.../apache/flink/table/utils/DateTimeUtils.java} | 554 +++++++++++++++++++--
.../flink/table/utils/TimestampStringUtils.java | 2 +-
.../functions/utils/HiveTableSqlFunction.java | 4 +-
.../planner/codegen/CodeGeneratorContext.scala | 10 +-
.../table/planner/codegen/ExprCodeGenerator.scala | 11 +-
.../planner/codegen/calls/BuiltInMethods.scala | 149 +++---
.../planner/codegen/calls/ExtractCallGen.scala | 3 +-
.../planner/codegen/calls/FloorCeilCallGen.scala | 6 +-
.../planner/codegen/calls/FunctionGenerator.scala | 22 +-
.../planner/codegen/calls/ScalarOperatorGens.scala | 27 +-
.../codegen/calls/TimestampDiffCallGen.scala | 5 +-
.../flink/table/planner/expressions/symbols.scala | 2 +-
.../table/planner/plan/utils/PartitionPruner.scala | 10 +-
.../stream/jsonplan/TableSourceJsonPlanITCase.java | 2 +-
.../jsonplan/WatermarkAssignerJsonPlanITCase.java | 2 +-
.../planner/runtime/batch/sql/CalcITCase.scala | 2 +-
.../runtime/batch/sql/TableScanITCase.scala | 2 +-
.../planner/runtime/batch/sql/UnnestITCase.scala | 2 +-
.../sql/agg/AggregateReduceGroupingITCase.scala | 2 +-
.../table/planner/runtime/utils/TestData.scala | 2 +-
.../table/planner/utils/DateTimeTestUtil.scala | 8 +-
flink-table/flink-table-runtime/pom.xml | 49 --
.../table/data/conversion/DateDateConverter.java | 6 +-
.../data/conversion/DateLocalDateConverter.java | 6 +-
.../data/conversion/TimeLocalTimeConverter.java | 6 +-
.../table/data/conversion/TimeTimeConverter.java | 6 +-
.../table/data/util/DataFormatConverters.java | 18 +-
.../operators/window/grouping/WindowsGrouping.java | 3 +-
.../flink/table/data/DataFormatConvertersTest.java | 8 +-
.../data/vector/VectorizedColumnBatchTest.java | 2 +-
39 files changed, 710 insertions(+), 258 deletions(-)
diff --git a/flink-connectors/flink-connector-hbase-1.4/src/test/java/org/apache/flink/connector/hbase1/util/HBaseTestBase.java b/flink-connectors/flink-connector-hbase-1.4/src/test/java/org/apache/flink/connector/hbase1/util/HBaseTestBase.java
index ee3926c..c39da66 100644
--- a/flink-connectors/flink-connector-hbase-1.4/src/test/java/org/apache/flink/connector/hbase1/util/HBaseTestBase.java
+++ b/flink-connectors/flink-connector-hbase-1.4/src/test/java/org/apache/flink/connector/hbase1/util/HBaseTestBase.java
@@ -35,9 +35,9 @@ import java.sql.Timestamp;
import java.util.ArrayList;
import java.util.List;
-import static org.apache.flink.table.runtime.functions.SqlDateTimeUtils.dateToInternal;
-import static org.apache.flink.table.runtime.functions.SqlDateTimeUtils.timeToInternal;
-import static org.apache.flink.table.runtime.functions.SqlDateTimeUtils.timestampToInternal;
+import static org.apache.flink.table.utils.DateTimeUtils.dateToInternal;
+import static org.apache.flink.table.utils.DateTimeUtils.timeToInternal;
+import static org.apache.flink.table.utils.DateTimeUtils.timestampToInternal;
/** Abstract IT case class for HBase. */
public abstract class HBaseTestBase extends HBaseTestingClusterAutoStarter {
diff --git a/flink-connectors/flink-connector-hbase-2.2/src/test/java/org/apache/flink/connector/hbase2/util/HBaseTestBase.java b/flink-connectors/flink-connector-hbase-2.2/src/test/java/org/apache/flink/connector/hbase2/util/HBaseTestBase.java
index d4333bf..f0e0bd2 100644
--- a/flink-connectors/flink-connector-hbase-2.2/src/test/java/org/apache/flink/connector/hbase2/util/HBaseTestBase.java
+++ b/flink-connectors/flink-connector-hbase-2.2/src/test/java/org/apache/flink/connector/hbase2/util/HBaseTestBase.java
@@ -35,9 +35,9 @@ import java.sql.Timestamp;
import java.util.ArrayList;
import java.util.List;
-import static org.apache.flink.table.runtime.functions.SqlDateTimeUtils.dateToInternal;
-import static org.apache.flink.table.runtime.functions.SqlDateTimeUtils.timeToInternal;
-import static org.apache.flink.table.runtime.functions.SqlDateTimeUtils.timestampToInternal;
+import static org.apache.flink.table.utils.DateTimeUtils.dateToInternal;
+import static org.apache.flink.table.utils.DateTimeUtils.timeToInternal;
+import static org.apache.flink.table.utils.DateTimeUtils.timestampToInternal;
/** Abstract IT case class for HBase. */
public abstract class HBaseTestBase extends HBaseTestingClusterAutoStarter {
diff --git a/flink-formats/flink-orc-nohive/src/main/java/org/apache/flink/orc/nohive/vector/AbstractOrcNoHiveVector.java b/flink-formats/flink-orc-nohive/src/main/java/org/apache/flink/orc/nohive/vector/AbstractOrcNoHiveVector.java
index 53888e4..ef70c7e 100644
--- a/flink-formats/flink-orc-nohive/src/main/java/org/apache/flink/orc/nohive/vector/AbstractOrcNoHiveVector.java
+++ b/flink-formats/flink-orc-nohive/src/main/java/org/apache/flink/orc/nohive/vector/AbstractOrcNoHiveVector.java
@@ -36,7 +36,7 @@ import java.sql.Timestamp;
import java.time.LocalDate;
import java.time.LocalDateTime;
-import static org.apache.flink.table.runtime.functions.SqlDateTimeUtils.dateToInternal;
+import static org.apache.flink.table.utils.DateTimeUtils.dateToInternal;
/** This column vector is used to adapt hive's ColumnVector to Flink's ColumnVector. */
public abstract class AbstractOrcNoHiveVector
diff --git a/flink-formats/flink-orc/src/main/java/org/apache/flink/orc/vector/AbstractOrcColumnVector.java b/flink-formats/flink-orc/src/main/java/org/apache/flink/orc/vector/AbstractOrcColumnVector.java
index 9699b7e..deeea5f 100644
--- a/flink-formats/flink-orc/src/main/java/org/apache/flink/orc/vector/AbstractOrcColumnVector.java
+++ b/flink-formats/flink-orc/src/main/java/org/apache/flink/orc/vector/AbstractOrcColumnVector.java
@@ -41,7 +41,7 @@ import java.nio.charset.StandardCharsets;
import java.sql.Date;
import java.time.LocalDate;
-import static org.apache.flink.table.runtime.functions.SqlDateTimeUtils.dateToInternal;
+import static org.apache.flink.table.utils.DateTimeUtils.dateToInternal;
/** This column vector is used to adapt hive's ColumnVector to Flink's ColumnVector. */
public abstract class AbstractOrcColumnVector
diff --git a/flink-formats/flink-orc/src/test/java/org/apache/flink/orc/OrcColumnarRowSplitReaderTest.java b/flink-formats/flink-orc/src/test/java/org/apache/flink/orc/OrcColumnarRowSplitReaderTest.java
index 86929bf..67a7952 100644
--- a/flink-formats/flink-orc/src/test/java/org/apache/flink/orc/OrcColumnarRowSplitReaderTest.java
+++ b/flink-formats/flink-orc/src/test/java/org/apache/flink/orc/OrcColumnarRowSplitReaderTest.java
@@ -54,7 +54,7 @@ import java.util.HashMap;
import java.util.Map;
import java.util.stream.IntStream;
-import static org.apache.flink.table.runtime.functions.SqlDateTimeUtils.internalToDate;
+import static org.apache.flink.table.utils.DateTimeUtils.internalToDate;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
diff --git a/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/vector/ParquetSplitReaderUtil.java b/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/vector/ParquetSplitReaderUtil.java
index f3cc0cf..be4e88f 100644
--- a/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/vector/ParquetSplitReaderUtil.java
+++ b/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/vector/ParquetSplitReaderUtil.java
@@ -70,7 +70,7 @@ import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;
-import static org.apache.flink.table.runtime.functions.SqlDateTimeUtils.dateToInternal;
+import static org.apache.flink.table.utils.DateTimeUtils.dateToInternal;
import static org.apache.parquet.Preconditions.checkArgument;
/** Util for generating {@link ParquetColumnarRowSplitReader}. */
diff --git a/flink-formats/flink-parquet/src/test/java/org/apache/flink/formats/parquet/ParquetColumnarRowInputFormatTest.java b/flink-formats/flink-parquet/src/test/java/org/apache/flink/formats/parquet/ParquetColumnarRowInputFormatTest.java
index 8cb2af3..f04a5f7 100644
--- a/flink-formats/flink-parquet/src/test/java/org/apache/flink/formats/parquet/ParquetColumnarRowInputFormatTest.java
+++ b/flink-formats/flink-parquet/src/test/java/org/apache/flink/formats/parquet/ParquetColumnarRowInputFormatTest.java
@@ -25,7 +25,6 @@ import org.apache.flink.core.fs.Path;
import org.apache.flink.table.data.DecimalData;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.filesystem.PartitionFieldExtractor;
-import org.apache.flink.table.runtime.functions.SqlDateTimeUtils;
import org.apache.flink.table.types.logical.BigIntType;
import org.apache.flink.table.types.logical.BooleanType;
import org.apache.flink.table.types.logical.DateType;
@@ -39,6 +38,7 @@ import org.apache.flink.table.types.logical.SmallIntType;
import org.apache.flink.table.types.logical.TimestampType;
import org.apache.flink.table.types.logical.TinyIntType;
import org.apache.flink.table.types.logical.VarCharType;
+import org.apache.flink.table.utils.DateTimeUtils;
import org.apache.flink.types.Row;
import org.apache.flink.util.InstantiationUtil;
@@ -602,7 +602,7 @@ public class ParquetColumnarRowInputFormatTest {
assertEquals(13, row.getFloat(8), 0);
assertEquals(6.6, row.getDouble(9), 0);
assertEquals(
- SqlDateTimeUtils.dateToInternal(Date.valueOf("2020-11-23")),
+ DateTimeUtils.dateToInternal(Date.valueOf("2020-11-23")),
row.getInt(10));
assertEquals(
LocalDateTime.of(1999, 1, 1, 1, 1),
diff --git a/flink-formats/flink-parquet/src/test/java/org/apache/flink/formats/parquet/vector/ParquetColumnarRowSplitReaderTest.java b/flink-formats/flink-parquet/src/test/java/org/apache/flink/formats/parquet/vector/ParquetColumnarRowSplitReaderTest.java
index 96c9950..53d9de0 100644
--- a/flink-formats/flink-parquet/src/test/java/org/apache/flink/formats/parquet/vector/ParquetColumnarRowSplitReaderTest.java
+++ b/flink-formats/flink-parquet/src/test/java/org/apache/flink/formats/parquet/vector/ParquetColumnarRowSplitReaderTest.java
@@ -22,7 +22,6 @@ import org.apache.flink.core.fs.Path;
import org.apache.flink.table.data.ColumnarRowData;
import org.apache.flink.table.data.DecimalData;
import org.apache.flink.table.data.vector.VectorizedColumnBatch;
-import org.apache.flink.table.runtime.functions.SqlDateTimeUtils;
import org.apache.flink.table.types.DataType;
import org.apache.flink.table.types.logical.BigIntType;
import org.apache.flink.table.types.logical.BooleanType;
@@ -37,6 +36,7 @@ import org.apache.flink.table.types.logical.TimestampType;
import org.apache.flink.table.types.logical.TinyIntType;
import org.apache.flink.table.types.logical.VarCharType;
import org.apache.flink.table.types.utils.TypeConversions;
+import org.apache.flink.table.utils.DateTimeUtils;
import org.apache.flink.types.Row;
import org.apache.hadoop.conf.Configuration;
@@ -580,8 +580,7 @@ public class ParquetColumnarRowSplitReaderTest {
assertEquals(13, row.getFloat(8), 0);
assertEquals(6.6, row.getDouble(9), 0);
assertEquals(
- SqlDateTimeUtils.dateToInternal(Date.valueOf("2020-11-23")),
- row.getInt(10));
+ DateTimeUtils.dateToInternal(Date.valueOf("2020-11-23")), row.getInt(10));
assertEquals(
LocalDateTime.of(1999, 1, 1, 1, 1),
row.getTimestamp(11, 9).toLocalDateTime());
diff --git a/flink-python/src/main/java/org/apache/flink/table/runtime/typeutils/PythonTypeUtils.java b/flink-python/src/main/java/org/apache/flink/table/runtime/typeutils/PythonTypeUtils.java
index 6a84045..ee13724 100644
--- a/flink-python/src/main/java/org/apache/flink/table/runtime/typeutils/PythonTypeUtils.java
+++ b/flink-python/src/main/java/org/apache/flink/table/runtime/typeutils/PythonTypeUtils.java
@@ -30,7 +30,6 @@ import org.apache.flink.api.common.typeutils.base.ShortSerializer;
import org.apache.flink.api.common.typeutils.base.array.BytePrimitiveArraySerializer;
import org.apache.flink.fnexecution.v1.FlinkFnApi;
import org.apache.flink.table.data.DecimalData;
-import org.apache.flink.table.runtime.functions.SqlDateTimeUtils;
import org.apache.flink.table.runtime.typeutils.serializers.python.ArrayDataSerializer;
import org.apache.flink.table.runtime.typeutils.serializers.python.DecimalDataSerializer;
import org.apache.flink.table.runtime.typeutils.serializers.python.MapDataSerializer;
@@ -57,6 +56,7 @@ import org.apache.flink.table.types.logical.TinyIntType;
import org.apache.flink.table.types.logical.VarBinaryType;
import org.apache.flink.table.types.logical.VarCharType;
import org.apache.flink.table.types.logical.utils.LogicalTypeDefaultVisitor;
+import org.apache.flink.table.utils.DateTimeUtils;
import java.math.BigDecimal;
import java.math.RoundingMode;
@@ -107,7 +107,7 @@ public final class PythonTypeUtils {
* Converts the internal representation of a SQL DATE (int) to the Java type used for UDF
* parameters ({@link java.sql.Date}).
*
- * <p>Note: The implementation refers to {@link SqlDateTimeUtils#internalToDate}.
+ * <p>Note: The implementation refers to {@link DateTimeUtils#internalToDate}.
*/
public static java.sql.Date internalToDate(int v) {
// note that, in this case, can't handle Daylight Saving Time
@@ -119,7 +119,7 @@ public final class PythonTypeUtils {
* Converts the Java type used for UDF parameters of SQL DATE type ({@link java.sql.Date}) to
* internal representation (int).
*
- * <p>Note: The implementation refers to {@link SqlDateTimeUtils#dateToInternal}.
+ * <p>Note: The implementation refers to {@link DateTimeUtils#dateToInternal}.
*/
public static int dateToInternal(java.sql.Date date) {
long ts = date.getTime() + LOCAL_TZ.getOffset(date.getTime());
@@ -130,7 +130,7 @@ public final class PythonTypeUtils {
* Converts the Java type used for UDF parameters of SQL TIMESTAMP type ({@link
* java.sql.Timestamp}) to internal representation (long).
*
- * <p>Note: The implementation refers to {@link SqlDateTimeUtils#timestampToInternal}.
+ * <p>Note: The implementation refers to {@link DateTimeUtils#timestampToInternal}.
*/
public static long timestampToInternal(java.sql.Timestamp ts) {
long time = ts.getTime();
diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/SqlDateTimeUtils.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/utils/DateTimeUtils.java
similarity index 75%
rename from flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/SqlDateTimeUtils.java
rename to flink-table/flink-table-common/src/main/java/org/apache/flink/table/utils/DateTimeUtils.java
index 8d033d0..09c393b 100644
--- a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/SqlDateTimeUtils.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/utils/DateTimeUtils.java
@@ -1,12 +1,13 @@
/*
- * 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
+ * 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
+ * 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,
@@ -15,20 +16,16 @@
* limitations under the License.
*/
-package org.apache.flink.table.runtime.functions;
+package org.apache.flink.table.utils;
+import org.apache.flink.annotation.Internal;
import org.apache.flink.table.api.TableException;
import org.apache.flink.table.data.DecimalData;
-import org.apache.flink.table.data.DecimalDataUtils;
import org.apache.flink.table.data.TimestampData;
import org.apache.flink.table.types.logical.DateType;
import org.apache.flink.table.types.logical.LogicalType;
import org.apache.flink.table.types.logical.TimestampType;
-import org.apache.flink.table.utils.ThreadLocalCache;
-import org.apache.calcite.avatica.util.DateTimeUtils;
-import org.apache.calcite.avatica.util.TimeUnit;
-import org.apache.calcite.avatica.util.TimeUnitRange;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -47,7 +44,11 @@ import java.time.format.DateTimeFormatter;
import java.time.format.DateTimeParseException;
import java.time.temporal.ChronoUnit;
import java.time.temporal.TemporalAccessor;
+import java.util.Collections;
import java.util.Date;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Objects;
import java.util.TimeZone;
import static java.time.temporal.ChronoField.DAY_OF_MONTH;
@@ -62,12 +63,13 @@ import static java.time.temporal.ChronoField.YEAR;
* Utility functions for datetime types: date, time, timestamp. Currently, it is a bit messy putting
* date time functions in various classes because the runtime module does not depend on calcite..
*/
-public class SqlDateTimeUtils {
+@Internal
+public class DateTimeUtils {
- private static final Logger LOG = LoggerFactory.getLogger(SqlDateTimeUtils.class);
+ private static final Logger LOG = LoggerFactory.getLogger(DateTimeUtils.class);
/** The julian date of the epoch, 1970-01-01. */
- private static final int EPOCH_JULIAN = 2440588;
+ public static final int EPOCH_JULIAN = 2440588;
/** The number of milliseconds in a second. */
private static final long MILLIS_PER_SECOND = 1000L;
@@ -83,7 +85,7 @@ public class SqlDateTimeUtils {
*
* <p>This is the modulo 'mask' used when converting TIMESTAMP values to DATE and TIME values.
*/
- private static final long MILLIS_PER_DAY = 86400000L; // = 24 * 60 * 60 * 1000
+ public static final long MILLIS_PER_DAY = 86400000L; // = 24 * 60 * 60 * 1000
/** The SimpleDateFormat string for ISO dates, "yyyy-MM-dd". */
private static final String DATE_FORMAT_STRING = "yyyy-MM-dd";
@@ -238,7 +240,7 @@ public class SqlDateTimeUtils {
}
public static long toTimestamp(DecimalData v) {
- return DecimalDataUtils.castToLong(v);
+ return v.toBigDecimal().setScale(0, RoundingMode.DOWN).longValue();
}
// --------------------------------------------------------------------------------------------
@@ -288,16 +290,12 @@ public class SqlDateTimeUtils {
long epochMills;
switch (precision) {
case 0:
- DecimalData timeInMills =
- DecimalDataUtils.multiply(
- v,
- DecimalData.fromUnscaledLong(MILLIS_PER_SECOND, 3, 0),
- v.precision() + 3,
- 0);
- epochMills = DecimalDataUtils.castToLong(timeInMills);
+ epochMills =
+ v.toBigDecimal().setScale(0, RoundingMode.DOWN).longValue()
+ * MILLIS_PER_SECOND;
return timestampDataFromEpochMills(epochMills);
case 3:
- epochMills = DecimalDataUtils.castToLong(v);
+ epochMills = toTimestamp(v);
return timestampDataFromEpochMills(epochMills);
default:
throw new TableException(
@@ -457,7 +455,7 @@ public class SqlDateTimeUtils {
ZoneId zoneId = ZoneId.of("UTC");
Instant instant = Instant.ofEpochMilli(ts);
ZonedDateTime zdt = ZonedDateTime.ofInstant(instant, zoneId);
- return DateTimeUtils.ymdToUnixDate(zdt.getYear(), zdt.getMonthValue(), zdt.getDayOfMonth());
+ return ymdToUnixDate(zdt.getYear(), zdt.getMonthValue(), zdt.getDayOfMonth());
}
// --------------------------------------------------------------------------------------------
@@ -599,7 +597,13 @@ public class SqlDateTimeUtils {
/** Helper for CAST({time} AS VARCHAR(n)). */
public static String unixTimeToString(int time) {
final StringBuilder buf = new StringBuilder(8);
- unixTimeToString(buf, time, 0); // set milli second precision to 0
+ unixTimeToString(buf, time, 0); // set millisecond precision to 0
+ return buf.toString();
+ }
+
+ public static String unixTimeToString(int time, int precision) {
+ final StringBuilder buf = new StringBuilder(8 + (precision > 0 ? precision + 1 : 0));
+ unixTimeToString(buf, time, precision);
return buf.toString();
}
@@ -643,6 +647,152 @@ public class SqlDateTimeUtils {
buf.append((char) ('0' + i % 10));
}
+ /** Helper for CAST({date} AS VARCHAR(n)). */
+ public static String unixDateToString(int date) {
+ final StringBuilder buf = new StringBuilder(10);
+ unixDateToString(buf, date);
+ return buf.toString();
+ }
+
+ private static void unixDateToString(StringBuilder buf, int date) {
+ julianToString(buf, date + EPOCH_JULIAN);
+ }
+
+ private static void julianToString(StringBuilder buf, int julian) {
+ // this shifts the epoch back to astronomical year -4800 instead of the
+ // start of the Christian era in year AD 1 of the proleptic Gregorian
+ // calendar.
+ int j = julian + 32044;
+ int g = j / 146097;
+ int dg = j % 146097;
+ int c = (dg / 36524 + 1) * 3 / 4;
+ int dc = dg - c * 36524;
+ int b = dc / 1461;
+ int db = dc % 1461;
+ int a = (db / 365 + 1) * 3 / 4;
+ int da = db - a * 365;
+
+ // integer number of full years elapsed since March 1, 4801 BC
+ int y = g * 400 + c * 100 + b * 4 + a;
+ // integer number of full months elapsed since the last March 1
+ int m = (da * 5 + 308) / 153 - 2;
+ // number of days elapsed since day 1 of the month
+ int d = da - (m + 4) * 153 / 5 + 122;
+ int year = y - 4800 + (m + 2) / 12;
+ int month = (m + 2) % 12 + 1;
+ int day = d + 1;
+ int4(buf, year);
+ buf.append('-');
+ int2(buf, month);
+ buf.append('-');
+ int2(buf, day);
+ }
+
+ public static String intervalYearMonthToString(int v) {
+ final StringBuilder buf = new StringBuilder();
+ if (v >= 0) {
+ buf.append('+');
+ } else {
+ buf.append('-');
+ v = -v;
+ }
+ final int y = v / 12;
+ final int m = v % 12;
+ buf.append(y);
+ buf.append('-');
+ number(buf, m, 2);
+ return buf.toString();
+ }
+
+ public static StringBuilder number(StringBuilder buf, int v, int n) {
+ for (int k = digitCount(v); k < n; k++) {
+ buf.append('0');
+ }
+ return buf.append(v);
+ }
+
+ public static int digitCount(int v) {
+ for (int n = 1; ; n++) {
+ v /= 10;
+ if (v == 0) {
+ return n;
+ }
+ }
+ }
+
+ private static int roundUp(int dividend, int divisor) {
+ int remainder = dividend % divisor;
+ dividend -= remainder;
+ if (remainder * 2 > divisor) {
+ dividend += divisor;
+ }
+ return dividend;
+ }
+
+ private static long roundUp(long dividend, long divisor) {
+ long remainder = dividend % divisor;
+ dividend -= remainder;
+ if (remainder * 2 > divisor) {
+ dividend += divisor;
+ }
+ return dividend;
+ }
+
+ private static void fraction(StringBuilder buf, int scale, long ms) {
+ if (scale > 0) {
+ buf.append('.');
+ long v1 = scale == 3 ? ms : scale == 2 ? ms / 10 : scale == 1 ? ms / 100 : 0;
+ number(buf, (int) v1, scale);
+ }
+ }
+
+ private static long powerX(long a, long b) {
+ long x = 1;
+ while (b > 0) {
+ x *= a;
+ --b;
+ }
+ return x;
+ }
+
+ public static String intervalDayTimeToString(long v) {
+ return intervalDayTimeToString(v, 3);
+ }
+
+ public static String intervalDayTimeToString(long v, int scale) {
+ final StringBuilder buf = new StringBuilder();
+ if (v >= 0) {
+ buf.append('+');
+ } else {
+ buf.append('-');
+ v = -v;
+ }
+ final long ms;
+ final long s;
+ final long m;
+ final long h;
+ final long d;
+ v = roundUp(v, powerX(10, 3 - scale));
+ ms = v % 1000;
+ v /= 1000;
+ s = v % 60;
+ v /= 60;
+ m = v % 60;
+ v /= 60;
+ h = v % 24;
+ v /= 24;
+ d = v;
+ buf.append((int) d);
+ buf.append(' ');
+ number(buf, (int) h, 2);
+ buf.append(':');
+ number(buf, (int) m, 2);
+ buf.append(':');
+ number(buf, (int) s, 2);
+ fraction(buf, scale, ms);
+ return buf.toString();
+ }
+
/**
* Parses a given datetime string to milli seconds since 1970-01-01 00:00:00 UTC using the
* default format "yyyy-MM-dd" or "yyyy-MM-dd HH:mm:ss" depends on the string length.
@@ -713,14 +863,143 @@ public class SqlDateTimeUtils {
private static final DateType REUSE_DATE_TYPE = new DateType();
private static final TimestampType REUSE_TIMESTAMP_TYPE = new TimestampType(9);
- public static long unixTimeExtract(TimeUnitRange range, int ts) {
- return DateTimeUtils.unixTimeExtract(range, ts);
+ public static int unixTimeExtract(TimeUnitRange range, int time) {
+ assert time >= 0;
+ assert time < MILLIS_PER_DAY;
+ switch (range) {
+ case HOUR:
+ return time / (int) MILLIS_PER_HOUR;
+ case MINUTE:
+ final int minutes = time / (int) MILLIS_PER_MINUTE;
+ return minutes % 60;
+ case SECOND:
+ final int seconds = time / (int) MILLIS_PER_SECOND;
+ return seconds % 60;
+ default:
+ throw new AssertionError(range);
+ }
}
public static long extractFromTimestamp(TimeUnitRange range, TimestampData ts, TimeZone tz) {
return convertExtract(range, ts, REUSE_TIMESTAMP_TYPE, tz);
}
+ public static long unixDateExtract(TimeUnitRange range, long date) {
+ return unixDateExtract(range, (int) date);
+ }
+
+ public static long unixDateExtract(TimeUnitRange range, int date) {
+ switch (range) {
+ case EPOCH:
+ return date * 86400L;
+ default:
+ return julianExtract(range, date + 2440588);
+ }
+ }
+
+ private static int julianExtract(TimeUnitRange range, int julian) {
+ int j = julian + 32044;
+ int g = j / 146097;
+ int dg = j % 146097;
+ int c = (dg / '躬' + 1) * 3 / 4;
+ int dc = dg - c * '躬';
+ int b = dc / 1461;
+ int db = dc % 1461;
+ int a = (db / 365 + 1) * 3 / 4;
+ int da = db - a * 365;
+ int y = g * 400 + c * 100 + b * 4 + a;
+ int m = (da * 5 + 308) / 153 - 2;
+ int d = da - (m + 4) * 153 / 5 + 122;
+ int year = y - 4800 + (m + 2) / 12;
+ int month = (m + 2) % 12 + 1;
+ int day = d + 1;
+ switch (range) {
+ case YEAR:
+ return year;
+ case YEAR_TO_MONTH:
+ case DAY_TO_SECOND:
+ case DAY_TO_MINUTE:
+ case DAY_TO_HOUR:
+ case HOUR:
+ case HOUR_TO_MINUTE:
+ case HOUR_TO_SECOND:
+ case MINUTE_TO_SECOND:
+ case MINUTE:
+ case SECOND:
+ case EPOCH:
+ default:
+ throw new AssertionError(range);
+ case MONTH:
+ return month;
+ case DAY:
+ return day;
+ case ISOYEAR:
+ int weekNumber = getIso8601WeekNumber(julian, year, month, day);
+ if (weekNumber == 1 && month == 12) {
+ return year + 1;
+ } else {
+ if (month == 1 && weekNumber > 50) {
+ return year - 1;
+ }
+
+ return year;
+ }
+ case QUARTER:
+ return (month + 2) / 3;
+ case DOW:
+ return (int) floorMod((long) (julian + 1), 7L) + 1;
+ case ISODOW:
+ return (int) floorMod((long) julian, 7L) + 1;
+ case WEEK:
+ return getIso8601WeekNumber(julian, year, month, day);
+ case DOY:
+ long janFirst = (long) ymdToJulian(year, 1, 1);
+ return (int) ((long) julian - janFirst) + 1;
+ case DECADE:
+ return year / 10;
+ case CENTURY:
+ return year > 0 ? (year + 99) / 100 : (year - 99) / 100;
+ case MILLENNIUM:
+ return year > 0 ? (year + 999) / 1000 : (year - 999) / 1000;
+ }
+ }
+
+ private static long firstMondayOfFirstWeek(int year) {
+ long janFirst = (long) ymdToJulian(year, 1, 1);
+ long janFirstDow = floorMod(janFirst + 1L, 7L);
+ return janFirst + (11L - janFirstDow) % 7L - 3L;
+ }
+
+ private static int getIso8601WeekNumber(int julian, int year, int month, int day) {
+ long fmofw = firstMondayOfFirstWeek(year);
+ if (month == 12 && day > 28) {
+ return 31 - day + 4 > 7 - ((int) floorMod((long) julian, 7L) + 1)
+ && 31 - day + (int) (floorMod((long) julian, 7L) + 1L) >= 4
+ ? (int) ((long) julian - fmofw) / 7 + 1
+ : 1;
+ } else if (month == 1 && day < 5) {
+ return 4 - day <= 7 - ((int) floorMod((long) julian, 7L) + 1)
+ && day - (int) (floorMod((long) julian, 7L) + 1L) >= -3
+ ? 1
+ : (int) ((long) julian - firstMondayOfFirstWeek(year - 1)) / 7 + 1;
+ } else {
+ return (int) ((long) julian - fmofw) / 7 + 1;
+ }
+ }
+
+ private static long floorDiv(long x, long y) {
+ long r = x / y;
+ if ((x ^ y) < 0L && r * y != x) {
+ --r;
+ }
+
+ return r;
+ }
+
+ private static long floorMod(long x, long y) {
+ return x - floorDiv(x, y) * y;
+ }
+
private static long convertExtract(
TimeUnitRange range, TimestampData ts, LogicalType type, TimeZone tz) {
TimeUnit startUnit = range.startUnit;
@@ -741,7 +1020,7 @@ public class SqlDateTimeUtils {
case WEEK:
if (type instanceof TimestampType) {
long d = divide(utcTs, TimeUnit.DAY.multiplier);
- return DateTimeUtils.unixDateExtract(range, d);
+ return unixDateExtract(range, d);
} else if (type instanceof DateType) {
return divide(utcTs, TimeUnit.DAY.multiplier);
} else {
@@ -930,24 +1209,24 @@ public class SqlDateTimeUtils {
if (!floor && (month > 1 || day > 1)) {
year += 1;
}
- return DateTimeUtils.ymdToUnixDate(year, 1, 1);
+ return ymdToUnixDate(year, 1, 1);
case MONTH:
if (!floor && day > 1) {
month += 1;
}
- return DateTimeUtils.ymdToUnixDate(year, month, 1);
+ return ymdToUnixDate(year, month, 1);
case QUARTER:
if (!floor && (month > 1 || day > 1)) {
quarter += 1;
}
- return DateTimeUtils.ymdToUnixDate(year, quarter * 3 - 2, 1);
+ return ymdToUnixDate(year, quarter * 3 - 2, 1);
case WEEK:
- int dow = (int) DateTimeUtils.floorMod(julian + 1, 7); // sun=0, sat=6
+ int dow = (int) floorMod(julian + 1, 7); // sun=0, sat=6
int offset = dow;
if (!floor && offset > 0) {
offset -= 7;
}
- return DateTimeUtils.ymdToUnixDate(year, month, day) - offset;
+ return ymdToUnixDate(year, month, day) - offset;
default:
throw new AssertionError(range);
}
@@ -1147,7 +1426,7 @@ public class SqlDateTimeUtils {
}
public static String fromUnixtime(DecimalData unixtime, TimeZone tz) {
- return fromUnixtime(DecimalDataUtils.castToLong(unixtime), tz);
+ return fromUnixtime(toTimestamp(unixtime), tz);
}
public static String fromUnixtime(long unixtime) {
@@ -1413,7 +1692,7 @@ public class SqlDateTimeUtils {
if (!isIllegalDate(y, m, d)) {
return null;
}
- return DateTimeUtils.ymdToUnixDate(y, m, d);
+ return ymdToUnixDate(y, m, d);
}
public static Integer timeStringToUnixDate(String v) {
@@ -1629,4 +1908,203 @@ public class SqlDateTimeUtils {
long tenToTheN = (long) Math.pow(10, n);
return (l / tenToTheN) * tenToTheN;
}
+
+ public static long unixDateCeil(TimeUnitRange range, long date) {
+ return julianDateFloor(range, (int) date + 2440588, false);
+ }
+
+ public static long unixDateFloor(TimeUnitRange range, long date) {
+ return julianDateFloor(range, (int) date + EPOCH_JULIAN, true);
+ }
+
+ public static long unixTimestampFloor(TimeUnitRange range, long timestamp) {
+ int date = (int) (timestamp / MILLIS_PER_DAY);
+ final long f = julianDateFloor(range, date + EPOCH_JULIAN, true);
+ return f * MILLIS_PER_DAY;
+ }
+
+ public static long unixTimestampCeil(TimeUnitRange range, long timestamp) {
+ int date = (int) (timestamp / MILLIS_PER_DAY);
+ final long f = julianDateFloor(range, date + EPOCH_JULIAN, false);
+ return f * MILLIS_PER_DAY;
+ }
+
+ // --------------------------------------------------------------------------------------------
+ // TimeUnit and TimeUnitRange enums
+ // --------------------------------------------------------------------------------------------
+
+ /**
+ * Enumeration of time units used to construct an interval.
+ *
+ * <p>Only {@link #YEAR}, {@link #MONTH}, {@link #DAY}, {@link #HOUR}, {@link #MINUTE}, {@link
+ * #SECOND} can be the unit of a SQL interval.
+ *
+ * <p>The others ({@link #QUARTER}, {@link #WEEK}, {@link #MILLISECOND}, {@link #DOW}, {@link
+ * #DOY}, {@link #EPOCH}, {@link #DECADE}, {@link #CENTURY}, {@link #MILLENNIUM}, {@link
+ * #MICROSECOND}, {@link #NANOSECOND}, {@link #ISODOW} and {@link #ISOYEAR}) are convenient to
+ * use internally, when converting to and from UNIX timestamps. And also may be arguments to the
+ * {@code EXTRACT}, {@code TIMESTAMPADD} and {@code TIMESTAMPDIFF} functions.
+ */
+ public enum TimeUnit {
+ YEAR(true, ' ', BigDecimal.valueOf(12) /* months */, null),
+ MONTH(true, '-', BigDecimal.ONE /* months */, BigDecimal.valueOf(12)),
+ DAY(false, '-', BigDecimal.valueOf(MILLIS_PER_DAY), null),
+ HOUR(false, ' ', BigDecimal.valueOf(MILLIS_PER_HOUR), BigDecimal.valueOf(24)),
+ MINUTE(false, ':', BigDecimal.valueOf(MILLIS_PER_MINUTE), BigDecimal.valueOf(60)),
+ SECOND(false, ':', BigDecimal.valueOf(MILLIS_PER_SECOND), BigDecimal.valueOf(60)),
+
+ QUARTER(true, '*', BigDecimal.valueOf(3) /* months */, BigDecimal.valueOf(4)),
+ ISOYEAR(true, ' ', BigDecimal.valueOf(12) /* months */, null),
+ WEEK(false, '*', BigDecimal.valueOf(MILLIS_PER_DAY * 7), BigDecimal.valueOf(53)),
+ MILLISECOND(false, '.', BigDecimal.ONE, BigDecimal.valueOf(1000)),
+ MICROSECOND(false, '.', BigDecimal.ONE.scaleByPowerOfTen(-3), BigDecimal.valueOf(1000_000)),
+ NANOSECOND(
+ false, '.', BigDecimal.ONE.scaleByPowerOfTen(-6), BigDecimal.valueOf(1000_000_000)),
+ DOW(false, '-', null, null),
+ ISODOW(false, '-', null, null),
+ DOY(false, '-', null, null),
+ EPOCH(false, '*', null, null),
+ DECADE(true, '*', BigDecimal.valueOf(120) /* months */, null),
+ CENTURY(true, '*', BigDecimal.valueOf(1200) /* months */, null),
+ MILLENNIUM(true, '*', BigDecimal.valueOf(12000) /* months */, null);
+
+ public final boolean yearMonth;
+ public final char separator;
+ public final BigDecimal multiplier;
+ private final BigDecimal limit;
+
+ private static final TimeUnit[] CACHED_VALUES = values();
+
+ TimeUnit(boolean yearMonth, char separator, BigDecimal multiplier, BigDecimal limit) {
+ this.yearMonth = yearMonth;
+ this.separator = separator;
+ this.multiplier = multiplier;
+ this.limit = limit;
+ }
+
+ /**
+ * Returns the TimeUnit associated with an ordinal. The value returned is null if the
+ * ordinal is not a member of the TimeUnit enumeration.
+ */
+ public static TimeUnit getValue(int ordinal) {
+ return ordinal < 0 || ordinal >= CACHED_VALUES.length ? null : CACHED_VALUES[ordinal];
+ }
+
+ /**
+ * Returns whether a given value is valid for a field of this time unit.
+ *
+ * @param field Field value
+ * @return Whether value
+ */
+ public boolean isValidValue(BigDecimal field) {
+ return field.compareTo(BigDecimal.ZERO) >= 0
+ && (limit == null || field.compareTo(limit) < 0);
+ }
+ }
+
+ /**
+ * A range of time units. The first is more significant than the other (e.g. year-to-day) or the
+ * same as the other (e.g. month).
+ */
+ public enum TimeUnitRange {
+ YEAR(TimeUnit.YEAR, null),
+ YEAR_TO_MONTH(TimeUnit.YEAR, TimeUnit.MONTH),
+ MONTH(TimeUnit.MONTH, null),
+ DAY(TimeUnit.DAY, null),
+ DAY_TO_HOUR(TimeUnit.DAY, TimeUnit.HOUR),
+ DAY_TO_MINUTE(TimeUnit.DAY, TimeUnit.MINUTE),
+ DAY_TO_SECOND(TimeUnit.DAY, TimeUnit.SECOND),
+ HOUR(TimeUnit.HOUR, null),
+ HOUR_TO_MINUTE(TimeUnit.HOUR, TimeUnit.MINUTE),
+ HOUR_TO_SECOND(TimeUnit.HOUR, TimeUnit.SECOND),
+ MINUTE(TimeUnit.MINUTE, null),
+ MINUTE_TO_SECOND(TimeUnit.MINUTE, TimeUnit.SECOND),
+ SECOND(TimeUnit.SECOND, null),
+
+ // non-standard time units cannot participate in ranges
+ ISOYEAR(TimeUnit.ISOYEAR, null),
+ QUARTER(TimeUnit.QUARTER, null),
+ WEEK(TimeUnit.WEEK, null),
+ MILLISECOND(TimeUnit.MILLISECOND, null),
+ MICROSECOND(TimeUnit.MICROSECOND, null),
+ NANOSECOND(TimeUnit.NANOSECOND, null),
+ DOW(TimeUnit.DOW, null),
+ ISODOW(TimeUnit.ISODOW, null),
+ DOY(TimeUnit.DOY, null),
+ EPOCH(TimeUnit.EPOCH, null),
+ DECADE(TimeUnit.DECADE, null),
+ CENTURY(TimeUnit.CENTURY, null),
+ MILLENNIUM(TimeUnit.MILLENNIUM, null);
+
+ public final TimeUnit startUnit;
+ public final TimeUnit endUnit;
+
+ private static final Map<Pair<TimeUnit>, TimeUnitRange> MAP = createMap();
+
+ /**
+ * Creates a TimeUnitRange.
+ *
+ * @param startUnit Start time unit
+ * @param endUnit End time unit
+ */
+ TimeUnitRange(TimeUnit startUnit, TimeUnit endUnit) {
+ assert startUnit != null;
+ this.startUnit = startUnit;
+ this.endUnit = endUnit;
+ }
+
+ /**
+ * Returns a {@code TimeUnitRange} with a given start and end unit.
+ *
+ * @param startUnit Start unit
+ * @param endUnit End unit
+ * @return Time unit range, or null if not valid
+ */
+ public static TimeUnitRange of(TimeUnit startUnit, TimeUnit endUnit) {
+ return MAP.get(new Pair<>(startUnit, endUnit));
+ }
+
+ private static Map<Pair<TimeUnit>, TimeUnitRange> createMap() {
+ Map<Pair<TimeUnit>, TimeUnitRange> map = new HashMap<>();
+ for (TimeUnitRange value : values()) {
+ map.put(new Pair<>(value.startUnit, value.endUnit), value);
+ }
+ return Collections.unmodifiableMap(map);
+ }
+
+ /** Whether this is in the YEAR-TO-MONTH family of intervals. */
+ public boolean monthly() {
+ return ordinal() <= MONTH.ordinal();
+ }
+
+ /**
+ * Immutable pair of values of the same type.
+ *
+ * @param <E> the element type
+ */
+ private static class Pair<E> {
+ final E left;
+ final E right;
+
+ private Pair(E left, E right) {
+ this.left = left;
+ this.right = right;
+ }
+
+ @Override
+ public int hashCode() {
+ int k = (left == null) ? 0 : left.hashCode();
+ int k1 = (right == null) ? 0 : right.hashCode();
+ return ((k << 4) | k) ^ k1;
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ return obj == this
+ || obj instanceof Pair
+ && Objects.equals(left, ((Pair) obj).left)
+ && Objects.equals(right, ((Pair) obj).right);
+ }
+ }
+ }
}
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/utils/TimestampStringUtils.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/utils/TimestampStringUtils.java
index ec4c320..c5ea10b 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/utils/TimestampStringUtils.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/utils/TimestampStringUtils.java
@@ -117,7 +117,7 @@ public class TimestampStringUtils {
/**
* Cast TIME type value to VARCHAR(N), we use same SQL format with codegen in
- * org.apache.flink.table.runtime.functions.SqlDateTimeUtils.
+ * org.apache.flink.table.utils.DateTimeUtils.
*/
public static String unixTimeToString(int time) {
final StringBuilder buf = new StringBuilder(8);
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/utils/HiveTableSqlFunction.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/utils/HiveTableSqlFunction.java
index 703f12d..d9252e6 100644
--- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/utils/HiveTableSqlFunction.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/utils/HiveTableSqlFunction.java
@@ -22,9 +22,9 @@ import org.apache.flink.table.functions.FunctionIdentifier;
import org.apache.flink.table.functions.TableFunction;
import org.apache.flink.table.planner.calcite.FlinkTypeFactory;
import org.apache.flink.table.planner.plan.schema.FlinkTableFunction;
-import org.apache.flink.table.runtime.functions.SqlDateTimeUtils;
import org.apache.flink.table.types.DataType;
import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.utils.DateTimeUtils;
import org.apache.flink.util.InstantiationUtil;
import org.apache.flink.util.Preconditions;
@@ -168,7 +168,7 @@ public class HiveTableSqlFunction extends TableSqlFunction {
case TIME:
return LocalTime.ofNanoOfDay(((TimeString) value).getMillisOfDay() * 1000_000);
case TIMESTAMP:
- return SqlDateTimeUtils.unixTimestampToLocalDateTime(
+ return DateTimeUtils.unixTimestampToLocalDateTime(
((TimestampString) value).getMillisSinceEpoch());
default:
throw new RuntimeException("Not support type: " + type);
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/CodeGeneratorContext.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/CodeGeneratorContext.scala
index d8df230..7e16cdd 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/CodeGeneratorContext.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/CodeGeneratorContext.scala
@@ -27,7 +27,7 @@ import org.apache.flink.table.functions.{FunctionContext, UserDefinedFunction}
import org.apache.flink.table.planner.codegen.CodeGenUtils._
import org.apache.flink.table.planner.codegen.GenerateUtils.generateRecordStatement
import org.apache.flink.table.planner.utils.InternalConfigOptions
-import org.apache.flink.table.runtime.functions.SqlDateTimeUtils
+import org.apache.flink.table.utils.DateTimeUtils
import org.apache.flink.table.runtime.operators.TableStreamOperator
import org.apache.flink.table.runtime.typeutils.{ExternalSerializer, InternalSerializers}
import org.apache.flink.table.runtime.util.collections._
@@ -526,7 +526,7 @@ class CodeGeneratorContext(val tableConfig: TableConfig) {
// declaration
reusableMemberStatements.add(s"private int $fieldTerm;")
- val utilsName = classOf[SqlDateTimeUtils].getCanonicalName
+ val utilsName = classOf[DateTimeUtils].getCanonicalName
// assignment
val field =
@@ -545,7 +545,7 @@ class CodeGeneratorContext(val tableConfig: TableConfig) {
val fieldTerm = s"queryStartLocaltime"
val queryStartLocalTimestamp = addReusableQueryLevelLocalDateTime()
- val utilsName = classOf[SqlDateTimeUtils].getCanonicalName
+ val utilsName = classOf[DateTimeUtils].getCanonicalName
// declaration
reusableMemberStatements.add(
s"""
@@ -562,7 +562,7 @@ class CodeGeneratorContext(val tableConfig: TableConfig) {
val fieldTerm = s"date"
val timestamp = addReusableRecordLevelLocalDateTime()
- val utilsName = classOf[SqlDateTimeUtils].getCanonicalName
+ val utilsName = classOf[DateTimeUtils].getCanonicalName
// declaration
reusableMemberStatements.add(s"private int $fieldTerm;")
@@ -579,7 +579,7 @@ class CodeGeneratorContext(val tableConfig: TableConfig) {
*/
def addReusableQueryLevelCurrentDate(): String = {
val fieldTerm = s"queryStartDate"
- val utilsName = classOf[SqlDateTimeUtils].getCanonicalName
+ val utilsName = classOf[DateTimeUtils].getCanonicalName
val timestamp = addReusableQueryLevelLocalDateTime()
reusableMemberStatements.add(
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/ExprCodeGenerator.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/ExprCodeGenerator.scala
index 3bebca4..627605c 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/ExprCodeGenerator.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/ExprCodeGenerator.scala
@@ -42,7 +42,6 @@ import org.apache.flink.table.runtime.typeutils.TypeCheckUtils.{isNumeric, isTem
import org.apache.flink.table.types.logical._
import org.apache.flink.table.types.logical.utils.LogicalTypeChecks.{getFieldCount, isCompositeType}
import org.apache.flink.table.typeutils.TimeIndicatorTypeInfo
-
import org.apache.calcite.rex._
import org.apache.calcite.sql.`type`.{ReturnTypes, SqlTypeName}
import org.apache.calcite.sql.{SqlKind, SqlOperator}
@@ -419,7 +418,15 @@ class ExprCodeGenerator(ctx: CodeGeneratorContext, nullableInput: Boolean)
case _ =>
literal.getValue3
}
- generateLiteral(ctx, resultType, value)
+ // Make sure to convert avatica time types to flink internal types
+ val convertedValue = value match {
+ case tu: org.apache.calcite.avatica.util.TimeUnit =>
+ org.apache.flink.table.utils.DateTimeUtils.TimeUnit.valueOf(tu.name())
+ case tur: org.apache.calcite.avatica.util.TimeUnitRange =>
+ org.apache.flink.table.utils.DateTimeUtils.TimeUnitRange.valueOf(tur.name())
+ case _ => value
+ }
+ generateLiteral(ctx, resultType, convertedValue)
}
override def visitCorrelVariable(correlVariable: RexCorrelVariable): GeneratedExpression = {
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/BuiltInMethods.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/BuiltInMethods.scala
index 5086a2b..08d4ba8 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/BuiltInMethods.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/BuiltInMethods.scala
@@ -20,12 +20,12 @@ package org.apache.flink.table.planner.codegen.calls
import org.apache.flink.table.data.{DecimalData, TimestampData}
import org.apache.flink.table.runtime.functions._
+import org.apache.flink.table.utils.DateTimeUtils
+import org.apache.flink.table.utils.DateTimeUtils.TimeUnitRange
-import org.apache.calcite.avatica.util.{DateTimeUtils, TimeUnitRange}
import org.apache.calcite.linq4j.tree.Types
import org.apache.calcite.runtime.{JsonFunctions, SqlFunctions}
-import org.apache.calcite.sql.{SqlJsonConstructorNullClause, SqlJsonExistsErrorBehavior,
- SqlJsonQueryEmptyOrErrorBehavior, SqlJsonQueryWrapperBehavior, SqlJsonValueEmptyOrErrorBehavior}
+import org.apache.calcite.sql.{SqlJsonConstructorNullClause, SqlJsonExistsErrorBehavior, SqlJsonQueryEmptyOrErrorBehavior, SqlJsonQueryWrapperBehavior, SqlJsonValueEmptyOrErrorBehavior}
import java.lang.reflect.Method
import java.lang.{Byte => JByte, Integer => JInteger, Long => JLong, Short => JShort}
@@ -227,240 +227,255 @@ object BuiltInMethods {
// SQL DATE TIME FUNCTIONS
val UNIX_TIME_TO_STRING = Types.lookupMethod(
- classOf[SqlDateTimeUtils],
+ classOf[DateTimeUtils],
"unixTimeToString",
classOf[Int])
+ val UNIX_DATE_TO_STRING = Types.lookupMethod(
+ classOf[DateTimeUtils],
+ "unixDateToString",
+ classOf[Int])
+
val TIMESTAMP_TO_STRING = Types.lookupMethod(
- classOf[SqlDateTimeUtils],
+ classOf[DateTimeUtils],
"timestampToString",
classOf[TimestampData], classOf[Int])
+ val INTERVAL_YEAR_MONTH_TO_STRING = Types.lookupMethod(
+ classOf[DateTimeUtils],
+ "intervalYearMonthToString",
+ classOf[Int])
+
+ val INTERVAL_DAY_TIME_TO_STRING = Types.lookupMethod(
+ classOf[DateTimeUtils],
+ "intervalDayTimeToString",
+ classOf[Long])
+
val TIMESTAMP_TO_STRING_TIME_ZONE = Types.lookupMethod(
- classOf[SqlDateTimeUtils],
+ classOf[DateTimeUtils],
"timestampToString",
classOf[TimestampData], classOf[TimeZone], classOf[Int])
val TIMESTAMP_TO_TIMESTAMP_WITH_LOCAL_ZONE = Types.lookupMethod(
- classOf[SqlDateTimeUtils],
+ classOf[DateTimeUtils],
"timestampToTimestampWithLocalZone",
classOf[TimestampData], classOf[TimeZone])
val TIMESTAMP_WITH_LOCAL_ZONE_TO_TIMESTAMP = Types.lookupMethod(
- classOf[SqlDateTimeUtils],
+ classOf[DateTimeUtils],
"timestampWithLocalZoneToTimestamp",
classOf[TimestampData], classOf[TimeZone])
val STRING_TO_DATE_WITH_FORMAT = Types.lookupMethod(
- classOf[SqlDateTimeUtils],
+ classOf[DateTimeUtils],
"strToDate",
classOf[String], classOf[String])
val DATE_FORMAT_STRING_STRING_STRING_TIME_ZONE = Types.lookupMethod(
- classOf[SqlDateTimeUtils], "dateFormat", classOf[String],
+ classOf[DateTimeUtils], "dateFormat", classOf[String],
classOf[String], classOf[String], classOf[TimeZone])
val DATE_FORMAT_STIRNG_STRING_TIME_ZONE = Types.lookupMethod(
- classOf[SqlDateTimeUtils], "dateFormat", classOf[String], classOf[String], classOf[TimeZone])
+ classOf[DateTimeUtils], "dateFormat", classOf[String], classOf[String], classOf[TimeZone])
val DATE_FORMAT_TIMESTAMP_STRING_TIME_ZONE = Types.lookupMethod(
- classOf[SqlDateTimeUtils],
+ classOf[DateTimeUtils],
"dateFormat",
classOf[TimestampData],
classOf[String],
classOf[TimeZone])
val DATE_FORMAT_STIRNG_STRING = Types.lookupMethod(
- classOf[SqlDateTimeUtils], "dateFormat", classOf[String], classOf[String])
+ classOf[DateTimeUtils], "dateFormat", classOf[String], classOf[String])
val DATE_FORMAT_TIMESTAMP_STRING = Types.lookupMethod(
- classOf[SqlDateTimeUtils], "dateFormat", classOf[TimestampData], classOf[String])
+ classOf[DateTimeUtils], "dateFormat", classOf[TimestampData], classOf[String])
val UNIX_TIMESTAMP_FORMAT = Types.lookupMethod(
- classOf[SqlDateTimeUtils],
+ classOf[DateTimeUtils],
"unixTimestamp",
classOf[String],
classOf[String],
classOf[TimeZone])
val UNIX_TIMESTAMP_STR = Types.lookupMethod(
- classOf[SqlDateTimeUtils], "unixTimestamp", classOf[String], classOf[TimeZone])
+ classOf[DateTimeUtils], "unixTimestamp", classOf[String], classOf[TimeZone])
val UNIX_TIMESTAMP = Types.lookupMethod(
- classOf[SqlDateTimeUtils], "unixTimestamp")
+ classOf[DateTimeUtils], "unixTimestamp")
val UNIX_TIMESTAMP_TS = Types.lookupMethod(
- classOf[SqlDateTimeUtils], "unixTimestamp", classOf[Long])
+ classOf[DateTimeUtils], "unixTimestamp", classOf[Long])
val FROM_UNIXTIME_FORMAT = Types.lookupMethod(
- classOf[SqlDateTimeUtils], "fromUnixtime", classOf[Long], classOf[String], classOf[TimeZone])
+ classOf[DateTimeUtils], "fromUnixtime", classOf[Long], classOf[String], classOf[TimeZone])
val FROM_UNIXTIME = Types.lookupMethod(
- classOf[SqlDateTimeUtils], "fromUnixtime", classOf[Long], classOf[TimeZone])
+ classOf[DateTimeUtils], "fromUnixtime", classOf[Long], classOf[TimeZone])
val DATEDIFF_T_S_TIME_ZONE = Types.lookupMethod(
- classOf[SqlDateTimeUtils], "dateDiff", classOf[Long], classOf[String], classOf[TimeZone])
+ classOf[DateTimeUtils], "dateDiff", classOf[Long], classOf[String], classOf[TimeZone])
val DATEDIFF_S_S_TIME_ZONE = Types.lookupMethod(
- classOf[SqlDateTimeUtils], "dateDiff", classOf[String], classOf[String], classOf[TimeZone])
+ classOf[DateTimeUtils], "dateDiff", classOf[String], classOf[String], classOf[TimeZone])
val DATEDIFF_S_T_TIME_ZONE = Types.lookupMethod(
- classOf[SqlDateTimeUtils], "dateDiff", classOf[String], classOf[Long], classOf[TimeZone])
+ classOf[DateTimeUtils], "dateDiff", classOf[String], classOf[Long], classOf[TimeZone])
val DATEDIFF_T_T_TIME_ZONE = Types.lookupMethod(
- classOf[SqlDateTimeUtils], "dateDiff", classOf[Long], classOf[Long], classOf[TimeZone])
+ classOf[DateTimeUtils], "dateDiff", classOf[Long], classOf[Long], classOf[TimeZone])
val DATEDIFF_T_S = Types.lookupMethod(
- classOf[SqlDateTimeUtils], "dateDiff", classOf[Long], classOf[String])
+ classOf[DateTimeUtils], "dateDiff", classOf[Long], classOf[String])
val DATEDIFF_S_S = Types.lookupMethod(
- classOf[SqlDateTimeUtils], "dateDiff", classOf[String], classOf[String])
+ classOf[DateTimeUtils], "dateDiff", classOf[String], classOf[String])
val DATEDIFF_S_T = Types.lookupMethod(
- classOf[SqlDateTimeUtils], "dateDiff", classOf[String], classOf[Long])
+ classOf[DateTimeUtils], "dateDiff", classOf[String], classOf[Long])
val DATEDIFF_T_T = Types.lookupMethod(
- classOf[SqlDateTimeUtils], "dateDiff", classOf[Long], classOf[Long])
+ classOf[DateTimeUtils], "dateDiff", classOf[Long], classOf[Long])
val DATE_SUB_S_TIME_ZONE = Types.lookupMethod(
- classOf[SqlDateTimeUtils], "dateSub", classOf[String], classOf[Int], classOf[TimeZone])
+ classOf[DateTimeUtils], "dateSub", classOf[String], classOf[Int], classOf[TimeZone])
val DATE_SUB_T_TIME_ZONE = Types.lookupMethod(
- classOf[SqlDateTimeUtils], "dateSub", classOf[Long], classOf[Int], classOf[TimeZone])
+ classOf[DateTimeUtils], "dateSub", classOf[Long], classOf[Int], classOf[TimeZone])
val DATE_SUB_S = Types.lookupMethod(
- classOf[SqlDateTimeUtils], "dateSub", classOf[String], classOf[Int])
+ classOf[DateTimeUtils], "dateSub", classOf[String], classOf[Int])
val DATE_SUB_T = Types.lookupMethod(
- classOf[SqlDateTimeUtils], "dateSub", classOf[Long], classOf[Int])
+ classOf[DateTimeUtils], "dateSub", classOf[Long], classOf[Int])
val DATE_ADD_S_TIME_ZONE = Types.lookupMethod(
- classOf[SqlDateTimeUtils], "dateAdd", classOf[String], classOf[Int], classOf[TimeZone])
+ classOf[DateTimeUtils], "dateAdd", classOf[String], classOf[Int], classOf[TimeZone])
val DATE_ADD_T_TIME_ZONE = Types.lookupMethod(
- classOf[SqlDateTimeUtils], "dateAdd", classOf[Long], classOf[Int], classOf[TimeZone])
+ classOf[DateTimeUtils], "dateAdd", classOf[Long], classOf[Int], classOf[TimeZone])
val DATE_ADD_S = Types.lookupMethod(
- classOf[SqlDateTimeUtils], "dateAdd", classOf[String], classOf[Int])
+ classOf[DateTimeUtils], "dateAdd", classOf[String], classOf[Int])
val DATE_ADD_T = Types.lookupMethod(
- classOf[SqlDateTimeUtils], "dateAdd", classOf[Long], classOf[Int])
+ classOf[DateTimeUtils], "dateAdd", classOf[Long], classOf[Int])
val LONG_TO_TIMESTAMP = Types.lookupMethod(
- classOf[SqlDateTimeUtils],
+ classOf[DateTimeUtils],
"toTimestamp",
classOf[Long])
val DOUBLE_TO_TIMESTAMP = Types.lookupMethod(
- classOf[SqlDateTimeUtils],
+ classOf[DateTimeUtils],
"toTimestamp",
classOf[Double])
val DECIMAL_TO_TIMESTAMP = Types.lookupMethod(
- classOf[SqlDateTimeUtils],
+ classOf[DateTimeUtils],
"toTimestamp",
classOf[DecimalData])
val LONG_TO_TIMESTAMP_LTZ_WITH_PRECISION = Types.lookupMethod(
- classOf[SqlDateTimeUtils],
+ classOf[DateTimeUtils],
"toTimestampData",
classOf[Long], classOf[Int])
val DOUBLE_TO_TIMESTAMP_LTZ_WITH_PRECISION = Types.lookupMethod(
- classOf[SqlDateTimeUtils],
+ classOf[DateTimeUtils],
"toTimestampData",
classOf[Double], classOf[Int])
val DECIMAL_TO_TIMESTAMP_LTZ_WITH_PRECISION = Types.lookupMethod(
- classOf[SqlDateTimeUtils],
+ classOf[DateTimeUtils],
"toTimestampData",
classOf[DecimalData], classOf[Int])
val STRING_TO_TIMESTAMP = Types.lookupMethod(
- classOf[SqlDateTimeUtils],
+ classOf[DateTimeUtils],
"toTimestampData",
classOf[String])
val STRING_TO_TIMESTAMP_WITH_FORMAT = Types.lookupMethod(
- classOf[SqlDateTimeUtils],
+ classOf[DateTimeUtils],
"toTimestampData",
classOf[String], classOf[String])
val STRING_TO_TIMESTAMP_TIME_ZONE = Types.lookupMethod(
- classOf[SqlDateTimeUtils],
+ classOf[DateTimeUtils],
"toTimestamp",
classOf[String], classOf[TimeZone])
val STRING_TO_TIMESTAMP_WITH_FORMAT_TIME_ZONE = Types.lookupMethod(
- classOf[SqlDateTimeUtils],
+ classOf[DateTimeUtils],
"toTimestamp",
classOf[String], classOf[String], classOf[TimeZone])
val TIMESTAMP_WITH_LOCAL_TIME_ZONE_TO_DATE = Types.lookupMethod(
- classOf[SqlDateTimeUtils],
+ classOf[DateTimeUtils],
"timestampWithLocalZoneToDate",
classOf[Long], classOf[TimeZone])
val TIMESTAMP_WITH_LOCAL_TIME_ZONE_TO_TIME = Types.lookupMethod(
- classOf[SqlDateTimeUtils],
+ classOf[DateTimeUtils],
"timestampWithLocalZoneToTime",
classOf[Long], classOf[TimeZone])
val DATE_TO_TIMESTAMP_WITH_LOCAL_TIME_ZONE = Types.lookupMethod(
- classOf[SqlDateTimeUtils],
+ classOf[DateTimeUtils],
"dateToTimestampWithLocalZone",
classOf[Int], classOf[TimeZone])
val TIME_TO_TIMESTAMP_WITH_LOCAL_TIME_ZONE = Types.lookupMethod(
- classOf[SqlDateTimeUtils],
+ classOf[DateTimeUtils],
"timeToTimestampWithLocalZone",
classOf[Int], classOf[TimeZone])
val TIMESTAMP_TO_BIGINT = Types.lookupMethod(
- classOf[SqlDateTimeUtils],
+ classOf[DateTimeUtils],
"fromTimestamp",
classOf[Long])
val EXTRACT_FROM_TIMESTAMP_TIME_ZONE = Types.lookupMethod(
- classOf[SqlDateTimeUtils],
+ classOf[DateTimeUtils],
"extractFromTimestamp",
classOf[TimeUnitRange], classOf[TimestampData], classOf[TimeZone])
val UNIX_TIME_EXTRACT = Types.lookupMethod(
- classOf[SqlDateTimeUtils],
+ classOf[DateTimeUtils],
"unixTimeExtract",
classOf[TimeUnitRange], classOf[Int])
val EXTRACT_YEAR_MONTH = Types.lookupMethod(
- classOf[SqlDateTimeUtils],
+ classOf[DateTimeUtils],
"extractYearMonth",
classOf[TimeUnitRange], classOf[Int])
val TIMESTAMP_FLOOR_TIME_ZONE = Types.lookupMethod(
- classOf[SqlDateTimeUtils],
+ classOf[DateTimeUtils],
"timestampFloor",
classOf[TimeUnitRange], classOf[Long], classOf[TimeZone])
val TIMESTAMP_CEIL_TIME_ZONE = Types.lookupMethod(
- classOf[SqlDateTimeUtils],
+ classOf[DateTimeUtils],
"timestampCeil",
classOf[TimeUnitRange], classOf[Long], classOf[TimeZone])
val CONVERT_TZ = Types.lookupMethod(
- classOf[SqlDateTimeUtils],
+ classOf[DateTimeUtils],
"convertTz",
classOf[String], classOf[String], classOf[String])
val CONVERT_FORMAT_TZ = Types.lookupMethod(
- classOf[SqlDateTimeUtils],
+ classOf[DateTimeUtils],
"convertTz",
classOf[String], classOf[String], classOf[String], classOf[String])
val STRING_TO_DATE = Types.lookupMethod(
- classOf[SqlDateTimeUtils], "dateStringToUnixDate", classOf[String])
+ classOf[DateTimeUtils], "dateStringToUnixDate", classOf[String])
val STRING_TO_TIME = Types.lookupMethod(
- classOf[SqlDateTimeUtils], "timeStringToUnixDate", classOf[String])
+ classOf[DateTimeUtils], "timeStringToUnixDate", classOf[String])
val TRUNCATE_DOUBLE_ONE = Types.lookupMethod(classOf[SqlFunctions], "struncate",
classOf[Double])
@@ -485,9 +500,21 @@ object BuiltInMethods {
classOf[DecimalData], classOf[Int])
val UNIX_DATE_CEIL = Types.lookupMethod(classOf[DateTimeUtils], "unixDateCeil",
+ classOf[TimeUnitRange], classOf[Long])
+
+ val UNIX_TIMESTAMP_CEIL = Types.lookupMethod(classOf[DateTimeUtils], "unixTimestampCeil",
+ classOf[TimeUnitRange], classOf[Long])
+
+ val UNIX_DATE_FLOOR = Types.lookupMethod(classOf[DateTimeUtils], "unixDateFloor",
+ classOf[TimeUnitRange], classOf[Long])
+
+ val UNIX_TIMESTAMP_FLOOR = Types.lookupMethod(classOf[DateTimeUtils], "unixTimestampFloor",
+ classOf[TimeUnitRange], classOf[Long])
+
+ val UNIX_DATE_EXTRACT = Types.lookupMethod(classOf[DateTimeUtils], "unixDateExtract",
classOf[TimeUnitRange], classOf[Int])
- val TRUNCATE_SQL_TIMESTAMP = Types.lookupMethod(classOf[SqlDateTimeUtils], "truncate",
+ val TRUNCATE_SQL_TIMESTAMP = Types.lookupMethod(classOf[DateTimeUtils], "truncate",
classOf[TimestampData], classOf[Int])
val ADD_MONTHS = Types.lookupMethod(classOf[SqlFunctions], "addMonths",
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/ExtractCallGen.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/ExtractCallGen.scala
index 8be3c3f..88c932f 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/ExtractCallGen.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/ExtractCallGen.scala
@@ -23,8 +23,7 @@ import org.apache.flink.table.planner.codegen.CodeGenUtils._
import org.apache.flink.table.planner.codegen.GenerateUtils.generateCallIfArgsNotNull
import org.apache.flink.table.planner.codegen.{CodeGenException, CodeGeneratorContext, GeneratedExpression}
import org.apache.flink.table.types.logical.{LogicalType, LogicalTypeRoot}
-
-import org.apache.calcite.avatica.util.{TimeUnit, TimeUnitRange}
+import org.apache.flink.table.utils.DateTimeUtils.{TimeUnit, TimeUnitRange}
import java.lang.reflect.Method
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/FloorCeilCallGen.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/FloorCeilCallGen.scala
index cd3d7b9..7ba55c5 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/FloorCeilCallGen.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/FloorCeilCallGen.scala
@@ -18,13 +18,13 @@
package org.apache.flink.table.planner.codegen.calls
-import org.apache.flink.table.planner.codegen.CodeGenUtils.{getEnum, primitiveTypeTermForType, qualifyMethod, TIMESTAMP_DATA}
+import org.apache.flink.table.planner.codegen.CodeGenUtils.{TIMESTAMP_DATA, getEnum, primitiveTypeTermForType, qualifyMethod}
import org.apache.flink.table.planner.codegen.GenerateUtils.generateCallIfArgsNotNull
import org.apache.flink.table.planner.codegen.{CodeGeneratorContext, GeneratedExpression}
import org.apache.flink.table.types.logical.{LogicalType, LogicalTypeRoot}
-import org.apache.calcite.avatica.util.TimeUnitRange
-import org.apache.calcite.avatica.util.TimeUnitRange._
+import org.apache.flink.table.utils.DateTimeUtils.TimeUnitRange
+import org.apache.flink.table.utils.DateTimeUtils.TimeUnitRange._
import java.lang.reflect.Method
import java.util.TimeZone
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/FunctionGenerator.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/FunctionGenerator.scala
index 36e9723..8884498 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/FunctionGenerator.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/FunctionGenerator.scala
@@ -403,22 +403,22 @@ class FunctionGenerator private(config: TableConfig) {
addSqlFunction(
EXTRACT,
Seq(RAW, BIGINT),
- new ExtractCallGen(BuiltInMethod.UNIX_DATE_EXTRACT.method))
+ new ExtractCallGen(BuiltInMethods.UNIX_DATE_EXTRACT))
addSqlFunction(
EXTRACT,
Seq(RAW, DATE),
- new ExtractCallGen(BuiltInMethod.UNIX_DATE_EXTRACT.method))
+ new ExtractCallGen(BuiltInMethods.UNIX_DATE_EXTRACT))
addSqlFunction(
EXTRACT,
Seq(RAW, TIME_WITHOUT_TIME_ZONE),
- new ExtractCallGen(BuiltInMethod.UNIX_DATE_EXTRACT.method))
+ new ExtractCallGen(BuiltInMethods.UNIX_DATE_EXTRACT))
addSqlFunction(
EXTRACT,
Seq(RAW, TIMESTAMP_WITHOUT_TIME_ZONE),
- new ExtractCallGen(BuiltInMethod.UNIX_DATE_EXTRACT.method))
+ new ExtractCallGen(BuiltInMethods.UNIX_DATE_EXTRACT))
addSqlFunction(
EXTRACT,
@@ -428,12 +428,12 @@ class FunctionGenerator private(config: TableConfig) {
addSqlFunction(
EXTRACT,
Seq(RAW, INTERVAL_DAY_TIME),
- new ExtractCallGen(BuiltInMethod.UNIX_DATE_EXTRACT.method))
+ new ExtractCallGen(BuiltInMethods.UNIX_DATE_EXTRACT))
addSqlFunction(
EXTRACT,
Seq(RAW, INTERVAL_YEAR_MONTH),
- new ExtractCallGen(BuiltInMethod.UNIX_DATE_EXTRACT.method))
+ new ExtractCallGen(BuiltInMethods.UNIX_DATE_EXTRACT))
addSqlFunction(
TIMESTAMP_DIFF,
@@ -463,21 +463,21 @@ class FunctionGenerator private(config: TableConfig) {
Seq(DATE, RAW),
new FloorCeilCallGen(
BuiltInMethod.FLOOR.method,
- Some(BuiltInMethod.UNIX_DATE_FLOOR.method)))
+ Some(BuiltInMethods.UNIX_DATE_FLOOR)))
addSqlFunction(
FLOOR,
Seq(TIME_WITHOUT_TIME_ZONE, RAW),
new FloorCeilCallGen(
BuiltInMethod.FLOOR.method,
- Some(BuiltInMethod.UNIX_DATE_FLOOR.method)))
+ Some(BuiltInMethods.UNIX_DATE_FLOOR)))
addSqlFunction(
FLOOR,
Seq(TIMESTAMP_WITHOUT_TIME_ZONE, RAW),
new FloorCeilCallGen(
BuiltInMethod.FLOOR.method,
- Some(BuiltInMethod.UNIX_TIMESTAMP_FLOOR.method)))
+ Some(BuiltInMethods.UNIX_TIMESTAMP_FLOOR)))
addSqlFunction(
FLOOR,
@@ -500,14 +500,14 @@ class FunctionGenerator private(config: TableConfig) {
Seq(TIME_WITHOUT_TIME_ZONE, RAW),
new FloorCeilCallGen(
BuiltInMethod.CEIL.method,
- Some(BuiltInMethod.UNIX_DATE_CEIL.method)))
+ Some(BuiltInMethods.UNIX_DATE_CEIL)))
addSqlFunction(
CEIL,
Seq(TIMESTAMP_WITHOUT_TIME_ZONE, RAW),
new FloorCeilCallGen(
BuiltInMethod.CEIL.method,
- Some(BuiltInMethod.UNIX_TIMESTAMP_CEIL.method)))
+ Some(BuiltInMethods.UNIX_TIMESTAMP_CEIL)))
addSqlFunction(
CEIL,
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/ScalarOperatorGens.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/ScalarOperatorGens.scala
index 9215423..034d092 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/ScalarOperatorGens.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/ScalarOperatorGens.scala
@@ -27,7 +27,7 @@ import org.apache.flink.table.planner.codegen.GenerateUtils._
import org.apache.flink.table.planner.codegen.GeneratedExpression.{ALWAYS_NULL, NEVER_NULL, NO_CODE}
import org.apache.flink.table.planner.codegen.{CodeGenException, CodeGenUtils, CodeGeneratorContext, GeneratedExpression}
import org.apache.flink.table.planner.utils.JavaScalaConversionUtil.toScala
-import org.apache.flink.table.runtime.functions.{SqlDateTimeUtils, SqlFunctionUtils}
+import org.apache.flink.table.runtime.functions.SqlFunctionUtils
import org.apache.flink.table.runtime.types.LogicalTypeDataTypeConverter.fromLogicalTypeToDataType
import org.apache.flink.table.runtime.types.PlannerTypeUtils
import org.apache.flink.table.runtime.types.PlannerTypeUtils.{isInteroperable, isPrimitive}
@@ -38,16 +38,13 @@ import org.apache.flink.table.types.logical._
import org.apache.flink.table.types.logical.utils.LogicalTypeCasts.supportsExplicitCast
import org.apache.flink.table.types.logical.utils.LogicalTypeChecks.{getFieldTypes, hasFamily}
import org.apache.flink.table.types.logical.utils.LogicalTypeMerging.findCommonType
+import org.apache.flink.table.utils.DateTimeUtils
import org.apache.flink.util.Preconditions.checkArgument
-
-import org.apache.calcite.avatica.util.DateTimeUtils.MILLIS_PER_DAY
-import org.apache.calcite.avatica.util.{DateTimeUtils, TimeUnitRange}
-import org.apache.calcite.util.BuiltInMethod
+import org.apache.flink.table.utils.DateTimeUtils.MILLIS_PER_DAY
import java.lang.{StringBuilder => JStringBuilder}
import java.nio.charset.StandardCharsets
import java.util.Arrays.asList
-
import scala.collection.JavaConversions._
/**
@@ -1003,18 +1000,16 @@ object ScalarOperatorGens {
// Interval Months -> String
case (INTERVAL_YEAR_MONTH, VARCHAR | CHAR) =>
- val method = qualifyMethod(BuiltInMethod.INTERVAL_YEAR_MONTH_TO_STRING.method)
- val timeUnitRange = qualifyEnum(TimeUnitRange.YEAR_TO_MONTH)
+ val method = qualifyMethod(BuiltInMethods.INTERVAL_YEAR_MONTH_TO_STRING)
generateStringResultCallIfArgsNotNull(ctx, Seq(operand), targetType) {
- terms => s"$method(${terms.head}, $timeUnitRange)"
+ terms => s"$method(${terms.head})"
}
// Interval Millis -> String
case (INTERVAL_DAY_TIME, VARCHAR | CHAR) =>
- val method = qualifyMethod(BuiltInMethod.INTERVAL_DAY_TIME_TO_STRING.method)
- val timeUnitRange = qualifyEnum(TimeUnitRange.DAY_TO_SECOND)
+ val method = qualifyMethod(BuiltInMethods.INTERVAL_DAY_TIME_TO_STRING)
generateStringResultCallIfArgsNotNull(ctx, Seq(operand), targetType) {
- terms => s"$method(${terms.head}, $timeUnitRange, 3)" // milli second precision
+ terms => s"$method(${terms.head})" // milli second precision
}
// Array -> String
@@ -2032,17 +2027,17 @@ object ScalarOperatorGens {
val stringValue = stringLiteral.literalValue.get.toString
val literalCode = expectType.getTypeRoot match {
case DATE =>
- SqlDateTimeUtils.dateStringToUnixDate(stringValue) match {
+ DateTimeUtils.dateStringToUnixDate(stringValue) match {
case null => throw new ValidationException(s"String '$stringValue' is not a valid date")
case v => v
}
case TIME_WITHOUT_TIME_ZONE =>
- SqlDateTimeUtils.timeStringToUnixDate(stringValue) match {
+ DateTimeUtils.timeStringToUnixDate(stringValue) match {
case null => throw new ValidationException(s"String '$stringValue' is not a valid time")
case v => v
}
case TIMESTAMP_WITHOUT_TIME_ZONE =>
- SqlDateTimeUtils.toTimestampData(stringValue) match {
+ DateTimeUtils.toTimestampData(stringValue) match {
case null =>
throw new ValidationException(s"String '$stringValue' is not a valid timestamp")
case v => s"${CodeGenUtils.TIMESTAMP_DATA}.fromEpochMillis(" +
@@ -2489,7 +2484,7 @@ object ScalarOperatorGens {
operandTerm: String): String =
fromType.getTypeRoot match {
case DATE =>
- s"${qualifyMethod(BuiltInMethod.UNIX_DATE_TO_STRING.method)}($operandTerm)"
+ s"${qualifyMethod(BuiltInMethods.UNIX_DATE_TO_STRING)}($operandTerm)"
case TIME_WITHOUT_TIME_ZONE =>
s"${qualifyMethod(BuiltInMethods.UNIX_TIME_TO_STRING)}($operandTerm)"
case TIMESTAMP_WITHOUT_TIME_ZONE => // including rowtime indicator
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/TimestampDiffCallGen.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/TimestampDiffCallGen.scala
index d7b4154..b5f72bf 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/TimestampDiffCallGen.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/TimestampDiffCallGen.scala
@@ -23,9 +23,8 @@ import org.apache.flink.table.planner.codegen.GenerateUtils.generateCallIfArgsNo
import org.apache.flink.table.planner.codegen.{CodeGenException, CodeGeneratorContext, GeneratedExpression}
import org.apache.flink.table.types.logical.LogicalTypeRoot._
import org.apache.flink.table.types.logical.{IntType, LogicalType}
-
-import org.apache.calcite.avatica.util.DateTimeUtils.MILLIS_PER_DAY
-import org.apache.calcite.avatica.util.TimeUnit
+import org.apache.flink.table.utils.DateTimeUtils.TimeUnit
+import org.apache.flink.table.utils.DateTimeUtils.MILLIS_PER_DAY
class TimestampDiffCallGen extends CallGenerator {
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/expressions/symbols.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/expressions/symbols.scala
index c8ff100..39ab3f6 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/expressions/symbols.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/expressions/symbols.scala
@@ -19,8 +19,8 @@
package org.apache.flink.table.planner.expressions
import org.apache.flink.api.common.typeinfo.TypeInformation
+import org.apache.flink.table.utils.DateTimeUtils.{TimeUnit, TimeUnitRange}
-import org.apache.calcite.avatica.util.{TimeUnit, TimeUnitRange}
import org.apache.calcite.sql.fun.SqlTrimFunction
import scala.language.{existentials, implicitConversions}
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/utils/PartitionPruner.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/utils/PartitionPruner.scala
index 9c397a8..0befd15 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/utils/PartitionPruner.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/utils/PartitionPruner.scala
@@ -25,7 +25,7 @@ import org.apache.flink.table.api.{TableConfig, TableException}
import org.apache.flink.table.data.{DecimalDataUtils, GenericRowData, StringData, TimestampData}
import org.apache.flink.table.planner.codegen.CodeGenUtils.DEFAULT_COLLECTOR_TERM
import org.apache.flink.table.planner.codegen.{ConstantCodeGeneratorContext, ExprCodeGenerator, FunctionCodeGenerator}
-import org.apache.flink.table.runtime.functions.SqlDateTimeUtils
+import org.apache.flink.table.utils.DateTimeUtils
import org.apache.flink.table.runtime.typeutils.InternalTypeInfo
import org.apache.flink.table.types.logical.LogicalTypeRoot._
import org.apache.flink.table.types.logical.{BooleanType, DecimalType, LogicalType}
@@ -177,11 +177,11 @@ object PartitionPruner {
case DECIMAL =>
val decimalType = t.asInstanceOf[DecimalType]
DecimalDataUtils.castFrom(v, decimalType.getPrecision, decimalType.getScale)
- case DATE => SqlDateTimeUtils.dateStringToUnixDate(v)
- case TIME_WITHOUT_TIME_ZONE => SqlDateTimeUtils.timeStringToUnixDate(v)
- case TIMESTAMP_WITHOUT_TIME_ZONE => SqlDateTimeUtils.toTimestampData(v)
+ case DATE => DateTimeUtils.dateStringToUnixDate(v)
+ case TIME_WITHOUT_TIME_ZONE => DateTimeUtils.timeStringToUnixDate(v)
+ case TIMESTAMP_WITHOUT_TIME_ZONE => DateTimeUtils.toTimestampData(v)
case TIMESTAMP_WITH_LOCAL_TIME_ZONE => TimestampData.fromInstant(
- SqlDateTimeUtils.toTimestampData(v).toLocalDateTime.atZone(timeZone).toInstant)
+ DateTimeUtils.toTimestampData(v).toLocalDateTime.atZone(timeZone).toInstant)
case _ =>
throw new TableException(s"$t is not supported in PartitionPruner")
}
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/stream/jsonplan/TableSourceJsonPlanITCase.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/stream/jsonplan/TableSourceJsonPlanITCase.java
index 3a7e5b2..2755a68 100644
--- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/stream/jsonplan/TableSourceJsonPlanITCase.java
+++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/stream/jsonplan/TableSourceJsonPlanITCase.java
@@ -29,7 +29,7 @@ import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
-import static org.apache.flink.table.runtime.functions.SqlDateTimeUtils.unixTimestampToLocalDateTime;
+import static org.apache.flink.table.utils.DateTimeUtils.unixTimestampToLocalDateTime;
/** Test for table source json plan. */
public class TableSourceJsonPlanITCase extends JsonPlanTestBase {
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/stream/jsonplan/WatermarkAssignerJsonPlanITCase.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/stream/jsonplan/WatermarkAssignerJsonPlanITCase.java
index 0059b41..4565b61 100644
--- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/stream/jsonplan/WatermarkAssignerJsonPlanITCase.java
+++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/stream/jsonplan/WatermarkAssignerJsonPlanITCase.java
@@ -28,7 +28,7 @@ import java.io.File;
import java.util.Arrays;
import java.util.HashMap;
-import static org.apache.flink.table.runtime.functions.SqlDateTimeUtils.unixTimestampToLocalDateTime;
+import static org.apache.flink.table.utils.DateTimeUtils.unixTimestampToLocalDateTime;
/** Test for watermark assigner json plan. */
public class WatermarkAssignerJsonPlanITCase extends JsonPlanTestBase {
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/batch/sql/CalcITCase.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/batch/sql/CalcITCase.scala
index a357ceb..0b8a62d 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/batch/sql/CalcITCase.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/batch/sql/CalcITCase.scala
@@ -40,7 +40,7 @@ import org.apache.flink.table.planner.runtime.utils.UserDefinedFunctionTestUtils
import org.apache.flink.table.planner.runtime.utils.{BatchTableEnvUtil, BatchTestBase, TestData, UserDefinedFunctionTestUtils}
import org.apache.flink.table.planner.utils.{DateTimeTestUtil, TestLegacyFilterableTableSource}
import org.apache.flink.table.planner.utils.DateTimeTestUtil._
-import org.apache.flink.table.runtime.functions.SqlDateTimeUtils.unixTimestampToLocalDateTime
+import org.apache.flink.table.utils.DateTimeUtils.unixTimestampToLocalDateTime
import org.apache.flink.types.Row
import org.junit.Assert.assertEquals
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/batch/sql/TableScanITCase.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/batch/sql/TableScanITCase.scala
index 867c12e..f8d6e62 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/batch/sql/TableScanITCase.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/batch/sql/TableScanITCase.scala
@@ -25,7 +25,7 @@ import org.apache.flink.table.api.{TableSchema, Types}
import org.apache.flink.table.planner.runtime.utils.BatchTestBase
import org.apache.flink.table.planner.runtime.utils.BatchTestBase.row
import org.apache.flink.table.planner.utils.{TestTableSourceWithTime, WithoutTimeAttributesTableSource}
-import org.apache.flink.table.runtime.functions.SqlDateTimeUtils.unixTimestampToLocalDateTime
+import org.apache.flink.table.utils.DateTimeUtils.unixTimestampToLocalDateTime
import org.junit.Test
import java.lang.{Integer => JInt}
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/batch/sql/UnnestITCase.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/batch/sql/UnnestITCase.scala
index 3300e0a..932bf71 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/batch/sql/UnnestITCase.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/batch/sql/UnnestITCase.scala
@@ -22,7 +22,7 @@ import org.apache.flink.api.java.typeutils.{ObjectArrayTypeInfo, RowTypeInfo}
import org.apache.flink.table.api.Types
import org.apache.flink.table.planner.runtime.utils.BatchTestBase.row
import org.apache.flink.table.planner.runtime.utils.{BatchTestBase, TestData}
-import org.apache.flink.table.runtime.functions.SqlDateTimeUtils.unixTimestampToLocalDateTime
+import org.apache.flink.table.utils.DateTimeUtils.unixTimestampToLocalDateTime
import org.apache.flink.types.Row
import org.junit.Test
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/batch/sql/agg/AggregateReduceGroupingITCase.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/batch/sql/agg/AggregateReduceGroupingITCase.scala
index ff85f1e..cd521c3 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/batch/sql/agg/AggregateReduceGroupingITCase.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/batch/sql/agg/AggregateReduceGroupingITCase.scala
@@ -25,7 +25,7 @@ import org.apache.flink.table.planner.plan.stats.FlinkStatistic
import org.apache.flink.table.planner.runtime.utils.BatchTestBase
import org.apache.flink.table.planner.runtime.utils.BatchTestBase.row
import org.apache.flink.table.planner.utils.DateTimeTestUtil.localDateTime
-import org.apache.flink.table.runtime.functions.SqlDateTimeUtils.unixTimestampToLocalDateTime
+import org.apache.flink.table.utils.DateTimeUtils.unixTimestampToLocalDateTime
import org.junit.{Before, Test}
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/utils/TestData.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/utils/TestData.scala
index 8c13216..538c5a8 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/utils/TestData.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/utils/TestData.scala
@@ -26,7 +26,7 @@ import org.apache.flink.table.planner.factories.TestValuesTableFactory.changelog
import org.apache.flink.table.planner.{JHashMap, JInt}
import org.apache.flink.table.planner.runtime.utils.BatchTestBase.row
import org.apache.flink.table.planner.utils.DateTimeTestUtil._
-import org.apache.flink.table.runtime.functions.SqlDateTimeUtils.unixTimestampToLocalDateTime
+import org.apache.flink.table.utils.DateTimeUtils.unixTimestampToLocalDateTime
import org.apache.flink.types.Row
import java.lang.{Boolean => JBool, Long => JLong}
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/utils/DateTimeTestUtil.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/utils/DateTimeTestUtil.scala
index 78f7043..c9b23c6 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/utils/DateTimeTestUtil.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/utils/DateTimeTestUtil.scala
@@ -19,10 +19,8 @@
package org.apache.flink.table.planner.utils
import org.apache.flink.table.data.util.DataFormatConverters.{LocalDateConverter, LocalTimeConverter}
-import org.apache.flink.table.runtime.functions.SqlDateTimeUtils
+import org.apache.flink.table.utils.DateTimeUtils
-import org.apache.calcite.avatica.util.DateTimeUtils
-import org.apache.calcite.avatica.util.DateTimeUtils.dateStringToUnixDate
import java.time.{LocalDate, LocalDateTime, LocalTime, ZoneId}
object DateTimeTestUtil {
@@ -31,7 +29,7 @@ object DateTimeTestUtil {
if (s == null) {
null
} else {
- LocalDateConverter.INSTANCE.toExternal(dateStringToUnixDate(s))
+ LocalDateConverter.INSTANCE.toExternal(DateTimeUtils.dateStringToUnixDate(s))
}
}
@@ -47,7 +45,7 @@ object DateTimeTestUtil {
if (s == null) {
null
} else {
- SqlDateTimeUtils.toTimestampData(s).toLocalDateTime
+ DateTimeUtils.toTimestampData(s).toLocalDateTime
}
}
diff --git a/flink-table/flink-table-runtime/pom.xml b/flink-table/flink-table-runtime/pom.xml
index e91abeb..557cbda 100644
--- a/flink-table/flink-table-runtime/pom.xml
+++ b/flink-table/flink-table-runtime/pom.xml
@@ -92,55 +92,6 @@ under the License.
<version>${janino.version}</version>
</dependency>
- <dependency>
- <groupId>org.apache.calcite.avatica</groupId>
- <artifactId>avatica-core</artifactId>
- <!-- When updating the Calcite version, make sure to update the version and dependency exclusions -->
- <version>1.17.0</version>
- <exclusions>
- <!--
-
- Dependencies that are not needed for how we use Avatica right now.
-
- We exclude all the dependencies of Avatica because currently we only use
- TimeUnit, TimeUnitRange and SqlDateTimeUtils which only dependent JDK.
-
- "mvn dependency:tree" as of Avatica 1.17:
-
- [INFO] +- org.apache.calcite.avatica:avatica-core:jar:1.17.0:compile
-
- -->
- <exclusion>
- <groupId>org.apache.calcite.avatica</groupId>
- <artifactId>avatica-metrics</artifactId>
- </exclusion>
- <exclusion>
- <groupId>com.google.protobuf</groupId>
- <artifactId>protobuf-java</artifactId>
- </exclusion>
- <exclusion>
- <groupId>org.apache.httpcomponents</groupId>
- <artifactId>httpclient</artifactId>
- </exclusion>
- <exclusion>
- <groupId>org.apache.httpcomponents</groupId>
- <artifactId>httpcore</artifactId>
- </exclusion>
- <exclusion>
- <groupId>com.fasterxml.jackson.core</groupId>
- <artifactId>jackson-core</artifactId>
- </exclusion>
- <exclusion>
- <groupId>com.fasterxml.jackson.core</groupId>
- <artifactId>jackson-annotations</artifactId>
- </exclusion>
- <exclusion>
- <groupId>com.fasterxml.jackson.core</groupId>
- <artifactId>jackson-databind</artifactId>
- </exclusion>
- </exclusions>
- </dependency>
-
<!-- test dependencies -->
<dependency>
diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/data/conversion/DateDateConverter.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/data/conversion/DateDateConverter.java
index bbbfc04..717d794 100644
--- a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/data/conversion/DateDateConverter.java
+++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/data/conversion/DateDateConverter.java
@@ -19,8 +19,8 @@
package org.apache.flink.table.data.conversion;
import org.apache.flink.annotation.Internal;
-import org.apache.flink.table.runtime.functions.SqlDateTimeUtils;
import org.apache.flink.table.types.logical.DateType;
+import org.apache.flink.table.utils.DateTimeUtils;
/** Converter for {@link DateType} of {@link java.sql.Date} external type. */
@Internal
@@ -30,11 +30,11 @@ public class DateDateConverter implements DataStructureConverter<Integer, java.s
@Override
public Integer toInternal(java.sql.Date external) {
- return SqlDateTimeUtils.dateToInternal(external);
+ return DateTimeUtils.dateToInternal(external);
}
@Override
public java.sql.Date toExternal(Integer internal) {
- return SqlDateTimeUtils.internalToDate(internal);
+ return DateTimeUtils.internalToDate(internal);
}
}
diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/data/conversion/DateLocalDateConverter.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/data/conversion/DateLocalDateConverter.java
index f825cfd..1cd5b89 100644
--- a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/data/conversion/DateLocalDateConverter.java
+++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/data/conversion/DateLocalDateConverter.java
@@ -19,8 +19,8 @@
package org.apache.flink.table.data.conversion;
import org.apache.flink.annotation.Internal;
-import org.apache.flink.table.runtime.functions.SqlDateTimeUtils;
import org.apache.flink.table.types.logical.DateType;
+import org.apache.flink.table.utils.DateTimeUtils;
/** Converter for {@link DateType} of {@link java.time.LocalDate} external type. */
@Internal
@@ -31,11 +31,11 @@ public class DateLocalDateConverter
@Override
public Integer toInternal(java.time.LocalDate external) {
- return SqlDateTimeUtils.localDateToUnixDate(external);
+ return DateTimeUtils.localDateToUnixDate(external);
}
@Override
public java.time.LocalDate toExternal(Integer internal) {
- return SqlDateTimeUtils.unixDateToLocalDate(internal);
+ return DateTimeUtils.unixDateToLocalDate(internal);
}
}
diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/data/conversion/TimeLocalTimeConverter.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/data/conversion/TimeLocalTimeConverter.java
index d2fb8a5..c67bd82 100644
--- a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/data/conversion/TimeLocalTimeConverter.java
+++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/data/conversion/TimeLocalTimeConverter.java
@@ -19,8 +19,8 @@
package org.apache.flink.table.data.conversion;
import org.apache.flink.annotation.Internal;
-import org.apache.flink.table.runtime.functions.SqlDateTimeUtils;
import org.apache.flink.table.types.logical.TimeType;
+import org.apache.flink.table.utils.DateTimeUtils;
/** Converter for {@link TimeType} of {@link java.time.LocalTime} external type. */
@Internal
@@ -31,11 +31,11 @@ public class TimeLocalTimeConverter
@Override
public Integer toInternal(java.time.LocalTime external) {
- return SqlDateTimeUtils.localTimeToUnixDate(external);
+ return DateTimeUtils.localTimeToUnixDate(external);
}
@Override
public java.time.LocalTime toExternal(Integer internal) {
- return SqlDateTimeUtils.unixTimeToLocalTime(internal);
+ return DateTimeUtils.unixTimeToLocalTime(internal);
}
}
diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/data/conversion/TimeTimeConverter.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/data/conversion/TimeTimeConverter.java
index 3dfb252..6fcf7e2 100644
--- a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/data/conversion/TimeTimeConverter.java
+++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/data/conversion/TimeTimeConverter.java
@@ -19,8 +19,8 @@
package org.apache.flink.table.data.conversion;
import org.apache.flink.annotation.Internal;
-import org.apache.flink.table.runtime.functions.SqlDateTimeUtils;
import org.apache.flink.table.types.logical.TimeType;
+import org.apache.flink.table.utils.DateTimeUtils;
/** Converter for {@link TimeType} of {@link java.sql.Time} external type. */
@Internal
@@ -30,11 +30,11 @@ public class TimeTimeConverter implements DataStructureConverter<Integer, java.s
@Override
public Integer toInternal(java.sql.Time external) {
- return SqlDateTimeUtils.timeToInternal(external);
+ return DateTimeUtils.timeToInternal(external);
}
@Override
public java.sql.Time toExternal(Integer internal) {
- return SqlDateTimeUtils.internalToTime(internal);
+ return DateTimeUtils.internalToTime(internal);
}
}
diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/data/util/DataFormatConverters.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/data/util/DataFormatConverters.java
index 8b52acc..4ac03da 100644
--- a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/data/util/DataFormatConverters.java
+++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/data/util/DataFormatConverters.java
@@ -44,7 +44,6 @@ import org.apache.flink.table.data.binary.BinaryArrayData;
import org.apache.flink.table.data.binary.BinaryMapData;
import org.apache.flink.table.data.writer.BinaryArrayWriter;
import org.apache.flink.table.data.writer.BinaryWriter;
-import org.apache.flink.table.runtime.functions.SqlDateTimeUtils;
import org.apache.flink.table.runtime.types.LogicalTypeDataTypeConverter;
import org.apache.flink.table.runtime.typeutils.BigDecimalTypeInfo;
import org.apache.flink.table.runtime.typeutils.DecimalDataTypeInfo;
@@ -66,6 +65,7 @@ import org.apache.flink.table.types.logical.RawType;
import org.apache.flink.table.types.logical.TimestampType;
import org.apache.flink.table.types.logical.TypeInformationRawType;
import org.apache.flink.table.types.utils.TypeConversions;
+import org.apache.flink.table.utils.DateTimeUtils;
import org.apache.flink.types.Row;
import org.apache.commons.lang3.ArrayUtils;
@@ -730,12 +730,12 @@ public class DataFormatConverters {
@Override
Integer toInternalImpl(LocalDate value) {
- return SqlDateTimeUtils.localDateToUnixDate(value);
+ return DateTimeUtils.localDateToUnixDate(value);
}
@Override
LocalDate toExternalImpl(Integer value) {
- return SqlDateTimeUtils.unixDateToLocalDate(value);
+ return DateTimeUtils.unixDateToLocalDate(value);
}
@Override
@@ -755,12 +755,12 @@ public class DataFormatConverters {
@Override
Integer toInternalImpl(LocalTime value) {
- return SqlDateTimeUtils.localTimeToUnixDate(value);
+ return DateTimeUtils.localTimeToUnixDate(value);
}
@Override
LocalTime toExternalImpl(Integer value) {
- return SqlDateTimeUtils.unixTimeToLocalTime(value);
+ return DateTimeUtils.unixTimeToLocalTime(value);
}
@Override
@@ -835,12 +835,12 @@ public class DataFormatConverters {
@Override
Integer toInternalImpl(Date value) {
- return SqlDateTimeUtils.dateToInternal(value);
+ return DateTimeUtils.dateToInternal(value);
}
@Override
Date toExternalImpl(Integer value) {
- return SqlDateTimeUtils.internalToDate(value);
+ return DateTimeUtils.internalToDate(value);
}
@Override
@@ -860,12 +860,12 @@ public class DataFormatConverters {
@Override
Integer toInternalImpl(Time value) {
- return SqlDateTimeUtils.timeToInternal(value);
+ return DateTimeUtils.timeToInternal(value);
}
@Override
Time toExternalImpl(Integer value) {
- return SqlDateTimeUtils.internalToTime(value);
+ return DateTimeUtils.internalToTime(value);
}
@Override
diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/window/grouping/WindowsGrouping.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/window/grouping/WindowsGrouping.java
index 759edc2..c15f72b 100644
--- a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/window/grouping/WindowsGrouping.java
+++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/window/grouping/WindowsGrouping.java
@@ -22,10 +22,9 @@ import org.apache.flink.table.data.RowData;
import org.apache.flink.table.data.binary.BinaryRowData;
import org.apache.flink.table.runtime.operators.window.TimeWindow;
import org.apache.flink.table.runtime.util.RowIterator;
+import org.apache.flink.table.utils.DateTimeUtils;
import org.apache.flink.util.Preconditions;
-import org.apache.calcite.avatica.util.DateTimeUtils;
-
import java.io.Closeable;
import java.io.IOException;
diff --git a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/data/DataFormatConvertersTest.java b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/data/DataFormatConvertersTest.java
index 50c8be9..0eb0baa 100644
--- a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/data/DataFormatConvertersTest.java
+++ b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/data/DataFormatConvertersTest.java
@@ -35,7 +35,6 @@ import org.apache.flink.api.java.typeutils.TypeExtractor;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.data.util.DataFormatConverters;
import org.apache.flink.table.data.util.DataFormatConverters.DataFormatConverter;
-import org.apache.flink.table.runtime.functions.SqlDateTimeUtils;
import org.apache.flink.table.runtime.typeutils.DecimalDataTypeInfo;
import org.apache.flink.table.runtime.typeutils.InternalTypeInfo;
import org.apache.flink.table.runtime.typeutils.LegacyTimestampTypeInfo;
@@ -47,6 +46,7 @@ import org.apache.flink.table.types.logical.LegacyTypeInformationType;
import org.apache.flink.table.types.logical.LogicalTypeRoot;
import org.apache.flink.table.types.logical.VarCharType;
import org.apache.flink.table.types.utils.TypeConversions;
+import org.apache.flink.table.utils.DateTimeUtils;
import org.apache.flink.types.Row;
import org.apache.flink.types.RowKind;
@@ -107,9 +107,9 @@ public class DataFormatConvertersTest {
new short[] {5, 1},
new byte[] {5, 1},
new char[] {5, 1},
- SqlDateTimeUtils.unixDateToLocalDate(5),
- SqlDateTimeUtils.unixTimeToLocalTime(11),
- SqlDateTimeUtils.unixTimestampToLocalDateTime(11),
+ DateTimeUtils.unixDateToLocalDate(5),
+ DateTimeUtils.unixTimeToLocalTime(11),
+ DateTimeUtils.unixTimestampToLocalDateTime(11),
StringData.fromString("hahah")
};
diff --git a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/data/vector/VectorizedColumnBatchTest.java b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/data/vector/VectorizedColumnBatchTest.java
index 2de8df8..2f3fa99 100644
--- a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/data/vector/VectorizedColumnBatchTest.java
+++ b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/data/vector/VectorizedColumnBatchTest.java
@@ -31,8 +31,8 @@ import org.apache.flink.table.data.vector.heap.HeapFloatVector;
import org.apache.flink.table.data.vector.heap.HeapIntVector;
import org.apache.flink.table.data.vector.heap.HeapLongVector;
import org.apache.flink.table.data.vector.heap.HeapShortVector;
+import org.apache.flink.table.utils.DateTimeUtils;
-import org.apache.calcite.avatica.util.DateTimeUtils;
import org.junit.Test;
import java.io.ByteArrayOutputStream;