fix: a select-only SQLRT session declines a non-select before running it

CHANGELOG.md

@@ -5,6 +5,7 @@
 - Answer the three endpoints the official JDBC driver calls before any statement runs: `POST /projects/p/authorization?sign_bearer_token` (placeholder logview token), `GET /logview/host`, and `GET /connection/mcqa` (refused by name, so a MaxQA misconfiguration stops resurfacing as an unrelated endpoint error).
 - Emit schema `columns`/`partitionKeys` as arrays instead of JSON `null`, which the Java SDK's `TunnelTableSchema` rejects — every DDL and INSERT result has no columns, so the download session the driver opens for one was unparseable.
 - Add `tests/jdbc`: acceptance against the published `odps-jdbc` artifact (offline mode), covering a statement with no result set end to end.
+- Decline non-select sub queries in a SQLRT session before executing them (`queryId: -1` with `ODPS-1850001 Non select query not supported.`, lift with `odps.sql.session.select.only=false`): a client that discovers a non-select at fetch time reruns the statement offline, so a session that had already executed an INSERT applied it twice.
 - Add resource CRUD: single-payload and Java SDK chunked part/merge uploads, `?meta` header metadata, ranged downloads, update, delete, prefix and paginated listing, plus TABLE resource metadata.
 - Add function registration (`/projects/p/registration/functions`) for Java, SQL and embedded functions with resource-reference validation; execution still returns `UnsupportedFeature`.
 - Declare `resources`, `functions` and `unsupported` surfaces in `/capabilities`.
@@ -14,6 +15,7 @@
 - Require write grants to create or update resources and functions under strict authentication; the Tunnel read-session `create` exception no longer applies to metadata verbs.
 - Add MCQA / SQLRT interactive sessions: an `SQLRT` task instance stays `Running` until the client stops it or it idles out, and its statements run as sub queries over the instance information KV (`?info`) using the Java SDK's own object status codes, with `query`/`cancel` writes and `status`/`progress`/`result`/`result_<id>` reads.
 - Declare `mcqa` in `/capabilities`, and name the remaining session gaps (named-session attach, MaxQA v2) separately from the surface that works.
+- Download an MCQA sub query's result over the instance tunnel (`?data&cached&taskname=..&queryid=..`): a record stream that carries its own schema, the `odps-tunnel-record-count` the SDK pages on, `rowrange` paging, `sizelimit` truncation and the `READ_TABLE_MAX_ROW` cap, with `InstanceTypeNotSupported` for statements that have no result set. This is the Java SDK's default interactive fetch and JDBC MaxQA's read path.
 
 ## 1.1.0 — 2026-09-17
 

README.md

@@ -88,7 +88,7 @@ session plane; `interactiveMode=maxqa` is refused by name.
 | --- | --- |
 | SQL | CREATE/DROP/TRUNCATE TABLE, INSERT INTO/OVERWRITE, SELECT/WITH; ODPS grammar validation with DuckDB execution |
 | Metadata | Table identity, schema, primary keys and properties; table listing; STRING partition create/delete/exists/list/pagination; persistent empty partitions |
-| MCQA session (SQLRT) | `Instance/Job/Tasks/SQLRT` creates a session instance that stays `Running` until the client stops it or it idles out; statements arrive as sub queries over the instance information KV (`?info&taskname=...`), with `status` / `progress` / `result_<id>` reads, `query` / `cancel` writes, and the object status codes the Java SDK polls on |
+| MCQA session (SQLRT) | `Instance/Job/Tasks/SQLRT` creates a session instance that stays `Running` until the client stops it or it idles out; statements arrive as sub queries over the instance information KV (`?info&taskname=...`), with `status` / `progress` / `result_<id>` reads, `query` / `cancel` writes, and the object status codes the Java SDK polls on. A sub query's result is also downloadable over the instance tunnel (`?data&cached&taskname=...&queryid=N`), which is the Java SDK's default fetch and JDBC MaxQA's read path |
 | Resources and functions | File-like resource upload (single payload or Java SDK part + merge), metadata read, download with `rOffset`/`rSize`, update, delete, prefix/paginated listing; TABLE resource metadata; Java/SQL/embedded function registration referencing existing resources |
 | Tunnel | Protobuf and Arrow batch upload/download; stream upload; Upsert/delete/partial updates; instance result download |
 | Storage API v2 | Arrow read/write, Batch/BatchCompatible/Streaming/StreamingRealtime sessions, commit/abort, projections and partition selection |
@@ -101,7 +101,7 @@ See [data transfer](docs/data-transfer.md) and [protocol details](docs/protocol.
 
 This is a local/CI test service. By default authentication signatures and permissions are **not validated**; optional strict mode uses local test credentials; keep it on a trusted test network. It is not a replacement for real MaxCompute acceptance tests.
 
-MCQA sessions cover the Java SDK's information-channel path (`SQLExecutor` with `useInstanceTunnel(false)`): sub-query answers are CSV whose first line is the column-name header, and session-scoped state (variables, temp objects) is not isolated — statements run against the shared engine. Not simulated: named-session attach (a session name is metadata; equal names create separate sessions), MaxQA v2 (`/mcqa` request prefix), instance-tunnel download of sub-query results, and per-statement statistics.
+MCQA sessions cover both of the Java SDK's read paths: the information channel (`SQLExecutor` with `useInstanceTunnel(false)`) answers with CSV whose first line is the column-name header, and the instance tunnel (`?data&cached&taskname=..&queryid=..`, the default fetch and the one JDBC MaxQA uses) answers with a record stream that carries its own schema, so a sub-query result arrives with the same rows and the same types either way. `instance_tunnel_limit_enabled` applies the service's `READ_TABLE_MAX_ROW` (10,000 rows) to that read, `rowrange` pages through it, and `sizelimit` shortens a response instead of failing it. Statements that produce no result set are refused on the tunnel with `InstanceTypeNotSupported`, like the service, so a client confirms them through the session API rather than reading an empty stream. A session is **select-only by default**: a `CREATE`/`INSERT`/`DROP` submitted as a sub query is declined before anything runs (`queryId: -1` plus `ODPS-1850001 Non select query not supported.`), because the Java client reacts to a non-select it learns about at fetch time by running the same statement offline again — and an INSERT that already ran in the session would land twice. Send `odps.sql.session.select.only=false`, in the session settings or per statement, to run them in the session anyway. Session-scoped state (variables, temp objects) is not isolated — statements run against the shared engine. Not simulated: named-session attach (a session name is metadata; equal names create separate sessions), MaxQA v2 (`/mcqa` request prefix), compressed or Arrow-encoded sub-query downloads (Protobuf records only), and per-statement statistics.
 Unsupported: Storage v1, Volume/Blob, CDC/incremental reads, filter predicate pushdown, explicit Schema management, column schema evolution, UDF **execution** (functions are registered metadata; SQL `CREATE FUNCTION`/`DROP FUNCTION` and calling a UDF return `UnsupportedFeature`), volume-backed resources, distributed scheduling, and full ODPS SQL semantics.
 
 Resource payloads are capped at 64 MiB each and 512 MiB per project/schema. Resource and function names resolve case-insensitively while the uploaded spelling is what listings return. Unsupported operations return errors rather than cloud behavior being assumed.

docs/data-transfer.md

@@ -11,6 +11,7 @@
 | StreamUploadSession | Protobuf pack | flush 即可见、动态/静态分区、retry trace-id 去重 |
 | UpsertSession | Protobuf U/D | PRIMARY KEY、按键替换/删除、部分列更新、提交/中止、重试去重 |
 | InstanceTunnel | Protobuf；HTTP Arrow 格式复用 | SQL 结果快照；Java InstanceTunnel 的 RecordReader 验收 |
+| InstanceTunnel MCQA 直读 | Protobuf（schema 随流） | 子查询按 queryid 读取、`rowrange` 分页、`sizelimit` 裁剪、`instance_tunnel_limit_enabled` 应用 10000 行上限、非 select 返回 InstanceTypeNotSupported；不做压缩与 Arrow |
 | Storage 表读取 | Arrow IPC | read session/get、Index 或 Offset/Count、MaxBatchRows、投影、分区列表、preview |
 | Storage 实例读取 | Arrow IPC | create/get session、Offset/Count 结果读取 |
 | Storage Batch | Arrow IPC | stream 创建/恢复/version、flush 暂存、close、选择 streams commit、abort、静态分区 overwrite |

docs/protocol.md

@@ -33,6 +33,10 @@ JDBC 驱动对每条语句都会先签一个 logview token，再为实例建立
 
 ## 编码
 
+MCQA 子查询直读没有 download session 可查，schema 只能随流走：帧头是 field 1 的 JSON schema（TunnelTableSchema 的 `columns`/`partitionKeys` 形状），后接 SCHEMA_END_TAG 33553920 与该帧自己的 CRC32C（覆盖 field number 的 4 字节小端值与 JSON 字节，不含 tag 与长度 varint），校验和后 CRC 归零，其后的记录部分与下面的 Protobuf 完全一致。响应头 `odps-tunnel-record-count` 是从本次 `rowrange` 起点起还能读到的行数，不是全量结果：Java 的 SessionRecordSetIterator 只取第一次响应的这个数并据此翻页，报大全量会让客户端越过结果末尾继续读取。查询仍在执行时该读取阻塞等待（长轮询），超时返回 Java SDK 认得的 `OdpsTaskTimeout`/“Wait for cache data timeout”，而不是先给一个空流。
+
+会话默认 select-only：`?info` 写入 `Key=query` 时，非 select 语句（含多语句提交里任何一段不是 select）在**执行前**就被拒绝，回 `{"queryId":-1,"status":"ok","result":"ODPS-1850001 Non select query not supported."}`——客户端对"取结果时才发现不是 select"的处置是 offline 重跑同一条语句，会话若已执行过，写语句就会落两次。`odps.sql.session.select.only=false`（会话设置或单条设置）可放开。
+
 Protobuf 按 public SDK 的列序编码，NULL 不写字段。记录结束 tag 33553408 后写 CRC32C，流尾 meta-count 33554430 与 checksum-of-checksums 33554431；CRC 数值按小端逻辑值更新，不包含传输长度和 NULL 标记。DATE 为天数，DATETIME 为毫秒，TIMESTAMP 为秒+纳秒，DECIMAL 为十进制文本。MAP 在引擎边界从 DuckDB OrderedMap 转换，ARRAY/STRUCT 递归编码。
 
 Tunnel Arrow 不是普通带 Schema 消息的 IPC stream：客户端从会话元数据构造 schema，服务端只发送 RecordBatch payload。帧头为大端 chunk size（65536），每满块追加 CRC32C，最后追加全流 CRC32C。ZSTD 和 LZ4 frame 压缩 Arrow IPC buffer，外层 Tunnel CRC 帧保持未压缩；deflate 仍包裹整个 Tunnel 帧。空表和 count=0 返回合法空数据帧。常规 Arrow 请求每批最多4096行；带 raw_size 的请求最多一批，CPP BufferArrowRecordReader 按实际行数推进；raw_size 过小时仍返回至少一行，以保证进度。
@@ -58,6 +62,7 @@ Tunnel Arrow 不是普通带 Schema 消息的 IPC stream：客户端从会话元
 | POST/GET/PUT `/projects/p/tables/t/streams` | 创建/恢复/写 pack；flush 后可见；dynamic_partition 支持逐 pack 分区 |
 | POST/GET/PUT/POST/DELETE `/projects/p/tables/t/upserts` | 创建/恢复/暂存 U/D/提交/中止；支持部分列更新 |
 | POST/GET `/projects/p/instances/i?downloads` / `?downloadid=id` | SQL 实例结果快照与 Tunnel 下载，复用表读取格式 |
+| GET `/projects/p/instances/i?data&cached&taskname=t[&queryid=n][&rowrange=(s,c)][&sizelimit=n][&instance_tunnel_limit_enabled]` | MCQA 子查询结果直读（Java SDK 默认交互取数、JDBC MaxQA 路径）：自带 schema 的 Protobuf 记录流，只支持未压缩 |
 | POST `/api/storage/v2` 或 `/api/storage/v3` | Action + Target 协议；Java 0.61.2-public 实际调用 v3 |
 
 Storage 表 Action：TableCreateReadSession、TableGetReadSession、TableRead、TablePreview、TableCreateWriteSession、TableGetWriteSession、TableCreateWriteStream、TableGetWriteStream、TableWrite、TableCloseWriteStream、TableCommitWriteSession、TableAbortWriteSession。实例 Action：InstanceCreateReadSession、InstanceGetReadSession（GET）、InstanceRead。

internal/server/mcqa.go

@@ -59,12 +59,22 @@ type mcqaQuery struct {
 	Warnings string
 	ID       int
 	Created  time.Time
+	// Data is the typed result behind the CSV text of the information channel. The
+	// instance-tunnel read serves it as records, so a consumer gets the same values and
+	// the same types through either path. It is retained for the same window as the CSV
+	// text — the mcqaRetainedQueries entries dropped by mcqaSession.drop, and counted by
+	// bytes into the instance budget.
+	Data engine.Result
 }
 
 // mcqaSession is the state behind a SQLRT instance. Fields are guarded by Server.mu;
 // statement execution runs outside that lock (see runSubQuery).
 type mcqaSession struct {
-	name       string
+	name string
+	// selectOnly is the session's promise that nothing but a whole select runs here.
+	// The service defaults it to true; a client that wants DDL or DML inside the session
+	// says so with odps.sql.session.select.only.
+	selectOnly bool
 	live       bool
 	nextID     int
 	lastID     int
@@ -73,8 +83,8 @@ type mcqaSession struct {
 	lastActive time.Time
 }
 
-func newMCQASession(name string) *mcqaSession {
-	return &mcqaSession{name: name, live: true, queries: map[int]*mcqaQuery{}, lastActive: time.Now()}
+func newMCQASession(name string, selectOnly bool) *mcqaSession {
+	return &mcqaSession{name: name, selectOnly: selectOnly, live: true, queries: map[int]*mcqaQuery{}, lastActive: time.Now()}
 }
 
 // expired reports whether the session has been idle past the configured TTL. Idle, not
@@ -115,7 +125,7 @@ func (m *mcqaSession) bytes() int64 {
 	}
 	var total int64
 	for _, q := range m.queries {
-		total += int64(len(q.Result) + len(q.Query))
+		total += q.Data.Bytes + int64(len(q.Result)+len(q.Query))
 	}
 	return total
 }
@@ -168,7 +178,7 @@ func (s *Server) mcqaCreate(w http.ResponseWriter, r *http.Request, p, sc, taskN
 	i := &instance{
 		Project: p, Schema: sc, ID: id(), Name: taskName, TaskType: "SQLRT",
 		Status: "Running", Created: time.Now(),
-		MCQA: newMCQASession(name),
+		MCQA: newMCQASession(name, !sessionRunsNonSelect(mcqaSettingsMap(settings))),
 	}
 	if !s.reserveInstance(i) {
 		fail(w, r, 429, "ResourceLimit", fmt.Errorf("instance limit"))
@@ -294,6 +304,18 @@ func (s *Server) mcqaSetInfo(w http.ResponseWriter, r *http.Request, i *instance
 			s.informationReply(w, mcqaInfoFailed, "Invalid sub query payload")
 			return
 		}
+		if m.selectOnly && !sessionRunsNonSelect(sub.Settings) && !isSelectStatement(sub.Query) {
+			// Refuse before running anything. A client that only learns the statement
+			// was not a select when it comes to fetch the result reruns it offline —
+			// and an INSERT that already ran in the session then applies twice. queryId
+			// -1 with status ok is how the service reports a sub query it declined to
+			// start, and the message is the pair FallbackPolicy matches on ("ODPS-185"),
+			// which is what turns the refusal into an offline rerun rather than an error.
+			s.mu.Unlock()
+			payload, _ := json.Marshal(map[string]any{"queryId": -1, "status": mcqaInfoOK, "result": mcqaSelectOnlyRefusal})
+			s.informationReply(w, mcqaInfoOK, string(payload))
+			return
+		}
 		m.nextID++
 		id := m.nextID
 		m.drop()
@@ -360,13 +382,16 @@ func (s *Server) runSubQuery(ctx context.Context, i *instance, id int, query str
 		q.Result = "Sub query cancelled"
 	default:
 		q.Status = mcqaStatusTerminated
+		q.Data = res
 		q.Result = sessionResultCSV(res)
 	}
 }
 
 // sessionResultCSV renders a result the way the SQLRT information channel does: the
 // first line is the column-name header, because CSVRecordParser#parse reads that line
-// to build the schema. Values are formatted like the offline instance result.
+// to build the schema. Values are formatted like the offline instance result, and every
+// row of the result is rendered — the tunnel read of the same sub query serves the same
+// rows, so neither path can look truncated relative to the other.
 func sessionResultCSV(res engine.Result) string {
 	header := make([]string, len(res.Columns))
 	for n, c := range res.Columns {