neuyilan commented on a change in pull request #2046: URL: https://github.com/apache/iotdb/pull/2046#discussion_r530074190
########## File path: client-go/src/main/client/session.go ########## @@ -0,0 +1,667 @@ +/** + * 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 client + +import ( + "bytes" + "context" + "encoding/binary" + "fmt" + "github.com/apache/thrift/lib/go/thrift" + log "github.com/sirupsen/logrus" + "github.com/yanhongwangg/go-thrift/rpc" + "github.com/yanhongwangg/incubator-iotdb/client-go/src/main/client/utils" + "net" + "os" + "strings" + "time" +) + +const protocolVersion = rpc.TSProtocolVersion_IOTDB_SERVICE_PROTOCOL_V3 + +func (s *Session) Open(enableRPCCompression bool, connectionTimeoutInMs int) { + dir, _ := os.Getwd() + os.Mkdir(dir+"\\logs", os.ModePerm) + logFile, _ := os.OpenFile(dir+"\\logs\\all.log", os.O_CREATE|os.O_RDWR|os.O_APPEND, 0666) + log.SetOutput(logFile) + log.SetLevel(log.InfoLevel) + var protocolFactory thrift.TProtocolFactory + s.trans, s.err = thrift.NewTSocketTimeout(net.JoinHostPort(s.Host, s.Port), time.Duration(connectionTimeoutInMs)) + if s.err != nil { + log.WithError(s.err).Error("connect failed") + } + s.trans = thrift.NewTFramedTransport(s.trans) + if !s.trans.IsOpen() { + s.err = s.trans.Open() + if s.err != nil { + log.WithError(s.err).Error("open the conn failed") + } + } + if enableRPCCompression { + protocolFactory = thrift.NewTCompactProtocolFactory() + } else { + protocolFactory = thrift.NewTBinaryProtocolFactoryDefault() + } + iProtocol := protocolFactory.GetProtocol(s.trans) + oProtocol := protocolFactory.GetProtocol(s.trans) + s.client = rpc.NewTSIServiceClient(thrift.NewTStandardClient(iProtocol, oProtocol)) + s.ZoneId = DefaultZoneId + tSOpenSessionReq := rpc.TSOpenSessionReq{ClientProtocol: protocolVersion, ZoneId: s.ZoneId, Username: &s.User, + Password: &s.Passwd} + tSOpenSessionResp, err := s.client.OpenSession(context.Background(), &tSOpenSessionReq) + if err != nil { + log.WithError(err).Error("open session failed") + } else { + log.WithField("code", tSOpenSessionResp.GetStatus().Code).Debug("open session success") + } + s.sessionId = tSOpenSessionResp.GetSessionId() + s.requestStatementId, err = s.client.RequestStatementId(context.Background(), s.sessionId) + s.FetchSize = DefaultFetchSize + if err != nil { + log.WithError(err).Error("request StatementId failed") + } + if s.ZoneId != "" { + s.SetTimeZone(s.ZoneId) + } else { + s.ZoneId = s.GetTimeZone() + } +} + +func (s *Session) CheckTimeseriesExists(path string) bool { + dataSet := s.ExecuteQueryStatement("SHOW TIMESERIES " + path) + result := dataSet.HasNext() + dataSet.CloseOperationHandle() + return result +} + +func (s *Session) Close() { + tSCloseSessionReq := rpc.NewTSCloseSessionReq() + tSCloseSessionReq.SessionId = s.sessionId + status, err := s.client.CloseSession(context.Background(), tSCloseSessionReq) + if err != nil { + log.WithError(err).Error("Error occurs when closing session at server. Maybe server is down.") + } else { + log.WithField("code", status.Code).Debug("close session success") + } + s.trans.Close() +} + +/* + *set one storage group + * + *param + *storageGroupId: string, storage group name (starts from root) + * + */ +func (s *Session) SetStorageGroup(storageGroupId string) { + status, err := s.client.SetStorageGroup(context.Background(), s.sessionId, storageGroupId) + if err != nil { + log.WithError(err).Error("setting storage group failed") + } else { + log.WithFields(log.Fields{ + "sg": storageGroupId, + "code": status.Code, + }).Debug("setting storage group success") + } +} + +/* + *delete one storage group + * + *param + *storageGroupId: string, storage group name (starts from root) + * + */ +func (s *Session) DeleteStorageGroup(storageGroupId string) { + status, err := s.client.DeleteStorageGroups(context.Background(), s.sessionId, []string{storageGroupId}) + if err != nil { + log.WithError(err).Error("delete storage group failed") + } else { + log.WithFields(log.Fields{ + "sg": storageGroupId, + "code": status.Code, + }).Debug("delete storage group success") + } +} + +/* + *delete multiple storage group + * + *param + *storageGroupIds: []string, paths of the target storage groups + * + */ +func (s *Session) DeleteStorageGroups(storageGroupIds []string) { + status, err := s.client.DeleteStorageGroups(context.Background(), s.sessionId, storageGroupIds) + s.sg = strings.Replace(strings.Trim(fmt.Sprint(storageGroupIds), "[]"), " ", ",", -1) Review comment: Do not do the code here as you just want to print the devices in debug log, for every operation, the code will be called, it'll cost cpu `s.sg = strings.Replace(strings.Trim(fmt.Sprint(storageGroupIds), "[]"), " ", ",", -1)` ########## File path: client-go/src/main/client/session.go ########## @@ -0,0 +1,667 @@ +/** + * 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 client + +import ( + "bytes" + "context" + "encoding/binary" + "fmt" + "github.com/apache/thrift/lib/go/thrift" + log "github.com/sirupsen/logrus" + "github.com/yanhongwangg/go-thrift/rpc" + "github.com/yanhongwangg/incubator-iotdb/client-go/src/main/client/utils" + "net" + "os" + "strings" + "time" +) + +const protocolVersion = rpc.TSProtocolVersion_IOTDB_SERVICE_PROTOCOL_V3 + +func (s *Session) Open(enableRPCCompression bool, connectionTimeoutInMs int) { + dir, _ := os.Getwd() + os.Mkdir(dir+"\\logs", os.ModePerm) + logFile, _ := os.OpenFile(dir+"\\logs\\all.log", os.O_CREATE|os.O_RDWR|os.O_APPEND, 0666) + log.SetOutput(logFile) + log.SetLevel(log.InfoLevel) + var protocolFactory thrift.TProtocolFactory + s.trans, s.err = thrift.NewTSocketTimeout(net.JoinHostPort(s.Host, s.Port), time.Duration(connectionTimeoutInMs)) + if s.err != nil { + log.WithError(s.err).Error("connect failed") + } + s.trans = thrift.NewTFramedTransport(s.trans) + if !s.trans.IsOpen() { + s.err = s.trans.Open() + if s.err != nil { + log.WithError(s.err).Error("open the conn failed") + } + } + if enableRPCCompression { + protocolFactory = thrift.NewTCompactProtocolFactory() + } else { + protocolFactory = thrift.NewTBinaryProtocolFactoryDefault() + } + iProtocol := protocolFactory.GetProtocol(s.trans) + oProtocol := protocolFactory.GetProtocol(s.trans) + s.client = rpc.NewTSIServiceClient(thrift.NewTStandardClient(iProtocol, oProtocol)) + s.ZoneId = DefaultZoneId + tSOpenSessionReq := rpc.TSOpenSessionReq{ClientProtocol: protocolVersion, ZoneId: s.ZoneId, Username: &s.User, + Password: &s.Passwd} + tSOpenSessionResp, err := s.client.OpenSession(context.Background(), &tSOpenSessionReq) + if err != nil { + log.WithError(err).Error("open session failed") + } else { + log.WithField("code", tSOpenSessionResp.GetStatus().Code).Debug("open session success") + } + s.sessionId = tSOpenSessionResp.GetSessionId() + s.requestStatementId, err = s.client.RequestStatementId(context.Background(), s.sessionId) + s.FetchSize = DefaultFetchSize + if err != nil { + log.WithError(err).Error("request StatementId failed") + } + if s.ZoneId != "" { + s.SetTimeZone(s.ZoneId) + } else { + s.ZoneId = s.GetTimeZone() + } +} + +func (s *Session) CheckTimeseriesExists(path string) bool { + dataSet := s.ExecuteQueryStatement("SHOW TIMESERIES " + path) + result := dataSet.HasNext() + dataSet.CloseOperationHandle() + return result +} + +func (s *Session) Close() { + tSCloseSessionReq := rpc.NewTSCloseSessionReq() + tSCloseSessionReq.SessionId = s.sessionId + status, err := s.client.CloseSession(context.Background(), tSCloseSessionReq) + if err != nil { + log.WithError(err).Error("Error occurs when closing session at server. Maybe server is down.") + } else { + log.WithField("code", status.Code).Debug("close session success") + } + s.trans.Close() +} + +/* + *set one storage group + * + *param + *storageGroupId: string, storage group name (starts from root) + * + */ +func (s *Session) SetStorageGroup(storageGroupId string) { + status, err := s.client.SetStorageGroup(context.Background(), s.sessionId, storageGroupId) + if err != nil { + log.WithError(err).Error("setting storage group failed") + } else { + log.WithFields(log.Fields{ + "sg": storageGroupId, + "code": status.Code, + }).Debug("setting storage group success") + } +} + +/* + *delete one storage group + * + *param + *storageGroupId: string, storage group name (starts from root) + * + */ +func (s *Session) DeleteStorageGroup(storageGroupId string) { + status, err := s.client.DeleteStorageGroups(context.Background(), s.sessionId, []string{storageGroupId}) + if err != nil { + log.WithError(err).Error("delete storage group failed") + } else { + log.WithFields(log.Fields{ + "sg": storageGroupId, + "code": status.Code, + }).Debug("delete storage group success") + } +} + +/* + *delete multiple storage group + * + *param + *storageGroupIds: []string, paths of the target storage groups + * + */ +func (s *Session) DeleteStorageGroups(storageGroupIds []string) { + status, err := s.client.DeleteStorageGroups(context.Background(), s.sessionId, storageGroupIds) + s.sg = strings.Replace(strings.Trim(fmt.Sprint(storageGroupIds), "[]"), " ", ",", -1) + if err != nil { + log.WithError(err).Error("delete storage groups failed") + } else { + log.WithFields(log.Fields{ + "sg": s.sg, + "code": status.Code, + }).Debug("delete storage groups success") + } +} + +/* + *create single time series + * + *params + *path: string, complete time series path (starts from root) + *dataType: int32, data type for this time series + *encoding: int32, data type for this time series + *compressor: int32, compressing type for this time series + * + */ +func (s *Session) CreateTimeseries(path string, dataType int32, encoding int32, compressor int32) { + request := rpc.TSCreateTimeseriesReq{SessionId: s.sessionId, Path: path, DataType: dataType, Encoding: encoding, + Compressor: compressor} + status, err := s.client.CreateTimeseries(context.Background(), &request) + if err != nil { + log.WithError(err).Error("creating time series failed") + } else { + log.WithFields(log.Fields{ + "ts": path, + "code": status.Code, + }).Debug("creating time series success") + } +} + +/* + *create multiple time series + * + *params + *paths: []string, complete time series paths (starts from root) + *dataTypes: []int32, data types for time series + *encodings: []int32, encodings for time series + *compressors: []int32, compressing types for time series + * + */ +func (s *Session) CreateMultiTimeseries(paths []string, dataTypes []int32, encodings []int32, compressors []int32) { + request := rpc.TSCreateMultiTimeseriesReq{SessionId: s.sessionId, Paths: paths, DataTypes: dataTypes, + Encodings: encodings, Compressors: compressors} + status, err := s.client.CreateMultiTimeseries(context.Background(), &request) + s.ts = strings.Replace(strings.Trim(fmt.Sprint(paths), "[]"), " ", ",", -1) Review comment: same as aboveļ¼ ########## File path: client-go/src/main/client/session.go ########## @@ -0,0 +1,698 @@ +/** + * 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 client + +import ( + "bytes" + "context" + "encoding/binary" + "fmt" + "github.com/apache/thrift/lib/go/thrift" + "github.com/sirupsen/logrus" + "github.com/yanhongwangg/go-thrift/rpc" + "github.com/yanhongwangg/incubator-iotdb/client-go/src/main/client/utils" + "net" + "os" + "strings" + "time" +) + +const protocolVersion = rpc.TSProtocolVersion_IOTDB_SERVICE_PROTOCOL_V3 + +var Log = logrus.New() + +func (s *Session) Open(enableRPCCompression bool, connectionTimeoutInMs int) { + dir, _ := os.Getwd() + os.Mkdir(dir+"\\logs", os.ModePerm) + logFile, _ := os.OpenFile(dir+"\\logs\\all.log", os.O_CREATE|os.O_RDWR|os.O_APPEND, 0666) + Log.SetOutput(logFile) + var protocolFactory thrift.TProtocolFactory + s.trans, s.err = thrift.NewTSocketTimeout(net.JoinHostPort(s.Host, s.Port), time.Duration(connectionTimeoutInMs)) + if s.err != nil { + Log.WithError(s.err).Error("connect failed") + } + s.trans = thrift.NewTFramedTransport(s.trans) + if !s.trans.IsOpen() { + s.err = s.trans.Open() + if s.err != nil { + Log.WithError(s.err).Error("open the conn failed") + } + } + if enableRPCCompression { + protocolFactory = thrift.NewTCompactProtocolFactory() + } else { + protocolFactory = thrift.NewTBinaryProtocolFactoryDefault() + } + iProtocol := protocolFactory.GetProtocol(s.trans) + oProtocol := protocolFactory.GetProtocol(s.trans) + s.client = rpc.NewTSIServiceClient(thrift.NewTStandardClient(iProtocol, oProtocol)) + s.ZoneId = DefaultZoneId + tSOpenSessionReq := rpc.TSOpenSessionReq{ClientProtocol: protocolVersion, ZoneId: s.ZoneId, Username: &s.User, + Password: &s.Passwd} + tSOpenSessionResp, err := s.client.OpenSession(context.Background(), &tSOpenSessionReq) + if err != nil { + Log.WithError(err).Error("open session failed") + } else { + Log.WithField("code", tSOpenSessionResp.GetStatus().Code).Info("open session success") + } + s.sessionId = tSOpenSessionResp.GetSessionId() + s.requestStatementId, err = s.client.RequestStatementId(context.Background(), s.sessionId) + s.FetchSize = DefaultFetchSize + if err != nil { + Log.WithError(err).Error("request StatementId failed") + } + if s.ZoneId != "" { + s.SetTimeZone(s.ZoneId) + } else { + s.ZoneId = s.GetTimeZone() + } +} + +func (s *Session) CheckTimeseriesExists(path string) bool { + dataSet := s.ExecuteQueryStatement("SHOW TIMESERIES " + path) + result := dataSet.HasNext() + dataSet.CloseOperationHandle() + return result +} + +func (s *Session) Close() error { + tSCloseSessionReq := rpc.NewTSCloseSessionReq() + tSCloseSessionReq.SessionId = s.sessionId + status, err := s.client.CloseSession(context.Background(), tSCloseSessionReq) + if err != nil { + Log.WithError(err).Error("Error occurs when closing session at server. Maybe server is down. Error message") + return err + } else { + Log.WithField("code", status.Code).Info("close session success") + } + s.trans.Close() + return nil +} + +/* + *set one storage group + * + *param + *storageGroupId: string, storage group name (starts from root) + * + *return + *error: correctness of operation + */ +func (s *Session) SetStorageGroup(storageGroupId string) error { + status, err := s.client.SetStorageGroup(context.Background(), s.sessionId, storageGroupId) + if err != nil { + Log.WithError(err).Error("setting storage group failed") + } else { + Log.WithFields(logrus.Fields{ + "sg": storageGroupId, + "code": status.Code, + }).Info("setting storage group success") + } + return err +} + +/* + *delete one storage group + * + *param + *storageGroupId: string, storage group name (starts from root) + * + *return + *error: correctness of operation + */ +func (s *Session) DeleteStorageGroup(storageGroupId string) { + status, err := s.client.DeleteStorageGroups(context.Background(), s.sessionId, []string{storageGroupId}) + if err != nil { + Log.WithError(err).Error("delete storage group failed") + } else { + Log.WithFields(logrus.Fields{ + "sg": storageGroupId, + "code": status.Code, + }).Info("delete storage group success") + } +} + +/* + *delete multiple storage group + * + *param + *storageGroupIds: []string, paths of the target storage groups + * + *return + *error: correctness of operation + */ +func (s *Session) DeleteStorageGroups(storageGroupIds []string) { + status, err := s.client.DeleteStorageGroups(context.Background(), s.sessionId, storageGroupIds) + s.sg = strings.Replace(strings.Trim(fmt.Sprint(storageGroupIds), "[]"), " ", ",", -1) + if err != nil { + Log.WithError(err).Error("delete storage groups failed") + } else { + Log.WithFields(logrus.Fields{ + "sg": s.sg, + "code": status.Code, + }).Info("delete storage groups success") + } +} + +/* + *create single time series + * + *params + *path: string, complete time series path (starts from root) + *dataType: int32, data type for this time series + *encoding: int32, data type for this time series + *compressor: int32, compressing type for this time series + * + *return + *error: correctness of operation + */ +func (s *Session) CreateTimeseries(path string, dataType int32, encoding int32, compressor int32) { + request := rpc.TSCreateTimeseriesReq{SessionId: s.sessionId, Path: path, DataType: dataType, Encoding: encoding, + Compressor: compressor} + status, err := s.client.CreateTimeseries(context.Background(), &request) + if err != nil { + Log.WithError(err).Error("creating time series failed") + } else { + Log.WithFields(logrus.Fields{ + "ts": path, + "code": status.Code, + }).Info("creating time series success") + } +} + +/* + *create multiple time series + * + *params + *paths: []string, complete time series paths (starts from root) + *dataTypes: []int32, data types for time series + *encodings: []int32, encodings for time series + *compressors: []int32, compressing types for time series + * + *return + *error: correctness of operation + */ +func (s *Session) CreateMultiTimeseries(paths []string, dataTypes []int32, encodings []int32, compressors []int32) { + request := rpc.TSCreateMultiTimeseriesReq{SessionId: s.sessionId, Paths: paths, DataTypes: dataTypes, + Encodings: encodings, Compressors: compressors} + status, err := s.client.CreateMultiTimeseries(context.Background(), &request) + s.ts = strings.Replace(strings.Trim(fmt.Sprint(paths), "[]"), " ", ",", -1) + if err != nil { + Log.WithError(err).Error("creating multi time series failed") + } else { + Log.WithFields(logrus.Fields{ + "ts": s.ts, + "code": status.Code, + }).Info("creating multi time series success") + } +} + +/* + *delete multiple time series, including data and schema + * + *params + *paths: []string, time series paths, which should be complete (starts from root) + * + *return + *error: correctness of operation + */ +func (s *Session) DeleteTimeseries(paths []string) { + status, err := s.client.DeleteTimeseries(context.Background(), s.sessionId, paths) + var ts = strings.Replace(strings.Trim(fmt.Sprint(paths), "[]"), " ", ",", -1) + if err != nil { + Log.WithError(err).Error("delete time series failed") + } else { + Log.WithFields(logrus.Fields{ + "ts": ts, + "code": status.Code, + }).Info("delete time series success") + } +} + +/* + *delete all startTime <= data <= endTime in multiple time series + * + *params + *paths: []string, time series array that the data in + *startTime: int64, start time of deletion range + *endTime: int64, end time of deletion range + * + *return + *error: correctness of operation + */ +func (s *Session) DeleteData(paths []string, startTime int64, endTime int64) { + request := rpc.TSDeleteDataReq{SessionId: s.sessionId, Paths: paths, StartTime: startTime, EndTime: endTime} + status, err := s.client.DeleteData(context.Background(), &request) + s.ts = strings.Replace(strings.Trim(fmt.Sprint(paths), "[]"), " ", ",", -1) + if err != nil { + Log.WithError(err).Error("delete data failed") + } else { + Log.WithFields(logrus.Fields{ + "ts": s.ts, + "code": status.Code, + }).Info("delete data success") + } +} + +/* + *special case for inserting one row of String (TEXT) value + * + *params + *deviceId: string, time series path for device + *measurements: []string, sensor names + *values: []string, values to be inserted, for each sensor + *timestamp: int64, indicate the timestamp of the row of data + * + *return + *error: correctness of operation + */ +func (s *Session) InsertStringRecord(deviceId string, measurements []string, values []string, timestamp int64) { + request := rpc.TSInsertStringRecordReq{SessionId: s.sessionId, DeviceId: deviceId, Measurements: measurements, + Values: values, Timestamp: timestamp} + status, err := s.client.InsertStringRecord(context.Background(), &request) + if err != nil { + Log.WithError(err).Error("insert one string record failed") + } else { + Log.WithFields(logrus.Fields{ + "dv": deviceId, + "code": status.Code, + }).Info("insert one string record success") + } +} + +func (s *Session) TestInsertStringRecord(deviceId string, measurements []string, values []string, timestamp int64) { + request := rpc.TSInsertStringRecordReq{SessionId: s.sessionId, DeviceId: deviceId, Measurements: measurements, + Values: values, Timestamp: timestamp} + status, err := s.client.TestInsertStringRecord(context.Background(), &request) + if err != nil { + Log.WithError(err).Error("insert one string record failed") + } else { + Log.WithFields(logrus.Fields{ + "dv": deviceId, + "code": status.Code, + }).Info("insert one string record success") + } +} + +/* + *special case for inserting multiple rows of String (TEXT) value + * + *params + *deviceIds: []string, time series paths for device + *measurements: [][]string, each element of outer list indicates measurements of a device + *values: [][]interface{}, values to be inserted, for each device + *timestamps: []int64, timestamps for records + * + *return + *error: correctness of operation + */ +func (s *Session) InsertStringRecords(deviceIds []string, measurements [][]string, values [][]string, + timestamps []int64) { + request := rpc.TSInsertStringRecordsReq{SessionId: s.sessionId, DeviceIds: deviceIds, MeasurementsList: measurements, + ValuesList: values, Timestamps: timestamps} + s.dv = strings.Replace(strings.Trim(fmt.Sprint(deviceIds), "[]"), " ", ",", -1) + status, err := s.client.InsertStringRecords(context.Background(), &request) + if err != nil { + Log.WithError(err).Error("insert multi string records failed") + } else { + Log.WithFields(logrus.Fields{ + "dv": s.dv, + "code": status.Code, + }).Info("insert multi string records success") + } +} + +func (s *Session) TestInsertStringRecords(deviceIds []string, measurements [][]string, values [][]string, + timestamps []int64) { + request := rpc.TSInsertStringRecordsReq{SessionId: s.sessionId, DeviceIds: deviceIds, MeasurementsList: measurements, + ValuesList: values, Timestamps: timestamps} + s.dv = strings.Replace(strings.Trim(fmt.Sprint(deviceIds), "[]"), " ", ",", -1) + status, err := s.client.TestInsertStringRecords(context.Background(), &request) + if err != nil { + Log.WithError(err).Error("insert multi string records failed") + } else { + Log.WithFields(logrus.Fields{ + "dv": s.dv, + "code": status.Code, + }).Info("insert multi string records success") + } +} + +/* + *insert one row of record into database, if you want improve your performance, please use insertTablet method + * + *params + *deviceId: string, time series path for device + *measurements: []string, sensor names + *dataTypes: []int32, list of dataType, indicate the data type for each sensor + *values: []interface{}, values to be inserted, for each sensor + *timestamp: int64, indicate the timestamp of the row of data + * + *return + *error: correctness of operation + */ +func (s *Session) InsertRecord(deviceId string, measurements []string, dataTypes []int16, values []interface{}, + timestamp int64) { + request := s.genInsertRecordReq(deviceId, measurements, dataTypes, values, timestamp) + status, err := s.client.InsertRecord(context.Background(), request) + if err != nil { + Log.WithError(err).Error("insert one record failed") + } else { + Log.WithFields(logrus.Fields{ + "dv": deviceId, + "code": status.Code, + }).Info("insert one record success") + } +} + +func (s *Session) TestInsertRecord(deviceId string, measurements []string, dataTypes []int16, values []interface{}, + timestamp int64) { + request := s.genInsertRecordReq(deviceId, measurements, dataTypes, values, timestamp) + status, err := s.client.TestInsertRecord(context.Background(), request) + if err != nil { + Log.WithError(err).Error("insert one record failed") + } else { + Log.WithFields(logrus.Fields{ + "dv": deviceId, + "code": status.Code, + }).Info("insert one record success") + } +} + +func (s *Session) genInsertRecordReq(deviceId string, measurements []string, dataTypes []int16, values []interface{}, + timestamp int64) *rpc.TSInsertRecordReq { + request := rpc.TSInsertRecordReq{SessionId: s.sessionId, DeviceId: deviceId, Measurements: measurements, + Timestamp: timestamp} + request.Values = valuesToBytes(dataTypes, values) + return &request +} + +/* + *insert multiple rows of data, records are independent to each other, in other words, there's no relationship + *between those records + * + *params + *deviceIds: []string, time series paths for device + *measurements: [][]string, each element of outer list indicates measurements of a device + *dataTypes: [][]int32, each element of outer list indicates sensor data types of a device + *values: [][]interface{}, values to be inserted, for each device + *timestamps: []int64, timestamps for records + * + *return + *error: correctness of operation + */ +func (s *Session) InsertRecords(deviceIds []string, measurements [][]string, dataTypes [][]int16, values [][]interface{}, + timestamps []int64) { + s.dv = strings.Replace(strings.Trim(fmt.Sprint(deviceIds), "[]"), " ", ",", -1) Review comment: then you can do it if the log debug level is enabled, if the log level is info, for every operation the code will be called once, it's unnecessary! ---------------------------------------------------------------- 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. For queries about this service, please contact Infrastructure at: [email protected]
