SessionPoolTest.cs

/*
 * 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.
 */

using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.Net.Sockets;
using System.Threading;
using System.Threading.Tasks;
using Apache.IoTDB.Data;
using Apache.IoTDB.DataStructure;
using ConsoleTableExt;

namespace Apache.IoTDB.Samples
{
    public partial class SessionPoolTest
    {
        public string host = "localhost";
        public int port = 6667;
        public string username = "root";
        public string password = "root";
        public List<string> nodeUrls = new();
        public int fetchSize = 500;
        public int processedSize = 4;
        public bool debug = false;
        private int poolSize = 2;
        public static string testTemplateName = "TEST_CSHARP_CLIENT_TEMPLATE_97209";
        public static string testDatabaseName = "root.TEST_CSHARP_CLIENT_GROUP_97209";
        public static string testDevice = "TEST_CSHARP_CLIENT_DEVICE";
        public static string testMeasurement = "TEST_CSHARP_CLIENT_TS";
        public static List<int> deviceCount = new List<int>() { 0, 1, 2, 3 };
        public static List<int> measurementCount = new List<int>() { 0, 1, 2, 3, 4, 5, 6 };
        public static List<string> testDevices = new List<string>(
            deviceCount.ConvertAll(x => testDevice + x.ToString()).ToArray()
        );
        public List<string> testMeasurements = new List<string>(
            measurementCount.ConvertAll(x => testMeasurement + x.ToString()).ToArray()
        );
        public SessionPoolTest(string _host = "localhost")
        {
            host = _host;
            nodeUrls.Add(host + ":" + port);
        }

        public SessionPoolTest(List<string> _nodeUrls)
        {
            nodeUrls = _nodeUrls;
        }
        public async Task Test()
        {
            if (nodeUrls.Count == 1)
            {
                await TestOpenWithNodeUrls();

                await TestOpenWith2NodeUrls();

                await TestOpenWithNodeUrlsAndInsertOneRecord();

                await TestInsertOneRecord();

                await TestInsertAlignedRecord();

                await TestInsertAlignedRecords();

                await TestInsertAlignedStringRecords();

                await TestInsertAlignedStringRecordsOfOneDevice();

                await TestInsertStringRecord();

                await TestInsertAlignedStringRecord();

                await TestInsertStringRecords();

                await TestInsertStringRecordsOfOneDevice();

                await TestInsertAlignedRecordsOfOneDevice();

                await TestInsertAlignedTablet();

                await TestInsertAlignedTablets();

                await TestInsertRecord();

                await TestCreateMultiTimeSeries();

                await TestInsertStrRecord();

                await TestInsertRecords();

                await TestInsertRecordsWithAllType();

                await TestInsertRecordsOfOneDevice();

                await TestInsertTablet();

                await TestInsertTabletWithAllType();

                await TestInsertTabletWithNullValue();

                await TestInsertTablets();

                await TestSetAndUnsetSchemaTemplate();

                await TestCreateAlignedTimeseries();

                await TestCreateAndDropSchemaTemplate();

                await TestGetTimeZone();

                await TestCreateAndDeleteDatabase();

                await TestCreateTimeSeries();

                await TestDeleteTimeSeries();

                await TestDeleteDatabase();

                await TestCheckTimeSeriesExists();

                await TestSetTimeZone();

                await TestDeleteData();

                await TestNonSql();

                await TestRawDataQuery();

                await TestLastDataQuery();

                await TestSqlQuery();

                await TestNonSqlBy_ADO();
            }
            else
            {
                await TestMultiNodeDataFetch();
            }
        }
        public async Task TestOpenWithNodeUrls()
        {
            var session_pool = new SessionPool(nodeUrls, 8);
            await session_pool.Open(false);
            Debug.Assert(session_pool.IsOpen());
            if (debug) session_pool.OpenDebugMode();
            await session_pool.Close();
            Console.WriteLine("TestOpenWithNodeUrls Passed!");
        }
        public async Task TestOpenWith2NodeUrls()
        {
            var session_pool = new SessionPool(new List<string>() { host + ":" + port, host + ":" + (port + 1) }, 8);
            await session_pool.Open(false);
            Debug.Assert(session_pool.IsOpen());
            if (debug) session_pool.OpenDebugMode();
            await session_pool.Close();

            session_pool = new SessionPool(new List<string>() { host + ":" + (port + 1), host + ":" + port }, 8);
            await session_pool.Open(false);
            Debug.Assert(session_pool.IsOpen());
            if (debug) session_pool.OpenDebugMode();
            await session_pool.Close();
            Console.WriteLine("TestOpenWith2NodeUrls Passed!");
        }
        public async Task TestOpenWithNodeUrlsAndInsertOneRecord()
        {
            var session_pool = new SessionPool(nodeUrls, 8);
            await session_pool.Open(false);
            if (debug) session_pool.OpenDebugMode();
            await session_pool.DeleteDatabaseAsync(testDatabaseName);
            var status = await session_pool.CreateTimeSeries(
                string.Format("{0}.{1}.{2}", testDatabaseName, testDevice, testMeasurements[0]),
                TSDataType.TEXT, TSEncoding.PLAIN, Compressor.SNAPPY);
            status = await session_pool.CreateTimeSeries(
                string.Format("{0}.{1}.{2}", testDatabaseName, testDevice, testMeasurements[1]),
                TSDataType.TEXT, TSEncoding.PLAIN, Compressor.SNAPPY);
            status = await session_pool.CreateTimeSeries(
                string.Format("{0}.{1}.{2}", testDatabaseName, testDevice, testMeasurements[2]),
                TSDataType.TEXT, TSEncoding.PLAIN, Compressor.SNAPPY);
            var rowRecord = new RowRecord(1668404120807, new() { "1111111", "22222", "333333" }, new() { testMeasurements[0], testMeasurements[1], testMeasurements[2] });
            status = await session_pool.InsertRecordsAsync(new List<string>() { string.Format("{0}.{1}", testDatabaseName, testDevice) }, new List<RowRecord>() { rowRecord });
            Debug.Assert(status == 0);
            Console.WriteLine("TestOpenWithNodeUrlsAndInsertOneRecord Passed!");
        }
        public async Task TestInsertOneRecord()
        {
            var session_pool = new SessionPool(host, port, 1);
            await session_pool.Open(false);
            if (debug) session_pool.OpenDebugMode();
            await session_pool.DeleteDatabaseAsync(testDatabaseName);
            var status = await session_pool.CreateTimeSeries(
                string.Format("{0}.{1}.{2}", testDatabaseName, testDevice, testMeasurements[0]),
                TSDataType.TEXT, TSEncoding.PLAIN, Compressor.SNAPPY);
            status = await session_pool.CreateTimeSeries(
                string.Format("{0}.{1}.{2}", testDatabaseName, testDevice, testMeasurements[1]),
                TSDataType.TEXT, TSEncoding.PLAIN, Compressor.SNAPPY);
            status = await session_pool.CreateTimeSeries(
                string.Format("{0}.{1}.{2}", testDatabaseName, testDevice, testMeasurements[2]),
                TSDataType.TEXT, TSEncoding.PLAIN, Compressor.SNAPPY);
            var rowRecord = new RowRecord(1668404120807, new() { "1111111", "22222", "333333" }, new() { testMeasurements[0], testMeasurements[1], testMeasurements[2] });
            status = await session_pool.InsertRecordsAsync(new List<string>() { string.Format("{0}.{1}", testDatabaseName, testDevice) }, new List<RowRecord>() { rowRecord });
        }
        public async Task TestGetTimeZone()
        {
            var session_pool = new SessionPool(host, port, poolSize);
            await session_pool.Open(false);
            if (debug) session_pool.OpenDebugMode();

            await session_pool.DeleteDatabaseAsync(testDatabaseName);
            System.Diagnostics.Debug.Assert(session_pool.IsOpen());
            var time_zone = await session_pool.GetTimeZone();
            System.Diagnostics.Debug.Assert(time_zone == "UTC+08:00");
            await session_pool.Close();
            Console.WriteLine("TestGetTimeZone Passed!");
        }



        public async Task TestCreateAndDeleteDatabase()
        {
            var session_pool = new SessionPool(host, port, poolSize);
            var status = 0;
            await session_pool.Open(false);
            if (debug) session_pool.OpenDebugMode();

            status = await session_pool.DeleteDatabaseAsync(testDatabaseName);
            System.Diagnostics.Debug.Assert(
                await session_pool.CreateDatabase(testDatabaseName) == 0);
            System.Diagnostics.Debug.Assert(
                await session_pool.DeleteDatabaseAsync(testDatabaseName) == 0);
            await session_pool.Close();
            Console.WriteLine("TestSetAndDeleteStorageGroup Passed!");
        }

        public async Task TestDeleteDatabase()
        {
            var session_pool = new SessionPool(host, port, poolSize);
            await session_pool.Open(false);
            if (debug) session_pool.OpenDebugMode();

            await session_pool.CreateDatabase(string.Format("{0}{1}", testDatabaseName, "_01"));
            await session_pool.CreateDatabase(string.Format("{0}{1}", testDatabaseName, "_02"));
            await session_pool.CreateDatabase(string.Format("{0}{1}", testDatabaseName, "_03"));
            await session_pool.CreateDatabase(string.Format("{0}{1}", testDatabaseName, "_04"));
            var database_names = new List<string>() { };
            database_names.Add(string.Format("{0}{1}", testDatabaseName, "_01"));
            database_names.Add(string.Format("{0}{1}", testDatabaseName, "_02"));
            database_names.Add(string.Format("{0}{1}", testDatabaseName, "_03"));
            database_names.Add(string.Format("{0}{1}", testDatabaseName, "_04"));
            System.Diagnostics.Debug.Assert(await session_pool.DeleteDatabasesAsync(database_names) == 0);
            await session_pool.Close();
            Console.WriteLine("TestDeleteStorageGroups Passed!");
        }


        public async Task TestSetTimeZone()
        {
            var session_pool = new SessionPool(host, port, poolSize);
            await session_pool.Open(false);
            if (debug) session_pool.OpenDebugMode();

            await session_pool.DeleteDatabaseAsync(testDatabaseName);
            System.Diagnostics.Debug.Assert(session_pool.IsOpen());
            await session_pool.SetTimeZone("GMT+8:00");
            System.Diagnostics.Debug.Assert(await session_pool.GetTimeZone() == "GMT+8:00");
            await session_pool.Close();
            Console.WriteLine("TestSetTimeZone Passed!");
        }

        public async Task TestDeleteData()
        {
            var session_pool = new SessionPool(host, port, poolSize);
            var status = 0;
            await session_pool.Open(false);
            if (debug) session_pool.OpenDebugMode();

            System.Diagnostics.Debug.Assert(session_pool.IsOpen());
            status = await session_pool.DeleteDatabaseAsync(testDatabaseName);

            status = await session_pool.CreateTimeSeries(
                string.Format("{0}.{1}.{2}", testDatabaseName, testDevice, testMeasurements[1]), TSDataType.TEXT,
                TSEncoding.PLAIN, Compressor.UNCOMPRESSED);
            System.Diagnostics.Debug.Assert(status == 0);
            status = await session_pool.CreateTimeSeries(
                string.Format("{0}.{1}.{2}", testDatabaseName, testDevice, testMeasurements[2]),
                TSDataType.BOOLEAN, TSEncoding.PLAIN, Compressor.UNCOMPRESSED);
            System.Diagnostics.Debug.Assert(status == 0);
            status = await session_pool.CreateTimeSeries(
                string.Format("{0}.{1}.{2}", testDatabaseName, testDevice, testMeasurements[3]),
                TSDataType.INT32, TSEncoding.PLAIN, Compressor.UNCOMPRESSED);
            System.Diagnostics.Debug.Assert(status == 0);

            var measures = new List<string>
            {
                testMeasurements[1], testMeasurements[2], testMeasurements[3]
            };
            var values = new List<object> { "test_text", true, (int)123 };
            status = await session_pool.InsertRecordAsync(
                string.Format("{0}.{1}", testDatabaseName, testDevice), new RowRecord(1, values, measures));
            System.Diagnostics.Debug.Assert(status == 0);
            status = await session_pool.InsertRecordAsync(
                string.Format("{0}.{1}", testDatabaseName, testDevice), new RowRecord(2, values, measures));
            System.Diagnostics.Debug.Assert(status == 0);
            status = await session_pool.InsertRecordAsync(
                string.Format("{0}.{1}", testDatabaseName, testDevice), new RowRecord(3, values, measures));
            System.Diagnostics.Debug.Assert(status == 0);
            status = await session_pool.InsertRecordAsync(
                string.Format("{0}.{1}", testDatabaseName, testDevice), new RowRecord(4, values, measures));
            System.Diagnostics.Debug.Assert(status == 0);
            var res = await session_pool.ExecuteQueryStatementAsync(
                "select * from " + string.Format("{0}.{1}", testDatabaseName, testDevice) + " where time<10");
            res.ShowTableNames();
            while (res.HasNext()) Console.WriteLine(res.Next());

            await res.Close();
            var ts_path_lst = new List<string>()
            {
                string.Format("{0}.{1}.{2}", testDatabaseName, testDevice, testMeasurements[1]),
                string.Format("{0}.{1}.{2}", testDatabaseName, testDevice, testMeasurements[2]),
            };
            await session_pool.DeleteDataAsync(ts_path_lst, 2, 3);
            res = await session_pool.ExecuteQueryStatementAsync(
                "select * from " + string.Format("{0}.{1}", testDatabaseName, testDevice) + " where time<10");
            res.ShowTableNames();
            while (res.HasNext()) Console.WriteLine(res.Next());

            await res.Close();
            status = await session_pool.DeleteDatabaseAsync(testDatabaseName);
            System.Diagnostics.Debug.Assert(status == 0);
            await session_pool.Close();
            Console.WriteLine("TestDeleteData Passed!");
        }

        public async Task TestNonSql()
        {
            var session_pool = new SessionPool(host, port, poolSize);
            var status = 0;
            await session_pool.Open(false);
            if (debug) session_pool.OpenDebugMode();

            System.Diagnostics.Debug.Assert(session_pool.IsOpen());
            status = await session_pool.DeleteDatabaseAsync(testDatabaseName);
            await session_pool.ExecuteNonQueryStatementAsync(
                "create timeseries " + string.Format("{0}.{1}", testDatabaseName, testDevice) + ".status with datatype=BOOLEAN,encoding=PLAIN");
            await session_pool.ExecuteNonQueryStatementAsync(
                "create timeseries " + string.Format("{0}.{1}", testDatabaseName, testDevice) + ".temperature with datatype=FLOAT,encoding=PLAIN");
            await session_pool.ExecuteNonQueryStatementAsync(
                "create timeseries " + string.Format("{0}.{1}", testDatabaseName, testDevice) + ".hardware with datatype=TEXT,encoding=PLAIN");
            status = await session_pool.ExecuteNonQueryStatementAsync(
                "insert into " + string.Format("{0}.{1}", testDatabaseName, testDevice) + "(timestamp, status, temperature, hardware) VALUES (4, false, 20, 'yxl')");
            System.Diagnostics.Debug.Assert(status == 0);
            await session_pool.ExecuteNonQueryStatementAsync(
                "insert into " + string.Format("{0}.{1}", testDatabaseName, testDevice) + "(timestamp, status, temperature, hardware) VALUES (5, true, 12, 'myy')");
            await session_pool.ExecuteNonQueryStatementAsync(
                "insert into " + string.Format("{0}.{1}", testDatabaseName, testDevice) + "(timestamp, status, temperature, hardware) VALUES (6, true, 21, 'lz')");
            await session_pool.ExecuteNonQueryStatementAsync(
                "insert into " + string.Format("{0}.{1}", testDatabaseName, testDevice) + "(timestamp, status, hardware) VALUES (7, true,'lz')");
            await session_pool.ExecuteNonQueryStatementAsync(
                "insert into " + string.Format("{0}.{1}", testDatabaseName, testDevice) + "(timestamp, status, hardware) VALUES (7, true,'lz')");
            var res = await session_pool.ExecuteQueryStatementAsync(
                "select * from " + string.Format("{0}.{1}", testDatabaseName, testDevice) + " where time<10");
            res.ShowTableNames();
            while (res.HasNext()) Console.WriteLine(res.Next());

            await res.Close();
            status = await session_pool.DeleteDatabaseAsync(testDatabaseName);
            System.Diagnostics.Debug.Assert(status == 0);
            await session_pool.Close();
            Console.WriteLine("TestNonSql Passed");
        }

        public async Task TestNonSqlBy_ADO()
        {
            var cnts = new IoTDB.Data.IoTDBConnectionStringBuilder();
            cnts.DataSource = host;
            cnts.TimeOut = (int)TimeSpan.FromSeconds(20).TotalMilliseconds;
            var cnt = new IoTDB.Data.IoTDBConnection(cnts.ConnectionString);
            await cnt.OpenAsync();
            var session_pool = cnt.SessionPool;
            System.Diagnostics.Debug.Assert(cnt.State == System.Data.ConnectionState.Open);
            var status = await session_pool.DeleteDatabaseAsync(testDatabaseName);
            await cnt.CreateCommand(
                 "create timeseries " + string.Format("{0}.{1}", testDatabaseName, testDevice) + ".status with datatype=BOOLEAN,encoding=PLAIN").ExecuteNonQueryAsync();
            await cnt.CreateCommand(
                "create timeseries " + string.Format("{0}.{1}", testDatabaseName, testDevice) + ".temperature with datatype=FLOAT,encoding=PLAIN").ExecuteNonQueryAsync();
            await cnt.CreateCommand(
                "create timeseries " + string.Format("{0}.{1}", testDatabaseName, testDevice) + ".hardware with datatype=TEXT,encoding=PLAIN").ExecuteNonQueryAsync();

            status = await cnt.CreateCommand(
    "insert into " + string.Format("{0}.{1}", testDatabaseName, testDevice) + "(timestamp, status, temperature, hardware) VALUES (3, false, 20, '1yxl')").ExecuteNonQueryAsync();
            status = await cnt.CreateCommand(
                "insert into " + string.Format("{0}.{1}", testDatabaseName, testDevice) + "(timestamp, status, temperature, hardware) VALUES (4, false, 20, 'yxl')").ExecuteNonQueryAsync();
            System.Diagnostics.Debug.Assert(status == 0);
            await cnt.CreateCommand(
                "insert into " + string.Format("{0}.{1}", testDatabaseName, testDevice) + "(timestamp, status, temperature, hardware) VALUES (5, true, 12, 'myy')").ExecuteNonQueryAsync();
            await cnt.CreateCommand(
                "insert into " + string.Format("{0}.{1}", testDatabaseName, testDevice) + "(timestamp, status, temperature, hardware) VALUES (6, true, 21, 'lz')").ExecuteNonQueryAsync();
            await cnt.CreateCommand(
                "insert into " + string.Format("{0}.{1}", testDatabaseName, testDevice) + "(timestamp, status, hardware) VALUES (7, true,'lz')").ExecuteNonQueryAsync();
            await cnt.CreateCommand(
                "insert into " + string.Format("{0}.{1}", testDatabaseName, testDevice) + "(timestamp, status, hardware) VALUES (7, true,'lz')").ExecuteNonQueryAsync();
            var reader = await cnt.CreateCommand(
                "select * from " + string.Format("{0}.{1}", testDatabaseName, testDevice) + " where time<10").ExecuteReaderAsync();
            ConsoleTableBuilder.From(reader.ToDataTable()).WithFormatter(0, fc => $"{fc:yyyy-MM-dd HH:mm:ss.fff}").WithFormat(ConsoleTableBuilderFormat.Default).ExportAndWriteLine();
            status = await session_pool.DeleteDatabaseAsync(testDatabaseName);
            await cnt.CloseAsync();

            System.Diagnostics.Debug.Assert(status == 0);

            Console.WriteLine("TestNonSqlBy_ADO Passed");
        }

        public async Task TestSqlQuery()
        {
            var session_pool = new SessionPool(host, port, poolSize);
            var status = 0;
            await session_pool.Open(false);
            if (debug) session_pool.OpenDebugMode();

            System.Diagnostics.Debug.Assert(session_pool.IsOpen());
            status = await session_pool.DeleteDatabaseAsync(testDatabaseName);
            await session_pool.ExecuteNonQueryStatementAsync(
                "create timeseries " + string.Format("{0}.{1}", testDatabaseName, testDevice) + ".status with datatype=BOOLEAN,encoding=PLAIN");
            await session_pool.ExecuteNonQueryStatementAsync(
                "create timeseries " + string.Format("{0}.{1}", testDatabaseName, testDevice) + ".temperature with datatype=FLOAT,encoding=PLAIN");
            await session_pool.ExecuteNonQueryStatementAsync(
                "create timeseries " + string.Format("{0}.{1}", testDatabaseName, testDevice) + ".hardware with datatype=TEXT,encoding=PLAIN");
            await session_pool.ExecuteNonQueryStatementAsync(
                "insert into " + string.Format("{0}.{1}", testDatabaseName, testDevice) + "(timestamp, status, temperature, hardware) VALUES (4, false, 20, 'yxl')");
            await session_pool.ExecuteNonQueryStatementAsync(
                "insert into " + string.Format("{0}.{1}", testDatabaseName, testDevice) + "(timestamp, status, temperature, hardware) VALUES (5, true, 12, 'myy')");
            await session_pool.ExecuteNonQueryStatementAsync(
                "insert into " + string.Format("{0}.{1}", testDatabaseName, testDevice) + "(timestamp, status, temperature, hardware) VALUES (6, true, 21, 'lz')");
            await session_pool.ExecuteNonQueryStatementAsync(
                "insert into " + string.Format("{0}.{1}", testDatabaseName, testDevice) + "(timestamp, status, hardware) VALUES (7, true,'lz')");
            await session_pool.ExecuteNonQueryStatementAsync(
                "insert into " + string.Format("{0}.{1}", testDatabaseName, testDevice) + "(timestamp, status, hardware) VALUES (7, true,'lz')");

            var res = await session_pool.ExecuteQueryStatementAsync("show timeseries root");
            res.ShowTableNames();
            while (res.HasNext()) Console.WriteLine(res.Next());

            await res.Close();
            Console.WriteLine("SHOW TIMESERIES ROOT sql passed!");
            res = await session_pool.ExecuteQueryStatementAsync("show devices");
            res.ShowTableNames();
            while (res.HasNext()) Console.WriteLine(res.Next());

            await res.Close();
            Console.WriteLine("SHOW DEVICES sql passed!");
            res = await session_pool.ExecuteQueryStatementAsync($"COUNT TIMESERIES {testDatabaseName}");
            res.ShowTableNames();
            while (res.HasNext()) Console.WriteLine(res.Next());

            await res.Close();
            Console.WriteLine("COUNT TIMESERIES root sql Passed");
            res = await session_pool.ExecuteQueryStatementAsync("select * from root.ln.wf01 where time<10");
            res.ShowTableNames();
            while (res.HasNext()) Console.WriteLine(res.Next());

            await res.Close();
            Console.WriteLine("SELECT sql Passed");
            res = await session_pool.ExecuteQueryStatementAsync(
                "select * from " + string.Format("{0}.{1}", testDatabaseName, testDevice) + " where time<10");
            res.ShowTableNames();
            while (res.HasNext()) Console.WriteLine(res.Next());

            await res.Close();
            status = await session_pool.DeleteDatabaseAsync(testDatabaseName);
            System.Diagnostics.Debug.Assert(status == 0);
            await session_pool.Close();
            Console.WriteLine("SELECT sql Passed");
        }
        public async Task TestRawDataQuery()
        {
            var session_pool = new SessionPool(host, port, poolSize);
            var status = 0;
            await session_pool.Open(false);
            if (debug) session_pool.OpenDebugMode();

            System.Diagnostics.Debug.Assert(session_pool.IsOpen());
            status = await session_pool.DeleteDatabaseAsync(testDatabaseName);

            var device_id = string.Format("{0}.{1}", testDatabaseName, testDevice);
            var measurements = new List<string> { testMeasurements[0], testMeasurements[1] };
            var data_type_lst = new List<TSDataType> { TSDataType.BOOLEAN, TSDataType.FLOAT };
            var encoding_lst = new List<TSEncoding> { TSEncoding.PLAIN, TSEncoding.PLAIN };
            var compressor_lst = new List<Compressor> { Compressor.SNAPPY, Compressor.SNAPPY };
            status = await session_pool.CreateAlignedTimeseriesAsync(device_id, measurements, data_type_lst, encoding_lst, compressor_lst);

            var records = new List<RowRecord>();
            var values = new List<object>() { true, 20.0f };
            var device_id_lst = new List<string>() { };
            for (int i = 1; i <= fetchSize * processedSize; i++)
            {
                var record = new RowRecord(i, values, measurements);
                records.Add(record);
                device_id_lst.Add(device_id);
            }
            status = await session_pool.InsertAlignedRecordsAsync(device_id_lst, records);
            System.Diagnostics.Debug.Assert(status == 0);

            var paths = new List<string>() { string.Format("{0}.{1}", device_id, testMeasurements[0]), string.Format("{0}.{1}", device_id, testMeasurements[1]) };

            var res = await session_pool.ExecuteRawDataQuery(paths, 10, fetchSize * processedSize);
            var count = 0;
            while (res.HasNext())
            {
                var record = res.Next();
                count++;
            }
            Console.WriteLine(count + " " + (fetchSize * processedSize - 10));
            System.Diagnostics.Debug.Assert(count == fetchSize * processedSize - 10);
            await res.Close();

            status = await session_pool.DeleteDatabaseAsync(testDatabaseName);
            System.Diagnostics.Debug.Assert(status == 0);
            await session_pool.Close();
            Console.WriteLine("RawDataQuery Passed");
        }
        public async Task TestLastDataQuery()
        {
            var session_pool = new SessionPool(host, port, poolSize);
            var status = 0;
            await session_pool.Open(false);
            if (debug) session_pool.OpenDebugMode();

            System.Diagnostics.Debug.Assert(session_pool.IsOpen());
            status = await session_pool.DeleteDatabaseAsync(testDatabaseName);

            var device_id = string.Format("{0}.{1}", testDatabaseName, testDevice);
            var measurements = new List<string> { testMeasurements[0], testMeasurements[1] };
            var data_type_lst = new List<TSDataType> { TSDataType.BOOLEAN, TSDataType.FLOAT };
            var encoding_lst = new List<TSEncoding> { TSEncoding.PLAIN, TSEncoding.PLAIN };
            var compressor_lst = new List<Compressor> { Compressor.SNAPPY, Compressor.SNAPPY };
            status = await session_pool.CreateAlignedTimeseriesAsync(device_id, measurements, data_type_lst, encoding_lst, compressor_lst);

            var records = new List<RowRecord>();
            var values = new List<object>() { true, 20.0f };
            var device_id_lst = new List<string>() { };
            for (int i = 1; i <= fetchSize * processedSize; i++)
            {
                var record = new RowRecord(i, values, measurements);
                records.Add(record);
                device_id_lst.Add(device_id);
            }
            status = await session_pool.InsertAlignedRecordsAsync(device_id_lst, records);
            System.Diagnostics.Debug.Assert(status == 0);

            var paths = new List<string>() { string.Format("{0}.{1}", device_id, testMeasurements[0]), string.Format("{0}.{1}", device_id, testMeasurements[1]) };

            var res = await session_pool.ExecuteLastDataQueryAsync(paths, fetchSize * processedSize - 10);
            var count = 0;
            while (res.HasNext())
            {
                var record = res.Next();
                Console.WriteLine(record);
                count++;
            }
            Console.WriteLine(count + " " + (fetchSize * processedSize - 10));
            System.Diagnostics.Debug.Assert(count == 2);
            await res.Close();

            status = await session_pool.DeleteDatabaseAsync(testDatabaseName);
            System.Diagnostics.Debug.Assert(status == 0);
            await session_pool.Close();
            Console.WriteLine("LastDataQuery Passed");
        }

        public async Task TestMultiNodeDataFetch()
        {
            System.Diagnostics.Debug.Assert(nodeUrls.Count > 1, "nodeUrls.Count should be greater than 1 in MultiNode Test");
            var session_pool = new SessionPool.Builder()
                .SetUsername(username)
                .SetPassword(password)
                .SetNodeUrl(nodeUrls)
                .SetPoolSize(4)
                .Build();
            await session_pool.Open(false);
            if (debug) session_pool.OpenDebugMode();
            var status = await session_pool.DeleteDatabaseAsync(testDatabaseName);
            var device_id = string.Format("{0}.{1}", testDatabaseName, testDevice);
            var measurements = new List<string> { testMeasurements[0], testMeasurements[1] };
            var data_type_lst = new List<TSDataType> { TSDataType.BOOLEAN, TSDataType.FLOAT };
            var encoding_lst = new List<TSEncoding> { TSEncoding.PLAIN, TSEncoding.PLAIN };
            var compressor_lst = new List<Compressor> { Compressor.SNAPPY, Compressor.SNAPPY };
            var ts_path_lst = new List<string>() {
                string.Format("{0}.{1}.{2}", testDatabaseName, testDevice, testMeasurements[0]),
                string.Format("{0}.{1}.{2}", testDatabaseName, testDevice, testMeasurements[1])
            };
            status = await session_pool.CreateMultiTimeSeriesAsync(ts_path_lst, data_type_lst, encoding_lst, compressor_lst);

            var records = new List<RowRecord>();
            var values = new List<object>() { true, 20.0f };
            var device_id_lst = new List<string>() { };
            for (int i = 1; i <= fetchSize * processedSize * 4 + 783; i++)
            {
                var record = new RowRecord(i, values, measurements);
                records.Add(record);
                device_id_lst.Add(device_id);
            }

            // insert data
            status = await session_pool.InsertRecordsAsync(device_id_lst, records);
            System.Diagnostics.Debug.Assert(status == 0);
            // fetch data
            var paths = new List<string>() { string.Format("{0}.{1}", device_id, testMeasurements[0]), string.Format("{0}.{1}", device_id, testMeasurements[1]) };
            var res = await session_pool.ExecuteQueryStatementAsync("select * from " + string.Format("{0}.{1}", testDatabaseName, testDevice));
            res.ShowTableNames();
            var count = 0;
            while (res.HasNext())
            {
                var record = res.Next();
                count++;
            }
            Console.WriteLine(count + " " + (fetchSize * processedSize * 4 + 783));
            System.Diagnostics.Debug.Assert(count == fetchSize * processedSize * 4 + 783);
            await res.Close();
            Console.WriteLine("MultiNodeDataFetch Passed");
        }
    }
}