From f2136bd8ae3100c5d44383f65481cfd0763fcbd6 Mon Sep 17 00:00:00 2001
From: camilesing <camilesing@gmail.com>
Date: Fri, 18 Sep 2026 23:29:02 +0800
Subject: [PATCH 1/3] fix(datatask): tolerate string wire formats for array
 parameters

Array-declared parameters (isArray=true) now accept three wire shapes besides a native JSON array: a serialized JSON array string (["a","b"]), a comma-separated string (a,b) and a single scalar, which becomes a one-element list. The same tolerance applies to arrays declared on OBJECT children.

Elements are still coerced one by one to the declared type, so the loosening does not weaken value validation; a JSON-looking string that fails to parse falls back to comma splitting.

Adds binder tests for the new formats (including elements with embedded commas and type coercion) and documents the accepted submission forms in docs/{en,zh}/data-task.md.
---
 .../cs/core/datatask/DataTaskParamBinder.java | 63 ++++++++++---
 .../datatask/DataTaskParamBinderTest.java     | 90 +++++++++++++++++++
 docs/en/data-task.md                          |  3 +
 docs/zh/data-task.md                          |  2 +
 4 files changed, 145 insertions(+), 13 deletions(-)

diff --git a/datapoly-core/src/main/java/com/cs/core/datatask/DataTaskParamBinder.java b/datapoly-core/src/main/java/com/cs/core/datatask/DataTaskParamBinder.java
index 83bc4b4..184084d 100644
--- a/datapoly-core/src/main/java/com/cs/core/datatask/DataTaskParamBinder.java
+++ b/datapoly-core/src/main/java/com/cs/core/datatask/DataTaskParamBinder.java
@@ -5,6 +5,7 @@
 import com.cs.common.dto.ItemParam;
 import com.cs.common.exception.CommonException;
 import com.cs.common.exception.ResponseErrorCode;
+import com.cs.persistence.util.JsonUtils;
 import org.apache.commons.lang3.StringUtils;
 
 import java.util.*;
@@ -46,19 +47,60 @@ private static Object bindScalarRoot(ItemParam decl, Map<String, Object> body, b
         if (!isArray) {
             return coerceOrNull(decl, raw, decl.getName(), Boolean.TRUE.equals(decl.getRequired()));
         }
-        List<Object> values = raw instanceof List ? (List<Object>) raw : null;
+        List<Object> values = arrayElements(decl, raw, decl.getName(), true);
         if (null == values || values.isEmpty()) {
             return requireOrNothing(decl, raw, decl.getName(),
                     () -> defaultList(decl));
         }
-        List<Object> out = new ArrayList<>(values.size());
-        for (int i = 0; i < values.size(); i++) {
-            out.add(coerce(decl.getType(), values.get(i),
-                    String.format("%s[%d]", decl.getName(), i), true));
+        return values;
+    }
+
+    /**
+     * Element list of an array-declared parameter. Besides a native JSON array, two
+     * string wire formats that frontends commonly emit are tolerated: a pre-serialized
+     * JSON array literal (["a","b"]) and a comma-separated list ("a,b"). Every element
+     * is still coerced to the declared type, so shape loosening does not weaken value
+     * validation; a String that carries nothing usable yields null for the caller.
+     */
+    private static List<Object> arrayElements(BaseParam decl, Object raw, String path, boolean enforce) {
+        List<Object> rawElements;
+        if (raw instanceof List) {
+            rawElements = (List<Object>) raw;
+        } else if (raw instanceof String && StringUtils.isNotBlank((String) raw)) {
+            rawElements = parseArrayString(StringUtils.trim((String) raw));
+        } else {
+            return null;
+        }
+        List<Object> out = new ArrayList<>(rawElements.size());
+        for (int i = 0; i < rawElements.size(); i++) {
+            out.add(coerce(decl.getType(), rawElements.get(i),
+                    String.format("%s[%d]", path, i), enforce));
         }
         return out;
     }
 
+    private static List<Object> parseArrayString(String value) {
+        if (value.startsWith("[") && value.endsWith("]")) {
+            try {
+                return JsonUtils.toBeanList(value, Object.class);
+            } catch (RuntimeException ignore) {
+                // not a JSON array after all; fall back to comma splitting
+            }
+        }
+        String[] parts = value.split(",");
+        List<Object> values = new ArrayList<>(parts.length);
+        for (String part : parts) {
+            String item = part.trim();
+            if (item.length() >= 2 && item.startsWith("\"") && item.endsWith("\"")) {
+                item = item.substring(1, item.length() - 1);
+            }
+            if (!item.isEmpty()) {
+                values.add(item);
+            }
+        }
+        return values;
+    }
+
     /**
      * OBJECT parameters accept either a nested map under the parameter name or flat
      * {@code name.child} keys on the request body; the nested form wins on conflicts.
@@ -92,19 +134,14 @@ private static Object bindObjectChildren(ItemParam decl, Map<String, Object> bod
         for (BaseParam child : children) {
             String path = root + "." + child.getName();
             if (Boolean.TRUE.equals(child.getIsArray())) {
-                List<Object> raws = container.get(child.getName()) instanceof List
-                        ? (List<Object>) container.get(child.getName()) : null;
-                if (null == raws || raws.isEmpty()) {
+                List<Object> items = arrayElements(child, container.get(child.getName()), path,
+                        Boolean.TRUE.equals(child.getRequired()));
+                if (null == items || items.isEmpty()) {
                     if (Boolean.TRUE.equals(child.getRequired())) {
                         throw missing(path);
                     }
                     continue;
                 }
-                List<Object> items = new ArrayList<>(raws.size());
-                for (int i = 0; i < raws.size(); i++) {
-                    items.add(coerce(child.getType(), raws.get(i), String.format("%s[%d]", path, i),
-                            Boolean.TRUE.equals(child.getRequired())));
-                }
                 out.put(child.getName(), items);
             } else {
                 Object value = coerceOrNull(child, container.get(child.getName()), path,
diff --git a/datapoly-test/src/test/java/com/cs/core/datatask/DataTaskParamBinderTest.java b/datapoly-test/src/test/java/com/cs/core/datatask/DataTaskParamBinderTest.java
index 0d4f1ba..b680523 100644
--- a/datapoly-test/src/test/java/com/cs/core/datatask/DataTaskParamBinderTest.java
+++ b/datapoly-test/src/test/java/com/cs/core/datatask/DataTaskParamBinderTest.java
@@ -119,6 +119,96 @@ public void scalarArraysConvertElementwise() {
         Assert.assertEquals(Arrays.asList(1L, 2L), bound.get("ids"));
     }
 
+    @Test
+    public void arrayDeclaredParamAcceptsJsonArrayString() {
+        Map<String, Object> body = new HashMap<>();
+        body.put("names", "[\"燕文\",\"顺友\"]");
+
+        Map<String, Object> bound = DataTaskParamBinder.bind(
+                Collections.singletonList(simple("names", ParamTypeEnum.STRING, true, false, null)), body);
+        Assert.assertEquals(Arrays.asList("燕文", "顺友"), bound.get("names"));
+    }
+
+    @Test
+    public void rawWireJsonWithStringifiedArrayBindsToList() {
+        String wire = "{\"params\":{\"logisticsProviderNameList\":\"[\\\"燕文\\\",\\\"顺友\\\"]\"}}";
+        Map<String, Object> body = com.cs.persistence.util.JsonUtils.toBeanType(wire,
+                new com.fasterxml.jackson.core.type.TypeReference<Map<String, Object>>() {
+                });
+
+        @SuppressWarnings("unchecked")
+        Map<String, Object> params = (Map<String, Object>) body.get("params");
+        Map<String, Object> bound = DataTaskParamBinder.bind(
+                Collections.singletonList(simple("logisticsProviderNameList", ParamTypeEnum.STRING, true, false, null)),
+                params);
+        Assert.assertEquals(Arrays.asList("燕文", "顺友"), bound.get("logisticsProviderNameList"));
+    }
+
+    @Test
+    public void arrayDeclaredParamAcceptsCommaSeparatedString() {
+        Map<String, Object> body = new HashMap<>();
+        body.put("names", " 燕文 , 顺友 ");
+
+        Map<String, Object> bound = DataTaskParamBinder.bind(
+                Collections.singletonList(simple("names", ParamTypeEnum.STRING, true, false, null)), body);
+        Assert.assertEquals(Arrays.asList("燕文", "顺友"), bound.get("names"));
+    }
+
+    @Test
+    public void arrayDeclaredParamAcceptsSingleScalarValue() {
+        Map<String, Object> body = new HashMap<>();
+        body.put("names", "燕文");
+
+        Map<String, Object> bound = DataTaskParamBinder.bind(
+                Collections.singletonList(simple("names", ParamTypeEnum.STRING, true, false, null)), body);
+        Assert.assertEquals(Collections.singletonList("燕文"), bound.get("names"));
+    }
+
+    @Test
+    public void jsonArrayStringElementsKeepEmbeddedCommas() {
+        Map<String, Object> body = new HashMap<>();
+        body.put("names", "[\"a,b\",\"c\"]");
+
+        Map<String, Object> bound = DataTaskParamBinder.bind(
+                Collections.singletonList(simple("names", ParamTypeEnum.STRING, true, false, null)), body);
+        Assert.assertEquals(Arrays.asList("a,b", "c"), bound.get("names"));
+    }
+
+    @Test
+    public void arrayStringElementsCoerceToDeclaredType() {
+        Map<String, Object> body = new HashMap<>();
+        body.put("ids", "[1,2,3]");
+
+        Map<String, Object> bound = DataTaskParamBinder.bind(
+                Collections.singletonList(simple("ids", ParamTypeEnum.LONG, true, false, null)), body);
+        Assert.assertEquals(Arrays.asList(1L, 2L, 3L), bound.get("ids"));
+    }
+
+    @Test
+    public void objectChildArrayAcceptsJsonArrayString() {
+        ItemParam decl = new ItemParam();
+        decl.setName("obj");
+        decl.setType(ParamTypeEnum.OBJECT);
+        decl.setIsArray(false);
+        decl.setRequired(false);
+        BaseParam child = new BaseParam();
+        child.setName("tags");
+        child.setType(ParamTypeEnum.STRING);
+        child.setIsArray(true);
+        child.setRequired(false);
+        decl.setChildren(new ArrayList<>(Collections.singletonList(child)));
+
+        Map<String, Object> body = new HashMap<>();
+        Map<String, Object> inner = new HashMap<>();
+        inner.put("tags", "[\"a\",\"b\"]");
+        body.put("obj", inner);
+
+        Map<String, Object> bound = DataTaskParamBinder.bind(Collections.singletonList(decl), body);
+        @SuppressWarnings("unchecked")
+        Map<String, Object> obj = (Map<String, Object>) bound.get("obj");
+        Assert.assertEquals(Arrays.asList("a", "b"), obj.get("tags"));
+    }
+
     @Test
     public void objectParamsAcceptNestedAndDottedForms() {
         ItemParam decl = simple("obj", ParamTypeEnum.OBJECT, false, false, null);
diff --git a/docs/en/data-task.md b/docs/en/data-task.md
index ae4fab6..bb702cc 100644
--- a/docs/en/data-task.md
+++ b/docs/en/data-task.md
@@ -131,6 +131,9 @@ Notes:
   `type` is one of `LONG/DOUBLE/STRING/DATE/TIME/BOOLEAN/OBJECT` (OBJECT requires `children`; submit accepts nested
   maps or flat `parent.sub` keys). Submission enforces required checks and type conversion; undeclared extra
   parameters are ignored.
+- Array parameters (`isArray=true`) accept a JSON array, a pre-serialized JSON array string (e.g.
+  `"[\"YANWEN\",\"SHUNYOU\"]"`), or a comma-separated string (e.g. `"YANWEN,SHUNYOU"`); elements are still
+  converted one by one to the declared type.
 - Reshaping order is **naming strategy → alias → column order**: `columnAlias` keys match column names **after** the
   naming strategy is applied; `columnOrder` fixes the output order and subset — unlisted columns are dropped.
 - In `formatMap`, date/time types take a pattern string and `BIG_DECIMAL` takes a scale (HALF_UP, default 6); with
diff --git a/docs/zh/data-task.md b/docs/zh/data-task.md
index 0ead322..a4b178c 100644
--- a/docs/zh/data-task.md
+++ b/docs/zh/data-task.md
@@ -121,6 +121,8 @@ curl -s -X POST $BASE/create -H "$AUTH" -H 'Content-Type: application/json' -d '
 - 入参声明字段与 API 配置一致：`name/type/location/isArray/required/defaultValue/remark`，`type` 取
   `LONG/DOUBLE/STRING/DATE/TIME/BOOLEAN/OBJECT`（OBJECT 需声明 `children`，支持嵌套 Map 或 `parent.sub`
   扁平键提交）；提交时按声明做必填校验与类型转换，未声明的多余入参被忽略。
+- 数组入参（`isArray=true`）接受 JSON 数组、序列化后的 JSON 数组字符串（如 `"[\"燕文\",\"顺友\"]"`）以及逗号
+  分隔字符串（如 `"燕文,顺友"`）三种提交形式，元素仍逐项按声明类型转换。
 - 整形顺序为**命名策略 → 别名 → 列顺序**：`columnAlias` 的 key 匹配命名策略转换**之后**的列名；
   `columnOrder` 给出输出列的顺序与子集，未列出的列被丢弃。
 - `formatMap` 中日期/时间类型给格式串，`BIG_DECIMAL` 给小数位数（HALF_UP，默认 6）；

From 8b44ee70469c16d1b71710df38c88337cecc2da2 Mon Sep 17 00:00:00 2001
From: camilesing <camilesing@gmail.com>
Date: Fri, 18 Sep 2026 23:29:16 +0800
Subject: [PATCH 2/3] feat(login): add a user role column, a login-page
 extension hook and Feishu config injection

Squashed port of two commits (user role foundation + Feishu login config plumbing).

Backend: DATAPOLY_SYSTEM_USER gains a user_role column through the v1.4.0 Liquibase migration in both dialects (new rows default to USER, existing rows are backfilled to ADMIN so nobody locks themselves out on upgrade). AccessToken carries the role, SystemUserService.login fills it from the row, SystemUserEntity maps user_role, and SystemUserDao.insert() lets account providers such as the Feishu extension create their users.

Login page: src/views/login/index.vue renders an extension.loginExtras component list below the password button (the fifth compile-time extension hook) and stores the role in sessionStorage; the in-repo stub contributes an empty array, so a plain build renders nothing new.

Deployment: the Feishu OAuth settings (DATAPOLY_FEISHU_ENABLED / APP_ID / APP_SECRET / REDIRECT_URI) are wired through docker-compose (values come from the ignored install/.env), conf/config.ini and datapolyctl.sh - a key is exported only when it has a value, since an empty string breaks relaxed binding of the boolean switch.

The Feishu OAuth backend and the login button live in the separate datapoly-extension repository.
---
 AGENTS.md                                     |  5 ++--
 build-docker/install/docker-compose.yml       |  6 +++++
 .../java/com/cs/common/dto/AccessToken.java   |  3 +++
 .../cs/core/service/SystemUserService.java    |  2 +-
 .../src/main/assembly/bin/datapolyctl.sh      | 14 +++++++++++
 .../src/main/assembly/conf/config.ini         | 12 ++++++++-
 .../src/extension-stub/index.js               |  5 +++-
 datapoly-manager-ui/src/views/login/index.vue | 25 +++++++++++++++++++
 .../db/changelog/db.changelog-master.yaml     |  2 ++
 .../resources/db/changelog/log-v1.4.0.yaml    |  9 +++++++
 .../db/migration/V1_4_0__user-role-ddl.sql    |  8 ++++++
 .../pg/changelog/db.changelog-master.yaml     |  2 ++
 .../resources/pg/changelog/log-v1.4.0.yaml    |  9 +++++++
 .../pg/migration/V1_4_0__user-role-ddl.sql    |  8 ++++++
 .../com/cs/persistence/dao/SystemUserDao.java |  9 +++++++
 .../persistence/entity/SystemUserEntity.java  |  4 +++
 .../core/service/SystemUserServiceTest.java   |  3 +++
 17 files changed, 121 insertions(+), 5 deletions(-)
 create mode 100644 datapoly-manager/src/main/resources/db/changelog/log-v1.4.0.yaml
 create mode 100644 datapoly-manager/src/main/resources/db/migration/V1_4_0__user-role-ddl.sql
 create mode 100644 datapoly-manager/src/main/resources/pg/changelog/log-v1.4.0.yaml
 create mode 100644 datapoly-manager/src/main/resources/pg/migration/V1_4_0__user-role-ddl.sql

diff --git a/AGENTS.md b/AGENTS.md
index 283fa1b..6a3497b 100644
--- a/AGENTS.md
+++ b/AGENTS.md
@@ -27,7 +27,7 @@ Groovy 沙箱默认启用但不是 JVM 隔离：勿把脚本编写权开放给
 
 ## 三、凭据外部化
 
-真实值一律环境变量注入，禁止写回仓库 yaml：`DATAPOLY_ADMIN_PASSWORD`、`DATAPOLY_REDIS_PASSWORD`、`DATAPOLY_DS_AES_KEY`（轮换前须用旧密钥导出重录）、`DATAPOLY_CORS_ALLOWED_ORIGINS`、compose 的 `MYSQL_ROOT_PASSWORD`/`MYSQL_PASSWORD`（演示默认 123456）。演示凭据 admin/123456、test/test 正式部署必须修改；actuator 已收窄为 health,info。
+真实值一律环境变量注入，禁止写回仓库 yaml：`DATAPOLY_ADMIN_PASSWORD`、`DATAPOLY_REDIS_PASSWORD`、`DATAPOLY_DS_AES_KEY`（轮换前须用旧密钥导出重录）、`DATAPOLY_CORS_ALLOWED_ORIGINS`、compose 的 `MYSQL_ROOT_PASSWORD`/`MYSQL_PASSWORD`（演示默认 123456）。演示凭据 admin/123456、test/test 正式部署必须修改；actuator 已收窄为 health,info。飞书登录凭证 `DATAPOLY_FEISHU_APP_ID`/`DATAPOLY_FEISHU_APP_SECRET` 同样只从环境变量注入：compose 取被忽略的 `install/.env`，发行版 `conf/config.ini` 留空即关闭（`datapolyctl.sh` 只在键有值时导出，空串会让布尔属性宽松绑定失败）。
 
 ## 四、端点防护
 
@@ -37,6 +37,7 @@ Groovy 沙箱默认启用但不是 JVM 隔离：勿把脚本编写权开放给
 
 - 新增 JDBC 代码资源必须 try-with-resources；firewall 规则行被删时网关按"全拒绝"处理（fail-closed，属预期）。
 - DataTask 投递 Sink 仓库内置零实现，外部以 Spring Bean / `META-INF/services` 注册（SPI `com.cs.common.datatask.DataTaskSink`）；宿主可自行维护本地扩展：在顶层 `datapoly-extension/`（已被 .gitignore 排除，独立 git 仓库）下用 `backend/` 放 Maven 扩展模块（依赖钉版在模块自身 pom、不进根 reactor，由入库脚本 build-extension.sh 在宿主机 JDK 25 构建后投放 lib-extra/，随发行版装配进各服务 classpath）、`front/` 放扩展 UI；API 扩展点 `ApiAssignmentPostProcessor` 注册方式相同、须同步执行且保持轻量。详见 docs/*/data-task.md。
-- 默认前端扩展目录 `datapoly-extension/front`（同被 .gitignore 排除）经 datapoly-manager-ui 编译期装配：webpack `@extension` 别名自动探测该目录 `src/index.js`（见 build/webpack.base.conf.js）、`src/extension-stub` 为缺省回退、扩展路由与 i18n 词条在 manager-ui 入口深合并——这四处钩子文件（build/webpack.base.conf.js、src/extension-stub、src/router、src/main.js）勿移除或改名；目录不存在时 CI 与普通构建不受影响。front 自带 `package.json` 可直接启动（`npm run dev`，复用宿主 webpack 链，前置为宿主 node_modules 已安装、Node 24——dev server 与生产构建均已实测）。
+- 默认前端扩展目录 `datapoly-extension/front`（同被 .gitignore 排除）经 datapoly-manager-ui 编译期装配：webpack `@extension` 别名自动探测该目录 `src/index.js`（见 build/webpack.base.conf.js）、`src/extension-stub` 为缺省回退、扩展路由与 i18n 词条在 manager-ui 入口深合并、登录页扩展区渲染 `@extension` 的 `loginExtras` 组件数组（`src/views/login/index.vue`，stub 为空数组）——这五处钩子（build/webpack.base.conf.js、src/extension-stub、src/router、src/main.js、src/views/login/index.vue 的 loginExtras 挂载点）勿移除或改名；目录不存在时 CI 与普通构建不受影响。front 自带 `package.json` 可直接启动（`npm run dev`，复用宿主 webpack 链，前置为宿主 node_modules 已安装、Node 24——dev server 与生产构建均已实测）。
+- 用户角色见 `DATAPOLY_SYSTEM_USER.user_role`（`ADMIN`/`USER`，v1.4.0 迁移：新行默认 `USER`、存量行回填 `ADMIN`），登录响应 `AccessToken.role` 带回该值；角色只决定界面可见范围，鉴权仍以 token 为准，按角色的端点拦截须同时校验数据库中的角色而非前端传来的值。
 - 宿主扩展 jar 经根目录 `lib-extra/` 投放点进入发行版 `lib/common/`（`package.xml` 打包该目录 `*.jar`；目录只占位入库，jar 永不入库）。扩展为独立 git 仓库（内部 GitLab，front+backend 一体）：`build-extension.sh` 按环境变量 `DATAPOLY_EXTENSION_GIT_URL`（真实地址不入库，CI 注入）+ `DATAPOLY_EXTENSION_GIT_REF`（默认 master）浅克隆到 `datapoly-extension/`（仍被 .gitignore 排除），目录已存在则按本地工作区构建（宿主机 JDK 25 优先、低于 25 不可用）、`DATAPOLY_EXTENSION_FORCE_SYNC=1` 强制覆盖本地改动；`build.sh`/`docker-maven-build.sh` 会先调用该脚本，未配置且目录不存在时无操作（纯开源构建零影响）。本地环境变量注入（env.sh）与防误提交钩子集中在被忽略的 `dev-local/`。
 - 一次性 token 在校验时即消费（含查库兜底路径；2026-09 修复兜底不消费导致的重放）；并发首用竞态下多 executor 仍可能各放行一次（无分布式锁，已知限制）。
diff --git a/build-docker/install/docker-compose.yml b/build-docker/install/docker-compose.yml
index af7033f..2275be7 100644
--- a/build-docker/install/docker-compose.yml
+++ b/build-docker/install/docker-compose.yml
@@ -44,6 +44,12 @@ services:
       DATAPOLY_DS_ENCRYPT: false
       # 部署后请务必设置：覆盖种子 admin 口令（S5）
       DATAPOLY_ADMIN_PASSWORD: ${DATAPOLY_ADMIN_PASSWORD:-}
+      # 飞书登录（datapoly-extension-feishu，relaxed binding 映射为 datapoly.feishu.*）：
+      # 真实值由 install/.env 注入，勿写回本文件；三项凭证缺一或开关关闭时登录页不显示飞书按钮
+      DATAPOLY_FEISHU_ENABLED: ${DATAPOLY_FEISHU_ENABLED:-false}
+      DATAPOLY_FEISHU_APP_ID: ${DATAPOLY_FEISHU_APP_ID:-}
+      DATAPOLY_FEISHU_APP_SECRET: ${DATAPOLY_FEISHU_APP_SECRET:-}
+      DATAPOLY_FEISHU_REDIRECT_URI: ${DATAPOLY_FEISHU_REDIRECT_URI:-}
     #      DATAPOLY_MANAGER_URL: http://www.example.com:8090
     #      DATAPOLY_GATEWAY_URL: http://www.example.com:8091
     depends_on:
diff --git a/datapoly-common/src/main/java/com/cs/common/dto/AccessToken.java b/datapoly-common/src/main/java/com/cs/common/dto/AccessToken.java
index 5e3031d..da3be45 100644
--- a/datapoly-common/src/main/java/com/cs/common/dto/AccessToken.java
+++ b/datapoly-common/src/main/java/com/cs/common/dto/AccessToken.java
@@ -30,4 +30,7 @@ public class AccessToken implements Serializable {
 
     @Schema(description = "有效期(时间段，单位:秒)")
     private Long expireSeconds;
+
+    @Schema(description = "角色(ADMIN/USER)")
+    private String role;
 }
diff --git a/datapoly-core/src/main/java/com/cs/core/service/SystemUserService.java b/datapoly-core/src/main/java/com/cs/core/service/SystemUserService.java
index dcbbc7a..5e2e3d3 100644
--- a/datapoly-core/src/main/java/com/cs/core/service/SystemUserService.java
+++ b/datapoly-core/src/main/java/com/cs/core/service/SystemUserService.java
@@ -46,7 +46,7 @@ public AccessToken login(String username, String password) {
         String token = TokenUtils.generateValue();
         CacheUtils.put(token, user);
         AccessToken accessTokenWrapper = new AccessToken(user.getRealName(), user.getUsername(), token,
-                System.currentTimeMillis() / 1000, CacheUtils.CACHE_DURATION_SECONDS);
+                System.currentTimeMillis() / 1000, CacheUtils.CACHE_DURATION_SECONDS, user.getRole());
         return accessTokenWrapper;
     }
 
diff --git a/datapoly-dist/src/main/assembly/bin/datapolyctl.sh b/datapoly-dist/src/main/assembly/bin/datapolyctl.sh
index 938d745..9176db1 100644
--- a/datapoly-dist/src/main/assembly/bin/datapolyctl.sh
+++ b/datapoly-dist/src/main/assembly/bin/datapolyctl.sh
@@ -55,6 +55,20 @@ export JSON_TIMEZONE=$(get_config_value "JSON_TIMEZONE" "${APP_CONF_PATH}/config
 export DATAPOLY_MANAGER_URL=$(get_config_value "DATAPOLY_MANAGER_URL" "${APP_CONF_PATH}/config.ini")
 export DATAPOLY_GATEWAY_URL=$(get_config_value "DATAPOLY_GATEWAY_URL" "${APP_CONF_PATH}/config.ini")
 
+# 飞书登录（datapoly-extension-feishu 扩展模块）：config.ini 里留空即关闭。仅在确实
+# 配置了值时才导出——否则会把 enabled 这类布尔属性导出成空串，宽松绑定会因无法转换而启动失败
+export_feishu_config() {
+    local key=$1
+    local value=$(get_config_value "$key" "${APP_CONF_PATH}/config.ini")
+    if [ -n "$value" ]; then
+        export "$key=$value"
+    fi
+}
+export_feishu_config "DATAPOLY_FEISHU_ENABLED"
+export_feishu_config "DATAPOLY_FEISHU_APP_ID"
+export_feishu_config "DATAPOLY_FEISHU_APP_SECRET"
+export_feishu_config "DATAPOLY_FEISHU_REDIRECT_URI"
+
 # JVM参数可以在这里设置
 # 堆 4G、年轻代/老年代 1:3：长驻对象（Hazelcast token/API 响应缓存、Eureka、Spring 框架
 # 对象）占堆内大头，老年代空间优先；年轻代 1G 足以容纳数据任务流式批次与 ≤200 行的
diff --git a/datapoly-dist/src/main/assembly/conf/config.ini b/datapoly-dist/src/main/assembly/conf/config.ini
index d2a4885..a09f0be 100644
--- a/datapoly-dist/src/main/assembly/conf/config.ini
+++ b/datapoly-dist/src/main/assembly/conf/config.ini
@@ -45,4 +45,14 @@ DATAPOLY_DS_ENCRYPT=false
 
 # 外部配置化网关/管理地址，默认为空
 # DATAPOLY_MANAGER_URL=http://www.example.com:8090
-# DATAPOLY_GATEWAY_URL=http://www.example.com:8091
\ No newline at end of file
+# DATAPOLY_GATEWAY_URL=http://www.example.com:8091
+
+# 飞书登录（datapoly-extension-feishu 扩展模块；未装配该扩展时全部留空即可）
+# App ID / App Secret 取自飞书开发者后台「凭证与基础信息」；真实值请只填在部署现场，
+# 不要写回仓库
+# DATAPOLY_FEISHU_ENABLED=true
+# DATAPOLY_FEISHU_APP_ID=
+# DATAPOLY_FEISHU_APP_SECRET=
+# 回调地址必须与开发者后台「安全设置 → 重定向 URL」逐字一致，取对外可达的网关地址；
+# 未配置或与登记值不符时飞书会直接跳失败页，登录页则不显示飞书按钮
+# DATAPOLY_FEISHU_REDIRECT_URI=https://www.example.com:8091/user/feishu/callback
\ No newline at end of file
diff --git a/datapoly-manager-ui/src/extension-stub/index.js b/datapoly-manager-ui/src/extension-stub/index.js
index 9672539..702e89e 100644
--- a/datapoly-manager-ui/src/extension-stub/index.js
+++ b/datapoly-manager-ui/src/extension-stub/index.js
@@ -1,8 +1,11 @@
 // Use of this source code is governed by a BSD-style license
 // Compile-time extension fallback: the webpack '@extension' alias resolves here when
-// ../../datapoly-extension/front/src is absent, keeping router and i18n assembly no-ops.
+// ../../datapoly-extension/front/src is absent, keeping router, i18n and login-page
+// assembly no-ops.
 export default {
   routes: [],
+  // Components rendered below the password login button by src/views/login/index.vue.
+  loginExtras: [],
   i18n: {
     'zh-CN': {},
     'en-US': {}
diff --git a/datapoly-manager-ui/src/views/login/index.vue b/datapoly-manager-ui/src/views/login/index.vue
index 4da167d..6938a3f 100644
--- a/datapoly-manager-ui/src/views/login/index.vue
+++ b/datapoly-manager-ui/src/views/login/index.vue
@@ -43,6 +43,16 @@
         </button>
       </form>
 
+      <!-- Extension mount point (contract: datapoly-extension/front exports `loginExtras`,
+           a component list rendered below the password login button; the in-repo stub keeps
+           it empty, so a plain open-source build renders nothing here). -->
+      <div class="extension-login"
+           v-if="loginExtras.length">
+        <component v-for="(extra, index) in loginExtras"
+                   :key="index"
+                   :is="extra"></component>
+      </div>
+
       <div class="footer-links">
         <p class="copyright">© {{ currentYear }} The DataPoly Authors</p>
       </div>
@@ -51,6 +61,8 @@
 </template>
 
 <script>
+import extension from '@extension'
+
 export default {
   name: 'LoginPage',
   data () {
@@ -59,6 +71,9 @@ export default {
         username: '',
         password: ''
       },
+      // Components contributed by the UI extension (e.g. the Feishu login button);
+      // empty when the '@extension' entry is the in-repo stub.
+      loginExtras: extension.loginExtras || [],
       errors: {},
       loading: false
     }
@@ -122,6 +137,7 @@ export default {
           window.sessionStorage.setItem('token', res.data.data.accessToken);
           window.sessionStorage.setItem('username', this.loginForm.username);
           window.sessionStorage.setItem('realname', res.data.data.realName);
+          window.sessionStorage.setItem('role', res.data.data.role || '');
           this.$http.get("/datapoly/manager/api/v1/health/version").then(
             res => {
               if (0 === res.data.code) {
@@ -356,6 +372,15 @@ export default {
   }
 }
 
+.extension-login {
+  position: relative;
+  z-index: 1;
+  margin-top: 24px;
+  padding-top: 20px;
+  border-top: 1px solid #334155;
+  text-align: center;
+}
+
 .footer-links {
   text-align: center;
   margin-top: 30px;
diff --git a/datapoly-manager/src/main/resources/db/changelog/db.changelog-master.yaml b/datapoly-manager/src/main/resources/db/changelog/db.changelog-master.yaml
index 691c684..e1f9c7d 100644
--- a/datapoly-manager/src/main/resources/db/changelog/db.changelog-master.yaml
+++ b/datapoly-manager/src/main/resources/db/changelog/db.changelog-master.yaml
@@ -7,3 +7,5 @@ databaseChangeLog:
       file: classpath:db/changelog/log-v1.2.0.yaml
   - include:
       file: classpath:db/changelog/log-v1.3.0.yaml
+  - include:
+      file: classpath:db/changelog/log-v1.4.0.yaml
diff --git a/datapoly-manager/src/main/resources/db/changelog/log-v1.4.0.yaml b/datapoly-manager/src/main/resources/db/changelog/log-v1.4.0.yaml
new file mode 100644
index 0000000..fe92924
--- /dev/null
+++ b/datapoly-manager/src/main/resources/db/changelog/log-v1.4.0.yaml
@@ -0,0 +1,9 @@
+databaseChangeLog:
+  - changeSet:
+      id: 1.4.0-user-role-ddl
+      author: datapoly
+      runOnChange: false
+      changes:
+        - sqlFile:
+            encoding: UTF-8
+            path: db/migration/V1_4_0__user-role-ddl.sql
\ No newline at end of file
diff --git a/datapoly-manager/src/main/resources/db/migration/V1_4_0__user-role-ddl.sql b/datapoly-manager/src/main/resources/db/migration/V1_4_0__user-role-ddl.sql
new file mode 100644
index 0000000..1bb3bbc
--- /dev/null
+++ b/datapoly-manager/src/main/resources/db/migration/V1_4_0__user-role-ddl.sql
@@ -0,0 +1,8 @@
+-- 系统用户角色（v1.4.0）：ADMIN=管理员（可增删改查），USER=普通用户（只读/调用）。
+-- 先以默认值 USER 建列（新行安全，扩展自动开通的账号即普通用户），再把存量账号（含 admin）
+-- 回填为 ADMIN：升级后既有登录者的权限与升级前完全一致，不会把自己锁死。
+ALTER TABLE `DATAPOLY_SYSTEM_USER`
+    ADD COLUMN `user_role` varchar(32) not null default 'USER' comment '角色:ADMIN/USER';
+
+UPDATE `DATAPOLY_SYSTEM_USER`
+SET `user_role` = 'ADMIN';
\ No newline at end of file
diff --git a/datapoly-manager/src/main/resources/pg/changelog/db.changelog-master.yaml b/datapoly-manager/src/main/resources/pg/changelog/db.changelog-master.yaml
index 0cd6696..16404e9 100644
--- a/datapoly-manager/src/main/resources/pg/changelog/db.changelog-master.yaml
+++ b/datapoly-manager/src/main/resources/pg/changelog/db.changelog-master.yaml
@@ -7,3 +7,5 @@ databaseChangeLog:
       file: classpath:pg/changelog/log-v1.2.0.yaml
   - include:
       file: classpath:pg/changelog/log-v1.3.0.yaml
+  - include:
+      file: classpath:pg/changelog/log-v1.4.0.yaml
diff --git a/datapoly-manager/src/main/resources/pg/changelog/log-v1.4.0.yaml b/datapoly-manager/src/main/resources/pg/changelog/log-v1.4.0.yaml
new file mode 100644
index 0000000..19cf14b
--- /dev/null
+++ b/datapoly-manager/src/main/resources/pg/changelog/log-v1.4.0.yaml
@@ -0,0 +1,9 @@
+databaseChangeLog:
+  - changeSet:
+      id: 1.4.0-user-role-ddl
+      author: datapoly
+      runOnChange: false
+      changes:
+        - sqlFile:
+            encoding: UTF-8
+            path: pg/migration/V1_4_0__user-role-ddl.sql
\ No newline at end of file
diff --git a/datapoly-manager/src/main/resources/pg/migration/V1_4_0__user-role-ddl.sql b/datapoly-manager/src/main/resources/pg/migration/V1_4_0__user-role-ddl.sql
new file mode 100644
index 0000000..60de5ae
--- /dev/null
+++ b/datapoly-manager/src/main/resources/pg/migration/V1_4_0__user-role-ddl.sql
@@ -0,0 +1,8 @@
+-- 系统用户角色（v1.4.0）：ADMIN=管理员（可增删改查），USER=普通用户（只读/调用）。
+-- 先以默认值 USER 建列（新行安全，扩展自动开通的账号即普通用户），再把存量账号（含 admin）
+-- 回填为 ADMIN：升级后既有登录者的权限与升级前完全一致，不会把自己锁死。
+ALTER TABLE datapoly_system_user
+    ADD COLUMN user_role varchar(32) NOT NULL DEFAULT 'USER';
+
+UPDATE datapoly_system_user
+SET user_role = 'ADMIN';
\ No newline at end of file
diff --git a/datapoly-persistence/src/main/java/com/cs/persistence/dao/SystemUserDao.java b/datapoly-persistence/src/main/java/com/cs/persistence/dao/SystemUserDao.java
index c83b1c3..4f0a435 100644
--- a/datapoly-persistence/src/main/java/com/cs/persistence/dao/SystemUserDao.java
+++ b/datapoly-persistence/src/main/java/com/cs/persistence/dao/SystemUserDao.java
@@ -34,4 +34,13 @@ public void updateUserPassword(String username, String newPassword) {
         }
     }
 
+    /**
+     * Creates a user row; the generated id is filled back into the entity.
+     * Account providers that authenticate outside DataPoly create their rows
+     * through this method, so the caller owns the password/salt and role values.
+     */
+    public void insert(SystemUserEntity userEntity) {
+        systemUserMapper.insert(userEntity);
+    }
+
 }
diff --git a/datapoly-persistence/src/main/java/com/cs/persistence/entity/SystemUserEntity.java b/datapoly-persistence/src/main/java/com/cs/persistence/entity/SystemUserEntity.java
index 6fd99b7..0d8399d 100644
--- a/datapoly-persistence/src/main/java/com/cs/persistence/entity/SystemUserEntity.java
+++ b/datapoly-persistence/src/main/java/com/cs/persistence/entity/SystemUserEntity.java
@@ -38,6 +38,10 @@ public class SystemUserEntity {
     @TableField("locked")
     private Boolean locked;
 
+    /** Role name: {@code ADMIN} (full access) or {@code USER} (read/invoke only). */
+    @TableField("user_role")
+    private String role;
+
     @TableField(value = "create_time", insertStrategy = FieldStrategy.NEVER, updateStrategy = FieldStrategy.NEVER)
     private Timestamp createTime;
 
diff --git a/datapoly-test/src/test/java/com/cs/core/service/SystemUserServiceTest.java b/datapoly-test/src/test/java/com/cs/core/service/SystemUserServiceTest.java
index 1193da7..8979179 100644
--- a/datapoly-test/src/test/java/com/cs/core/service/SystemUserServiceTest.java
+++ b/datapoly-test/src/test/java/com/cs/core/service/SystemUserServiceTest.java
@@ -59,6 +59,8 @@ public void loginSuccessReturnsAccessToken() throws Exception {
         Assert.assertNotNull(accessToken);
         Assert.assertEquals(USERNAME, accessToken.getAppKey());
         Assert.assertFalse(accessToken.getAccessToken().isEmpty());
+        // The role travels with the login response so the UI can filter by it
+        Assert.assertEquals("ADMIN", accessToken.getRole());
     }
 
     @Test
@@ -131,6 +133,7 @@ private static SystemUserEntity buildUser(String username, String rawPassword) {
         user.setRealName(username);
         user.setSalt(salt);
         user.setPassword(PasswordUtils.encryptPassword(rawPassword, salt));
+        user.setRole("ADMIN");
         return user;
     }
 

From 062ac4317d4cccf388c4f1c1bfb1695f6730366e Mon Sep 17 00:00:00 2001
From: camilesing <camilesing@gmail.com>
Date: Fri, 18 Sep 2026 23:33:35 +0800
Subject: [PATCH 3/3] feat(datatask): add a statement-delivery extension point
 for source-side exports

A delivery provider can now claim a definition and run its rendered statement itself. The new optional interface com.cs.common.datatask.DataTaskStatementSink is consulted once per job, after rendering and before any JDBC work; on a claim the engine skips the whole row pipeline (no session, no result set, no row limit or reshaping), calls executeStatement and records the returned SinkOutcome like any other artifact with artifactInfo.statementDelegated=true. Declined definitions and non-query statements keep their previous path, so one sink can serve both modes.

A delegated statement routinely outlives the lease window, so a lazily started daemon (data-task-lease-keeper, stopped by @PreDestroy) refreshes the lease every lease-seconds/3 while it blocks. Cancellation only reaches a statement before submission: a late cancel keeps the artifact and records artifactInfo.cancelRequested=true instead of dropping it.

DataTaskStatementRequest carries the rendered SQL, its remaining bind values, the query flag, the pooled datasource, the submitter and the cancel probe. Adds seven engine tests and documents both semantics in docs/{en,zh}/data-task.md.
---
 AGENTS.md                                     |   2 +-
 CHANGELOG.md                                  |  15 ++
 .../datatask/DataTaskStatementRequest.java    |  85 ++++++
 .../datatask/DataTaskStatementSink.java       |  40 +++
 .../cs/core/datatask/DataTaskJobEngine.java   | 193 ++++++++++---
 .../core/datatask/DataTaskJobEngineTest.java  | 254 +++++++++++++++++-
 docs/en/data-task.md                          |  53 ++++
 docs/zh/data-task.md                          |  46 ++++
 8 files changed, 646 insertions(+), 42 deletions(-)
 create mode 100644 datapoly-common/src/main/java/com/cs/common/datatask/DataTaskStatementRequest.java
 create mode 100644 datapoly-common/src/main/java/com/cs/common/datatask/DataTaskStatementSink.java

diff --git a/AGENTS.md b/AGENTS.md
index 6a3497b..e3603cf 100644
--- a/AGENTS.md
+++ b/AGENTS.md
@@ -36,7 +36,7 @@ Groovy 沙箱默认启用但不是 JVM 隔离：勿把脚本编写权开放给
 ## 五、其他约束
 
 - 新增 JDBC 代码资源必须 try-with-resources；firewall 规则行被删时网关按"全拒绝"处理（fail-closed，属预期）。
-- DataTask 投递 Sink 仓库内置零实现，外部以 Spring Bean / `META-INF/services` 注册（SPI `com.cs.common.datatask.DataTaskSink`）；宿主可自行维护本地扩展：在顶层 `datapoly-extension/`（已被 .gitignore 排除，独立 git 仓库）下用 `backend/` 放 Maven 扩展模块（依赖钉版在模块自身 pom、不进根 reactor，由入库脚本 build-extension.sh 在宿主机 JDK 25 构建后投放 lib-extra/，随发行版装配进各服务 classpath）、`front/` 放扩展 UI；API 扩展点 `ApiAssignmentPostProcessor` 注册方式相同、须同步执行且保持轻量。详见 docs/*/data-task.md。
+- DataTask 投递 Sink 仓库内置零实现，外部以 Spring Bean / `META-INF/services` 注册（SPI `com.cs.common.datatask.DataTaskSink`；服务端导出类投递——如 MaxCompute `UNLOAD` 直写对象存储——另实现可选接口 `DataTaskStatementSink`：`handlesStatement` 判定、`executeStatement` 执行，引擎跳过整条行式管线并在阻塞期间代为续租，取消只在提交前拦得住）；宿主可自行维护本地扩展：在顶层 `datapoly-extension/`（已被 .gitignore 排除，独立 git 仓库）下用 `backend/` 放 Maven 扩展模块（依赖钉版在模块自身 pom、不进根 reactor，由入库脚本 build-extension.sh 在宿主机 JDK 25 构建后投放 lib-extra/，随发行版装配进各服务 classpath）、`front/` 放扩展 UI；API 扩展点 `ApiAssignmentPostProcessor` 注册方式相同、须同步执行且保持轻量。详见 docs/*/data-task.md。
 - 默认前端扩展目录 `datapoly-extension/front`（同被 .gitignore 排除）经 datapoly-manager-ui 编译期装配：webpack `@extension` 别名自动探测该目录 `src/index.js`（见 build/webpack.base.conf.js）、`src/extension-stub` 为缺省回退、扩展路由与 i18n 词条在 manager-ui 入口深合并、登录页扩展区渲染 `@extension` 的 `loginExtras` 组件数组（`src/views/login/index.vue`，stub 为空数组）——这五处钩子（build/webpack.base.conf.js、src/extension-stub、src/router、src/main.js、src/views/login/index.vue 的 loginExtras 挂载点）勿移除或改名；目录不存在时 CI 与普通构建不受影响。front 自带 `package.json` 可直接启动（`npm run dev`，复用宿主 webpack 链，前置为宿主 node_modules 已安装、Node 24——dev server 与生产构建均已实测）。
 - 用户角色见 `DATAPOLY_SYSTEM_USER.user_role`（`ADMIN`/`USER`，v1.4.0 迁移：新行默认 `USER`、存量行回填 `ADMIN`），登录响应 `AccessToken.role` 带回该值；角色只决定界面可见范围，鉴权仍以 token 为准，按角色的端点拦截须同时校验数据库中的角色而非前端传来的值。
 - 宿主扩展 jar 经根目录 `lib-extra/` 投放点进入发行版 `lib/common/`（`package.xml` 打包该目录 `*.jar`；目录只占位入库，jar 永不入库）。扩展为独立 git 仓库（内部 GitLab，front+backend 一体）：`build-extension.sh` 按环境变量 `DATAPOLY_EXTENSION_GIT_URL`（真实地址不入库，CI 注入）+ `DATAPOLY_EXTENSION_GIT_REF`（默认 master）浅克隆到 `datapoly-extension/`（仍被 .gitignore 排除），目录已存在则按本地工作区构建（宿主机 JDK 25 优先、低于 25 不可用）、`DATAPOLY_EXTENSION_FORCE_SYNC=1` 强制覆盖本地改动；`build.sh`/`docker-maven-build.sh` 会先调用该脚本，未配置且目录不存在时无操作（纯开源构建零影响）。本地环境变量注入（env.sh）与防误提交钩子集中在被忽略的 `dev-local/`。
diff --git a/CHANGELOG.md b/CHANGELOG.md
index ee724de..15ee6f6 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -12,6 +12,21 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/).
 
 ### Added
 
+- DataTask statement sinks (optional capability, backward compatible): the new
+  interface `com.cs.common.datatask.DataTaskStatementSink` lets a delivery
+  provider claim a definition and run its rendered statement itself, for
+  exports that must complete inside the source engine (MaxCompute
+  `UNLOAD ... INTO LOCATION 'oss://...'`) instead of streaming rows through the
+  executor. `DataTaskJobEngine` consults the sink once, right after rendering
+  and before any JDBC work: on a claim it skips the whole row pipeline (no
+  session, no result set, no row limit or reshaping) and records the returned
+  `SinkOutcome` like any other artifact. Because such a statement can outlive
+  `lease-seconds`, a lazily started daemon refreshes the job lease every
+  `lease-seconds/3` while it blocks; the cancel probe can only stop a statement
+  before submission, so a late cancel keeps the artifact and is recorded as
+  `artifactInfo.cancelRequested`. Existing sinks are untouched — a sink that
+  never claims a definition behaves exactly as before. Documented in
+  `docs/{zh,en}/data-task.md` §5.
 - DataTask sink contract (net-neutral extension): `SinkRequest` now carries
   `columnMetadata` (per-column JDBC type hints, shaped through the same
   projection as the columns) and the terminal `DataTaskEvent` includes the
diff --git a/datapoly-common/src/main/java/com/cs/common/datatask/DataTaskStatementRequest.java b/datapoly-common/src/main/java/com/cs/common/datatask/DataTaskStatementRequest.java
new file mode 100644
index 0000000..8312ddc
--- /dev/null
+++ b/datapoly-common/src/main/java/com/cs/common/datatask/DataTaskStatementRequest.java
@@ -0,0 +1,85 @@
+// Use of this source code is governed by a BSD-style license
+package com.cs.common.datatask;
+
+import com.cs.common.enums.ProductTypeEnum;
+import lombok.Builder;
+import lombok.Data;
+
+import javax.sql.DataSource;
+import java.util.List;
+import java.util.Map;
+import java.util.function.BooleanSupplier;
+
+/**
+ * Everything a {@link DataTaskStatementSink} needs to decide about, and then run, one
+ * definition's statement outside the engine's row pipeline: the rendered SQL, the
+ * definition's delivery configuration, the datasource it was authored against and the
+ * cooperative cancellation probe.
+ *
+ * <p>The engine builds this object right after rendering the template and before any
+ * JDBC work, so a sink can inspect it in
+ * {@link DataTaskStatementSink#handlesStatement(DataTaskStatementRequest)} without
+ * paying for a connection.</p>
+ */
+@Data
+@Builder
+public class DataTaskStatementRequest {
+
+    private Long jobId;
+
+    private String taskName;
+
+    /** Registration type of the sink this definition selected. */
+    private String sinkType;
+
+    /**
+     * Opaque JSON configuration stored on the task definition; its schema belongs to
+     * the sink implementation. Secrets placed here are persisted in plain text in the
+     * meta store — providers should reference server-side credentials instead.
+     */
+    private String sinkConfig;
+
+    /**
+     * Statement rendered from the definition template. For server-side execution this
+     * text must be complete: drivers of these engines generally accept no bind
+     * parameters, so callers author such definitions with {@code ${param}} inlining
+     * ({@code dollarAllowed}) rather than {@code #{param}}.
+     */
+    private String sql;
+
+    /**
+     * Bind values the renderer left behind as placeholders; a non-empty list means
+     * {@link #sql} is not executable as-is and the definition must be rewritten with
+     * inlined parameters.
+     */
+    private List<Object> sqlParameters;
+
+    /**
+     * Whether {@link #sql} is a query (SELECT/WITH, ignoring leading whitespace, parens and
+     * comments). Server-side exports wrap the statement in their own command — MaxCompute's
+     * {@code UNLOAD FROM (<sql>)}, for instance — which only accepts a query, so a sink
+     * claiming such definitions must leave everything else (DML, DDL, engine-specific
+     * statements) on the ordinary path.
+     */
+    private boolean query;
+
+    /** Boundary values that produced {@link #sql}, for logging and diagnostics. */
+    private Map<String, Object> params;
+
+    private Long datasourceId;
+
+    private ProductTypeEnum product;
+
+    /** Pooled datasource of the definition, shared with the synchronous API path. */
+    private DataSource dataSource;
+
+    /** Account id of the submitter, as carried by the job record. */
+    private String submittedBy;
+
+    /**
+     * True once the job row left RUNNING or a cancel was requested — the same predicate
+     * the row pipeline uses. Long-running statements should poll it before submitting;
+     * an already-submitted server-side job will finish regardless.
+     */
+    private BooleanSupplier cancelled;
+}
\ No newline at end of file
diff --git a/datapoly-common/src/main/java/com/cs/common/datatask/DataTaskStatementSink.java b/datapoly-common/src/main/java/com/cs/common/datatask/DataTaskStatementSink.java
new file mode 100644
index 0000000..ae7b3c7
--- /dev/null
+++ b/datapoly-common/src/main/java/com/cs/common/datatask/DataTaskStatementSink.java
@@ -0,0 +1,40 @@
+// Use of this source code is governed by a BSD-style license
+package com.cs.common.datatask;
+
+/**
+ * Optional capability of a {@link DataTaskSink}: some deliveries must not be produced by
+ * this JVM at all. Server-side exports (MaxCompute {@code UNLOAD}, warehouse
+ * {@code INSERT OVERWRITE DIRECTORY}, ...) complete inside the source engine and write
+ * straight to the target store, so the engine must hand over the statement instead of
+ * executing it and pushing rows.
+ *
+ * <p>{@link #handlesStatement(DataTaskStatementRequest)} is consulted once per job, right
+ * after the template is rendered and before any JDBC execution. Returning {@code true}
+ * makes the engine skip the row pipeline entirely — no {@link SinkSession}, no row limit,
+ * no column reshaping — and call {@link #executeStatement(DataTaskStatementRequest)}
+ * instead; the outcome is recorded on the job like any other sink outcome. Returning
+ * {@code false} leaves the definition on the ordinary path, so one sink can serve both
+ * modes (e.g. stream to object storage for most engines, delegate to the engine's own
+ * export command for one of them).</p>
+ *
+ * <p>Implementations own resource cleanup for the delegated path: the engine has no
+ * session to abort when the statement fails or the job is cancelled mid-flight.</p>
+ */
+public interface DataTaskStatementSink extends DataTaskSink {
+
+    /**
+     * Non-statement implementations never reach this method — a sink that claims a
+     * definition via {@link #handlesStatement} is driven through
+     * {@link #executeStatement} only. Sinks serving both modes override it normally.
+     */
+    @Override
+    default SinkSession openSession(SinkRequest request) throws Exception {
+        throw new UnsupportedOperationException(
+                type() + ": this sink is driven through executeStatement, not openSession");
+    }
+
+    /** @return true to have the engine delegate this definition's statement to the sink */
+    boolean handlesStatement(DataTaskStatementRequest request);
+
+    SinkOutcome executeStatement(DataTaskStatementRequest request) throws Exception;
+}
\ No newline at end of file
diff --git a/datapoly-core/src/main/java/com/cs/core/datatask/DataTaskJobEngine.java b/datapoly-core/src/main/java/com/cs/core/datatask/DataTaskJobEngine.java
index 4b0e624..f18fab1 100644
--- a/datapoly-core/src/main/java/com/cs/core/datatask/DataTaskJobEngine.java
+++ b/datapoly-core/src/main/java/com/cs/core/datatask/DataTaskJobEngine.java
@@ -4,6 +4,8 @@
 import com.cs.common.datatask.CellDecorator;
 import com.cs.common.datatask.ColumnMetadata;
 import com.cs.common.datatask.DataTaskSink;
+import com.cs.common.datatask.DataTaskStatementRequest;
+import com.cs.common.datatask.DataTaskStatementSink;
 import com.cs.common.datatask.SinkOutcome;
 import com.cs.common.datatask.SinkRequest;
 import com.cs.common.datatask.SinkSession;
@@ -36,10 +38,15 @@
 import org.springframework.context.ApplicationEventPublisher;
 import org.springframework.stereotype.Service;
 
+import jakarta.annotation.PreDestroy;
 import jakarta.annotation.Resource;
 import java.io.File;
 import java.sql.*;
 import java.util.*;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
 import java.util.function.BooleanSupplier;
 import java.util.function.Consumer;
 import java.util.function.Function;
@@ -52,6 +59,12 @@
  * batches through the reshaping plan into the delivery provider session, refreshes
  * progress/lease while scanning and finalizes the job row together with a Spring
  * {@link DataTaskEvent} carrying the terminal state.
+ *
+ * <p>A sink that claims the definition as a
+ * {@link DataTaskStatementSink statement sink} (server-side exports such as MaxCompute
+ * {@code UNLOAD}) bypasses the row pipeline entirely: the statement is handed over, the
+ * lease is refreshed from a background keeper while it blocks, and the returned outcome
+ * is recorded on the job like any other artifact.</p>
  */
 @Slf4j
 @Service
@@ -91,6 +104,9 @@ public class DataTaskJobEngine {
     @Value("${datapoly.data-task.max-rows-default:1000000}")
     private long maxRowsDefault;
 
+    /** Lazily started daemon keeping delegated statements' leases alive; see {@link #runDelegatedStatement}. */
+    private volatile ScheduledExecutorService leaseKeeper;
+
     /** Claim exactly one PENDING job for this worker; null once the queue is drained. */
     public Long claimNext(String workerAddr) {
         long now = System.currentTimeMillis();
@@ -116,6 +132,68 @@ protected HikariDataSource loadDataSource(DataSourceEntity dsEntity) {
         return DataSourceUtils.getHikariDataSource(dsEntity, driverPath.getAbsolutePath());
     }
 
+    /**
+     * Runs one statement a {@link DataTaskStatementSink} claimed, refreshing the job lease
+     * while it blocks: a server-side export routinely outlives the lease window, and the
+     * reaper would otherwise fail a perfectly healthy job. Progress is not reported here —
+     * the engine only sees the terminal outcome.
+     */
+    private SinkOutcome runDelegatedStatement(DataTaskStatementSink sink, DataTaskStatementRequest request,
+                                              Long jobId) throws Exception {
+        long intervalMs = Math.max(1000L, leaseSeconds * 1000L / 3);
+        ScheduledFuture<?> lease = leaseKeeper().scheduleAtFixedRate(() -> refreshLease(jobId),
+                intervalMs, intervalMs, TimeUnit.MILLISECONDS);
+        try {
+            return sink.executeStatement(request);
+        } finally {
+            lease.cancel(false);
+        }
+    }
+
+    private void refreshLease(Long jobId) {
+        try {
+            if (!dataTaskJobDao.heartbeat(jobId, 0L,
+                    new Timestamp(System.currentTimeMillis() + leaseSeconds * 1000L))) {
+                log.warn("Data task job {} is no longer RUNNING; the delegated statement is left to finish", jobId);
+            }
+        } catch (Exception e) {
+            log.warn("Failed to refresh the lease of data task job {}: {}", jobId, e.getMessage());
+        }
+    }
+
+    private ScheduledExecutorService leaseKeeper() {
+        ScheduledExecutorService keeper = leaseKeeper;
+        if (null == keeper) {
+            synchronized (this) {
+                keeper = leaseKeeper;
+                if (null == keeper) {
+                    keeper = Executors.newSingleThreadScheduledExecutor(runnable -> {
+                        Thread thread = new Thread(runnable, "data-task-lease-keeper");
+                        thread.setDaemon(true);
+                        return thread;
+                    });
+                    leaseKeeper = keeper;
+                }
+            }
+        }
+        return keeper;
+    }
+
+    @PreDestroy
+    void shutdownLeaseKeeper() {
+        ScheduledExecutorService keeper = leaseKeeper;
+        if (null != keeper) {
+            keeper.shutdownNow();
+        }
+    }
+
+    /** Cooperative-cancel predicate shared by the row pipeline and delegated statements. */
+    private boolean isJobCancelled(Long jobId) {
+        DataTaskJobEntity current = dataTaskJobDao.getById(jobId);
+        return null == current || !Objects.equals(current.getStatus(), DataTaskStatus.RUNNING)
+                || Boolean.TRUE.equals(current.getCancelRequested());
+    }
+
     public void run(Long jobId) {
         DataTaskJobEntity job = dataTaskJobDao.getById(jobId);
         if (null == job) {
@@ -157,43 +235,83 @@ public void run(Long jobId) {
                         "datatask.sink.unknown", snapshot.getSinkType());
             }
 
-            channel = new SessionChannel(dataTaskJobDao, jobId, snapshot,
-                    sinkHolder.get(), job);
-            StreamSpec streamSpec = StreamSpec.builder()
-                    .dataSource(dataSource)
-                    .product(dsEntity.getType())
-                    .sqlMeta(sqlMeta)
-                    .naming(null == snapshot.getNamingStrategy()
-                            ? NamingStrategyEnum.CAMEL_CASE : snapshot.getNamingStrategy())
-                    .cancelSupplier(channel::isJobCancelled)
-                    .rowLimit(effectiveMaxRows(snapshot.getMaxRows()))
-                    .flushIntervalMs(flushIntervalMs)
-                    .fetchSize(fetchSize)
-                    .timeoutSeconds(queryTimeoutSeconds)
-                    .progress(rows -> onProgress(jobId, rows))
-                    .build();
+            DataTaskSink resolved = sinkHolder.get();
+            boolean delegated = false;
+            if (resolved instanceof DataTaskStatementSink) {
+                // server-side delivery: the sink runs the statement itself (the data never
+                // travels through this process), so the row pipeline stays out of the way
+                DataTaskStatementSink statementSink = (DataTaskStatementSink) resolved;
+                DataTaskStatementRequest statementRequest = DataTaskStatementRequest.builder()
+                        .jobId(jobId)
+                        .taskName(job.getDefName())
+                        .sinkType(snapshot.getSinkType())
+                        .sinkConfig(snapshot.getSinkConfig())
+                        .sql(sqlMeta.getSql())
+                        .sqlParameters(sqlMeta.getParameter())
+                        .query(sqlMeta.isQuerySQL())
+                        .params(params)
+                        .datasourceId(dsEntity.getId())
+                        .product(dsEntity.getType())
+                        .dataSource(dataSource)
+                        .submittedBy(job.getSubmittedBy())
+                        .cancelled(() -> isJobCancelled(jobId))
+                        .build();
+                if (statementSink.handlesStatement(statementRequest)) {
+                    delegated = true;
+                    SinkOutcome outcome = runDelegatedStatement(statementSink, statementRequest, jobId);
+                    artifactInfo.put("statementDelegated", Boolean.TRUE);
+                    if (null != outcome) {
+                        artifactUri = outcome.getArtifactUri();
+                        if (null != outcome.getInfo()) {
+                            artifactInfo.putAll(outcome.getInfo());
+                        }
+                    }
+                    if (isRunningAndCancelled(jobId)) {
+                        // an accepted server-side export cannot be stopped: the payload is
+                        // already in the target store, so the artifact is kept and the late
+                        // cancel is recorded rather than dropped
+                        artifactInfo.put("cancelRequested", Boolean.TRUE);
+                    }
+                }
+            }
 
-            StreamResult result = streamQuery(streamSpec, channel);
-            delivered = result.getRows();
-            session = channel.session();
-            artifactInfo.put("sinkStopped", channel.isSinkAskedStop());
-
-            if (!result.isRowset()) {
-                artifactInfo.put("updateCount", result.getUpdateCount());
-                terminal = DataTaskStatus.SUCCESS;
-            } else {
-                SinkOutcome outcome = session.complete();
-                session = null; // ownership moved into complete(); nothing left to abort
-                if (null != outcome) {
-                    artifactUri = outcome.getArtifactUri();
-                    if (null != outcome.getInfo()) {
-                        artifactInfo.putAll(outcome.getInfo());
+            if (!delegated) {
+                channel = new SessionChannel(jobId, snapshot, resolved, job);
+                StreamSpec streamSpec = StreamSpec.builder()
+                        .dataSource(dataSource)
+                        .product(dsEntity.getType())
+                        .sqlMeta(sqlMeta)
+                        .naming(null == snapshot.getNamingStrategy()
+                                ? NamingStrategyEnum.CAMEL_CASE : snapshot.getNamingStrategy())
+                        .cancelSupplier(channel::isJobCancelled)
+                        .rowLimit(effectiveMaxRows(snapshot.getMaxRows()))
+                        .flushIntervalMs(flushIntervalMs)
+                        .fetchSize(fetchSize)
+                        .timeoutSeconds(queryTimeoutSeconds)
+                        .progress(rows -> onProgress(jobId, rows))
+                        .build();
+
+                StreamResult result = streamQuery(streamSpec, channel);
+                delivered = result.getRows();
+                session = channel.session();
+                artifactInfo.put("sinkStopped", channel.isSinkAskedStop());
+
+                if (!result.isRowset()) {
+                    artifactInfo.put("updateCount", result.getUpdateCount());
+                } else {
+                    SinkOutcome outcome = session.complete();
+                    session = null; // ownership moved into complete(); nothing left to abort
+                    if (null != outcome) {
+                        artifactUri = outcome.getArtifactUri();
+                        if (null != outcome.getInfo()) {
+                            artifactInfo.putAll(outcome.getInfo());
+                        }
                     }
+                    artifactInfo.put("truncated", result.isTruncated());
+                    artifactInfo.put("deliveredRows", delivered);
                 }
-                artifactInfo.put("truncated", result.isTruncated());
-                artifactInfo.put("deliveredRows", delivered);
-                terminal = DataTaskStatus.SUCCESS;
             }
+            terminal = DataTaskStatus.SUCCESS;
 
             dataTaskJobDao.finishSuccess(jobId, delivered, artifactUri,
                     JsonUtils.toJsonString(artifactInfo), now());
@@ -480,7 +598,6 @@ public static class StreamSpec {
     }
 
     private class SessionChannel implements ResultChannel {
-        private final DataTaskJobDao jobDao;
         private final Long jobId;
         private final DataTaskDefEntity spec;
         private final DataTaskSink sink;
@@ -489,9 +606,7 @@ private class SessionChannel implements ResultChannel {
         private volatile SinkSession sessionRef;
         private volatile boolean sinkAskedStop;
 
-        SessionChannel(DataTaskJobDao jobDao, Long jobId, DataTaskDefEntity spec,
-                       DataTaskSink sink, DataTaskJobEntity job) {
-            this.jobDao = jobDao;
+        SessionChannel(Long jobId, DataTaskDefEntity spec, DataTaskSink sink, DataTaskJobEntity job) {
             this.jobId = jobId;
             this.spec = spec;
             this.sink = sink;
@@ -507,9 +622,7 @@ boolean isSinkAskedStop() {
         }
 
         boolean isJobCancelled() {
-            DataTaskJobEntity current = jobDao.getById(jobId);
-            return null == current || !Objects.equals(current.getStatus(), DataTaskStatus.RUNNING)
-                    || Boolean.TRUE.equals(current.getCancelRequested());
+            return DataTaskJobEngine.this.isJobCancelled(jobId);
         }
 
         @Override
diff --git a/datapoly-test/src/test/java/com/cs/core/datatask/DataTaskJobEngineTest.java b/datapoly-test/src/test/java/com/cs/core/datatask/DataTaskJobEngineTest.java
index c6ec3bc..1a54124 100644
--- a/datapoly-test/src/test/java/com/cs/core/datatask/DataTaskJobEngineTest.java
+++ b/datapoly-test/src/test/java/com/cs/core/datatask/DataTaskJobEngineTest.java
@@ -3,6 +3,8 @@
 
 import com.cs.common.datatask.ColumnMetadata;
 import com.cs.common.datatask.DataTaskSink;
+import com.cs.common.datatask.DataTaskStatementRequest;
+import com.cs.common.datatask.DataTaskStatementSink;
 import com.cs.common.datatask.SinkOutcome;
 import com.cs.common.datatask.SinkRequest;
 import com.cs.common.datatask.SinkSession;
@@ -74,12 +76,47 @@ public SinkSession openSession(SinkRequest request) {
         }
     }
 
+    /** Same registration type as {@link StubSink}: one sink, two delivery modes. */
+    private static class StubStatementSink extends StubSink implements DataTaskStatementSink {
+        final List<DataTaskStatementRequest> executed = new ArrayList<>();
+        boolean handles = true;
+        boolean throwOnExecute;
+        boolean cancelledAtSubmission;
+        Runnable duringExecution;
+
+        @Override
+        public boolean handlesStatement(DataTaskStatementRequest request) {
+            // mirrors the real contract: only queries can be wrapped by a server-side export
+            return handles && request.isQuery();
+        }
+
+        @Override
+        public SinkOutcome executeStatement(DataTaskStatementRequest request) throws Exception {
+            executed.add(request);
+            cancelledAtSubmission = request.getCancelled().getAsBoolean();
+            if (cancelledAtSubmission) {
+                // mirrors a well-behaved server-side sink: never submit once the job asks out
+                throw new IllegalStateException("cancelled before submission");
+            }
+            if (null != duringExecution) {
+                duringExecution.run();
+            }
+            if (throwOnExecute) {
+                throw new IllegalStateException("unload rejected by MaxCompute");
+            }
+            Map<String, Object> info = new LinkedHashMap<>();
+            info.put("location", ARTIFACT_LOCATION);
+            return SinkOutcome.builder().artifactUri(ARTIFACT_LOCATION).info(info).build();
+        }
+    }
+
     private static class RecordingJobDao extends com.cs.persistence.dao.DataTaskJobDao {
         DataTaskJobEntity current;
         Long successRows;
         String successUri;
         String successInfoJson;
         String failureMessage;
+        int heartbeats;
 
         @Override
         public DataTaskJobEntity getById(Long id) {
@@ -88,6 +125,7 @@ public DataTaskJobEntity getById(Long id) {
 
         @Override
         public boolean heartbeat(Long id, long totalRows, java.sql.Timestamp leaseExpireAt) {
+            heartbeats++;
             current.setTotalRows(totalRows);
             return true;
         }
@@ -149,8 +187,11 @@ private void collect(Object event) {
 
     // ------------------------------------------------------------------ state
 
+    private static final String ARTIFACT_LOCATION = "oss://oss-cn-hangzhou-internal.aliyuncs.com/reports/orders/";
+
     private int cannedRows;
     private boolean cannedTruncated;
+    private boolean rowPipelineRan;
     private List<DataTaskEvent> events = new ArrayList<>();
 
     // ------------------------------------------------------------------ helpers
@@ -190,7 +231,7 @@ private DataTaskJobEntity jobFor(DataTaskDefEntity def) {
     }
 
     /** Overridden streamQuery plays the JDBC layer role with canned batches/results. */
-    private DataTaskJobEngine engine(RecordingJobDao jobDao, StubSink sink, final ChannelScript script) {
+    private DataTaskJobEngine engine(RecordingJobDao jobDao, DataTaskSink sink, final ChannelScript script) {
         DataTaskJobEngine engineInstance = new DataTaskJobEngine() {
             @Override
             protected HikariDataSource loadDataSource(DataSourceEntity dsEntity) {
@@ -199,6 +240,7 @@ protected HikariDataSource loadDataSource(DataSourceEntity dsEntity) {
 
             @Override
             protected StreamResult streamQuery(StreamSpec spec, ResultChannel channel) throws Exception {
+                rowPipelineRan = true;
                 script.drive(channel);
                 return StreamResult.rows(cannedRows, cannedTruncated);
             }
@@ -213,6 +255,38 @@ protected StreamResult streamQuery(StreamSpec spec, ResultChannel channel) throw
         return engineInstance;
     }
 
+    /** Definition authored for server-side delivery: one query, ${} inlined parameters. */
+    private DataTaskDefEntity statementDefinition() {
+        DataTaskDefEntity def = DataTaskDefEntity.builder()
+                .name("odps-export")
+                .datasourceId(3L)
+                .sqlText("SELECT shop_name FROM sale_detail WHERE sale_date = '${sale_date}'")
+                .namingStrategy(NamingStrategyEnum.NONE)
+                .dollarAllowed(Boolean.TRUE)
+                .maxRows(100L)
+                .sinkType("stub")
+                .sinkConfig("{\"format\":\"PARQUET\"}")
+                .enabled(Boolean.TRUE)
+                .build();
+        def.setResponseFormat(Collections.emptyMap());
+        return def;
+    }
+
+    private DataTaskJobEntity statementJob(DataTaskDefEntity def) {
+        DataTaskJobEntity job = jobFor(def);
+        job.setParamsJson("{\"sale_date\":\"2013\"}");
+        return job;
+    }
+
+    private static ChannelScript mustNotStream() {
+        return new ChannelScript() {
+            @Override
+            public void drive(DataTaskJobEngine.ResultChannel channel) {
+                Assert.fail("the row pipeline must not run for a delegated statement");
+            }
+        };
+    }
+
     // ------------------------------------------------------------------ cases
 
     @Test
@@ -355,4 +429,182 @@ public void drive(DataTaskJobEngine.ResultChannel channel) throws Exception {
         Assert.assertNotNull(jobDao.successInfoJson);
         Assert.assertTrue(jobDao.successInfoJson.contains("\"truncated\":true"));
     }
+
+    // -------------------------------------------------- delegated (statement) delivery
+
+    @Test
+    public void delegatedStatementSkipsTheRowPipelineAndRecordsItsOutcome() throws Exception {
+        final StubStatementSink sink = new StubStatementSink();
+        final RecordingJobDao jobDao = new RecordingJobDao();
+        jobDao.current = statementJob(statementDefinition());
+
+        DataTaskJobEngine localEngine = engine(jobDao, sink, mustNotStream());
+
+        localEngine.run(77L);
+
+        Assert.assertFalse(rowPipelineRan);
+        Assert.assertTrue(sink.sessions.isEmpty());
+        Assert.assertEquals(1, sink.executed.size());
+        DataTaskStatementRequest request = sink.executed.get(0);
+        Assert.assertEquals(Long.valueOf(77L), request.getJobId());
+        Assert.assertEquals("odps-export", request.getTaskName());
+        Assert.assertEquals("stub", request.getSinkType());
+        Assert.assertEquals("{\"format\":\"PARQUET\"}", request.getSinkConfig());
+        // ${} inlining leaves a complete statement: no placeholders for the engine to bind
+        Assert.assertEquals("SELECT shop_name FROM sale_detail WHERE sale_date = '2013'", request.getSql());
+        Assert.assertTrue(request.getSqlParameters().isEmpty());
+        Assert.assertEquals("2013", request.getParams().get("sale_date"));
+        Assert.assertEquals(Long.valueOf(3L), request.getDatasourceId());
+        Assert.assertEquals(ProductTypeEnum.MYSQL, request.getProduct());
+        Assert.assertNotNull(request.getDataSource());
+        Assert.assertEquals("tester", request.getSubmittedBy());
+        Assert.assertTrue("a SELECT is a query the sink may export server-side", request.isQuery());
+        // the cancel probe is read while the job is still RUNNING
+        Assert.assertFalse(sink.cancelledAtSubmission);
+
+        Assert.assertEquals(Long.valueOf(0L), jobDao.successRows);
+        Assert.assertEquals(ARTIFACT_LOCATION, jobDao.successUri);
+        Assert.assertTrue(jobDao.successInfoJson.contains("\"statementDelegated\":true"));
+        Assert.assertTrue(jobDao.successInfoJson.contains("\"sinkType\":\"stub\""));
+
+        Assert.assertEquals(1, events.size());
+        Assert.assertEquals(DataTaskStatus.SUCCESS, events.get(0).getStatus());
+        Assert.assertEquals(ARTIFACT_LOCATION, events.get(0).getArtifactUri());
+    }
+
+    @Test
+    public void statementSinkDecliningKeepsTheDefinitionOnTheRowPipeline() throws Exception {
+        final StubStatementSink sink = new StubStatementSink();
+        sink.handles = false;
+        final RecordingJobDao jobDao = new RecordingJobDao();
+        jobDao.current = statementJob(statementDefinition());
+
+        cannedRows = 1;
+        DataTaskJobEngine localEngine = engine(jobDao, sink, new ChannelScript() {
+            @Override
+            public void drive(DataTaskJobEngine.ResultChannel channel) throws Exception {
+                channel.start(new ArrayList<>(Arrays.asList("shop_name")),
+                        Collections.<ColumnMetadata>emptyList());
+                List<Object[]> batch = new ArrayList<>();
+                batch.add(new Object[]{"s1"});
+                channel.batch(batch);
+            }
+        });
+
+        localEngine.run(77L);
+
+        Assert.assertTrue(rowPipelineRan);
+        Assert.assertTrue(sink.executed.isEmpty());
+        Assert.assertEquals(1, sink.sessions.size());
+        Assert.assertTrue(sink.sessions.get(0).completed);
+        Assert.assertEquals(Long.valueOf(1L), jobDao.successRows);
+    }
+
+    @Test
+    public void nonQueryDefinitionsStayOnThePathTheyHadBefore() throws Exception {
+        final StubStatementSink sink = new StubStatementSink();
+        final RecordingJobDao jobDao = new RecordingJobDao();
+        DataTaskDefEntity def = statementDefinition();
+        // DML/DDL cannot be wrapped by a server-side export command
+        def.setSqlText("DELETE FROM sale_detail WHERE sale_date = '${sale_date}'");
+        jobDao.current = statementJob(def);
+
+        DataTaskJobEngine localEngine = engine(jobDao, sink, new ChannelScript() {
+            @Override
+            public void drive(DataTaskJobEngine.ResultChannel channel) throws Exception {
+                channel.start(new ArrayList<>(Arrays.asList("cnt")),
+                        Collections.<ColumnMetadata>emptyList());
+            }
+        });
+
+        localEngine.run(77L);
+
+        Assert.assertTrue(rowPipelineRan);
+        Assert.assertTrue(sink.executed.isEmpty());
+        Assert.assertEquals("stub://artifact", jobDao.successUri);
+        Assert.assertEquals(DataTaskStatus.SUCCESS, events.get(0).getStatus());
+    }
+
+    @Test
+    public void delegatedStatementFailureMarksTheJobFailed() throws Exception {
+        final StubStatementSink sink = new StubStatementSink();
+        sink.throwOnExecute = true;
+        final RecordingJobDao jobDao = new RecordingJobDao();
+        jobDao.current = statementJob(statementDefinition());
+
+        DataTaskJobEngine localEngine = engine(jobDao, sink, mustNotStream());
+
+        localEngine.run(77L);
+
+        Assert.assertTrue(String.valueOf(jobDao.failureMessage).contains("unload rejected"));
+        Assert.assertNull(jobDao.successUri);
+        Assert.assertEquals(1, events.size());
+        Assert.assertEquals(DataTaskStatus.FAILED, events.get(0).getStatus());
+    }
+
+    @Test
+    public void cancelRequestedBeforeSubmissionMarksTheDelegatedJobCanceled() throws Exception {
+        final StubStatementSink sink = new StubStatementSink();
+        final RecordingJobDao jobDao = new RecordingJobDao();
+        jobDao.current = statementJob(statementDefinition());
+        jobDao.current.setCancelRequested(Boolean.TRUE); // cooperatively flagged beforehand
+
+        DataTaskJobEngine localEngine = engine(jobDao, sink, mustNotStream());
+
+        localEngine.run(77L);
+
+        Assert.assertTrue(sink.cancelledAtSubmission);
+        Assert.assertNull(jobDao.successUri);
+        Assert.assertEquals(1, events.size());
+        Assert.assertEquals(DataTaskStatus.CANCELED, events.get(0).getStatus());
+    }
+
+    @Test
+    public void lateCancelKeepsTheArtifactAndIsRecordedOnIt() throws Exception {
+        final StubStatementSink sink = new StubStatementSink();
+        final RecordingJobDao jobDao = new RecordingJobDao();
+        jobDao.current = statementJob(statementDefinition());
+        // the export is already accepted by the source engine when the cancel arrives
+        sink.duringExecution = new Runnable() {
+            @Override
+            public void run() {
+                jobDao.current.setCancelRequested(Boolean.TRUE);
+            }
+        };
+
+        DataTaskJobEngine localEngine = engine(jobDao, sink, mustNotStream());
+
+        localEngine.run(77L);
+
+        Assert.assertEquals(ARTIFACT_LOCATION, jobDao.successUri);
+        Assert.assertTrue(jobDao.successInfoJson.contains("\"cancelRequested\":true"));
+        Assert.assertEquals(1, events.size());
+        Assert.assertEquals(DataTaskStatus.SUCCESS, events.get(0).getStatus());
+    }
+
+    @Test
+    public void delegatedStatementKeepsItsLeaseAliveWhileItBlocks() throws Exception {
+        final StubStatementSink sink = new StubStatementSink();
+        sink.duringExecution = new Runnable() {
+            @Override
+            public void run() {
+                try {
+                    Thread.sleep(2500L);
+                } catch (InterruptedException e) {
+                    Thread.currentThread().interrupt();
+                }
+            }
+        };
+        final RecordingJobDao jobDao = new RecordingJobDao();
+        jobDao.current = statementJob(statementDefinition());
+
+        DataTaskJobEngine localEngine = engine(jobDao, sink, mustNotStream());
+
+        localEngine.run(77L);
+
+        // without the keeper the lease would expire mid-export and the reaper would fail the job
+        Assert.assertTrue("expected periodic lease refreshes, saw " + jobDao.heartbeats,
+                jobDao.heartbeats >= 2);
+        Assert.assertEquals(DataTaskStatus.SUCCESS, events.get(0).getStatus());
+    }
 }
diff --git a/docs/en/data-task.md b/docs/en/data-task.md
index bb702cc..f40df46 100644
--- a/docs/en/data-task.md
+++ b/docs/en/data-task.md
@@ -279,6 +279,59 @@ Two auxiliary extension points:
   Listen for it in the host application to wire WebSocket/SSE/webhooks; the default interaction remains frontend
   polling of `/job/{id}`.
 
+### Statement sinks (server-side exports)
+
+Some deliveries should never be produced by this process at all: a warehouse's own export command (MaxCompute's
+`UNLOAD ... INTO LOCATION 'oss://...'`) writes the data into object storage inside the source engine, so a whole
+table never has to travel through the executor. Implement the optional interface
+`com.cs.common.datatask.DataTaskStatementSink` for those:
+
+```java
+public interface DataTaskStatementSink extends DataTaskSink {
+
+    @Override
+    default SinkSession openSession(SinkRequest request) {   // no session on this path
+        throw new UnsupportedOperationException(type() + ": driven through executeStatement");
+    }
+
+    /** Claim this definition, or leave it on the row pipeline */
+    boolean handlesStatement(DataTaskStatementRequest request);
+
+    SinkOutcome executeStatement(DataTaskStatementRequest request) throws Exception;
+}
+```
+
+`handlesStatement` is consulted once per job, after the template is rendered and before any JDBC work (no
+connection exists yet, so the decision must be cheap). Returning `true` makes the engine **skip the entire row
+pipeline** — no session, no result set, no row limit or column reshaping — and call `executeStatement` instead,
+merging the returned `artifactUri`/`info` into the job record exactly like any other sink outcome. Returning
+`false` keeps the ordinary path, so one sink can serve both modes (see the extension module's `oss` delivery,
+which switches on the datasource type).
+
+`DataTaskStatementRequest` carries `jobId / taskName / sinkType / sinkConfig / sql` (already rendered) /
+`sqlParameters` (the `#{}` bind values; a non-empty list means the statement cannot be handed over wholesale and
+has to be rewritten with `${}` inlining) / `query` (whether the rendered statement is a query — `SELECT`/`WITH`
+after leading whitespace, parens and comments) / `params` (raw boundary values, for diagnostics) /
+`datasourceId / product / dataSource` (the definition's pooled datasource, shared with the synchronous API path) /
+`submittedBy` / `cancelled` (the same cooperative-cancel predicate the row pipeline uses).
+
+Server-side export commands usually accept **queries only** (`UNLOAD FROM (<sql>)` does), so
+`handlesStatement` should include `isQuery()` in its decision: DML, DDL and definitions that already spell out a
+full export command then stay on the execution they had before, instead of changing behaviour because this path
+was introduced.
+
+Two behaviours differ from the row pipeline and implementations must know them:
+
+- **Lease**: while the statement blocks, a background keeper in the engine refreshes the job lease every
+  `lease-seconds/3`, so a server-side export may run for tens of minutes without the reaper failing it — but
+  `totalRows` stays 0, since there is no row stream to count.
+- **Cancellation**: the probe can only stop a statement *before* submission. Once the source engine accepted it
+  (an UNLOAD already submitted), MaxCompute finishes writing the files regardless; the job is still marked
+  SUCCESS with `artifactInfo.cancelRequested=true` so that files in the object store never end up without a
+  recorded artifact. Deliveries needing true mid-flight cancellation should not take this path.
+
+Cleanup on failure is the implementation's responsibility — there is no session to `abort`.
+
 ## 6. Worker configuration reference
 
 Executor-side prefix `datapoly.data-task.*` (every key can be overridden with a `DATAPOLY_DATA_TASK_*` environment
diff --git a/docs/zh/data-task.md b/docs/zh/data-task.md
index a4b178c..18ff2df 100644
--- a/docs/zh/data-task.md
+++ b/docs/zh/data-task.md
@@ -261,6 +261,52 @@ public class CsvFileSink implements DataTaskSink {
   `com.cs.core.datatask.DataTaskEvent`（字段：`jobId/defId/defName/status/totalRows/artifactUri/errorMessage/sinkType`）。
   在宿主应用监听即可对接 WebSocket/SSE/Webhook；默认交互仍是前端轮询 `/job/{id}`。
 
+### 语句型投递（服务端导出）
+
+有些投递根本不该由本进程产出：数据仓库自带的导出命令（MaxCompute 的
+`UNLOAD ... INTO LOCATION 'oss://...'`）在源引擎里跑完就把数据写进对象存储，整表数据不必穿过 executor
+再转发一次。这种需求实现可选接口 `com.cs.common.datatask.DataTaskStatementSink`：
+
+```java
+public interface DataTaskStatementSink extends DataTaskSink {
+
+    @Override
+    default SinkSession openSession(SinkRequest request) {   // 语句型 sink 不走会话
+        throw new UnsupportedOperationException(type() + ": driven through executeStatement");
+    }
+
+    /** 本次定义是否交给 sink 直接执行；false = 照常走行式管线 */
+    boolean handlesStatement(DataTaskStatementRequest request);
+
+    SinkOutcome executeStatement(DataTaskStatementRequest request) throws Exception;
+}
+```
+
+`handlesStatement` 在模板渲染之后、任何 JDBC 动作之前被调用一次（此时尚未建立连接，判定必须廉价）；
+返回 `true` 时引擎**跳过整个行式管线**——不开会话、不读结果集、不施行数上限与列整形——改为调用
+`executeStatement`，把返回的 `artifactUri`/`info` 与普通 sink 一样合并进执行记录。返回 `false` 则走常规路径，
+因此同一个 sink 可以两种模式并存（按数据源类型切换的现成例子见扩展模块的 `oss` 投递）。
+
+`DataTaskStatementRequest` 携带：`jobId / taskName / sinkType / sinkConfig / sql`（**已渲染**的语句）/
+`sqlParameters`（`#{}` 绑定值；非空即说明该语句不能整条交给源引擎，需改成 `${}` 内联） / `query`
+（渲染后的语句是否为查询，忽略前导空白/括号/注释后的 `SELECT`/`WITH`） / `params`（原始参数，诊断用）/
+`datasourceId / product / dataSource`（定义所指向数据源的连接池，与 API 数据面共用）/
+`submittedBy` / `cancelled`（与行式路径同一判据的取消探针）。
+
+服务端导出命令通常只接受**查询语句**（`UNLOAD FROM (<sql>)` 即如此），所以 `handlesStatement` 应把
+`isQuery()` 纳入判定：DML、DDL 以及用户已手写完整导出命令的定义都会因此留在原有执行路径上，不会因为
+新增这条通道而改变行为。
+
+两处与行式路径不同的语义，实现前必须知道：
+
+- **租约**：语句阻塞期间由引擎的后台续租线程按 `lease-seconds/3` 刷新租约，服务端导出跑几十分钟也不会被
+  reaper 判失联；但 `totalRows` 恒为 0，没有行流可计数。
+- **取消**：探针只在**提交前**拦得住——语句一旦被源引擎接受（UNLOAD 已提交），MaxCompute 会照常把文件写完。
+  此时引擎仍记 SUCCESS，并在 `artifactInfo.cancelRequested=true` 标注，避免出现"OSS 里有文件、记录里没产物"
+  的孤儿数据。需要行级取消语义的投递不要走这条路。
+
+失败清理由实现方负责（没有会话可以 `abort`）。
+
 ## 6、Worker 配置参考
 
 executor 侧前缀 `datapoly.data-task.*`（`application.yaml` 均可用 `DATAPOLY_DATA_TASK_*` 环境变量覆盖）：