[FLINK-28936][sql-gateway] Fix REST endpoint can not serialize CHAR(0) by introducing a new serializer for LogicalType #20617

yuzelin · 2022-08-18T08:14:12Z

[FLINK-28936][sql-gateway] Fix REST endpoint can not serialize CHAR(0) by introducing a new serializer for LogicalType

What is the purpose of the change

Fix the absence of LogicalType serializer.

Brief change log

Add LogicalTypeJsonSerializer in sql-gateway-api.
Add LogicalTypeJsonDeserializer in sql-gateway-api.

Verifying this change

This change added tests and can be verified as follows:

Added test prove that a LogicalType instance after serializing and deserializing is equal to original logical type.

Does this pull request potentially affect one of the following parts:

Dependencies (does it add or upgrade a dependency): (yes / no)
The public API, i.e., is any changed class annotated with @Public(Evolving): (yes / no)
The serializers: (yes / no / don't know)
The runtime per-record code paths (performance sensitive): (yes / no / don't know)
Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: (yes / no / don't know)
The S3 file system connector: (yes / no / don't know)

Documentation

Does this pull request introduce a new feature? (yes / no)
If yes, how is the feature documented? (not applicable / docs / JavaDocs / not documented)

flinkbot · 2022-08-18T08:25:49Z

CI report:

05f1385 Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run azure re-run the last Azure build

fsk119

Thanks for your contribution. I left some comments.

.../main/java/org/apache/flink/table/gateway/api/results/serde/LogicalTypeJsonDeserializer.java

fsk119 · 2022-08-22T03:18:50Z

...rc/main/java/org/apache/flink/table/gateway/api/results/serde/LogicalTypeJsonSerializer.java

+
+    // Common fields
+    public static final String FIELD_NAME_TYPE_NAME = "type";
+    public static final String FIELD_NAME_NULLABLE = "nullable";


miss FIELD_NAME_DESCRPITION

As far as I know, description is not a common field shared by all logical types, so I think we don't need this.

...src/test/java/org/apache/flink/table/gateway/api/results/serde/LogicalTypeJsonSerDeTest.java

fsk119 · 2022-08-22T04:01:02Z

...rc/main/java/org/apache/flink/table/gateway/api/results/serde/LogicalTypeJsonSerializer.java

+    // ROW
+    public static final String FIELD_NAME_FIELDS = "fields";
+    public static final String FIELD_NAME_FIELD_NAME = "name";
+    public static final String FIELD_NAME_FIELD_TYPE = "fieldType";


missing description.

Added code for serialization and deserialization for RowType#description and corresponding test.

fsk119 · 2022-08-22T06:20:20Z

...rc/main/java/org/apache/flink/table/gateway/api/results/serde/LogicalTypeJsonSerializer.java

+            case FLOAT:
+            case DOUBLE:
+            case DATE:
+            case NULL:


NullType is alway nullable. I think we dont't need to add "nullable": true in the json.

Added codes to handle NullType.

...rc/main/java/org/apache/flink/table/gateway/api/results/serde/LogicalTypeJsonSerializer.java

fsk119 · 2022-08-22T06:32:55Z

...src/test/java/org/apache/flink/table/gateway/api/results/serde/LogicalTypeJsonSerDeTest.java

+
+    private static List<LogicalType> generateTestData() {
+        List<LogicalType> types =
+                Arrays.asList(


Please also add

new TimeType(), new TimeType(3), new LocalZonedTimestampType(false, TimestampKind.PROCTIME, 3), new TimestampType(false, TimestampKind.ROWTIME, 3), new ZonedTimestampType(), new ZonedTimestampType(3), new ZonedTimestampType(false, TimestampKind.ROWTIME, 3), new LocalZonedTimestampType(), new LocalZonedTimestampType(3), new LocalZonedTimestampType(false, TimestampKind.PROCTIME, 3), new LocalZonedTimestampType(false, TimestampKind.ROWTIME, 3),

Add necessary test case.

.../main/java/org/apache/flink/table/gateway/api/results/serde/LogicalTypeJsonDeserializer.java

fsk119 · 2022-08-22T06:38:36Z

.../main/java/org/apache/flink/table/gateway/api/results/serde/LogicalTypeJsonDeserializer.java

+            case VARBINARY:
+                return length == 0 ? VarBinaryType.ofEmptyLiteral() : new VarBinaryType(length);
+            default:
+                throw new TableException("Logical type with attribute 'length' expected.");


"Can't convert json '%s' to the logical type %s."

BTW, I think it's better to throw SqlGatewayException here.

fsk119 · 2022-08-22T06:48:50Z

.../main/java/org/apache/flink/table/gateway/api/results/serde/LogicalTypeJsonDeserializer.java

+
+    private LogicalType deserializeRaw(JsonNode logicalTypeNode) {
+        String className = logicalTypeNode.get(FIELD_NAME_CLASS).asText();
+        String specialSerializer = logicalTypeNode.get(FIELD_NAME_SERIALIZER).asText();


rename to serializer. We don't have special serializer.

fsk119

Thanks for your update. I left some comments.

.../src/test/java/org/apache/flink/table/gateway/rest/util/SqlGatewayRestEndpointExtension.java

fsk119 · 2022-08-24T06:27:04Z

.../src/test/java/org/apache/flink/table/gateway/rest/util/SqlGatewayRestEndpointExtension.java

+
+        InetSocketAddress serverAddress = checkNotNull(sqlGatewayRestEndpoint.getServerAddress());
+        targetAddress = serverAddress.getHostName();
+        targetPort = serverAddress.getPort();


Can we use NetUtils to get a random port? If other CI runs in the same machine, it may fail to bind the port.

After checking the implementation, I found the port is 0 in default configuration used by sqlGatewayRestEndpoint.start(), and that means server would bind a random port.

fsk119 · 2022-08-24T06:41:02Z

.../main/java/org/apache/flink/table/gateway/api/results/serde/LogicalTypeJsonDeserializer.java

    private LogicalType deserializeInternal(JsonNode logicalTypeNode) {
        LogicalTypeRoot typeRoot =
                LogicalTypeRoot.valueOf(logicalTypeNode.get(FIELD_NAME_TYPE_NAME).asText());
+        // handle the special case of NullType


Can we describe the root reason here. For example, the NullType is always nullable.

Made it clearer.

fsk119 · 2022-08-24T06:52:17Z

...src/test/java/org/apache/flink/table/gateway/api/results/serde/LogicalTypeJsonSerDeTest.java

+    // final constants for testing unsupported case
+    private final LogicalType unsupportedType =
+            new DayTimeIntervalType(DayTimeIntervalType.DayTimeResolution.DAY_TO_HOUR);
+    private final String serializerExceptionMessageFormat =
+            "Unable to serialize logical type '%s'. Please check the documentation for supported types.";
+    private final String unsupportedTypeString = "INTERVAL_DAY_TIME";
+    private final String json =
+            String.format(
+                    "{\"%s\": \"%s\", \"%s\": %s}",
+                    "type", unsupportedTypeString, "nullable", "true");
+    private final String deserializerExceptionMessageFormat =
+            "Unable to deserialize a logical type of type root '%s'. Please check the documentation for supported types.";


Local variables are enough. Move into the test case.

fsk119 · 2022-08-24T06:53:19Z

...src/test/java/org/apache/flink/table/gateway/api/results/serde/LogicalTypeJsonSerDeTest.java

+
+    private final ObjectMapper mapper = getObjectMapper();
+
+    // final constants for testing unsupported case


Remove useless comments... I think variable names are straightforward for us.

fsk119 · 2022-08-24T06:55:27Z

...src/test/java/org/apache/flink/table/gateway/api/results/serde/LogicalTypeJsonSerDeTest.java

+    /**
+     * LogicalType isn't annotated with Jackson annotations, so it's necessary to register the
+     * customer serializer and deserializer when testing LogicalType Serde alone.
+     */


Remove this comment. It's the Jackson's API, we don't need to tell others how to use jackson.

fsk119 · 2022-08-24T06:55:41Z

...src/test/java/org/apache/flink/table/gateway/api/results/serde/LogicalTypeJsonSerDeTest.java

+     * LogicalType isn't annotated with Jackson annotations, so it's necessary to register the
+     * customer serializer and deserializer when testing LogicalType Serde alone.
+     */
+    private ObjectMapper getObjectMapper() {


rename to buildObjectMapper

fsk119 · 2022-08-24T07:13:20Z

...src/test/java/org/apache/flink/table/gateway/api/results/serde/LogicalTypeJsonSerDeTest.java

                        new LocalZonedTimestampType(),
                        new LocalZonedTimestampType(3),
                        new LocalZonedTimestampType(false, 3),
+                        // LocalZonedTimestampType#eaquals doesn't compare TimestampKind


Remove this ...

fsk119 · 2022-08-24T07:13:53Z

...src/test/java/org/apache/flink/table/gateway/api/results/serde/LogicalTypeJsonSerDeTest.java

+        // test to deserialize unsupported JSON string
+        String unsupportedTypeString = "INTERVAL_DAY_TIME";
+        String json =
+                String.format(
+                        "{\"%s\": \"%s\", \"%s\": %s}",
+                        "type", unsupportedTypeString, "nullable", "true");
+        assertThatThrownBy(() -> mapper.readValue(json, LogicalType.class))
+                .satisfies(
+                        FlinkAssertions.anyCauseMatches(
+                                UnsupportedOperationException.class,
+                                String.format(
+                                        "Unable to deserialize a logical type of type root '%s'. Please check the documentation for supported types.",
+                                        unsupportedTypeString)));


It's better to split into two test cases.

…) by introducing a new serializer for LogicalType

…ame to SqlGatewayRestEndpointStatementITCase).

fsk119

Thanks for your update. LGTM

…acted LogicalType (apache#20617)

fsk119 self-requested a review August 18, 2022 08:15

fsk119 reviewed Aug 22, 2022

View reviewed changes

yuzelin force-pushed the LogicalTypeSerDe branch 2 times, most recently from 2ed8067 to 5c4fe03 Compare August 23, 2022 12:34

fsk119 self-requested a review August 24, 2022 02:03

yuzelin force-pushed the LogicalTypeSerDe branch 2 times, most recently from 9d127f6 to 5988c33 Compare August 24, 2022 02:41

fsk119 reviewed Aug 24, 2022

View reviewed changes

yuzelin added 7 commits August 24, 2022 20:06

[FLINK-28936][sql-gateway] Fix RSET endpoint can not serialize CHAR(0…

be68cdf

…) by introducing a new serializer for LogicalType

[fix] fix rule violations.

5fff82b

[fix] fix some problems.

a9c69ad

[fix] minor changes according to the review comments.

1c28bd1

[fix] fix bugs and slightly improve unstable StatementCaseITTest (ren…

e8c8591

…ame to SqlGatewayRestEndpointStatementITCase).

[fix] improve codes according to the review comment.

ca2a7db

[fix] fix a improper use of method.

05f1385

yuzelin force-pushed the LogicalTypeSerDe branch from 6fef77c to 05f1385 Compare August 24, 2022 12:23

fsk119 approved these changes Aug 25, 2022

View reviewed changes

fsk119 merged commit 64f11ee into apache:master Aug 25, 2022

yuzelin deleted the LogicalTypeSerDe branch August 26, 2022 03:30

huangxiaofeng10047 pushed a commit to huangxiaofeng10047/flink that referenced this pull request Nov 3, 2022

[FLINK-28936][sql-gateway] Fix REST endpoint can not serialize uncomp…

b4fe7d5

…acted LogicalType (apache#20617)


		private final ObjectMapper mapper = getObjectMapper();

		// final constants for testing unsupported case

[FLINK-28936][sql-gateway] Fix REST endpoint can not serialize CHAR(0) by introducing a new serializer for LogicalType #20617

[FLINK-28936][sql-gateway] Fix REST endpoint can not serialize CHAR(0) by introducing a new serializer for LogicalType #20617

Uh oh!

Conversation

yuzelin commented Aug 18, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What is the purpose of the change

Brief change log

Verifying this change

Does this pull request potentially affect one of the following parts:

Documentation

Uh oh!

flinkbot commented Aug 18, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

CI report:

Uh oh!

fsk119 left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

yuzelin Aug 22, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

fsk119 left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

yuzelin Aug 24, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

yuzelin commented Aug 18, 2022 •

edited

Loading

flinkbot commented Aug 18, 2022 •

edited

Loading

yuzelin Aug 22, 2022 •

edited

Loading

yuzelin Aug 24, 2022 •

edited

Loading