You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by GitBox <gi...@apache.org> on 2022/08/04 09:05:16 UTC
[GitHub] [flink] WencongLiu opened a new pull request, #20451: [FLINK-28162][sql-gateway]Introduce the operation related API for REST endpoint
WencongLiu opened a new pull request, #20451:
URL: https://github.com/apache/flink/pull/20451
<!--
*Thank you very much for contributing to Apache Flink - we are happy that you want to help us improve Flink. To help the community review your contribution in the best possible way, please go through the checklist below, which will get the contribution into a shape in which it can be best reviewed.*
*Please understand that we do not do this to make contributions to Flink a hassle. In order to uphold a high standard of quality for code contributions, while at the same time managing a large number of contributions, we need contributors to prepare the contributions well, and give reviewers enough contextual information for the review. Please also understand that contributions that do not follow this guide will take longer to review and thus typically be picked up with lower priority by the community.*
## Contribution Checklist
- Make sure that the pull request corresponds to a [JIRA issue](https://issues.apache.org/jira/projects/FLINK/issues). Exceptions are made for typos in JavaDoc or documentation files, which need no JIRA issue.
- Name the pull request in the form "[FLINK-XXXX] [component] Title of the pull request", where *FLINK-XXXX* should be replaced by the actual issue number. Skip *component* if you are unsure about which is the best component.
Typo fixes that have no associated JIRA issue should be named following this pattern: `[hotfix] [docs] Fix typo in event time introduction` or `[hotfix] [javadocs] Expand JavaDoc for PuncuatedWatermarkGenerator`.
- Fill out the template below to describe the changes contributed by the pull request. That will give reviewers the context they need to do the review.
- Make sure that the change passes the automated tests, i.e., `mvn clean verify` passes. You can set up Azure Pipelines CI to do that following [this guide](https://cwiki.apache.org/confluence/display/FLINK/Azure+Pipelines#AzurePipelines-Tutorial:SettingupAzurePipelinesforaforkoftheFlinkrepository).
- Each pull request should address only one issue, not mix up code from multiple issues.
- Each commit in the pull request has a meaningful commit message (including the JIRA id)
- Once all items of the checklist are addressed, remove the above text and this checklist, leaving only the filled out template below.
**(The sections below can be removed for hotfixes of typos)**
-->
## What is the purpose of the change
*Introduce the operation related api in sql gateway*
## Brief change log
- *Introduce operation related handlers and message headers*
## Verifying this change
This change added tests and can be verified as follows:
- *Added integration tests for operation related apis*
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): (no)
- The public API, i.e., is any changed class annotated with `@Public(Evolving)`: (no)
- The serializers: (no )
- The runtime per-record code paths (performance sensitive): (no)
- Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: (no)
- The S3 file system connector: (no)
## Documentation
- Does this pull request introduce a new feature? (yes)
- If yes, how is the feature documented? (not documented)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] WencongLiu commented on a diff in pull request #20451: [FLINK-28162][sql-gateway]Introduce the operation related API for REST endpoint
Posted by GitBox <gi...@apache.org>.
WencongLiu commented on code in PR #20451:
URL: https://github.com/apache/flink/pull/20451#discussion_r938461762
##########
flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/rest/message/statement/ExecuteStatementRequestBody.java:
##########
@@ -0,0 +1,59 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.gateway.rest.message.statement;
+
+import org.apache.flink.runtime.rest.messages.RequestBody;
+
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonInclude;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
+
+import javax.annotation.Nullable;
+
+/** {@link RequestBody} for execute a statement. */
+@JsonInclude(JsonInclude.Include.NON_NULL)
+public class ExecuteStatementRequestBody implements RequestBody {
+
+ private static final String FIELD_NAME_STATEMENT = "statement";
+ private static final String FIELD_NAME_EXECUTION_TIMEOUT = "execution_timeout";
+
+ @JsonProperty(FIELD_NAME_STATEMENT)
+ @Nullable
+ private final String statement;
Review Comment:
Fixed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] xintongsong commented on a diff in pull request #20451: [FLINK-28162][sql-gateway]Introduce the operation related API for REST endpoint
Posted by GitBox <gi...@apache.org>.
xintongsong commented on code in PR #20451:
URL: https://github.com/apache/flink/pull/20451#discussion_r938390193
##########
flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/rest/message/statement/FetchResultsRequestBody.java:
##########
@@ -0,0 +1,47 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.gateway.rest.message.statement;
+
+import org.apache.flink.runtime.rest.messages.RequestBody;
+
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonInclude;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
+
+import javax.annotation.Nullable;
+
+/** {@link RequestBody} for fetching job result. */
+@JsonInclude(JsonInclude.Include.NON_NULL)
+public class FetchResultsRequestBody implements RequestBody {
+
+ private static final String FIELD_NAME_MAX_FETCH_SIZE = "max_fetch_size";
Review Comment:
I agree that it would be nice to use consistent naming conventions for the rest field in 2.0, and we should start to do that early. Is there already a consensus that camelCase will be used in long term? Could you point us to it?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] WencongLiu commented on a diff in pull request #20451: [FLINK-28162][sql-gateway]Introduce the operation related API for REST endpoint
Posted by GitBox <gi...@apache.org>.
WencongLiu commented on code in PR #20451:
URL: https://github.com/apache/flink/pull/20451#discussion_r938471294
##########
flink-table/flink-sql-gateway-api/src/main/java/org/apache/flink/table/gateway/api/results/ResultSet.java:
##########
@@ -23,16 +23,30 @@
import org.apache.flink.table.catalog.ResolvedSchema;
import org.apache.flink.table.data.RowData;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.annotation.JsonDeserialize;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.annotation.JsonSerialize;
+
import javax.annotation.Nullable;
+import java.io.Serializable;
import java.util.Collections;
import java.util.List;
import java.util.Objects;
import java.util.stream.Collectors;
/** The collection of the results. */
@PublicEvolving
-public class ResultSet {
+@JsonSerialize(using = ResultSetJsonSerializer.class)
+@JsonDeserialize(using = ResultSetJsonDeserializer.class)
+public class ResultSet implements Serializable {
+
+ public static final String FIELD_NAME_RESULT_TYPE = "result_type";
Review Comment:
Removed.
##########
flink-table/flink-sql-gateway-api/src/main/java/org/apache/flink/table/gateway/api/results/ResultSet.java:
##########
@@ -23,16 +23,30 @@
import org.apache.flink.table.catalog.ResolvedSchema;
import org.apache.flink.table.data.RowData;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.annotation.JsonDeserialize;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.annotation.JsonSerialize;
+
import javax.annotation.Nullable;
+import java.io.Serializable;
import java.util.Collections;
import java.util.List;
import java.util.Objects;
import java.util.stream.Collectors;
/** The collection of the results. */
@PublicEvolving
-public class ResultSet {
+@JsonSerialize(using = ResultSetJsonSerializer.class)
+@JsonDeserialize(using = ResultSetJsonDeserializer.class)
+public class ResultSet implements Serializable {
+
+ public static final String FIELD_NAME_RESULT_TYPE = "result_type";
+
+ public static final String FIELD_NAME_COLUMN_INFOS = "columns";
+
+ public static final String FIELD_NAME_DATA = "data";
+
+ public static final String FIELD_NAME_NEXT_TOKEN = "next_token";
Review Comment:
Removed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] WencongLiu commented on a diff in pull request #20451: [FLINK-28162][sql-gateway]Introduce the operation related API for REST endpoint
Posted by GitBox <gi...@apache.org>.
WencongLiu commented on code in PR #20451:
URL: https://github.com/apache/flink/pull/20451#discussion_r939501511
##########
flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/rest/message/statement/FetchResultsRequestBody.java:
##########
@@ -0,0 +1,47 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.gateway.rest.message.statement;
+
+import org.apache.flink.runtime.rest.messages.RequestBody;
+
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonInclude;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
+
+import javax.annotation.Nullable;
+
+/** {@link RequestBody} for fetching job result. */
+@JsonInclude(JsonInclude.Include.NON_NULL)
+public class FetchResultsRequestBody implements RequestBody {
+
+ private static final String FIELD_NAME_MAX_FETCH_SIZE = "max_fetch_size";
+
+ @JsonProperty(FIELD_NAME_MAX_FETCH_SIZE)
+ @Nullable
+ private final Integer maxFetchSize;
Review Comment:
This field has been deleted.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] WencongLiu commented on a diff in pull request #20451: [FLINK-28162][sql-gateway]Introduce the operation related API for REST endpoint
Posted by GitBox <gi...@apache.org>.
WencongLiu commented on code in PR #20451:
URL: https://github.com/apache/flink/pull/20451#discussion_r938468545
##########
flink-table/flink-sql-gateway-api/src/main/java/org/apache/flink/table/gateway/api/results/ExceptionInfo.java:
##########
@@ -0,0 +1,64 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.gateway.api.results;
+
+import org.apache.flink.util.Preconditions;
+
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
+
+/** ExceptionInfo for SQL gateway service. */
+public class ExceptionInfo {
+
+ private static final String FIELD_EXCEPTION_ROOT_CLASS_NAME = "root";
Review Comment:
Fixed.
The ExceptionInfo contains the same attributes as the FLIP.
"exception": {
"rootCause": "....",
"exceptionStack": "..."
}
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] WencongLiu commented on a diff in pull request #20451: [FLINK-28162][sql-gateway]Introduce the operation related API for REST endpoint
Posted by GitBox <gi...@apache.org>.
WencongLiu commented on code in PR #20451:
URL: https://github.com/apache/flink/pull/20451#discussion_r939497965
##########
flink-table/flink-sql-gateway-api/src/main/java/org/apache/flink/table/gateway/api/results/ResultSetJsonSerializer.java:
##########
@@ -0,0 +1,89 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.gateway.api.results;
+
+import org.apache.flink.formats.common.TimestampFormat;
+import org.apache.flink.formats.json.JsonFormatOptions;
+import org.apache.flink.formats.json.JsonRowDataSerializationSchema;
+import org.apache.flink.table.catalog.Column;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.types.logical.RowType;
+
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonGenerator;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.JsonSerializer;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.SerializerProvider;
+
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.apache.flink.table.gateway.api.results.ResultSet.FIELD_NAME_COLUMN_INFOS;
+import static org.apache.flink.table.gateway.api.results.ResultSet.FIELD_NAME_DATA;
+import static org.apache.flink.table.gateway.api.results.ResultSet.FIELD_NAME_NEXT_TOKEN;
+import static org.apache.flink.table.gateway.api.results.ResultSet.FIELD_NAME_RESULT_TYPE;
+
+/** Json serializer for {@link ResultSet}. */
+public class ResultSetJsonSerializer extends JsonSerializer<ResultSet> {
+
+ @Override
+ public void serialize(
+ ResultSet resultSet, JsonGenerator jsonGenerator, SerializerProvider serializerProvider)
+ throws IOException {
+ jsonGenerator.writeStartObject();
+
+ // Serialize result type
+ serializerProvider.defaultSerializeField(
+ FIELD_NAME_RESULT_TYPE, resultSet.getResultType(), jsonGenerator);
+
+ // Serialize next token
+ serializerProvider.defaultSerializeField(
+ FIELD_NAME_NEXT_TOKEN, resultSet.getNextToken(), jsonGenerator);
+
+ // Serialize column infos
+ List<Column> columns = resultSet.getResultSchema().getColumns();
+ List<ColumnInfo> columnInfos = new ArrayList<>();
+ for (Column column : columns) {
+ columnInfos.add(
+ ColumnInfo.create(column.getName(), column.getDataType().getLogicalType()));
+ }
+ serializerProvider.defaultSerializeField(
+ FIELD_NAME_COLUMN_INFOS, columnInfos, jsonGenerator);
+
+ // Serialize RowData
+ JsonRowDataSerializationSchema serializationSchema =
+ new JsonRowDataSerializationSchema(
+ (RowType)
+ resultSet
+ .getResultSchema()
+ .toPhysicalRowDataType()
+ .getLogicalType(),
+ TimestampFormat.ISO_8601,
+ JsonFormatOptions.MapNullKeyMode.LITERAL,
+ "null",
+ true);
+ List<String> rowDataStrings = new ArrayList<>();
+ for (RowData row : resultSet.getData()) {
+ byte[] rowByte = serializationSchema.serialize(row);
+ rowDataStrings.add(new String(rowByte, StandardCharsets.UTF_8));
+ }
Review Comment:
Fixed.
Now `data` contains a list of Json Objects. Each of Json Object contains a RowKind and a list of JsonNode.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] WencongLiu commented on a diff in pull request #20451: [FLINK-28162][sql-gateway]Introduce the operation related API for REST endpoint
Posted by GitBox <gi...@apache.org>.
WencongLiu commented on code in PR #20451:
URL: https://github.com/apache/flink/pull/20451#discussion_r939498653
##########
flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/rest/message/statement/ExecuteStatementResponseBody.java:
##########
@@ -0,0 +1,53 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.gateway.rest.message.statement;
+
+import org.apache.flink.runtime.rest.messages.ResponseBody;
+import org.apache.flink.table.gateway.api.operation.OperationType;
+
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonInclude;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
+
+/** {@link ResponseBody} for execute a statement. */
+@JsonInclude(JsonInclude.Include.NON_NULL)
+public class ExecuteStatementResponseBody implements ResponseBody {
+
+ private static final String FIELD_OPERATION_HANDLE = "operation_handle";
+
+ private static final String FIELD_OPERATION_TYPE = "operation_type";
+
+ @JsonProperty(FIELD_OPERATION_HANDLE)
+ private final String operationHandle;
+
+ @JsonProperty(FIELD_OPERATION_TYPE)
+ private final String operationTYpe = OperationType.EXECUTE_STATEMENT.name();
+
+ public ExecuteStatementResponseBody(
Review Comment:
The field `has_result` will be removed in FLIP.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] WencongLiu commented on a diff in pull request #20451: [FLINK-28162][sql-gateway]Introduce the operation related API for REST endpoint
Posted by GitBox <gi...@apache.org>.
WencongLiu commented on code in PR #20451:
URL: https://github.com/apache/flink/pull/20451#discussion_r938489323
##########
flink-table/flink-sql-gateway-api/src/main/java/org/apache/flink/table/gateway/api/results/ResultSetJsonSerializer.java:
##########
@@ -0,0 +1,89 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.gateway.api.results;
+
+import org.apache.flink.formats.common.TimestampFormat;
+import org.apache.flink.formats.json.JsonFormatOptions;
+import org.apache.flink.formats.json.JsonRowDataSerializationSchema;
+import org.apache.flink.table.catalog.Column;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.types.logical.RowType;
+
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonGenerator;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.JsonSerializer;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.SerializerProvider;
+
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.apache.flink.table.gateway.api.results.ResultSet.FIELD_NAME_COLUMN_INFOS;
+import static org.apache.flink.table.gateway.api.results.ResultSet.FIELD_NAME_DATA;
+import static org.apache.flink.table.gateway.api.results.ResultSet.FIELD_NAME_NEXT_TOKEN;
+import static org.apache.flink.table.gateway.api.results.ResultSet.FIELD_NAME_RESULT_TYPE;
+
+/** Json serializer for {@link ResultSet}. */
+public class ResultSetJsonSerializer extends JsonSerializer<ResultSet> {
+
+ @Override
+ public void serialize(
+ ResultSet resultSet, JsonGenerator jsonGenerator, SerializerProvider serializerProvider)
+ throws IOException {
+ jsonGenerator.writeStartObject();
+
+ // Serialize result type
+ serializerProvider.defaultSerializeField(
+ FIELD_NAME_RESULT_TYPE, resultSet.getResultType(), jsonGenerator);
+
+ // Serialize next token
+ serializerProvider.defaultSerializeField(
+ FIELD_NAME_NEXT_TOKEN, resultSet.getNextToken(), jsonGenerator);
Review Comment:
Removed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] WencongLiu commented on a diff in pull request #20451: [FLINK-28162][sql-gateway]Introduce the operation related API for REST endpoint
Posted by GitBox <gi...@apache.org>.
WencongLiu commented on code in PR #20451:
URL: https://github.com/apache/flink/pull/20451#discussion_r939497733
##########
flink-table/flink-sql-gateway-api/pom.xml:
##########
@@ -64,7 +64,19 @@
<version>${project.version}</version>
<scope>test</scope>
</dependency>
- </dependencies>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-json</artifactId>
+ <version>1.16-SNAPSHOT</version>
Review Comment:
Fixed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] WencongLiu commented on a diff in pull request #20451: [FLINK-28162][sql-gateway]Introduce the operation related API for REST endpoint
Posted by GitBox <gi...@apache.org>.
WencongLiu commented on code in PR #20451:
URL: https://github.com/apache/flink/pull/20451#discussion_r938462535
##########
flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/rest/message/statement/ExecuteStatementResponseBody.java:
##########
@@ -0,0 +1,53 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.gateway.rest.message.statement;
+
+import org.apache.flink.runtime.rest.messages.ResponseBody;
+import org.apache.flink.table.gateway.api.operation.OperationType;
+
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonInclude;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
+
+/** {@link ResponseBody} for execute a statement. */
+@JsonInclude(JsonInclude.Include.NON_NULL)
+public class ExecuteStatementResponseBody implements ResponseBody {
+
+ private static final String FIELD_OPERATION_HANDLE = "operation_handle";
+
+ private static final String FIELD_OPERATION_TYPE = "operation_type";
+
+ @JsonProperty(FIELD_OPERATION_HANDLE)
+ private final String operationHandle;
+
+ @JsonProperty(FIELD_OPERATION_TYPE)
+ private final String operationTYpe = OperationType.EXECUTE_STATEMENT.name();
Review Comment:
Fixed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] xintongsong commented on a diff in pull request #20451: [FLINK-28162][sql-gateway]Introduce the operation related API for REST endpoint
Posted by GitBox <gi...@apache.org>.
xintongsong commented on code in PR #20451:
URL: https://github.com/apache/flink/pull/20451#discussion_r937556239
##########
flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/rest/handler/operation/AbstractOperationHandler.java:
##########
@@ -0,0 +1,91 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.gateway.rest.handler.operation;
+
+import org.apache.flink.runtime.rest.handler.HandlerRequest;
+import org.apache.flink.runtime.rest.handler.RestHandlerException;
+import org.apache.flink.runtime.rest.messages.EmptyRequestBody;
+import org.apache.flink.runtime.rest.messages.MessageHeaders;
+import org.apache.flink.table.gateway.api.SqlGatewayService;
+import org.apache.flink.table.gateway.api.operation.OperationHandle;
+import org.apache.flink.table.gateway.api.operation.OperationStatus;
+import org.apache.flink.table.gateway.api.session.SessionHandle;
+import org.apache.flink.table.gateway.api.utils.SqlGatewayException;
+import org.apache.flink.table.gateway.rest.handler.AbstractSqlGatewayRestHandler;
+import org.apache.flink.table.gateway.rest.message.operation.OperationHandleIdPathParameter;
+import org.apache.flink.table.gateway.rest.message.operation.OperationMessageParameters;
+import org.apache.flink.table.gateway.rest.message.operation.OperationStatusResponseBody;
+import org.apache.flink.table.gateway.rest.message.session.SessionHandleIdPathParameter;
+import org.apache.flink.table.gateway.rest.util.SqlGatewayRestAPIVersion;
+
+import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpResponseStatus;
+
+import javax.annotation.Nonnull;
+
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+
+/** Abstract operation handler. */
+public abstract class AbstractOperationHandler
+ extends AbstractSqlGatewayRestHandler<
+ EmptyRequestBody, OperationStatusResponseBody, OperationMessageParameters> {
+
+ public AbstractOperationHandler(
+ SqlGatewayService service,
+ Map<String, String> responseHeaders,
+ MessageHeaders<
+ EmptyRequestBody,
+ OperationStatusResponseBody,
+ OperationMessageParameters>
+ messageHeaders) {
+ super(service, responseHeaders, messageHeaders);
+ }
+
+ private String getStatus(SessionHandle sessionHandle, OperationHandle operationHandle) {
+ if (execute(sessionHandle, operationHandle)) {
+ return this.service
+ .getOperationInfo(sessionHandle, operationHandle)
+ .getStatus()
+ .toString();
+ } else {
+ return OperationStatus.CLOSED.toString();
+ }
+ }
+
+ boolean execute(SessionHandle sessionHandle, OperationHandle operationHandle) {
Review Comment:
What is the semantic of the return value?
##########
flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/rest/message/statement/ExecuteStatementRequestBody.java:
##########
@@ -0,0 +1,59 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.gateway.rest.message.statement;
+
+import org.apache.flink.runtime.rest.messages.RequestBody;
+
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonInclude;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
+
+import javax.annotation.Nullable;
+
+/** {@link RequestBody} for execute a statement. */
+@JsonInclude(JsonInclude.Include.NON_NULL)
+public class ExecuteStatementRequestBody implements RequestBody {
+
+ private static final String FIELD_NAME_STATEMENT = "statement";
+ private static final String FIELD_NAME_EXECUTION_TIMEOUT = "execution_timeout";
+
+ @JsonProperty(FIELD_NAME_STATEMENT)
+ @Nullable
+ private final String statement;
Review Comment:
This should not be `@Nullable`. It's required according to the FLIP.
##########
flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/rest/message/statement/ExecuteStatementResponseBody.java:
##########
@@ -0,0 +1,53 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.gateway.rest.message.statement;
+
+import org.apache.flink.runtime.rest.messages.ResponseBody;
+import org.apache.flink.table.gateway.api.operation.OperationType;
+
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonInclude;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
+
+/** {@link ResponseBody} for execute a statement. */
+@JsonInclude(JsonInclude.Include.NON_NULL)
+public class ExecuteStatementResponseBody implements ResponseBody {
+
+ private static final String FIELD_OPERATION_HANDLE = "operation_handle";
+
+ private static final String FIELD_OPERATION_TYPE = "operation_type";
+
+ @JsonProperty(FIELD_OPERATION_HANDLE)
+ private final String operationHandle;
+
+ @JsonProperty(FIELD_OPERATION_TYPE)
+ private final String operationTYpe = OperationType.EXECUTE_STATEMENT.name();
+
+ public ExecuteStatementResponseBody(
Review Comment:
Where is the field `has_result` as described in FLIP?
##########
flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/rest/message/statement/FetchResultsRequestBody.java:
##########
@@ -0,0 +1,47 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.gateway.rest.message.statement;
+
+import org.apache.flink.runtime.rest.messages.RequestBody;
+
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonInclude;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
+
+import javax.annotation.Nullable;
+
+/** {@link RequestBody} for fetching job result. */
+@JsonInclude(JsonInclude.Include.NON_NULL)
+public class FetchResultsRequestBody implements RequestBody {
+
+ private static final String FIELD_NAME_MAX_FETCH_SIZE = "max_fetch_size";
+
+ @JsonProperty(FIELD_NAME_MAX_FETCH_SIZE)
+ @Nullable
+ private final Integer maxFetchSize;
Review Comment:
This is not defined in FLIP.
##########
flink-table/flink-sql-gateway-api/src/main/java/org/apache/flink/table/gateway/api/results/ExceptionInfo.java:
##########
@@ -0,0 +1,64 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.gateway.api.results;
+
+import org.apache.flink.util.Preconditions;
+
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
+
+/** ExceptionInfo for SQL gateway service. */
+public class ExceptionInfo {
+
+ private static final String FIELD_EXCEPTION_ROOT_CLASS_NAME = "root";
+ private static final String FIELD_EXCEPTION_MESSAGE = "message";
Review Comment:
This is not specified in FLIP.
##########
flink-table/flink-sql-gateway-api/src/main/java/org/apache/flink/table/gateway/api/results/ResultSet.java:
##########
@@ -23,16 +23,30 @@
import org.apache.flink.table.catalog.ResolvedSchema;
import org.apache.flink.table.data.RowData;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.annotation.JsonDeserialize;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.annotation.JsonSerialize;
+
import javax.annotation.Nullable;
+import java.io.Serializable;
import java.util.Collections;
import java.util.List;
import java.util.Objects;
import java.util.stream.Collectors;
/** The collection of the results. */
@PublicEvolving
-public class ResultSet {
+@JsonSerialize(using = ResultSetJsonSerializer.class)
+@JsonDeserialize(using = ResultSetJsonDeserializer.class)
+public class ResultSet implements Serializable {
Review Comment:
It seems this class needs a serial version id
##########
flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/rest/message/statement/FetchResultsResponseBody.java:
##########
@@ -0,0 +1,75 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.gateway.rest.message.statement;
+
+import org.apache.flink.runtime.rest.messages.ResponseBody;
+import org.apache.flink.table.gateway.api.results.ExceptionInfo;
+import org.apache.flink.table.gateway.api.results.ResultSet;
+
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonInclude;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
+
+/** {@link ResponseBody} for execute a statement. */
+@JsonInclude(JsonInclude.Include.NON_NULL)
+public class FetchResultsResponseBody implements ResponseBody {
+
+ private static final String FIELD_RESULT_TYPE = "result_type";
+ private static final String FIELD_RESULTS = "results";
+ private static final String FIELD_NEXT_RESULT_URI = "next_result_uri";
+ private static final String FIELD_EXCEPTION = "exception";
+
+ @JsonProperty(FIELD_RESULTS)
+ private final ResultSet results;
+
+ @JsonProperty(FIELD_RESULT_TYPE)
+ private final String resultType;
+
+ @JsonProperty(FIELD_NEXT_RESULT_URI)
+ private final String nextResultUri;
+
+ @JsonProperty(FIELD_EXCEPTION)
+ private final ExceptionInfo exceptionInfo;
Review Comment:
IIUC, these two should be nullable?
##########
flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/rest/message/statement/ExecuteStatementResponseBody.java:
##########
@@ -0,0 +1,53 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.gateway.rest.message.statement;
+
+import org.apache.flink.runtime.rest.messages.ResponseBody;
+import org.apache.flink.table.gateway.api.operation.OperationType;
+
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonInclude;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
+
+/** {@link ResponseBody} for execute a statement. */
+@JsonInclude(JsonInclude.Include.NON_NULL)
+public class ExecuteStatementResponseBody implements ResponseBody {
+
+ private static final String FIELD_OPERATION_HANDLE = "operation_handle";
+
+ private static final String FIELD_OPERATION_TYPE = "operation_type";
+
+ @JsonProperty(FIELD_OPERATION_HANDLE)
+ private final String operationHandle;
+
+ @JsonProperty(FIELD_OPERATION_TYPE)
+ private final String operationTYpe = OperationType.EXECUTE_STATEMENT.name();
Review Comment:
```suggestion
private final String operationType = OperationType.EXECUTE_STATEMENT.name();
```
##########
flink-table/flink-sql-gateway-api/src/main/java/org/apache/flink/table/gateway/api/results/ExceptionInfo.java:
##########
@@ -0,0 +1,64 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.gateway.api.results;
+
+import org.apache.flink.util.Preconditions;
+
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
+
+/** ExceptionInfo for SQL gateway service. */
+public class ExceptionInfo {
+
+ private static final String FIELD_EXCEPTION_ROOT_CLASS_NAME = "root";
Review Comment:
Should be "root_cause" as specified in FLIP.
##########
flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/rest/message/statement/ExecuteStatementResponseBody.java:
##########
@@ -0,0 +1,53 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.gateway.rest.message.statement;
+
+import org.apache.flink.runtime.rest.messages.ResponseBody;
+import org.apache.flink.table.gateway.api.operation.OperationType;
+
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonInclude;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
+
+/** {@link ResponseBody} for execute a statement. */
+@JsonInclude(JsonInclude.Include.NON_NULL)
+public class ExecuteStatementResponseBody implements ResponseBody {
+
+ private static final String FIELD_OPERATION_HANDLE = "operation_handle";
+
+ private static final String FIELD_OPERATION_TYPE = "operation_type";
+
+ @JsonProperty(FIELD_OPERATION_HANDLE)
+ private final String operationHandle;
+
+ @JsonProperty(FIELD_OPERATION_TYPE)
+ private final String operationTYpe = OperationType.EXECUTE_STATEMENT.name();
+
+ public ExecuteStatementResponseBody(
+ @JsonProperty(FIELD_OPERATION_HANDLE) String operationHandle) {
+ this.operationHandle = operationHandle;
+ }
+
+ public String getOperationHandle() {
+ return operationHandle;
+ }
+
+ public String getOperationTYpe() {
Review Comment:
```suggestion
public String getOperationType() {
```
##########
flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/rest/handler/util/GetApiVersionHandler.java:
##########
@@ -0,0 +1,58 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.gateway.rest.handler.util;
+
+import org.apache.flink.runtime.rest.handler.HandlerRequest;
+import org.apache.flink.runtime.rest.messages.EmptyMessageParameters;
+import org.apache.flink.runtime.rest.messages.EmptyRequestBody;
+import org.apache.flink.runtime.rest.messages.MessageHeaders;
+import org.apache.flink.table.gateway.api.SqlGatewayService;
+import org.apache.flink.table.gateway.rest.handler.AbstractSqlGatewayRestHandler;
+import org.apache.flink.table.gateway.rest.message.util.GetApiVersionResponseBody;
+import org.apache.flink.table.gateway.rest.util.SqlGatewayRestAPIVersion;
+
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
+
+import java.util.Collections;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+
+/** Handler to get rest api version. */
+public class GetApiVersionHandler
+ extends AbstractSqlGatewayRestHandler<
+ EmptyRequestBody, GetApiVersionResponseBody, EmptyMessageParameters> {
+
+ public GetApiVersionHandler(
+ SqlGatewayService service,
+ Map<String, String> responseHeaders,
+ MessageHeaders<EmptyRequestBody, GetApiVersionResponseBody, EmptyMessageParameters>
+ messageHeaders) {
+ super(service, responseHeaders, messageHeaders);
+ }
+
+ @Override
+ protected CompletableFuture<GetApiVersionResponseBody> handleRequest(
+ @Nullable SqlGatewayRestAPIVersion version,
+ @Nonnull HandlerRequest<EmptyRequestBody> request) {
+ return CompletableFuture.completedFuture(
+ new GetApiVersionResponseBody(
+ Collections.singletonList(SqlGatewayRestAPIVersion.V1.toString())));
Review Comment:
```suggestion
return CompletableFuture.completedFuture(
new GetApiVersionResponseBody(
Arrays.stream(SqlGatewayRestAPIVersion.values())
.filter(SqlGatewayRestAPIVersion::isStableVersion)
.map(Objects::toString)
.collect(Collectors.toList())));
```
##########
flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/rest/handler/statement/ExecuteStatementHandler.java:
##########
@@ -0,0 +1,89 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.gateway.rest.handler.statement;
+
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
Review Comment:
Extra license.
##########
flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/rest/handler/statement/ExecuteStatementHandler.java:
##########
@@ -0,0 +1,89 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.gateway.rest.handler.statement;
+
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.runtime.rest.handler.HandlerRequest;
+import org.apache.flink.runtime.rest.messages.MessageHeaders;
+import org.apache.flink.table.gateway.api.SqlGatewayService;
+import org.apache.flink.table.gateway.api.operation.OperationHandle;
+import org.apache.flink.table.gateway.api.session.SessionHandle;
+import org.apache.flink.table.gateway.rest.handler.AbstractSqlGatewayRestHandler;
+import org.apache.flink.table.gateway.rest.message.session.SessionHandleIdPathParameter;
+import org.apache.flink.table.gateway.rest.message.session.SessionMessageParameters;
+import org.apache.flink.table.gateway.rest.message.statement.ExecuteStatementRequestBody;
+import org.apache.flink.table.gateway.rest.message.statement.ExecuteStatementResponseBody;
+import org.apache.flink.table.gateway.rest.util.SqlGatewayRestAPIVersion;
+
+import javax.annotation.Nonnull;
+
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+
+/** Handler to execute a statement. */
+public class ExecuteStatementHandler
+ extends AbstractSqlGatewayRestHandler<
+ ExecuteStatementRequestBody,
+ ExecuteStatementResponseBody,
+ SessionMessageParameters> {
+
+ public ExecuteStatementHandler(
+ SqlGatewayService service,
+ Map<String, String> responseHeaders,
+ MessageHeaders<
+ ExecuteStatementRequestBody,
+ ExecuteStatementResponseBody,
+ SessionMessageParameters>
+ messageHeaders) {
+ super(service, responseHeaders, messageHeaders);
+ }
+
+ @Override
+ protected CompletableFuture<ExecuteStatementResponseBody> handleRequest(
+ SqlGatewayRestAPIVersion version,
+ @Nonnull HandlerRequest<ExecuteStatementRequestBody> request) {
+ String statement = request.getRequestBody().getStatement();
+ Long timeout = request.getRequestBody().getTimeout();
+ timeout = timeout == null ? 0L : timeout;
+ SessionHandle sessionHandle = request.getPathParameter(SessionHandleIdPathParameter.class);
+ OperationHandle operationHandle =
+ service.executeStatement(sessionHandle, statement, timeout, new Configuration());
Review Comment:
This `new Configuration` doesn't feel right. The FLIP says "the RESET statement will reset all properties set by SET xx=yy statement."
##########
flink-table/flink-sql-gateway-api/src/main/java/org/apache/flink/table/gateway/api/results/ResultSet.java:
##########
@@ -23,16 +23,30 @@
import org.apache.flink.table.catalog.ResolvedSchema;
import org.apache.flink.table.data.RowData;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.annotation.JsonDeserialize;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.annotation.JsonSerialize;
+
import javax.annotation.Nullable;
+import java.io.Serializable;
import java.util.Collections;
import java.util.List;
import java.util.Objects;
import java.util.stream.Collectors;
/** The collection of the results. */
@PublicEvolving
-public class ResultSet {
+@JsonSerialize(using = ResultSetJsonSerializer.class)
+@JsonDeserialize(using = ResultSetJsonDeserializer.class)
+public class ResultSet implements Serializable {
+
+ public static final String FIELD_NAME_RESULT_TYPE = "result_type";
+
+ public static final String FIELD_NAME_COLUMN_INFOS = "columns";
+
+ public static final String FIELD_NAME_DATA = "data";
+
+ public static final String FIELD_NAME_NEXT_TOKEN = "next_token";
Review Comment:
This is not defined in FLIP
##########
flink-table/flink-sql-gateway-api/src/main/java/org/apache/flink/table/gateway/api/results/ResultSetJsonSerializer.java:
##########
@@ -0,0 +1,89 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.gateway.api.results;
+
+import org.apache.flink.formats.common.TimestampFormat;
+import org.apache.flink.formats.json.JsonFormatOptions;
+import org.apache.flink.formats.json.JsonRowDataSerializationSchema;
+import org.apache.flink.table.catalog.Column;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.types.logical.RowType;
+
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonGenerator;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.JsonSerializer;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.SerializerProvider;
+
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.apache.flink.table.gateway.api.results.ResultSet.FIELD_NAME_COLUMN_INFOS;
+import static org.apache.flink.table.gateway.api.results.ResultSet.FIELD_NAME_DATA;
+import static org.apache.flink.table.gateway.api.results.ResultSet.FIELD_NAME_NEXT_TOKEN;
+import static org.apache.flink.table.gateway.api.results.ResultSet.FIELD_NAME_RESULT_TYPE;
+
+/** Json serializer for {@link ResultSet}. */
+public class ResultSetJsonSerializer extends JsonSerializer<ResultSet> {
+
+ @Override
+ public void serialize(
+ ResultSet resultSet, JsonGenerator jsonGenerator, SerializerProvider serializerProvider)
+ throws IOException {
+ jsonGenerator.writeStartObject();
+
+ // Serialize result type
+ serializerProvider.defaultSerializeField(
+ FIELD_NAME_RESULT_TYPE, resultSet.getResultType(), jsonGenerator);
+
+ // Serialize next token
+ serializerProvider.defaultSerializeField(
+ FIELD_NAME_NEXT_TOKEN, resultSet.getNextToken(), jsonGenerator);
+
+ // Serialize column infos
+ List<Column> columns = resultSet.getResultSchema().getColumns();
+ List<ColumnInfo> columnInfos = new ArrayList<>();
+ for (Column column : columns) {
+ columnInfos.add(
+ ColumnInfo.create(column.getName(), column.getDataType().getLogicalType()));
+ }
+ serializerProvider.defaultSerializeField(
+ FIELD_NAME_COLUMN_INFOS, columnInfos, jsonGenerator);
+
+ // Serialize RowData
+ JsonRowDataSerializationSchema serializationSchema =
+ new JsonRowDataSerializationSchema(
+ (RowType)
+ resultSet
+ .getResultSchema()
+ .toPhysicalRowDataType()
+ .getLogicalType(),
+ TimestampFormat.ISO_8601,
+ JsonFormatOptions.MapNullKeyMode.LITERAL,
+ "null",
+ true);
+ List<String> rowDataStrings = new ArrayList<>();
+ for (RowData row : resultSet.getData()) {
+ byte[] rowByte = serializationSchema.serialize(row);
+ rowDataStrings.add(new String(rowByte, StandardCharsets.UTF_8));
+ }
Review Comment:
These doesn't align with the FLIP.
##########
flink-table/flink-sql-gateway-api/pom.xml:
##########
@@ -64,7 +64,19 @@
<version>${project.version}</version>
<scope>test</scope>
</dependency>
- </dependencies>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-json</artifactId>
+ <version>1.16-SNAPSHOT</version>
+ <scope>compile</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-table-runtime</artifactId>
+ <version>1.16-SNAPSHOT</version>
Review Comment:
```suggestion
<version>${project.version}</version>
```
##########
flink-table/flink-sql-gateway-api/src/main/java/org/apache/flink/table/gateway/api/results/ResultSetJsonSerializer.java:
##########
@@ -0,0 +1,89 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.gateway.api.results;
+
+import org.apache.flink.formats.common.TimestampFormat;
+import org.apache.flink.formats.json.JsonFormatOptions;
+import org.apache.flink.formats.json.JsonRowDataSerializationSchema;
+import org.apache.flink.table.catalog.Column;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.types.logical.RowType;
+
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonGenerator;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.JsonSerializer;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.SerializerProvider;
+
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.apache.flink.table.gateway.api.results.ResultSet.FIELD_NAME_COLUMN_INFOS;
+import static org.apache.flink.table.gateway.api.results.ResultSet.FIELD_NAME_DATA;
+import static org.apache.flink.table.gateway.api.results.ResultSet.FIELD_NAME_NEXT_TOKEN;
+import static org.apache.flink.table.gateway.api.results.ResultSet.FIELD_NAME_RESULT_TYPE;
+
+/** Json serializer for {@link ResultSet}. */
+public class ResultSetJsonSerializer extends JsonSerializer<ResultSet> {
Review Comment:
We need tests for this serializer and the corresponding deserializer.
##########
flink-table/flink-sql-gateway-api/src/main/java/org/apache/flink/table/gateway/api/results/ResultSet.java:
##########
@@ -23,16 +23,30 @@
import org.apache.flink.table.catalog.ResolvedSchema;
import org.apache.flink.table.data.RowData;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.annotation.JsonDeserialize;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.annotation.JsonSerialize;
+
import javax.annotation.Nullable;
+import java.io.Serializable;
import java.util.Collections;
import java.util.List;
import java.util.Objects;
import java.util.stream.Collectors;
/** The collection of the results. */
@PublicEvolving
-public class ResultSet {
+@JsonSerialize(using = ResultSetJsonSerializer.class)
+@JsonDeserialize(using = ResultSetJsonDeserializer.class)
+public class ResultSet implements Serializable {
+
+ public static final String FIELD_NAME_RESULT_TYPE = "result_type";
Review Comment:
This is not defined in FLIP
##########
flink-table/flink-sql-gateway-api/src/main/java/org/apache/flink/table/gateway/api/results/ResultSetJsonSerializer.java:
##########
@@ -0,0 +1,89 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.gateway.api.results;
+
+import org.apache.flink.formats.common.TimestampFormat;
+import org.apache.flink.formats.json.JsonFormatOptions;
+import org.apache.flink.formats.json.JsonRowDataSerializationSchema;
+import org.apache.flink.table.catalog.Column;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.types.logical.RowType;
+
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonGenerator;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.JsonSerializer;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.SerializerProvider;
+
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.apache.flink.table.gateway.api.results.ResultSet.FIELD_NAME_COLUMN_INFOS;
+import static org.apache.flink.table.gateway.api.results.ResultSet.FIELD_NAME_DATA;
+import static org.apache.flink.table.gateway.api.results.ResultSet.FIELD_NAME_NEXT_TOKEN;
+import static org.apache.flink.table.gateway.api.results.ResultSet.FIELD_NAME_RESULT_TYPE;
+
+/** Json serializer for {@link ResultSet}. */
+public class ResultSetJsonSerializer extends JsonSerializer<ResultSet> {
+
+ @Override
+ public void serialize(
+ ResultSet resultSet, JsonGenerator jsonGenerator, SerializerProvider serializerProvider)
+ throws IOException {
+ jsonGenerator.writeStartObject();
+
+ // Serialize result type
+ serializerProvider.defaultSerializeField(
+ FIELD_NAME_RESULT_TYPE, resultSet.getResultType(), jsonGenerator);
+
+ // Serialize next token
+ serializerProvider.defaultSerializeField(
+ FIELD_NAME_NEXT_TOKEN, resultSet.getNextToken(), jsonGenerator);
Review Comment:
These are not defined in FLIP.
##########
flink-table/flink-sql-gateway-api/pom.xml:
##########
@@ -64,7 +64,19 @@
<version>${project.version}</version>
<scope>test</scope>
</dependency>
- </dependencies>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-json</artifactId>
+ <version>1.16-SNAPSHOT</version>
Review Comment:
```suggestion
<version>${project.version}</version>
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] WencongLiu commented on a diff in pull request #20451: [FLINK-28162][sql-gateway]Introduce the operation related API for REST endpoint
Posted by GitBox <gi...@apache.org>.
WencongLiu commented on code in PR #20451:
URL: https://github.com/apache/flink/pull/20451#discussion_r938463223
##########
flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/rest/message/statement/ExecuteStatementResponseBody.java:
##########
@@ -0,0 +1,53 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.gateway.rest.message.statement;
+
+import org.apache.flink.runtime.rest.messages.ResponseBody;
+import org.apache.flink.table.gateway.api.operation.OperationType;
+
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonInclude;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
+
+/** {@link ResponseBody} for execute a statement. */
+@JsonInclude(JsonInclude.Include.NON_NULL)
+public class ExecuteStatementResponseBody implements ResponseBody {
+
+ private static final String FIELD_OPERATION_HANDLE = "operation_handle";
+
+ private static final String FIELD_OPERATION_TYPE = "operation_type";
+
+ @JsonProperty(FIELD_OPERATION_HANDLE)
+ private final String operationHandle;
+
+ @JsonProperty(FIELD_OPERATION_TYPE)
+ private final String operationTYpe = OperationType.EXECUTE_STATEMENT.name();
+
+ public ExecuteStatementResponseBody(
+ @JsonProperty(FIELD_OPERATION_HANDLE) String operationHandle) {
+ this.operationHandle = operationHandle;
+ }
+
+ public String getOperationHandle() {
+ return operationHandle;
+ }
+
+ public String getOperationTYpe() {
Review Comment:
Fixed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] WencongLiu commented on a diff in pull request #20451: [FLINK-28162][sql-gateway]Introduce the operation related API for REST endpoint
Posted by GitBox <gi...@apache.org>.
WencongLiu commented on code in PR #20451:
URL: https://github.com/apache/flink/pull/20451#discussion_r938461506
##########
flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/rest/handler/statement/ExecuteStatementHandler.java:
##########
@@ -0,0 +1,89 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.gateway.rest.handler.statement;
+
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
Review Comment:
Removed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] WencongLiu commented on a diff in pull request #20451: [FLINK-28162][sql-gateway]Introduce the operation related API for REST endpoint
Posted by GitBox <gi...@apache.org>.
WencongLiu commented on code in PR #20451:
URL: https://github.com/apache/flink/pull/20451#discussion_r938460160
##########
flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/rest/handler/util/GetApiVersionHandler.java:
##########
@@ -0,0 +1,58 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.gateway.rest.handler.util;
+
+import org.apache.flink.runtime.rest.handler.HandlerRequest;
+import org.apache.flink.runtime.rest.messages.EmptyMessageParameters;
+import org.apache.flink.runtime.rest.messages.EmptyRequestBody;
+import org.apache.flink.runtime.rest.messages.MessageHeaders;
+import org.apache.flink.table.gateway.api.SqlGatewayService;
+import org.apache.flink.table.gateway.rest.handler.AbstractSqlGatewayRestHandler;
+import org.apache.flink.table.gateway.rest.message.util.GetApiVersionResponseBody;
+import org.apache.flink.table.gateway.rest.util.SqlGatewayRestAPIVersion;
+
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
+
+import java.util.Collections;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+
+/** Handler to get rest api version. */
+public class GetApiVersionHandler
+ extends AbstractSqlGatewayRestHandler<
+ EmptyRequestBody, GetApiVersionResponseBody, EmptyMessageParameters> {
+
+ public GetApiVersionHandler(
+ SqlGatewayService service,
+ Map<String, String> responseHeaders,
+ MessageHeaders<EmptyRequestBody, GetApiVersionResponseBody, EmptyMessageParameters>
+ messageHeaders) {
+ super(service, responseHeaders, messageHeaders);
+ }
+
+ @Override
+ protected CompletableFuture<GetApiVersionResponseBody> handleRequest(
+ @Nullable SqlGatewayRestAPIVersion version,
+ @Nonnull HandlerRequest<EmptyRequestBody> request) {
+ return CompletableFuture.completedFuture(
+ new GetApiVersionResponseBody(
+ Collections.singletonList(SqlGatewayRestAPIVersion.V1.toString())));
Review Comment:
Done.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] WencongLiu commented on a diff in pull request #20451: [FLINK-28162][sql-gateway]Introduce the operation related API for REST endpoint
Posted by GitBox <gi...@apache.org>.
WencongLiu commented on code in PR #20451:
URL: https://github.com/apache/flink/pull/20451#discussion_r939493280
##########
flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/rest/message/statement/FetchResultsRequestBody.java:
##########
@@ -0,0 +1,47 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.gateway.rest.message.statement;
+
+import org.apache.flink.runtime.rest.messages.RequestBody;
+
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonInclude;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
+
+import javax.annotation.Nullable;
+
+/** {@link RequestBody} for fetching job result. */
+@JsonInclude(JsonInclude.Include.NON_NULL)
+public class FetchResultsRequestBody implements RequestBody {
+
+ private static final String FIELD_NAME_MAX_FETCH_SIZE = "max_fetch_size";
Review Comment:
@zentol @xintongsong thanks, I'll fix it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] WencongLiu commented on a diff in pull request #20451: [FLINK-28162][sql-gateway]Introduce the operation related API for REST endpoint
Posted by GitBox <gi...@apache.org>.
WencongLiu commented on code in PR #20451:
URL: https://github.com/apache/flink/pull/20451#discussion_r939497689
##########
flink-table/flink-sql-gateway-api/src/main/java/org/apache/flink/table/gateway/api/results/ResultSetJsonSerializer.java:
##########
@@ -0,0 +1,89 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.gateway.api.results;
+
+import org.apache.flink.formats.common.TimestampFormat;
+import org.apache.flink.formats.json.JsonFormatOptions;
+import org.apache.flink.formats.json.JsonRowDataSerializationSchema;
+import org.apache.flink.table.catalog.Column;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.types.logical.RowType;
+
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonGenerator;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.JsonSerializer;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.SerializerProvider;
+
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.apache.flink.table.gateway.api.results.ResultSet.FIELD_NAME_COLUMN_INFOS;
+import static org.apache.flink.table.gateway.api.results.ResultSet.FIELD_NAME_DATA;
+import static org.apache.flink.table.gateway.api.results.ResultSet.FIELD_NAME_NEXT_TOKEN;
+import static org.apache.flink.table.gateway.api.results.ResultSet.FIELD_NAME_RESULT_TYPE;
+
+/** Json serializer for {@link ResultSet}. */
+public class ResultSetJsonSerializer extends JsonSerializer<ResultSet> {
Review Comment:
Added tests.
##########
flink-table/flink-sql-gateway-api/pom.xml:
##########
@@ -64,7 +64,19 @@
<version>${project.version}</version>
<scope>test</scope>
</dependency>
- </dependencies>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-json</artifactId>
+ <version>1.16-SNAPSHOT</version>
+ <scope>compile</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-table-runtime</artifactId>
+ <version>1.16-SNAPSHOT</version>
Review Comment:
Fixed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] xintongsong commented on a diff in pull request #20451: [FLINK-28162][sql-gateway]Introduce the operation related API for REST endpoint
Posted by GitBox <gi...@apache.org>.
xintongsong commented on code in PR #20451:
URL: https://github.com/apache/flink/pull/20451#discussion_r938390193
##########
flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/rest/message/statement/FetchResultsRequestBody.java:
##########
@@ -0,0 +1,47 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.gateway.rest.message.statement;
+
+import org.apache.flink.runtime.rest.messages.RequestBody;
+
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonInclude;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
+
+import javax.annotation.Nullable;
+
+/** {@link RequestBody} for fetching job result. */
+@JsonInclude(JsonInclude.Include.NON_NULL)
+public class FetchResultsRequestBody implements RequestBody {
+
+ private static final String FIELD_NAME_MAX_FETCH_SIZE = "max_fetch_size";
Review Comment:
I agree that it would be nice to use consistent naming conventions for the REST fields in 2.0, and we should start to do that early. Is there already a consensus that camelCase will be used in long term? Could you point us to it?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] WencongLiu commented on pull request #20451: [FLINK-28162][sql-gateway]Introduce the operation related API for REST endpoint
Posted by GitBox <gi...@apache.org>.
WencongLiu commented on PR #20451:
URL: https://github.com/apache/flink/pull/20451#issuecomment-1207801814
> Thanks @WencongLiu. I think this PR is very close to a mergable state.
>
> I have rebased the latest master branch, squashed your fixup commits, and made a few minor changes in two new fixup commits. I've also moved the session related api changes to a hotfix commit.
>
> I have only two remaining issues.
>
> * I left some of the comments from the previous round of review unresolved, most of which requires updates to the FLIP. @fsk119, could you help confirm them?
> * I'm not fully convinced by making `nextResultUri` and `exception` in `FetchResultsResponseBody` required, even when there's not a next result or exception, especially for `exception`. Why do we need this complexity?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] zentol commented on a diff in pull request #20451: [FLINK-28162][sql-gateway]Introduce the operation related API for REST endpoint
Posted by GitBox <gi...@apache.org>.
zentol commented on code in PR #20451:
URL: https://github.com/apache/flink/pull/20451#discussion_r938635524
##########
flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/rest/message/statement/FetchResultsRequestBody.java:
##########
@@ -0,0 +1,47 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.gateway.rest.message.statement;
+
+import org.apache.flink.runtime.rest.messages.RequestBody;
+
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonInclude;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
+
+import javax.annotation.Nullable;
+
+/** {@link RequestBody} for fetching job result. */
+@JsonInclude(JsonInclude.Include.NON_NULL)
+public class FetchResultsRequestBody implements RequestBody {
+
+ private static final String FIELD_NAME_MAX_FETCH_SIZE = "max_fetch_size";
Review Comment:
See the discussion in https://issues.apache.org/jira/browse/FLINK-15787
I also made sure all recently added REST APIs followed camelCase.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] WencongLiu commented on a diff in pull request #20451: [FLINK-28162][sql-gateway]Introduce the operation related API for REST endpoint
Posted by GitBox <gi...@apache.org>.
WencongLiu commented on code in PR #20451:
URL: https://github.com/apache/flink/pull/20451#discussion_r938460160
##########
flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/rest/handler/util/GetApiVersionHandler.java:
##########
@@ -0,0 +1,58 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.gateway.rest.handler.util;
+
+import org.apache.flink.runtime.rest.handler.HandlerRequest;
+import org.apache.flink.runtime.rest.messages.EmptyMessageParameters;
+import org.apache.flink.runtime.rest.messages.EmptyRequestBody;
+import org.apache.flink.runtime.rest.messages.MessageHeaders;
+import org.apache.flink.table.gateway.api.SqlGatewayService;
+import org.apache.flink.table.gateway.rest.handler.AbstractSqlGatewayRestHandler;
+import org.apache.flink.table.gateway.rest.message.util.GetApiVersionResponseBody;
+import org.apache.flink.table.gateway.rest.util.SqlGatewayRestAPIVersion;
+
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
+
+import java.util.Collections;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+
+/** Handler to get rest api version. */
+public class GetApiVersionHandler
+ extends AbstractSqlGatewayRestHandler<
+ EmptyRequestBody, GetApiVersionResponseBody, EmptyMessageParameters> {
+
+ public GetApiVersionHandler(
+ SqlGatewayService service,
+ Map<String, String> responseHeaders,
+ MessageHeaders<EmptyRequestBody, GetApiVersionResponseBody, EmptyMessageParameters>
+ messageHeaders) {
+ super(service, responseHeaders, messageHeaders);
+ }
+
+ @Override
+ protected CompletableFuture<GetApiVersionResponseBody> handleRequest(
+ @Nullable SqlGatewayRestAPIVersion version,
+ @Nonnull HandlerRequest<EmptyRequestBody> request) {
+ return CompletableFuture.completedFuture(
+ new GetApiVersionResponseBody(
+ Collections.singletonList(SqlGatewayRestAPIVersion.V1.toString())));
Review Comment:
Fixed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] zentol commented on a diff in pull request #20451: [FLINK-28162][sql-gateway]Introduce the operation related API for REST endpoint
Posted by GitBox <gi...@apache.org>.
zentol commented on code in PR #20451:
URL: https://github.com/apache/flink/pull/20451#discussion_r937708827
##########
flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/rest/message/statement/FetchResultsRequestBody.java:
##########
@@ -0,0 +1,47 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.gateway.rest.message.statement;
+
+import org.apache.flink.runtime.rest.messages.RequestBody;
+
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonInclude;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
+
+import javax.annotation.Nullable;
+
+/** {@link RequestBody} for fetching job result. */
+@JsonInclude(JsonInclude.Include.NON_NULL)
+public class FetchResultsRequestBody implements RequestBody {
+
+ private static final String FIELD_NAME_MAX_FETCH_SIZE = "max_fetch_size";
Review Comment:
Please make sure that _all_ request/response body fields use camelCase. Let's write this from the get-go in such a way that we don't have to break everything in 2.0.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] zentol commented on a diff in pull request #20451: [FLINK-28162][sql-gateway]Introduce the operation related API for REST endpoint
Posted by GitBox <gi...@apache.org>.
zentol commented on code in PR #20451:
URL: https://github.com/apache/flink/pull/20451#discussion_r937708827
##########
flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/rest/message/statement/FetchResultsRequestBody.java:
##########
@@ -0,0 +1,47 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.gateway.rest.message.statement;
+
+import org.apache.flink.runtime.rest.messages.RequestBody;
+
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonInclude;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
+
+import javax.annotation.Nullable;
+
+/** {@link RequestBody} for fetching job result. */
+@JsonInclude(JsonInclude.Include.NON_NULL)
+public class FetchResultsRequestBody implements RequestBody {
+
+ private static final String FIELD_NAME_MAX_FETCH_SIZE = "max_fetch_size";
Review Comment:
Please make sure that _all_ request/response body fields use camelCase.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] WencongLiu commented on a diff in pull request #20451: [FLINK-28162][sql-gateway]Introduce the operation related API for REST endpoint
Posted by GitBox <gi...@apache.org>.
WencongLiu commented on code in PR #20451:
URL: https://github.com/apache/flink/pull/20451#discussion_r938459685
##########
flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/rest/handler/operation/AbstractOperationHandler.java:
##########
@@ -0,0 +1,91 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.gateway.rest.handler.operation;
+
+import org.apache.flink.runtime.rest.handler.HandlerRequest;
+import org.apache.flink.runtime.rest.handler.RestHandlerException;
+import org.apache.flink.runtime.rest.messages.EmptyRequestBody;
+import org.apache.flink.runtime.rest.messages.MessageHeaders;
+import org.apache.flink.table.gateway.api.SqlGatewayService;
+import org.apache.flink.table.gateway.api.operation.OperationHandle;
+import org.apache.flink.table.gateway.api.operation.OperationStatus;
+import org.apache.flink.table.gateway.api.session.SessionHandle;
+import org.apache.flink.table.gateway.api.utils.SqlGatewayException;
+import org.apache.flink.table.gateway.rest.handler.AbstractSqlGatewayRestHandler;
+import org.apache.flink.table.gateway.rest.message.operation.OperationHandleIdPathParameter;
+import org.apache.flink.table.gateway.rest.message.operation.OperationMessageParameters;
+import org.apache.flink.table.gateway.rest.message.operation.OperationStatusResponseBody;
+import org.apache.flink.table.gateway.rest.message.session.SessionHandleIdPathParameter;
+import org.apache.flink.table.gateway.rest.util.SqlGatewayRestAPIVersion;
+
+import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpResponseStatus;
+
+import javax.annotation.Nonnull;
+
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+
+/** Abstract operation handler. */
+public abstract class AbstractOperationHandler
+ extends AbstractSqlGatewayRestHandler<
+ EmptyRequestBody, OperationStatusResponseBody, OperationMessageParameters> {
+
+ public AbstractOperationHandler(
+ SqlGatewayService service,
+ Map<String, String> responseHeaders,
+ MessageHeaders<
+ EmptyRequestBody,
+ OperationStatusResponseBody,
+ OperationMessageParameters>
+ messageHeaders) {
+ super(service, responseHeaders, messageHeaders);
+ }
+
+ private String getStatus(SessionHandle sessionHandle, OperationHandle operationHandle) {
+ if (execute(sessionHandle, operationHandle)) {
+ return this.service
+ .getOperationInfo(sessionHandle, operationHandle)
+ .getStatus()
+ .toString();
+ } else {
+ return OperationStatus.CLOSED.toString();
+ }
+ }
+
+ boolean execute(SessionHandle sessionHandle, OperationHandle operationHandle) {
Review Comment:
The class inheritance structure has been refactored.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] WencongLiu commented on a diff in pull request #20451: [FLINK-28162][sql-gateway]Introduce the operation related API for REST endpoint
Posted by GitBox <gi...@apache.org>.
WencongLiu commented on code in PR #20451:
URL: https://github.com/apache/flink/pull/20451#discussion_r938468545
##########
flink-table/flink-sql-gateway-api/src/main/java/org/apache/flink/table/gateway/api/results/ExceptionInfo.java:
##########
@@ -0,0 +1,64 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.gateway.api.results;
+
+import org.apache.flink.util.Preconditions;
+
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
+
+/** ExceptionInfo for SQL gateway service. */
+public class ExceptionInfo {
+
+ private static final String FIELD_EXCEPTION_ROOT_CLASS_NAME = "root";
Review Comment:
Fixed.
The ExceptionInfo contains the same attributes as the FLIP.
"exception": {
"root_cause": "....",
"exception_stack": "..."
}
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] zentol commented on a diff in pull request #20451: [FLINK-28162][sql-gateway]Introduce the operation related API for REST endpoint
Posted by GitBox <gi...@apache.org>.
zentol commented on code in PR #20451:
URL: https://github.com/apache/flink/pull/20451#discussion_r938635524
##########
flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/rest/message/statement/FetchResultsRequestBody.java:
##########
@@ -0,0 +1,47 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.gateway.rest.message.statement;
+
+import org.apache.flink.runtime.rest.messages.RequestBody;
+
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonInclude;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
+
+import javax.annotation.Nullable;
+
+/** {@link RequestBody} for fetching job result. */
+@JsonInclude(JsonInclude.Include.NON_NULL)
+public class FetchResultsRequestBody implements RequestBody {
+
+ private static final String FIELD_NAME_MAX_FETCH_SIZE = "max_fetch_size";
Review Comment:
See the discussion in https://issues.apache.org/jira/browse/FLINK-15787
I also made sure all recently added REST APIs followed camelCase (at least where I was involved...).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] WencongLiu commented on a diff in pull request #20451: [FLINK-28162][sql-gateway]Introduce the operation related API for REST endpoint
Posted by GitBox <gi...@apache.org>.
WencongLiu commented on code in PR #20451:
URL: https://github.com/apache/flink/pull/20451#discussion_r938489165
##########
flink-table/flink-sql-gateway-api/src/main/java/org/apache/flink/table/gateway/api/results/ResultSet.java:
##########
@@ -23,16 +23,30 @@
import org.apache.flink.table.catalog.ResolvedSchema;
import org.apache.flink.table.data.RowData;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.annotation.JsonDeserialize;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.annotation.JsonSerialize;
+
import javax.annotation.Nullable;
+import java.io.Serializable;
import java.util.Collections;
import java.util.List;
import java.util.Objects;
import java.util.stream.Collectors;
/** The collection of the results. */
@PublicEvolving
-public class ResultSet {
+@JsonSerialize(using = ResultSetJsonSerializer.class)
+@JsonDeserialize(using = ResultSetJsonDeserializer.class)
+public class ResultSet implements Serializable {
Review Comment:
Fixed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] WencongLiu closed pull request #20451: [FLINK-28162][sql-gateway]Introduce the operation related API for REST endpoint
Posted by GitBox <gi...@apache.org>.
WencongLiu closed pull request #20451: [FLINK-28162][sql-gateway]Introduce the operation related API for REST endpoint
URL: https://github.com/apache/flink/pull/20451
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] xintongsong commented on a diff in pull request #20451: [FLINK-28162][sql-gateway]Introduce the operation related API for REST endpoint
Posted by GitBox <gi...@apache.org>.
xintongsong commented on code in PR #20451:
URL: https://github.com/apache/flink/pull/20451#discussion_r939983157
##########
flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/rest/message/statement/FetchResultsResponseBody.java:
##########
@@ -0,0 +1,67 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.gateway.rest.message.statement;
+
+import org.apache.flink.runtime.rest.messages.ResponseBody;
+import org.apache.flink.table.gateway.api.results.ResultSet;
+
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
+
+import javax.annotation.Nullable;
+
+/** {@link ResponseBody} for execute a statement. */
+public class FetchResultsResponseBody implements ResponseBody {
+
+ private static final String FIELD_RESULT_TYPE = "resultType";
+ private static final String FIELD_RESULTS = "results";
+ private static final String FIELD_NEXT_RESULT_URI = "nextResultUri";
+ private static final String FIELD_EXCEPTION = "exception";
Review Comment:
```suggestion
```
##########
flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/rest/header/statement/FetchResultsHeaders.java:
##########
@@ -0,0 +1,97 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.gateway.rest.header.statement;
+
+import org.apache.flink.runtime.rest.HttpMethodWrapper;
+import org.apache.flink.runtime.rest.messages.EmptyRequestBody;
+import org.apache.flink.table.gateway.rest.header.SqlGatewayMessageHeaders;
+import org.apache.flink.table.gateway.rest.message.operation.OperationHandleIdPathParameter;
+import org.apache.flink.table.gateway.rest.message.session.SessionHandleIdPathParameter;
+import org.apache.flink.table.gateway.rest.message.statement.FetchResultsResponseBody;
+import org.apache.flink.table.gateway.rest.message.statement.FetchResultsTokenParameters;
+import org.apache.flink.table.gateway.rest.message.statement.FetchResultsTokenPathParameter;
+
+import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpResponseStatus;
+
+/** Message headers for fetching results. */
+public class FetchResultsHeaders
+ implements SqlGatewayMessageHeaders<
+ EmptyRequestBody, FetchResultsResponseBody, FetchResultsTokenParameters> {
+
+ private static final FetchResultsHeaders INSTANCE = new FetchResultsHeaders();
+
+ public static final String URL =
+ "/sessions/:"
+ + SessionHandleIdPathParameter.KEY
+ + "/operations/:"
+ + OperationHandleIdPathParameter.KEY
+ + "/result/:"
+ + FetchResultsTokenPathParameter.KEY;
+
+ @Override
+ public Class<FetchResultsResponseBody> getResponseClass() {
+ return FetchResultsResponseBody.class;
+ }
+
+ @Override
+ public HttpResponseStatus getResponseStatusCode() {
+ return HttpResponseStatus.OK;
+ }
+
+ @Override
+ public String getDescription() {
+ return "Fetch results of Operation.";
+ }
+
+ @Override
+ public HttpMethodWrapper getHttpMethod() {
+ return HttpMethodWrapper.GET;
+ }
+
+ @Override
+ public String getTargetRestEndpointURL() {
+ return URL;
+ }
+
+ public static FetchResultsHeaders getInstance() {
+ return INSTANCE;
+ }
+
+ public static String buildNextUri(
Review Comment:
```suggestion
@Nullable
public static String buildNextUri(
```
##########
flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/rest/handler/statement/FetchResultsHandler.java:
##########
@@ -0,0 +1,97 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.gateway.rest.handler.statement;
+
+import org.apache.flink.runtime.rest.handler.HandlerRequest;
+import org.apache.flink.runtime.rest.messages.EmptyRequestBody;
+import org.apache.flink.runtime.rest.messages.MessageHeaders;
+import org.apache.flink.table.gateway.api.SqlGatewayService;
+import org.apache.flink.table.gateway.api.operation.OperationHandle;
+import org.apache.flink.table.gateway.api.results.ResultSet;
+import org.apache.flink.table.gateway.api.session.SessionHandle;
+import org.apache.flink.table.gateway.api.utils.SqlGatewayException;
+import org.apache.flink.table.gateway.rest.handler.AbstractSqlGatewayRestHandler;
+import org.apache.flink.table.gateway.rest.header.statement.FetchResultsHeaders;
+import org.apache.flink.table.gateway.rest.message.operation.OperationHandleIdPathParameter;
+import org.apache.flink.table.gateway.rest.message.session.SessionHandleIdPathParameter;
+import org.apache.flink.table.gateway.rest.message.statement.FetchResultsResponseBody;
+import org.apache.flink.table.gateway.rest.message.statement.FetchResultsTokenParameters;
+import org.apache.flink.table.gateway.rest.message.statement.FetchResultsTokenPathParameter;
+import org.apache.flink.table.gateway.rest.util.SqlGatewayRestAPIVersion;
+
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
+
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+
+/** Handler to fetch results. */
+public class FetchResultsHandler
+ extends AbstractSqlGatewayRestHandler<
+ EmptyRequestBody, FetchResultsResponseBody, FetchResultsTokenParameters> {
+
+ public FetchResultsHandler(
+ SqlGatewayService service,
+ Map<String, String> responseHeaders,
+ MessageHeaders<EmptyRequestBody, FetchResultsResponseBody, FetchResultsTokenParameters>
+ messageHeaders) {
+ super(service, responseHeaders, messageHeaders);
+ }
+
+ @Override
+ protected CompletableFuture<FetchResultsResponseBody> handleRequest(
+ SqlGatewayRestAPIVersion version, @Nonnull HandlerRequest<EmptyRequestBody> request) {
+ // Parse the parameters
+ SessionHandle sessionHandle = request.getPathParameter(SessionHandleIdPathParameter.class);
+ OperationHandle operationHandle =
+ request.getPathParameter(OperationHandleIdPathParameter.class);
+ Long token = request.getPathParameter(FetchResultsTokenPathParameter.class);
+
+ // Get the statement results
+ @Nullable ResultSet resultSet;
+ @Nullable String resultType;
+ Long nextToken;
+
+ try {
+ resultSet =
+ service.fetchResults(sessionHandle, operationHandle, token, Integer.MAX_VALUE);
+ nextToken = resultSet.getNextToken();
+ resultType = resultSet.getResultType().toString();
+ } catch (Exception e) {
+ Throwable root = e;
+ while (root.getCause() != null
+ && root.getCause().getMessage() != null
+ && !root.getCause().getMessage().isEmpty()) {
+ root = root.getCause();
+ }
+ throw new SqlGatewayException(root);
Review Comment:
I think we should preserve the complete error stack.
```suggestion
throw new SqlGatewayException(e);
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] WencongLiu commented on a diff in pull request #20451: [FLINK-28162][sql-gateway]Introduce the operation related API for REST endpoint
Posted by GitBox <gi...@apache.org>.
WencongLiu commented on code in PR #20451:
URL: https://github.com/apache/flink/pull/20451#discussion_r939503076
##########
flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/rest/message/statement/FetchResultsResponseBody.java:
##########
@@ -0,0 +1,75 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.gateway.rest.message.statement;
+
+import org.apache.flink.runtime.rest.messages.ResponseBody;
+import org.apache.flink.table.gateway.api.results.ExceptionInfo;
+import org.apache.flink.table.gateway.api.results.ResultSet;
+
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonInclude;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
+
+/** {@link ResponseBody} for execute a statement. */
+@JsonInclude(JsonInclude.Include.NON_NULL)
+public class FetchResultsResponseBody implements ResponseBody {
+
+ private static final String FIELD_RESULT_TYPE = "result_type";
+ private static final String FIELD_RESULTS = "results";
+ private static final String FIELD_NEXT_RESULT_URI = "next_result_uri";
+ private static final String FIELD_EXCEPTION = "exception";
+
+ @JsonProperty(FIELD_RESULTS)
+ private final ResultSet results;
+
+ @JsonProperty(FIELD_RESULT_TYPE)
+ private final String resultType;
+
+ @JsonProperty(FIELD_NEXT_RESULT_URI)
+ private final String nextResultUri;
+
+ @JsonProperty(FIELD_EXCEPTION)
+ private final ExceptionInfo exceptionInfo;
Review Comment:
These two fields should not be nullable.
If there is no more data, `nextResultUri ` should be an empty string `""`.
If there is no exception throwed, all fields in 'exceptionInfo' should be null.
The content of these two field in the response will be
"next_result_uri": "".
"exception": {
"root_cause": null,
"exception_stack": null
}
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot commented on pull request #20451: [FLINK-28162][sql-gateway]Introduce the operation related API for REST endpoint
Posted by GitBox <gi...@apache.org>.
flinkbot commented on PR #20451:
URL: https://github.com/apache/flink/pull/20451#issuecomment-1204999675
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "702c2a8eba94ff74bff699b600e93233d1a02338",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "702c2a8eba94ff74bff699b600e93233d1a02338",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 702c2a8eba94ff74bff699b600e93233d1a02338 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] xintongsong commented on a diff in pull request #20451: [FLINK-28162][sql-gateway]Introduce the operation related API for REST endpoint
Posted by GitBox <gi...@apache.org>.
xintongsong commented on code in PR #20451:
URL: https://github.com/apache/flink/pull/20451#discussion_r938642247
##########
flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/rest/message/statement/FetchResultsRequestBody.java:
##########
@@ -0,0 +1,47 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.gateway.rest.message.statement;
+
+import org.apache.flink.runtime.rest.messages.RequestBody;
+
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonInclude;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
+
+import javax.annotation.Nullable;
+
+/** {@link RequestBody} for fetching job result. */
+@JsonInclude(JsonInclude.Include.NON_NULL)
+public class FetchResultsRequestBody implements RequestBody {
+
+ private static final String FIELD_NAME_MAX_FETCH_SIZE = "max_fetch_size";
Review Comment:
@zentol, thanks a lot for the pointer.
@WencongLiu, let's also add a hotfix commit and fix this for the APIs we added in FLINK-28161.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] WencongLiu commented on a diff in pull request #20451: [FLINK-28162][sql-gateway]Introduce the operation related API for REST endpoint
Posted by GitBox <gi...@apache.org>.
WencongLiu commented on code in PR #20451:
URL: https://github.com/apache/flink/pull/20451#discussion_r939500886
##########
flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/rest/handler/statement/ExecuteStatementHandler.java:
##########
@@ -0,0 +1,89 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.gateway.rest.handler.statement;
+
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.runtime.rest.handler.HandlerRequest;
+import org.apache.flink.runtime.rest.messages.MessageHeaders;
+import org.apache.flink.table.gateway.api.SqlGatewayService;
+import org.apache.flink.table.gateway.api.operation.OperationHandle;
+import org.apache.flink.table.gateway.api.session.SessionHandle;
+import org.apache.flink.table.gateway.rest.handler.AbstractSqlGatewayRestHandler;
+import org.apache.flink.table.gateway.rest.message.session.SessionHandleIdPathParameter;
+import org.apache.flink.table.gateway.rest.message.session.SessionMessageParameters;
+import org.apache.flink.table.gateway.rest.message.statement.ExecuteStatementRequestBody;
+import org.apache.flink.table.gateway.rest.message.statement.ExecuteStatementResponseBody;
+import org.apache.flink.table.gateway.rest.util.SqlGatewayRestAPIVersion;
+
+import javax.annotation.Nonnull;
+
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+
+/** Handler to execute a statement. */
+public class ExecuteStatementHandler
+ extends AbstractSqlGatewayRestHandler<
+ ExecuteStatementRequestBody,
+ ExecuteStatementResponseBody,
+ SessionMessageParameters> {
+
+ public ExecuteStatementHandler(
+ SqlGatewayService service,
+ Map<String, String> responseHeaders,
+ MessageHeaders<
+ ExecuteStatementRequestBody,
+ ExecuteStatementResponseBody,
+ SessionMessageParameters>
+ messageHeaders) {
+ super(service, responseHeaders, messageHeaders);
+ }
+
+ @Override
+ protected CompletableFuture<ExecuteStatementResponseBody> handleRequest(
+ SqlGatewayRestAPIVersion version,
+ @Nonnull HandlerRequest<ExecuteStatementRequestBody> request) {
+ String statement = request.getRequestBody().getStatement();
+ Long timeout = request.getRequestBody().getTimeout();
+ timeout = timeout == null ? 0L : timeout;
+ SessionHandle sessionHandle = request.getPathParameter(SessionHandleIdPathParameter.class);
+ OperationHandle operationHandle =
+ service.executeStatement(sessionHandle, statement, timeout, new Configuration());
Review Comment:
Have add a `Map<String, String>` in `ExecuteStatementResponseBody` as the value of executionConfig.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] WencongLiu commented on a diff in pull request #20451: [FLINK-28162][sql-gateway]Introduce the operation related API for REST endpoint
Posted by GitBox <gi...@apache.org>.
WencongLiu commented on code in PR #20451:
URL: https://github.com/apache/flink/pull/20451#discussion_r939501719
##########
flink-table/flink-sql-gateway-api/src/main/java/org/apache/flink/table/gateway/api/results/ExceptionInfo.java:
##########
@@ -0,0 +1,64 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.gateway.api.results;
+
+import org.apache.flink.util.Preconditions;
+
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
+
+/** ExceptionInfo for SQL gateway service. */
+public class ExceptionInfo {
+
+ private static final String FIELD_EXCEPTION_ROOT_CLASS_NAME = "root";
+ private static final String FIELD_EXCEPTION_MESSAGE = "message";
Review Comment:
Fixed.
The ExceptionInfo contains the same attributes as the FLIP.
"exception": {
"rootCause": "....",
"exceptionStack": "..."
}
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] WencongLiu commented on pull request #20451: [FLINK-28162][sql-gateway]Introduce the operation related API for REST endpoint
Posted by GitBox <gi...@apache.org>.
WencongLiu commented on PR #20451:
URL: https://github.com/apache/flink/pull/20451#issuecomment-1208260761
@flinkbot run azure
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] xintongsong commented on pull request #20451: [FLINK-28162][sql-gateway]Introduce the operation related API for REST endpoint
Posted by GitBox <gi...@apache.org>.
xintongsong commented on PR #20451:
URL: https://github.com/apache/flink/pull/20451#issuecomment-1208270030
@flinkbot run azure
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] xintongsong closed pull request #20451: [FLINK-28162][sql-gateway]Introduce the operation related API for REST endpoint
Posted by GitBox <gi...@apache.org>.
xintongsong closed pull request #20451: [FLINK-28162][sql-gateway]Introduce the operation related API for REST endpoint
URL: https://github.com/apache/flink/pull/20451
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] WencongLiu commented on pull request #20451: [FLINK-28162][sql-gateway]Introduce the operation related API for REST endpoint
Posted by GitBox <gi...@apache.org>.
WencongLiu commented on PR #20451:
URL: https://github.com/apache/flink/pull/20451#issuecomment-1207604935
@flinkbot run azure
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org