[Fix] Fix unrecognized global variables defined by registry in applic…

…ation mode (#3266)

Co-authored-by: Zzm0809 <[email protected]>

dinky-app/dinky-app-base/src/main/java/org/dinky/app/db/DBUtil.java

@@ -21,13 +21,15 @@
 
 import org.dinky.app.model.SysConfig;
 import org.dinky.data.app.AppDatabase;
+import org.dinky.data.app.AppGlobalVariable;
 import org.dinky.data.app.AppParamConfig;
 import org.dinky.data.app.AppTask;
 
 import java.sql.SQLException;
 import java.util.List;
 
 import cn.hutool.core.text.StrFormatter;
+import cn.hutool.core.util.StrUtil;
 import cn.hutool.db.Db;
 import cn.hutool.db.Entity;
 import cn.hutool.db.ds.simple.SimpleDataSource;
@@ -61,9 +63,31 @@ public static String getDbSourceSQLStatement() throws SQLException {
         Entity option = Entity.create("dinky_database").set("enabled", true);
         List<AppDatabase> entities = db.find(option, AppDatabase.class);
         for (AppDatabase entity : entities) {
+            // Filter out items with empty FlinkConfiguration, as this item is optional in the front-end form and does
+            // not need to be generated when it is empty
+            if (StrUtil.isNotBlank(entity.getFlinkConfig())) {
+                sb.append(entity.getName())
+                        .append(":=")
+                        .append(entity.getFlinkConfig())
+                        .append("\n;\n");
+            }
+        }
+        return sb.toString();
+    }
+
+    /**
+     * Get the global variables statement获取全局变量
+     * @return the global variables statement
+     * @throws SQLException if a database access error occurs
+     */
+    public static String getGlobalVariablesStatement() throws SQLException {
+        StringBuilder sb = new StringBuilder();
+        Entity option = Entity.create("dinky_fragment").set("enabled", true);
+        List<AppGlobalVariable> entities = db.find(option, AppGlobalVariable.class);
+        for (AppGlobalVariable entity : entities) {
             sb.append(entity.getName())
                     .append(":=")
-                    .append(entity.getFlinkConfig())
+                    .append(entity.getFragmentValue())
                     .append("\n;\n");
         }
         return sb.toString();

dinky-app/dinky-app-base/src/main/java/org/dinky/app/flinksql/Submitter.java

@@ -161,8 +161,11 @@ public static String buildSql(AppTask appTask) throws SQLException {
         }
         // build Database golbal varibals
         if (appTask.getFragment()) {
-            log.info("Global env is enable, load database flink config env.");
+            log.info("Global env is enable, load database flink config env and global variables.");
+            // append database flink config env
             sb.append(DBUtil.getDbSourceSQLStatement()).append("\n");
+            // append global variables
+            sb.append(DBUtil.getGlobalVariablesStatement()).append("\n");
         }
         sb.append(appTask.getStatement());
         return sb.toString();

dinky-common/src/main/java/org/dinky/data/app/AppGlobalVariable.java

@@ -0,0 +1,35 @@
+/*
+ *
+ *  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.dinky.data.app;
+
+import io.swagger.annotations.ApiModelProperty;
+import lombok.Data;
+
+@Data
+public class AppGlobalVariable {
+    @ApiModelProperty(value = "ID", required = true, dataType = "Integer", example = "1", notes = "Primary Key")
+    private Integer id;
+
+    @ApiModelProperty(value = "Name", required = true, dataType = "String", example = "Name")
+    private String name;
+
+    @ApiModelProperty(value = "flinkConfig", dataType = "String", example = "flinkConfig")
+    private String fragmentValue;
+}