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)