You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by sh...@apache.org on 2026/03/26 06:28:48 UTC
(iotdb-client-go) 02/03: Call VerifySuccess before return to user (#151)
This is an automated email from the ASF dual-hosted git repository.
shuwenwei pushed a commit to branch fixBugs-0326
in repository https://gitbox.apache.org/repos/asf/iotdb-client-go.git
commit 35e9dc4496c3ed6c5e849a2ed1af1025bef4c53a
Author: Haonan <hh...@outlook.com>
AuthorDate: Thu Feb 12 18:03:50 2026 +0800
Call VerifySuccess before return to user (#151)
* Move VerifySuccess
* fix missing code
* fix missing code
* fix copilot review
---
client/errors.go | 29 +--
client/session.go | 252 ++++++++++++++++++---------
client/sessionpool.go | 16 --
client/utils.go | 8 +-
example/session_example.go | 16 +-
example/session_pool/session_pool_example.go | 17 +-
test/e2e/e2e_test.go | 29 ++-
7 files changed, 208 insertions(+), 159 deletions(-)
diff --git a/client/errors.go b/client/errors.go
index 2c54bde..0f1aabb 100644
--- a/client/errors.go
+++ b/client/errors.go
@@ -20,28 +20,33 @@
package client
import (
- "bytes"
+ "fmt"
"github.com/apache/iotdb-client-go/common"
)
+// ExecutionError represents an error returned by the server via TSStatus.
+// It is NOT a connection error and should not cause session drops.
+type ExecutionError struct {
+ Code int32
+ Message string
+}
+
+func (e *ExecutionError) Error() string {
+ if e.Message != "" {
+ return fmt.Sprintf("error code: %d, message: %v", e.Code, e.Message)
+ }
+ return fmt.Sprintf("error code: %d", e.Code)
+}
+
type BatchError struct {
statuses []*common.TSStatus
+ Message string
}
func (e *BatchError) Error() string {
- buff := bytes.Buffer{}
- for _, status := range e.statuses {
- buff.WriteString(*status.Message + ";")
- }
- return buff.String()
+ return e.Message
}
func (e *BatchError) GetStatuses() []*common.TSStatus {
return e.statuses
}
-
-func NewBatchError(statuses []*common.TSStatus) *BatchError {
- return &BatchError{
- statuses: statuses,
- }
-}
diff --git a/client/session.go b/client/session.go
index fff4cf0..98a9fee 100644
--- a/client/session.go
+++ b/client/session.go
@@ -186,14 +186,17 @@ func (s *Session) Close() error {
*return
*error: correctness of operation
*/
-func (s *Session) SetStorageGroup(storageGroupId string) (r *common.TSStatus, err error) {
- r, err = s.client.SetStorageGroup(context.Background(), s.sessionId, storageGroupId)
+func (s *Session) SetStorageGroup(storageGroupId string) error {
+ r, err := s.client.SetStorageGroup(context.Background(), s.sessionId, storageGroupId)
if err != nil && r == nil {
if s.reconnect() {
r, err = s.client.SetStorageGroup(context.Background(), s.sessionId, storageGroupId)
}
}
- return r, err
+ if err != nil {
+ return err
+ }
+ return VerifySuccess(r)
}
/*
@@ -203,14 +206,17 @@ func (s *Session) SetStorageGroup(storageGroupId string) (r *common.TSStatus, er
*return
*error: correctness of operation
*/
-func (s *Session) DeleteStorageGroup(storageGroupId string) (r *common.TSStatus, err error) {
- r, err = s.client.DeleteStorageGroups(context.Background(), s.sessionId, []string{storageGroupId})
+func (s *Session) DeleteStorageGroup(storageGroupId string) error {
+ r, err := s.client.DeleteStorageGroups(context.Background(), s.sessionId, []string{storageGroupId})
if err != nil && r == nil {
if s.reconnect() {
r, err = s.client.DeleteStorageGroups(context.Background(), s.sessionId, []string{storageGroupId})
}
}
- return r, err
+ if err != nil {
+ return err
+ }
+ return VerifySuccess(r)
}
/*
@@ -220,14 +226,17 @@ func (s *Session) DeleteStorageGroup(storageGroupId string) (r *common.TSStatus,
*return
*error: correctness of operation
*/
-func (s *Session) DeleteStorageGroups(storageGroupIds ...string) (r *common.TSStatus, err error) {
- r, err = s.client.DeleteStorageGroups(context.Background(), s.sessionId, storageGroupIds)
+func (s *Session) DeleteStorageGroups(storageGroupIds ...string) error {
+ r, err := s.client.DeleteStorageGroups(context.Background(), s.sessionId, storageGroupIds)
if err != nil && r == nil {
if s.reconnect() {
r, err = s.client.DeleteStorageGroups(context.Background(), s.sessionId, storageGroupIds)
}
}
- return r, err
+ if err != nil {
+ return err
+ }
+ return VerifySuccess(r)
}
/*
@@ -240,9 +249,11 @@ func (s *Session) DeleteStorageGroups(storageGroupIds ...string) (r *common.TSSt
*return
*error: correctness of operation
*/
-func (s *Session) CreateTimeseries(path string, dataType TSDataType, encoding TSEncoding, compressor TSCompressionType, attributes map[string]string, tags map[string]string) (r *common.TSStatus, err error) {
- request := rpc.TSCreateTimeseriesReq{SessionId: s.sessionId, Path: path, DataType: int32(dataType), Encoding: int32(encoding),
- Compressor: int32(compressor), Attributes: attributes, Tags: tags}
+func (s *Session) CreateTimeseries(path string, dataType TSDataType, encoding TSEncoding, compressor TSCompressionType, attributes map[string]string, tags map[string]string) error {
+ request := rpc.TSCreateTimeseriesReq{
+ SessionId: s.sessionId, Path: path, DataType: int32(dataType), Encoding: int32(encoding),
+ Compressor: int32(compressor), Attributes: attributes, Tags: tags,
+ }
status, err := s.client.CreateTimeseries(context.Background(), &request)
if err != nil && status == nil {
if s.reconnect() {
@@ -250,7 +261,10 @@ func (s *Session) CreateTimeseries(path string, dataType TSDataType, encoding TS
status, err = s.client.CreateTimeseries(context.Background(), &request)
}
}
- return status, err
+ if err != nil {
+ return err
+ }
+ return VerifySuccess(status)
}
/*
@@ -265,7 +279,7 @@ func (s *Session) CreateTimeseries(path string, dataType TSDataType, encoding TS
*return
*error: correctness of operation
*/
-func (s *Session) CreateAlignedTimeseries(prefixPath string, measurements []string, dataTypes []TSDataType, encodings []TSEncoding, compressors []TSCompressionType, measurementAlias []string) (r *common.TSStatus, err error) {
+func (s *Session) CreateAlignedTimeseries(prefixPath string, measurements []string, dataTypes []TSDataType, encodings []TSEncoding, compressors []TSCompressionType, measurementAlias []string) error {
destTypes := make([]int32, len(dataTypes))
for i, t := range dataTypes {
destTypes[i] = int32(t)
@@ -297,7 +311,10 @@ func (s *Session) CreateAlignedTimeseries(prefixPath string, measurements []stri
status, err = s.client.CreateAlignedTimeseries(context.Background(), &request)
}
}
- return status, err
+ if err != nil {
+ return err
+ }
+ return VerifySuccess(status)
}
/*
@@ -310,7 +327,7 @@ func (s *Session) CreateAlignedTimeseries(prefixPath string, measurements []stri
*return
*error: correctness of operation
*/
-func (s *Session) CreateMultiTimeseries(paths []string, dataTypes []TSDataType, encodings []TSEncoding, compressors []TSCompressionType) (r *common.TSStatus, err error) {
+func (s *Session) CreateMultiTimeseries(paths []string, dataTypes []TSDataType, encodings []TSEncoding, compressors []TSCompressionType) error {
destTypes := make([]int32, len(dataTypes))
for i, t := range dataTypes {
destTypes[i] = int32(t)
@@ -326,9 +343,11 @@ func (s *Session) CreateMultiTimeseries(paths []string, dataTypes []TSDataType,
destCompressions[i] = int32(e)
}
- request := rpc.TSCreateMultiTimeseriesReq{SessionId: s.sessionId, Paths: paths, DataTypes: destTypes,
- Encodings: destEncodings, Compressors: destCompressions}
- r, err = s.client.CreateMultiTimeseries(context.Background(), &request)
+ request := rpc.TSCreateMultiTimeseriesReq{
+ SessionId: s.sessionId, Paths: paths, DataTypes: destTypes,
+ Encodings: destEncodings, Compressors: destCompressions,
+ }
+ r, err := s.client.CreateMultiTimeseries(context.Background(), &request)
if err != nil && r == nil {
if s.reconnect() {
@@ -337,7 +356,10 @@ func (s *Session) CreateMultiTimeseries(paths []string, dataTypes []TSDataType,
}
}
- return r, err
+ if err != nil {
+ return err
+ }
+ return VerifySuccess(r)
}
/*
@@ -347,14 +369,17 @@ func (s *Session) CreateMultiTimeseries(paths []string, dataTypes []TSDataType,
*return
*error: correctness of operation
*/
-func (s *Session) DeleteTimeseries(paths []string) (r *common.TSStatus, err error) {
- r, err = s.client.DeleteTimeseries(context.Background(), s.sessionId, paths)
+func (s *Session) DeleteTimeseries(paths []string) error {
+ r, err := s.client.DeleteTimeseries(context.Background(), s.sessionId, paths)
if err != nil && r == nil {
if s.reconnect() {
r, err = s.client.DeleteTimeseries(context.Background(), s.sessionId, paths)
}
}
- return r, err
+ if err != nil {
+ return err
+ }
+ return VerifySuccess(r)
}
/*
@@ -366,16 +391,19 @@ func (s *Session) DeleteTimeseries(paths []string) (r *common.TSStatus, err erro
*return
*error: correctness of operation
*/
-func (s *Session) DeleteData(paths []string, startTime int64, endTime int64) (r *common.TSStatus, err error) {
+func (s *Session) DeleteData(paths []string, startTime int64, endTime int64) error {
request := rpc.TSDeleteDataReq{SessionId: s.sessionId, Paths: paths, StartTime: startTime, EndTime: endTime}
- r, err = s.client.DeleteData(context.Background(), &request)
+ r, err := s.client.DeleteData(context.Background(), &request)
if err != nil && r == nil {
if s.reconnect() {
request.SessionId = s.sessionId
r, err = s.client.DeleteData(context.Background(), &request)
}
}
- return r, err
+ if err != nil {
+ return err
+ }
+ return VerifySuccess(r)
}
/*
@@ -388,17 +416,22 @@ func (s *Session) DeleteData(paths []string, startTime int64, endTime int64) (r
*return
*error: correctness of operation
*/
-func (s *Session) InsertStringRecord(deviceId string, measurements []string, values []string, timestamp int64) (r *common.TSStatus, err error) {
- request := rpc.TSInsertStringRecordReq{SessionId: s.sessionId, PrefixPath: deviceId, Measurements: measurements,
- Values: values, Timestamp: timestamp}
- r, err = s.client.InsertStringRecord(context.Background(), &request)
+func (s *Session) InsertStringRecord(deviceId string, measurements []string, values []string, timestamp int64) error {
+ request := rpc.TSInsertStringRecordReq{
+ SessionId: s.sessionId, PrefixPath: deviceId, Measurements: measurements,
+ Values: values, Timestamp: timestamp,
+ }
+ r, err := s.client.InsertStringRecord(context.Background(), &request)
if err != nil && r == nil {
if s.reconnect() {
request.SessionId = s.sessionId
r, err = s.client.InsertStringRecord(context.Background(), &request)
}
}
- return r, err
+ if err != nil {
+ return err
+ }
+ return VerifySuccess(r)
}
func (s *Session) GetTimeZone() (string, error) {
@@ -409,11 +442,17 @@ func (s *Session) GetTimeZone() (string, error) {
return resp.TimeZone, nil
}
-func (s *Session) SetTimeZone(timeZone string) (r *common.TSStatus, err error) {
+func (s *Session) SetTimeZone(timeZone string) error {
request := rpc.TSSetTimeZoneReq{SessionId: s.sessionId, TimeZone: timeZone}
- r, err = s.client.SetTimeZone(context.Background(), &request)
+ r, err := s.client.SetTimeZone(context.Background(), &request)
+ if err != nil {
+ return err
+ }
+ if err := VerifySuccess(r); err != nil {
+ return err
+ }
s.config.TimeZone = timeZone
- return r, err
+ return nil
}
func (s *Session) ExecuteStatementWithContext(ctx context.Context, sql string) (*SessionDataSet, error) {
@@ -444,7 +483,7 @@ func (s *Session) ExecuteStatement(sql string) (*SessionDataSet, error) {
return s.ExecuteStatementWithContext(context.Background(), sql)
}
-func (s *Session) ExecuteNonQueryStatement(sql string) (r *common.TSStatus, err error) {
+func (s *Session) ExecuteNonQueryStatement(sql string) error {
request := rpc.TSExecuteStatementReq{
SessionId: s.sessionId,
Statement: sql,
@@ -460,8 +499,10 @@ func (s *Session) ExecuteNonQueryStatement(sql string) (r *common.TSStatus, err
resp, err = s.client.ExecuteStatementV2(context.Background(), &request)
}
}
-
- return resp.Status, err
+ if err != nil {
+ return err
+ }
+ return VerifySuccess(resp.Status)
}
func (s *Session) ExecuteQueryStatement(sql string, timeoutMs *int64) (*SessionDataSet, error) {
@@ -560,12 +601,12 @@ func (s *Session) genTSInsertRecordReq(deviceId string, time int64,
return request, nil
}
-func (s *Session) InsertRecord(deviceId string, measurements []string, dataTypes []TSDataType, values []interface{}, timestamp int64) (r *common.TSStatus, err error) {
+func (s *Session) InsertRecord(deviceId string, measurements []string, dataTypes []TSDataType, values []interface{}, timestamp int64) error {
request, err := s.genTSInsertRecordReq(deviceId, timestamp, measurements, dataTypes, values, false)
if err != nil {
- return nil, err
+ return err
}
- r, err = s.client.InsertRecord(context.Background(), request)
+ r, err := s.client.InsertRecord(context.Background(), request)
if err != nil && r == nil {
if s.reconnect() {
@@ -574,15 +615,18 @@ func (s *Session) InsertRecord(deviceId string, measurements []string, dataTypes
}
}
- return r, err
+ if err != nil {
+ return err
+ }
+ return VerifySuccess(r)
}
-func (s *Session) InsertAlignedRecord(deviceId string, measurements []string, dataTypes []TSDataType, values []interface{}, timestamp int64) (r *common.TSStatus, err error) {
+func (s *Session) InsertAlignedRecord(deviceId string, measurements []string, dataTypes []TSDataType, values []interface{}, timestamp int64) error {
request, err := s.genTSInsertRecordReq(deviceId, timestamp, measurements, dataTypes, values, true)
if err != nil {
- return nil, err
+ return err
}
- r, err = s.client.InsertRecord(context.Background(), request)
+ r, err := s.client.InsertRecord(context.Background(), request)
if err != nil && r == nil {
if s.reconnect() {
@@ -591,7 +635,10 @@ func (s *Session) InsertAlignedRecord(deviceId string, measurements []string, da
}
}
- return r, err
+ if err != nil {
+ return err
+ }
+ return VerifySuccess(r)
}
type deviceData struct {
@@ -620,11 +667,11 @@ func (d *deviceData) Swap(i, j int) {
// InsertRecordsOfOneDevice Insert multiple rows, which can reduce the overhead of network. This method is just like jdbc
// executeBatch, we pack some insert request in batch and send them to server. If you want improve
// your performance, please see insertTablet method
-// Each row is independent, which could have different deviceId, time, number of measurements
-func (s *Session) InsertRecordsOfOneDevice(deviceId string, timestamps []int64, measurementsSlice [][]string, dataTypesSlice [][]TSDataType, valuesSlice [][]interface{}, sorted bool) (r *common.TSStatus, err error) {
+// Each row is independent, which could have different insertTargetName, time, number of measurements
+func (s *Session) InsertRecordsOfOneDevice(deviceId string, timestamps []int64, measurementsSlice [][]string, dataTypesSlice [][]TSDataType, valuesSlice [][]interface{}, sorted bool) error {
length := len(timestamps)
if len(measurementsSlice) != length || len(dataTypesSlice) != length || len(valuesSlice) != length {
- return nil, errors.New("timestamps, measurementsSlice and valuesSlice's size should be equal")
+ return errors.New("timestamps, measurementsSlice and valuesSlice's size should be equal")
}
if !sorted {
@@ -636,10 +683,11 @@ func (s *Session) InsertRecordsOfOneDevice(deviceId string, timestamps []int64,
})
}
+ var err error
valuesList := make([][]byte, length)
for i := 0; i < length; i++ {
if valuesList[i], err = valuesToBytes(dataTypesSlice[i], valuesSlice[i], measurementsSlice[i]); err != nil {
- return nil, err
+ return err
}
}
@@ -651,7 +699,7 @@ func (s *Session) InsertRecordsOfOneDevice(deviceId string, timestamps []int64,
ValuesList: valuesList,
}
- r, err = s.client.InsertRecordsOfOneDevice(context.Background(), request)
+ r, err := s.client.InsertRecordsOfOneDevice(context.Background(), request)
if err != nil && r == nil {
if s.reconnect() {
@@ -660,13 +708,16 @@ func (s *Session) InsertRecordsOfOneDevice(deviceId string, timestamps []int64,
}
}
- return r, err
+ if err != nil {
+ return err
+ }
+ return VerifySuccess(r)
}
-func (s *Session) InsertAlignedRecordsOfOneDevice(deviceId string, timestamps []int64, measurementsSlice [][]string, dataTypesSlice [][]TSDataType, valuesSlice [][]interface{}, sorted bool) (r *common.TSStatus, err error) {
+func (s *Session) InsertAlignedRecordsOfOneDevice(deviceId string, timestamps []int64, measurementsSlice [][]string, dataTypesSlice [][]TSDataType, valuesSlice [][]interface{}, sorted bool) error {
length := len(timestamps)
if len(measurementsSlice) != length || len(dataTypesSlice) != length || len(valuesSlice) != length {
- return nil, errors.New("timestamps, measurementsSlice and valuesSlice's size should be equal")
+ return errors.New("timestamps, measurementsSlice and valuesSlice's size should be equal")
}
if !sorted {
@@ -678,10 +729,11 @@ func (s *Session) InsertAlignedRecordsOfOneDevice(deviceId string, timestamps []
})
}
+ var err error
valuesList := make([][]byte, length)
for i := 0; i < length; i++ {
if valuesList[i], err = valuesToBytes(dataTypesSlice[i], valuesSlice[i], measurementsSlice[i]); err != nil {
- return nil, err
+ return err
}
}
var isAligned = true
@@ -694,7 +746,7 @@ func (s *Session) InsertAlignedRecordsOfOneDevice(deviceId string, timestamps []
IsAligned: &isAligned,
}
- r, err = s.client.InsertRecordsOfOneDevice(context.Background(), request)
+ r, err := s.client.InsertRecordsOfOneDevice(context.Background(), request)
if err != nil && r == nil {
if s.reconnect() {
@@ -703,7 +755,10 @@ func (s *Session) InsertAlignedRecordsOfOneDevice(deviceId string, timestamps []
}
}
- return r, err
+ if err != nil {
+ return err
+ }
+ return VerifySuccess(r)
}
/*
@@ -719,36 +774,44 @@ func (s *Session) InsertAlignedRecordsOfOneDevice(deviceId string, timestamps []
*
*/
func (s *Session) InsertRecords(deviceIds []string, measurements [][]string, dataTypes [][]TSDataType, values [][]interface{},
- timestamps []int64) (r *common.TSStatus, err error) {
+ timestamps []int64,
+) error {
request, err := s.genInsertRecordsReq(deviceIds, measurements, dataTypes, values, timestamps, false)
if err != nil {
- return nil, err
+ return err
} else {
- r, err = s.client.InsertRecords(context.Background(), request)
+ r, err := s.client.InsertRecords(context.Background(), request)
if err != nil && r == nil {
if s.reconnect() {
request.SessionId = s.sessionId
r, err = s.client.InsertRecords(context.Background(), request)
}
}
- return r, err
+ if err != nil {
+ return err
+ }
+ return VerifySuccess(r)
}
}
func (s *Session) InsertAlignedRecords(deviceIds []string, measurements [][]string, dataTypes [][]TSDataType, values [][]interface{},
- timestamps []int64) (r *common.TSStatus, err error) {
+ timestamps []int64,
+) error {
request, err := s.genInsertRecordsReq(deviceIds, measurements, dataTypes, values, timestamps, true)
if err != nil {
- return nil, err
+ return err
} else {
- r, err = s.client.InsertRecords(context.Background(), request)
+ r, err := s.client.InsertRecords(context.Background(), request)
if err != nil && r == nil {
if s.reconnect() {
request.SessionId = s.sessionId
r, err = s.client.InsertRecords(context.Background(), request)
}
}
- return r, err
+ if err != nil {
+ return err
+ }
+ return VerifySuccess(r)
}
}
@@ -757,63 +820,72 @@ func (s *Session) InsertAlignedRecords(deviceIds []string, measurements [][]stri
*params
*tablets: []*client.Tablet, list of tablets
*/
-func (s *Session) InsertTablets(tablets []*Tablet, sorted bool) (r *common.TSStatus, err error) {
+func (s *Session) InsertTablets(tablets []*Tablet, sorted bool) error {
if !sorted {
for _, t := range tablets {
if err := t.Sort(); err != nil {
- return nil, err
+ return err
}
}
}
request, err := s.genInsertTabletsReq(tablets, false)
if err != nil {
- return nil, err
+ return err
}
- r, err = s.client.InsertTablets(context.Background(), request)
+ r, err := s.client.InsertTablets(context.Background(), request)
if err != nil && r == nil {
if s.reconnect() {
request.SessionId = s.sessionId
r, err = s.client.InsertTablets(context.Background(), request)
}
}
- return r, err
+ if err != nil {
+ return err
+ }
+ return VerifySuccess(r)
}
-func (s *Session) InsertAlignedTablets(tablets []*Tablet, sorted bool) (r *common.TSStatus, err error) {
+func (s *Session) InsertAlignedTablets(tablets []*Tablet, sorted bool) error {
if !sorted {
for _, t := range tablets {
if err := t.Sort(); err != nil {
- return nil, err
+ return err
}
}
}
request, err := s.genInsertTabletsReq(tablets, true)
if err != nil {
- return nil, err
+ return err
}
- r, err = s.client.InsertTablets(context.Background(), request)
+ r, err := s.client.InsertTablets(context.Background(), request)
if err != nil && r == nil {
if s.reconnect() {
request.SessionId = s.sessionId
r, err = s.client.InsertTablets(context.Background(), request)
}
}
- return r, err
+ if err != nil {
+ return err
+ }
+ return VerifySuccess(r)
}
-func (s *Session) ExecuteBatchStatement(inserts []string) (r *common.TSStatus, err error) {
+func (s *Session) ExecuteBatchStatement(inserts []string) error {
request := rpc.TSExecuteBatchStatementReq{
SessionId: s.sessionId,
Statements: inserts,
}
- r, err = s.client.ExecuteBatchStatement(context.Background(), &request)
+ r, err := s.client.ExecuteBatchStatement(context.Background(), &request)
if err != nil && r == nil {
if s.reconnect() {
request.SessionId = s.sessionId
r, err = s.client.ExecuteBatchStatement(context.Background(), &request)
}
}
- return r, err
+ if err != nil {
+ return err
+ }
+ return VerifySuccess(r)
}
func (s *Session) ExecuteRawDataQuery(paths []string, startTime int64, endTime int64) (*SessionDataSet, error) {
@@ -1020,18 +1092,18 @@ func valuesToBytes(dataTypes []TSDataType, values []interface{}, measurementName
return buff.Bytes(), nil
}
-func (s *Session) InsertTablet(tablet *Tablet, sorted bool) (r *common.TSStatus, err error) {
+func (s *Session) InsertTablet(tablet *Tablet, sorted bool) error {
if !sorted {
if err := tablet.Sort(); err != nil {
- return nil, err
+ return err
}
}
request, err := s.genTSInsertTabletReq(tablet, false)
if err != nil {
- return nil, err
+ return err
}
- r, err = s.client.InsertTablet(context.Background(), request)
+ r, err := s.client.InsertTablet(context.Background(), request)
if err != nil && r == nil {
if s.reconnect() {
@@ -1040,21 +1112,24 @@ func (s *Session) InsertTablet(tablet *Tablet, sorted bool) (r *common.TSStatus,
}
}
- return r, err
+ if err != nil {
+ return err
+ }
+ return VerifySuccess(r)
}
-func (s *Session) InsertAlignedTablet(tablet *Tablet, sorted bool) (r *common.TSStatus, err error) {
+func (s *Session) InsertAlignedTablet(tablet *Tablet, sorted bool) error {
if !sorted {
if err := tablet.Sort(); err != nil {
- return nil, err
+ return err
}
}
request, err := s.genTSInsertTabletReq(tablet, true)
if err != nil {
- return nil, err
+ return err
}
- r, err = s.client.InsertTablet(context.Background(), request)
+ r, err := s.client.InsertTablet(context.Background(), request)
if err != nil && r == nil {
if s.reconnect() {
@@ -1063,7 +1138,10 @@ func (s *Session) InsertAlignedTablet(tablet *Tablet, sorted bool) (r *common.TS
}
}
- return r, err
+ if err != nil {
+ return err
+ }
+ return VerifySuccess(r)
}
func (s *Session) genTSInsertTabletReq(tablet *Tablet, isAligned bool) (*rpc.TSInsertTabletReq, error) {
diff --git a/client/sessionpool.go b/client/sessionpool.go
index a05cc4e..6054e7d 100644
--- a/client/sessionpool.go
+++ b/client/sessionpool.go
@@ -151,22 +151,6 @@ func (spool *SessionPool) PutBack(session Session) {
<-spool.sem
}
-func (spool *SessionPool) dropSession(session Session) {
- defer func() {
- if e := recover(); e != nil {
- if session.trans != nil && session.trans.IsOpen() {
- session.Close()
- }
- }
- }()
- err := session.Close()
- if err != nil {
- log.Println("Failed to close session ", session)
- }
- <-spool.sem
-}
-
->>>>>>> f00cf99 (Fix multiple issues in SessionPool and PooledTableSession (#153))
func (spool *SessionPool) Close() {
close(spool.ch)
for s := range spool.ch {
diff --git a/client/utils.go b/client/utils.go
index 41cc783..0f05dac 100644
--- a/client/utils.go
+++ b/client/utils.go
@@ -124,7 +124,7 @@ func verifySuccesses(statuses []*common.TSStatus) error {
}
errMsg := buff.String()
if len(errMsg) > 0 {
- return NewBatchError(statuses)
+ return &BatchError{statuses, errMsg}
}
return nil
}
@@ -141,11 +141,11 @@ func VerifySuccess(status *common.TSStatus) error {
return nil
}
if status.Code != SuccessStatus {
+ msg := ""
if status.Message != nil {
- return fmt.Errorf("error code: %d, message: %v", status.Code, *status.Message)
- } else {
- return fmt.Errorf("error code: %d", status.Code)
+ msg = *status.Message
}
+ return &ExecutionError{Code: status.Code, Message: msg}
}
return nil
}
diff --git a/example/session_example.go b/example/session_example.go
index 1b31514..8ebe918 100644
--- a/example/session_example.go
+++ b/example/session_example.go
@@ -465,9 +465,9 @@ func deleteData() {
func insertTablet() {
if tablet, err := createTablet(12); err == nil {
- status, err := session.InsertTablet(tablet, false)
+ err = session.InsertTablet(tablet, false)
tablet.Reset()
- checkError(status, err)
+ checkError(err)
} else {
log.Fatal(err)
}
@@ -475,9 +475,9 @@ func insertTablet() {
func insertAlignedTablet() {
if tablet, err := createTablet(12); err == nil {
- status, err := session.InsertAlignedTablet(tablet, false)
+ err = session.InsertAlignedTablet(tablet, false)
tablet.Reset()
- checkError(status, err)
+ checkError(err)
} else {
log.Fatal(err)
}
@@ -642,14 +642,8 @@ func executeBatchStatement() {
}
}
-func checkError(status *common.TSStatus, err error) {
+func checkError(err error) {
if err != nil {
log.Fatal(err)
}
-
- if status != nil {
- if err = client.VerifySuccess(status); err != nil {
- log.Println(err)
- }
- }
}
diff --git a/example/session_pool/session_pool_example.go b/example/session_pool/session_pool_example.go
index 6f52ffe..2dfb5bb 100644
--- a/example/session_pool/session_pool_example.go
+++ b/example/session_pool/session_pool_example.go
@@ -22,7 +22,6 @@ package main
import (
"flag"
"fmt"
- "github.com/apache/iotdb-client-go/common"
"log"
"math/rand"
"strings"
@@ -410,9 +409,9 @@ func insertTablet() {
defer sessionPool.PutBack(session)
if err == nil {
if tablet, err := createTablet(12); err == nil {
- status, err := session.InsertTablet(tablet, false)
+ err := session.InsertTablet(tablet, false)
tablet.Reset()
- checkError(status, err)
+ checkError(err)
} else {
log.Fatal(err)
}
@@ -425,9 +424,9 @@ func insertAlignedTablet() {
defer sessionPool.PutBack(session)
if err == nil {
if tablet, err := createTablet(12); err == nil {
- status, err := session.InsertAlignedTablet(tablet, false)
+ err := session.InsertAlignedTablet(tablet, false)
tablet.Reset()
- checkError(status, err)
+ checkError(err)
} else {
log.Fatal(err)
}
@@ -725,14 +724,8 @@ func printDataSet2(sds *client.SessionDataSet) {
}
}
-func checkError(status *common.TSStatus, err error) {
+func checkError(err error) {
if err != nil {
log.Fatal(err)
}
-
- if status != nil {
- if err = client.VerifySuccess(status); err != nil {
- log.Println(err)
- }
- }
}
diff --git a/test/e2e/e2e_test.go b/test/e2e/e2e_test.go
index a225546..be83bbf 100644
--- a/test/e2e/e2e_test.go
+++ b/test/e2e/e2e_test.go
@@ -28,8 +28,6 @@ import (
"testing"
"time"
- "github.com/apache/iotdb-client-go/common"
-
"github.com/apache/iotdb-client-go/client"
"github.com/stretchr/testify/suite"
)
@@ -61,20 +59,17 @@ func (s *e2eTestSuite) TearDownSuite() {
}
func (s *e2eTestSuite) SetupTest() {
- r, err := s.session.SetStorageGroup("root.tsg1")
- s.checkError(r, err)
+ err := s.session.SetStorageGroup("root.tsg1")
+ s.checkError(err)
}
func (s *e2eTestSuite) TearDownTest() {
- r, err := s.session.DeleteStorageGroup("root.tsg1")
- s.checkError(r, err)
+ err := s.session.DeleteStorageGroup("root.tsg1")
+ s.checkError(err)
}
-func (s *e2eTestSuite) checkError(status *common.TSStatus, err error) {
+func (s *e2eTestSuite) checkError(err error) {
s.Require().NoError(err)
- if status != nil {
- s.Require().NoError(client.VerifySuccess(status))
- }
}
func (s *e2eTestSuite) Test_NonQuery() {
@@ -174,7 +169,7 @@ func (s *e2eTestSuite) Test_InsertRecordsWithWrongType() {
values = [][]interface{}{{100.0, true}, {"aaa"}}
timestamp = []int64{1, 2}
)
- _, err := s.session.InsertRecords(deviceId, measurements, dataTypes, values, timestamp)
+ err := s.session.InsertRecords(deviceId, measurements, dataTypes, values, timestamp)
assert := s.Require()
assert.NotNil(err)
assert.Equal("measurement s1 values[0] 100(float64) must be bool", err.Error())
@@ -255,8 +250,8 @@ func (s *e2eTestSuite) Test_InsertAlignedTablet() {
var timeseries = []string{"root.ln.device1.**"}
s.session.DeleteTimeseries(timeseries)
if tablet, err := createTablet(12); err == nil {
- status, err := s.session.InsertAlignedTablet(tablet, false)
- s.checkError(status, err)
+ err := s.session.InsertAlignedTablet(tablet, false)
+ s.checkError(err)
tablet.Reset()
} else {
log.Fatal(err)
@@ -277,8 +272,8 @@ func (s *e2eTestSuite) Test_InsertAlignedTabletWithNilValue() {
var timeseries = []string{"root.ln.device1.**"}
s.session.DeleteTimeseries(timeseries)
if tablet, err := createTabletWithNil(12); err == nil {
- status, err := s.session.InsertAlignedTablet(tablet, false)
- s.checkError(status, err)
+ err := s.session.InsertAlignedTablet(tablet, false)
+ s.checkError(err)
tablet.Reset()
} else {
log.Fatal(err)
@@ -499,8 +494,8 @@ func (s *e2eTestSuite) Test_QueryAllDataType() {
tablet.SetValueAt("string", 9, 0)
tablet.RowSize = 1
- r, err := s.session.InsertAlignedTablet(tablet, true)
- s.checkError(r, err)
+ err = s.session.InsertAlignedTablet(tablet, true)
+ s.checkError(err)
sessionDataSet, err := s.session.ExecuteQueryStatement("select s0, s1, s2, s3, s4, s5, s6, s7, s8, s9 from root.tsg1.d1 limit 1", nil)
for {