You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by sn...@apache.org on 2023/02/22 14:44:51 UTC
[flink] branch release-1.17 updated: [FLINK-31091][sql-gateway] Add Ser/de for Interval types (#21991)
This is an automated email from the ASF dual-hosted git repository.
snuyanzin pushed a commit to branch release-1.17
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-1.17 by this push:
new 8535c26198d [FLINK-31091][sql-gateway] Add Ser/de for Interval types (#21991)
8535c26198d is described below
commit 8535c26198d0e3372465aa283055ce39f4577f86
Author: Sergey Nuyanzin <sn...@gmail.com>
AuthorDate: Wed Feb 22 15:44:42 2023 +0100
[FLINK-31091][sql-gateway] Add Ser/de for Interval types (#21991)
---
.../src/test/resources/sql/select.q | 9 ++++
.../rest/serde/LogicalTypeJsonDeserializer.java | 29 +++++++++++
.../rest/serde/LogicalTypeJsonSerializer.java | 33 +++++++++++-
.../rest/serde/LogicalTypeJsonSerDeTest.java | 59 +++++++++++++++++++---
.../rest/serde/ResultInfoJsonSerDeTest.java | 38 +++++++++-----
.../src/test/resources/result_info_json_format.txt | 2 +-
.../resources/result_info_plain_text_format.txt | 2 +-
7 files changed, 149 insertions(+), 23 deletions(-)
diff --git a/flink-table/flink-sql-client/src/test/resources/sql/select.q b/flink-table/flink-sql-client/src/test/resources/sql/select.q
index d28d4c14507..99f9e047834 100644
--- a/flink-table/flink-sql-client/src/test/resources/sql/select.q
+++ b/flink-table/flink-sql-client/src/test/resources/sql/select.q
@@ -221,3 +221,12 @@ FROM (VALUES
+----------+---------------------+-------------------------+-------------------------------+-------------------------+-------------------------+-------------------------------+
2 rows in set
!ok
+
+SELECT INTERVAL '1' DAY as dayInterval, INTERVAL '1' YEAR as yearInterval;
++-----------------+--------------+
+| dayInterval | yearInterval |
++-----------------+--------------+
+| +1 00:00:00.000 | +1-00 |
++-----------------+--------------+
+1 row in set
+!ok
diff --git a/flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/rest/serde/LogicalTypeJsonDeserializer.java b/flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/rest/serde/LogicalTypeJsonDeserializer.java
index 63d2bbd1b4b..c1da541dc39 100644
--- a/flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/rest/serde/LogicalTypeJsonDeserializer.java
+++ b/flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/rest/serde/LogicalTypeJsonDeserializer.java
@@ -27,6 +27,7 @@ import org.apache.flink.table.types.logical.BinaryType;
import org.apache.flink.table.types.logical.BooleanType;
import org.apache.flink.table.types.logical.CharType;
import org.apache.flink.table.types.logical.DateType;
+import org.apache.flink.table.types.logical.DayTimeIntervalType;
import org.apache.flink.table.types.logical.DecimalType;
import org.apache.flink.table.types.logical.DoubleType;
import org.apache.flink.table.types.logical.FloatType;
@@ -46,6 +47,7 @@ import org.apache.flink.table.types.logical.TimestampType;
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.YearMonthIntervalType;
import org.apache.flink.table.types.logical.ZonedTimestampType;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonParser;
@@ -64,10 +66,12 @@ import static org.apache.flink.table.gateway.rest.serde.LogicalTypeJsonSerialize
import static org.apache.flink.table.gateway.rest.serde.LogicalTypeJsonSerializer.FIELD_NAME_FIELD_NAME;
import static org.apache.flink.table.gateway.rest.serde.LogicalTypeJsonSerializer.FIELD_NAME_FIELD_TYPE;
import static org.apache.flink.table.gateway.rest.serde.LogicalTypeJsonSerializer.FIELD_NAME_FILED_DESCRIPTION;
+import static org.apache.flink.table.gateway.rest.serde.LogicalTypeJsonSerializer.FIELD_NAME_FRACTIONAL_PRECISION;
import static org.apache.flink.table.gateway.rest.serde.LogicalTypeJsonSerializer.FIELD_NAME_KEY_TYPE;
import static org.apache.flink.table.gateway.rest.serde.LogicalTypeJsonSerializer.FIELD_NAME_LENGTH;
import static org.apache.flink.table.gateway.rest.serde.LogicalTypeJsonSerializer.FIELD_NAME_NULLABLE;
import static org.apache.flink.table.gateway.rest.serde.LogicalTypeJsonSerializer.FIELD_NAME_PRECISION;
+import static org.apache.flink.table.gateway.rest.serde.LogicalTypeJsonSerializer.FIELD_NAME_RESOLUTION;
import static org.apache.flink.table.gateway.rest.serde.LogicalTypeJsonSerializer.FIELD_NAME_SCALE;
import static org.apache.flink.table.gateway.rest.serde.LogicalTypeJsonSerializer.FIELD_NAME_SERIALIZER;
import static org.apache.flink.table.gateway.rest.serde.LogicalTypeJsonSerializer.FIELD_NAME_TYPE_NAME;
@@ -143,6 +147,9 @@ public final class LogicalTypeJsonDeserializer extends StdDeserializer<LogicalTy
case TIMESTAMP_WITH_TIME_ZONE:
case TIMESTAMP_WITH_LOCAL_TIME_ZONE:
return deserializeTimestamp(typeRoot, logicalTypeNode).copy(isNullable);
+ case INTERVAL_DAY_TIME:
+ case INTERVAL_YEAR_MONTH:
+ return deserializeInterval(isNullable, typeRoot, logicalTypeNode);
case MAP:
return deserializeMap(logicalTypeNode).copy(isNullable);
case ARRAY:
@@ -204,6 +211,28 @@ public final class LogicalTypeJsonDeserializer extends StdDeserializer<LogicalTy
}
}
+ private LogicalType deserializeInterval(
+ boolean isNullable, LogicalTypeRoot typeRoot, JsonNode logicalTypeNode) {
+ int precision = logicalTypeNode.get(FIELD_NAME_PRECISION).asInt();
+ switch (typeRoot) {
+ case INTERVAL_YEAR_MONTH:
+ return new YearMonthIntervalType(
+ isNullable,
+ YearMonthIntervalType.YearMonthResolution.valueOf(
+ logicalTypeNode.get(FIELD_NAME_RESOLUTION).asText()),
+ precision);
+ case INTERVAL_DAY_TIME:
+ return new DayTimeIntervalType(
+ isNullable,
+ DayTimeIntervalType.DayTimeResolution.valueOf(
+ logicalTypeNode.get(FIELD_NAME_RESOLUTION).asText()),
+ precision,
+ logicalTypeNode.get(FIELD_NAME_FRACTIONAL_PRECISION).asInt());
+ default:
+ throw new TableException("Interval type root expected.");
+ }
+ }
+
private LogicalType deserializeMap(JsonNode logicalTypeNode) {
JsonNode keyNode = logicalTypeNode.get(FIELD_NAME_KEY_TYPE);
LogicalType keyType = deserializeInternal(keyNode);
diff --git a/flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/rest/serde/LogicalTypeJsonSerializer.java b/flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/rest/serde/LogicalTypeJsonSerializer.java
index 89a7324e12c..f50eecb4086 100644
--- a/flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/rest/serde/LogicalTypeJsonSerializer.java
+++ b/flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/rest/serde/LogicalTypeJsonSerializer.java
@@ -23,6 +23,7 @@ import org.apache.flink.table.api.TableException;
import org.apache.flink.table.types.logical.ArrayType;
import org.apache.flink.table.types.logical.BinaryType;
import org.apache.flink.table.types.logical.CharType;
+import org.apache.flink.table.types.logical.DayTimeIntervalType;
import org.apache.flink.table.types.logical.DecimalType;
import org.apache.flink.table.types.logical.LocalZonedTimestampType;
import org.apache.flink.table.types.logical.LogicalType;
@@ -35,6 +36,7 @@ import org.apache.flink.table.types.logical.TimeType;
import org.apache.flink.table.types.logical.TimestampType;
import org.apache.flink.table.types.logical.VarBinaryType;
import org.apache.flink.table.types.logical.VarCharType;
+import org.apache.flink.table.types.logical.YearMonthIntervalType;
import org.apache.flink.table.types.logical.ZonedTimestampType;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonGenerator;
@@ -64,7 +66,7 @@ public final class LogicalTypeJsonSerializer extends StdSerializer<LogicalType>
// CHAR, VARCHAR, BINARY, VARBINARY
public static final String FIELD_NAME_LENGTH = "length";
- // DECIMAL, TIMESTAMP_WITHOUT_TIME_ZONE, TIMESTAMP_WITH_LOCAL_TIME_ZONE
+ // DECIMAL, TIMESTAMP_WITHOUT_TIME_ZONE, TIMESTAMP_WITH_LOCAL_TIME_ZONE, INTERVAL
public static final String FIELD_NAME_PRECISION = "precision";
// DECIMAL
@@ -87,6 +89,10 @@ public final class LogicalTypeJsonSerializer extends StdSerializer<LogicalType>
public static final String FIELD_NAME_CLASS = "class";
public static final String FIELD_NAME_SERIALIZER = "serializer";
+ // INTERVAL
+ public static final String FIELD_NAME_RESOLUTION = "resolution";
+ public static final String FIELD_NAME_FRACTIONAL_PRECISION = "fractionalPrecision";
+
public LogicalTypeJsonSerializer() {
super(LogicalType.class);
}
@@ -163,6 +169,12 @@ public final class LogicalTypeJsonSerializer extends StdSerializer<LogicalType>
case ROW:
serializeRow((RowType) logicalType, jsonGenerator);
break;
+ case INTERVAL_YEAR_MONTH:
+ serializeYearMonthInterval((YearMonthIntervalType) logicalType, jsonGenerator);
+ break;
+ case INTERVAL_DAY_TIME:
+ serializeDayTimeInterval((DayTimeIntervalType) logicalType, jsonGenerator);
+ break;
case RAW:
if (logicalType instanceof RawType) {
serializeRaw((RawType<?>) logicalType, jsonGenerator);
@@ -207,6 +219,25 @@ public final class LogicalTypeJsonSerializer extends StdSerializer<LogicalType>
}
}
+ private void serializeYearMonthInterval(
+ YearMonthIntervalType yearMonthIntervalType, JsonGenerator jsonGenerator)
+ throws IOException {
+ jsonGenerator.writeNumberField(
+ FIELD_NAME_PRECISION, yearMonthIntervalType.getYearPrecision());
+ jsonGenerator.writeStringField(
+ FIELD_NAME_RESOLUTION, yearMonthIntervalType.getResolution().toString());
+ }
+
+ private void serializeDayTimeInterval(
+ DayTimeIntervalType dayTimeIntervalType, JsonGenerator jsonGenerator)
+ throws IOException {
+ jsonGenerator.writeNumberField(FIELD_NAME_PRECISION, dayTimeIntervalType.getDayPrecision());
+ jsonGenerator.writeNumberField(
+ FIELD_NAME_FRACTIONAL_PRECISION, dayTimeIntervalType.getFractionalPrecision());
+ jsonGenerator.writeStringField(
+ FIELD_NAME_RESOLUTION, dayTimeIntervalType.getResolution().toString());
+ }
+
private void serializeMap(MapType mapType, JsonGenerator jsonGenerator) throws IOException {
jsonGenerator.writeFieldName(FIELD_NAME_KEY_TYPE);
serializeInternal(mapType.getKeyType(), jsonGenerator);
diff --git a/flink-table/flink-sql-gateway/src/test/java/org/apache/flink/table/gateway/rest/serde/LogicalTypeJsonSerDeTest.java b/flink-table/flink-sql-gateway/src/test/java/org/apache/flink/table/gateway/rest/serde/LogicalTypeJsonSerDeTest.java
index a207c5d3538..457af18618a 100644
--- a/flink-table/flink-sql-gateway/src/test/java/org/apache/flink/table/gateway/rest/serde/LogicalTypeJsonSerDeTest.java
+++ b/flink-table/flink-sql-gateway/src/test/java/org/apache/flink/table/gateway/rest/serde/LogicalTypeJsonSerDeTest.java
@@ -21,6 +21,7 @@ package org.apache.flink.table.gateway.rest.serde;
import org.apache.flink.api.common.typeutils.base.LocalDateTimeSerializer;
import org.apache.flink.core.testutils.FlinkAssertions;
import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.planner.plan.nodes.exec.serde.DataTypeJsonSerdeTest;
import org.apache.flink.table.runtime.typeutils.ExternalSerializer;
import org.apache.flink.table.types.logical.ArrayType;
import org.apache.flink.table.types.logical.BigIntType;
@@ -41,12 +42,14 @@ import org.apache.flink.table.types.logical.NullType;
import org.apache.flink.table.types.logical.RawType;
import org.apache.flink.table.types.logical.RowType;
import org.apache.flink.table.types.logical.SmallIntType;
+import org.apache.flink.table.types.logical.StructuredType;
import org.apache.flink.table.types.logical.TimeType;
import org.apache.flink.table.types.logical.TimestampKind;
import org.apache.flink.table.types.logical.TimestampType;
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.YearMonthIntervalType;
import org.apache.flink.table.types.logical.ZonedTimestampType;
import org.apache.flink.types.Row;
@@ -71,13 +74,13 @@ import static org.junit.jupiter.api.parallel.ExecutionMode.CONCURRENT;
/** Tests for {@link LogicalType} serialization and deserialization. */
@Execution(CONCURRENT)
-public class LogicalTypeJsonSerDeTest {
+class LogicalTypeJsonSerDeTest {
private final ObjectMapper mapper = buildObjectMapper();
@ParameterizedTest
@MethodSource("generateTestData")
- public void testLogicalTypeJsonSerDe(LogicalType logicalType) throws IOException {
+ void testLogicalTypeJsonSerDe(LogicalType logicalType) throws IOException {
String json = mapper.writeValueAsString(logicalType);
LogicalType actualType = mapper.readValue(json, LogicalType.class);
@@ -85,9 +88,18 @@ public class LogicalTypeJsonSerDeTest {
}
@Test
- public void testSerializeUnsupportedType() {
+ void testSerializeUnsupportedType() {
LogicalType unsupportedType =
- new DayTimeIntervalType(DayTimeIntervalType.DayTimeResolution.DAY_TO_HOUR);
+ StructuredType.newBuilder(DataTypeJsonSerdeTest.PojoClass.class)
+ .attributes(
+ Arrays.asList(
+ new StructuredType.StructuredAttribute(
+ "f0", new IntType(true)),
+ new StructuredType.StructuredAttribute(
+ "f1", new BigIntType(true)),
+ new StructuredType.StructuredAttribute(
+ "f2", new VarCharType(200), "desc")))
+ .build();
assertThatThrownBy(() -> mapper.writeValueAsString(unsupportedType))
.satisfies(
FlinkAssertions.anyCauseMatches(
@@ -98,8 +110,8 @@ public class LogicalTypeJsonSerDeTest {
}
@Test
- public void testDeserializeUnsupportedType() {
- String unsupportedTypeString = "INTERVAL_DAY_TIME";
+ void testDeserializeUnsupportedType() {
+ String unsupportedTypeString = "STRUCTURED_TYPE";
String json =
String.format(
"{\"%s\": \"%s\", \"%s\": %s}",
@@ -114,7 +126,7 @@ public class LogicalTypeJsonSerDeTest {
}
@Test
- public void testDeserializeUnsupportedJson() {
+ void testDeserializeUnsupportedJson() {
String json = String.format("{\"%s\": \"%s\"}", "unknown", "whatever");
assertThatThrownBy(() -> mapper.readValue(json, LogicalType.class))
.satisfies(
@@ -208,7 +220,27 @@ public class LogicalTypeJsonSerDeTest {
new RawType<>(
Row.class,
ExternalSerializer.of(
- DataTypes.ROW(DataTypes.INT(), DataTypes.STRING()))));
+ DataTypes.ROW(DataTypes.INT(), DataTypes.STRING()))),
+ // interval types
+ new YearMonthIntervalType(YearMonthIntervalType.YearMonthResolution.MONTH),
+ new YearMonthIntervalType(
+ YearMonthIntervalType.YearMonthResolution.YEAR_TO_MONTH),
+ new YearMonthIntervalType(YearMonthIntervalType.YearMonthResolution.YEAR),
+ new DayTimeIntervalType(DayTimeIntervalType.DayTimeResolution.DAY),
+ new DayTimeIntervalType(DayTimeIntervalType.DayTimeResolution.DAY_TO_HOUR),
+ new DayTimeIntervalType(
+ DayTimeIntervalType.DayTimeResolution.DAY_TO_MINUTE),
+ new DayTimeIntervalType(
+ DayTimeIntervalType.DayTimeResolution.DAY_TO_SECOND),
+ new DayTimeIntervalType(DayTimeIntervalType.DayTimeResolution.HOUR),
+ new DayTimeIntervalType(
+ DayTimeIntervalType.DayTimeResolution.HOUR_TO_MINUTE),
+ new DayTimeIntervalType(
+ DayTimeIntervalType.DayTimeResolution.HOUR_TO_SECOND),
+ new DayTimeIntervalType(DayTimeIntervalType.DayTimeResolution.MINUTE),
+ new DayTimeIntervalType(
+ DayTimeIntervalType.DayTimeResolution.MINUTE_TO_SECOND),
+ new DayTimeIntervalType(DayTimeIntervalType.DayTimeResolution.SECOND));
List<LogicalType> testTypes =
Stream.concat(
@@ -230,4 +262,15 @@ public class LogicalTypeJsonSerDeTest {
mapper.registerModule(module);
return mapper;
}
+
+ // --------------------------------------------------------------------------------------------
+ // Helper POJOs
+ // --------------------------------------------------------------------------------------------
+
+ /** Testing class. */
+ public static class PojoClass {
+ public int f0;
+ public long f1;
+ public String f2;
+ }
}
diff --git a/flink-table/flink-sql-gateway/src/test/java/org/apache/flink/table/gateway/rest/serde/ResultInfoJsonSerDeTest.java b/flink-table/flink-sql-gateway/src/test/java/org/apache/flink/table/gateway/rest/serde/ResultInfoJsonSerDeTest.java
index 5cb765be07e..99d1c32a522 100644
--- a/flink-table/flink-sql-gateway/src/test/java/org/apache/flink/table/gateway/rest/serde/ResultInfoJsonSerDeTest.java
+++ b/flink-table/flink-sql-gateway/src/test/java/org/apache/flink/table/gateway/rest/serde/ResultInfoJsonSerDeTest.java
@@ -47,11 +47,14 @@ import java.math.BigDecimal;
import java.net.URL;
import java.nio.charset.StandardCharsets;
import java.sql.Timestamp;
+import java.time.Duration;
import java.time.Instant;
import java.time.LocalDate;
import java.time.LocalDateTime;
import java.time.LocalTime;
+import java.time.Period;
import java.time.ZoneOffset;
+import java.time.temporal.ChronoUnit;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
@@ -67,29 +70,34 @@ import static org.apache.flink.table.api.DataTypes.BIGINT;
import static org.apache.flink.table.api.DataTypes.BOOLEAN;
import static org.apache.flink.table.api.DataTypes.BYTES;
import static org.apache.flink.table.api.DataTypes.DATE;
+import static org.apache.flink.table.api.DataTypes.DAY;
import static org.apache.flink.table.api.DataTypes.DECIMAL;
import static org.apache.flink.table.api.DataTypes.DOUBLE;
import static org.apache.flink.table.api.DataTypes.FIELD;
import static org.apache.flink.table.api.DataTypes.FLOAT;
import static org.apache.flink.table.api.DataTypes.INT;
+import static org.apache.flink.table.api.DataTypes.INTERVAL;
import static org.apache.flink.table.api.DataTypes.MAP;
+import static org.apache.flink.table.api.DataTypes.MONTH;
import static org.apache.flink.table.api.DataTypes.MULTISET;
import static org.apache.flink.table.api.DataTypes.ROW;
+import static org.apache.flink.table.api.DataTypes.SECOND;
import static org.apache.flink.table.api.DataTypes.SMALLINT;
import static org.apache.flink.table.api.DataTypes.STRING;
import static org.apache.flink.table.api.DataTypes.TIME;
import static org.apache.flink.table.api.DataTypes.TIMESTAMP;
import static org.apache.flink.table.api.DataTypes.TIMESTAMP_WITH_LOCAL_TIME_ZONE;
import static org.apache.flink.table.api.DataTypes.TINYINT;
+import static org.apache.flink.table.api.DataTypes.YEAR;
import static org.assertj.core.api.Assertions.assertThat;
/** Tests for {@link ResultInfoSerializer} and {@link ResultInfoDeserializer}. */
-public class ResultInfoJsonSerDeTest {
+class ResultInfoJsonSerDeTest {
private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
private static final Row testRow = initRow();
@BeforeAll
- public static void setUp() {
+ static void setUp() {
SimpleModule simpleModule = new SimpleModule();
simpleModule.addSerializer(ResultInfo.class, new ResultInfoSerializer());
simpleModule.addDeserializer(ResultInfo.class, new ResultInfoDeserializer());
@@ -98,22 +106,22 @@ public class ResultInfoJsonSerDeTest {
@ParameterizedTest
@EnumSource(RowFormat.class)
- public void testResultInfoSerDeWithSingleRow(RowFormat rowFormat) throws Exception {
+ void testResultInfoSerDeWithSingleRow(RowFormat rowFormat) throws Exception {
serDeTest(Collections.singletonList(testRow), rowFormat);
}
@ParameterizedTest
@EnumSource(RowFormat.class)
- public void testResultInfoSerDeWithMultiRowData(RowFormat rowFormat) throws Exception {
+ void testResultInfoSerDeWithMultiRowData(RowFormat rowFormat) throws Exception {
serDeTest(Collections.nCopies(10, testRow), rowFormat);
}
@ParameterizedTest
@EnumSource(RowFormat.class)
- public void testResultInfoSerDeWithNullValues(RowFormat rowFormat) throws Exception {
+ void testResultInfoSerDeWithNullValues(RowFormat rowFormat) throws Exception {
List<Integer> positions =
- IntStream.range(0, 18)
- .mapToObj(i -> new Random().nextInt(18))
+ IntStream.range(0, 20)
+ .mapToObj(i -> new Random().nextInt(20))
.collect(Collectors.toList());
serDeTest(
@@ -123,7 +131,7 @@ public class ResultInfoJsonSerDeTest {
@ParameterizedTest
@ValueSource(strings = {"result_info_json_format.txt", "result_info_plain_text_format.txt"})
- public void testDeserializationFromJson(String fileName) throws Exception {
+ void testDeserializationFromJson(String fileName) throws Exception {
URL url = ResultInfoJsonSerDeTest.class.getClassLoader().getResource(fileName);
String input =
IOUtils.toString(Preconditions.checkNotNull(url), StandardCharsets.UTF_8).trim();
@@ -155,8 +163,8 @@ public class ResultInfoJsonSerDeTest {
ResultInfo resultInfo = OBJECT_MAPPER.readValue(result, ResultInfo.class);
// validate schema
- assertThat(resultInfo.getResultSchema().toString())
- .isEqualTo(testResultInfo.getResultSchema().toString());
+ assertThat(resultInfo.getResultSchema())
+ .hasToString(testResultInfo.getResultSchema().toString());
// validate data
assertDataWithFormat(resultInfo.getData(), rows, rowFormat);
@@ -177,6 +185,8 @@ public class ResultInfoJsonSerDeTest {
final LocalTime time = LocalTime.parse("12:12:43");
final Timestamp timestamp3 = Timestamp.valueOf("1990-10-14 12:12:43.123");
final Timestamp timestamp9 = Timestamp.valueOf("1990-10-14 12:12:43.123456789");
+ final long duration = Duration.of(123, ChronoUnit.HOURS).getSeconds();
+ final int period = Period.of(1, 2, 3).getDays();
final Instant timestampWithLocalZone =
LocalDateTime.of(1990, 10, 14, 12, 12, 43, 123456789)
.atOffset(ZoneOffset.of("Z"))
@@ -193,7 +203,7 @@ public class ResultInfoJsonSerDeTest {
innerMap.put("key", 234);
nestedMap.put("inner_map", innerMap);
- Row testRow = new Row(18);
+ Row testRow = new Row(20);
setRandomKind(testRow);
testRow.setField(0, true);
testRow.setField(1, tinyint);
@@ -213,6 +223,8 @@ public class ResultInfoJsonSerDeTest {
testRow.setField(15, map);
testRow.setField(16, multiSet);
testRow.setField(17, nestedMap);
+ testRow.setField(18, duration);
+ testRow.setField(19, period);
return testRow;
}
@@ -253,7 +265,9 @@ public class ResultInfoJsonSerDeTest {
FIELD("timestampWithLocalZone", TIMESTAMP_WITH_LOCAL_TIME_ZONE(9)),
FIELD("map", MAP(STRING(), BIGINT())),
FIELD("multiSet", MULTISET(STRING())),
- FIELD("map2map", MAP(STRING(), MAP(STRING(), INT()))));
+ FIELD("map2map", MAP(STRING(), MAP(STRING(), INT()))),
+ FIELD("dayTimeInterval", INTERVAL(DAY(), SECOND(3))),
+ FIELD("yearMonthInterval", INTERVAL(YEAR(), MONTH())));
}
private ResolvedSchema getTestResolvedSchema(List<DataTypes.Field> fields) {
diff --git a/flink-table/flink-sql-gateway/src/test/resources/result_info_json_format.txt b/flink-table/flink-sql-gateway/src/test/resources/result_info_json_format.txt
index 486f9a54e66..c0bfdbe840e 100644
--- a/flink-table/flink-sql-gateway/src/test/resources/result_info_json_format.txt
+++ b/flink-table/flink-sql-gateway/src/test/resources/result_info_json_format.txt
@@ -1 +1 @@
-{"columns":[{"name":"bool","logicalType":{"type":"BOOLEAN","nullable":true},"comment":null},{"name":"tinyint","logicalType":{"type":"TINYINT","nullable":true},"comment":null},{"name":"smallint","logicalType":{"type":"SMALLINT","nullable":true},"comment":null},{"name":"int","logicalType":{"type":"INTEGER","nullable":true},"comment":null},{"name":"bigint","logicalType":{"type":"BIGINT","nullable":true},"comment":null},{"name":"float","logicalType":{"type":"FLOAT","nullable":true},"comment" [...]
+{"columns":[{"name":"bool","logicalType":{"type":"BOOLEAN","nullable":true},"comment":null},{"name":"tinyint","logicalType":{"type":"TINYINT","nullable":true},"comment":null},{"name":"smallint","logicalType":{"type":"SMALLINT","nullable":true},"comment":null},{"name":"int","logicalType":{"type":"INTEGER","nullable":true},"comment":null},{"name":"bigint","logicalType":{"type":"BIGINT","nullable":true},"comment":null},{"name":"float","logicalType":{"type":"FLOAT","nullable":true},"comment" [...]
diff --git a/flink-table/flink-sql-gateway/src/test/resources/result_info_plain_text_format.txt b/flink-table/flink-sql-gateway/src/test/resources/result_info_plain_text_format.txt
index a640abfc9cc..1cc10d6ecae 100644
--- a/flink-table/flink-sql-gateway/src/test/resources/result_info_plain_text_format.txt
+++ b/flink-table/flink-sql-gateway/src/test/resources/result_info_plain_text_format.txt
@@ -1 +1 @@
-{"columns":[{"name":"bool","logicalType":{"type":"BOOLEAN","nullable":true},"comment":null},{"name":"tinyint","logicalType":{"type":"TINYINT","nullable":true},"comment":null},{"name":"smallint","logicalType":{"type":"SMALLINT","nullable":true},"comment":null},{"name":"int","logicalType":{"type":"INTEGER","nullable":true},"comment":null},{"name":"bigint","logicalType":{"type":"BIGINT","nullable":true},"comment":null},{"name":"float","logicalType":{"type":"FLOAT","nullable":true},"comment" [...]
+{"columns":[{"name":"bool","logicalType":{"type":"BOOLEAN","nullable":true},"comment":null},{"name":"tinyint","logicalType":{"type":"TINYINT","nullable":true},"comment":null},{"name":"smallint","logicalType":{"type":"SMALLINT","nullable":true},"comment":null},{"name":"int","logicalType":{"type":"INTEGER","nullable":true},"comment":null},{"name":"bigint","logicalType":{"type":"BIGINT","nullable":true},"comment":null},{"name":"float","logicalType":{"type":"FLOAT","nullable":true},"comment" [...]