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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
