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