HTHou commented on code in PR #162: URL: https://github.com/apache/iotdb-client-go/pull/162#discussion_r3412187253
########## database/conn.go: ########## @@ -0,0 +1,171 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package iotdb_go + +import ( + "context" + "fmt" + "log" + "net" + "os" + "time" + + "github.com/apache/iotdb-client-go/v2/client" + "github.com/apache/iotdb-client-go/v2/database/column" + "github.com/pkg/errors" +) + +func dial(ctx context.Context, addr string, num int, opt *Options) (*connect, error) { + if addr == "" { + return nil, errors.New("empty addr") + } + // 使用 net.SplitHostPort 分割地址和端口 + host, port, err := net.SplitHostPort(addr) + if err != nil { + return nil, err + } + var ( + conn client.SessionPool + debugf = func(format string, v ...any) {} + ) + + if opt.Debug { + if opt.Debugf != nil { + debugf = func(format string, v ...any) { + opt.Debugf( + "[iotdb][%s][id=%d] "+format, + append([]interface{}{opt.Addr, num}, v...)..., + ) + } + } else { + debugf = log.New(os.Stdout, fmt.Sprintf("[iotdb][%s][id=%d]", opt.Addr, num), 0).Printf + } + } + var ( + config = &client.PoolConfig{ + Host: host, + Port: port, + UserName: opt.UserName, + Password: opt.Password, + } + poolMaxSize = 3 + poolWaitToGetSessionTimeoutInMs = 60000 + poolConnectionTimeoutInMs = 60000 + poolEnableCompression = false + ) + if opt.PoolMaxSize != nil { + poolMaxSize = *opt.PoolMaxSize + } + if opt.PoolWaitToGetSessionTimeoutInMs != nil { + poolWaitToGetSessionTimeoutInMs = *opt.PoolWaitToGetSessionTimeoutInMs + } + if opt.PoolConnectionTimeoutInMs != nil { + poolConnectionTimeoutInMs = *opt.PoolConnectionTimeoutInMs + } + if opt.PoolEnableCompression != nil { + poolEnableCompression = *opt.PoolEnableCompression + } + conn = client.NewSessionPool(config, poolMaxSize, poolConnectionTimeoutInMs, poolWaitToGetSessionTimeoutInMs, poolEnableCompression) + + var ( + netConn = &connect{ + id: num, + opt: opt, + conn: conn, + debugfFunc: debugf, + connectedAt: time.Now(), + } + ) + + return netConn, nil +} + +type connect struct { + id int + opt *Options + conn client.SessionPool + debugfFunc func(format string, v ...any) + connectedAt time.Time + timeZone *time.Location +} + +func (c *connect) debugf(format string, v ...any) { + c.debugfFunc(format, v...) +} + +func (c *connect) isBad() bool { + return false +} +func (c *connect) close() error { + c.conn.Close() + return nil +} + +func (c *connect) ping(ctx context.Context) (err error) { + session, err := c.conn.GetSession() + if err != nil { + return err + } + defer c.conn.PutBack(session) + return session.Ping(ctx) +} + +func (c *connect) query(ctx context.Context, release nativeTransportRelease, query string, args ...any) (*rows, error) { + options := queryOptions(ctx) + body, err := bindQueryOrAppendParameters(&options, query, c.timeZone, args...) + if err != nil { + return nil, err + } + session, err := c.conn.GetSession() + if err != nil { + release(c, err) + return nil, err + } + defer c.conn.PutBack(session) Review Comment: The session is returned to the pool before the returned rows are consumed. `SessionDataSet` keeps using this session’s RPC client/session id for later `FetchResultsV2` and `CloseOperation` calls from `Rows.Next`/`Rows.Close`, so another goroutine can borrow the same session and use the transport concurrently while the result set is still active. Please hold the session until rows are closed/EOF, or fully materialize the result before `PutBack`. ########## database/iotdb_std.go: ########## @@ -0,0 +1,230 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package iotdb_go + +import ( + "context" + "database/sql" + "database/sql/driver" + "errors" + "io" + "log" + "net" + "os" + "syscall" +) + +var globalConnID int64 + +func init() { + var debugf = func(format string, v ...any) {} + sql.Register("iotdb", &stdDriver{debugf: debugf}) +} + +// isConnBrokenError returns true if the error class indicates that the +// db connection is no longer usable and should be marked bad +func isConnBrokenError(err error) bool { + if errors.Is(err, io.EOF) || errors.Is(err, syscall.EPIPE) || errors.Is(err, syscall.ECONNRESET) { + return true + } + if _, ok := err.(*net.OpError); ok { + return true + } + return false +} + +type stdDriver struct { + opt *Options + conn stdConnect + commit func() error + debugf func(format string, v ...any) +} + +var _ driver.Conn = (*stdDriver)(nil) +var _ driver.ConnBeginTx = (*stdDriver)(nil) +var _ driver.ExecerContext = (*stdDriver)(nil) +var _ driver.QueryerContext = (*stdDriver)(nil) +var _ driver.ConnPrepareContext = (*stdDriver)(nil) + +func (std *stdDriver) Open(dsn string) (_ driver.Conn, err error) { + var opt Options + if err := opt.fromDSN(dsn); err != nil { + std.debugf("Open dsn error: %v\n", err) + return nil, err + } + var debugf = func(format string, v ...any) {} + if opt.Debug { + debugf = log.New(os.Stdout, "[iotdb-std][opener] ", 0).Printf + } + opt.ClientInfo.Comment = []string{"database/sql"} + return (&stdConnOpener{opt: &opt, debugf: debugf}).Connect(context.Background()) +} + +var _ driver.Driver = (*stdDriver)(nil) + +func (std *stdDriver) ResetSession(ctx context.Context) error { + if std.conn.isBad() { + std.debugf("Resetting session because connection is bad") + return driver.ErrBadConn + } + return nil +} + +var _ driver.SessionResetter = (*stdDriver)(nil) + +func (std *stdDriver) Ping(ctx context.Context) error { + if std.conn.isBad() { + std.debugf("Ping: connection is bad") + return driver.ErrBadConn + } + + return std.conn.ping(ctx) +} + +var _ driver.Pinger = (*stdDriver)(nil) + +// Begin starts and returns a new transaction. +// +// Deprecated: Drivers should implement ConnBeginTx instead (or additionally). +func (std *stdDriver) Begin() (driver.Tx, error) { + if std.conn.isBad() { + std.debugf("Begin: connection is bad") + return nil, driver.ErrBadConn + } + return std, nil +} + +func (std *stdDriver) BeginTx(ctx context.Context, opts driver.TxOptions) (driver.Tx, error) { Review Comment: This advertises transaction support, but no transaction is actually started. `Commit` is usually a no-op and `Rollback` only clears a callback, so operations executed through `*sql.Tx` are already applied and cannot be rolled back. If IoTDB does not support transactions here, `Begin`/`BeginTx` should return an explicit unsupported error instead of returning this connection as `driver.Tx`. ########## database/iotdb_std.go: ########## @@ -0,0 +1,230 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package iotdb_go + +import ( + "context" + "database/sql" + "database/sql/driver" + "errors" + "io" + "log" + "net" + "os" + "syscall" +) + +var globalConnID int64 + +func init() { + var debugf = func(format string, v ...any) {} + sql.Register("iotdb", &stdDriver{debugf: debugf}) +} + +// isConnBrokenError returns true if the error class indicates that the +// db connection is no longer usable and should be marked bad +func isConnBrokenError(err error) bool { + if errors.Is(err, io.EOF) || errors.Is(err, syscall.EPIPE) || errors.Is(err, syscall.ECONNRESET) { + return true + } + if _, ok := err.(*net.OpError); ok { + return true + } + return false +} + +type stdDriver struct { + opt *Options + conn stdConnect + commit func() error + debugf func(format string, v ...any) +} + +var _ driver.Conn = (*stdDriver)(nil) +var _ driver.ConnBeginTx = (*stdDriver)(nil) +var _ driver.ExecerContext = (*stdDriver)(nil) +var _ driver.QueryerContext = (*stdDriver)(nil) +var _ driver.ConnPrepareContext = (*stdDriver)(nil) + +func (std *stdDriver) Open(dsn string) (_ driver.Conn, err error) { + var opt Options + if err := opt.fromDSN(dsn); err != nil { + std.debugf("Open dsn error: %v\n", err) + return nil, err + } + var debugf = func(format string, v ...any) {} + if opt.Debug { + debugf = log.New(os.Stdout, "[iotdb-std][opener] ", 0).Printf + } + opt.ClientInfo.Comment = []string{"database/sql"} + return (&stdConnOpener{opt: &opt, debugf: debugf}).Connect(context.Background()) +} + +var _ driver.Driver = (*stdDriver)(nil) + +func (std *stdDriver) ResetSession(ctx context.Context) error { + if std.conn.isBad() { + std.debugf("Resetting session because connection is bad") + return driver.ErrBadConn + } + return nil +} + +var _ driver.SessionResetter = (*stdDriver)(nil) + +func (std *stdDriver) Ping(ctx context.Context) error { + if std.conn.isBad() { + std.debugf("Ping: connection is bad") + return driver.ErrBadConn + } + + return std.conn.ping(ctx) +} + +var _ driver.Pinger = (*stdDriver)(nil) + +// Begin starts and returns a new transaction. +// +// Deprecated: Drivers should implement ConnBeginTx instead (or additionally). +func (std *stdDriver) Begin() (driver.Tx, error) { + if std.conn.isBad() { + std.debugf("Begin: connection is bad") + return nil, driver.ErrBadConn + } + return std, nil +} + +func (std *stdDriver) BeginTx(ctx context.Context, opts driver.TxOptions) (driver.Tx, error) { + if std.conn.isBad() { + std.debugf("BeginTx: connection is bad") + return nil, driver.ErrBadConn + } + + return std, nil +} + +func (std *stdDriver) Commit() error { + if std.commit == nil { + return nil + } + defer func() { + std.commit = nil + }() + + if err := std.commit(); err != nil { + if isConnBrokenError(err) { + std.debugf("Commit got EOF error: resetting connection") + return driver.ErrBadConn + } + std.debugf("Commit error: %v\n", err) + return err + } + return nil +} + +func (std *stdDriver) Rollback() error { + std.commit = nil + //std.conn.close() + return nil +} + +var _ driver.Tx = (*stdDriver)(nil) + +func (std *stdDriver) CheckNamedValue(nv *driver.NamedValue) error { return nil } + +var _ driver.NamedValueChecker = (*stdDriver)(nil) + +// Prepare returns a prepared statement, bound to this connection. +func (std *stdDriver) Prepare(query string) (driver.Stmt, error) { + return std.PrepareContext(context.Background(), query) +} + +func (std *stdDriver) QueryContext(ctx context.Context, query string, args []driver.NamedValue) (driver.Rows, error) { + if std.conn.isBad() { + std.debugf("QueryContext: connection is bad") + return nil, driver.ErrBadConn + } + + r, err := std.conn.query(ctx, func(nativeTransport, error) {}, query, rebind(args)...) + if isConnBrokenError(err) { + std.debugf("QueryContext got a fatal error, resetting connection: %v\n", err) + return nil, driver.ErrBadConn + } + if err != nil { + std.debugf("QueryContext error: %v\n", err) + return nil, err + } + return &stdRows{ + rows: r, + debugf: std.debugf, + }, nil +} + +func (std *stdDriver) PrepareContext(ctx context.Context, query string) (driver.Stmt, error) { + if std.conn.isBad() { + std.debugf("PrepareContext: connection is bad") + return nil, driver.ErrBadConn + } + + std.commit = std.conn.commit Review Comment: Prepared statements are currently unusable: `PrepareContext` ignores the query and returns `stdBatch`, whose `ExecContext` returns `driver.ErrSkip`, `Exec` returns `not implemented`, and `Query` errors. Normal `db.Prepare(...).Exec/Query` calls will fail even though the driver advertises `ConnPrepareContext`. Please implement a statement that stores the query and delegates to the connection, or do not advertise prepare support. ########## database/bind.go: ########## @@ -0,0 +1,370 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package iotdb_go + +import ( + std_driver "database/sql/driver" + "fmt" + "reflect" + "regexp" + "strings" + "time" + + "github.com/apache/iotdb-client-go/v2/database/driver" + "github.com/pkg/errors" +) + +var ( + ErrInvalidTimezone = errors.New("invalid timezone value") +) + +func Named(name string, value any) driver.NamedValue { + return driver.NamedValue{ + Name: name, + Value: value, + } +} + +type TimeUnit uint8 + +const ( + Seconds TimeUnit = iota + MilliSeconds + MicroSeconds + NanoSeconds +) + +type GroupSet struct { + Value []any +} + +type ArraySet []any + +func DateNamed(name string, value time.Time, scale TimeUnit) driver.NamedDateValue { + return driver.NamedDateValue{ + Name: name, + Value: value, + Scale: uint8(scale), + } +} + +var ( + bindNumericRe = regexp.MustCompile(`\$[0-9]+`) + bindPositionalRe = regexp.MustCompile(`[^\\][?]`) +) + +func bind(tz *time.Location, query string, args ...any) (string, error) { + if len(args) == 0 { + return query, nil + } + var ( + haveNumeric bool + havePositional bool + ) + + allArgumentsNamed, err := checkAllNamedArguments(args...) + if err != nil { + return "", err + } + + if allArgumentsNamed { + return bindNamed(tz, query, args...) + } + + haveNumeric = bindNumericRe.MatchString(query) + havePositional = bindPositionalRe.MatchString(query) + if haveNumeric && havePositional { + return "", ErrBindMixedParamsFormats + } + if haveNumeric { + return bindNumeric(tz, query, args...) + } + return bindPositional(tz, query, args...) +} + +func checkAllNamedArguments(args ...any) (bool, error) { + var ( + haveNamed bool + haveAnonymous bool + ) + for _, v := range args { + switch v.(type) { + case driver.NamedValue, driver.NamedDateValue: + haveNamed = true + default: + haveAnonymous = true + } + if haveNamed && haveAnonymous { + return haveNamed, ErrBindMixedParamsFormats + } + } + return haveNamed, nil +} + +func bindPositional(tz *time.Location, query string, args ...any) (_ string, err error) { + var ( + lastMatchIndex = -1 // Position of previous match for copying + argIndex = 0 // Index for the argument at current position + buf = make([]byte, 0, len(query)) + unbindCount = 0 // Number of positional arguments that couldn't be matched + ) + + for i := 0; i < len(query); i++ { + // It's fine looping through the query string as bytes, because the (fixed) characters we're looking for + // are in the ASCII range to won't take up more than one byte. + if query[i] == '?' { Review Comment: This binding code treats every `?` byte as a placeholder, including `?` inside quoted strings or comments. The numeric and named bind paths have the same class of issue because they regex-replace globally. Statements like `SELECT '?'` or `WHERE text = '@name'` will be rewritten incorrectly. Binding needs to skip SQL string literals/comments, preferably via a small lexer rather than raw byte/regex replacement. ########## database/rows_std.go: ########## @@ -0,0 +1,89 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package iotdb_go + +import ( + "database/sql/driver" + "io" + + "github.com/pkg/errors" +) + +type stdRows struct { + rows *rows + debugf func(format string, v ...any) +} + +// Columns returns the names of the columns. The number of +// columns of the result is inferred from the length of the +// slice. If a particular column name isn't known, an empty +// string should be returned for that entry. +func (s *stdRows) Columns() []string { + return s.rows.set.GetColumnNames() +} + +// Close closes the rows iterator. +func (s *stdRows) Close() error { + s.rows.set.Close() + return nil +} + +// Next is called to populate the next row of data into +// the provided slice. The provided slice will be the same +// size as the Columns() are wide. +// +// Next should return io.EOF when there are no more rows. +// +// The dest should not be written to outside of Next. Care +// should be taken when closing Rows not to modify +// a buffer held in dest. +func (s *stdRows) Next(dest []driver.Value) error { + if len(s.rows.set.GetColumnNames()) != len(dest) { + return errors.New("column count mismatch") + } + next, err := s.rows.Next() + if err != nil { + s.debugf("rows.Next() failed: %v", err) + return err + } + if next { + for i := range dest { + if s.rows.columns[i] == nil { + dest[i] = nil + continue + } + switch value := s.rows.columns[i].Row(s.rows.set, false).(type) { Review Comment: This loses SQL NULLs. The column getters return typed zero values when the underlying `SessionDataSet` value is NULL, and `Next` assigns those zero values into `dest`. `database/sql` expects `dest[i] = nil` for NULL so `sql.Null*`, pointer scans, and `*any` behave correctly. Please check nullness before reading the value or have the column layer return nil for NULL. ########## database/conn_exec.go: ########## @@ -0,0 +1,46 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package iotdb_go + +import ( + "context" +) + +func (c *connect) exec(ctx context.Context, query string, args ...any) error { + var ( + options = queryOptions(ctx) + body, err = bindQueryOrAppendParameters(&options, query, c.timeZone, args...) + ) + if err != nil { + return err + } + + session, err := c.conn.GetSession() + if err != nil { + return err + } + defer c.conn.PutBack(session) + + _, err = session.ExecuteStatement(body) Review Comment: `ExecContext` receives a caller context, but this call goes into `Session.ExecuteStatement`, which uses `context.Background()` internally for the RPC. That means cancellation and deadlines from `database/sql` are ignored after this point. `QueryContext` has the same issue through `ExecuteQueryStatement`; the context needs to be threaded down to the actual RPC calls. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
