fix: include column headers in offline SQL results

.github/workflows/ci.yml

@@ -26,6 +26,8 @@ jobs:
           docker run --rm emulator-build sh -c 'go test -race ./... && go vet -unreachable=false ./...'
       - run: docker build --platform linux/amd64 -t maxcompute-emulator:ci .
       - run: mvn -B -f tests/java/pom.xml -Demulator.image=maxcompute-emulator:ci test
+      - name: Legacy Java SDK offline SQL result regression
+        run: mvn -B -f tests/java/pom.xml -Plegacy-sql-sdk -Demulator.image=maxcompute-emulator:ci clean test
       - name: Official JDBC driver acceptance
         run: mvn -B -f tests/jdbc/pom.xml -Demulator.image=maxcompute-emulator:ci test
       - name: Set up Python for the PyODPS consumer probe

README.md

@@ -101,6 +101,8 @@ 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.
 
+Offline SQL instance results (`?result`) also include the CSV column-name header expected by `SQLTask.getResult`. Empty SELECTs retain that header, while statements with no result schema return an empty result. The legacy Java SDK regression can be run with `mvn -B -f tests/java/pom.xml -Plegacy-sql-sdk -Demulator.image=maxcompute-emulator:dev clean test`.
+
 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.
 

internal/server/mcqa_test.go

@@ -2,11 +2,15 @@ package server
 
 import (
 	"bytes"
+	"context"
+	"encoding/csv"
 	"encoding/json"
+	"encoding/xml"
 	"fmt"
 	"io"
 	"net/http"
 	"net/http/httptest"
+	"reflect"
 	"strings"
 	"testing"
 	"time"
@@ -334,3 +338,50 @@ func TestMCQASessionIdleExpiry(t *testing.T) {
 		t.Fatalf("idle session must expire: code=%d body=%s", code, b)
 	}
 }
+
+// Offline SQL results must include the CSV schema line consumed by SQLTask.getResult.
+func TestOfflineSQLResultCSV(t *testing.T) {
+	s, h := fixture(t, Config{})
+	if _, err := s.Engine.Execute(context.Background(), "p", "default", "insert overwrite table t values (1,'one'),(2,'two'),(3,'three')"); err != nil {
+		t.Fatal(err)
+	}
+	for _, tc := range []struct {
+		name, query string
+		want        [][]string
+	}{
+		{"three rows", "select id, s from t order by id", [][]string{{"id", "s"}, {"1", "one"}, {"2", "two"}, {"3", "three"}}},
+		{"empty select", "select id, s from t where id = 99", [][]string{{"id", "s"}}},
+		{"escaped names and values", "select 'value,one' as `a,b`, 'value\"two' as `a\"b`, 'line\nvalue' as `line\nname`, cast(null as string) as missing, '' as blank, '<&>' as xml", [][]string{{"a,b", `a"b`, "line\nname", "missing", "blank", "xml"}, {"value,one", `value"two`, "line\nvalue", `\N`, "", "<&>"}}},
+		{"non query", "create table csv_empty(id bigint)", nil},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			body := `<Instance><Job><Tasks><SQL><Name>csv_task</Name><Query>` + esc(tc.query) + `</Query></SQL></Tasks></Job></Instance>`
+			code, header, b := mcqaCall(t, h, "POST", "/projects/p/instances", []byte(body))
+			if code != 201 {
+				t.Fatalf("create: %d %s", code, b)
+			}
+			code, _, b = mcqaCall(t, h, "GET", header.Get("Location")+"?result", nil)
+			if code != 200 {
+				t.Fatalf("result: %d %s", code, b)
+			}
+			var result struct {
+				Tasks struct {
+					Task struct{ Status, Result string }
+				}
+			}
+			if err := xml.Unmarshal(b, &result); err != nil {
+				t.Fatal(err)
+			}
+			if result.Tasks.Task.Status != "Success" {
+				t.Fatalf("failed SQL: %s", b)
+			}
+			got, err := csv.NewReader(strings.NewReader(result.Tasks.Task.Result)).ReadAll()
+			if err != nil {
+				t.Fatal(err)
+			}
+			if !reflect.DeepEqual(got, tc.want) {
+				t.Fatalf("CSV = %#v, want %#v", got, tc.want)
+			}
+		})
+	}
+}

internal/server/server.go

@@ -395,6 +395,14 @@ func (s *Server) instance(w http.ResponseWriter, r *http.Request, p, sc string,
 		} else {
 			var b bytes.Buffer
 			cw := csv.NewWriter(&b)
+			// SQLTask.getResult consumes the first CSV record as column names.
+			if len(res.Columns) > 0 {
+				header := make([]string, len(res.Columns))
+				for n, c := range res.Columns {
+					header[n] = c.Name
+				}
+				cw.Write(header)
+			}
 			for _, row := range res.Rows {
 				line := make([]string, len(row))
 				for n, v := range row {

tests/java/pom.xml

@@ -9,12 +9,12 @@
     <maven.compiler.target>17</maven.compiler.target>
     <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
     <!-- Docker 29 rejects docker-java's default API 1.32. -->
+    <odps.core.version>0.61.2-public</odps.core.version>
     <docker.api.version>1.44</docker.api.version>
   </properties>
   <dependencies>
-    <dependency><groupId>com.aliyun.odps</groupId><artifactId>odps-sdk-storage-api</artifactId><version>0.61.2-public</version></dependency>
     <dependency>
-      <groupId>com.aliyun.odps</groupId><artifactId>odps-sdk-core</artifactId><version>0.61.2-public</version>
+      <groupId>com.aliyun.odps</groupId><artifactId>odps-sdk-core</artifactId><version>${odps.core.version}</version>
     </dependency>
     <dependency>
       <groupId>org.junit.jupiter</groupId><artifactId>junit-jupiter</artifactId><version>5.10.2</version><scope>test</scope>
@@ -37,4 +37,28 @@
       </plugin>
     </plugins>
   </build>
+  <profiles>
+    <profile>
+      <id>storage-sdk</id>
+      <activation><activeByDefault>true</activeByDefault></activation>
+      <dependencies>
+        <dependency><groupId>com.aliyun.odps</groupId><artifactId>odps-sdk-storage-api</artifactId><version>0.61.2-public</version></dependency>
+      </dependencies>
+    </profile>
+    <!-- Compile only the offline API regression against SeaTunnel's legacy SDK. -->
+    <profile>
+      <id>legacy-sql-sdk</id>
+      <properties><odps.core.version>0.51.2-public</odps.core.version></properties>
+      <build><plugins>
+        <plugin>
+          <groupId>org.apache.maven.plugins</groupId><artifactId>maven-compiler-plugin</artifactId><version>3.13.0</version>
+          <configuration><testIncludes><testInclude>**/SQLResultTest.java</testInclude></testIncludes></configuration>
+        </plugin>
+        <plugin>
+          <groupId>org.apache.maven.plugins</groupId><artifactId>maven-surefire-plugin</artifactId>
+          <configuration><includes><include>**/SQLResultTest.java</include></includes></configuration>
+        </plugin>
+      </plugins></build>
+    </profile>
+  </profiles>
 </project>

tests/java/src/test/java/SQLResultTest.java

@@ -0,0 +1,112 @@
+import static org.junit.jupiter.api.Assertions.*;
+
+import com.aliyun.odps.Instance;
+import com.aliyun.odps.Odps;
+import com.aliyun.odps.account.AliyunAccount;
+import com.aliyun.odps.data.Record;
+import com.aliyun.odps.task.SQLTask;
+import com.aliyun.odps.tunnel.TableTunnel;
+import com.aliyun.odps.utils.CSVRecordParser;
+import java.util.List;
+import java.util.UUID;
+import org.junit.jupiter.api.*;
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.containers.wait.strategy.Wait;
+
+/** Covers the offline CSV API used by SeaTunnel's MaxCompute E2E assertions. */
+public class SQLResultTest {
+  static GenericContainer<?> container;
+  static String endpoint;
+  Odps odps;
+
+  @BeforeAll
+  static void start() {
+    endpoint = System.getProperty("emulator.endpoint");
+    if (endpoint == null) {
+      container = new GenericContainer<>(
+          System.getProperty("emulator.image", "maxcompute/maxcompute-emulator:1.1.0"))
+          .withExposedPorts(8080).waitingFor(Wait.forHttp("/readyz"));
+      container.start();
+      endpoint = "http://" + container.getHost() + ":" + container.getMappedPort(8080);
+    }
+  }
+
+  @AfterAll
+  static void stop() {
+    if (container != null) container.stop();
+  }
+
+  @BeforeEach
+  void client() {
+    odps = new Odps(new AliyunAccount("ak", "sk"));
+    odps.setDefaultProject("test_project");
+    odps.setEndpoint(endpoint);
+    odps.setTunnelEndpoint(endpoint);
+  }
+
+  Instance sql(String statement) throws Exception {
+    Instance instance = SQLTask.run(odps, statement);
+    instance.waitForSuccess();
+    return instance;
+  }
+
+  @Test
+  void bufferedBatchUploadRetainsAllThreeRowsInSQLResult() throws Exception {
+    String table = "csv_" + UUID.randomUUID().toString().replace("-", "");
+    try {
+      sql("create table " + table + " (id INT, name STRING, age INT)");
+      TableTunnel tunnel = new TableTunnel(odps);
+      tunnel.setEndpoint(endpoint);
+      TableTunnel.UploadSession upload = tunnel.createUploadSession("test_project", table);
+      try (var writer = upload.openBufferedWriter()) {
+        for (int i = 1; i <= 3; i++) {
+          Record row = upload.newRecord();
+          row.set(0, i);
+          row.setString(1, "INSERT_TEST" + i);
+          row.set(2, 10 + i * 10);
+          writer.write(row);
+        }
+      }
+      upload.commit();
+      assertEquals(3, tunnel.createDownloadSession("test_project", table).getRecordCount());
+      List<Record> rows = SQLTask.getResult(sql("select * from " + table + " order by id"));
+      assertEquals(3, rows.size());
+      for (int i = 0; i < 3; i++) {
+        assertEquals(String.valueOf(i + 1), rows.get(i).getString(0));
+        assertEquals("INSERT_TEST" + (i + 1), rows.get(i).getString(1));
+        assertEquals(String.valueOf(20 + i * 10), rows.get(i).getString(2));
+      }
+      assertEquals("id", rows.get(0).getColumns()[0].getName());
+    } finally {
+      odps.tables().delete("test_project", table, true);
+    }
+  }
+
+  @Test
+  void emptySelectRetainsItsColumnSchema() throws Exception {
+    Instance instance = sql("select 1 as id, 'empty' as name where 1 = 0");
+    assertTrue(SQLTask.getResult(instance).isEmpty());
+    var parsed = CSVRecordParser.parse(instance.getTaskResults().values().iterator().next());
+    assertEquals("id", parsed.getSchema().getColumn(0).getName());
+    assertEquals("name", parsed.getSchema().getColumn(1).getName());
+    assertTrue(parsed.getRecords().isEmpty());
+  }
+
+  @Test
+  void quotedColumnNamesAndValuesRoundTripThroughXMLAndCSV() throws Exception {
+    Instance instance = sql("select 'value,one' as `a,b`, 'value\"two' as `a\"b`, "
+        + "'line\nvalue' as `line\nname`, cast(null as string) as missing, '' as blank, '<&>' as xml");
+    List<Record> rows = SQLTask.getResult(instance);
+    assertEquals(1, rows.size());
+    Record row = rows.get(0);
+    assertEquals("a,b", row.getColumns()[0].getName());
+    assertEquals("a\"b", row.getColumns()[1].getName());
+    assertEquals("line\nname", row.getColumns()[2].getName());
+    assertEquals("value,one", row.getString(0));
+    assertEquals("value\"two", row.getString(1));
+    assertEquals("line\nvalue", row.getString(2));
+    assertEquals("\\N", row.getString(3));
+    assertEquals("", row.getString(4));
+    assertEquals("<&>", row.getString(5));
+  }
+}