diff --git a/datavines-common/src/main/java/io/datavines/common/ConfigConstants.java b/datavines-common/src/main/java/io/datavines/common/ConfigConstants.java
index 6a0e3625c..bb33180fa 100644
--- a/datavines-common/src/main/java/io/datavines/common/ConfigConstants.java
+++ b/datavines-common/src/main/java/io/datavines/common/ConfigConstants.java
@@ -75,6 +75,10 @@ public class ConfigConstants {
     public static final String INVALIDATE_ITEM_CAN_OUTPUT = "invalidate_item_can_output";
     public static final String ERROR_DATA_OUTPUT_TO_DATASOURCE_DATABASE = "error_data_output_to_datasource_database";
     public static final String DATABASE = "database";
+
+    public static final String DATABASE_NAME = "database_name";
+    public static final String TABLE_NAME = "table_name";
+    public static final String COLUMN_NAME = "column_name";
     public static final String SCHEMA = "schema";
     public static final String SCHEMA2 = "schema2";
     public static final String USER = "user";
diff --git a/datavines-connector/datavines-connector-api/src/main/java/io/datavines/connector/api/Dialect.java b/datavines-connector/datavines-connector-api/src/main/java/io/datavines/connector/api/Dialect.java
index 7b64c81c2..de59d65b0 100644
--- a/datavines-connector/datavines-connector-api/src/main/java/io/datavines/connector/api/Dialect.java
+++ b/datavines-connector/datavines-connector-api/src/main/java/io/datavines/connector/api/Dialect.java
@@ -54,6 +54,20 @@ default String getFullQualifiedTableName(String database, String schema, String
         return table;
     }
 
+    default String getFullQualifiedTableNameForSpark(String database, String schema, String table) {
+        table = "`" + table + "`";
+
+        if (!StringUtils.isEmptyOrNullStr(schema)) {
+            table = "`" + schema + "`." + table ;
+        }
+
+        if (!StringUtils.isEmptyOrNullStr(database)) {
+            table = "`" + database + "`." + table;
+        }
+
+        return table;
+    }
+
     default boolean invalidateItemCanOutput(){
         return true;
     }
diff --git a/datavines-engine/datavines-engine-plugins/datavines-engine-local/datavines-engine-local-config/src/main/java/io/datavines/engine/local/config/BaseLocalConfigurationBuilder.java b/datavines-engine/datavines-engine-plugins/datavines-engine-local/datavines-engine-local-config/src/main/java/io/datavines/engine/local/config/BaseLocalConfigurationBuilder.java
index 83c3ebd4f..85bc02cd2 100644
--- a/datavines-engine/datavines-engine-plugins/datavines-engine-local/datavines-engine-local-config/src/main/java/io/datavines/engine/local/config/BaseLocalConfigurationBuilder.java
+++ b/datavines-engine/datavines-engine-plugins/datavines-engine-local/datavines-engine-local-config/src/main/java/io/datavines/engine/local/config/BaseLocalConfigurationBuilder.java
@@ -70,6 +70,10 @@ protected List<SourceConfig> getSourceConfigs() throws DataVinesException {
                     connectorParameterMap = connectorFactory.getConnectorParameterConverter().converter(connectorParameterMap);
                     String connectorUUID = connectorFactory.getConnectorParameterConverter().getConnectorUUID(connectorParameterMap);
 
+                    metricInputParameter.put(DATABASE_NAME,metricInputParameter.get(DATABASE));
+                    metricInputParameter.put(TABLE_NAME,metricInputParameter.get(TABLE));
+                    metricInputParameter.put(COLUMN_NAME,metricInputParameter.get(COLUMN));
+
                     String table = connectorFactory.getDialect()
                             .getFullQualifiedTableName(metricInputParameter.get(DATABASE),metricInputParameter.get(SCHEMA),metricInputParameter.get(TABLE));
                     connectorParameterMap.put(TABLE, table);
diff --git a/datavines-engine/datavines-engine-plugins/datavines-engine-spark/datavines-engine-spark-config/src/main/java/io/datavines/engine/spark/config/BaseSparkConfigurationBuilder.java b/datavines-engine/datavines-engine-plugins/datavines-engine-spark/datavines-engine-spark-config/src/main/java/io/datavines/engine/spark/config/BaseSparkConfigurationBuilder.java
index 940f313ce..10b708bca 100644
--- a/datavines-engine/datavines-engine-plugins/datavines-engine-spark/datavines-engine-spark-config/src/main/java/io/datavines/engine/spark/config/BaseSparkConfigurationBuilder.java
+++ b/datavines-engine/datavines-engine-plugins/datavines-engine-spark/datavines-engine-spark-config/src/main/java/io/datavines/engine/spark/config/BaseSparkConfigurationBuilder.java
@@ -72,11 +72,16 @@ protected List<SourceConfig> getSourceConfigs() throws DataVinesException {
                         metricInputParameter.put(SCHEMA, (String)connectorParameter.getParameters().get(SCHEMA));
                     }
 
+                    metricInputParameter.put(DATABASE_NAME,metricInputParameter.get(DATABASE));
+                    metricInputParameter.put(TABLE_NAME,metricInputParameter.get(TABLE));
+                    metricInputParameter.put(COLUMN_NAME,metricInputParameter.get(COLUMN));
+
                     ConnectorFactory connectorFactory = PluginLoader
                             .getPluginLoader(ConnectorFactory.class)
                             .getNewPlugin(connectorParameter.getType());
                     String table = connectorFactory.getDialect()
-                            .getFullQualifiedTableName(metricInputParameter.get(DATABASE),metricInputParameter.get(SCHEMA),metricInputParameter.get(TABLE));
+                            .getFullQualifiedTableNameForSpark(metricInputParameter.get(DATABASE),
+                                    metricInputParameter.get(SCHEMA),metricInputParameter.get(TABLE));
 
                     connectorParameterMap.put(TABLE, table);
                     connectorParameterMap.put(DATABASE, metricInputParameter.get(DATABASE));
@@ -126,7 +131,7 @@ protected List<SourceConfig> getSourceConfigs() throws DataVinesException {
                     ConnectorFactory connectorFactory = PluginLoader
                             .getPluginLoader(ConnectorFactory.class)
                             .getNewPlugin(connectorParameter2.getType());
-                    String table = connectorFactory.getDialect().getFullQualifiedTableName(metricInputParameter.get(DATABASE2),
+                    String table = connectorFactory.getDialect().getFullQualifiedTableNameForSpark(metricInputParameter.get(DATABASE2),
                                                                                            metricInputParameter.get(SCHEMA2),
                                                                                            metricInputParameter.get(TABLE2));
 
diff --git a/datavines-engine/datavines-engine-plugins/datavines-engine-spark/datavines-engine-spark-core/src/main/java/io/datavines/engine/spark/core/SparkDataVinesBootstrap.java b/datavines-engine/datavines-engine-plugins/datavines-engine-spark/datavines-engine-spark-core/src/main/java/io/datavines/engine/spark/core/SparkDataVinesBootstrap.java
index b821a77d9..4e308d720 100644
--- a/datavines-engine/datavines-engine-plugins/datavines-engine-spark/datavines-engine-spark-core/src/main/java/io/datavines/engine/spark/core/SparkDataVinesBootstrap.java
+++ b/datavines-engine/datavines-engine-plugins/datavines-engine-spark/datavines-engine-spark-core/src/main/java/io/datavines/engine/spark/core/SparkDataVinesBootstrap.java
@@ -18,10 +18,16 @@
 
 import io.datavines.engine.core.BaseDataVinesBootstrap;
 
+import java.util.Base64;
+
 public class SparkDataVinesBootstrap extends BaseDataVinesBootstrap {
 
     public static void main(String[] args) {
         SparkDataVinesBootstrap bootstrap = new SparkDataVinesBootstrap();
-        bootstrap.execute(args);
+        if (args.length == 1) {
+            String arg = args[0];
+            args[0] = new String(Base64.getDecoder().decode(arg));
+            bootstrap.execute(args);
+        }
     }
 }
diff --git a/datavines-engine/datavines-engine-plugins/datavines-engine-spark/datavines-engine-spark-executor/src/main/java/io/datavines/engine/spark/executor/SparkEngineExecutor.java b/datavines-engine/datavines-engine-plugins/datavines-engine-spark/datavines-engine-spark-executor/src/main/java/io/datavines/engine/spark/executor/SparkEngineExecutor.java
index 236ee6044..387649d29 100644
--- a/datavines-engine/datavines-engine-plugins/datavines-engine-spark/datavines-engine-spark-executor/src/main/java/io/datavines/engine/spark/executor/SparkEngineExecutor.java
+++ b/datavines-engine/datavines-engine-plugins/datavines-engine-spark/datavines-engine-spark-executor/src/main/java/io/datavines/engine/spark/executor/SparkEngineExecutor.java
@@ -19,6 +19,8 @@
 import java.io.File;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Base64;
+import java.util.Objects;
 
 import io.datavines.common.entity.JobExecutionRequest;
 import io.datavines.common.utils.FileUtils;
@@ -127,7 +129,7 @@ protected String buildCommand() {
                 JSONUtils.parseObject(jobExecutionRequest.getApplicationParameter(), DataVinesJobConfig.class);
 
         sparkParameters.setMainArgs("\""
-                + StringUtils.replaceDoubleBrackets(StringUtils.escapeJava(JSONUtils.toJsonString(configuration))) + "\"");
+                + Base64.getEncoder().encodeToString(Objects.requireNonNull(JSONUtils.toJsonString(configuration)).getBytes()) + "\"");
 
         sparkParameters.setMainClass("io.datavines.engine.spark.core.SparkDataVinesBootstrap");
 
diff --git a/datavines-metric/datavines-metric-api/src/main/java/io/datavines/metric/api/MetricConstants.java b/datavines-metric/datavines-metric-api/src/main/java/io/datavines/metric/api/MetricConstants.java
index 9341bd22b..faac57b68 100644
--- a/datavines-metric/datavines-metric-api/src/main/java/io/datavines/metric/api/MetricConstants.java
+++ b/datavines-metric/datavines-metric-api/src/main/java/io/datavines/metric/api/MetricConstants.java
@@ -33,9 +33,9 @@ public class MetricConstants {
         RESULT_COLUMN_LIST.add(new ColumnInfo("metric_type",false));
         RESULT_COLUMN_LIST.add(new ColumnInfo("metric_name",false));
         RESULT_COLUMN_LIST.add(new ColumnInfo("metric_dimension",false));
-        RESULT_COLUMN_LIST.add(new ColumnInfo("database_name",true,"database"));
-        RESULT_COLUMN_LIST.add(new ColumnInfo("table_name",true,"table"));
-        RESULT_COLUMN_LIST.add(new ColumnInfo("column_name",true,"column"));
+        RESULT_COLUMN_LIST.add(new ColumnInfo("database_name",true));
+        RESULT_COLUMN_LIST.add(new ColumnInfo("table_name",true));
+        RESULT_COLUMN_LIST.add(new ColumnInfo("column_name",true));
         RESULT_COLUMN_LIST.add(new ColumnInfo("actual_value",false));
         RESULT_COLUMN_LIST.add(new ColumnInfo("expected_value",false));
         RESULT_COLUMN_LIST.add(new ColumnInfo("expected_type",false));