diff --git a/dolphinscheduler-task-plugin/dolphinscheduler-task-sql/src/main/java/org/apache/dolphinscheduler/plugin/task/sql/SqlTask.java b/dolphinscheduler-task-plugin/dolphinscheduler-task-sql/src/main/java/org/apache/dolphinscheduler/plugin/task/sql/SqlTask.java
index 92bf2e6e2d..f58817efd1 100644
--- a/dolphinscheduler-task-plugin/dolphinscheduler-task-sql/src/main/java/org/apache/dolphinscheduler/plugin/task/sql/SqlTask.java
+++ b/dolphinscheduler-task-plugin/dolphinscheduler-task-sql/src/main/java/org/apache/dolphinscheduler/plugin/task/sql/SqlTask.java
@@ -261,6 +261,7 @@ public class SqlTask extends AbstractTask {
                 resultJSONArray.add(mapOfColValues);
                 rowCount++;
             }
+
             int displayRows = sqlParameters.getDisplayRows() > 0 ? sqlParameters.getDisplayRows() : TaskConstants.DEFAULT_DISPLAY_ROWS;
             displayRows = Math.min(displayRows, rowCount);
             logger.info("display sql result {} rows as follows:", displayRows);
@@ -269,7 +270,10 @@ public class SqlTask extends AbstractTask {
                 logger.info("row {} : {}", i + 1, row);
             }
         }
-        String result = JSONUtils.toJsonString(resultJSONArray);
+
+        String result = resultJSONArray.isEmpty() ?
+            JSONUtils.toJsonString(generateEmptyRow(resultSet)) : JSONUtils.toJsonString(resultJSONArray);
+
         if (sqlParameters.getSendEmail() == null || sqlParameters.getSendEmail()) {
             sendAttachment(sqlParameters.getGroupId(), StringUtils.isNotEmpty(sqlParameters.getTitle())
                     ? sqlParameters.getTitle()
@@ -279,6 +283,27 @@ public class SqlTask extends AbstractTask {
         return result;
     }
 
+    /**
+     * generate empty Results as ArrayNode
+     */
+    private ArrayNode generateEmptyRow(ResultSet resultSet) throws SQLException {
+        ArrayNode resultJSONArray = JSONUtils.createArrayNode();
+        ObjectNode emptyOfColValues = JSONUtils.createObjectNode();
+        if (resultSet != null) {
+            ResultSetMetaData metaData = resultSet.getMetaData();
+            int columnsNum = metaData.getColumnCount();
+            logger.info("sql query results is empty");
+            for (int i = 1; i <= columnsNum; i++) {
+                emptyOfColValues.set(metaData.getColumnLabel(i), JSONUtils.toJsonNode(""));
+            }
+        } else {
+            emptyOfColValues.set("error", JSONUtils.toJsonNode("resultSet is null"));
+        }
+        resultJSONArray.add(emptyOfColValues);
+        return resultJSONArray;
+    }
+
+
     /**
      * send alert as an attachment
      *
@@ -296,6 +321,7 @@ public class SqlTask extends AbstractTask {
 
     private String executeQuery(Connection connection, SqlBinds sqlBinds, String handlerType) throws Exception {
         try (PreparedStatement statement = prepareStatementAndBind(connection, sqlBinds)) {
+            logger.info("{} statement execute query, for sql: {}", handlerType, sqlBinds.getSql());
             ResultSet resultSet = statement.executeQuery();
             return resultProcess(resultSet);
         }