| # 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. | |
| # | |
| # Uncomment the following line to use apache-iotdb module installed by pip3 | |
| import numpy as np | |
| from iotdb.Session import Session | |
| from iotdb.SessionPool import PoolConfig, create_session_pool | |
| from iotdb.utils.BitMap import BitMap | |
| from iotdb.utils.IoTDBConstants import TSDataType, TSEncoding, Compressor | |
| from iotdb.utils.NumpyTablet import NumpyTablet | |
| from iotdb.utils.Tablet import Tablet | |
| from .iotdb_container import IoTDBContainer | |
| # whether the test has passed | |
| final_flag = True | |
| failed_count = 0 | |
| def test_fail(): | |
| global failed_count | |
| global final_flag | |
| final_flag = False | |
| failed_count += 1 | |
| def print_message(message): | |
| print("*********") | |
| print(message) | |
| print("*********") | |
| assert False | |
| def test_session(): | |
| session_test() | |
| def test_session_pool(): | |
| session_test(True) | |
| def session_test(use_session_pool=False): | |
| with IoTDBContainer("iotdb:dev") as db: | |
| db: IoTDBContainer | |
| if use_session_pool: | |
| pool_config = PoolConfig( | |
| db.get_container_host_ip(), | |
| db.get_exposed_port(6667), | |
| "root", | |
| "root", | |
| None, | |
| 1024, | |
| "Asia/Shanghai", | |
| 3, | |
| ) | |
| session_pool = create_session_pool(pool_config, 1, 3000) | |
| session = session_pool.get_session() | |
| else: | |
| session = Session(db.get_container_host_ip(), db.get_exposed_port(6667)) | |
| session.open(False) | |
| if not session.is_open(): | |
| print("can't open session") | |
| exit(1) | |
| # set and delete databases | |
| session.set_storage_group("root.sg_test_01") | |
| session.set_storage_group("root.sg_test_02") | |
| session.set_storage_group("root.sg_test_03") | |
| session.set_storage_group("root.sg_test_04") | |
| try: | |
| session.delete_storage_group("root.sg_test_02") | |
| except Exception: | |
| test_fail() | |
| print_message("delete database failed") | |
| try: | |
| session.delete_storage_groups(["root.sg_test_03", "root.sg_test_04"]) | |
| except Exception: | |
| test_fail() | |
| print_message("delete databases failed") | |
| # setting time series. | |
| session.create_time_series( | |
| "root.sg_test_01.d_01.s_01", | |
| TSDataType.BOOLEAN, | |
| TSEncoding.PLAIN, | |
| Compressor.SNAPPY, | |
| ) | |
| session.create_time_series( | |
| "root.sg_test_01.d_01.s_02", | |
| TSDataType.INT32, | |
| TSEncoding.PLAIN, | |
| Compressor.SNAPPY, | |
| ) | |
| session.create_time_series( | |
| "root.sg_test_01.d_01.s_03", | |
| TSDataType.INT64, | |
| TSEncoding.PLAIN, | |
| Compressor.SNAPPY, | |
| ) | |
| session.create_time_series( | |
| "root.sg_test_01.d_02.s_01", | |
| TSDataType.BOOLEAN, | |
| TSEncoding.PLAIN, | |
| Compressor.SNAPPY, | |
| None, | |
| {"tag1": "v1"}, | |
| {"description": "v1"}, | |
| "temperature", | |
| ) | |
| # setting multiple time series once. | |
| ts_path_lst_ = [ | |
| "root.sg_test_01.d_01.s_04", | |
| "root.sg_test_01.d_01.s_05", | |
| "root.sg_test_01.d_01.s_06", | |
| "root.sg_test_01.d_01.s_07", | |
| "root.sg_test_01.d_01.s_08", | |
| "root.sg_test_01.d_01.s_09", | |
| ] | |
| data_type_lst_ = [ | |
| TSDataType.FLOAT, | |
| TSDataType.DOUBLE, | |
| TSDataType.TEXT, | |
| TSDataType.FLOAT, | |
| TSDataType.DOUBLE, | |
| TSDataType.TEXT, | |
| ] | |
| encoding_lst_ = [TSEncoding.PLAIN for _ in range(len(data_type_lst_))] | |
| compressor_lst_ = [Compressor.SNAPPY for _ in range(len(data_type_lst_))] | |
| session.create_multi_time_series( | |
| ts_path_lst_, data_type_lst_, encoding_lst_, compressor_lst_ | |
| ) | |
| ts_path_lst_ = [ | |
| "root.sg_test_01.d_02.s_04", | |
| "root.sg_test_01.d_02.s_05", | |
| "root.sg_test_01.d_02.s_06", | |
| "root.sg_test_01.d_02.s_07", | |
| "root.sg_test_01.d_02.s_08", | |
| "root.sg_test_01.d_02.s_09", | |
| ] | |
| data_type_lst_ = [ | |
| TSDataType.FLOAT, | |
| TSDataType.DOUBLE, | |
| TSDataType.TEXT, | |
| TSDataType.FLOAT, | |
| TSDataType.DOUBLE, | |
| TSDataType.TEXT, | |
| ] | |
| encoding_lst_ = [TSEncoding.PLAIN for _ in range(len(data_type_lst_))] | |
| compressor_lst_ = [Compressor.SNAPPY for _ in range(len(data_type_lst_))] | |
| tags_lst_ = [{"tag2": "v2"} for _ in range(len(data_type_lst_))] | |
| attributes_lst_ = [{"description": "v2"} for _ in range(len(data_type_lst_))] | |
| session.create_multi_time_series( | |
| ts_path_lst_, | |
| data_type_lst_, | |
| encoding_lst_, | |
| compressor_lst_, | |
| None, | |
| tags_lst_, | |
| attributes_lst_, | |
| None, | |
| ) | |
| # delete time series | |
| try: | |
| session.delete_time_series( | |
| [ | |
| "root.sg_test_01.d_01.s_07", | |
| "root.sg_test_01.d_01.s_08", | |
| "root.sg_test_01.d_01.s_09", | |
| ] | |
| ) | |
| except Exception: | |
| test_fail() | |
| print_message("delete time series failed") | |
| # checking time series | |
| # s_07 expecting False | |
| if session.check_time_series_exists("root.sg_test_01.d_01.s_07"): | |
| test_fail() | |
| print_message("root.sg_test_01.d_01.s_07 shouldn't exist") | |
| # s_03 expecting True | |
| if not session.check_time_series_exists("root.sg_test_01.d_01.s_03"): | |
| test_fail() | |
| print_message("root.sg_test_01.d_01.s_03 should exist") | |
| # d_02.s_01 expecting True | |
| if not session.check_time_series_exists("root.sg_test_01.d_02.s_01"): | |
| test_fail() | |
| print_message("root.sg_test_01.d_02.s_01 should exist") | |
| # d_02.s_06 expecting True | |
| if not session.check_time_series_exists("root.sg_test_01.d_02.s_06"): | |
| test_fail() | |
| print_message("root.sg_test_01.d_02.s_06 should exist") | |
| # insert one record into the database. | |
| measurements_ = ["s_01", "s_02", "s_03", "s_04", "s_05", "s_06"] | |
| values_ = [False, 10, 11, 1.1, 10011.1, "test_record"] | |
| data_types_ = [ | |
| TSDataType.BOOLEAN, | |
| TSDataType.INT32, | |
| TSDataType.INT64, | |
| TSDataType.FLOAT, | |
| TSDataType.DOUBLE, | |
| TSDataType.TEXT, | |
| ] | |
| try: | |
| session.insert_record( | |
| "root.sg_test_01.d_01", 1, measurements_, data_types_, values_ | |
| ) | |
| except Exception: | |
| test_fail() | |
| print_message("insert record failed") | |
| # insert one record with none into the database. | |
| measurements_ = ["s_01", "s_02", "s_03", "s_04", "s_05", "s_06"] | |
| values_ = [False, 10, 11, None, None, None] | |
| data_types_ = [ | |
| TSDataType.BOOLEAN, | |
| TSDataType.INT32, | |
| TSDataType.INT64, | |
| TSDataType.FLOAT, | |
| TSDataType.DOUBLE, | |
| TSDataType.TEXT, | |
| ] | |
| try: | |
| session.insert_record( | |
| "root.sg_test_01.d_01", 1, measurements_, data_types_, values_ | |
| ) | |
| except Exception: | |
| test_fail() | |
| print_message("insert record failed") | |
| # insert multiple records into database | |
| measurements_list_ = [ | |
| ["s_01", "s_02", "s_03", "s_04", "s_05", "s_06"], | |
| ["s_01", "s_02", "s_03", "s_04", "s_05", "s_06"], | |
| ] | |
| values_list_ = [ | |
| [False, 22, 33, 4.4, 55.1, "test_records01"], | |
| [True, 77, 88, 1.25, 8.125, bytes("test_records02", "utf-8")], | |
| ] | |
| data_type_list_ = [data_types_, data_types_] | |
| device_ids_ = ["root.sg_test_01.d_01", "root.sg_test_01.d_02"] | |
| try: | |
| session.insert_records( | |
| device_ids_, [2, 3], measurements_list_, data_type_list_, values_list_ | |
| ) | |
| except Exception: | |
| test_fail() | |
| print_message("insert records failed") | |
| # insert multiple records with none into database | |
| measurements_list_ = [ | |
| ["s_01", "s_02", "s_03", "s_04", "s_05", "s_06"], | |
| ["s_01", "s_02", "s_03", "s_04", "s_05", "s_06"], | |
| ] | |
| values_list_ = [ | |
| [False, 22, 33, 4.4, 55.1, "test_records01"], | |
| [None, None, None, None, 8.125, bytes("test_records02", "utf-8")], | |
| ] | |
| data_type_list_ = [data_types_, data_types_] | |
| device_ids_ = ["root.sg_test_01.d_01", "root.sg_test_01.d_02"] | |
| try: | |
| session.insert_records( | |
| device_ids_, [2, 3], measurements_list_, data_type_list_, values_list_ | |
| ) | |
| except Exception: | |
| test_fail() | |
| print_message("insert records failed") | |
| values_list_ = [ | |
| [False, 22, 33, 4.4, 55.1, "test_records01"], | |
| [None, None, None, None, None, None], | |
| ] | |
| data_type_list_ = [data_types_, data_types_] | |
| device_ids_ = ["root.sg_test_01.d_01", "root.sg_test_01.d_02"] | |
| try: | |
| session.insert_records( | |
| device_ids_, [2, 3], measurements_list_, data_type_list_, values_list_ | |
| ) | |
| except Exception: | |
| test_fail() | |
| print_message("insert records failed") | |
| # insert one tablet into the database. | |
| values_ = [ | |
| [False, 10, 11, 1.1, 10011.1, "test01"], | |
| [True, 100, 11111, 1.25, 101.0, "test02"], | |
| [False, 100, 1, 188.1, 688.25, "test03"], | |
| [True, 0, 0, 0, 6.25, "test04"], | |
| ] # Non-ASCII text will cause error since bytes can only hold 0-128 nums. | |
| timestamps_ = [4, 5, 6, 7] | |
| tablet_ = Tablet( | |
| "root.sg_test_01.d_01", measurements_, data_types_, values_, timestamps_ | |
| ) | |
| try: | |
| session.insert_tablet(tablet_) | |
| except Exception: | |
| test_fail() | |
| print_message("insert tablet failed") | |
| # insert one numpy tablet into the database. | |
| np_values_ = [ | |
| np.array([False, True, False, True], np.dtype(">?")), | |
| np.array([10, 100, 100, 0], np.dtype(">i4")), | |
| np.array([11, 11111, 1, 0], np.dtype(">i8")), | |
| np.array([1.1, 1.25, 188.1, 0], np.dtype(">f4")), | |
| np.array([10011.1, 101.0, 688.25, 6.25], np.dtype(">f8")), | |
| np.array(["test01", "test02", "test03", "test04"]), | |
| ] | |
| np_timestamps_ = np.array([1, 2, 3, 4], np.dtype(">i8")) | |
| np_tablet_ = NumpyTablet( | |
| "root.sg_test_01.d_02", | |
| measurements_, | |
| data_types_, | |
| np_values_, | |
| np_timestamps_, | |
| ) | |
| try: | |
| session.insert_tablet(np_tablet_) | |
| except Exception: | |
| test_fail() | |
| print_message("insert numpy tablet failed") | |
| # insert multiple tablets into database | |
| tablet_01 = Tablet( | |
| "root.sg_test_01.d_01", measurements_, data_types_, values_, [8, 9, 10, 11] | |
| ) | |
| tablet_02 = Tablet( | |
| "root.sg_test_01.d_02", | |
| measurements_, | |
| data_types_, | |
| values_, | |
| [12, 13, 14, 15], | |
| ) | |
| try: | |
| session.insert_tablets([tablet_01, tablet_02]) | |
| except Exception: | |
| test_fail() | |
| print_message("insert tablets failed") | |
| # insert one tablet with empty cells into the database. | |
| values_ = [ | |
| [None, 10, 11, 1.1, 10011.1, "test01"], | |
| [True, None, 11111, 1.25, 101.0, "test02"], | |
| [False, 100, 1, None, 688.25, "test03"], | |
| [True, 0, 0, 0, None, None], | |
| ] # Non-ASCII text will cause error since bytes can only hold 0-128 nums. | |
| timestamps_ = [20, 21, 22, 23] | |
| tablet_ = Tablet( | |
| "root.sg_test_01.d_01", measurements_, data_types_, values_, timestamps_ | |
| ) | |
| try: | |
| session.insert_tablet(tablet_) | |
| except Exception: | |
| test_fail() | |
| print_message("insert tablet with empty cells failed") | |
| # insert one numpy tablet with empty cells into the database. | |
| np_values_ = [ | |
| np.array([False, True, False, True], np.dtype(">?")), | |
| np.array([10, 0, 100, 0], np.dtype(">i4")), | |
| np.array([11, 11111, 0, 0], np.dtype(">i8")), | |
| np.array([1.1, 1.25, 188.1, 0], np.dtype(">f4")), | |
| np.array([10011.1, 101.0, 688.25, 0], np.dtype(">f8")), | |
| np.array(["test01", "test02", "test03", ""]), | |
| ] | |
| np_timestamps_ = np.array([30, 31, 32, 33], np.dtype(">i8")) | |
| np_bitmaps_ = [] | |
| for _ in range(len(measurements_)): | |
| np_bitmaps_.append(BitMap(len(np_timestamps_))) | |
| np_bitmaps_[0].mark(0) | |
| np_bitmaps_[1].mark(1) | |
| np_bitmaps_[2].mark(2) | |
| np_bitmaps_[4].mark(3) | |
| np_bitmaps_[5].mark(3) | |
| np_tablet_ = NumpyTablet( | |
| "root.sg_test_01.d_01", | |
| measurements_, | |
| data_types_, | |
| np_values_, | |
| np_timestamps_, | |
| np_bitmaps_, | |
| ) | |
| try: | |
| session.insert_tablet(np_tablet_) | |
| except Exception: | |
| test_fail() | |
| print_message("insert numpy tablet with empty cells failed") | |
| # insert records of one device | |
| time_list = [1, 2, 3] | |
| measurements_list = [ | |
| ["s_01", "s_02", "s_03"], | |
| ["s_01", "s_02", "s_03"], | |
| ["s_01", "s_02", "s_03"], | |
| ] | |
| data_types_list = [ | |
| [TSDataType.BOOLEAN, TSDataType.INT32, TSDataType.INT64], | |
| [TSDataType.BOOLEAN, TSDataType.INT32, TSDataType.INT64], | |
| [TSDataType.BOOLEAN, TSDataType.INT32, TSDataType.INT64], | |
| ] | |
| values_list = [[False, 22, 33], [True, 1, 23], [False, 15, 26]] | |
| try: | |
| session.insert_records_of_one_device( | |
| "root.sg_test_01.d_01", | |
| time_list, | |
| measurements_list, | |
| data_types_list, | |
| values_list, | |
| ) | |
| except Exception: | |
| test_fail() | |
| print_message("insert records of one device failed") | |
| # execute non-query sql statement | |
| try: | |
| session.execute_non_query_statement( | |
| "insert into root.sg_test_01.d_01(timestamp, s_02) values(16, 188)" | |
| ) | |
| except Exception: | |
| test_fail() | |
| print_message( | |
| "execute 'insert into root.sg_test_01.d_01(timestamp, s_02) values(16, 188)' failed" | |
| ) | |
| # execute sql query statement | |
| session_data_set = session.execute_query_statement( | |
| "select * from root.sg_test_01.d_01" | |
| ) | |
| session_data_set.set_fetch_size(1024) | |
| expect_count = 20 | |
| actual_count = 0 | |
| while session_data_set.has_next(): | |
| print(session_data_set.next()) | |
| actual_count += 1 | |
| session_data_set.close_operation_handle() | |
| if actual_count != expect_count: | |
| test_fail() | |
| print_message( | |
| "query count mismatch: expect count: " | |
| + str(expect_count) | |
| + " actual count: " | |
| + str(actual_count) | |
| ) | |
| # close session connection. | |
| session.close() | |
| if final_flag: | |
| print("All executions done!!") | |
| else: | |
| print("Some test failed, please have a check") | |
| print("failed count: ", failed_count) | |
| exit(1) |