feat: refine openmldb batch log and offline job info with SQL

java/openmldb-batch/src/main/scala/com/_4paradigm/openmldb/batch/api/OpenmldbSession.scala

@@ -174,6 +174,8 @@
    * @return
    */
   def openmldbSql(sqlText: String): OpenmldbDataframe = {
+    logger.info("Try to execute OpenMLDB SQL: " + sqlText)
+
     if (config.enableSparksql) {
       return OpenmldbDataframe(this, sparksql(sqlText))
     }
@@ -278,7 +280,15 @@
   def close(): Unit = stop()
 
   def registerOpenmldbOfflineTable(catalogService: OpenmldbCatalogService): Unit = {
+    if (catalogService == null) {
+      return
+    }
+
     val databases = catalogService.getDatabases
+    if (databases == null) {
+      return
+    }
+
     databases.map(dbName => {
       val tableInfos = catalogService.getTableInfos(dbName)
       tableInfos.map(tableInfo => {
@@ -323,7 +333,7 @@
             }
           } catch {
             case e: Exception => {
-              logger.warn(s"Fail to register table $dbName.$tableName " + ExceptionUtils.getStackTrace(e))
+              logger.warn(s"Fail to register table $dbName.$tableName, exception: " + ExceptionUtils.getStackTrace(e))
             }
           }
         }

java/openmldb-taskmanager/src/main/scala/com/_4paradigm/openmldb/taskmanager/OpenmldbBatchjobManager.scala

@@ -57,10 +57,10 @@
 
     if (TaskManagerConfig.isK8s) {
       val args = List(sql)
-      K8sJobManager.submitSparkJob(jobType, mainClass, args, tempSqlFile.getAbsolutePath, sparkConf.asScala.toMap,
+      K8sJobManager.submitSparkJob(jobType, mainClass, args, sql, tempSqlFile.getAbsolutePath, sparkConf.asScala.toMap,
         defaultDb)
     } else {
-      SparkJobManager.submitSparkJob(jobType, mainClass, args, tempSqlFile.getAbsolutePath,
+      SparkJobManager.submitSparkJob(jobType, mainClass, args, sql, tempSqlFile.getAbsolutePath,
         sparkConf.asScala.toMap, defaultDb, blocking = true)
     }
   }
@@ -73,11 +73,11 @@
 
     if (TaskManagerConfig.isK8s) {
       val args = List(sql)
-      K8sJobManager.submitSparkJob(jobType, mainClass, args, tempSqlFile.getAbsolutePath, sparkConf.asScala.toMap,
+      K8sJobManager.submitSparkJob(jobType, mainClass, args, sql, tempSqlFile.getAbsolutePath, sparkConf.asScala.toMap,
         defaultDb)
     } else {
       val args = List(tempSqlFile.getAbsolutePath)
-      SparkJobManager.submitSparkJob(jobType, mainClass, args, tempSqlFile.getAbsolutePath, sparkConf.asScala.toMap,
+      SparkJobManager.submitSparkJob(jobType, mainClass, args, sql, tempSqlFile.getAbsolutePath, sparkConf.asScala.toMap,
         defaultDb)
     }
   }
@@ -90,11 +90,11 @@
 
     if (TaskManagerConfig.isK8s) {
       val args = List(sql)
-      K8sJobManager.submitSparkJob(jobType, mainClass, args, tempSqlFile.getAbsolutePath, sparkConf.asScala.toMap,
+      K8sJobManager.submitSparkJob(jobType, mainClass, args, sql, tempSqlFile.getAbsolutePath, sparkConf.asScala.toMap,
         defaultDb)
     } else {
       val args = List(tempSqlFile.getAbsolutePath)
-      SparkJobManager.submitSparkJob(jobType, mainClass, args, tempSqlFile.getAbsolutePath, sparkConf.asScala.toMap,
+      SparkJobManager.submitSparkJob(jobType, mainClass, args, sql, tempSqlFile.getAbsolutePath, sparkConf.asScala.toMap,
         defaultDb)
     }
   }
@@ -107,11 +107,11 @@
 
     if (TaskManagerConfig.isK8s) {
       val args = List(sql)
-      K8sJobManager.submitSparkJob(jobType, mainClass, args, tempSqlFile.getAbsolutePath, sparkConf.asScala.toMap,
+      K8sJobManager.submitSparkJob(jobType, mainClass, args, sql, tempSqlFile.getAbsolutePath, sparkConf.asScala.toMap,
         defaultDb)
     } else {
       val args = List(tempSqlFile.getAbsolutePath)
-      SparkJobManager.submitSparkJob(jobType, mainClass, args, tempSqlFile.getAbsolutePath, sparkConf.asScala.toMap,
+      SparkJobManager.submitSparkJob(jobType, mainClass, args, sql, tempSqlFile.getAbsolutePath, sparkConf.asScala.toMap,
         defaultDb)
     }
   }
@@ -124,11 +124,11 @@
 
     if (TaskManagerConfig.isK8s) {
       val args = List(sql)
-      K8sJobManager.submitSparkJob(jobType, mainClass, args, tempSqlFile.getAbsolutePath, sparkConf.asScala.toMap,
+      K8sJobManager.submitSparkJob(jobType, mainClass, args, sql, tempSqlFile.getAbsolutePath, sparkConf.asScala.toMap,
         defaultDb)
     } else {
       val args = List(tempSqlFile.getAbsolutePath)
-      SparkJobManager.submitSparkJob(jobType, mainClass, args, tempSqlFile.getAbsolutePath, sparkConf.asScala.toMap,
+      SparkJobManager.submitSparkJob(jobType, mainClass, args, sql, tempSqlFile.getAbsolutePath, sparkConf.asScala.toMap,
         defaultDb)
     }
   }

java/openmldb-taskmanager/src/main/scala/com/_4paradigm/openmldb/taskmanager/k8s/K8sJobManager.scala

@@ -44,12 +44,18 @@
 
   def submitSparkJob(jobType: String, mainClass: String,
     args: List[String] = List(),
+    sql: String = "",
     localSqlFile: String = "",
     sparkConf: Map[String, String] = Map(),
     defaultDb: String = "",
     blocking: Boolean = false): JobInfo = {
 
-    val jobInfo = JobInfoManager.createJobInfo(jobType, args, sparkConf)
+    val jobInfoArgs = if (sql.nonEmpty) {
+      List(sql)
+    } else {
+      args
+    }
+    val jobInfo = JobInfoManager.createJobInfo(jobType, jobInfoArgs, sparkConf)
 
     val jobName = getK8sJobName(jobInfo.getId)
     jobInfo.setApplicationId(jobName)

java/openmldb-taskmanager/src/main/scala/com/_4paradigm/openmldb/taskmanager/spark/SparkJobManager.scala

@@ -77,11 +77,18 @@
 
   def submitSparkJob(jobType: String, mainClass: String,
                      args: List[String] = List(),
+                     sql: String = "",
                      localSqlFile: String = "",
                      sparkConf: Map[String, String] = Map(),
                      defaultDb: String = "",
                      blocking: Boolean = false): JobInfo = {
-    val jobInfo = JobInfoManager.createJobInfo(jobType, args, sparkConf)
+
+    val jobInfoArgs = if (sql.nonEmpty) {
+      List(sql)
+    } else {
+      args
+    }
+    val jobInfo = JobInfoManager.createJobInfo(jobType, jobInfoArgs, sparkConf)
 
     // Submit Spark application with SparkLauncher
     val launcher = createSparkLauncher(mainClass)

java/openmldb-taskmanager/src/test/scala/com/_4paradigm/openmldb/taskmanager/spark/TestSparkJobManager.scala

@@ -36,7 +36,7 @@ class TestSparkJobManager extends FunSuite {
     val jobType = "DummySparkApp"
     val sparkConf = Map(SparkLauncher.DRIVER_EXTRA_CLASSPATH -> System.getProperty("java.class.path"))
 
-    SparkJobManager.submitSparkJob(jobType, mainClass, List[String](), "", sparkConf)
+    SparkJobManager.submitSparkJob(jobType, mainClass, List[String](), "", "", sparkConf)
 
     JobInfoManager.getAllJobs().map(println)
     Thread.sleep(5000)

java/pom.xml

@@ -106,6 +106,20 @@
             </resource>
         </resources>
         <plugins>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-source-plugin</artifactId>
+                <version>3.0.1</version>
+                <executions>
+                    <execution>
+                        <phase>package</phase>
+                        <goals>
+                            <goal>jar-no-fork</goal>
+                        </goals>
+                    </execution>
+                </executions>
+            </plugin>
+
             <!-- those plugins are enable for all submodules -->
             <plugin>
                 <groupId>org.apache.maven.plugins</groupId>

src/sdk/file_option_parser.h

@@ -40,6 +40,7 @@
         check_map_.emplace("header", std::make_pair(CheckHeader(), hybridse::node::kBool));
         check_map_.emplace("quote", std::make_pair(CheckQuote(), hybridse::node::kVarchar));
         check_map_.emplace("mode", std::make_pair(CheckMode(), hybridse::node::kVarchar));
+        check_map_.emplace("sql", std::make_pair(CheckSql(), hybridse::node::kVarchar));
     }
 
     ::openmldb::base::Status Parse(const std::shared_ptr<hybridse::node::OptionsMap>& options_map) {
@@ -82,6 +83,7 @@
     std::string null_value_ = "null";
     std::string delimiter_ = ",";
     bool header_ = true;
+    std::string sql_ = "";
 
     ::openmldb::base::Status GetOption(const hybridse::node::ConstNode* node, const std::string& option_name,
                                        std::function<bool(const hybridse::node::ConstNode* node)> const& f,
@@ -157,6 +159,12 @@
             return true;
         };
     }
+    std::function<bool(const hybridse::node::ConstNode* node)> CheckSql() {
+        return [this](const hybridse::node::ConstNode* node) {
+            sql_ = node->GetAsString();
+            return true;
+        };
+    }
 };
 
 class ReadFileOptionsParser : public FileOptionsParser {