diff --git a/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/SQLTaskExecutionContext.java b/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/SQLTaskExecutionContext.java
deleted file mode 100644
index 40d9c5632eb1..000000000000
--- a/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/SQLTaskExecutionContext.java
+++ /dev/null
@@ -1,73 +0,0 @@
-/*
- * 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.apache.dolphinscheduler.plugin.task.api;
-
-import java.io.Serializable;
-
-/**
- * SQL Task ExecutionContext
- */
-public class SQLTaskExecutionContext implements Serializable {
-
-    /**
-     * warningGroupId
-     */
-    private int warningGroupId;
-
-    /**
-     * connectionParams
-     */
-    private String connectionParams;
-
-    /**
-     * DefaultFS
-     */
-    private String defaultFS;
-
-    public int getWarningGroupId() {
-        return warningGroupId;
-    }
-
-    public void setWarningGroupId(int warningGroupId) {
-        this.warningGroupId = warningGroupId;
-    }
-
-    public String getConnectionParams() {
-        return connectionParams;
-    }
-
-    public void setConnectionParams(String connectionParams) {
-        this.connectionParams = connectionParams;
-    }
-
-    public String getDefaultFS() {
-        return defaultFS;
-    }
-
-    public void setDefaultFS(String defaultFS) {
-        this.defaultFS = defaultFS;
-    }
-
-    @Override
-    public String toString() {
-        return "SQLTaskExecutionContext{"
-                + "warningGroupId=" + warningGroupId
-                + ", connectionParams='" + connectionParams + '\''
-                + ", defaultFS='" + defaultFS + '\'' + '}';
-    }
-}
diff --git a/dolphinscheduler-task-plugin/dolphinscheduler-task-sql/src/main/java/org/apache/dolphinscheduler/plugin/task/sql/SqlParametersFactory.java b/dolphinscheduler-task-plugin/dolphinscheduler-task-sql/src/main/java/org/apache/dolphinscheduler/plugin/task/sql/SqlParametersFactory.java
new file mode 100644
index 000000000000..6b778462948a
--- /dev/null
+++ b/dolphinscheduler-task-plugin/dolphinscheduler-task-sql/src/main/java/org/apache/dolphinscheduler/plugin/task/sql/SqlParametersFactory.java
@@ -0,0 +1,4 @@
+package org.apache.dolphinscheduler.plugin.task.sql;
+
+public class SqlParametersFactory {
+}
diff --git a/dolphinscheduler-task-plugin/dolphinscheduler-task-sql/src/main/java/org/apache/dolphinscheduler/plugin/task/sql/executor/ISqlExecutionEngine.java b/dolphinscheduler-task-plugin/dolphinscheduler-task-sql/src/main/java/org/apache/dolphinscheduler/plugin/task/sql/executor/ISqlExecutionEngine.java
new file mode 100644
index 000000000000..6a9f7146d5d2
--- /dev/null
+++ b/dolphinscheduler-task-plugin/dolphinscheduler-task-sql/src/main/java/org/apache/dolphinscheduler/plugin/task/sql/executor/ISqlExecutionEngine.java
@@ -0,0 +1,19 @@
+package org.apache.dolphinscheduler.plugin.task.sql.executor;
+
+import java.sql.SQLException;
+
+public interface ISqlExecutionEngine extends AutoCloseable {
+
+    /**
+     * Execute the given sql, the provided sql might contain multiple lines.
+     */
+    void execute(String sql);
+
+    /**
+     * Close the engine, once the Engine is closed, then it cannot execute again.
+     * <p> And if the Engine is executing sql, will cancel the execution.
+     */
+    @Override
+    void close() throws SQLException;
+
+}
diff --git a/dolphinscheduler-task-plugin/dolphinscheduler-task-sql/src/main/java/org/apache/dolphinscheduler/plugin/task/sql/executor/SqlExecutionEngine.java b/dolphinscheduler-task-plugin/dolphinscheduler-task-sql/src/main/java/org/apache/dolphinscheduler/plugin/task/sql/executor/SqlExecutionEngine.java
new file mode 100644
index 000000000000..9fc108da8879
--- /dev/null
+++ b/dolphinscheduler-task-plugin/dolphinscheduler-task-sql/src/main/java/org/apache/dolphinscheduler/plugin/task/sql/executor/SqlExecutionEngine.java
@@ -0,0 +1,48 @@
+package org.apache.dolphinscheduler.plugin.task.sql.executor;
+
+import lombok.extern.slf4j.Slf4j;
+import org.apache.dolphinscheduler.plugin.task.api.parameters.SqlParameters;
+
+import java.sql.Connection;
+import java.sql.SQLException;
+
+@Slf4j
+public class SqlExecutionEngine implements ISqlExecutionEngine {
+
+    protected SqlParameters sqlParameters;
+
+    protected volatile Connection connection;
+
+    // If set true, means the engine is canceled
+    // todo: 使用状态机来控制连接初始化和close
+    protected volatile boolean cancelFlag = false;
+
+    public SqlExecutionEngine(SqlParameters sqlParameters) {
+        this.sqlParameters = sqlParameters;
+        this.connection = connect();
+    }
+
+    @Override
+    public void execute(String sql) {
+        // 1. SQL拆分
+
+        // 2. SQL参数替换
+
+        // 3. 前置执行执行
+
+        // 4. 后置执行
+    }
+
+    protected Connection connect() {
+        // todo: 初始化连接
+        return null;
+    }
+
+    @Override
+    public void close() throws SQLException {
+        try (Connection connection1 = connection) {
+            log.info("Closed SqlTaskEngine.");
+        }
+    }
+
+}