You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ja...@apache.org on 2021/04/21 14:30:43 UTC
[flink] 03/03: [FLINK-22354][table] Fix TIME field display in
sql-client
This is an automated email from the ASF dual-hosted git repository.
jark pushed a commit to branch release-1.13
in repository https://gitbox.apache.org/repos/asf/flink.git
commit cdf65b690c79cc940ea1b42d0b8506540071900f
Author: Leonard Xu <xb...@gmail.com>
AuthorDate: Wed Apr 21 15:14:15 2021 +0800
[FLINK-22354][table] Fix TIME field display in sql-client
This closes #15689
---
.../org/apache/flink/table/utils/PrintUtils.java | 24 ++++++-
.../flink/table/utils/TimestampStringUtils.java | 77 +++++++++++++++++++++-
.../apache/flink/table/utils/PrintUtilsTest.java | 18 +++--
3 files changed, 109 insertions(+), 10 deletions(-)
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/utils/PrintUtils.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/utils/PrintUtils.java
index 48ff9e6..891bfa0 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/utils/PrintUtils.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/utils/PrintUtils.java
@@ -46,9 +46,11 @@ import com.ibm.icu.lang.UProperty;
import javax.annotation.Nullable;
import java.io.PrintWriter;
+import java.sql.Time;
import java.sql.Timestamp;
import java.time.Instant;
import java.time.LocalDateTime;
+import java.time.LocalTime;
import java.time.ZoneId;
import java.util.ArrayList;
import java.util.Arrays;
@@ -59,7 +61,10 @@ import java.util.Map;
import java.util.stream.Stream;
import static org.apache.flink.table.types.logical.utils.LogicalTypeChecks.getPrecision;
+import static org.apache.flink.table.utils.TimestampStringUtils.localTimeToUnixDate;
+import static org.apache.flink.table.utils.TimestampStringUtils.timeToInternal;
import static org.apache.flink.table.utils.TimestampStringUtils.timestampToString;
+import static org.apache.flink.table.utils.TimestampStringUtils.unixTimeToString;
/** Utilities for print formatting. */
@Internal
@@ -238,7 +243,7 @@ public class PrintUtils {
}
/**
- * Normalizes field that contains TIMESTAMP and TIMESTAMP_LTZ type data.
+ * Normalizes field that contains TIMESTAMP, TIMESTAMP_LTZ and TIME type data.
*
* <p>This method also supports nested type ARRAY, ROW, MAP.
*/
@@ -252,6 +257,8 @@ public class PrintUtils {
case TIMESTAMP_WITHOUT_TIME_ZONE:
case TIMESTAMP_WITH_LOCAL_TIME_ZONE:
return formatTimestampField(field, fieldType, sessionTimeZone);
+ case TIME_WITHOUT_TIME_ZONE:
+ return formatTimeField(field);
case ARRAY:
LogicalType elementType = ((ArrayType) fieldType).getElementType();
if (field instanceof List) {
@@ -453,6 +460,21 @@ public class PrintUtils {
}
}
+ /** Formats the print content of TIME type data. */
+ private static Object formatTimeField(Object timeField) {
+ if (timeField.getClass().isAssignableFrom(int.class) || timeField instanceof Integer) {
+ return unixTimeToString((int) timeField);
+ } else if (timeField.getClass().isAssignableFrom(long.class) || timeField instanceof Long) {
+ return unixTimeToString(((Long) timeField).intValue());
+ } else if (timeField instanceof Time) {
+ return unixTimeToString(timeToInternal((Time) timeField));
+ } else if (timeField instanceof LocalTime) {
+ return unixTimeToString(localTimeToUnixDate((LocalTime) timeField));
+ } else {
+ return timeField;
+ }
+ }
+
public static String genBorderLine(int[] colWidths) {
StringBuilder sb = new StringBuilder();
sb.append("+");
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 ce41110..abde5b8 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
@@ -21,13 +21,27 @@ package org.apache.flink.table.utils;
import org.apache.flink.annotation.Internal;
import java.time.LocalDateTime;
+import java.time.LocalTime;
+import java.util.TimeZone;
-/** Utils to represent a LocalDateTime to String, considered the precision. */
+/**
+ * Utils to represent a LocalDateTime to String, considered the precision.
+ *
+ * <p>TODO://This class keep same SQL formats with {@code
+ * org.apache.flink.table.runtime.functions.SqlDateTimeUtils} which used in Flink SQL codegen, The
+ * two utils will be unified once FLINK-21456 finished.
+ */
@Internal
public class TimestampStringUtils {
- // TODO this method is copied from org.apache.flink.table.runtime.functions.SqlDateTimeUtils,
- // we can refactor these utils in the future
+ private static final long MILLIS_PER_SECOND = 1000L;
+ private static final long MILLIS_PER_MINUTE = 60000L;
+ private static final long MILLIS_PER_HOUR = 3600000L;
+ private static final long MILLIS_PER_DAY = 86400000L;
+
+ /** The local time zone, used to deal {@link java.sql.Time} value. */
+ private static final TimeZone LOCAL_TZ = TimeZone.getDefault();
+
public static String timestampToString(LocalDateTime ldt, int precision) {
String fraction = pad(9, (long) ldt.getNano());
while (fraction.length() > precision && fraction.endsWith("0")) {
@@ -99,4 +113,61 @@ public class TimestampStringUtils {
buf.append((char) ('0' + (i / 10) % 10));
buf.append((char) ('0' + i % 10));
}
+
+ /**
+ * Cast TIME type value to VARCHAR(N), we use same SQL format with codegen in
+ * org.apache.flink.table.runtime.functions.SqlDateTimeUtils.
+ */
+ public static String unixTimeToString(int time) {
+ final StringBuilder buf = new StringBuilder(8);
+ unixTimeToString(buf, time, 0); // set milli second precision to 0
+ return buf.toString();
+ }
+
+ private static void unixTimeToString(StringBuilder buf, int time, int precision) {
+ // we copy this method from Calcite DateTimeUtils but add the following changes
+ // time may be negative which means time milli seconds before 00:00:00
+ // this maybe a bug in calcite avatica
+ while (time < 0) {
+ time += MILLIS_PER_DAY;
+ }
+ int h = time / 3600000;
+ int time2 = time % 3600000;
+ int m = time2 / 60000;
+ int time3 = time2 % 60000;
+ int s = time3 / 1000;
+ int ms = time3 % 1000;
+ int2(buf, h);
+ buf.append(':');
+ int2(buf, m);
+ buf.append(':');
+ int2(buf, s);
+ if (precision > 0) {
+ buf.append('.');
+ while (precision > 0) {
+ buf.append((char) ('0' + (ms / 100)));
+ ms = ms % 100;
+ ms = ms * 10;
+
+ // keep consistent with Timestamp.toString()
+ if (ms == 0) {
+ break;
+ }
+
+ --precision;
+ }
+ }
+ }
+
+ public static int timeToInternal(java.sql.Time time) {
+ long ts = time.getTime() + LOCAL_TZ.getOffset(time.getTime());
+ return (int) (ts % MILLIS_PER_DAY);
+ }
+
+ public static int localTimeToUnixDate(LocalTime time) {
+ return time.getHour() * (int) MILLIS_PER_HOUR
+ + time.getMinute() * (int) MILLIS_PER_MINUTE
+ + time.getSecond() * (int) MILLIS_PER_SECOND
+ + time.getNano() / 1000_000;
+ }
}
diff --git a/flink-table/flink-table-common/src/test/java/org/apache/flink/table/utils/PrintUtilsTest.java b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/utils/PrintUtilsTest.java
index 0fe823f..e883944 100644
--- a/flink-table/flink-table-common/src/test/java/org/apache/flink/table/utils/PrintUtilsTest.java
+++ b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/utils/PrintUtilsTest.java
@@ -50,7 +50,7 @@ public class PrintUtilsTest {
@Test
public void testArrayToString() {
- Row row = new Row(6);
+ Row row = new Row(7);
row.setField(0, new int[] {1, 2});
row.setField(1, new Integer[] {3, 4});
row.setField(2, new Object[] {new int[] {5, 6}, new int[] {7, 8}});
@@ -67,6 +67,7 @@ public class PrintUtilsTest {
new Instant[] {Instant.ofEpochMilli(1), Instant.ofEpochMilli(10)},
new Instant[] {Instant.ofEpochSecond(1), Instant.ofEpochSecond(10)}
});
+ row.setField(6, new int[] {1123, 2123});
ResolvedSchema resolvedSchema =
ResolvedSchema.of(
@@ -81,18 +82,20 @@ public class PrintUtilsTest {
Column.physical(
"f5",
DataTypes.ARRAY(
- DataTypes.ARRAY(DataTypes.TIMESTAMP_LTZ(3))))));
+ DataTypes.ARRAY(DataTypes.TIMESTAMP_LTZ(3)))),
+ Column.physical("f6", DataTypes.ARRAY(DataTypes.TIME()))));
assertEquals(
"[[1, 2], [3, 4], [[5, 6], [7, 8]], [[9, 10], [11, 12]],"
+ " [2021-04-18 18:00:00.123456, 2021-04-18 18:00:00.000001],"
+ " [[1970-01-01 00:00:00.001, 1970-01-01 00:00:00.010],"
- + " [1970-01-01 00:00:01.000, 1970-01-01 00:00:10.000]]]",
+ + " [1970-01-01 00:00:01.000, 1970-01-01 00:00:10.000]],"
+ + " [00:00:01, 00:00:02]]",
Arrays.toString(PrintUtils.rowToString(row, resolvedSchema, UTC_ZONE_ID)));
}
@Test
public void testNestedRowToString() {
- Row row = new Row(3);
+ Row row = new Row(4);
row.setField(0, new int[] {1, 2});
Row row1 = new Row(4);
row1.setField(0, "hello");
@@ -111,6 +114,7 @@ public class PrintUtilsTest {
new int[] {1, 10},
new int[] {2, 20}
});
+ row.setField(3, new Integer[] {3000, 4000});
ResolvedSchema resolvedSchema =
ResolvedSchema.of(
@@ -124,11 +128,13 @@ public class PrintUtilsTest {
DataTypes.ARRAY(DataTypes.TIMESTAMP(6)),
DataTypes.ARRAY(DataTypes.TIMESTAMP_LTZ(6)))),
Column.physical(
- "f2", DataTypes.ARRAY(DataTypes.ARRAY(DataTypes.INT())))));
+ "f2", DataTypes.ARRAY(DataTypes.ARRAY(DataTypes.INT()))),
+ Column.physical("f3", DataTypes.ARRAY(DataTypes.TIME()))));
assertEquals(
"[[1, 2], +I[hello, [true, false],"
+ " [2021-04-18 18:00:00.123456, 2021-04-18 18:00:00.000001],"
- + " [1970-01-01 00:00:00.100000, 1970-01-01 00:00:00.200000]], [[1, 10], [2, 20]]]",
+ + " [1970-01-01 00:00:00.100000, 1970-01-01 00:00:00.200000]], [[1, 10], [2, 20]],"
+ + " [00:00:03, 00:00:04]]",
Arrays.toString(PrintUtils.rowToString(row, resolvedSchema, UTC_ZONE_ID)));
}