JackieTien97 commented on code in PR #18035: URL: https://github.com/apache/iotdb/pull/18035#discussion_r3479692122
########## iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/udf/InternalQueryExecutor.java: ########## @@ -0,0 +1,125 @@ +/* + * 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.iotdb.db.queryengine.udf; + +import org.apache.iotdb.commons.exception.IoTDBException; +import org.apache.iotdb.commons.exception.SemanticException; +import org.apache.iotdb.commons.queryengine.common.SessionInfo; +import org.apache.iotdb.commons.queryengine.plan.relational.sql.ast.Statement; +import org.apache.iotdb.db.protocol.session.IClientSession; +import org.apache.iotdb.db.protocol.session.InternalClientSession; +import org.apache.iotdb.db.protocol.session.SessionManager; +import org.apache.iotdb.db.protocol.thrift.impl.ClientRPCServiceImpl; +import org.apache.iotdb.db.queryengine.common.QueryId; +import org.apache.iotdb.db.queryengine.plan.Coordinator; +import org.apache.iotdb.db.queryengine.plan.analyze.QueryType; +import org.apache.iotdb.db.queryengine.plan.execution.ExecutionResult; +import org.apache.iotdb.db.queryengine.plan.execution.IQueryExecution; +import org.apache.iotdb.db.queryengine.plan.planner.LocalExecutionPlanner; +import org.apache.iotdb.db.queryengine.plan.relational.metadata.Metadata; +import org.apache.iotdb.db.queryengine.plan.relational.sql.parser.SqlParser; +import org.apache.iotdb.rpc.TSStatusCode; + +/** Stateless utility for UDF embedded table-model queries. */ +public final class InternalQueryExecutor { + + private static final Coordinator COORDINATOR = Coordinator.getInstance(); + private static final SessionManager SESSION_MANAGER = SessionManager.getInstance(); + private static final Metadata METADATA = LocalExecutionPlanner.getInstance().metadata; + private static final SqlParser RELATION_SQL_PARSER = new SqlParser(); + + private InternalQueryExecutor() {} + + public static InternalQueryResult executeInternalQuery( + SessionInfo sessionInfo, + String fragmentInstanceId, + QueryId outerQueryId, + String sql, + long timeoutMs) + throws IoTDBException { + + IClientSession previousSession = SESSION_MANAGER.getCurrSession(); + + InternalClientSession internalSession = + new InternalClientSession(formatInternalClientId(fragmentInstanceId, outerQueryId)); + internalSession.setSqlDialect(sessionInfo.getSqlDialect()); + sessionInfo.getDatabaseName().ifPresent(internalSession::setDatabaseName); + + SESSION_MANAGER.supplySession( + internalSession, + sessionInfo.getUserId(), + sessionInfo.getUserName(), + sessionInfo.getZoneId(), + sessionInfo.getVersion()); + + long statementId = -1; + long queryId = -1; + try { + SESSION_MANAGER.exchangeCurrSession(internalSession); + + statementId = SESSION_MANAGER.requestStatementId(internalSession); + queryId = SESSION_MANAGER.requestQueryId(internalSession, statementId); + + Statement parsedStatement = + RELATION_SQL_PARSER.createStatement(sql, sessionInfo.getZoneId(), internalSession); + + ExecutionResult result = + COORDINATOR.executeForTableModel( + parsedStatement, + RELATION_SQL_PARSER, + internalSession, + queryId, + sessionInfo, + sql, + METADATA, + timeoutMs, + false, + false); + + if (result.status.code != TSStatusCode.SUCCESS_STATUS.getStatusCode()) { + throw new IoTDBException(result.status.message, result.status.code); + } + + IQueryExecution queryExecution = COORDINATOR.getQueryExecution(queryId); + if (queryExecution == null) { + throw new IoTDBException( + "Internal query execution not found", + TSStatusCode.INTERNAL_SERVER_ERROR.getStatusCode()); + } + + return new InternalQueryResult(queryExecution, internalSession, statementId, queryId, sql); + } catch (Exception e) { Review Comment: catch IoTDBException and IoTDBRuntimeException seprately, and rethrow them instead of using another TSStatusCode.INTERNAL_SERVER_ERROR.getStatusCode() IoTDBException ########## iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/Coordinator.java: ########## @@ -308,6 +308,17 @@ private ExecutionResult execution( boolean userQuery, boolean debug, BiFunction<MPPQueryContext, Long, IQueryExecution> iQueryExecutionFactory) { + return execution(queryId, session, sql, userQuery, debug, false, iQueryExecutionFactory); + } + + private ExecutionResult execution( + long queryId, + SessionInfo session, + String sql, + boolean userQuery, + boolean debug, + boolean readOnlyInternalQuery, Review Comment: It will always be false -- 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]
