[WIP] Add catalog management and sql gateway api

streampark-console/streampark-console-service/src/main/assembly/script/upgrade/mysql/2.2.0.sql

@@ -87,7 +87,7 @@ create table `t_flink_gateway` (
 ) engine=innodb auto_increment=100000 default charset=utf8mb4 collate=utf8mb4_general_ci;
 
 -- menu level 2
-insert into `t_menu` values (120500, 130000, 'setting.flinkGateway', '/setting/FlinkGateway', 'setting/FlinkGateway/index', null, 'apartment', '0', 1, 3, now(), now());
+insert into `t_menu` values (120500, 130000, 'setting.flinkGateway', '/setting/FlinkGateway', 'setting/FlinkGateway/index', null, 'api', '0', 1, 3, now(), now());
 -- menu level 3
 insert into `t_menu` values (120501, 120500, 'add', NULL, NULL, 'gateway:add', NULL, '1', 1, NULL, now(), now());
 insert into `t_menu` values (120502, 120500, 'update', NULL, NULL, 'gateway:update', NULL, '1', 1, NULL, now(), now());
@@ -104,4 +104,37 @@ insert into `t_role_menu` (role_id, menu_id) values (100002, 120501);
 insert into `t_role_menu` (role_id, menu_id) values (100002, 120502);
 insert into `t_role_menu` (role_id, menu_id) values (100002, 120503);
 
+-- ----------------------------
+-- Table of t_flink_catalog
+-- ----------------------------
+drop table if exists `t_flink_catalog`;
+create table `t_flink_catalog` (
+                                `id` bigint not null auto_increment,
+                                `catalog_name` varchar(128) collate utf8mb4_general_ci not null comment 'The name of the catalog',
+                                `properties` text not null comment 'properties of catalog',
+                                `description` varchar(255) collate utf8mb4_general_ci default null comment 'description',
+                                `create_time` datetime not null default current_timestamp comment 'create time',
+                                `modify_time` datetime not null default current_timestamp on update current_timestamp comment 'modify time',
+                                primary key (`id`) using btree,
+                                unique key `un_catalog_name_inx` (`catalog_name`) using btree
+) engine=innodb auto_increment=100000 default charset=utf8mb4 collate=utf8mb4_general_ci;
+
+-- menu level 2
+insert into `t_menu` values (120600, 130000, 'setting.flinkCatalog', '/setting/FlinkCatalog', 'setting/FlinkCatalog/index', null, 'folder', '0', 1, 3, now(), now());
+-- menu level 3
+insert into `t_menu` values (120601, 120600, 'add', NULL, NULL, 'catalog:add', NULL, '1', 1, NULL, now(), now());
+insert into `t_menu` values (120602, 120600, 'update', NULL, NULL, 'catalog:update', NULL, '1', 1, NULL, now(), now());
+insert into `t_menu` values (120603, 120600, 'delete', NULL, NULL, 'catalog:delete', NULL, '1', 1, NULL, now(), now());
+
+-- role menu script
+insert into `t_role_menu` (role_id, menu_id) values (100001, 120600);
+insert into `t_role_menu` (role_id, menu_id) values (100001, 120601);
+insert into `t_role_menu` (role_id, menu_id) values (100001, 120602);
+insert into `t_role_menu` (role_id, menu_id) values (100001, 120603);
+
+insert into `t_role_menu` (role_id, menu_id) values (100002, 120600);
+insert into `t_role_menu` (role_id, menu_id) values (100002, 120601);
+insert into `t_role_menu` (role_id, menu_id) values (100002, 120602);
+insert into `t_role_menu` (role_id, menu_id) values (100002, 120603);
+
 set foreign_key_checks = 1;

streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/bean/SqlGatewayExecuteResult.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.streampark.console.core.bean;
+
+import org.apache.streampark.gateway.OperationHandle;
+import org.apache.streampark.gateway.results.OperationInfo;
+import org.apache.streampark.gateway.results.ResultSet;
+import org.apache.streampark.gateway.session.SessionHandle;
+
+import io.swagger.annotations.ApiModel;
+import io.swagger.annotations.ApiModelProperty;
+import lombok.Data;
+
+import java.io.Serializable;
+
+@Data
+@ApiModel(description = "Result of sql gateway execute")
+public class SqlGatewayExecuteResult implements Serializable {
+
+  @ApiModelProperty(value = "sessionHandle")
+  private SessionHandle sessionHandle;
+
+  @ApiModelProperty(value = "operationHandle")
+  private OperationHandle operationHandle;
+
+  @ApiModelProperty(value = "operationInfo")
+  private OperationInfo operationInfo;
+
+  @ApiModelProperty(value = "resultSet")
+  private ResultSet resultSet;
+
+  public SqlGatewayExecuteResult(
+      SessionHandle sessionHandle,
+      OperationHandle operationHandle,
+      OperationInfo operationInfo,
+      ResultSet resultSet) {
+    this.sessionHandle = sessionHandle;
+    this.operationHandle = operationHandle;
+    this.operationInfo = operationInfo;
+    this.resultSet = resultSet;
+  }
+}

streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/controller/FlinkCatalogController.java

@@ -0,0 +1,94 @@
+/*
+ * 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.streampark.console.core.controller;
+
+import org.apache.streampark.console.base.domain.RestRequest;
+import org.apache.streampark.console.base.domain.RestResponse;
+import org.apache.streampark.console.base.mybatis.pager.MybatisPager;
+import org.apache.streampark.console.core.entity.FlinkCatalog;
+import org.apache.streampark.console.core.service.FlinkCatalogService;
+
+import io.swagger.v3.oas.annotations.Operation;
+import io.swagger.v3.oas.annotations.tags.Tag;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.validation.annotation.Validated;
+import org.springframework.web.bind.annotation.DeleteMapping;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.PathVariable;
+import org.springframework.web.bind.annotation.PostMapping;
+import org.springframework.web.bind.annotation.PutMapping;
+import org.springframework.web.bind.annotation.RequestBody;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RequestParam;
+import org.springframework.web.bind.annotation.RestController;
+
+import javax.validation.constraints.NotNull;
+
+@Tag(name = "FLINK_CATALOG_TAG")
+@Slf4j
+@Validated
+@RestController
+@RequiredArgsConstructor
+@RequestMapping("flink/catalog")
+public class FlinkCatalogController {
+
+  private final FlinkCatalogService flinkCatalogService;
+
+  @Operation(summary = "List catalog")
+  @GetMapping("list")
+  public RestResponse list(RestRequest request) {
+    return RestResponse.success(
+        flinkCatalogService.page(new MybatisPager<FlinkCatalog>().getDefaultPage(request)));
+  }
+
+  @Operation(summary = "Create flink catalog")
+  @PostMapping("create")
+  public RestResponse create(@RequestBody FlinkCatalog flinkCatalog) {
+    flinkCatalogService.create(flinkCatalog);
+    return RestResponse.success();
+  }
+
+  @Operation(summary = "Check flink catalog name")
+  @GetMapping("check/name")
+  public RestResponse checkName(
+      @NotNull(message = "The flink catalog name cannot be null") @RequestParam("name")
+          String name) {
+    return RestResponse.success(flinkCatalogService.existsByCatalogName(name));
+  }
+
+  @Operation(summary = "Update flink catalog")
+  @PutMapping("update")
+  public RestResponse update(@RequestBody FlinkCatalog flinkCatalog) {
+    flinkCatalogService.update(flinkCatalog);
+    return RestResponse.success();
+  }
+
+  @Operation(summary = "Get flink catalog by id")
+  @GetMapping("get/{id}")
+  public RestResponse get(@PathVariable Long id) {
+    return RestResponse.success(flinkCatalogService.getById(id));
+  }
+
+  @Operation(summary = "Delete flink catalog by id")
+  @DeleteMapping("delete/{id}")
+  public RestResponse delete(@PathVariable Long id) {
+    flinkCatalogService.removeById(id);
+    return RestResponse.success();
+  }
+}

streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/controller/FlinkGateWayController.java

@@ -93,9 +93,9 @@ public RestResponse get(@PathVariable Long id) {
   }
 
   @Operation(summary = "Delete flink gateway by id")
-  @DeleteMapping("delete")
+  @DeleteMapping("delete/{id}")
   public RestResponse delete(
-      @NotNull(message = "The Gateway id cannot be null") @RequestParam("id") Long id) {
+      @NotNull(message = "The Gateway id cannot be null") @PathVariable Long id) {
     flinkGatewayService.removeById(id);
     return RestResponse.success();
   }