From ff7bd72772cd8c580487e5fee139f8c461896e02 Mon Sep 17 00:00:00 2001 From: dingxin Date: Mon, 5 Oct 2026 00:04:22 +0800 Subject: [PATCH] fix: include column headers in offline SQL results --- .github/workflows/ci.yml | 2 + README.md | 2 + internal/server/mcqa_test.go | 51 +++++++++ internal/server/server.go | 8 ++ tests/java/pom.xml | 28 ++++- tests/java/src/test/java/SQLResultTest.java | 112 ++++++++++++++++++++ 6 files changed, 201 insertions(+), 2 deletions(-) create mode 100644 tests/java/src/test/java/SQLResultTest.java diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 3d1b9af..2534922 100644 --- a/.github/workflows/ci.yml +++ b/.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 diff --git a/README.md b/README.md index 767a6c3..5535316 100644 --- a/README.md +++ b/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. diff --git a/internal/server/mcqa_test.go b/internal/server/mcqa_test.go index cae9436..037eee6 100644 --- a/internal/server/mcqa_test.go +++ b/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 := `csv_task` + esc(tc.query) + `` + 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) + } + }) + } +} diff --git a/internal/server/server.go b/internal/server/server.go index 52ce2b3..43afb5b 100644 --- a/internal/server/server.go +++ b/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 { diff --git a/tests/java/pom.xml b/tests/java/pom.xml index 8e550a2..b2f454e 100644 --- a/tests/java/pom.xml +++ b/tests/java/pom.xml @@ -9,12 +9,12 @@ 17 UTF-8 + 0.61.2-public 1.44 - com.aliyun.odpsodps-sdk-storage-api0.61.2-public - com.aliyun.odpsodps-sdk-core0.61.2-public + com.aliyun.odpsodps-sdk-core${odps.core.version} org.junit.jupiterjunit-jupiter5.10.2test @@ -37,4 +37,28 @@ + + + storage-sdk + true + + com.aliyun.odpsodps-sdk-storage-api0.61.2-public + + + + + legacy-sql-sdk + 0.51.2-public + + + org.apache.maven.pluginsmaven-compiler-plugin3.13.0 + **/SQLResultTest.java + + + org.apache.maven.pluginsmaven-surefire-plugin + **/SQLResultTest.java + + + + diff --git a/tests/java/src/test/java/SQLResultTest.java b/tests/java/src/test/java/SQLResultTest.java new file mode 100644 index 0000000..a435f98 --- /dev/null +++ b/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 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 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)); + } +}