This is an automated email from the ASF dual-hosted git repository.
HTHou pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iotdb-client-go.git
The following commit(s) were added to refs/heads/main by this push:
new a25f68a fix(client): re-send session configuration on reconnect (#178)
a25f68a is described below
commit a25f68aa9681acba92851562bbc05d07c4d0ba91
Author: Takashi Honzawa <[email protected]>
AuthorDate: Mon Sep 14 03:44:54 2026 -0400
fix(client): re-send session configuration on reconnect (#178)
Session.initClusterConn() built its TSOpenSessionReq without a
Configuration map, while Open() and OpenCluster() populate
Configuration["sql_dialect"], ["version"], and ["db"]. Because
reconnect() routes exclusively through initClusterConn() and is the sole
transparent-reconnect path, any session that reconnects after a server
restart silently reverted to the server-default dialect with no database
bound. For table-model sessions this poisons the session: every
subsequent statement fails with a SqlParseError (700), and because that
is a valid server response rather than a transport error, TableSessionPool
keeps the session in the pool.
Extract the Configuration-building block into a single helper and call it
from all three open paths so the reconnect path sends what open sends and
the three cannot drift again. Add a table-driven test for the helper.
Verified against apache/iotdb:2.0.3-standalone: a table-model insert on a
pooled session that reconnects after `docker restart` fails on the current
code and succeeds with this change.
Fixes #177
Co-authored-by: Claude <[email protected]>
---
client/session.go | 42 ++++++++++++++++++++++--------------------
client/session_test.go | 37 ++++++++++++++++++++++++++++++++++++-
2 files changed, 58 insertions(+), 21 deletions(-)
diff --git a/client/session.go b/client/session.go
index 787bd93..42424a0 100644
--- a/client/session.go
+++ b/client/session.go
@@ -86,6 +86,25 @@ type endPoint struct {
Port string
}
+// buildOpenSessionConfiguration returns the Configuration map every
+// OpenSession request must carry so a (re)connected session speaks the
+// caller's dialect and stays bound to its database. Shared by Open,
+// OpenCluster, and initClusterConn (the reconnect path) so the three cannot
+// drift: omitting it on reconnect silently reverts the session to server
+// defaults.
+func buildOpenSessionConfiguration(cfg *Config) map[string]string {
+ c := map[string]string{"sql_dialect": cfg.sqlDialect}
+ if cfg.Version == "" {
+ c["version"] = string(DEFAULT_VERSION)
+ } else {
+ c["version"] = string(cfg.Version)
+ }
+ if cfg.Database != "" {
+ c["db"] = cfg.Database
+ }
+ return c
+}
+
func (s *Session) Open(enableRPCCompression bool, connectionTimeoutInMs int)
error {
if s.config.FetchSize <= 0 {
s.config.FetchSize = DefaultFetchSize
@@ -118,16 +137,7 @@ func (s *Session) Open(enableRPCCompression bool,
connectionTimeoutInMs int) err
ClientProtocol:
rpc.TSProtocolVersion_IOTDB_SERVICE_PROTOCOL_V3, ZoneId: s.config.TimeZone,
Username: s.config.UserName,
Password: &s.config.Password,
}
- req.Configuration = make(map[string]string)
- req.Configuration["sql_dialect"] = s.config.sqlDialect
- if s.config.Version == "" {
- req.Configuration["version"] = string(DEFAULT_VERSION)
- } else {
- req.Configuration["version"] = string(s.config.Version)
- }
- if s.config.Database != "" {
- req.Configuration["db"] = s.config.Database
- }
+ req.Configuration = buildOpenSessionConfiguration(s.config)
resp, err := s.client.OpenSession(context.Background(), &req)
if err != nil {
return err
@@ -176,16 +186,7 @@ func (s *Session) OpenCluster(enableRPCCompression bool)
error {
ClientProtocol:
rpc.TSProtocolVersion_IOTDB_SERVICE_PROTOCOL_V3, ZoneId: s.config.TimeZone,
Username: s.config.UserName,
Password: &s.config.Password,
}
- req.Configuration = make(map[string]string)
- req.Configuration["sql_dialect"] = s.config.sqlDialect
- if s.config.Version == "" {
- req.Configuration["version"] = string(DEFAULT_VERSION)
- } else {
- req.Configuration["version"] = string(s.config.Version)
- }
- if s.config.Database != "" {
- req.Configuration["db"] = s.config.Database
- }
+ req.Configuration = buildOpenSessionConfiguration(s.config)
resp, err := s.client.OpenSession(context.Background(), &req)
if err != nil {
@@ -1425,6 +1426,7 @@ func (s *Session) initClusterConn(node endPoint) error {
ClientProtocol:
rpc.TSProtocolVersion_IOTDB_SERVICE_PROTOCOL_V3, ZoneId: s.config.TimeZone,
Username: s.config.UserName,
Password: &s.config.Password,
}
+ req.Configuration = buildOpenSessionConfiguration(s.config)
resp, err := s.client.OpenSession(context.Background(), &req)
if err != nil {
diff --git a/client/session_test.go b/client/session_test.go
index 67f83e9..af14dae 100644
--- a/client/session_test.go
+++ b/client/session_test.go
@@ -19,7 +19,10 @@
package client
-import "testing"
+import (
+ "reflect"
+ "testing"
+)
func TestParseNodeURL(t *testing.T) {
tests := []struct {
@@ -60,3 +63,35 @@ func TestParseNodeURL(t *testing.T) {
})
}
}
+
+func TestBuildOpenSessionConfiguration(t *testing.T) {
+ cases := []struct {
+ name string
+ cfg *Config
+ want map[string]string
+ }{
+ {
+ name: "table dialect with db",
+ cfg: &Config{sqlDialect: TableSqlDialect, Database:
"mydb"},
+ want: map[string]string{"sql_dialect": TableSqlDialect,
"version": string(DEFAULT_VERSION), "db": "mydb"},
+ },
+ {
+ name: "tree dialect, no db (db key omitted)",
+ cfg: &Config{sqlDialect: TreeSqlDialect},
+ want: map[string]string{"sql_dialect": TreeSqlDialect,
"version": string(DEFAULT_VERSION)},
+ },
+ {
+ name: "explicit version passes through",
+ cfg: &Config{sqlDialect: TableSqlDialect, Version:
V_1_0},
+ want: map[string]string{"sql_dialect": TableSqlDialect,
"version": string(V_1_0)},
+ },
+ }
+ for _, tc := range cases {
+ t.Run(tc.name, func(t *testing.T) {
+ got := buildOpenSessionConfiguration(tc.cfg)
+ if !reflect.DeepEqual(got, tc.want) {
+ t.Errorf("got %v, want %v", got, tc.want)
+ }
+ })
+ }
+}
