| # 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. |
| # |
| import logging |
| import struct |
| import time |
| from thrift.protocol import TBinaryProtocol, TCompactProtocol |
| from thrift.transport import TSocket, TTransport |
| |
| from iotdb.utils.SessionDataSet import SessionDataSet |
| from .template.Template import Template |
| from .template.TemplateQueryType import TemplateQueryType |
| from .thrift.rpc.IClientRPCService import ( |
| Client, |
| TSCreateTimeseriesReq, |
| TSCreateAlignedTimeseriesReq, |
| TSInsertRecordReq, |
| TSInsertStringRecordReq, |
| TSInsertTabletReq, |
| TSExecuteStatementReq, |
| TSOpenSessionReq, |
| TSCreateMultiTimeseriesReq, |
| TSCloseSessionReq, |
| TSInsertTabletsReq, |
| TSInsertRecordsReq, |
| TSInsertRecordsOfOneDeviceReq, |
| TSCreateSchemaTemplateReq, |
| TSDropSchemaTemplateReq, |
| TSAppendSchemaTemplateReq, |
| TSPruneSchemaTemplateReq, |
| TSSetSchemaTemplateReq, |
| TSUnsetSchemaTemplateReq, |
| TSQueryTemplateReq, |
| ) |
| from .thrift.rpc.ttypes import ( |
| TSDeleteDataReq, |
| TSProtocolVersion, |
| TSSetTimeZoneReq, |
| TSRawDataQueryReq, |
| TSLastDataQueryReq, |
| TSInsertStringRecordsOfOneDeviceReq, |
| ) |
| # for debug |
| # from IoTDBConstants import * |
| # from SessionDataSet import SessionDataSet |
| # |
| # from thrift.protocol import TBinaryProtocol, TCompactProtocol |
| # from thrift.transport import TSocket, TTransport |
| # |
| # from iotdb.rpc.IClientRPCService import Client, TSCreateTimeseriesReq, TSInsertRecordReq, TSInsertTabletReq, \ |
| # TSExecuteStatementReq, TSOpenSessionReq, TSQueryDataSet, TSFetchResultsReq, TSCloseOperationReq, \ |
| # TSCreateMultiTimeseriesReq, TSCloseSessionReq, TSInsertTabletsReq, TSInsertRecordsReq |
| # from iotdb.rpc.ttypes import TSDeleteDataReq, TSProtocolVersion, TSSetTimeZoneReq |
| from .utils.IoTDBConstants import TSDataType |
| |
| logger = logging.getLogger("IoTDB") |
| |
| |
| class Session(object): |
| SUCCESS_CODE = 200 |
| DEFAULT_FETCH_SIZE = 10000 |
| DEFAULT_USER = "root" |
| DEFAULT_PASSWORD = "root" |
| DEFAULT_ZONE_ID = time.strftime("%z") |
| |
| def __init__( |
| self, |
| host, |
| port, |
| user=DEFAULT_USER, |
| password=DEFAULT_PASSWORD, |
| fetch_size=DEFAULT_FETCH_SIZE, |
| zone_id=DEFAULT_ZONE_ID, |
| ): |
| self.__host = host |
| self.__port = port |
| self.__user = user |
| self.__password = password |
| self.__fetch_size = fetch_size |
| self.__is_close = True |
| self.__transport = None |
| self.__client = None |
| self.protocol_version = TSProtocolVersion.IOTDB_SERVICE_PROTOCOL_V3 |
| self.__session_id = None |
| self.__statement_id = None |
| self.__zone_id = zone_id |
| |
| def open(self, enable_rpc_compression): |
| if not self.__is_close: |
| return |
| self.__transport = TTransport.TFramedTransport( |
| TSocket.TSocket(self.__host, self.__port) |
| ) |
| |
| if not self.__transport.isOpen(): |
| try: |
| self.__transport.open() |
| except TTransport.TTransportException as e: |
| logger.exception("TTransportException!", exc_info=e) |
| |
| if enable_rpc_compression: |
| self.__client = Client(TCompactProtocol.TCompactProtocol(self.__transport)) |
| else: |
| self.__client = Client(TBinaryProtocol.TBinaryProtocol(self.__transport)) |
| |
| open_req = TSOpenSessionReq( |
| client_protocol=self.protocol_version, |
| username=self.__user, |
| password=self.__password, |
| zoneId=self.__zone_id, |
| configuration={"version": "V_0_13"}, |
| ) |
| |
| try: |
| open_resp = self.__client.openSession(open_req) |
| |
| if self.protocol_version != open_resp.serverProtocolVersion: |
| logger.exception( |
| "Protocol differ, Client version is {}, but Server version is {}".format( |
| self.protocol_version, open_resp.serverProtocolVersion |
| ) |
| ) |
| # version is less than 0.10 |
| if open_resp.serverProtocolVersion == 0: |
| raise TTransport.TException(message="Protocol not supported.") |
| |
| self.__session_id = open_resp.sessionId |
| self.__statement_id = self.__client.requestStatementId(self.__session_id) |
| |
| except Exception as e: |
| self.__transport.close() |
| logger.exception("session closed because: ", exc_info=e) |
| |
| if self.__zone_id is not None: |
| self.set_time_zone(self.__zone_id) |
| else: |
| self.__zone_id = self.get_time_zone() |
| |
| self.__is_close = False |
| |
| def is_open(self): |
| return not self.__is_close |
| |
| def close(self): |
| if self.__is_close: |
| return |
| req = TSCloseSessionReq(self.__session_id) |
| try: |
| self.__client.closeSession(req) |
| except TTransport.TException as e: |
| logger.exception( |
| "Error occurs when closing session at server. Maybe server is down. Error message: ", |
| exc_info=e, |
| ) |
| finally: |
| self.__is_close = True |
| if self.__transport is not None: |
| self.__transport.close() |
| |
| def set_storage_group(self, group_name): |
| """ |
| set one storage group |
| :param group_name: String, storage group name (starts from root) |
| """ |
| status = self.__client.setStorageGroup(self.__session_id, group_name) |
| logger.debug( |
| "setting storage group {} message: {}".format(group_name, status.message) |
| ) |
| |
| return Session.verify_success(status) |
| |
| def delete_storage_group(self, storage_group): |
| """ |
| delete one storage group. |
| :param storage_group: String, path of the target storage group. |
| """ |
| groups = [storage_group] |
| return self.delete_storage_groups(groups) |
| |
| def delete_storage_groups(self, storage_group_lst): |
| """ |
| delete multiple storage groups. |
| :param storage_group_lst: List, paths of the target storage groups. |
| """ |
| status = self.__client.deleteStorageGroups(self.__session_id, storage_group_lst) |
| logger.debug( |
| "delete storage group(s) {} message: {}".format( |
| storage_group_lst, status.message |
| ) |
| ) |
| |
| return Session.verify_success(status) |
| |
| def create_time_series( |
| self, |
| ts_path, |
| data_type, |
| encoding, |
| compressor, |
| props=None, |
| tags=None, |
| attributes=None, |
| alias=None, |
| ): |
| """ |
| create single time series |
| :param ts_path: String, complete time series path (starts from root) |
| :param data_type: TSDataType, data type for this time series |
| :param encoding: TSEncoding, encoding for this time series |
| :param compressor: Compressor, compressing type for this time series |
| :param props: Dictionary, properties for time series |
| :param tags: Dictionary, tag map for time series |
| :param attributes: Dictionary, attribute map for time series |
| :param alias: String, measurement alias for time series |
| """ |
| data_type = data_type.value |
| encoding = encoding.value |
| compressor = compressor.value |
| request = TSCreateTimeseriesReq( |
| self.__session_id, |
| ts_path, |
| data_type, |
| encoding, |
| compressor, |
| props, |
| tags, |
| attributes, |
| alias, |
| ) |
| status = self.__client.createTimeseries(request) |
| logger.debug( |
| "creating time series {} message: {}".format(ts_path, status.message) |
| ) |
| |
| return Session.verify_success(status) |
| |
| def create_aligned_time_series( |
| self, device_id, measurements_lst, data_type_lst, encoding_lst, compressor_lst |
| ): |
| """ |
| create aligned time series |
| :param device_id: String, device id for timeseries (starts from root) |
| :param measurements_lst: List of String, measurement ids for time series |
| :param data_type_lst: List of TSDataType, data types for time series |
| :param encoding_lst: List of TSEncoding, encodings for time series |
| :param compressor_lst: List of Compressor, compressing types for time series |
| """ |
| data_type_lst = [data_type.value for data_type in data_type_lst] |
| encoding_lst = [encoding.value for encoding in encoding_lst] |
| compressor_lst = [compressor.value for compressor in compressor_lst] |
| |
| request = TSCreateAlignedTimeseriesReq( |
| self.__session_id, |
| device_id, |
| measurements_lst, |
| data_type_lst, |
| encoding_lst, |
| compressor_lst, |
| ) |
| status = self.__client.createAlignedTimeseries(request) |
| logger.debug( |
| "creating aligned time series of device {} message: {}".format( |
| measurements_lst, status.message |
| ) |
| ) |
| |
| return Session.verify_success(status) |
| |
| def create_multi_time_series( |
| self, |
| ts_path_lst, |
| data_type_lst, |
| encoding_lst, |
| compressor_lst, |
| props_lst=None, |
| tags_lst=None, |
| attributes_lst=None, |
| alias_lst=None, |
| ): |
| """ |
| create multiple time series |
| :param ts_path_lst: List of String, complete time series paths (starts from root) |
| :param data_type_lst: List of TSDataType, data types for time series |
| :param encoding_lst: List of TSEncoding, encodings for time series |
| :param compressor_lst: List of Compressor, compressing types for time series |
| :param props_lst: List of Props Dictionary, properties for time series |
| :param tags_lst: List of tag Dictionary, tag maps for time series |
| :param attributes_lst: List of attribute Dictionary, attribute maps for time series |
| :param alias_lst: List of alias, measurement alias for time series |
| """ |
| data_type_lst = [data_type.value for data_type in data_type_lst] |
| encoding_lst = [encoding.value for encoding in encoding_lst] |
| compressor_lst = [compressor.value for compressor in compressor_lst] |
| |
| request = TSCreateMultiTimeseriesReq( |
| self.__session_id, |
| ts_path_lst, |
| data_type_lst, |
| encoding_lst, |
| compressor_lst, |
| props_lst, |
| tags_lst, |
| attributes_lst, |
| alias_lst, |
| ) |
| status = self.__client.createMultiTimeseries(request) |
| logger.debug( |
| "creating multiple time series {} message: {}".format( |
| ts_path_lst, status.message |
| ) |
| ) |
| |
| return Session.verify_success(status) |
| |
| def delete_time_series(self, paths_list): |
| """ |
| delete multiple time series, including data and schema |
| :param paths_list: List of time series path, which should be complete (starts from root) |
| """ |
| status = self.__client.deleteTimeseries(self.__session_id, paths_list) |
| logger.debug( |
| "deleting multiple time series {} message: {}".format( |
| paths_list, status.message |
| ) |
| ) |
| |
| return Session.verify_success(status) |
| |
| def check_time_series_exists(self, path): |
| """ |
| check whether a specific time series exists |
| :param path: String, complete path of time series for checking |
| :return Boolean value indicates whether it exists. |
| """ |
| data_set = self.execute_query_statement("SHOW TIMESERIES {}".format(path)) |
| result = data_set.has_next() |
| data_set.close_operation_handle() |
| return result |
| |
| def delete_data(self, paths_list, timestamp): |
| """ |
| delete all data <= time in multiple time series |
| :param paths_list: time series list that the data in. |
| :param timestamp: data with time stamp less than or equal to time will be deleted. |
| """ |
| request = TSDeleteDataReq(self.__session_id, paths_list, timestamp) |
| try: |
| status = self.__client.deleteData(request) |
| logger.debug( |
| "delete data from {}, message: {}".format(paths_list, status.message) |
| ) |
| except TTransport.TException as e: |
| logger.exception("data deletion fails because: ", e) |
| |
| def insert_str_record(self, device_id, timestamp, measurements, string_values): |
| """special case for inserting one row of String (TEXT) value""" |
| if type(string_values) == str: |
| string_values = [string_values] |
| if type(measurements) == str: |
| measurements = [measurements] |
| data_types = [TSDataType.TEXT.value for _ in string_values] |
| request = self.gen_insert_str_record_req( |
| device_id, timestamp, measurements, data_types, string_values |
| ) |
| status = self.__client.insertStringRecord(request) |
| logger.debug( |
| "insert one record to device {} message: {}".format( |
| device_id, status.message |
| ) |
| ) |
| |
| return Session.verify_success(status) |
| |
| def insert_aligned_str_record( |
| self, device_id, timestamp, measurements, string_values |
| ): |
| """special case for inserting one row of String (TEXT) value""" |
| if type(string_values) == str: |
| string_values = [string_values] |
| if type(measurements) == str: |
| measurements = [measurements] |
| data_types = [TSDataType.TEXT.value for _ in string_values] |
| request = self.gen_insert_str_record_req( |
| device_id, timestamp, measurements, data_types, string_values, True |
| ) |
| status = self.__client.insertStringRecord(request) |
| logger.debug( |
| "insert one record to device {} message: {}".format( |
| device_id, status.message |
| ) |
| ) |
| |
| return Session.verify_success(status) |
| |
| def insert_record(self, device_id, timestamp, measurements, data_types, values): |
| """ |
| insert one row of record into database, if you want improve your performance, please use insertTablet method |
| for example a record at time=10086 with three measurements is: |
| timestamp, m1, m2, m3 |
| 10086, 125.3, True, text1 |
| :param device_id: String, time series path for device |
| :param timestamp: Integer, indicate the timestamp of the row of data |
| :param measurements: List of String, sensor names |
| :param data_types: List of TSDataType, indicate the data type for each sensor |
| :param values: List, values to be inserted, for each sensor |
| """ |
| data_types = [data_type.value for data_type in data_types] |
| request = self.gen_insert_record_req( |
| device_id, timestamp, measurements, data_types, values |
| ) |
| status = self.__client.insertRecord(request) |
| logger.debug( |
| "insert one record to device {} message: {}".format( |
| device_id, status.message |
| ) |
| ) |
| |
| return Session.verify_success(status) |
| |
| def insert_records( |
| self, device_ids, times, measurements_lst, types_lst, values_lst |
| ): |
| """ |
| insert multiple rows of data, records are independent to each other, in other words, there's no relationship |
| between those records |
| :param device_ids: List of String, time series paths for device |
| :param times: List of Integer, timestamps for records |
| :param measurements_lst: 2-D List of String, each element of outer list indicates measurements of a device |
| :param types_lst: 2-D List of TSDataType, each element of outer list indicates sensor data types of a device |
| :param values_lst: 2-D List, values to be inserted, for each device |
| """ |
| type_values_lst = [] |
| for types in types_lst: |
| data_types = [data_type.value for data_type in types] |
| type_values_lst.append(data_types) |
| request = self.gen_insert_records_req( |
| device_ids, times, measurements_lst, type_values_lst, values_lst |
| ) |
| status = self.__client.insertRecords(request) |
| logger.debug( |
| "insert multiple records to devices {} message: {}".format( |
| device_ids, status.message |
| ) |
| ) |
| |
| return Session.verify_success(status) |
| |
| def insert_aligned_record( |
| self, device_id, timestamp, measurements, data_types, values |
| ): |
| """ |
| insert one row of aligned record into database, if you want improve your performance, please use insertTablet method |
| for example a record at time=10086 with three measurements is: |
| timestamp, m1, m2, m3 |
| 10086, 125.3, True, text1 |
| :param device_id: String, time series path for device |
| :param timestamp: Integer, indicate the timestamp of the row of data |
| :param measurements: List of String, sensor names |
| :param data_types: List of TSDataType, indicate the data type for each sensor |
| :param values: List, values to be inserted, for each sensor |
| """ |
| data_types = [data_type.value for data_type in data_types] |
| request = self.gen_insert_record_req( |
| device_id, timestamp, measurements, data_types, values, True |
| ) |
| status = self.__client.insertRecord(request) |
| logger.debug( |
| "insert one record to device {} message: {}".format( |
| device_id, status.message |
| ) |
| ) |
| |
| return Session.verify_success(status) |
| |
| def insert_aligned_records( |
| self, device_ids, times, measurements_lst, types_lst, values_lst |
| ): |
| """ |
| insert multiple aligned rows of data, records are independent to each other, in other words, there's no relationship |
| between those records |
| :param device_ids: List of String, time series paths for device |
| :param times: List of Integer, timestamps for records |
| :param measurements_lst: 2-D List of String, each element of outer list indicates measurements of a device |
| :param types_lst: 2-D List of TSDataType, each element of outer list indicates sensor data types of a device |
| :param values_lst: 2-D List, values to be inserted, for each device |
| """ |
| type_values_lst = [] |
| for types in types_lst: |
| data_types = [data_type.value for data_type in types] |
| type_values_lst.append(data_types) |
| request = self.gen_insert_records_req( |
| device_ids, times, measurements_lst, type_values_lst, values_lst, True |
| ) |
| status = self.__client.insertRecords(request) |
| logger.debug( |
| "insert multiple records to devices {} message: {}".format( |
| device_ids, status.message |
| ) |
| ) |
| |
| return Session.verify_success(status) |
| |
| def test_insert_record( |
| self, device_id, timestamp, measurements, data_types, values |
| ): |
| """ |
| this method NOT insert data into database and the server just return after accept the request, this method |
| should be used to test other time cost in client |
| :param device_id: String, time series path for device |
| :param timestamp: Integer, indicate the timestamp of the row of data |
| :param measurements: List of String, sensor names |
| :param data_types: List of TSDataType, indicate the data type for each sensor |
| :param values: List, values to be inserted, for each sensor |
| """ |
| data_types = [data_type.value for data_type in data_types] |
| request = self.gen_insert_record_req( |
| device_id, timestamp, measurements, data_types, values |
| ) |
| status = self.__client.testInsertRecord(request) |
| logger.debug( |
| "testing! insert one record to device {} message: {}".format( |
| device_id, status.message |
| ) |
| ) |
| |
| return Session.verify_success(status) |
| |
| def test_insert_records( |
| self, device_ids, times, measurements_lst, types_lst, values_lst |
| ): |
| """ |
| this method NOT insert data into database and the server just return after accept the request, this method |
| should be used to test other time cost in client |
| :param device_ids: List of String, time series paths for device |
| :param times: List of Integer, timestamps for records |
| :param measurements_lst: 2-D List of String, each element of outer list indicates measurements of a device |
| :param types_lst: 2-D List of TSDataType, each element of outer list indicates sensor data types of a device |
| :param values_lst: 2-D List, values to be inserted, for each device |
| """ |
| type_values_lst = [] |
| for types in types_lst: |
| data_types = [data_type.value for data_type in types] |
| type_values_lst.append(data_types) |
| request = self.gen_insert_records_req( |
| device_ids, times, measurements_lst, type_values_lst, values_lst |
| ) |
| status = self.__client.testInsertRecords(request) |
| logger.debug( |
| "testing! insert multiple records, message: {}".format(status.message) |
| ) |
| |
| return Session.verify_success(status) |
| |
| def gen_insert_record_req( |
| self, device_id, timestamp, measurements, data_types, values, is_aligned=False |
| ): |
| if (len(values) != len(data_types)) or (len(values) != len(measurements)): |
| raise RuntimeError( |
| "length of data types does not equal to length of values!" |
| ) |
| values_in_bytes = Session.value_to_bytes(data_types, values) |
| return TSInsertRecordReq( |
| self.__session_id, |
| device_id, |
| measurements, |
| values_in_bytes, |
| timestamp, |
| is_aligned, |
| ) |
| |
| def gen_insert_str_record_req( |
| self, device_id, timestamp, measurements, data_types, values, is_aligned=False |
| ): |
| if (len(values) != len(data_types)) or (len(values) != len(measurements)): |
| raise RuntimeError( |
| "length of data types does not equal to length of values!" |
| ) |
| return TSInsertStringRecordReq( |
| self.__session_id, device_id, measurements, values, timestamp, is_aligned |
| ) |
| |
| def gen_insert_records_req( |
| self, |
| device_ids, |
| times, |
| measurements_lst, |
| types_lst, |
| values_lst, |
| is_aligned=False, |
| ): |
| if ( |
| (len(device_ids) != len(measurements_lst)) |
| or (len(times) != len(types_lst)) |
| or (len(device_ids) != len(times)) |
| or (len(times) != len(values_lst)) |
| ): |
| raise RuntimeError( |
| "deviceIds, times, measurementsList and valuesList's size should be equal" |
| ) |
| |
| value_lst = [] |
| for values, data_types, measurements in zip( |
| values_lst, types_lst, measurements_lst |
| ): |
| if (len(values) != len(data_types)) or (len(values) != len(measurements)): |
| raise RuntimeError( |
| "deviceIds, times, measurementsList and valuesList's size should be equal" |
| ) |
| values_in_bytes = Session.value_to_bytes(data_types, values) |
| value_lst.append(values_in_bytes) |
| |
| return TSInsertRecordsReq( |
| self.__session_id, |
| device_ids, |
| measurements_lst, |
| value_lst, |
| times, |
| is_aligned, |
| ) |
| |
| def insert_tablet(self, tablet): |
| """ |
| insert one tablet, in a tablet, for each timestamp, the number of measurements is same |
| for example three records in the same device can form a tablet: |
| timestamps, m1, m2, m3 |
| 1, 125.3, True, text1 |
| 2, 111.6, False, text2 |
| 3, 688.6, True, text3 |
| Notice: From 0.13.0, the tablet can contain empty cell |
| The tablet itself is sorted (see docs of Tablet.py) |
| :param tablet: a tablet specified above |
| """ |
| status = self.__client.insertTablet(self.gen_insert_tablet_req(tablet)) |
| logger.debug( |
| "insert one tablet to device {} message: {}".format( |
| tablet.get_device_id(), status.message |
| ) |
| ) |
| |
| return Session.verify_success(status) |
| |
| def insert_tablets(self, tablet_lst): |
| """ |
| insert multiple tablets, tablets are independent to each other |
| :param tablet_lst: List of tablets |
| """ |
| status = self.__client.insertTablets(self.gen_insert_tablets_req(tablet_lst)) |
| logger.debug("insert multiple tablets, message: {}".format(status.message)) |
| |
| return Session.verify_success(status) |
| |
| def insert_aligned_tablet(self, tablet): |
| """ |
| insert one aligned tablet, in a tablet, for each timestamp, the number of measurements is same |
| for example three records in the same device can form a tablet: |
| timestamps, m1, m2, m3 |
| 1, 125.3, True, text1 |
| 2, 111.6, False, text2 |
| 3, 688.6, True, text3 |
| Notice: From 0.13.0, the tablet can contain empty cell |
| The tablet itself is sorted (see docs of Tablet.py) |
| :param tablet: a tablet specified above |
| """ |
| status = self.__client.insertTablet(self.gen_insert_tablet_req(tablet, True)) |
| logger.debug( |
| "insert one tablet to device {} message: {}".format( |
| tablet.get_device_id(), status.message |
| ) |
| ) |
| |
| return Session.verify_success(status) |
| |
| def insert_aligned_tablets(self, tablet_lst): |
| """ |
| insert multiple aligned tablets, tablets are independent to each other |
| :param tablet_lst: List of tablets |
| """ |
| status = self.__client.insertTablets( |
| self.gen_insert_tablets_req(tablet_lst, True) |
| ) |
| logger.debug("insert multiple tablets, message: {}".format(status.message)) |
| |
| return Session.verify_success(status) |
| |
| def insert_records_of_one_device( |
| self, device_id, times_list, measurements_list, types_list, values_list |
| ): |
| # sort by timestamp |
| sorted_zipped = sorted( |
| zip(times_list, measurements_list, types_list, values_list) |
| ) |
| result = zip(*sorted_zipped) |
| times_list, measurements_list, types_list, values_list = [ |
| list(x) for x in result |
| ] |
| |
| return self.insert_records_of_one_device_sorted( |
| device_id, times_list, measurements_list, types_list, values_list |
| ) |
| |
| def insert_records_of_one_device_sorted( |
| self, device_id, times_list, measurements_list, types_list, values_list |
| ): |
| """ |
| 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 |
| |
| :param device_id: device id |
| :param times_list: timestamps list |
| :param measurements_list: measurements list |
| :param types_list: types list |
| :param values_list: values list |
| :param have_sorted: have these list been sorted by timestamp |
| """ |
| # check parameter |
| size = len(times_list) |
| if ( |
| size != len(measurements_list) |
| or size != len(types_list) |
| or size != len(values_list) |
| ): |
| raise RuntimeError( |
| "insert records of one device error: types, times, measurementsList and valuesList's size should be equal" |
| ) |
| |
| # check sorted |
| if not Session.check_sorted(times_list): |
| raise RuntimeError( |
| "insert records of one device error: timestamp not sorted" |
| ) |
| |
| request = self.gen_insert_records_of_one_device_request( |
| device_id, times_list, measurements_list, values_list, types_list |
| ) |
| |
| # send request |
| status = self.__client.insertRecordsOfOneDevice(request) |
| logger.debug("insert records of one device, message: {}".format(status.message)) |
| |
| return Session.verify_success(status) |
| |
| def insert_aligned_records_of_one_device( |
| self, device_id, times_list, measurements_list, types_list, values_list |
| ): |
| # sort by timestamp |
| sorted_zipped = sorted( |
| zip(times_list, measurements_list, types_list, values_list) |
| ) |
| result = zip(*sorted_zipped) |
| times_list, measurements_list, types_list, values_list = [ |
| list(x) for x in result |
| ] |
| |
| return self.insert_aligned_records_of_one_device_sorted( |
| device_id, times_list, measurements_list, types_list, values_list |
| ) |
| |
| def insert_aligned_records_of_one_device_sorted( |
| self, device_id, times_list, measurements_list, types_list, values_list |
| ): |
| """ |
| Insert multiple aligned 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 to improve |
| your performance, please see insertTablet method |
| |
| :param device_id: device id |
| :param times_list: timestamps list |
| :param measurements_list: measurements list |
| :param types_list: types list |
| :param values_list: values list |
| """ |
| # check parameter |
| size = len(times_list) |
| if ( |
| size != len(measurements_list) |
| or size != len(types_list) |
| or size != len(values_list) |
| ): |
| raise RuntimeError( |
| "insert records of one device error: types, times, measurementsList and valuesList's size should be equal" |
| ) |
| |
| # check sorted |
| if not Session.check_sorted(times_list): |
| raise RuntimeError( |
| "insert records of one device error: timestamp not sorted" |
| ) |
| |
| request = self.gen_insert_records_of_one_device_request( |
| device_id, times_list, measurements_list, values_list, types_list, True |
| ) |
| |
| # send request |
| status = self.__client.insertRecordsOfOneDevice(request) |
| logger.debug("insert records of one device, message: {}".format(status.message)) |
| |
| return Session.verify_success(status) |
| |
| def gen_insert_records_of_one_device_request( |
| self, |
| device_id, |
| times_list, |
| measurements_list, |
| values_list, |
| types_list, |
| is_aligned=False, |
| ): |
| binary_value_list = [] |
| for values, data_types, measurements in zip( |
| values_list, types_list, measurements_list |
| ): |
| data_types = [data_type.value for data_type in data_types] |
| if (len(values) != len(data_types)) or (len(values) != len(measurements)): |
| raise RuntimeError( |
| "insert records of one device error: deviceIds, times, measurementsList and valuesList's size should be equal" |
| ) |
| values_in_bytes = Session.value_to_bytes(data_types, values) |
| binary_value_list.append(values_in_bytes) |
| |
| return TSInsertRecordsOfOneDeviceReq( |
| self.__session_id, |
| device_id, |
| measurements_list, |
| binary_value_list, |
| times_list, |
| is_aligned, |
| ) |
| |
| def test_insert_tablet(self, tablet): |
| """ |
| this method NOT insert data into database and the server just return after accept the request, this method |
| should be used to test other time cost in client |
| :param tablet: a tablet of data |
| """ |
| status = self.__client.testInsertTablet(self.gen_insert_tablet_req(tablet)) |
| logger.debug( |
| "testing! insert one tablet to device {} message: {}".format( |
| tablet.get_device_id(), status.message |
| ) |
| ) |
| |
| return Session.verify_success(status) |
| |
| def test_insert_tablets(self, tablet_list): |
| """ |
| this method NOT insert data into database and the server just return after accept the request, this method |
| should be used to test other time cost in client |
| :param tablet_list: List of tablets |
| """ |
| status = self.__client.testInsertTablets( |
| self.gen_insert_tablets_req(tablet_list) |
| ) |
| logger.debug( |
| "testing! insert multiple tablets, message: {}".format(status.message) |
| ) |
| |
| return Session.verify_success(status) |
| |
| def gen_insert_tablet_req(self, tablet, is_aligned=False): |
| data_type_values = [data_type.value for data_type in tablet.get_data_types()] |
| return TSInsertTabletReq( |
| self.__session_id, |
| tablet.get_device_id(), |
| tablet.get_measurements(), |
| tablet.get_binary_values(), |
| tablet.get_binary_timestamps(), |
| data_type_values, |
| tablet.get_row_number(), |
| is_aligned, |
| ) |
| |
| def gen_insert_tablets_req(self, tablet_lst, is_aligned=False): |
| device_id_lst = [] |
| measurements_lst = [] |
| values_lst = [] |
| timestamps_lst = [] |
| type_lst = [] |
| size_lst = [] |
| for tablet in tablet_lst: |
| data_type_values = [ |
| data_type.value for data_type in tablet.get_data_types() |
| ] |
| device_id_lst.append(tablet.get_device_id()) |
| measurements_lst.append(tablet.get_measurements()) |
| values_lst.append(tablet.get_binary_values()) |
| timestamps_lst.append(tablet.get_binary_timestamps()) |
| type_lst.append(data_type_values) |
| size_lst.append(tablet.get_row_number()) |
| return TSInsertTabletsReq( |
| self.__session_id, |
| device_id_lst, |
| measurements_lst, |
| values_lst, |
| timestamps_lst, |
| type_lst, |
| size_lst, |
| is_aligned, |
| ) |
| |
| def execute_query_statement(self, sql, timeout=0): |
| """ |
| execute query sql statement and returns SessionDataSet |
| :param sql: String, query sql statement |
| :return: SessionDataSet, contains query results and relevant info (see SessionDataSet.py) |
| """ |
| request = TSExecuteStatementReq( |
| self.__session_id, sql, self.__statement_id, self.__fetch_size, timeout |
| ) |
| resp = self.__client.executeQueryStatement(request) |
| return SessionDataSet( |
| sql, |
| resp.columns, |
| resp.dataTypeList, |
| resp.columnNameIndexMap, |
| resp.queryId, |
| self.__client, |
| self.__statement_id, |
| self.__session_id, |
| resp.queryDataSet, |
| resp.ignoreTimeStamp, |
| ) |
| |
| def execute_non_query_statement(self, sql): |
| """ |
| execute non-query sql statement |
| :param sql: String, non-query sql statement |
| """ |
| request = TSExecuteStatementReq(self.__session_id, sql, self.__statement_id) |
| try: |
| resp = self.__client.executeUpdateStatement(request) |
| status = resp.status |
| logger.debug( |
| "execute non-query statement {} message: {}".format(sql, status.message) |
| ) |
| return Session.verify_success(status) |
| except TTransport.TException as e: |
| raise RuntimeError("execution of non-query statement fails because: ", e) |
| |
| @staticmethod |
| def value_to_bytes(data_types, values): |
| format_str_list = [">"] |
| values_tobe_packed = [] |
| for data_type, value in zip(data_types, values): |
| if data_type == TSDataType.BOOLEAN.value: |
| format_str_list.append("c") |
| format_str_list.append("?") |
| values_tobe_packed.append(bytes([TSDataType.BOOLEAN.value])) |
| values_tobe_packed.append(value) |
| elif data_type == TSDataType.INT32.value: |
| format_str_list.append("c") |
| format_str_list.append("i") |
| values_tobe_packed.append(bytes([TSDataType.INT32.value])) |
| values_tobe_packed.append(value) |
| elif data_type == TSDataType.INT64.value: |
| format_str_list.append("c") |
| format_str_list.append("q") |
| values_tobe_packed.append(bytes([TSDataType.INT64.value])) |
| values_tobe_packed.append(value) |
| elif data_type == TSDataType.FLOAT.value: |
| format_str_list.append("c") |
| format_str_list.append("f") |
| values_tobe_packed.append(bytes([TSDataType.FLOAT.value])) |
| values_tobe_packed.append(value) |
| elif data_type == TSDataType.DOUBLE.value: |
| format_str_list.append("c") |
| format_str_list.append("d") |
| values_tobe_packed.append(bytes([TSDataType.DOUBLE.value])) |
| values_tobe_packed.append(value) |
| elif data_type == TSDataType.TEXT.value: |
| value_bytes = bytes(value, "utf-8") |
| format_str_list.append("c") |
| format_str_list.append("i") |
| format_str_list.append(str(len(value_bytes))) |
| format_str_list.append("s") |
| values_tobe_packed.append(bytes([TSDataType.TEXT.value])) |
| values_tobe_packed.append(len(value_bytes)) |
| values_tobe_packed.append(value_bytes) |
| else: |
| raise RuntimeError("Unsupported data type:" + str(data_type)) |
| format_str = "".join(format_str_list) |
| return struct.pack(format_str, *values_tobe_packed) |
| |
| def get_time_zone(self): |
| if self.__zone_id is not None: |
| return self.__zone_id |
| try: |
| resp = self.__client.getTimeZone(self.__session_id) |
| except TTransport.TException as e: |
| raise RuntimeError("Could not get time zone because: ", e) |
| return resp.timeZone |
| |
| def set_time_zone(self, zone_id): |
| request = TSSetTimeZoneReq(self.__session_id, zone_id) |
| try: |
| status = self.__client.setTimeZone(request) |
| logger.debug( |
| "setting time zone_id as {}, message: {}".format( |
| zone_id, status.message |
| ) |
| ) |
| except TTransport.TException as e: |
| raise RuntimeError("Could not set time zone because: ", e) |
| self.__zone_id = zone_id |
| |
| @staticmethod |
| def check_sorted(timestamps): |
| for i in range(1, len(timestamps)): |
| if timestamps[i] < timestamps[i - 1]: |
| return False |
| return True |
| |
| @staticmethod |
| def verify_success(status): |
| """ |
| verify success of operation |
| :param status: execution result status |
| """ |
| if status.code == Session.SUCCESS_CODE: |
| return 0 |
| |
| logger.error("error status is %s", status) |
| return -1 |
| |
| def execute_raw_data_query( |
| self, paths: list, start_time: int, end_time: int |
| ) -> SessionDataSet: |
| """ |
| execute query statement and returns SessionDataSet |
| :param paths: String path list |
| :param start_time: Query start time |
| :param end_time: Query end time |
| :return: SessionDataSet, contains query results and relevant info (see SessionDataSet.py) |
| """ |
| request = TSRawDataQueryReq( |
| self.__session_id, |
| paths, |
| self.__fetch_size, |
| startTime=start_time, |
| endTime=end_time, |
| statementId=self.__statement_id, |
| enableRedirectQuery=False, |
| ) |
| resp = self.__client.executeRawDataQuery(request) |
| return SessionDataSet( |
| "", |
| resp.columns, |
| resp.dataTypeList, |
| resp.columnNameIndexMap, |
| resp.queryId, |
| self.__client, |
| self.__statement_id, |
| self.__session_id, |
| resp.queryDataSet, |
| resp.ignoreTimeStamp, |
| ) |
| |
| def execute_last_data_query(self, paths: list, last_time: int) -> SessionDataSet: |
| """ |
| execute query statement and returns SessionDataSet |
| :param paths: String path list |
| :param last_time: Query last time |
| :return: SessionDataSet, contains query results and relevant info (see SessionDataSet.py) |
| """ |
| request = TSLastDataQueryReq( |
| self.__session_id, |
| paths, |
| self.__fetch_size, |
| last_time, |
| self.__statement_id, |
| enableRedirectQuery=False, |
| ) |
| |
| resp = self.__client.executeLastDataQuery(request) |
| return SessionDataSet( |
| "", |
| resp.columns, |
| resp.dataTypeList, |
| resp.columnNameIndexMap, |
| resp.queryId, |
| self.__client, |
| self.__statement_id, |
| self.__session_id, |
| resp.queryDataSet, |
| resp.ignoreTimeStamp, |
| ) |
| |
| def insert_string_records_of_one_device( |
| self, |
| device_id: str, |
| times: list, |
| measurements_list: list, |
| values_list: list, |
| have_sorted: bool = False, |
| ): |
| """ |
| insert multiple row of string record into database: |
| timestamp, m1, m2, m3 |
| 0, text1, text2, text3 |
| :param device_id: String, device id |
| :param times: Timestamp list |
| :param measurements_list: Measurements list |
| :param values_list: Value list |
| :param have_sorted: have these list been sorted by timestamp |
| """ |
| if (len(times) != len(measurements_list)) or (len(times) != len(values_list)): |
| raise RuntimeError( |
| "insert records of one device error: times, measurementsList and valuesList's size should be equal!" |
| ) |
| request = self.gen_insert_string_records_of_one_device_request( |
| device_id, times, measurements_list, values_list, have_sorted, False |
| ) |
| status = self.__client.insertStringRecordsOfOneDevice(request) |
| logger.debug( |
| "insert one device {} message: {}".format(device_id, status.message) |
| ) |
| |
| return Session.verify_success(status) |
| |
| def insert_aligned_string_records_of_one_device( |
| self, |
| device_id: str, |
| times: list, |
| measurements_list: list, |
| values: list, |
| have_sorted: bool = False, |
| ): |
| if (len(times) != len(measurements_list)) or (len(times) != len(values)): |
| raise RuntimeError( |
| "insert records of one device error: times, measurementsList and valuesList's size should be equal!" |
| ) |
| request = self.gen_insert_string_records_of_one_device_request( |
| device_id, times, measurements_list, values, have_sorted, True |
| ) |
| status = self.__client.insertStringRecordsOfOneDevice(request) |
| logger.debug( |
| "insert one device {} message: {}".format(device_id, status.message) |
| ) |
| |
| return Session.verify_success(status) |
| |
| def gen_insert_string_records_of_one_device_request( |
| self, |
| device_id, |
| times, |
| measurements_list, |
| values_list, |
| have_sorted, |
| is_aligned=False, |
| ): |
| if (len(times) != len(measurements_list)) or (len(times) != len(values_list)): |
| raise RuntimeError( |
| "insert records of one device error: times, measurementsList and valuesList's size should be equal!" |
| ) |
| if not Session.check_sorted(times): |
| # sort by timestamp |
| sorted_zipped = sorted(zip(times, measurements_list, values_list)) |
| result = zip(*sorted_zipped) |
| times_list, measurements_list, values_list = [list(x) for x in result] |
| request = TSInsertStringRecordsOfOneDeviceReq( |
| self.__session_id, |
| device_id, |
| measurements_list, |
| values_list, |
| times, |
| is_aligned, |
| ) |
| return request |
| |
| def create_schema_template(self, template: Template): |
| """ |
| create schema template, users using this method should use the template class as an argument |
| :param template: The template contains multiple child node(see Template.py) |
| """ |
| bytes_array = template.serialize |
| request = TSCreateSchemaTemplateReq( |
| self.__session_id, template.get_name(), bytes_array |
| ) |
| status = self.__client.createSchemaTemplate(request) |
| logger.debug( |
| "create one template {} template name: {}".format( |
| self.__session_id, template.get_name() |
| ) |
| ) |
| return Session.verify_success(status) |
| |
| def drop_schema_template(self, template_name: str): |
| """ |
| drop schema template, this method should be used to the template unset anything |
| :param template_name: template name |
| """ |
| request = TSDropSchemaTemplateReq(self.__session_id, template_name) |
| status = self.__client.dropSchemaTemplate(request) |
| logger.debug( |
| "drop one template {} template name: {}".format( |
| self.__session_id, template_name |
| ) |
| ) |
| return Session.verify_success(status) |
| |
| def execute_statement(self, sql: str, timeout=0): |
| request = TSExecuteStatementReq( |
| self.__session_id, sql, self.__statement_id, self.__fetch_size, timeout |
| ) |
| try: |
| resp = self.__client.executeStatement(request) |
| status = resp.status |
| logger.debug("execute statement {} message: {}".format(sql, status.message)) |
| if Session.verify_success(status) == 0: |
| if resp.columns: |
| return SessionDataSet( |
| sql, |
| resp.columns, |
| resp.dataTypeList, |
| resp.columnNameIndexMap, |
| resp.queryId, |
| self.__client, |
| self.__statement_id, |
| self.__session_id, |
| resp.queryDataSet, |
| resp.ignoreTimeStamp, |
| ) |
| else: |
| return None |
| else: |
| raise RuntimeError( |
| "execution of statement fails because: {}", status.message |
| ) |
| except TTransport.TException as e: |
| raise RuntimeError("execution of statement fails because: ", e) |
| |
| def add_measurements_in_template( |
| self, |
| template_name: str, |
| measurements_path: list, |
| data_types: list, |
| encodings: list, |
| compressors: list, |
| is_aligned: bool = False, |
| ): |
| """ |
| add measurements in the template, the template must already create. This function adds some measurements node. |
| :param template_name: template name, string list, like ["name_x", "name_y", "name_z"] |
| :param measurements_path: when ths is_aligned is True, recommend the name like a.b, like [python.x, python.y, iotdb.z] |
| :param data_types: using TSDataType(see IoTDBConstants.py) |
| :param encodings: using TSEncoding(see IoTDBConstants.py) |
| :param compressors: using Compressor(see IoTDBConstants.py) |
| :param is_aligned: True is aligned, False is unaligned |
| """ |
| request = TSAppendSchemaTemplateReq( |
| self.__session_id, |
| template_name, |
| is_aligned, |
| measurements_path, |
| list(map(lambda x: x.value, data_types)), |
| list(map(lambda x: x.value, encodings)), |
| list(map(lambda x: x.value, compressors)), |
| ) |
| status = self.__client.appendSchemaTemplate(request) |
| logger.debug( |
| "append unaligned template {} template name: {}".format( |
| self.__session_id, template_name |
| ) |
| ) |
| return Session.verify_success(status) |
| |
| def delete_node_in_template(self, template_name: str, path: str): |
| """ |
| delete a node in the template, this node must be already in the template |
| :param template_name: template name |
| :param path: measurements path |
| """ |
| request = TSPruneSchemaTemplateReq(self.__session_id, template_name, path) |
| status = self.__client.pruneSchemaTemplate(request) |
| logger.debug( |
| "append unaligned template {} template name: {}".format( |
| self.__session_id, template_name |
| ) |
| ) |
| return Session.verify_success(status) |
| |
| def set_schema_template(self, template_name, prefix_path): |
| """ |
| set template in prefix path, template already exit, prefix path is not measurements |
| :param template_name: template name |
| :param prefix_path: prefix path |
| """ |
| request = TSSetSchemaTemplateReq(self.__session_id, template_name, prefix_path) |
| status = self.__client.setSchemaTemplate(request) |
| logger.debug( |
| "set schema template to path{} template name: {}, path:{}".format( |
| self.__session_id, template_name, prefix_path |
| ) |
| ) |
| return Session.verify_success(status) |
| |
| def unset_schema_template(self, template_name, prefix_path): |
| """ |
| unset schema template from prefix path, this method unsetting the template from entities, |
| which have already inserted records using the template, is not supported. |
| :param template_name: template name |
| :param prefix_path: |
| """ |
| request = TSUnsetSchemaTemplateReq( |
| self.__session_id, prefix_path, template_name |
| ) |
| status = self.__client.unsetSchemaTemplate(request) |
| logger.debug( |
| "set schema template to path{} template name: {}, path:{}".format( |
| self.__session_id, template_name, prefix_path |
| ) |
| ) |
| return Session.verify_success(status) |
| |
| def count_measurements_in_template(self, template_name: str): |
| """ |
| drop schema template, this method should be used to the template unset anything |
| :param template_name: template name |
| """ |
| request = TSQueryTemplateReq( |
| self.__session_id, |
| template_name, |
| TemplateQueryType.COUNT_MEASUREMENTS.value, |
| ) |
| response = self.__client.querySchemaTemplate(request) |
| logger.debug( |
| "count measurements template {}, template name is {}, count is {}".format( |
| self.__session_id, template_name, response.measurements |
| ) |
| ) |
| return response.count |
| |
| def is_measurement_in_template(self, template_name: str, path: str): |
| """ |
| judge the node in the template is measurement or not, this node must in the template |
| :param template_name: template name |
| :param path: |
| """ |
| request = TSQueryTemplateReq( |
| self.__session_id, |
| template_name, |
| TemplateQueryType.IS_MEASUREMENT.value, |
| path, |
| ) |
| response = self.__client.querySchemaTemplate(request) |
| logger.debug( |
| "judge the path is measurement or not in template {}, template name is {}, result is {}".format( |
| self.__session_id, template_name, response.result |
| ) |
| ) |
| return response.result |
| |
| def is_path_exist_in_template(self, template_name: str, path: str): |
| """ |
| judge whether the node is a measurement or not in the template, this node must be in the template |
| :param template_name: template name |
| :param path: |
| """ |
| request = TSQueryTemplateReq( |
| self.__session_id, template_name, TemplateQueryType.PATH_EXIST.value, path |
| ) |
| response = self.__client.querySchemaTemplate(request) |
| logger.debug( |
| "judge the path is in template or not {}, template name is {}, result is {}".format( |
| self.__session_id, template_name, response.result |
| ) |
| ) |
| return response.result |
| |
| def show_measurements_in_template(self, template_name: str, pattern: str = ""): |
| """ |
| show all measurements under the pattern in template |
| :param template_name: template name |
| :param pattern: parent path, if default, show all measurements |
| """ |
| request = TSQueryTemplateReq( |
| self.__session_id, |
| template_name, |
| TemplateQueryType.SHOW_MEASUREMENTS.value, |
| pattern, |
| ) |
| response = self.__client.querySchemaTemplate(request) |
| logger.debug( |
| "show measurements in template {}, template name is {}, result is {}".format( |
| self.__session_id, template_name, response.measurements |
| ) |
| ) |
| return response.measurements |
| |
| def show_all_templates(self): |
| """ |
| show all schema templates |
| """ |
| request = TSQueryTemplateReq( |
| self.__session_id, |
| "", |
| TemplateQueryType.SHOW_TEMPLATES.value, |
| ) |
| response = self.__client.querySchemaTemplate(request) |
| logger.debug( |
| "show all template {}, measurements is {}".format( |
| self.__session_id, response.measurements |
| ) |
| ) |
| return response.measurements |
| |
| def show_paths_template_set_on(self, template_name): |
| """ |
| show the path prefix where a schema template is set |
| :param template_name: |
| """ |
| request = TSQueryTemplateReq( |
| self.__session_id, template_name, TemplateQueryType.SHOW_SET_TEMPLATES.value |
| ) |
| response = self.__client.querySchemaTemplate(request) |
| logger.debug( |
| "show paths template set {}, on {}".format( |
| self.__session_id, response.measurements |
| ) |
| ) |
| return response.measurements |
| |
| def show_paths_template_using_on(self, template_name): |
| """ |
| show the path prefix where a schema template is used |
| :param template_name: |
| """ |
| request = TSQueryTemplateReq( |
| self.__session_id, |
| template_name, |
| TemplateQueryType.SHOW_USING_TEMPLATES.value, |
| ) |
| response = self.__client.querySchemaTemplate(request) |
| logger.debug( |
| "show paths template using {}, on {}".format( |
| self.__session_id, response.measurements |
| ) |
| ) |
| return response.measurements |
