neuyilan commented on a change in pull request #2046: URL: https://github.com/apache/iotdb/pull/2046#discussion_r528556583
########## File path: client-go/README.md ########## @@ -0,0 +1,304 @@ +<!-- + + 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. + +--> + +To use go client, add "import github.com/yanhongwangg/incubator-iotdb/client-go" + +## Requirement +* GoLang 1.14+ + +## Getting started +*Create project eg:session + +*Enter project directory eg:cd session + +*Download dependency : go get github.com/yanhongwangg/incubator-iotdb@client-go + +*eg:session_example.go + +``` +package main + +import ( Review comment: if those codes same as the session_example.go, I think we can just add the session_example.go link here. ########## 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") Review comment: please make the error message more clear. ########## 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) { Review comment: the method no return value, the annotation should be consistent with the codes, the same for other methods ########## 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") Review comment: I think if the operation succeeds, we do not need to print the info message, as this will print so many logs for users who use the go-client. I think the debug message is ok. and the default log level is info. ########## 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: for every inset operation, this may time-consume, If you want to print the devices, I think the `deviceIds[]` is just enough. ---------------------------------------------------------------- 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: us...@infra.apache.org
