You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@paimon.apache.org by "s7monk (via GitHub)" <gi...@apache.org> on 2023/12/01 16:33:13 UTC
[PR] [Feature] Job Submission - Refreshed Baseline Implementation [incubator-paimon-webui]
s7monk opened a new pull request, #107:
URL: https://github.com/apache/incubator-paimon-webui/pull/107
<!-- Please specify the module before the PR name: [core] ... or [flink] ... -->
### Purpose
Job submit baseline implementation.
### Tests
<!-- List UT and IT cases to verify this change -->
### API and Format
<!-- Does this change affect API or storage format -->
### Documentation
<!-- Does this change introduce a new feature -->
--
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@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature] Job Submission - Refreshed Baseline Implementation [incubator-paimon-webui]
Posted by "s7monk (via GitHub)" <gi...@apache.org>.
s7monk commented on code in PR #107:
URL: https://github.com/apache/incubator-paimon-webui/pull/107#discussion_r1428819469
##########
paimon-web-flink/src/main/java/org/apache/paimon/web/flink/executor/FlinkExecutor.java:
##########
@@ -0,0 +1,156 @@
+/*
+ * 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.paimon.web.flink.executor;
+
+import org.apache.paimon.web.common.executor.Executor;
+import org.apache.paimon.web.common.result.FetchResultParams;
+import org.apache.paimon.web.common.result.SubmitResult;
+import org.apache.paimon.web.flink.exception.SqlExecutionException;
+import org.apache.paimon.web.flink.operation.FlinkSqlOperationType;
+import org.apache.paimon.web.flink.parser.StatementParser;
+import org.apache.paimon.web.flink.utils.CollectResultUtil;
+import org.apache.paimon.web.flink.utils.FormatSqlExceptionUtil;
+
+import org.apache.commons.collections.CollectionUtils;
+import org.apache.flink.core.execution.JobClient;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.table.api.StatementSet;
+import org.apache.flink.table.api.TableEnvironment;
+import org.apache.flink.table.api.TableResult;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Optional;
+
+/** The flink implementation of the {@link Executor}. */
+public class FlinkExecutor implements Executor {
+
+ private static final String EXECUTE_SUCCESS = "OK";
+ private final StreamExecutionEnvironment env;
+ private final TableEnvironment tableEnv;
+
+ public FlinkExecutor(StreamExecutionEnvironment env, TableEnvironment tableEnv) {
+ this.env = env;
+ this.tableEnv = tableEnv;
+ }
+
+ @Override
+ public SubmitResult executeSql(String multiStatement) throws SqlExecutionException {
+ List<String> insertStatements = new ArrayList<>();
+ String[] statements = StatementParser.parse(multiStatement);
+ for (String statement : statements) {
+ FlinkSqlOperationType operationType = FlinkSqlOperationType.getOperationType(statement);
+ switch (operationType.getCategory()) {
+ case DQL:
+ if (insertStatements.isEmpty()) {
Review Comment:
The statements passed by the sql editor can only contain one of DQL and Insert. If there is an insert statement in front of it, the DQL statement cannot be executed, so it is judged here first.
--
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@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature] Job Submission - Refreshed Baseline Implementation [incubator-paimon-webui]
Posted by "s7monk (via GitHub)" <gi...@apache.org>.
s7monk commented on code in PR #107:
URL: https://github.com/apache/incubator-paimon-webui/pull/107#discussion_r1428799616
##########
paimon-web-common/pom.xml:
##########
@@ -38,11 +38,6 @@ under the License.
</properties>
<dependencies>
- <dependency>
Review Comment:
Thank you for bringing this to my attention. You're correct; this part isn't related to the current PR. I will take the revert 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@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature] Job Submission - Refreshed Baseline Implementation [incubator-paimon-webui]
Posted by "s7monk (via GitHub)" <gi...@apache.org>.
s7monk commented on code in PR #107:
URL: https://github.com/apache/incubator-paimon-webui/pull/107#discussion_r1428806277
##########
paimon-web-common/src/main/java/org/apache/paimon/web/common/executor/Executor.java:
##########
@@ -0,0 +1,32 @@
+/*
+ * 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.paimon.web.common.executor;
+
+import org.apache.paimon.web.common.result.FetchResultParams;
+import org.apache.paimon.web.common.result.SubmitResult;
+
+/** The Executor interface. */
+public interface Executor {
+
+ SubmitResult executeSql(String statement) throws Exception;
+
+ SubmitResult fetchResults(FetchResultParams params) throws Exception;
+
+ boolean stop(String jobId, boolean withSavepoint, boolean withDrain) throws Exception;
Review Comment:
Spark's stop method may not need withSavepoint and withDrain, I will pass a variable parameter.
--
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@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature] Job Submission - Refreshed Baseline Implementation [incubator-paimon-webui]
Posted by "s7monk (via GitHub)" <gi...@apache.org>.
s7monk commented on code in PR #107:
URL: https://github.com/apache/incubator-paimon-webui/pull/107#discussion_r1428797325
##########
paimon-web-common/pom.xml:
##########
@@ -38,11 +38,6 @@ under the License.
</properties>
<dependencies>
- <dependency>
Review Comment:
Thank you for bringing this to my attention. You're correct; this part isn't related to the current PR. I will take the revert it.
##########
paimon-web-common/pom.xml:
##########
@@ -38,11 +38,6 @@ under the License.
</properties>
<dependencies>
- <dependency>
Review Comment:
Thank you for bringing this to my attention. You're correct; this part isn't related to the current PR. I will take the revert 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@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature] Job Submission - Refreshed Baseline Implementation [incubator-paimon-webui]
Posted by "s7monk (via GitHub)" <gi...@apache.org>.
s7monk commented on code in PR #107:
URL: https://github.com/apache/incubator-paimon-webui/pull/107#discussion_r1428815907
##########
paimon-web-flink/src/main/java/org/apache/paimon/web/flink/executor/FlinkExecutorFactory.java:
##########
@@ -0,0 +1,57 @@
+/*
+ * 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.paimon.web.flink.executor;
+
+import org.apache.paimon.web.common.executor.Executor;
+import org.apache.paimon.web.common.executor.ExecutorFactory;
+
+import org.apache.flink.api.common.RuntimeExecutionMode;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.table.api.EnvironmentSettings;
+import org.apache.flink.table.api.TableEnvironment;
+import org.apache.flink.table.api.bridge.java.internal.StreamTableEnvironmentImpl;
+
+/** Factory class for creating executors that interface with the Flink Table API. */
+public class FlinkExecutorFactory implements ExecutorFactory {
+
+ private final RuntimeExecutionMode mode;
+
+ private final Configuration configuration;
+
+ public FlinkExecutorFactory(RuntimeExecutionMode mode, Configuration configuration) {
+ this.mode = mode;
+ this.configuration = configuration;
+ }
+
+ @Override
+ public Executor createExecutor() {
+ EnvironmentSettings environmentSettings = getEnvironmentSettings(mode);
+ StreamExecutionEnvironment env =
+ StreamExecutionEnvironment.getExecutionEnvironment(configuration);
+ TableEnvironment tableEvn = StreamTableEnvironmentImpl.create(env, environmentSettings);
+ return new FlinkExecutor(env, tableEvn);
+ }
+
+ private EnvironmentSettings getEnvironmentSettings(RuntimeExecutionMode mode) {
+ return mode == RuntimeExecutionMode.BATCH
Review Comment:
I ignored this, I will modify 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@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature] Job Submission - Refreshed Baseline Implementation [incubator-paimon-webui]
Posted by "s7monk (via GitHub)" <gi...@apache.org>.
s7monk commented on code in PR #107:
URL: https://github.com/apache/incubator-paimon-webui/pull/107#discussion_r1428809475
##########
paimon-web-common/src/main/java/org/apache/paimon/web/common/function/Supplier.java:
##########
@@ -0,0 +1,25 @@
+/*
+ * 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.paimon.web.common.function;
+
+/** Supplier Functional interface that can use exceptions. */
+@FunctionalInterface
+public interface Supplier<T> {
Review Comment:
OK, i will remove this.
--
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@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature] Job Submission - Refreshed Baseline Implementation [incubator-paimon-webui]
Posted by "hdygxsj (via GitHub)" <gi...@apache.org>.
hdygxsj commented on code in PR #107:
URL: https://github.com/apache/incubator-paimon-webui/pull/107#discussion_r1415631152
##########
paimon-web-server/src/main/java/org/apache/paimon/web/server/controller/SessionController.java:
##########
@@ -0,0 +1,115 @@
+/*
+ * 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.paimon.web.server.controller;
+
+import org.apache.paimon.web.server.data.dto.SessionDTO;
+import org.apache.paimon.web.server.data.result.R;
+import org.apache.paimon.web.server.data.result.enums.Status;
+import org.apache.paimon.web.server.data.vo.SessionVO;
+import org.apache.paimon.web.server.service.SessionService;
+
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.PostMapping;
+import org.springframework.web.bind.annotation.RequestBody;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RestController;
+
+import java.util.List;
+
+/** Session api controller. */
+@Slf4j
+@RestController
+@RequestMapping("/api/session")
+public class SessionController {
+
+ @Autowired private SessionService sessionService;
+
+ /**
+ * Creates a session based on the provided session data transfer object (DTO).
+ *
+ * @param sessionDTO the session DTO containing the necessary data to create a session
+ * @return a response entity indicating success or failure
+ */
Review Comment:
How about adding a daemon thread to scan the status of the session
##########
paimon-web-server/src/main/java/org/apache/paimon/web/server/controller/SessionController.java:
##########
@@ -0,0 +1,115 @@
+/*
+ * 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.paimon.web.server.controller;
+
+import org.apache.paimon.web.server.data.dto.SessionDTO;
+import org.apache.paimon.web.server.data.result.R;
+import org.apache.paimon.web.server.data.result.enums.Status;
+import org.apache.paimon.web.server.data.vo.SessionVO;
+import org.apache.paimon.web.server.service.SessionService;
+
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.PostMapping;
+import org.springframework.web.bind.annotation.RequestBody;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RestController;
+
+import java.util.List;
+
+/** Session api controller. */
+@Slf4j
+@RestController
+@RequestMapping("/api/session")
+public class SessionController {
+
+ @Autowired private SessionService sessionService;
+
+ /**
+ * Creates a session based on the provided session data transfer object (DTO).
+ *
+ * @param sessionDTO the session DTO containing the necessary data to create a session
+ * @return a response entity indicating success or failure
+ */
Review Comment:
How about adding a daemon thread to scan the status of the session
--
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@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature] Job Submission - Refreshed Baseline Implementation [incubator-paimon-webui]
Posted by "s7monk (via GitHub)" <gi...@apache.org>.
s7monk commented on code in PR #107:
URL: https://github.com/apache/incubator-paimon-webui/pull/107#discussion_r1428814345
##########
paimon-web-common/src/main/java/org/apache/paimon/web/common/result/SubmitResult.java:
##########
@@ -0,0 +1,129 @@
+/*
+ * 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.paimon.web.common.result;
+
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+/** This class represents the result of a job. Including flink job and spark Job. */
+public class SubmitResult implements Serializable {
+
+ private static final long serialVersionUID = 1L;
+
+ private final String submitId;
Review Comment:
If it is sql gateway, submitId will store operationHandle in order to obtain the results, and others will store a uuid.
--
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@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature] Job Submission - Refreshed Baseline Implementation [incubator-paimon-webui]
Posted by "s7monk (via GitHub)" <gi...@apache.org>.
s7monk commented on code in PR #107:
URL: https://github.com/apache/incubator-paimon-webui/pull/107#discussion_r1428812950
##########
paimon-web-common/src/main/java/org/apache/paimon/web/common/result/SubmitResult.java:
##########
@@ -0,0 +1,129 @@
+/*
+ * 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.paimon.web.common.result;
+
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+/** This class represents the result of a job. Including flink job and spark Job. */
Review Comment:
ok.
--
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@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature] Job Submission - Refreshed Baseline Implementation [incubator-paimon-webui]
Posted by "s7monk (via GitHub)" <gi...@apache.org>.
s7monk commented on PR #107:
URL: https://github.com/apache/incubator-paimon-webui/pull/107#issuecomment-1839193377
@SteNicholas @JingsongLi PTAL.
--
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@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature] Job Submission - Refreshed Baseline Implementation [incubator-paimon-webui]
Posted by "s7monk (via GitHub)" <gi...@apache.org>.
s7monk commented on code in PR #107:
URL: https://github.com/apache/incubator-paimon-webui/pull/107#discussion_r1428797325
##########
paimon-web-common/pom.xml:
##########
@@ -38,11 +38,6 @@ under the License.
</properties>
<dependencies>
- <dependency>
Review Comment:
Thank you for bringing this to my attention. You're correct; this part isn't related to the current PR. I will take the revert it.
##########
paimon-web-common/pom.xml:
##########
@@ -38,11 +38,6 @@ under the License.
</properties>
<dependencies>
- <dependency>
Review Comment:
Thank you for bringing this to my attention. You're correct; this part isn't related to the current PR. I will take the revert 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@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature] Job Submission - Refreshed Baseline Implementation [incubator-paimon-webui]
Posted by "s7monk (via GitHub)" <gi...@apache.org>.
s7monk commented on code in PR #107:
URL: https://github.com/apache/incubator-paimon-webui/pull/107#discussion_r1428809690
##########
paimon-web-common/src/main/java/org/apache/paimon/web/common/result/FetchResultParams.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.paimon.web.common.result;
+
+/** Represents the parameters required to fetch the results of a certain operation. */
+public class FetchResultParams {
+
+ private final String sessionId;
+ private final String submitId;
+ private final Long token;
+
+ private FetchResultParams(Builder builder) {
Review Comment:
Yes, I will modify this class.
--
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@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature] Job Submission - Refreshed Baseline Implementation [incubator-paimon-webui]
Posted by "SteNicholas (via GitHub)" <gi...@apache.org>.
SteNicholas commented on code in PR #107:
URL: https://github.com/apache/incubator-paimon-webui/pull/107#discussion_r1438071048
##########
paimon-web-flink/src/main/java/org/apache/paimon/web/flink/executor/FlinkExecutor.java:
##########
@@ -0,0 +1,156 @@
+/*
+ * 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.paimon.web.flink.executor;
+
+import org.apache.paimon.web.common.executor.Executor;
+import org.apache.paimon.web.common.result.FetchResultParams;
+import org.apache.paimon.web.common.result.SubmitResult;
+import org.apache.paimon.web.flink.exception.SqlExecutionException;
+import org.apache.paimon.web.flink.operation.FlinkSqlOperationType;
+import org.apache.paimon.web.flink.parser.StatementParser;
+import org.apache.paimon.web.flink.utils.CollectResultUtil;
+import org.apache.paimon.web.flink.utils.FormatSqlExceptionUtil;
+
+import org.apache.commons.collections.CollectionUtils;
+import org.apache.flink.core.execution.JobClient;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.table.api.StatementSet;
+import org.apache.flink.table.api.TableEnvironment;
+import org.apache.flink.table.api.TableResult;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Optional;
+
+/** The flink implementation of the {@link Executor}. */
+public class FlinkExecutor implements Executor {
+
+ private static final String EXECUTE_SUCCESS = "OK";
+ private final StreamExecutionEnvironment env;
+ private final TableEnvironment tableEnv;
+
+ public FlinkExecutor(StreamExecutionEnvironment env, TableEnvironment tableEnv) {
+ this.env = env;
+ this.tableEnv = tableEnv;
+ }
+
+ @Override
+ public SubmitResult executeSql(String multiStatement) throws SqlExecutionException {
+ List<String> insertStatements = new ArrayList<>();
+ String[] statements = StatementParser.parse(multiStatement);
+ for (String statement : statements) {
+ FlinkSqlOperationType operationType = FlinkSqlOperationType.getOperationType(statement);
+ switch (operationType.getCategory()) {
+ case DQL:
+ if (insertStatements.isEmpty()) {
+ return executeQueryStatement(statement);
+ }
+ break;
+ case DML:
+ if (operationType.getType().equals(FlinkSqlOperationType.INSERT.getType())) {
+ insertStatements.add(statement);
+ } else if (insertStatements.isEmpty()) {
+ return executeDmlStatement(statement);
+ }
+ break;
+ default:
+ executeStatementWithoutResult(statement);
+ break;
+ }
+ }
+ return executeInsertStatements(insertStatements);
Review Comment:
@s7monk, IMO, this could merge into DML branch, no need to distiguish here.
--
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@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature] Job Submission - Refreshed Baseline Implementation [incubator-paimon-webui]
Posted by "s7monk (via GitHub)" <gi...@apache.org>.
s7monk closed pull request #107: [Feature] Job Submission - Refreshed Baseline Implementation
URL: https://github.com/apache/incubator-paimon-webui/pull/107
--
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@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature] Job Submission - Refreshed Baseline Implementation [incubator-paimon-webui]
Posted by "SteNicholas (via GitHub)" <gi...@apache.org>.
SteNicholas commented on code in PR #107:
URL: https://github.com/apache/incubator-paimon-webui/pull/107#discussion_r1438072311
##########
paimon-web-flink/src/main/java/org/apache/paimon/web/flink/executor/FlinkExecutor.java:
##########
@@ -0,0 +1,157 @@
+/*
+ * 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.paimon.web.flink.executor;
+
+import org.apache.paimon.web.common.executor.Executor;
+import org.apache.paimon.web.common.result.FetchResultParams;
+import org.apache.paimon.web.common.result.SubmissionResult;
+import org.apache.paimon.web.flink.exception.SqlExecutionException;
+import org.apache.paimon.web.flink.operation.FlinkSqlOperationType;
+import org.apache.paimon.web.flink.parser.StatementParser;
+import org.apache.paimon.web.flink.utils.CollectResultUtil;
+import org.apache.paimon.web.flink.utils.FormatSqlExceptionUtil;
+
+import org.apache.commons.collections.CollectionUtils;
+import org.apache.flink.core.execution.JobClient;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.table.api.StatementSet;
+import org.apache.flink.table.api.TableEnvironment;
+import org.apache.flink.table.api.TableResult;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Optional;
+
+/** The flink implementation of the {@link Executor}. */
+public class FlinkExecutor implements Executor {
+
+ private static final String EXECUTE_SUCCESS = "OK";
Review Comment:
Where is the execution failure status?
--
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@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature] Job Submission - Refreshed Baseline Implementation [incubator-paimon-webui]
Posted by "SteNicholas (via GitHub)" <gi...@apache.org>.
SteNicholas commented on code in PR #107:
URL: https://github.com/apache/incubator-paimon-webui/pull/107#discussion_r1428776828
##########
paimon-web-common/src/main/java/org/apache/paimon/web/common/executor/Executor.java:
##########
@@ -0,0 +1,32 @@
+/*
+ * 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.paimon.web.common.executor;
+
+import org.apache.paimon.web.common.result.FetchResultParams;
+import org.apache.paimon.web.common.result.SubmitResult;
+
+/** The Executor interface. */
Review Comment:
Could you describe this interface more detailed?
##########
paimon-web-common/src/main/java/org/apache/paimon/web/common/executor/Executor.java:
##########
@@ -0,0 +1,32 @@
+/*
+ * 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.paimon.web.common.executor;
+
+import org.apache.paimon.web.common.result.FetchResultParams;
+import org.apache.paimon.web.common.result.SubmitResult;
+
+/** The Executor interface. */
+public interface Executor {
+
+ SubmitResult executeSql(String statement) throws Exception;
+
+ SubmitResult fetchResults(FetchResultParams params) throws Exception;
+
+ boolean stop(String jobId, boolean withSavepoint, boolean withDrain) throws Exception;
Review Comment:
Is the executor only defined for Flink job? What's the behavior of Spark job for `withSavepoint` and `withDrain` param?
##########
paimon-web-flink/src/main/java/org/apache/paimon/web/flink/executor/FlinkExecutorFactory.java:
##########
@@ -0,0 +1,57 @@
+/*
+ * 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.paimon.web.flink.executor;
+
+import org.apache.paimon.web.common.executor.Executor;
+import org.apache.paimon.web.common.executor.ExecutorFactory;
+
+import org.apache.flink.api.common.RuntimeExecutionMode;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.table.api.EnvironmentSettings;
+import org.apache.flink.table.api.TableEnvironment;
+import org.apache.flink.table.api.bridge.java.internal.StreamTableEnvironmentImpl;
+
+/** Factory class for creating executors that interface with the Flink Table API. */
+public class FlinkExecutorFactory implements ExecutorFactory {
+
+ private final RuntimeExecutionMode mode;
+
+ private final Configuration configuration;
+
+ public FlinkExecutorFactory(RuntimeExecutionMode mode, Configuration configuration) {
+ this.mode = mode;
+ this.configuration = configuration;
+ }
+
+ @Override
+ public Executor createExecutor() {
+ EnvironmentSettings environmentSettings = getEnvironmentSettings(mode);
+ StreamExecutionEnvironment env =
+ StreamExecutionEnvironment.getExecutionEnvironment(configuration);
+ TableEnvironment tableEvn = StreamTableEnvironmentImpl.create(env, environmentSettings);
+ return new FlinkExecutor(env, tableEvn);
+ }
+
+ private EnvironmentSettings getEnvironmentSettings(RuntimeExecutionMode mode) {
+ return mode == RuntimeExecutionMode.BATCH
Review Comment:
What's the behavior of `RuntimeExecutionMode.DYNAMIC`?
##########
paimon-web-common/src/main/java/org/apache/paimon/web/common/executor/Executor.java:
##########
@@ -0,0 +1,32 @@
+/*
+ * 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.paimon.web.common.executor;
+
+import org.apache.paimon.web.common.result.FetchResultParams;
+import org.apache.paimon.web.common.result.SubmitResult;
+
+/** The Executor interface. */
+public interface Executor {
+
+ SubmitResult executeSql(String statement) throws Exception;
Review Comment:
Add the comment of all interface methods.
##########
paimon-web-common/src/main/java/org/apache/paimon/web/common/result/SubmitResult.java:
##########
@@ -0,0 +1,129 @@
+/*
+ * 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.paimon.web.common.result;
+
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+/** This class represents the result of a job. Including flink job and spark Job. */
Review Comment:
Why `SubmitRsult` represents the result of a job? IMO, the `SubmitRsult` only means the submission result of job. BTW, the `Including flink job and spark Job.` could be removed.
##########
paimon-web-common/src/main/java/org/apache/paimon/web/common/function/Supplier.java:
##########
@@ -0,0 +1,25 @@
+/*
+ * 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.paimon.web.common.function;
+
+/** Supplier Functional interface that can use exceptions. */
+@FunctionalInterface
+public interface Supplier<T> {
Review Comment:
Why introduce this interface? Why not use `java.util.function.Supplier`?
##########
paimon-web-common/src/main/java/org/apache/paimon/web/common/result/SubmitResult.java:
##########
@@ -0,0 +1,129 @@
+/*
+ * 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.paimon.web.common.result;
+
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+/** This class represents the result of a job. Including flink job and spark Job. */
+public class SubmitResult implements Serializable {
+
+ private static final long serialVersionUID = 1L;
+
+ private final String submitId;
+ private final String jobId;
+ private final String status;
+ private final List<Map<String, Object>> data;
+ private final boolean shouldFetchResult;
+
+ private SubmitResult(Builder builder) {
+ this.jobId = builder.jobId;
+ this.status = builder.status;
+ this.data = builder.data;
+ this.submitId = builder.submitId;
+ this.shouldFetchResult = builder.shouldFetchResult;
+ }
+
+ public String getSubmitId() {
+ return submitId;
+ }
+
+ public String getJobId() {
+ return jobId;
+ }
+
+ public String getStatus() {
+ return status;
+ }
+
+ public List<Map<String, Object>> getData() {
+ return data;
+ }
+
+ public boolean shouldFetchResult() {
+ return shouldFetchResult;
+ }
+
+ public static Builder builder() {
+ return new Builder();
+ }
+
+ /** The builder for SubmitResult. */
+ public static class Builder {
+
+ private String submitId;
+ private String jobId;
+ private String status;
+ private List<Map<String, Object>> data = new ArrayList<>();
+ private boolean shouldFetchResult;
+
+ public Builder submitId(String submitId) {
+ this.submitId = submitId;
+ return this;
+ }
+
+ public Builder jobId(String jobId) {
+ this.jobId = jobId;
+ return this;
+ }
+
+ public Builder status(String status) {
+ this.status = status;
+ return this;
+ }
+
+ public Builder data(List<Map<String, Object>> data) {
+ this.data = data;
+ return this;
+ }
+
+ public Builder addData(Map<String, Object> dataItem) {
+ this.data.add(dataItem);
+ return this;
+ }
+
+ public Builder shouldFetchResult(boolean shouldFetchResult) {
+ this.shouldFetchResult = shouldFetchResult;
+ return this;
+ }
+
+ public SubmitResult build() {
+ return new SubmitResult(this);
+ }
+ }
+
+ @Override
+ public String toString() {
+ return "SubmitResult{"
Review Comment:
Does the `shouldFetchResult` need to be included in `toString`?
##########
paimon-web-common/pom.xml:
##########
@@ -38,11 +38,6 @@ under the License.
</properties>
<dependencies>
- <dependency>
Review Comment:
Why remove this dependency? Is this change related with job submission?
##########
paimon-web-common/src/main/java/org/apache/paimon/web/common/result/SubmitResult.java:
##########
@@ -0,0 +1,129 @@
+/*
+ * 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.paimon.web.common.result;
+
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+/** This class represents the result of a job. Including flink job and spark Job. */
+public class SubmitResult implements Serializable {
+
+ private static final long serialVersionUID = 1L;
+
+ private final String submitId;
Review Comment:
What's the purpose of definition for `submitId`?
##########
paimon-web-common/src/main/java/org/apache/paimon/web/common/executor/ExecutorFactory.java:
##########
@@ -0,0 +1,25 @@
+/*
+ * 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.paimon.web.common.executor;
+
+/** The ExecutorFactory interface provides a method to create an Executor. */
+public interface ExecutorFactory {
+
+ Executor createExecutor() throws Exception;
Review Comment:
Add the comment of all interface methods.
##########
paimon-web-common/src/main/java/org/apache/paimon/web/common/result/FetchResultParams.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.paimon.web.common.result;
+
+/** Represents the parameters required to fetch the results of a certain operation. */
+public class FetchResultParams {
+
+ private final String sessionId;
+ private final String submitId;
+ private final Long token;
+
+ private FetchResultParams(Builder builder) {
Review Comment:
I don't think the constructor need to use builder as parameter. The builder mode isn't used via this way.
##########
paimon-web-common/src/main/java/org/apache/paimon/web/common/result/SubmitResult.java:
##########
@@ -0,0 +1,129 @@
+/*
+ * 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.paimon.web.common.result;
+
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+/** This class represents the result of a job. Including flink job and spark Job. */
+public class SubmitResult implements Serializable {
Review Comment:
```suggestion
public class SubmissionResult implements Serializable {
```
##########
paimon-web-common/src/main/java/org/apache/paimon/web/common/executor/ExecutorFactory.java:
##########
@@ -0,0 +1,25 @@
+/*
+ * 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.paimon.web.common.executor;
+
+/** The ExecutorFactory interface provides a method to create an Executor. */
Review Comment:
```suggestion
/** This factory is defined to create an {@link Executor} instance. */
```
##########
paimon-web-flink/src/main/java/org/apache/paimon/web/flink/executor/FlinkExecutorFactory.java:
##########
@@ -0,0 +1,57 @@
+/*
+ * 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.paimon.web.flink.executor;
+
+import org.apache.paimon.web.common.executor.Executor;
+import org.apache.paimon.web.common.executor.ExecutorFactory;
+
+import org.apache.flink.api.common.RuntimeExecutionMode;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.table.api.EnvironmentSettings;
+import org.apache.flink.table.api.TableEnvironment;
+import org.apache.flink.table.api.bridge.java.internal.StreamTableEnvironmentImpl;
+
+/** Factory class for creating executors that interface with the Flink Table API. */
+public class FlinkExecutorFactory implements ExecutorFactory {
+
+ private final RuntimeExecutionMode mode;
+
+ private final Configuration configuration;
+
+ public FlinkExecutorFactory(RuntimeExecutionMode mode, Configuration configuration) {
+ this.mode = mode;
+ this.configuration = configuration;
+ }
+
+ @Override
+ public Executor createExecutor() {
+ EnvironmentSettings environmentSettings = getEnvironmentSettings(mode);
+ StreamExecutionEnvironment env =
+ StreamExecutionEnvironment.getExecutionEnvironment(configuration);
+ TableEnvironment tableEvn = StreamTableEnvironmentImpl.create(env, environmentSettings);
Review Comment:
```suggestion
TableEnvironment tableEnv = StreamTableEnvironmentImpl.create(env, environmentSettings);
```
##########
paimon-web-flink/src/main/java/org/apache/paimon/web/flink/exception/SqlExecutionException.java:
##########
@@ -0,0 +1,27 @@
+/*
+ * 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.paimon.web.flink.exception;
+
+/** Exception class for SQL execution issues. */
Review Comment:
```suggestion
/** Exception of SQL execution. */
```
##########
paimon-web-flink/src/main/java/org/apache/paimon/web/flink/executor/FlinkExecutor.java:
##########
@@ -0,0 +1,156 @@
+/*
+ * 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.paimon.web.flink.executor;
+
+import org.apache.paimon.web.common.executor.Executor;
+import org.apache.paimon.web.common.result.FetchResultParams;
+import org.apache.paimon.web.common.result.SubmitResult;
+import org.apache.paimon.web.flink.exception.SqlExecutionException;
+import org.apache.paimon.web.flink.operation.FlinkSqlOperationType;
+import org.apache.paimon.web.flink.parser.StatementParser;
+import org.apache.paimon.web.flink.utils.CollectResultUtil;
+import org.apache.paimon.web.flink.utils.FormatSqlExceptionUtil;
+
+import org.apache.commons.collections.CollectionUtils;
+import org.apache.flink.core.execution.JobClient;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.table.api.StatementSet;
+import org.apache.flink.table.api.TableEnvironment;
+import org.apache.flink.table.api.TableResult;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Optional;
+
+/** The flink implementation of the {@link Executor}. */
+public class FlinkExecutor implements Executor {
+
+ private static final String EXECUTE_SUCCESS = "OK";
+ private final StreamExecutionEnvironment env;
+ private final TableEnvironment tableEnv;
+
+ public FlinkExecutor(StreamExecutionEnvironment env, TableEnvironment tableEnv) {
+ this.env = env;
+ this.tableEnv = tableEnv;
+ }
+
+ @Override
+ public SubmitResult executeSql(String multiStatement) throws SqlExecutionException {
+ List<String> insertStatements = new ArrayList<>();
+ String[] statements = StatementParser.parse(multiStatement);
+ for (String statement : statements) {
+ FlinkSqlOperationType operationType = FlinkSqlOperationType.getOperationType(statement);
+ switch (operationType.getCategory()) {
+ case DQL:
+ if (insertStatements.isEmpty()) {
+ return executeQueryStatement(statement);
+ }
+ break;
+ case DML:
+ if (operationType.getType().equals(FlinkSqlOperationType.INSERT.getType())) {
+ insertStatements.add(statement);
+ } else if (insertStatements.isEmpty()) {
+ return executeDmlStatement(statement);
+ }
+ break;
+ default:
+ executeStatementWithoutResult(statement);
+ break;
+ }
+ }
+ return executeInsertStatements(insertStatements);
Review Comment:
Is insert statement same as DML?
##########
paimon-web-flink/src/main/java/org/apache/paimon/web/flink/executor/FlinkExecutor.java:
##########
@@ -0,0 +1,156 @@
+/*
+ * 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.paimon.web.flink.executor;
+
+import org.apache.paimon.web.common.executor.Executor;
+import org.apache.paimon.web.common.result.FetchResultParams;
+import org.apache.paimon.web.common.result.SubmitResult;
+import org.apache.paimon.web.flink.exception.SqlExecutionException;
+import org.apache.paimon.web.flink.operation.FlinkSqlOperationType;
+import org.apache.paimon.web.flink.parser.StatementParser;
+import org.apache.paimon.web.flink.utils.CollectResultUtil;
+import org.apache.paimon.web.flink.utils.FormatSqlExceptionUtil;
+
+import org.apache.commons.collections.CollectionUtils;
+import org.apache.flink.core.execution.JobClient;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.table.api.StatementSet;
+import org.apache.flink.table.api.TableEnvironment;
+import org.apache.flink.table.api.TableResult;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Optional;
+
+/** The flink implementation of the {@link Executor}. */
+public class FlinkExecutor implements Executor {
+
+ private static final String EXECUTE_SUCCESS = "OK";
+ private final StreamExecutionEnvironment env;
+ private final TableEnvironment tableEnv;
+
+ public FlinkExecutor(StreamExecutionEnvironment env, TableEnvironment tableEnv) {
+ this.env = env;
+ this.tableEnv = tableEnv;
+ }
+
+ @Override
+ public SubmitResult executeSql(String multiStatement) throws SqlExecutionException {
+ List<String> insertStatements = new ArrayList<>();
+ String[] statements = StatementParser.parse(multiStatement);
+ for (String statement : statements) {
+ FlinkSqlOperationType operationType = FlinkSqlOperationType.getOperationType(statement);
+ switch (operationType.getCategory()) {
+ case DQL:
+ if (insertStatements.isEmpty()) {
Review Comment:
Could empty insert statement be handled early?
--
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@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature] Job Submission - Refreshed Baseline Implementation [incubator-paimon-webui]
Posted by "s7monk (via GitHub)" <gi...@apache.org>.
s7monk commented on code in PR #107:
URL: https://github.com/apache/incubator-paimon-webui/pull/107#discussion_r1428797529
##########
paimon-web-common/src/main/java/org/apache/paimon/web/common/executor/Executor.java:
##########
@@ -0,0 +1,32 @@
+/*
+ * 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.paimon.web.common.executor;
+
+import org.apache.paimon.web.common.result.FetchResultParams;
+import org.apache.paimon.web.common.result.SubmitResult;
+
+/** The Executor interface. */
Review Comment:
ok.
--
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@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature] Job Submission - Refreshed Baseline Implementation [incubator-paimon-webui]
Posted by "s7monk (via GitHub)" <gi...@apache.org>.
s7monk commented on code in PR #107:
URL: https://github.com/apache/incubator-paimon-webui/pull/107#discussion_r1428813597
##########
paimon-web-common/src/main/java/org/apache/paimon/web/common/result/SubmitResult.java:
##########
@@ -0,0 +1,129 @@
+/*
+ * 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.paimon.web.common.result;
+
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+/** This class represents the result of a job. Including flink job and spark Job. */
+public class SubmitResult implements Serializable {
+
+ private static final long serialVersionUID = 1L;
+
+ private final String submitId;
+ private final String jobId;
+ private final String status;
+ private final List<Map<String, Object>> data;
+ private final boolean shouldFetchResult;
+
+ private SubmitResult(Builder builder) {
+ this.jobId = builder.jobId;
+ this.status = builder.status;
+ this.data = builder.data;
+ this.submitId = builder.submitId;
+ this.shouldFetchResult = builder.shouldFetchResult;
+ }
+
+ public String getSubmitId() {
+ return submitId;
+ }
+
+ public String getJobId() {
+ return jobId;
+ }
+
+ public String getStatus() {
+ return status;
+ }
+
+ public List<Map<String, Object>> getData() {
+ return data;
+ }
+
+ public boolean shouldFetchResult() {
+ return shouldFetchResult;
+ }
+
+ public static Builder builder() {
+ return new Builder();
+ }
+
+ /** The builder for SubmitResult. */
+ public static class Builder {
+
+ private String submitId;
+ private String jobId;
+ private String status;
+ private List<Map<String, Object>> data = new ArrayList<>();
+ private boolean shouldFetchResult;
+
+ public Builder submitId(String submitId) {
+ this.submitId = submitId;
+ return this;
+ }
+
+ public Builder jobId(String jobId) {
+ this.jobId = jobId;
+ return this;
+ }
+
+ public Builder status(String status) {
+ this.status = status;
+ return this;
+ }
+
+ public Builder data(List<Map<String, Object>> data) {
+ this.data = data;
+ return this;
+ }
+
+ public Builder addData(Map<String, Object> dataItem) {
+ this.data.add(dataItem);
+ return this;
+ }
+
+ public Builder shouldFetchResult(boolean shouldFetchResult) {
+ this.shouldFetchResult = shouldFetchResult;
+ return this;
+ }
+
+ public SubmitResult build() {
+ return new SubmitResult(this);
+ }
+ }
+
+ @Override
+ public String toString() {
+ return "SubmitResult{"
Review Comment:
I will remove.
--
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@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature] Job Submission - Refreshed Baseline Implementation [incubator-paimon-webui]
Posted by "s7monk (via GitHub)" <gi...@apache.org>.
s7monk commented on code in PR #107:
URL: https://github.com/apache/incubator-paimon-webui/pull/107#discussion_r1428820728
##########
paimon-web-flink/src/main/java/org/apache/paimon/web/flink/executor/FlinkExecutor.java:
##########
@@ -0,0 +1,156 @@
+/*
+ * 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.paimon.web.flink.executor;
+
+import org.apache.paimon.web.common.executor.Executor;
+import org.apache.paimon.web.common.result.FetchResultParams;
+import org.apache.paimon.web.common.result.SubmitResult;
+import org.apache.paimon.web.flink.exception.SqlExecutionException;
+import org.apache.paimon.web.flink.operation.FlinkSqlOperationType;
+import org.apache.paimon.web.flink.parser.StatementParser;
+import org.apache.paimon.web.flink.utils.CollectResultUtil;
+import org.apache.paimon.web.flink.utils.FormatSqlExceptionUtil;
+
+import org.apache.commons.collections.CollectionUtils;
+import org.apache.flink.core.execution.JobClient;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.table.api.StatementSet;
+import org.apache.flink.table.api.TableEnvironment;
+import org.apache.flink.table.api.TableResult;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Optional;
+
+/** The flink implementation of the {@link Executor}. */
+public class FlinkExecutor implements Executor {
+
+ private static final String EXECUTE_SUCCESS = "OK";
+ private final StreamExecutionEnvironment env;
+ private final TableEnvironment tableEnv;
+
+ public FlinkExecutor(StreamExecutionEnvironment env, TableEnvironment tableEnv) {
+ this.env = env;
+ this.tableEnv = tableEnv;
+ }
+
+ @Override
+ public SubmitResult executeSql(String multiStatement) throws SqlExecutionException {
+ List<String> insertStatements = new ArrayList<>();
+ String[] statements = StatementParser.parse(multiStatement);
+ for (String statement : statements) {
+ FlinkSqlOperationType operationType = FlinkSqlOperationType.getOperationType(statement);
+ switch (operationType.getCategory()) {
+ case DQL:
+ if (insertStatements.isEmpty()) {
+ return executeQueryStatement(statement);
+ }
+ break;
+ case DML:
+ if (operationType.getType().equals(FlinkSqlOperationType.INSERT.getType())) {
+ insertStatements.add(statement);
+ } else if (insertStatements.isEmpty()) {
+ return executeDmlStatement(statement);
+ }
+ break;
+ default:
+ executeStatementWithoutResult(statement);
+ break;
+ }
+ }
+ return executeInsertStatements(insertStatements);
Review Comment:
If it contains multiple insert statements, statementSet will be opened, so insert is executed last.
--
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@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org