Fix SessionDataSet sync-over-async deadlock and migrate to 4-arg RowRecord constructor - Add HasNextAsync() and FetchResultsAsync() to SessionDataSet to avoid sync-over-async deadlock in FetchResults() - Mark HasNext() as [Obsolete] with guidance to use HasNextAsync() - Update ConstructOneRow() to use 4-arg RowRecord constructor with dataTypes - Fix IoTDBCommand.BindParameters to use 4-arg RowRecord constructor - Add #pragma warning suppress in IoTDBDataReader for required sync calls - Update SessionPool.CheckTimeSeriesExistsAsync to use HasNextAsync() - Migrate all sample code to use await HasNextAsync() and 4-arg RowRecord
diff --git a/Apache-IoTDB-Client-CSharp-UserCase/Apache-IoTDB-Client-CSharp-UserCase.csproj b/Apache-IoTDB-Client-CSharp-UserCase/Apache-IoTDB-Client-CSharp-UserCase.csproj index a6b222f..01a0dd6 100644 --- a/Apache-IoTDB-Client-CSharp-UserCase/Apache-IoTDB-Client-CSharp-UserCase.csproj +++ b/Apache-IoTDB-Client-CSharp-UserCase/Apache-IoTDB-Client-CSharp-UserCase.csproj
@@ -1,4 +1,4 @@ -<Project Sdk="Microsoft.NET.Sdk"> +<Project Sdk="Microsoft.NET.Sdk"> <PropertyGroup> <OutputType>Exe</OutputType> @@ -8,7 +8,7 @@ </PropertyGroup> <ItemGroup> - <PackageReference Include="Apache.IoTDB" Version="1.0.0.3" /> + <ProjectReference Include="..\src\Apache.IoTDB\Apache.IoTDB.csproj" /> </ItemGroup> -</Project> +</Project> \ No newline at end of file
diff --git a/Apache-IoTDB-Client-CSharp-UserCase/Program.cs b/Apache-IoTDB-Client-CSharp-UserCase/Program.cs index 345e9a7..86bea34 100644 --- a/Apache-IoTDB-Client-CSharp-UserCase/Program.cs +++ b/Apache-IoTDB-Client-CSharp-UserCase/Program.cs
@@ -19,7 +19,6 @@ using System; using System.Collections.Generic; -using System.Threading; using System.Threading.Tasks; using Apache.IoTDB; using Apache.IoTDB.DataStructure; @@ -32,10 +31,19 @@ static int port = 6667; static int pool_size = 2; + static SessionPool CreateSessionPool() + { + return new SessionPool.Builder() + .SetHost(host) + .SetPort(port) + .SetPoolSize(pool_size) + .Build(); + } + static async Task OpenAndCloseSessionPool() { - var session_pool = new SessionPool(host, port, pool_size); + var session_pool = CreateSessionPool(); await session_pool.Open(false); if (session_pool.IsOpen()) { @@ -50,7 +58,7 @@ static async Task CreateTimeseries() { - var session_pool = new SessionPool(host, port, pool_size); + var session_pool = CreateSessionPool(); await session_pool.Open(false); await session_pool.DeleteDatabaseAsync("root.ln.wf01.wt01"); @@ -63,12 +71,13 @@ static async Task InsertRecord() { - var session_pool = new SessionPool(host, port, pool_size); + var session_pool = CreateSessionPool(); await session_pool.Open(false); long timestamp = 1; var values = new List<object> { true, (double)1.1, "test" }; var measures = new List<string> { "status", "temperature", "hardware" }; - var rowRecord = new RowRecord(timestamp, values, measures); + var dataTypes = new List<TSDataType> { TSDataType.BOOLEAN, TSDataType.DOUBLE, TSDataType.TEXT }; + var rowRecord = new RowRecord(timestamp, values, measures, dataTypes); var status = await session_pool.InsertRecordAsync("root.ln.wf01.wt01", rowRecord); await session_pool.Close(); @@ -76,7 +85,7 @@ static async Task InsertTablet() { - var session_pool = new SessionPool(host, port, pool_size); + var session_pool = CreateSessionPool(); await session_pool.Open(false); var device_id = "root.ln.wf01.wt01"; var measurement_lst = new List<string> { "status", "temperature", "hardware" }; @@ -95,11 +104,11 @@ static async Task ExecuteQueryStatement() { - var session_pool = new SessionPool(host, port, pool_size); + var session_pool = CreateSessionPool(); await session_pool.Open(false); var res = await session_pool.ExecuteQueryStatementAsync("select * from root.ln.wf01.wt01"); res.ShowTableNames(); - while (res.HasNext()) + while (await res.HasNextAsync()) { Console.WriteLine(res.Next()); }
diff --git a/samples/Apache.IoTDB.Samples/SessionPoolTest.AlignedRecord.cs b/samples/Apache.IoTDB.Samples/SessionPoolTest.AlignedRecord.cs index ef65bb9..7433c17 100644 --- a/samples/Apache.IoTDB.Samples/SessionPoolTest.AlignedRecord.cs +++ b/samples/Apache.IoTDB.Samples/SessionPoolTest.AlignedRecord.cs
@@ -24,103 +24,104 @@ using Apache.IoTDB.DataStructure; namespace Apache.IoTDB.Samples { - public partial class SessionPoolTest - { - public async Task TestInsertAlignedRecord() + public partial class SessionPoolTest { - var session_pool = new SessionPool(host, port, poolSize); - int status; - await session_pool.Open(false); - if (debug) session_pool.OpenDebugMode(); + public async Task TestInsertAlignedRecord() + { + var session_pool = new SessionPool(host, port, poolSize); + int status; + await session_pool.Open(false); + if (debug) session_pool.OpenDebugMode(); - System.Diagnostics.Debug.Assert(session_pool.IsOpen()); - status = await session_pool.DeleteDatabaseAsync(testDatabaseName); + System.Diagnostics.Debug.Assert(session_pool.IsOpen()); + status = await session_pool.DeleteDatabaseAsync(testDatabaseName); - string prefixPath = string.Format("{0}.{1}", testDatabaseName, testDevice); - var measurements = new List<string> { testMeasurements[1], testMeasurements[2], testMeasurements[3] }; - var types = new List<TSDataType> { TSDataType.TEXT, TSDataType.BOOLEAN, TSDataType.INT32 }; - var encodings = new List<TSEncoding> { TSEncoding.PLAIN, TSEncoding.PLAIN, TSEncoding.PLAIN }; - var compressors = new List<Compressor> { Compressor.UNCOMPRESSED, Compressor.UNCOMPRESSED, Compressor.UNCOMPRESSED }; - //status = await session_pool.CreateAlignedTimeseriesAsync(prefixPath, measurements, types, encodings, compressors); - //System.Diagnostics.Debug.Assert(status == 0); + string prefixPath = string.Format("{0}.{1}", testDatabaseName, testDevice); + var measurements = new List<string> { testMeasurements[1], testMeasurements[2], testMeasurements[3] }; + var types = new List<TSDataType> { TSDataType.TEXT, TSDataType.BOOLEAN, TSDataType.INT32 }; + var encodings = new List<TSEncoding> { TSEncoding.PLAIN, TSEncoding.PLAIN, TSEncoding.PLAIN }; + var compressors = new List<Compressor> { Compressor.UNCOMPRESSED, Compressor.UNCOMPRESSED, Compressor.UNCOMPRESSED }; + //status = await session_pool.CreateAlignedTimeseriesAsync(prefixPath, measurements, types, encodings, compressors); + //System.Diagnostics.Debug.Assert(status == 0); - var measures = new List<string> + var measures = new List<string> { testMeasurements[1], testMeasurements[2], testMeasurements[3] }; - var values = new List<object> { "test_text", true, (int)123 }; - var tasks = new List<Task<int>>(); - var start_ms = DateTime.Now.Ticks / 10000; - for (var timestamp = 1; timestamp <= fetchSize * processedSize; timestamp++) - { - var rowRecord = new RowRecord(timestamp, values, measures); - var task = session_pool.InsertAlignedRecordAsync( - string.Format("{0}.{1}", testDatabaseName, testDevice), rowRecord); - tasks.Add(task); - } - Task.WaitAll(tasks.ToArray()); - var end_ms = DateTime.Now.Ticks / 10000; - Console.WriteLine(string.Format("total insert record time is {0}", end_ms - start_ms)); - status = await session_pool.DeleteDatabaseAsync(testDatabaseName); - await session_pool.Close(); - Console.WriteLine("TestInsertAlignedRecordAsync Passed"); - } - public async Task TestInsertAlignedStringRecord() - { - var session_pool = new SessionPool(host, port, poolSize); - var status = 0; - await session_pool.Open(false); - if (debug) session_pool.OpenDebugMode(); + var values = new List<object> { "test_text", true, (int)123 }; + var dataTypes = new List<string>() { "TEXT", "BOOLEAN", "INT32" }; + var tasks = new List<Task<int>>(); + var start_ms = DateTime.Now.Ticks / 10000; + for (var timestamp = 1; timestamp <= fetchSize * processedSize; timestamp++) + { + var rowRecord = new RowRecord(timestamp, values, measures, dataTypes); + var task = session_pool.InsertAlignedRecordAsync( + string.Format("{0}.{1}", testDatabaseName, testDevice), rowRecord); + tasks.Add(task); + } + Task.WaitAll(tasks.ToArray()); + var end_ms = DateTime.Now.Ticks / 10000; + Console.WriteLine(string.Format("total insert record time is {0}", end_ms - start_ms)); + status = await session_pool.DeleteDatabaseAsync(testDatabaseName); + await session_pool.Close(); + Console.WriteLine("TestInsertAlignedRecordAsync Passed"); + } + public async Task TestInsertAlignedStringRecord() + { + 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()); - await session_pool.DeleteDatabaseAsync(testDatabaseName); + System.Diagnostics.Debug.Assert(session_pool.IsOpen()); + await session_pool.DeleteDatabaseAsync(testDatabaseName); - status = await session_pool.CreateAlignedTimeseriesAsync( - string.Format("{0}.{1}", testDatabaseName, testDevice), - new List<string>() { testMeasurements[0], testMeasurements[1], testMeasurements[2] }, - new List<TSDataType>() { TSDataType.TEXT, TSDataType.TEXT, TSDataType.TEXT }, - new List<TSEncoding>() { TSEncoding.PLAIN, TSEncoding.PLAIN, TSEncoding.PLAIN }, - new List<Compressor>() { Compressor.UNCOMPRESSED, Compressor.UNCOMPRESSED, Compressor.UNCOMPRESSED }); + status = await session_pool.CreateAlignedTimeseriesAsync( + string.Format("{0}.{1}", testDatabaseName, testDevice), + new List<string>() { testMeasurements[0], testMeasurements[1], testMeasurements[2] }, + new List<TSDataType>() { TSDataType.TEXT, TSDataType.TEXT, TSDataType.TEXT }, + new List<TSEncoding>() { TSEncoding.PLAIN, TSEncoding.PLAIN, TSEncoding.PLAIN }, + new List<Compressor>() { Compressor.UNCOMPRESSED, Compressor.UNCOMPRESSED, Compressor.UNCOMPRESSED }); - System.Diagnostics.Debug.Assert(status == 0); - var measurements = new List<string> + System.Diagnostics.Debug.Assert(status == 0); + var measurements = new List<string> {testMeasurements[0], testMeasurements[1], testMeasurements[2]}; - var values = new List<string> { "test_text1", "test_text2", "test_text3" }; - var tasks = new List<Task<int>>(); - var start_ms = DateTime.Now.Ticks / 10000; - for (var timestamp = 1; timestamp <= fetchSize * processedSize; timestamp++) - { - var task = session_pool.InsertAlignedStringRecordAsync( - string.Format("{0}.{1}", testDatabaseName, testDevice), measurements, values, timestamp); - tasks.Add(task); - } + var values = new List<string> { "test_text1", "test_text2", "test_text3" }; + var tasks = new List<Task<int>>(); + var start_ms = DateTime.Now.Ticks / 10000; + for (var timestamp = 1; timestamp <= fetchSize * processedSize; timestamp++) + { + var task = session_pool.InsertAlignedStringRecordAsync( + string.Format("{0}.{1}", testDatabaseName, testDevice), measurements, values, timestamp); + tasks.Add(task); + } - Task.WaitAll(tasks.ToArray()); - var end_ms = DateTime.Now.Ticks / 10000; - Console.WriteLine(string.Format("total insert aligned string record time is {0}", end_ms - start_ms)); - var res = await session_pool.ExecuteQueryStatementAsync("select * from " + string.Format("{0}.{1}", testDatabaseName, testDevice)); - var res_cnt = 0; - while (res.HasNext()) - { - res.Next(); - res_cnt++; - } - Console.WriteLine(res_cnt + " " + fetchSize * processedSize); - System.Diagnostics.Debug.Assert(res_cnt == fetchSize * processedSize); - await session_pool.DeleteDatabaseAsync(testDatabaseName); - await session_pool.Close(); - Console.WriteLine("TestInsertAlignedStringRecordAsync Passed"); - } - public async Task TestInsertAlignedRecords() - { - var session_pool = new SessionPool(host, port, poolSize); - await session_pool.Open(false); - if (debug) session_pool.OpenDebugMode(); + Task.WaitAll(tasks.ToArray()); + var end_ms = DateTime.Now.Ticks / 10000; + Console.WriteLine(string.Format("total insert aligned string record time is {0}", end_ms - start_ms)); + var res = await session_pool.ExecuteQueryStatementAsync("select * from " + string.Format("{0}.{1}", testDatabaseName, testDevice)); + var res_cnt = 0; + while (await res.HasNextAsync()) + { + res.Next(); + res_cnt++; + } + Console.WriteLine(res_cnt + " " + fetchSize * processedSize); + System.Diagnostics.Debug.Assert(res_cnt == fetchSize * processedSize); + await session_pool.DeleteDatabaseAsync(testDatabaseName); + await session_pool.Close(); + Console.WriteLine("TestInsertAlignedStringRecordAsync Passed"); + } + public async Task TestInsertAlignedRecords() + { + var session_pool = new SessionPool(host, port, poolSize); + await session_pool.Open(false); + if (debug) session_pool.OpenDebugMode(); - System.Diagnostics.Debug.Assert(session_pool.IsOpen()); - var status = 0; - await session_pool.DeleteDatabaseAsync(testDatabaseName); + System.Diagnostics.Debug.Assert(session_pool.IsOpen()); + var status = 0; + await session_pool.DeleteDatabaseAsync(testDatabaseName); - string prefixPath = string.Format("{0}.{1}", testDatabaseName, testDevice); - var measurement_lst = new List<string>() + string prefixPath = string.Format("{0}.{1}", testDatabaseName, testDevice); + var measurement_lst = new List<string>() { testMeasurements[1], testMeasurements[2], @@ -129,37 +130,37 @@ testMeasurements[5], testMeasurements[6] }; - var data_type_lst = new List<TSDataType>() + var data_type_lst = new List<TSDataType>() { TSDataType.BOOLEAN, TSDataType.INT32, TSDataType.INT64, TSDataType.DOUBLE, TSDataType.FLOAT, TSDataType.TEXT }; - var encoding_lst = new List<TSEncoding>() + var encoding_lst = new List<TSEncoding>() { TSEncoding.PLAIN, TSEncoding.PLAIN, TSEncoding.PLAIN, TSEncoding.PLAIN, TSEncoding.PLAIN, TSEncoding.PLAIN }; - var compressor_lst = new List<Compressor>() + var compressor_lst = new List<Compressor>() { Compressor.SNAPPY, Compressor.SNAPPY, Compressor.SNAPPY, Compressor.SNAPPY, Compressor.SNAPPY, Compressor.SNAPPY }; - status = await session_pool.CreateAlignedTimeseriesAsync(prefixPath, measurement_lst, data_type_lst, encoding_lst, - compressor_lst); - System.Diagnostics.Debug.Assert(status == 0); + status = await session_pool.CreateAlignedTimeseriesAsync(prefixPath, measurement_lst, data_type_lst, encoding_lst, + compressor_lst); + System.Diagnostics.Debug.Assert(status == 0); - var device_id = new List<string>() { }; - for (var i = 0; i < 3; i++) device_id.Add(string.Format("{0}.{1}", testDatabaseName, testDevice)); - var measurements_lst = new List<List<string>>() { }; - measurements_lst.Add(new List<string>() { testMeasurements[1], testMeasurements[2] }); - measurements_lst.Add(new List<string>() + var device_id = new List<string>() { }; + for (var i = 0; i < 3; i++) device_id.Add(string.Format("{0}.{1}", testDatabaseName, testDevice)); + var measurements_lst = new List<List<string>>() { }; + measurements_lst.Add(new List<string>() { testMeasurements[1], testMeasurements[2] }); + measurements_lst.Add(new List<string>() { testMeasurements[1], testMeasurements[2], testMeasurements[3], testMeasurements[4] }); - measurements_lst.Add(new List<string>() + measurements_lst.Add(new List<string>() { testMeasurements[1], testMeasurements[2], @@ -168,161 +169,166 @@ testMeasurements[5], testMeasurements[6] }); - var values_lst = new List<List<object>>() { }; - values_lst.Add(new List<object>() { true, (int)123 }); - values_lst.Add(new List<object>() { true, (int)123, (long)456, (double)1.1 }); - values_lst.Add(new List<object>() + var values_lst = new List<List<object>>() { }; + values_lst.Add(new List<object>() { true, (int)123 }); + values_lst.Add(new List<object>() { true, (int)123, (long)456, (double)1.1 }); + values_lst.Add(new List<object>() {true, (int) 123, (long) 456, (double) 1.1, (float) 10001.1, "test_record"}); - var timestamp_lst = new List<long>() { 1, 2, 3 }; - var rowRecords = new List<RowRecord>() { }; - for (var i = 0; i < 3; i++) - { - var rowRecord = new RowRecord(timestamp_lst[i], values_lst[i], measurements_lst[i]); - rowRecords.Add(rowRecord); - } + var timestamp_lst = new List<long>() { 1, 2, 3 }; + var dataTypes_lst = new List<List<string>>() { }; + dataTypes_lst.Add(new List<string>() { "BOOLEAN", "INT32" }); + dataTypes_lst.Add(new List<string>() { "BOOLEAN", "INT32", "INT64", "DOUBLE" }); + dataTypes_lst.Add(new List<string>() { "BOOLEAN", "INT32", "INT64", "DOUBLE", "FLOAT", "TEXT" }); + var rowRecords = new List<RowRecord>() { }; + for (var i = 0; i < 3; i++) + { + var rowRecord = new RowRecord(timestamp_lst[i], values_lst[i], measurements_lst[i], dataTypes_lst[i]); + rowRecords.Add(rowRecord); + } - status = await session_pool.InsertAlignedRecordsAsync(device_id, rowRecords); - 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()); + status = await session_pool.InsertAlignedRecordsAsync(device_id, rowRecords); + 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 (await res.HasNextAsync()) Console.WriteLine(res.Next()); - await res.Close(); - Console.WriteLine(status); + await res.Close(); + Console.WriteLine(status); - // large data test - device_id = new List<string>() { }; - rowRecords = new List<RowRecord>() { }; - var tasks = new List<Task<int>>(); - for (var timestamp = 4; timestamp <= fetchSize * processedSize; timestamp++) - { - device_id.Add(string.Format("{0}.{1}", testDatabaseName, testDevice)); - rowRecords.Add(new RowRecord(timestamp, new List<object>() { true, (int)123 }, - new List<string>() { testMeasurements[1], testMeasurements[2] })); - if (timestamp % fetchSize == 0) - { - tasks.Add(session_pool.InsertAlignedRecordsAsync(device_id, rowRecords)); - device_id = new List<string>() { }; - rowRecords = new List<RowRecord>() { }; + // large data test + device_id = new List<string>() { }; + rowRecords = new List<RowRecord>() { }; + var tasks = new List<Task<int>>(); + var dataTypes = new List<string>() { "BOOLEAN", "INT32" }; + for (var timestamp = 4; timestamp <= fetchSize * processedSize; timestamp++) + { + device_id.Add(string.Format("{0}.{1}", testDatabaseName, testDevice)); + rowRecords.Add(new RowRecord(timestamp, new List<object>() { true, (int)123 }, + new List<string>() { testMeasurements[1], testMeasurements[2] }, dataTypes)); + if (timestamp % fetchSize == 0) + { + tasks.Add(session_pool.InsertAlignedRecordsAsync(device_id, rowRecords)); + device_id = new List<string>() { }; + rowRecords = new List<RowRecord>() { }; + } + } + + Task.WaitAll(tasks.ToArray()); + res = await session_pool.ExecuteQueryStatementAsync( + "select * from " + string.Format("{0}.{1}", testDatabaseName, testDevice)); + res.ShowTableNames(); + var record_count = fetchSize * processedSize; + var res_count = 0; + while (await res.HasNextAsync()) + { + res.Next(); + res_count += 1; + } + + await res.Close(); + Console.WriteLine(res_count + " " + fetchSize * processedSize); + System.Diagnostics.Debug.Assert(res_count == record_count); + System.Diagnostics.Debug.Assert(status == 0); + + status = await session_pool.DeleteDatabaseAsync(testDatabaseName); + System.Diagnostics.Debug.Assert(status == 0); + await session_pool.Close(); + Console.WriteLine("TestInsertAlignedRecords Passed!"); } - } - - Task.WaitAll(tasks.ToArray()); - res = await session_pool.ExecuteQueryStatementAsync( - "select * from " + string.Format("{0}.{1}", testDatabaseName, testDevice)); - res.ShowTableNames(); - var record_count = fetchSize * processedSize; - var res_count = 0; - while (res.HasNext()) - { - res.Next(); - res_count += 1; - } - - await res.Close(); - Console.WriteLine(res_count + " " + fetchSize * processedSize); - System.Diagnostics.Debug.Assert(res_count == record_count); - System.Diagnostics.Debug.Assert(status == 0); - - status = await session_pool.DeleteDatabaseAsync(testDatabaseName); - System.Diagnostics.Debug.Assert(status == 0); - await session_pool.Close(); - Console.WriteLine("TestInsertAlignedRecords Passed!"); - } - public async Task TestInsertAlignedStringRecords() - { - var session_pool = new SessionPool(host, port, poolSize); - await session_pool.Open(false); - if (debug) session_pool.OpenDebugMode(); - - System.Diagnostics.Debug.Assert(session_pool.IsOpen()); - var status = 0; - await session_pool.DeleteDatabaseAsync(testDatabaseName); - - string prefixPath = string.Format("{0}.{1}", testDatabaseName, testDevice); - var measurement_lst = new List<string>() { testMeasurements[1], testMeasurements[2] }; - var data_type_lst = new List<TSDataType>() { TSDataType.TEXT, TSDataType.TEXT }; - 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(prefixPath, measurement_lst, data_type_lst, encoding_lst, - compressor_lst); - System.Diagnostics.Debug.Assert(status == 0); - - var device_id = new List<string>() { }; - for (var i = 0; i < 3; i++) device_id.Add(string.Format("{0}.{1}", testDatabaseName, testDevice)); - var measurements_lst = new List<List<string>>() { }; - measurements_lst.Add(new List<string>() { testMeasurements[1], testMeasurements[2] }); - measurements_lst.Add(new List<string>() { testMeasurements[1], testMeasurements[2] }); - measurements_lst.Add(new List<string>() { testMeasurements[1], testMeasurements[2] }); - var values_lst = new List<List<string>>() { }; - values_lst.Add(new List<string>() { "test1", "test2" }); - values_lst.Add(new List<string>() { "test3", "test4" }); - values_lst.Add(new List<string>() { "test5", "test6" }); - List<long> timestamp_lst = new List<long>() { 1, 2, 3 }; - - status = await session_pool.InsertAlignedStringRecordsAsync(device_id, measurements_lst, values_lst, timestamp_lst); - 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(); - - // large data test - device_id = new List<string>() { }; - measurements_lst = new List<List<string>>() { }; - values_lst = new List<List<string>>() { }; - timestamp_lst = new List<long>() { }; - List<Task<int>> tasks = new List<Task<int>>(); - for (var timestamp = 4; timestamp <= fetchSize * processedSize; timestamp++) - { - device_id.Add(string.Format("{0}.{1}", testDatabaseName, testDevice)); - measurements_lst.Add(new List<string>() { testMeasurements[1], testMeasurements[2] }); - values_lst.Add(new List<string>() { "test1", "test2" }); - timestamp_lst.Add(timestamp); - if (timestamp % fetchSize == 0) + public async Task TestInsertAlignedStringRecords() { - tasks.Add(session_pool.InsertAlignedStringRecordsAsync(device_id, measurements_lst, values_lst, timestamp_lst)); - device_id = new List<string>() { }; - measurements_lst = new List<List<string>>() { }; - values_lst = new List<List<string>>() { }; - timestamp_lst = new List<long>() { }; + var session_pool = new SessionPool(host, port, poolSize); + await session_pool.Open(false); + if (debug) session_pool.OpenDebugMode(); + + System.Diagnostics.Debug.Assert(session_pool.IsOpen()); + var status = 0; + await session_pool.DeleteDatabaseAsync(testDatabaseName); + + string prefixPath = string.Format("{0}.{1}", testDatabaseName, testDevice); + var measurement_lst = new List<string>() { testMeasurements[1], testMeasurements[2] }; + var data_type_lst = new List<TSDataType>() { TSDataType.TEXT, TSDataType.TEXT }; + 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(prefixPath, measurement_lst, data_type_lst, encoding_lst, + compressor_lst); + System.Diagnostics.Debug.Assert(status == 0); + + var device_id = new List<string>() { }; + for (var i = 0; i < 3; i++) device_id.Add(string.Format("{0}.{1}", testDatabaseName, testDevice)); + var measurements_lst = new List<List<string>>() { }; + measurements_lst.Add(new List<string>() { testMeasurements[1], testMeasurements[2] }); + measurements_lst.Add(new List<string>() { testMeasurements[1], testMeasurements[2] }); + measurements_lst.Add(new List<string>() { testMeasurements[1], testMeasurements[2] }); + var values_lst = new List<List<string>>() { }; + values_lst.Add(new List<string>() { "test1", "test2" }); + values_lst.Add(new List<string>() { "test3", "test4" }); + values_lst.Add(new List<string>() { "test5", "test6" }); + List<long> timestamp_lst = new List<long>() { 1, 2, 3 }; + + status = await session_pool.InsertAlignedStringRecordsAsync(device_id, measurements_lst, values_lst, timestamp_lst); + 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 (await res.HasNextAsync()) Console.WriteLine(res.Next()); + + await res.Close(); + + // large data test + device_id = new List<string>() { }; + measurements_lst = new List<List<string>>() { }; + values_lst = new List<List<string>>() { }; + timestamp_lst = new List<long>() { }; + List<Task<int>> tasks = new List<Task<int>>(); + for (var timestamp = 4; timestamp <= fetchSize * processedSize; timestamp++) + { + device_id.Add(string.Format("{0}.{1}", testDatabaseName, testDevice)); + measurements_lst.Add(new List<string>() { testMeasurements[1], testMeasurements[2] }); + values_lst.Add(new List<string>() { "test1", "test2" }); + timestamp_lst.Add(timestamp); + if (timestamp % fetchSize == 0) + { + tasks.Add(session_pool.InsertAlignedStringRecordsAsync(device_id, measurements_lst, values_lst, timestamp_lst)); + device_id = new List<string>() { }; + measurements_lst = new List<List<string>>() { }; + values_lst = new List<List<string>>() { }; + timestamp_lst = new List<long>() { }; + } + } + + Task.WaitAll(tasks.ToArray()); + res = await session_pool.ExecuteQueryStatementAsync( + "select * from " + string.Format("{0}.{1}", testDatabaseName, testDevice)); + res.ShowTableNames(); + var res_count = 0; + while (await res.HasNextAsync()) + { + res.Next(); + res_count += 1; + } + + await res.Close(); + Console.WriteLine(res_count + " " + fetchSize * processedSize); + System.Diagnostics.Debug.Assert(res_count == fetchSize * processedSize); + status = await session_pool.DeleteDatabaseAsync(testDatabaseName); + System.Diagnostics.Debug.Assert(status == 0); + await session_pool.Close(); + Console.WriteLine("TestInsertAlignedStringRecords Passed!"); } - } + public async Task TestInsertAlignedRecordsOfOneDevice() + { + var session_pool = new SessionPool(host, port, poolSize); + await session_pool.Open(false); + if (debug) session_pool.OpenDebugMode(); - Task.WaitAll(tasks.ToArray()); - res = await session_pool.ExecuteQueryStatementAsync( - "select * from " + string.Format("{0}.{1}", testDatabaseName, testDevice)); - res.ShowTableNames(); - var res_count = 0; - while (res.HasNext()) - { - res.Next(); - res_count += 1; - } + System.Diagnostics.Debug.Assert(session_pool.IsOpen()); + var status = 0; + await session_pool.DeleteDatabaseAsync(testDatabaseName); - await res.Close(); - Console.WriteLine(res_count + " " + fetchSize * processedSize); - System.Diagnostics.Debug.Assert(res_count == fetchSize * processedSize); - status = await session_pool.DeleteDatabaseAsync(testDatabaseName); - System.Diagnostics.Debug.Assert(status == 0); - await session_pool.Close(); - Console.WriteLine("TestInsertAlignedStringRecords Passed!"); - } - public async Task TestInsertAlignedRecordsOfOneDevice() - { - var session_pool = new SessionPool(host, port, poolSize); - await session_pool.Open(false); - if (debug) session_pool.OpenDebugMode(); - - System.Diagnostics.Debug.Assert(session_pool.IsOpen()); - var status = 0; - await session_pool.DeleteDatabaseAsync(testDatabaseName); - - string prefixPath = string.Format("{0}.{1}", testDatabaseName, testDevice); - var measurement_lst = new List<string>() + string prefixPath = string.Format("{0}.{1}", testDatabaseName, testDevice); + var measurement_lst = new List<string>() { testMeasurements[1], testMeasurements[2], @@ -331,37 +337,37 @@ testMeasurements[5], testMeasurements[6] }; - var data_type_lst = new List<TSDataType>() + var data_type_lst = new List<TSDataType>() { TSDataType.BOOLEAN, TSDataType.INT32, TSDataType.INT64, TSDataType.DOUBLE, TSDataType.FLOAT, TSDataType.TEXT }; - var encoding_lst = new List<TSEncoding>() + var encoding_lst = new List<TSEncoding>() { TSEncoding.PLAIN, TSEncoding.PLAIN, TSEncoding.PLAIN, TSEncoding.PLAIN, TSEncoding.PLAIN, TSEncoding.PLAIN }; - var compressor_lst = new List<Compressor>() + var compressor_lst = new List<Compressor>() { Compressor.SNAPPY, Compressor.SNAPPY, Compressor.SNAPPY, Compressor.SNAPPY, Compressor.SNAPPY, Compressor.SNAPPY }; - status = await session_pool.CreateAlignedTimeseriesAsync(prefixPath, measurement_lst, data_type_lst, encoding_lst, - compressor_lst); - System.Diagnostics.Debug.Assert(status == 0); + status = await session_pool.CreateAlignedTimeseriesAsync(prefixPath, measurement_lst, data_type_lst, encoding_lst, + compressor_lst); + System.Diagnostics.Debug.Assert(status == 0); - var device_id = string.Format("{0}.{1}", testDatabaseName, testDevice); - var measurements_lst = new List<List<string>>() { }; - measurements_lst.Add(new List<string>() { testMeasurements[1], testMeasurements[2] }); - measurements_lst.Add(new List<string>() + var device_id = string.Format("{0}.{1}", testDatabaseName, testDevice); + var measurements_lst = new List<List<string>>() { }; + measurements_lst.Add(new List<string>() { testMeasurements[1], testMeasurements[2] }); + measurements_lst.Add(new List<string>() { testMeasurements[1], testMeasurements[2], testMeasurements[3], testMeasurements[4] }); - measurements_lst.Add(new List<string>() + measurements_lst.Add(new List<string>() { testMeasurements[1], testMeasurements[2], @@ -370,130 +376,135 @@ testMeasurements[5], testMeasurements[6] }); - var values_lst = new List<List<object>>() { }; - values_lst.Add(new List<object>() { true, (int)123 }); - values_lst.Add(new List<object>() { true, (int)123, (long)456, (double)1.1 }); - values_lst.Add(new List<object>() + var values_lst = new List<List<object>>() { }; + values_lst.Add(new List<object>() { true, (int)123 }); + values_lst.Add(new List<object>() { true, (int)123, (long)456, (double)1.1 }); + values_lst.Add(new List<object>() {true, (int) 123, (long) 456, (double) 1.1, (float) 10001.1, "test_record"}); - var timestamp_lst = new List<long>() { 1, 2, 3 }; - var rowRecords = new List<RowRecord>() { }; - for (var i = 0; i < 3; i++) - { - var rowRecord = new RowRecord(timestamp_lst[i], values_lst[i], measurements_lst[i]); - rowRecords.Add(rowRecord); - } - status = await session_pool.InsertAlignedRecordsOfOneDeviceAsync(device_id, rowRecords); - 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()); + var timestamp_lst = new List<long>() { 1, 2, 3 }; + var dataTypes_lst = new List<List<string>>() { }; + dataTypes_lst.Add(new List<string>() { "BOOLEAN", "INT32" }); + dataTypes_lst.Add(new List<string>() { "BOOLEAN", "INT32", "INT64", "DOUBLE" }); + dataTypes_lst.Add(new List<string>() { "BOOLEAN", "INT32", "INT64", "DOUBLE", "FLOAT", "TEXT" }); + var rowRecords = new List<RowRecord>() { }; + for (var i = 0; i < 3; i++) + { + var rowRecord = new RowRecord(timestamp_lst[i], values_lst[i], measurements_lst[i], dataTypes_lst[i]); + rowRecords.Add(rowRecord); + } + status = await session_pool.InsertAlignedRecordsOfOneDeviceAsync(device_id, rowRecords); + 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 (await res.HasNextAsync()) Console.WriteLine(res.Next()); - await res.Close(); - rowRecords = new List<RowRecord>() { }; - var tasks = new List<Task<int>>(); - for (var timestamp = 4; timestamp <= fetchSize * processedSize; timestamp++) - { - rowRecords.Add(new RowRecord(timestamp, new List<object>() { true, (int)123 }, - new List<string>() { testMeasurements[1], testMeasurements[2] })); - if (timestamp % fetchSize == 0) - { - tasks.Add(session_pool.InsertAlignedRecordsOfOneDeviceAsync(device_id, rowRecords)); - rowRecords = new List<RowRecord>() { }; + await res.Close(); + rowRecords = new List<RowRecord>() { }; + var tasks = new List<Task<int>>(); + var dataTypes = new List<string>() { "BOOLEAN", "INT32" }; + for (var timestamp = 4; timestamp <= fetchSize * processedSize; timestamp++) + { + rowRecords.Add(new RowRecord(timestamp, new List<object>() { true, (int)123 }, + new List<string>() { testMeasurements[1], testMeasurements[2] }, dataTypes)); + if (timestamp % fetchSize == 0) + { + tasks.Add(session_pool.InsertAlignedRecordsOfOneDeviceAsync(device_id, rowRecords)); + rowRecords = new List<RowRecord>() { }; + } + } + + Task.WaitAll(tasks.ToArray()); + res = await session_pool.ExecuteQueryStatementAsync( + "select * from " + string.Format("{0}.{1}", testDatabaseName, testDevice)); + var res_count = 0; + while (await res.HasNextAsync()) + { + res.Next(); + res_count += 1; + } + + await res.Close(); + Console.WriteLine(res_count + " " + fetchSize * processedSize); + System.Diagnostics.Debug.Assert(res_count == fetchSize * processedSize); + status = await session_pool.DeleteDatabaseAsync(testDatabaseName); + System.Diagnostics.Debug.Assert(status == 0); + await session_pool.Close(); + Console.WriteLine("TestInsertAlignedRecordsOfOneDevice Passed!"); } - } - - Task.WaitAll(tasks.ToArray()); - res = await session_pool.ExecuteQueryStatementAsync( - "select * from " + string.Format("{0}.{1}", testDatabaseName, testDevice)); - var res_count = 0; - while (res.HasNext()) - { - res.Next(); - res_count += 1; - } - - await res.Close(); - Console.WriteLine(res_count + " " + fetchSize * processedSize); - System.Diagnostics.Debug.Assert(res_count == fetchSize * processedSize); - status = await session_pool.DeleteDatabaseAsync(testDatabaseName); - System.Diagnostics.Debug.Assert(status == 0); - await session_pool.Close(); - Console.WriteLine("TestInsertAlignedRecordsOfOneDevice Passed!"); - } - public async Task TestInsertAlignedStringRecordsOfOneDevice() - { - var session_pool = new SessionPool(host, port, poolSize); - await session_pool.Open(false); - if (debug) session_pool.OpenDebugMode(); - - System.Diagnostics.Debug.Assert(session_pool.IsOpen()); - var status = 0; - await session_pool.DeleteDatabaseAsync(testDatabaseName); - var device_id = string.Format("{0}.{1}", testDatabaseName, testDevice); - var measurements = new List<string>() { testMeasurements[0], testMeasurements[1], testMeasurements[2] }; - var data_type_lst = new List<TSDataType>() { TSDataType.TEXT, TSDataType.TEXT, TSDataType.TEXT }; - var encoding_lst = new List<TSEncoding>() { TSEncoding.PLAIN, TSEncoding.PLAIN, TSEncoding.PLAIN }; - var compressor_lst = new List<Compressor>() { Compressor.SNAPPY, Compressor.SNAPPY, Compressor.SNAPPY }; - status = await session_pool.CreateAlignedTimeseriesAsync(device_id, measurements, data_type_lst, encoding_lst, compressor_lst); - System.Diagnostics.Debug.Assert(status == 0); - - var measurements_lst = new List<List<string>>() { }; - measurements_lst.Add(new List<string>() { testMeasurements[0], testMeasurements[1], testMeasurements[2] }); - measurements_lst.Add(new List<string>() { testMeasurements[0], testMeasurements[1], testMeasurements[2] }); - measurements_lst.Add(new List<string>() { testMeasurements[0], testMeasurements[1], testMeasurements[2] }); - - var values_lst = new List<List<string>>() { }; - values_lst.Add(new List<string>() { "test1", "test2", "test3" }); - values_lst.Add(new List<string>() { "test4", "test5", "test6" }); - values_lst.Add(new List<string>() { "test7", "test8", "test9" }); - - var timestamp_lst = new List<long>() { 1, 2, 3 }; - - status = await session_pool.InsertAlignedStringRecordsOfOneDeviceAsync(device_id, timestamp_lst, measurements_lst, values_lst); - 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(); - // large data test - values_lst = new List<List<string>>() { }; - var tasks = new List<Task<int>>(); - measurements_lst = new List<List<string>>() { }; - timestamp_lst = new List<long>() { }; - for (var timestamp = 4; timestamp <= fetchSize * processedSize; timestamp++) - { - values_lst.Add(new List<string>() { "test1", "test2" }); - measurements_lst.Add(new List<string>() { testMeasurements[1], testMeasurements[2] }); - timestamp_lst.Add(timestamp); - if (timestamp % fetchSize == 0) + public async Task TestInsertAlignedStringRecordsOfOneDevice() { - tasks.Add(session_pool.InsertAlignedStringRecordsOfOneDeviceAsync(device_id, timestamp_lst, measurements_lst, values_lst)); - values_lst = new List<List<string>>() { }; - measurements_lst = new List<List<string>>() { }; - timestamp_lst = new List<long>() { }; + var session_pool = new SessionPool(host, port, poolSize); + await session_pool.Open(false); + if (debug) session_pool.OpenDebugMode(); + + System.Diagnostics.Debug.Assert(session_pool.IsOpen()); + var status = 0; + await session_pool.DeleteDatabaseAsync(testDatabaseName); + var device_id = string.Format("{0}.{1}", testDatabaseName, testDevice); + var measurements = new List<string>() { testMeasurements[0], testMeasurements[1], testMeasurements[2] }; + var data_type_lst = new List<TSDataType>() { TSDataType.TEXT, TSDataType.TEXT, TSDataType.TEXT }; + var encoding_lst = new List<TSEncoding>() { TSEncoding.PLAIN, TSEncoding.PLAIN, TSEncoding.PLAIN }; + var compressor_lst = new List<Compressor>() { Compressor.SNAPPY, Compressor.SNAPPY, Compressor.SNAPPY }; + status = await session_pool.CreateAlignedTimeseriesAsync(device_id, measurements, data_type_lst, encoding_lst, compressor_lst); + System.Diagnostics.Debug.Assert(status == 0); + + var measurements_lst = new List<List<string>>() { }; + measurements_lst.Add(new List<string>() { testMeasurements[0], testMeasurements[1], testMeasurements[2] }); + measurements_lst.Add(new List<string>() { testMeasurements[0], testMeasurements[1], testMeasurements[2] }); + measurements_lst.Add(new List<string>() { testMeasurements[0], testMeasurements[1], testMeasurements[2] }); + + var values_lst = new List<List<string>>() { }; + values_lst.Add(new List<string>() { "test1", "test2", "test3" }); + values_lst.Add(new List<string>() { "test4", "test5", "test6" }); + values_lst.Add(new List<string>() { "test7", "test8", "test9" }); + + var timestamp_lst = new List<long>() { 1, 2, 3 }; + + status = await session_pool.InsertAlignedStringRecordsOfOneDeviceAsync(device_id, timestamp_lst, measurements_lst, values_lst); + 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 (await res.HasNextAsync()) Console.WriteLine(res.Next()); + + await res.Close(); + // large data test + values_lst = new List<List<string>>() { }; + var tasks = new List<Task<int>>(); + measurements_lst = new List<List<string>>() { }; + timestamp_lst = new List<long>() { }; + for (var timestamp = 4; timestamp <= fetchSize * processedSize; timestamp++) + { + values_lst.Add(new List<string>() { "test1", "test2" }); + measurements_lst.Add(new List<string>() { testMeasurements[1], testMeasurements[2] }); + timestamp_lst.Add(timestamp); + if (timestamp % fetchSize == 0) + { + tasks.Add(session_pool.InsertAlignedStringRecordsOfOneDeviceAsync(device_id, timestamp_lst, measurements_lst, values_lst)); + values_lst = new List<List<string>>() { }; + measurements_lst = new List<List<string>>() { }; + timestamp_lst = new List<long>() { }; + } + } + + Task.WaitAll(tasks.ToArray()); + res = await session_pool.ExecuteQueryStatementAsync( + "select * from " + string.Format("{0}.{1}", testDatabaseName, testDevice)); + var res_count = 0; + while (await res.HasNextAsync()) + { + res.Next(); + res_count += 1; + } + + await res.Close(); + Console.WriteLine(res_count + " " + fetchSize * processedSize); + System.Diagnostics.Debug.Assert(res_count == fetchSize * processedSize); + status = await session_pool.DeleteDatabaseAsync(testDatabaseName); + System.Diagnostics.Debug.Assert(status == 0); + await session_pool.Close(); + Console.WriteLine("TestInsertAlignedStringRecordsOfOneDevice Passed!"); } - } - - Task.WaitAll(tasks.ToArray()); - res = await session_pool.ExecuteQueryStatementAsync( - "select * from " + string.Format("{0}.{1}", testDatabaseName, testDevice)); - var res_count = 0; - while (res.HasNext()) - { - res.Next(); - res_count += 1; - } - - await res.Close(); - Console.WriteLine(res_count + " " + fetchSize * processedSize); - System.Diagnostics.Debug.Assert(res_count == fetchSize * processedSize); - status = await session_pool.DeleteDatabaseAsync(testDatabaseName); - System.Diagnostics.Debug.Assert(status == 0); - await session_pool.Close(); - Console.WriteLine("TestInsertAlignedStringRecordsOfOneDevice Passed!"); } - } }
diff --git a/samples/Apache.IoTDB.Samples/SessionPoolTest.AlignedTablet.cs b/samples/Apache.IoTDB.Samples/SessionPoolTest.AlignedTablet.cs index 393869f..e975e69 100644 --- a/samples/Apache.IoTDB.Samples/SessionPoolTest.AlignedTablet.cs +++ b/samples/Apache.IoTDB.Samples/SessionPoolTest.AlignedTablet.cs
@@ -25,100 +25,100 @@ namespace Apache.IoTDB.Samples { - public partial class SessionPoolTest - { - public async Task TestInsertAlignedTablet() + public partial class SessionPoolTest { - var session_pool = new SessionPool(host, port, poolSize); - var status = 0; - await session_pool.Open(false); - if (debug) session_pool.OpenDebugMode(); + public async Task TestInsertAlignedTablet() + { + 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()); - await session_pool.DeleteDatabaseAsync(testDatabaseName); - var device_id = string.Format("{0}.{1}", testDatabaseName, testDevice); - var measurement_lst = new List<string> + System.Diagnostics.Debug.Assert(session_pool.IsOpen()); + await session_pool.DeleteDatabaseAsync(testDatabaseName); + var device_id = string.Format("{0}.{1}", testDatabaseName, testDevice); + var measurement_lst = new List<string> { testMeasurements[1], testMeasurements[2], testMeasurements[3] }; - var value_lst = new List<List<object>> + var value_lst = new List<List<object>> { new() {"iotdb", true, (int) 12}, new() {"c#", false, (int) 13}, new() {"client", true, (int) 14} }; - var timestamp_lst = new List<long> { 1, 2, 3 }; - var datatype_lst = new List<TSDataType> { TSDataType.TEXT, TSDataType.BOOLEAN, TSDataType.INT32 }; - var tablet = new Tablet(device_id, measurement_lst, datatype_lst, value_lst, timestamp_lst); - status = await session_pool.InsertAlignedTabletAsync(tablet); - System.Diagnostics.Debug.Assert(status == 0); - var res = await session_pool.ExecuteQueryStatementAsync( - "select * from " + string.Format("{0}.{1}", testDatabaseName, testDevice) + " where time<15"); - res.ShowTableNames(); - while (res.HasNext()) Console.WriteLine(res.Next()); + var timestamp_lst = new List<long> { 1, 2, 3 }; + var datatype_lst = new List<TSDataType> { TSDataType.TEXT, TSDataType.BOOLEAN, TSDataType.INT32 }; + var tablet = new Tablet(device_id, measurement_lst, datatype_lst, value_lst, timestamp_lst); + status = await session_pool.InsertAlignedTabletAsync(tablet); + System.Diagnostics.Debug.Assert(status == 0); + var res = await session_pool.ExecuteQueryStatementAsync( + "select * from " + string.Format("{0}.{1}", testDatabaseName, testDevice) + " where time<15"); + res.ShowTableNames(); + while (await res.HasNextAsync()) Console.WriteLine(res.Next()); - await res.Close(); - // large data test - value_lst = new List<List<object>>() { }; - timestamp_lst = new List<long>() { }; - var tasks = new List<Task<int>>(); - var start_ms = DateTime.Now.Ticks / 10000; - for (var timestamp = 4; timestamp <= fetchSize * processedSize; timestamp++) - { - timestamp_lst.Add(timestamp); - value_lst.Add(new List<object>() { "iotdb", true, (int)timestamp }); - if (timestamp % fetchSize == 0) - { - tablet = new Tablet(device_id, measurement_lst, datatype_lst, value_lst, timestamp_lst); - tasks.Add(session_pool.InsertAlignedTabletAsync(tablet)); - value_lst = new List<List<object>>() { }; - timestamp_lst = new List<long>() { }; + await res.Close(); + // large data test + value_lst = new List<List<object>>() { }; + timestamp_lst = new List<long>() { }; + var tasks = new List<Task<int>>(); + var start_ms = DateTime.Now.Ticks / 10000; + for (var timestamp = 4; timestamp <= fetchSize * processedSize; timestamp++) + { + timestamp_lst.Add(timestamp); + value_lst.Add(new List<object>() { "iotdb", true, (int)timestamp }); + if (timestamp % fetchSize == 0) + { + tablet = new Tablet(device_id, measurement_lst, datatype_lst, value_lst, timestamp_lst); + tasks.Add(session_pool.InsertAlignedTabletAsync(tablet)); + value_lst = new List<List<object>>() { }; + timestamp_lst = new List<long>() { }; + } + } + Console.WriteLine(tasks.Count); + + Task.WaitAll(tasks.ToArray()); + var end_ms = DateTime.Now.Ticks / 10000; + Console.WriteLine(string.Format("total tablet insert time is {0}", end_ms - start_ms)); + res = await session_pool.ExecuteQueryStatementAsync( + "select * from " + string.Format("{0}.{1}", testDatabaseName, testDevice)); + res.ShowTableNames(); + var res_count = 0; + while (await res.HasNextAsync()) + { + res.Next(); + res_count += 1; + } + + await res.Close(); + Console.WriteLine(res_count + " " + fetchSize * processedSize); + System.Diagnostics.Debug.Assert(res_count == fetchSize * processedSize); + status = await session_pool.DeleteDatabaseAsync(testDatabaseName); + System.Diagnostics.Debug.Assert(status == 0); + await session_pool.Close(); + Console.WriteLine("TestInsertAlignedTablet Passed!"); } - } - Console.WriteLine(tasks.Count); - Task.WaitAll(tasks.ToArray()); - var end_ms = DateTime.Now.Ticks / 10000; - Console.WriteLine(string.Format("total tablet insert time is {0}", end_ms - start_ms)); - res = await session_pool.ExecuteQueryStatementAsync( - "select * from " + string.Format("{0}.{1}", testDatabaseName, testDevice)); - res.ShowTableNames(); - var res_count = 0; - while (res.HasNext()) - { - res.Next(); - res_count += 1; - } + public async Task TestInsertAlignedTablets() + { + var session_pool = new SessionPool(host, port, poolSize); + var status = 0; + await session_pool.Open(false); + if (debug) session_pool.OpenDebugMode(); - await res.Close(); - Console.WriteLine(res_count + " " + fetchSize * processedSize); - System.Diagnostics.Debug.Assert(res_count == fetchSize * processedSize); - status = await session_pool.DeleteDatabaseAsync(testDatabaseName); - System.Diagnostics.Debug.Assert(status == 0); - await session_pool.Close(); - Console.WriteLine("TestInsertAlignedTablet Passed!"); - } - - public async Task TestInsertAlignedTablets() - { - 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()); - await session_pool.DeleteDatabaseAsync(testDatabaseName); - var device_id = new List<string>() + System.Diagnostics.Debug.Assert(session_pool.IsOpen()); + await session_pool.DeleteDatabaseAsync(testDatabaseName); + var device_id = new List<string>() { string.Format("{0}.{1}", testDatabaseName, testDevices[1]), string.Format("{0}.{1}", testDatabaseName, testDevices[2]) }; - var measurements_lst = new List<List<string>>() + var measurements_lst = new List<List<string>>() { new() {testMeasurements[1], testMeasurements[2], testMeasurements[3] }, new() {testMeasurements[1], testMeasurements[2], testMeasurements[3] } }; - var values_lst = new List<List<List<object>>>() + var values_lst = new List<List<List<object>>>() { new() { @@ -131,65 +131,65 @@ new List<object>() {"client_2", true, (int) 3} } }; - var datatype_lst = new List<List<TSDataType>>() + var datatype_lst = new List<List<TSDataType>>() { new() {TSDataType.TEXT, TSDataType.BOOLEAN, TSDataType.INT32}, new() {TSDataType.TEXT, TSDataType.BOOLEAN, TSDataType.INT32} }; - var timestamp_lst = new List<List<long>>() + var timestamp_lst = new List<List<long>>() {new() {2, 1, 3}, new() {3, 1, 2}}; - var tablets = new List<Tablet>() { }; - for (var i = 0; i < device_id.Count; i++) - { - var tablet = new Tablet(device_id[i], measurements_lst[i], datatype_lst[i], values_lst[i], timestamp_lst[i]); - tablets.Add(tablet); - } + var tablets = new List<Tablet>() { }; + for (var i = 0; i < device_id.Count; i++) + { + var tablet = new Tablet(device_id[i], measurements_lst[i], datatype_lst[i], values_lst[i], timestamp_lst[i]); + tablets.Add(tablet); + } - status = await session_pool.InsertAlignedTabletsAsync(tablets); - System.Diagnostics.Debug.Assert(status == 0); - var res = await session_pool.ExecuteQueryStatementAsync( - "select * from " + string.Format("{0}.{1}", testDatabaseName, testDevices[1]) + " where time<15"); - res.ShowTableNames(); - while (res.HasNext()) Console.WriteLine(res.Next()); + status = await session_pool.InsertAlignedTabletsAsync(tablets); + System.Diagnostics.Debug.Assert(status == 0); + var res = await session_pool.ExecuteQueryStatementAsync( + "select * from " + string.Format("{0}.{1}", testDatabaseName, testDevices[1]) + " where time<15"); + res.ShowTableNames(); + while (await res.HasNextAsync()) Console.WriteLine(res.Next()); - // large data test - var tasks = new List<Task<int>>(); - // tablets = new List<Tablet>() { }; - for (var timestamp = 4; timestamp <= processedSize * fetchSize; timestamp++) - { - var local_device_id = string.Format("{0}.{1}", testDatabaseName, testDevices[1]); - var local_measurements = new List<string>() + // large data test + var tasks = new List<Task<int>>(); + // tablets = new List<Tablet>() { }; + for (var timestamp = 4; timestamp <= processedSize * fetchSize; timestamp++) + { + var local_device_id = string.Format("{0}.{1}", testDatabaseName, testDevices[1]); + var local_measurements = new List<string>() {testMeasurements[1], testMeasurements[2], testMeasurements[3]}; - var local_value = new List<List<object>>() { new() { "iotdb", true, (int)timestamp } }; - var local_data_type = new List<TSDataType>() { TSDataType.TEXT, TSDataType.BOOLEAN, TSDataType.INT32 }; - var local_timestamp = new List<long> { timestamp }; - var tablet = new Tablet(local_device_id, local_measurements, local_data_type, local_value, local_timestamp); - tablets.Add(tablet); - if (timestamp % fetchSize == 0) - { - tasks.Add(session_pool.InsertAlignedTabletsAsync(tablets)); - tablets = new List<Tablet>() { }; + var local_value = new List<List<object>>() { new() { "iotdb", true, (int)timestamp } }; + var local_data_type = new List<TSDataType>() { TSDataType.TEXT, TSDataType.BOOLEAN, TSDataType.INT32 }; + var local_timestamp = new List<long> { timestamp }; + var tablet = new Tablet(local_device_id, local_measurements, local_data_type, local_value, local_timestamp); + tablets.Add(tablet); + if (timestamp % fetchSize == 0) + { + tasks.Add(session_pool.InsertAlignedTabletsAsync(tablets)); + tablets = new List<Tablet>() { }; + } + } + + Task.WaitAll(tasks.ToArray()); + res = await session_pool.ExecuteQueryStatementAsync( + "select * from " + string.Format("{0}.{1}", testDatabaseName, testDevices[1])); + res.ShowTableNames(); + var res_count = 0; + while (await res.HasNextAsync()) + { + res.Next(); + res_count += 1; + } + + await res.Close(); + Console.WriteLine(res_count + " " + fetchSize * processedSize); + System.Diagnostics.Debug.Assert(res_count == fetchSize * processedSize); + status = await session_pool.DeleteDatabaseAsync(testDatabaseName); + System.Diagnostics.Debug.Assert(status == 0); + await session_pool.Close(); + Console.WriteLine("TestInsertAlignedTablets Passed!"); } - } - - Task.WaitAll(tasks.ToArray()); - res = await session_pool.ExecuteQueryStatementAsync( - "select * from " + string.Format("{0}.{1}", testDatabaseName, testDevices[1])); - res.ShowTableNames(); - var res_count = 0; - while (res.HasNext()) - { - res.Next(); - res_count += 1; - } - - await res.Close(); - Console.WriteLine(res_count + " " + fetchSize * processedSize); - System.Diagnostics.Debug.Assert(res_count == fetchSize * processedSize); - status = await session_pool.DeleteDatabaseAsync(testDatabaseName); - System.Diagnostics.Debug.Assert(status == 0); - await session_pool.Close(); - Console.WriteLine("TestInsertAlignedTablets Passed!"); } - } }
diff --git a/samples/Apache.IoTDB.Samples/SessionPoolTest.Record.cs b/samples/Apache.IoTDB.Samples/SessionPoolTest.Record.cs index 7604c9a..767b93f 100644 --- a/samples/Apache.IoTDB.Samples/SessionPoolTest.Record.cs +++ b/samples/Apache.IoTDB.Samples/SessionPoolTest.Record.cs
@@ -25,212 +25,214 @@ namespace Apache.IoTDB.Samples { - public partial class SessionPoolTest - { - - public async Task TestInsertRecord() + public partial class SessionPoolTest { - var session_pool = new SessionPool(host, port, poolSize); - int status; - await session_pool.Open(false); - if (debug) session_pool.OpenDebugMode(); - System.Diagnostics.Debug.Assert(session_pool.IsOpen()); - status = await session_pool.DeleteDatabaseAsync(testDatabaseName); + public async Task TestInsertRecord() + { + var session_pool = new SessionPool(host, port, poolSize); + int status; + await session_pool.Open(false); + if (debug) session_pool.OpenDebugMode(); - status = await session_pool.CreateTimeSeries( - string.Format("{0}.{1}.{2}", testDatabaseName, testDevice, testMeasurements[1]), TSDataType.TEXT, - TSEncoding.PLAIN, Compressor.UNCOMPRESSED); + System.Diagnostics.Debug.Assert(session_pool.IsOpen()); + status = await session_pool.DeleteDatabaseAsync(testDatabaseName); - 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> + 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 }; - var tasks = new List<Task<int>>(); - var start_ms = DateTime.Now.Ticks / 10000; - for (var timestamp = 1; timestamp <= fetchSize * processedSize; timestamp++) - { - var rowRecord = new RowRecord(timestamp, values, measures); - var task = session_pool.InsertRecordAsync( - string.Format("{0}.{1}", testDatabaseName, testDevice), rowRecord); - tasks.Add(task); - } + var values = new List<object> { "test_text", true, (int)123 }; + var dataTypes = new List<string>() { "TEXT", "BOOLEAN", "INT32" }; + var tasks = new List<Task<int>>(); + var start_ms = DateTime.Now.Ticks / 10000; + for (var timestamp = 1; timestamp <= fetchSize * processedSize; timestamp++) + { + var rowRecord = new RowRecord(timestamp, values, measures, dataTypes); + var task = session_pool.InsertRecordAsync( + string.Format("{0}.{1}", testDatabaseName, testDevice), rowRecord); + tasks.Add(task); + } - Task.WaitAll(tasks.ToArray()); - var end_ms = DateTime.Now.Ticks / 10000; - Console.WriteLine(string.Format("total insert aligned record time is {0}", end_ms - start_ms)); - status = await session_pool.DeleteDatabaseAsync(testDatabaseName); - await session_pool.Close(); - Console.WriteLine("TestInsertRecordAsync Passed"); - } - public async Task TestInsertStringRecord() - { - var session_pool = new SessionPool(host, port, poolSize); - var status = 0; - await session_pool.Open(false); - if (debug) session_pool.OpenDebugMode(); + Task.WaitAll(tasks.ToArray()); + var end_ms = DateTime.Now.Ticks / 10000; + Console.WriteLine(string.Format("total insert aligned record time is {0}", end_ms - start_ms)); + status = await session_pool.DeleteDatabaseAsync(testDatabaseName); + await session_pool.Close(); + Console.WriteLine("TestInsertRecordAsync Passed"); + } + public async Task TestInsertStringRecord() + { + 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()); - await session_pool.DeleteDatabaseAsync(testDatabaseName); + System.Diagnostics.Debug.Assert(session_pool.IsOpen()); + await session_pool.DeleteDatabaseAsync(testDatabaseName); - status = await session_pool.CreateTimeSeries( - string.Format("{0}.{1}.{2}", testDatabaseName, testDevice, testMeasurements[0]), TSDataType.TEXT, - TSEncoding.PLAIN, Compressor.UNCOMPRESSED); + status = await session_pool.CreateTimeSeries( + string.Format("{0}.{1}.{2}", testDatabaseName, testDevice, testMeasurements[0]), 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[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.TEXT, TSEncoding.PLAIN, Compressor.UNCOMPRESSED); - System.Diagnostics.Debug.Assert(status == 0); - var measurements = new List<string> + System.Diagnostics.Debug.Assert(status == 0); + 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.TEXT, TSEncoding.PLAIN, Compressor.UNCOMPRESSED); + System.Diagnostics.Debug.Assert(status == 0); + var measurements = new List<string> {testMeasurements[0], testMeasurements[1], testMeasurements[2]}; - var values = new List<string> { "test_text1", "test_text2", "test_text3" }; - var tasks = new List<Task<int>>(); - var start_ms = DateTime.Now.Ticks / 10000; - for (var timestamp = 1; timestamp <= fetchSize * processedSize; timestamp++) - { - var task = session_pool.InsertStringRecordAsync( - string.Format("{0}.{1}", testDatabaseName, testDevice), measurements, values, timestamp); - tasks.Add(task); - } + var values = new List<string> { "test_text1", "test_text2", "test_text3" }; + var tasks = new List<Task<int>>(); + var start_ms = DateTime.Now.Ticks / 10000; + for (var timestamp = 1; timestamp <= fetchSize * processedSize; timestamp++) + { + var task = session_pool.InsertStringRecordAsync( + string.Format("{0}.{1}", testDatabaseName, testDevice), measurements, values, timestamp); + tasks.Add(task); + } - Task.WaitAll(tasks.ToArray()); - var end_ms = DateTime.Now.Ticks / 10000; - Console.WriteLine(string.Format("total insert string record time is {0}", end_ms - start_ms)); - var res = await session_pool.ExecuteQueryStatementAsync("select * from " + string.Format("{0}.{1}", testDatabaseName, testDevice)); - var res_cnt = 0; - while (res.HasNext()) - { - res.Next(); - res_cnt++; - } - Console.WriteLine(res_cnt + " " + fetchSize * processedSize); - System.Diagnostics.Debug.Assert(res_cnt == fetchSize * processedSize); - await session_pool.DeleteDatabaseAsync(testDatabaseName); - await session_pool.Close(); - Console.WriteLine("TestInsertStringRecordAsync Passed"); - } - public async Task TestInsertStrRecord() - { - var session_pool = new SessionPool(host, port, poolSize); - var status = 0; - await session_pool.Open(false); - if (debug) session_pool.OpenDebugMode(); + Task.WaitAll(tasks.ToArray()); + var end_ms = DateTime.Now.Ticks / 10000; + Console.WriteLine(string.Format("total insert string record time is {0}", end_ms - start_ms)); + var res = await session_pool.ExecuteQueryStatementAsync("select * from " + string.Format("{0}.{1}", testDatabaseName, testDevice)); + var res_cnt = 0; + while (await res.HasNextAsync()) + { + res.Next(); + res_cnt++; + } + Console.WriteLine(res_cnt + " " + fetchSize * processedSize); + System.Diagnostics.Debug.Assert(res_cnt == fetchSize * processedSize); + await session_pool.DeleteDatabaseAsync(testDatabaseName); + await session_pool.Close(); + Console.WriteLine("TestInsertStringRecordAsync Passed"); + } + public async Task TestInsertStrRecord() + { + 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()); - await session_pool.DeleteDatabaseAsync(testDatabaseName); + System.Diagnostics.Debug.Assert(session_pool.IsOpen()); + await session_pool.DeleteDatabaseAsync(testDatabaseName); - status = await session_pool.CreateTimeSeries( - string.Format("{0}.{1}.{2}", testDatabaseName, testDevice, testMeasurements[1]), - TSDataType.INT32, 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.INT32, TSEncoding.PLAIN, Compressor.UNCOMPRESSED); - System.Diagnostics.Debug.Assert(status == 0); + status = await session_pool.CreateTimeSeries( + string.Format("{0}.{1}.{2}", testDatabaseName, testDevice, testMeasurements[1]), + TSDataType.INT32, 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.INT32, TSEncoding.PLAIN, Compressor.UNCOMPRESSED); + System.Diagnostics.Debug.Assert(status == 0); - var measures = new List<string> { testMeasurements[1], testMeasurements[2] }; - var values = new List<object> { (int)1, (int)2 }; - var rowRecord = new RowRecord(1, values, measures); - status = await session_pool.InsertRecordAsync( - string.Format("{0}.{1}", testDatabaseName, testDevice), rowRecord); - System.Diagnostics.Debug.Assert(status == 0); - var res = await session_pool.ExecuteQueryStatementAsync( - "select * from " + string.Format("{0}.{1}", testDatabaseName, testDevice) + " where time<2"); - res.ShowTableNames(); - while (res.HasNext()) Console.WriteLine(res.Next()); + var measures = new List<string> { testMeasurements[1], testMeasurements[2] }; + var values = new List<object> { (int)1, (int)2 }; + var dataTypes = new List<string>() { "INT32", "INT32" }; + var rowRecord = new RowRecord(1, values, measures, dataTypes); + status = await session_pool.InsertRecordAsync( + string.Format("{0}.{1}", testDatabaseName, testDevice), rowRecord); + System.Diagnostics.Debug.Assert(status == 0); + var res = await session_pool.ExecuteQueryStatementAsync( + "select * from " + string.Format("{0}.{1}", testDatabaseName, testDevice) + " where time<2"); + res.ShowTableNames(); + while (await res.HasNextAsync()) Console.WriteLine(res.Next()); - await res.Close(); + await res.Close(); - var tasks = new List<Task<int>>(); - // large data test - var rowRecords = new List<RowRecord>() { }; - for (var timestamp = 2; timestamp <= fetchSize * processedSize; timestamp++) - rowRecords.Add(new RowRecord(timestamp, values, measures)); + var tasks = new List<Task<int>>(); + // large data test + var rowRecords = new List<RowRecord>() { }; + for (var timestamp = 2; timestamp <= fetchSize * processedSize; timestamp++) + rowRecords.Add(new RowRecord(timestamp, values, measures, dataTypes)); - for (var timestamp = 2; timestamp <= fetchSize * processedSize; timestamp++) - { - var task = session_pool.InsertRecordAsync( - string.Format("{0}.{1}", testDatabaseName, testDevice), rowRecords[timestamp - 2]); - tasks.Add(task); - } + for (var timestamp = 2; timestamp <= fetchSize * processedSize; timestamp++) + { + var task = session_pool.InsertRecordAsync( + string.Format("{0}.{1}", testDatabaseName, testDevice), rowRecords[timestamp - 2]); + tasks.Add(task); + } - Task.WaitAll(tasks.ToArray()); - res = await session_pool.ExecuteQueryStatementAsync( - "select * from " + string.Format("{0}.{1}", testDatabaseName, testDevice)); - var res_count = 0; - while (res.HasNext()) - { - res.Next(); - res_count += 1; - } + Task.WaitAll(tasks.ToArray()); + res = await session_pool.ExecuteQueryStatementAsync( + "select * from " + string.Format("{0}.{1}", testDatabaseName, testDevice)); + var res_count = 0; + while (await res.HasNextAsync()) + { + res.Next(); + res_count += 1; + } - await res.Close(); - Console.WriteLine(res_count + " " + fetchSize * processedSize); - System.Diagnostics.Debug.Assert(res_count == fetchSize * processedSize); - await session_pool.DeleteDatabaseAsync(testDatabaseName); - await session_pool.Close(); - Console.WriteLine("TestInsertStrRecord Passed!"); - } - public async Task TestInsertRecords() - { - var session_pool = new SessionPool(host, port, poolSize); - await session_pool.Open(false); - if (debug) session_pool.OpenDebugMode(); + await res.Close(); + Console.WriteLine(res_count + " " + fetchSize * processedSize); + System.Diagnostics.Debug.Assert(res_count == fetchSize * processedSize); + await session_pool.DeleteDatabaseAsync(testDatabaseName); + await session_pool.Close(); + Console.WriteLine("TestInsertStrRecord Passed!"); + } + public async Task TestInsertRecords() + { + var session_pool = new SessionPool(host, port, poolSize); + await session_pool.Open(false); + if (debug) session_pool.OpenDebugMode(); - System.Diagnostics.Debug.Assert(session_pool.IsOpen()); - var status = 0; - await session_pool.DeleteDatabaseAsync(testDatabaseName); - status = await session_pool.CreateTimeSeries( - string.Format("{0}.{1}.{2}", testDatabaseName, testDevice, testMeasurements[1]), - TSDataType.BOOLEAN, TSEncoding.PLAIN, Compressor.SNAPPY); - System.Diagnostics.Debug.Assert(status == 0); - status = await session_pool.CreateTimeSeries( - string.Format("{0}.{1}.{2}", testDatabaseName, testDevice, testMeasurements[2]), - TSDataType.INT32, TSEncoding.PLAIN, Compressor.SNAPPY); - System.Diagnostics.Debug.Assert(status == 0); - status = await session_pool.CreateTimeSeries( - string.Format("{0}.{1}.{2}", testDatabaseName, testDevice, testMeasurements[3]), - TSDataType.INT64, TSEncoding.PLAIN, Compressor.SNAPPY); - System.Diagnostics.Debug.Assert(status == 0); - status = await session_pool.CreateTimeSeries( - string.Format("{0}.{1}.{2}", testDatabaseName, testDevice, testMeasurements[4]), - TSDataType.DOUBLE, TSEncoding.PLAIN, Compressor.SNAPPY); - System.Diagnostics.Debug.Assert(status == 0); - status = await session_pool.CreateTimeSeries( - string.Format("{0}.{1}.{2}", testDatabaseName, testDevice, testMeasurements[5]), - TSDataType.FLOAT, TSEncoding.PLAIN, Compressor.SNAPPY); - System.Diagnostics.Debug.Assert(status == 0); - status = await session_pool.CreateTimeSeries( - string.Format("{0}.{1}.{2}", testDatabaseName, testDevice, testMeasurements[6]), TSDataType.TEXT, - TSEncoding.PLAIN, Compressor.SNAPPY); - System.Diagnostics.Debug.Assert(status == 0); + System.Diagnostics.Debug.Assert(session_pool.IsOpen()); + var status = 0; + await session_pool.DeleteDatabaseAsync(testDatabaseName); + status = await session_pool.CreateTimeSeries( + string.Format("{0}.{1}.{2}", testDatabaseName, testDevice, testMeasurements[1]), + TSDataType.BOOLEAN, TSEncoding.PLAIN, Compressor.SNAPPY); + System.Diagnostics.Debug.Assert(status == 0); + status = await session_pool.CreateTimeSeries( + string.Format("{0}.{1}.{2}", testDatabaseName, testDevice, testMeasurements[2]), + TSDataType.INT32, TSEncoding.PLAIN, Compressor.SNAPPY); + System.Diagnostics.Debug.Assert(status == 0); + status = await session_pool.CreateTimeSeries( + string.Format("{0}.{1}.{2}", testDatabaseName, testDevice, testMeasurements[3]), + TSDataType.INT64, TSEncoding.PLAIN, Compressor.SNAPPY); + System.Diagnostics.Debug.Assert(status == 0); + status = await session_pool.CreateTimeSeries( + string.Format("{0}.{1}.{2}", testDatabaseName, testDevice, testMeasurements[4]), + TSDataType.DOUBLE, TSEncoding.PLAIN, Compressor.SNAPPY); + System.Diagnostics.Debug.Assert(status == 0); + status = await session_pool.CreateTimeSeries( + string.Format("{0}.{1}.{2}", testDatabaseName, testDevice, testMeasurements[5]), + TSDataType.FLOAT, TSEncoding.PLAIN, Compressor.SNAPPY); + System.Diagnostics.Debug.Assert(status == 0); + status = await session_pool.CreateTimeSeries( + string.Format("{0}.{1}.{2}", testDatabaseName, testDevice, testMeasurements[6]), TSDataType.TEXT, + TSEncoding.PLAIN, Compressor.SNAPPY); + System.Diagnostics.Debug.Assert(status == 0); - var device_id = new List<string>() { }; - for (var i = 0; i < 3; i++) device_id.Add(string.Format("{0}.{1}", testDatabaseName, testDevice)); + var device_id = new List<string>() { }; + for (var i = 0; i < 3; i++) device_id.Add(string.Format("{0}.{1}", testDatabaseName, testDevice)); - var measurements_lst = new List<List<string>>() { }; - measurements_lst.Add(new List<string>() { testMeasurements[1], testMeasurements[2] }); - measurements_lst.Add(new List<string>() + var measurements_lst = new List<List<string>>() { }; + measurements_lst.Add(new List<string>() { testMeasurements[1], testMeasurements[2] }); + measurements_lst.Add(new List<string>() { testMeasurements[1], testMeasurements[2], testMeasurements[3], testMeasurements[4] }); - measurements_lst.Add(new List<string>() + measurements_lst.Add(new List<string>() { testMeasurements[1], testMeasurements[2], @@ -239,210 +241,215 @@ testMeasurements[5], testMeasurements[6] }); - var values_lst = new List<List<object>>() { }; - values_lst.Add(new List<object>() { true, (int)123 }); - values_lst.Add(new List<object>() { true, (int)123, (long)456, (double)1.1 }); - values_lst.Add(new List<object>() + var values_lst = new List<List<object>>() { }; + values_lst.Add(new List<object>() { true, (int)123 }); + values_lst.Add(new List<object>() { true, (int)123, (long)456, (double)1.1 }); + values_lst.Add(new List<object>() {true, (int) 123, (long) 456, (double) 1.1, (float) 10001.1, "test_record"}); - var timestamp_lst = new List<long>() { 1, 2, 3 }; - var rowRecords = new List<RowRecord>() { }; - for (var i = 0; i < 3; i++) - { - var rowRecord = new RowRecord(timestamp_lst[i], values_lst[i], measurements_lst[i]); - rowRecords.Add(rowRecord); - } + var timestamp_lst = new List<long>() { 1, 2, 3 }; + var dataTypes_lst = new List<List<string>>() { }; + dataTypes_lst.Add(new List<string>() { "BOOLEAN", "INT32" }); + dataTypes_lst.Add(new List<string>() { "BOOLEAN", "INT32", "INT64", "DOUBLE" }); + dataTypes_lst.Add(new List<string>() { "BOOLEAN", "INT32", "INT64", "DOUBLE", "FLOAT", "TEXT" }); + var rowRecords = new List<RowRecord>() { }; + for (var i = 0; i < 3; i++) + { + var rowRecord = new RowRecord(timestamp_lst[i], values_lst[i], measurements_lst[i], dataTypes_lst[i]); + rowRecords.Add(rowRecord); + } - status = await session_pool.InsertRecordsAsync(device_id, rowRecords); - 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()); + status = await session_pool.InsertRecordsAsync(device_id, rowRecords); + 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 (await res.HasNextAsync()) Console.WriteLine(res.Next()); - await res.Close(); - Console.WriteLine(status); + await res.Close(); + Console.WriteLine(status); - // large data test - device_id = new List<string>() { }; - rowRecords = new List<RowRecord>() { }; - var tasks = new List<Task<int>>(); - for (var timestamp = 4; timestamp <= fetchSize * processedSize; timestamp++) - { - device_id.Add(string.Format("{0}.{1}", testDatabaseName, testDevice)); - rowRecords.Add(new RowRecord(timestamp, new List<object>() { true, (int)123 }, - new List<string>() { testMeasurements[1], testMeasurements[2] })); - if (timestamp % fetchSize == 0) - { - tasks.Add(session_pool.InsertRecordsAsync(device_id, rowRecords)); - device_id = new List<string>() { }; - rowRecords = new List<RowRecord>() { }; + // large data test + device_id = new List<string>() { }; + rowRecords = new List<RowRecord>() { }; + var tasks = new List<Task<int>>(); + for (var timestamp = 4; timestamp <= fetchSize * processedSize; timestamp++) + { + device_id.Add(string.Format("{0}.{1}", testDatabaseName, testDevice)); + rowRecords.Add(new RowRecord(timestamp, new List<object>() { true, (int)123 }, + new List<string>() { testMeasurements[1], testMeasurements[2] }, + new List<string>() { "BOOLEAN", "INT32" })); + if (timestamp % fetchSize == 0) + { + tasks.Add(session_pool.InsertRecordsAsync(device_id, rowRecords)); + device_id = new List<string>() { }; + rowRecords = new List<RowRecord>() { }; + } + } + + Task.WaitAll(tasks.ToArray()); + res = await session_pool.ExecuteQueryStatementAsync( + "select * from " + string.Format("{0}.{1}", testDatabaseName, testDevice)); + res.ShowTableNames(); + var record_count = fetchSize * processedSize; + var res_count = 0; + while (await res.HasNextAsync()) + { + res.Next(); + res_count += 1; + } + + await res.Close(); + Console.WriteLine(res_count + " " + fetchSize * processedSize); + System.Diagnostics.Debug.Assert(res_count == record_count); + System.Diagnostics.Debug.Assert(status == 0); + + string sql = string.Format("select {0}, {1}, {2} from ", testMeasurements[3], testMeasurements[1], testMeasurements[2]) + string.Format("{0}.{1}", testDatabaseName, testDevice); + res = await session_pool.ExecuteQueryStatementAsync(sql); + res.ShowTableNames(); + RowRecord row = null; + while (await res.HasNextAsync()) + { + row = res.Next(); + break; + } + + Console.WriteLine($"{testDatabaseName}.{testDevice}.{row.Measurements[0]} {testMeasurements[3]}"); + System.Diagnostics.Debug.Assert($"{testDatabaseName}.{testDevice}.{testMeasurements[3]}" == row.Measurements[0]); + System.Diagnostics.Debug.Assert($"{testDatabaseName}.{testDevice}.{testMeasurements[1]}" == row.Measurements[1]); + System.Diagnostics.Debug.Assert($"{testDatabaseName}.{testDevice}.{testMeasurements[2]}" == row.Measurements[2]); + + status = await session_pool.DeleteDatabaseAsync(testDatabaseName); + System.Diagnostics.Debug.Assert(status == 0); + await session_pool.Close(); + Console.WriteLine("TestInsertRecords Passed!"); } - } - - Task.WaitAll(tasks.ToArray()); - res = await session_pool.ExecuteQueryStatementAsync( - "select * from " + string.Format("{0}.{1}", testDatabaseName, testDevice)); - res.ShowTableNames(); - var record_count = fetchSize * processedSize; - var res_count = 0; - while (res.HasNext()) - { - res.Next(); - res_count += 1; - } - - await res.Close(); - Console.WriteLine(res_count + " " + fetchSize * processedSize); - System.Diagnostics.Debug.Assert(res_count == record_count); - System.Diagnostics.Debug.Assert(status == 0); - - string sql = string.Format("select {0}, {1}, {2} from ", testMeasurements[3], testMeasurements[1], testMeasurements[2]) + string.Format("{0}.{1}", testDatabaseName, testDevice); - res = await session_pool.ExecuteQueryStatementAsync(sql); - res.ShowTableNames(); - RowRecord row = null; - while (res.HasNext()) - { - row = res.Next(); - break; - } - - Console.WriteLine($"{testDatabaseName}.{testDevice}.{row.Measurements[0]} {testMeasurements[3]}"); - System.Diagnostics.Debug.Assert($"{testDatabaseName}.{testDevice}.{testMeasurements[3]}" == row.Measurements[0]); - System.Diagnostics.Debug.Assert($"{testDatabaseName}.{testDevice}.{testMeasurements[1]}" == row.Measurements[1]); - System.Diagnostics.Debug.Assert($"{testDatabaseName}.{testDevice}.{testMeasurements[2]}" == row.Measurements[2]); - - status = await session_pool.DeleteDatabaseAsync(testDatabaseName); - System.Diagnostics.Debug.Assert(status == 0); - await session_pool.Close(); - Console.WriteLine("TestInsertRecords Passed!"); - } - public async Task TestInsertStringRecords() - { - var session_pool = new SessionPool(host, port, poolSize); - await session_pool.Open(false); - if (debug) session_pool.OpenDebugMode(); - - System.Diagnostics.Debug.Assert(session_pool.IsOpen()); - var status = 0; - await session_pool.DeleteDatabaseAsync(testDatabaseName); - status = await session_pool.CreateTimeSeries( - string.Format("{0}.{1}.{2}", testDatabaseName, testDevice, testMeasurements[1]), TSDataType.TEXT, - TSEncoding.PLAIN, Compressor.SNAPPY); - System.Diagnostics.Debug.Assert(status == 0); - status = await session_pool.CreateTimeSeries( - string.Format("{0}.{1}.{2}", testDatabaseName, testDevice, testMeasurements[2]), TSDataType.TEXT, - TSEncoding.PLAIN, Compressor.SNAPPY); - System.Diagnostics.Debug.Assert(status == 0); - - var device_id = new List<string>() { }; - for (var i = 0; i < 3; i++) device_id.Add(string.Format("{0}.{1}", testDatabaseName, testDevice)); - - var measurements_lst = new List<List<string>>() { }; - measurements_lst.Add(new List<string>() { testMeasurements[1], testMeasurements[2] }); - measurements_lst.Add(new List<string>() { testMeasurements[1], testMeasurements[2] }); - measurements_lst.Add(new List<string>() { testMeasurements[1], testMeasurements[2] }); - var values_lst = new List<List<string>>() { }; - values_lst.Add(new List<string>() { "test1", "test2" }); - values_lst.Add(new List<string>() { "test3", "test4" }); - values_lst.Add(new List<string>() { "test5", "test6" }); - var timestamp_lst = new List<long>() { 1, 2, 3 }; - - status = await session_pool.InsertStringRecordsAsync(device_id, measurements_lst, values_lst, timestamp_lst); - 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(); - - // large data test - device_id = new List<string>() { }; - measurements_lst = new List<List<string>>() { }; - values_lst = new List<List<string>>() { }; - timestamp_lst = new List<long>() { }; - var tasks = new List<Task<int>>(); - for (var timestamp = 4; timestamp <= fetchSize * processedSize; timestamp++) - { - device_id.Add(string.Format("{0}.{1}", testDatabaseName, testDevice)); - measurements_lst.Add(new List<string>() { testMeasurements[1], testMeasurements[2] }); - values_lst.Add(new List<string>() { "test" + timestamp, "test" + timestamp }); - timestamp_lst.Add(timestamp); - if (timestamp % fetchSize == 0) + public async Task TestInsertStringRecords() { - tasks.Add(session_pool.InsertStringRecordsAsync(device_id, measurements_lst, values_lst, timestamp_lst)); - device_id = new List<string>() { }; - measurements_lst = new List<List<string>>() { }; - values_lst = new List<List<string>>() { }; - timestamp_lst = new List<long>() { }; + var session_pool = new SessionPool(host, port, poolSize); + await session_pool.Open(false); + if (debug) session_pool.OpenDebugMode(); + + System.Diagnostics.Debug.Assert(session_pool.IsOpen()); + var status = 0; + await session_pool.DeleteDatabaseAsync(testDatabaseName); + status = await session_pool.CreateTimeSeries( + string.Format("{0}.{1}.{2}", testDatabaseName, testDevice, testMeasurements[1]), TSDataType.TEXT, + TSEncoding.PLAIN, Compressor.SNAPPY); + System.Diagnostics.Debug.Assert(status == 0); + status = await session_pool.CreateTimeSeries( + string.Format("{0}.{1}.{2}", testDatabaseName, testDevice, testMeasurements[2]), TSDataType.TEXT, + TSEncoding.PLAIN, Compressor.SNAPPY); + System.Diagnostics.Debug.Assert(status == 0); + + var device_id = new List<string>() { }; + for (var i = 0; i < 3; i++) device_id.Add(string.Format("{0}.{1}", testDatabaseName, testDevice)); + + var measurements_lst = new List<List<string>>() { }; + measurements_lst.Add(new List<string>() { testMeasurements[1], testMeasurements[2] }); + measurements_lst.Add(new List<string>() { testMeasurements[1], testMeasurements[2] }); + measurements_lst.Add(new List<string>() { testMeasurements[1], testMeasurements[2] }); + var values_lst = new List<List<string>>() { }; + values_lst.Add(new List<string>() { "test1", "test2" }); + values_lst.Add(new List<string>() { "test3", "test4" }); + values_lst.Add(new List<string>() { "test5", "test6" }); + var timestamp_lst = new List<long>() { 1, 2, 3 }; + + status = await session_pool.InsertStringRecordsAsync(device_id, measurements_lst, values_lst, timestamp_lst); + 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 (await res.HasNextAsync()) Console.WriteLine(res.Next()); + + await res.Close(); + + // large data test + device_id = new List<string>() { }; + measurements_lst = new List<List<string>>() { }; + values_lst = new List<List<string>>() { }; + timestamp_lst = new List<long>() { }; + var tasks = new List<Task<int>>(); + for (var timestamp = 4; timestamp <= fetchSize * processedSize; timestamp++) + { + device_id.Add(string.Format("{0}.{1}", testDatabaseName, testDevice)); + measurements_lst.Add(new List<string>() { testMeasurements[1], testMeasurements[2] }); + values_lst.Add(new List<string>() { "test" + timestamp, "test" + timestamp }); + timestamp_lst.Add(timestamp); + if (timestamp % fetchSize == 0) + { + tasks.Add(session_pool.InsertStringRecordsAsync(device_id, measurements_lst, values_lst, timestamp_lst)); + device_id = new List<string>() { }; + measurements_lst = new List<List<string>>() { }; + values_lst = new List<List<string>>() { }; + timestamp_lst = new List<long>() { }; + } + } + + Task.WaitAll(tasks.ToArray()); + res = await session_pool.ExecuteQueryStatementAsync( + "select * from " + string.Format("{0}.{1}", testDatabaseName, testDevice)); + res.ShowTableNames(); + var record_count = fetchSize * processedSize; + var res_count = 0; + while (await res.HasNextAsync()) + { + res.Next(); + res_count += 1; + } + + await res.Close(); + Console.WriteLine(res_count + " " + fetchSize * processedSize); + System.Diagnostics.Debug.Assert(res_count == record_count); + System.Diagnostics.Debug.Assert(status == 0); + status = await session_pool.DeleteDatabaseAsync(testDatabaseName); + System.Diagnostics.Debug.Assert(status == 0); + await session_pool.Close(); + Console.WriteLine("TestInsertStringRecords Passed!"); } - } + public async Task TestInsertRecordsOfOneDevice() + { + var session_pool = new SessionPool(host, port, poolSize); + await session_pool.Open(false); + if (debug) session_pool.OpenDebugMode(); - Task.WaitAll(tasks.ToArray()); - res = await session_pool.ExecuteQueryStatementAsync( - "select * from " + string.Format("{0}.{1}", testDatabaseName, testDevice)); - res.ShowTableNames(); - var record_count = fetchSize * processedSize; - var res_count = 0; - while (res.HasNext()) - { - res.Next(); - res_count += 1; - } - - await res.Close(); - Console.WriteLine(res_count + " " + fetchSize * processedSize); - System.Diagnostics.Debug.Assert(res_count == record_count); - System.Diagnostics.Debug.Assert(status == 0); - status = await session_pool.DeleteDatabaseAsync(testDatabaseName); - System.Diagnostics.Debug.Assert(status == 0); - await session_pool.Close(); - Console.WriteLine("TestInsertStringRecords Passed!"); - } - public async Task TestInsertRecordsOfOneDevice() - { - var session_pool = new SessionPool(host, port, poolSize); - await session_pool.Open(false); - if (debug) session_pool.OpenDebugMode(); - - System.Diagnostics.Debug.Assert(session_pool.IsOpen()); - var status = 0; - await session_pool.DeleteDatabaseAsync(testDatabaseName); - status = await session_pool.CreateTimeSeries( - string.Format("{0}.{1}.{2}", testDatabaseName, testDevice, testMeasurements[1]), - TSDataType.BOOLEAN, TSEncoding.PLAIN, Compressor.SNAPPY); - System.Diagnostics.Debug.Assert(status == 0); - status = await session_pool.CreateTimeSeries( - string.Format("{0}.{1}.{2}", testDatabaseName, testDevice, testMeasurements[2]), - TSDataType.INT32, TSEncoding.PLAIN, Compressor.SNAPPY); - System.Diagnostics.Debug.Assert(status == 0); - status = await session_pool.CreateTimeSeries( - string.Format("{0}.{1}.{2}", testDatabaseName, testDevice, testMeasurements[3]), - TSDataType.INT64, TSEncoding.PLAIN, Compressor.SNAPPY); - System.Diagnostics.Debug.Assert(status == 0); - status = await session_pool.CreateTimeSeries( - string.Format("{0}.{1}.{2}", testDatabaseName, testDevice, testMeasurements[4]), - TSDataType.DOUBLE, TSEncoding.PLAIN, Compressor.SNAPPY); - System.Diagnostics.Debug.Assert(status == 0); - status = await session_pool.CreateTimeSeries( - string.Format("{0}.{1}.{2}", testDatabaseName, testDevice, testMeasurements[5]), - TSDataType.FLOAT, TSEncoding.PLAIN, Compressor.SNAPPY); - System.Diagnostics.Debug.Assert(status == 0); - status = await session_pool.CreateTimeSeries( - string.Format("{0}.{1}.{2}", testDatabaseName, testDevice, testMeasurements[6]), TSDataType.TEXT, - TSEncoding.PLAIN, Compressor.SNAPPY); - System.Diagnostics.Debug.Assert(status == 0); - var device_id = string.Format("{0}.{1}", testDatabaseName, testDevice); - var measurements_lst = new List<List<string>>() { }; - measurements_lst.Add(new List<string>() { testMeasurements[1], testMeasurements[2] }); - measurements_lst.Add(new List<string>() + System.Diagnostics.Debug.Assert(session_pool.IsOpen()); + var status = 0; + await session_pool.DeleteDatabaseAsync(testDatabaseName); + status = await session_pool.CreateTimeSeries( + string.Format("{0}.{1}.{2}", testDatabaseName, testDevice, testMeasurements[1]), + TSDataType.BOOLEAN, TSEncoding.PLAIN, Compressor.SNAPPY); + System.Diagnostics.Debug.Assert(status == 0); + status = await session_pool.CreateTimeSeries( + string.Format("{0}.{1}.{2}", testDatabaseName, testDevice, testMeasurements[2]), + TSDataType.INT32, TSEncoding.PLAIN, Compressor.SNAPPY); + System.Diagnostics.Debug.Assert(status == 0); + status = await session_pool.CreateTimeSeries( + string.Format("{0}.{1}.{2}", testDatabaseName, testDevice, testMeasurements[3]), + TSDataType.INT64, TSEncoding.PLAIN, Compressor.SNAPPY); + System.Diagnostics.Debug.Assert(status == 0); + status = await session_pool.CreateTimeSeries( + string.Format("{0}.{1}.{2}", testDatabaseName, testDevice, testMeasurements[4]), + TSDataType.DOUBLE, TSEncoding.PLAIN, Compressor.SNAPPY); + System.Diagnostics.Debug.Assert(status == 0); + status = await session_pool.CreateTimeSeries( + string.Format("{0}.{1}.{2}", testDatabaseName, testDevice, testMeasurements[5]), + TSDataType.FLOAT, TSEncoding.PLAIN, Compressor.SNAPPY); + System.Diagnostics.Debug.Assert(status == 0); + status = await session_pool.CreateTimeSeries( + string.Format("{0}.{1}.{2}", testDatabaseName, testDevice, testMeasurements[6]), TSDataType.TEXT, + TSEncoding.PLAIN, Compressor.SNAPPY); + System.Diagnostics.Debug.Assert(status == 0); + var device_id = string.Format("{0}.{1}", testDatabaseName, testDevice); + var measurements_lst = new List<List<string>>() { }; + measurements_lst.Add(new List<string>() { testMeasurements[1], testMeasurements[2] }); + measurements_lst.Add(new List<string>() { testMeasurements[1], testMeasurements[2], testMeasurements[3], testMeasurements[4] }); - measurements_lst.Add(new List<string>() + measurements_lst.Add(new List<string>() { testMeasurements[1], testMeasurements[2], @@ -451,151 +458,156 @@ testMeasurements[5], testMeasurements[6] }); - var values_lst = new List<List<object>>() { }; - values_lst.Add(new List<object>() { true, (int)123 }); - values_lst.Add(new List<object>() { true, (int)123, (long)456, (double)1.1 }); - values_lst.Add(new List<object>() + var values_lst = new List<List<object>>() { }; + values_lst.Add(new List<object>() { true, (int)123 }); + values_lst.Add(new List<object>() { true, (int)123, (long)456, (double)1.1 }); + values_lst.Add(new List<object>() {true, (int) 123, (long) 456, (double) 1.1, (float) 10001.1, "test_record"}); - var timestamp_lst = new List<long>() { 1, 2, 3 }; - var rowRecords = new List<RowRecord>() { }; - for (var i = 0; i < 3; i++) - { - var rowRecord = new RowRecord(timestamp_lst[i], values_lst[i], measurements_lst[i]); - rowRecords.Add(rowRecord); - } + var timestamp_lst = new List<long>() { 1, 2, 3 }; + var dataTypes_lst = new List<List<string>>() { }; + dataTypes_lst.Add(new List<string>() { "BOOLEAN", "INT32" }); + dataTypes_lst.Add(new List<string>() { "BOOLEAN", "INT32", "INT64", "DOUBLE" }); + dataTypes_lst.Add(new List<string>() { "BOOLEAN", "INT32", "INT64", "DOUBLE", "FLOAT", "TEXT" }); + var rowRecords = new List<RowRecord>() { }; + for (var i = 0; i < 3; i++) + { + var rowRecord = new RowRecord(timestamp_lst[i], values_lst[i], measurements_lst[i], dataTypes_lst[i]); + rowRecords.Add(rowRecord); + } - status = await session_pool.InsertRecordsOfOneDeviceAsync(device_id, rowRecords); - 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()); + status = await session_pool.InsertRecordsOfOneDeviceAsync(device_id, rowRecords); + 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 (await res.HasNextAsync()) Console.WriteLine(res.Next()); - await res.Close(); - // large data test - rowRecords = new List<RowRecord>() { }; - var tasks = new List<Task<int>>(); - for (var timestamp = 4; timestamp <= fetchSize * processedSize; timestamp++) - { - rowRecords.Add(new RowRecord(timestamp, new List<object>() { true, (int)123 }, - new List<string>() { testMeasurements[1], testMeasurements[2] })); - if (timestamp % fetchSize == 0) - { - tasks.Add(session_pool.InsertRecordsOfOneDeviceAsync(device_id, rowRecords)); - rowRecords = new List<RowRecord>() { }; + await res.Close(); + // large data test + rowRecords = new List<RowRecord>() { }; + var tasks = new List<Task<int>>(); + for (var timestamp = 4; timestamp <= fetchSize * processedSize; timestamp++) + { + rowRecords.Add(new RowRecord(timestamp, new List<object>() { true, (int)123 }, + new List<string>() { testMeasurements[1], testMeasurements[2] }, + new List<string>() { "BOOLEAN", "INT32" })); + if (timestamp % fetchSize == 0) + { + tasks.Add(session_pool.InsertRecordsOfOneDeviceAsync(device_id, rowRecords)); + rowRecords = new List<RowRecord>() { }; + } + } + + Task.WaitAll(tasks.ToArray()); + res = await session_pool.ExecuteQueryStatementAsync( + "select * from " + string.Format("{0}.{1}", testDatabaseName, testDevice)); + var res_count = 0; + while (await res.HasNextAsync()) + { + res.Next(); + res_count += 1; + } + + await res.Close(); + Console.WriteLine(res_count + " " + fetchSize * processedSize); + System.Diagnostics.Debug.Assert(res_count == fetchSize * processedSize); + status = await session_pool.DeleteDatabaseAsync(testDatabaseName); + System.Diagnostics.Debug.Assert(status == 0); + await session_pool.Close(); + Console.WriteLine("TestInsertRecordsOfOneDevice Passed!"); } - } - - Task.WaitAll(tasks.ToArray()); - res = await session_pool.ExecuteQueryStatementAsync( - "select * from " + string.Format("{0}.{1}", testDatabaseName, testDevice)); - var res_count = 0; - while (res.HasNext()) - { - res.Next(); - res_count += 1; - } - - await res.Close(); - Console.WriteLine(res_count + " " + fetchSize * processedSize); - System.Diagnostics.Debug.Assert(res_count == fetchSize * processedSize); - status = await session_pool.DeleteDatabaseAsync(testDatabaseName); - System.Diagnostics.Debug.Assert(status == 0); - await session_pool.Close(); - Console.WriteLine("TestInsertRecordsOfOneDevice Passed!"); - } - public async Task TestInsertStringRecordsOfOneDevice() - { - var session_pool = new SessionPool(host, port, poolSize); - await session_pool.Open(false); - if (debug) session_pool.OpenDebugMode(); - - System.Diagnostics.Debug.Assert(session_pool.IsOpen()); - var status = 0; - await session_pool.DeleteDatabaseAsync(testDatabaseName); - status = await session_pool.CreateTimeSeries( - string.Format("{0}.{1}.{2}", testDatabaseName, testDevice, testMeasurements[0]), - TSDataType.TEXT, TSEncoding.PLAIN, Compressor.SNAPPY); - System.Diagnostics.Debug.Assert(status == 0); - status = await session_pool.CreateTimeSeries( - string.Format("{0}.{1}.{2}", testDatabaseName, testDevice, testMeasurements[1]), - TSDataType.TEXT, TSEncoding.PLAIN, Compressor.SNAPPY); - System.Diagnostics.Debug.Assert(status == 0); - status = await session_pool.CreateTimeSeries( - string.Format("{0}.{1}.{2}", testDatabaseName, testDevice, testMeasurements[2]), - TSDataType.TEXT, TSEncoding.PLAIN, Compressor.SNAPPY); - System.Diagnostics.Debug.Assert(status == 0); - - var device_id = string.Format("{0}.{1}", testDatabaseName, testDevice); - var measurements_lst = new List<List<string>>() { }; - measurements_lst.Add(new List<string>() { testMeasurements[0], testMeasurements[1], testMeasurements[2] }); - measurements_lst.Add(new List<string>() { testMeasurements[0], testMeasurements[1], testMeasurements[2] }); - measurements_lst.Add(new List<string>() { testMeasurements[0], testMeasurements[1], testMeasurements[2] }); - - var values_lst = new List<List<string>>() { }; - values_lst.Add(new List<string>() { "test1", "test2", "test3" }); - values_lst.Add(new List<string>() { "test4", "test5", "test6" }); - values_lst.Add(new List<string>() { "test7", "test8", "test9" }); - - var timestamp_lst = new List<long>() { 1, 2, 3 }; - - status = await session_pool.InsertStringRecordsOfOneDeviceAsync(device_id, timestamp_lst, measurements_lst, values_lst); - 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(); - // large data test - values_lst = new List<List<string>>() { }; - var tasks = new List<Task<int>>(); - measurements_lst = new List<List<string>>() { }; - timestamp_lst = new List<long>() { }; - for (var timestamp = 4; timestamp <= fetchSize * processedSize; timestamp++) - { - values_lst.Add(new List<string>() { "test1", "test2" }); - measurements_lst.Add(new List<string>() { testMeasurements[1], testMeasurements[2] }); - timestamp_lst.Add(timestamp); - if (timestamp % fetchSize == 0) + public async Task TestInsertStringRecordsOfOneDevice() { - tasks.Add(session_pool.InsertStringRecordsOfOneDeviceAsync(device_id, timestamp_lst, measurements_lst, values_lst)); - values_lst = new List<List<string>>() { }; - measurements_lst = new List<List<string>>() { }; - timestamp_lst = new List<long>() { }; + var session_pool = new SessionPool(host, port, poolSize); + await session_pool.Open(false); + if (debug) session_pool.OpenDebugMode(); + + System.Diagnostics.Debug.Assert(session_pool.IsOpen()); + var status = 0; + await session_pool.DeleteDatabaseAsync(testDatabaseName); + status = await session_pool.CreateTimeSeries( + string.Format("{0}.{1}.{2}", testDatabaseName, testDevice, testMeasurements[0]), + TSDataType.TEXT, TSEncoding.PLAIN, Compressor.SNAPPY); + System.Diagnostics.Debug.Assert(status == 0); + status = await session_pool.CreateTimeSeries( + string.Format("{0}.{1}.{2}", testDatabaseName, testDevice, testMeasurements[1]), + TSDataType.TEXT, TSEncoding.PLAIN, Compressor.SNAPPY); + System.Diagnostics.Debug.Assert(status == 0); + status = await session_pool.CreateTimeSeries( + string.Format("{0}.{1}.{2}", testDatabaseName, testDevice, testMeasurements[2]), + TSDataType.TEXT, TSEncoding.PLAIN, Compressor.SNAPPY); + System.Diagnostics.Debug.Assert(status == 0); + + var device_id = string.Format("{0}.{1}", testDatabaseName, testDevice); + var measurements_lst = new List<List<string>>() { }; + measurements_lst.Add(new List<string>() { testMeasurements[0], testMeasurements[1], testMeasurements[2] }); + measurements_lst.Add(new List<string>() { testMeasurements[0], testMeasurements[1], testMeasurements[2] }); + measurements_lst.Add(new List<string>() { testMeasurements[0], testMeasurements[1], testMeasurements[2] }); + + var values_lst = new List<List<string>>() { }; + values_lst.Add(new List<string>() { "test1", "test2", "test3" }); + values_lst.Add(new List<string>() { "test4", "test5", "test6" }); + values_lst.Add(new List<string>() { "test7", "test8", "test9" }); + + var timestamp_lst = new List<long>() { 1, 2, 3 }; + + status = await session_pool.InsertStringRecordsOfOneDeviceAsync(device_id, timestamp_lst, measurements_lst, values_lst); + 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 (await res.HasNextAsync()) Console.WriteLine(res.Next()); + + await res.Close(); + // large data test + values_lst = new List<List<string>>() { }; + var tasks = new List<Task<int>>(); + measurements_lst = new List<List<string>>() { }; + timestamp_lst = new List<long>() { }; + for (var timestamp = 4; timestamp <= fetchSize * processedSize; timestamp++) + { + values_lst.Add(new List<string>() { "test1", "test2" }); + measurements_lst.Add(new List<string>() { testMeasurements[1], testMeasurements[2] }); + timestamp_lst.Add(timestamp); + if (timestamp % fetchSize == 0) + { + tasks.Add(session_pool.InsertStringRecordsOfOneDeviceAsync(device_id, timestamp_lst, measurements_lst, values_lst)); + values_lst = new List<List<string>>() { }; + measurements_lst = new List<List<string>>() { }; + timestamp_lst = new List<long>() { }; + } + } + + Task.WaitAll(tasks.ToArray()); + res = await session_pool.ExecuteQueryStatementAsync( + "select * from " + string.Format("{0}.{1}", testDatabaseName, testDevice)); + var res_count = 0; + while (await res.HasNextAsync()) + { + res.Next(); + res_count += 1; + } + + await res.Close(); + Console.WriteLine(res_count + " " + fetchSize * processedSize); + System.Diagnostics.Debug.Assert(res_count == fetchSize * processedSize); + status = await session_pool.DeleteDatabaseAsync(testDatabaseName); + System.Diagnostics.Debug.Assert(status == 0); + await session_pool.Close(); + Console.WriteLine("TestInsertStringRecordsOfOneDevice Passed!"); } - } - Task.WaitAll(tasks.ToArray()); - res = await session_pool.ExecuteQueryStatementAsync( - "select * from " + string.Format("{0}.{1}", testDatabaseName, testDevice)); - var res_count = 0; - while (res.HasNext()) - { - res.Next(); - res_count += 1; - } + public async Task TestInsertRecordsWithAllType() + { + var session_pool = new SessionPool(host, port, poolSize); + var status = 0; + await session_pool.Open(false); + if (debug) session_pool.OpenDebugMode(); - await res.Close(); - Console.WriteLine(res_count + " " + fetchSize * processedSize); - System.Diagnostics.Debug.Assert(res_count == fetchSize * processedSize); - status = await session_pool.DeleteDatabaseAsync(testDatabaseName); - System.Diagnostics.Debug.Assert(status == 0); - await session_pool.Close(); - Console.WriteLine("TestInsertStringRecordsOfOneDevice Passed!"); - } + System.Diagnostics.Debug.Assert(session_pool.IsOpen()); + await session_pool.DeleteDatabaseAsync(testDatabaseName); - public async Task TestInsertRecordsWithAllType() - { - 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()); - await session_pool.DeleteDatabaseAsync(testDatabaseName); - - var measurements = new List<string> + var measurements = new List<string> { "boolean_measurement", "int32_measurement", @@ -609,7 +621,7 @@ "string_measurement" }; - var dataTypes = new List<string> + var dataTypes = new List<string> { "BOOLEAN", "INT32", @@ -624,44 +636,44 @@ }; - var values1 = new List<object> + var values1 = new List<object> { true, 123, 123456789L, 1.23f, 1.23456789, "iotdb", ((DateTimeOffset)DateTime.Now).ToUnixTimeMilliseconds(), DateTime.Today, new byte[] {0x01, 0x02}, "string1" }; - var values2 = new List<object> + var values2 = new List<object> { false, 456, 987654321L, 4.56f, 9.87654321, "iotdb2", ((DateTimeOffset)DateTime.Now.AddSeconds(1)).ToUnixTimeMilliseconds(), DateTime.Today.AddDays(1), new byte[] {0x03, 0x04}, "string2" }; - var values3 = new List<object> + var values3 = new List<object> { true, 789, 123123123L, 7.89f, 7.89101112, "iotdb3", ((DateTimeOffset)DateTime.Now.AddSeconds(2)).ToUnixTimeMilliseconds(), DateTime.Today.AddDays(2), new byte[] {0x05, 0x06}, "string3" }; - var rowRecord1 = new RowRecord(1, values1, measurements, dataTypes); - var rowRecord2 = new RowRecord(2, values2, measurements, dataTypes); - var rowRecord3 = new RowRecord(3, values3, measurements, dataTypes); + var rowRecord1 = new RowRecord(1, values1, measurements, dataTypes); + var rowRecord2 = new RowRecord(2, values2, measurements, dataTypes); + var rowRecord3 = new RowRecord(3, values3, measurements, dataTypes); - var device_id = new List<string> { string.Format("{0}.{1}", testDatabaseName, testDevice), string.Format("{0}.{1}", testDatabaseName, testDevice), string.Format("{0}.{1}", testDatabaseName, testDevice) }; - var rowRecords = new List<RowRecord> { rowRecord1, rowRecord2, rowRecord3 }; + var device_id = new List<string> { string.Format("{0}.{1}", testDatabaseName, testDevice), string.Format("{0}.{1}", testDatabaseName, testDevice), string.Format("{0}.{1}", testDatabaseName, testDevice) }; + var rowRecords = new List<RowRecord> { rowRecord1, rowRecord2, rowRecord3 }; - status = await session_pool.InsertRecordsAsync(device_id, rowRecords); - System.Diagnostics.Debug.Assert(status == 0); + status = await session_pool.InsertRecordsAsync(device_id, rowRecords); + System.Diagnostics.Debug.Assert(status == 0); - var res = await session_pool.ExecuteQueryStatementAsync( - "select * from " + string.Format("{0}.{1}", testDatabaseName, testDevice)); - res.ShowTableNames(); - var res_count = 0; - while (res.HasNext()) - { - Console.WriteLine(res.Next()); - res_count += 1; - } + var res = await session_pool.ExecuteQueryStatementAsync( + "select * from " + string.Format("{0}.{1}", testDatabaseName, testDevice)); + res.ShowTableNames(); + var res_count = 0; + while (await res.HasNextAsync()) + { + Console.WriteLine(res.Next()); + res_count += 1; + } - await res.Close(); - status = await session_pool.DeleteDatabaseAsync(testDatabaseName); - System.Diagnostics.Debug.Assert(status == 0); - await session_pool.Close(); - Console.WriteLine("TestInsertRecordsWithAllType Passed!"); + await res.Close(); + status = await session_pool.DeleteDatabaseAsync(testDatabaseName); + System.Diagnostics.Debug.Assert(status == 0); + await session_pool.Close(); + Console.WriteLine("TestInsertRecordsWithAllType Passed!"); + } } - } }
diff --git a/samples/Apache.IoTDB.Samples/SessionPoolTest.Tablet.cs b/samples/Apache.IoTDB.Samples/SessionPoolTest.Tablet.cs index 1f14fdd..eed1141 100644 --- a/samples/Apache.IoTDB.Samples/SessionPoolTest.Tablet.cs +++ b/samples/Apache.IoTDB.Samples/SessionPoolTest.Tablet.cs
@@ -25,100 +25,100 @@ namespace Apache.IoTDB.Samples { - public partial class SessionPoolTest - { - public async Task TestInsertTablet() + public partial class SessionPoolTest { - var session_pool = new SessionPool(host, port, poolSize); - var status = 0; - await session_pool.Open(false); - if (debug) session_pool.OpenDebugMode(); + public async Task TestInsertTablet() + { + 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()); - await session_pool.DeleteDatabaseAsync(testDatabaseName); - var device_id = string.Format("{0}.{1}", testDatabaseName, testDevice); - var measurement_lst = new List<string> + System.Diagnostics.Debug.Assert(session_pool.IsOpen()); + await session_pool.DeleteDatabaseAsync(testDatabaseName); + var device_id = string.Format("{0}.{1}", testDatabaseName, testDevice); + var measurement_lst = new List<string> { testMeasurements[1], testMeasurements[2], testMeasurements[3] }; - var value_lst = new List<List<object>> + var value_lst = new List<List<object>> { new() {"iotdb", true, (int) 12}, new() {"c#", false, (int) 13}, new() {"client", true, (int) 14} }; - var timestamp_lst = new List<long> { 1, 2, 3 }; - var datatype_lst = new List<TSDataType> { TSDataType.TEXT, TSDataType.BOOLEAN, TSDataType.INT32 }; - var tablet = new Tablet(device_id, measurement_lst, datatype_lst, value_lst, timestamp_lst); - status = await session_pool.InsertTabletAsync(tablet); - System.Diagnostics.Debug.Assert(status == 0); - var res = await session_pool.ExecuteQueryStatementAsync( - "select * from " + string.Format("{0}.{1}", testDatabaseName, testDevice) + " where time<15"); - res.ShowTableNames(); - while (res.HasNext()) Console.WriteLine(res.Next()); + var timestamp_lst = new List<long> { 1, 2, 3 }; + var datatype_lst = new List<TSDataType> { TSDataType.TEXT, TSDataType.BOOLEAN, TSDataType.INT32 }; + var tablet = new Tablet(device_id, measurement_lst, datatype_lst, value_lst, timestamp_lst); + status = await session_pool.InsertTabletAsync(tablet); + System.Diagnostics.Debug.Assert(status == 0); + var res = await session_pool.ExecuteQueryStatementAsync( + "select * from " + string.Format("{0}.{1}", testDatabaseName, testDevice) + " where time<15"); + res.ShowTableNames(); + while (await res.HasNextAsync()) Console.WriteLine(res.Next()); - await res.Close(); - // large data test - value_lst = new List<List<object>>() { }; - timestamp_lst = new List<long>() { }; - var tasks = new List<Task<int>>(); - var start_ms = DateTime.Now.Ticks / 10000; - for (var timestamp = 4; timestamp <= fetchSize * processedSize; timestamp++) - { - timestamp_lst.Add(timestamp); - value_lst.Add(new List<object>() { "iotdb", true, (int)timestamp }); - if (timestamp % fetchSize == 0) - { - tablet = new Tablet(device_id, measurement_lst, datatype_lst, value_lst, timestamp_lst); - tasks.Add(session_pool.InsertTabletAsync(tablet)); - value_lst = new List<List<object>>() { }; - timestamp_lst = new List<long>() { }; + await res.Close(); + // large data test + value_lst = new List<List<object>>() { }; + timestamp_lst = new List<long>() { }; + var tasks = new List<Task<int>>(); + var start_ms = DateTime.Now.Ticks / 10000; + for (var timestamp = 4; timestamp <= fetchSize * processedSize; timestamp++) + { + timestamp_lst.Add(timestamp); + value_lst.Add(new List<object>() { "iotdb", true, (int)timestamp }); + if (timestamp % fetchSize == 0) + { + tablet = new Tablet(device_id, measurement_lst, datatype_lst, value_lst, timestamp_lst); + tasks.Add(session_pool.InsertTabletAsync(tablet)); + value_lst = new List<List<object>>() { }; + timestamp_lst = new List<long>() { }; + } + } + + Task.WaitAll(tasks.ToArray()); + var end_ms = DateTime.Now.Ticks / 10000; + Console.WriteLine(string.Format("total tablet insert time is {0}", end_ms - start_ms)); + res = await session_pool.ExecuteQueryStatementAsync( + "select * from " + string.Format("{0}.{1}", testDatabaseName, testDevice)); + res.ShowTableNames(); + var res_count = 0; + while (await res.HasNextAsync()) + { + res.Next(); + res_count += 1; + } + + await res.Close(); + Console.WriteLine(res_count + " " + fetchSize * processedSize); + System.Diagnostics.Debug.Assert(res_count == fetchSize * processedSize); + status = await session_pool.DeleteDatabaseAsync(testDatabaseName); + System.Diagnostics.Debug.Assert(status == 0); + await session_pool.Close(); + Console.WriteLine("TestInsertTablet Passed!"); } - } - Task.WaitAll(tasks.ToArray()); - var end_ms = DateTime.Now.Ticks / 10000; - Console.WriteLine(string.Format("total tablet insert time is {0}", end_ms - start_ms)); - res = await session_pool.ExecuteQueryStatementAsync( - "select * from " + string.Format("{0}.{1}", testDatabaseName, testDevice)); - res.ShowTableNames(); - var res_count = 0; - while (res.HasNext()) - { - res.Next(); - res_count += 1; - } + public async Task TestInsertTablets() + { + var session_pool = new SessionPool(host, port, poolSize); + var status = 0; + await session_pool.Open(false); + if (debug) session_pool.OpenDebugMode(); - await res.Close(); - Console.WriteLine(res_count + " " + fetchSize * processedSize); - System.Diagnostics.Debug.Assert(res_count == fetchSize * processedSize); - status = await session_pool.DeleteDatabaseAsync(testDatabaseName); - System.Diagnostics.Debug.Assert(status == 0); - await session_pool.Close(); - Console.WriteLine("TestInsertTablet Passed!"); - } - - public async Task TestInsertTablets() - { - 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()); - await session_pool.DeleteDatabaseAsync(testDatabaseName); - var device_id = new List<string>() + System.Diagnostics.Debug.Assert(session_pool.IsOpen()); + await session_pool.DeleteDatabaseAsync(testDatabaseName); + var device_id = new List<string>() { string.Format("{0}.{1}", testDatabaseName, testDevices[1]), string.Format("{0}.{1}", testDatabaseName, testDevices[2]) }; - var measurements_lst = new List<List<string>>() + var measurements_lst = new List<List<string>>() { new() {testMeasurements[1], testMeasurements[2], testMeasurements[3]}, new() {testMeasurements[1], testMeasurements[2], testMeasurements[3]} }; - var values_lst = new List<List<List<object>>>() + var values_lst = new List<List<List<object>>>() { new() { @@ -131,116 +131,116 @@ new List<object>() {"client_2", true, (int) 3} } }; - var datatype_lst = new List<List<TSDataType>>() + var datatype_lst = new List<List<TSDataType>>() { new() {TSDataType.TEXT, TSDataType.BOOLEAN, TSDataType.INT32}, new() {TSDataType.TEXT, TSDataType.BOOLEAN, TSDataType.INT32} }; - var timestamp_lst = new List<List<long>>() + var timestamp_lst = new List<List<long>>() {new() {2, 1, 3}, new() {3, 1, 2}}; - var tablets = new List<Tablet>() { }; - for (var i = 0; i < device_id.Count; i++) - { - var tablet = new Tablet(device_id[i], measurements_lst[i], datatype_lst[i], values_lst[i], timestamp_lst[i]); - tablets.Add(tablet); - } + var tablets = new List<Tablet>() { }; + for (var i = 0; i < device_id.Count; i++) + { + var tablet = new Tablet(device_id[i], measurements_lst[i], datatype_lst[i], values_lst[i], timestamp_lst[i]); + tablets.Add(tablet); + } - status = await session_pool.InsertTabletsAsync(tablets); - // System.Diagnostics.Debug.Assert(status == 0); - var res = await session_pool.ExecuteQueryStatementAsync( - "select * from " + string.Format("{0}.{1}", testDatabaseName, testDevices[1]) + " where time<15"); - res.ShowTableNames(); - while (res.HasNext()) Console.WriteLine(res.Next()); + status = await session_pool.InsertTabletsAsync(tablets); + // System.Diagnostics.Debug.Assert(status == 0); + var res = await session_pool.ExecuteQueryStatementAsync( + "select * from " + string.Format("{0}.{1}", testDatabaseName, testDevices[1]) + " where time<15"); + res.ShowTableNames(); + while (await res.HasNextAsync()) Console.WriteLine(res.Next()); - // large data test + // large data test - var tasks = new List<Task<int>>(); - // tablets = new List<Tablet>(); - for (var timestamp = 4; timestamp <= processedSize * fetchSize; timestamp++) - { - var local_device_id = string.Format("{0}.{1}", testDatabaseName, testDevices[1]); - var local_measurements = new List<string>() + var tasks = new List<Task<int>>(); + // tablets = new List<Tablet>(); + for (var timestamp = 4; timestamp <= processedSize * fetchSize; timestamp++) + { + var local_device_id = string.Format("{0}.{1}", testDatabaseName, testDevices[1]); + var local_measurements = new List<string>() {testMeasurements[1], testMeasurements[2], testMeasurements[3]}; - var local_value = new List<List<object>>() { new() { "iotdb", true, (int)timestamp } }; - var local_timestamp = new List<long> { timestamp }; - var local_data_type = new List<TSDataType> { TSDataType.TEXT, TSDataType.BOOLEAN, TSDataType.INT32 }; - var tablet = new Tablet(local_device_id, local_measurements, local_data_type, local_value, local_timestamp); - tablets.Add(tablet); - if (timestamp % fetchSize == 0) - { - tasks.Add(session_pool.InsertTabletsAsync(tablets)); - tablets = new List<Tablet>() { }; + var local_value = new List<List<object>>() { new() { "iotdb", true, (int)timestamp } }; + var local_timestamp = new List<long> { timestamp }; + var local_data_type = new List<TSDataType> { TSDataType.TEXT, TSDataType.BOOLEAN, TSDataType.INT32 }; + var tablet = new Tablet(local_device_id, local_measurements, local_data_type, local_value, local_timestamp); + tablets.Add(tablet); + if (timestamp % fetchSize == 0) + { + tasks.Add(session_pool.InsertTabletsAsync(tablets)); + tablets = new List<Tablet>() { }; + } + } + + Task.WaitAll(tasks.ToArray()); + res = await session_pool.ExecuteQueryStatementAsync( + "select * from " + string.Format("{0}.{1}", testDatabaseName, testDevices[1])); + res.ShowTableNames(); + var res_count = 0; + while (await res.HasNextAsync()) + { + res.Next(); + res_count += 1; + } + + await res.Close(); + Console.WriteLine(res_count + " " + fetchSize * processedSize); + System.Diagnostics.Debug.Assert(res_count == fetchSize * processedSize); + status = await session_pool.DeleteDatabaseAsync(testDatabaseName); + System.Diagnostics.Debug.Assert(status == 0); + await session_pool.Close(); + Console.WriteLine("TestInsertTablets Passed!"); } - } + public async Task TestInsertTabletWithNullValue() + { + var session_pool = new SessionPool(host, port, poolSize); + var status = 0; + await session_pool.Open(false); + if (debug) session_pool.OpenDebugMode(); - Task.WaitAll(tasks.ToArray()); - res = await session_pool.ExecuteQueryStatementAsync( - "select * from " + string.Format("{0}.{1}", testDatabaseName, testDevices[1])); - res.ShowTableNames(); - var res_count = 0; - while (res.HasNext()) - { - res.Next(); - res_count += 1; - } - - await res.Close(); - Console.WriteLine(res_count + " " + fetchSize * processedSize); - System.Diagnostics.Debug.Assert(res_count == fetchSize * processedSize); - status = await session_pool.DeleteDatabaseAsync(testDatabaseName); - System.Diagnostics.Debug.Assert(status == 0); - await session_pool.Close(); - Console.WriteLine("TestInsertTablets Passed!"); - } - public async Task TestInsertTabletWithNullValue() - { - 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()); - await session_pool.DeleteDatabaseAsync(testDatabaseName); - var device_id = string.Format("{0}.{1}", testDatabaseName, testDevice); - var measurements = new List<string>() { testMeasurements[1], testMeasurements[2], testMeasurements[3] }; - var values = new List<List<object>>() + System.Diagnostics.Debug.Assert(session_pool.IsOpen()); + await session_pool.DeleteDatabaseAsync(testDatabaseName); + var device_id = string.Format("{0}.{1}", testDatabaseName, testDevice); + var measurements = new List<string>() { testMeasurements[1], testMeasurements[2], testMeasurements[3] }; + var values = new List<List<object>>() { new List<object>() {null, true, (int) 12}, new List<object>() {"c#", null, (int) 13}, new List<object>() {"client", true, null} }; - var datatype = new List<TSDataType>() { TSDataType.TEXT, TSDataType.BOOLEAN, TSDataType.INT32 }; - var timestamp = new List<long>() { 2, 1, 3 }; - var tablet = new Tablet(device_id, measurements, datatype, values, timestamp); - status = await session_pool.InsertTabletAsync(tablet); - System.Diagnostics.Debug.Assert(status == 0); - var res = await session_pool.ExecuteQueryStatementAsync( - "select * from " + string.Format("{0}.{1}", testDatabaseName, testDevice)); - res.ShowTableNames(); - var res_count = 0; - while (res.HasNext()) - { - Console.WriteLine(res.Next()); - res_count += 1; - } + var datatype = new List<TSDataType>() { TSDataType.TEXT, TSDataType.BOOLEAN, TSDataType.INT32 }; + var timestamp = new List<long>() { 2, 1, 3 }; + var tablet = new Tablet(device_id, measurements, datatype, values, timestamp); + status = await session_pool.InsertTabletAsync(tablet); + System.Diagnostics.Debug.Assert(status == 0); + var res = await session_pool.ExecuteQueryStatementAsync( + "select * from " + string.Format("{0}.{1}", testDatabaseName, testDevice)); + res.ShowTableNames(); + var res_count = 0; + while (await res.HasNextAsync()) + { + Console.WriteLine(res.Next()); + res_count += 1; + } - await res.Close(); - status = await session_pool.DeleteDatabaseAsync(testDatabaseName); - System.Diagnostics.Debug.Assert(status == 0); - await session_pool.Close(); - Console.WriteLine("TestInsertTabletsWithNullValue Passed!"); - } + await res.Close(); + status = await session_pool.DeleteDatabaseAsync(testDatabaseName); + System.Diagnostics.Debug.Assert(status == 0); + await session_pool.Close(); + Console.WriteLine("TestInsertTabletsWithNullValue Passed!"); + } - public async Task TestInsertTabletWithAllType() - { - var session_pool = new SessionPool(host, port, poolSize); - var status = 0; - await session_pool.Open(false); - if (debug) session_pool.OpenDebugMode(); + public async Task TestInsertTabletWithAllType() + { + 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()); - await session_pool.DeleteDatabaseAsync(testDatabaseName); - var device_id = string.Format("{0}.{1}", testDatabaseName, testDevice); - var measurements = new List<string> + System.Diagnostics.Debug.Assert(session_pool.IsOpen()); + await session_pool.DeleteDatabaseAsync(testDatabaseName); + var device_id = string.Format("{0}.{1}", testDatabaseName, testDevice); + var measurements = new List<string> { "boolean_measurement", "int32_measurement", @@ -253,12 +253,12 @@ "blob_measurement", "string_measurement" }; - var values = new List<List<object>> + var values = new List<List<object>> { new() {true, 123, 123456789L, 1.23f, 1.23456789, "iotdb", ((DateTimeOffset)DateTime.Now).ToUnixTimeMilliseconds(), DateTime.Today, new byte[] {0x01, 0x02}, "string1"}, new() {false, 456, 987654321L, 4.56f, 9.87654321, "iotdb2", ((DateTimeOffset)DateTime.Now.AddSeconds(1)).ToUnixTimeMilliseconds(), DateTime.Today.AddDays(1), new byte[] {0x03, 0x04}, "string2"} }; - var datatypes = new List<TSDataType> + var datatypes = new List<TSDataType> { TSDataType.BOOLEAN, TSDataType.INT32, @@ -271,26 +271,26 @@ TSDataType.BLOB, TSDataType.STRING }; - var timestamps = new List<long> { 1, 2 }; - var tablet = new Tablet(device_id, measurements, datatypes, values, timestamps); - status = await session_pool.InsertTabletAsync(tablet); - System.Diagnostics.Debug.Assert(status == 0); + var timestamps = new List<long> { 1, 2 }; + var tablet = new Tablet(device_id, measurements, datatypes, values, timestamps); + status = await session_pool.InsertTabletAsync(tablet); + System.Diagnostics.Debug.Assert(status == 0); - var res = await session_pool.ExecuteQueryStatementAsync( - "select * from " + string.Format("{0}.{1}", testDatabaseName, testDevice)); - res.ShowTableNames(); - var res_count = 0; - while (res.HasNext()) - { - Console.WriteLine(res.Next()); - res_count += 1; - } + var res = await session_pool.ExecuteQueryStatementAsync( + "select * from " + string.Format("{0}.{1}", testDatabaseName, testDevice)); + res.ShowTableNames(); + var res_count = 0; + while (await res.HasNextAsync()) + { + Console.WriteLine(res.Next()); + res_count += 1; + } - await res.Close(); - status = await session_pool.DeleteDatabaseAsync(testDatabaseName); - System.Diagnostics.Debug.Assert(status == 0); - await session_pool.Close(); - Console.WriteLine("TestInsertTabletWithAllType Passed!"); + await res.Close(); + status = await session_pool.DeleteDatabaseAsync(testDatabaseName); + System.Diagnostics.Debug.Assert(status == 0); + await session_pool.Close(); + Console.WriteLine("TestInsertTabletWithAllType Passed!"); + } } - } }
diff --git a/samples/Apache.IoTDB.Samples/SessionPoolTest.Template.cs b/samples/Apache.IoTDB.Samples/SessionPoolTest.Template.cs index 00e443b..866a7ab 100644 --- a/samples/Apache.IoTDB.Samples/SessionPoolTest.Template.cs +++ b/samples/Apache.IoTDB.Samples/SessionPoolTest.Template.cs
@@ -24,81 +24,81 @@ using Apache.IoTDB.DataStructure; namespace Apache.IoTDB.Samples { - public partial class SessionPoolTest - { - public async Task TestCreateAndDropSchemaTemplate() + public partial class SessionPoolTest { - var session_pool = new SessionPool(host, port, poolSize); - await session_pool.Open(false); - if (debug) session_pool.OpenDebugMode(); + public async Task TestCreateAndDropSchemaTemplate() + { + var session_pool = new SessionPool(host, port, poolSize); + await session_pool.Open(false); + if (debug) session_pool.OpenDebugMode(); - System.Diagnostics.Debug.Assert(session_pool.IsOpen()); - var status = 0; - await session_pool.DropSchemaTemplateAsync(testTemplateName); + System.Diagnostics.Debug.Assert(session_pool.IsOpen()); + var status = 0; + await session_pool.DropSchemaTemplateAsync(testTemplateName); - MeasurementNode node1 = new MeasurementNode(testMeasurements[1], TSDataType.INT32, TSEncoding.PLAIN, Compressor.SNAPPY); - MeasurementNode node2 = new MeasurementNode(testMeasurements[2], TSDataType.INT64, TSEncoding.PLAIN, Compressor.SNAPPY); - MeasurementNode node3 = new MeasurementNode(testMeasurements[3], TSDataType.DOUBLE, TSEncoding.PLAIN, Compressor.SNAPPY); - MeasurementNode node4 = new MeasurementNode(testMeasurements[4], TSDataType.FLOAT, TSEncoding.PLAIN, Compressor.SNAPPY); + MeasurementNode node1 = new MeasurementNode(testMeasurements[1], TSDataType.INT32, TSEncoding.PLAIN, Compressor.SNAPPY); + MeasurementNode node2 = new MeasurementNode(testMeasurements[2], TSDataType.INT64, TSEncoding.PLAIN, Compressor.SNAPPY); + MeasurementNode node3 = new MeasurementNode(testMeasurements[3], TSDataType.DOUBLE, TSEncoding.PLAIN, Compressor.SNAPPY); + MeasurementNode node4 = new MeasurementNode(testMeasurements[4], TSDataType.FLOAT, TSEncoding.PLAIN, Compressor.SNAPPY); - Template template = new Template(testTemplateName); - template.addToTemplate(node1); - template.addToTemplate(node2); - template.addToTemplate(node3); - template.addToTemplate(node4); + Template template = new Template(testTemplateName); + template.addToTemplate(node1); + template.addToTemplate(node2); + template.addToTemplate(node3); + template.addToTemplate(node4); - status = await session_pool.CreateSchemaTemplateAsync(template); - System.Diagnostics.Debug.Assert(status == 0); - var templates = await session_pool.ShowAllTemplatesAsync(); - foreach (var t in templates) - { - Console.WriteLine("template name :\t{0}", t); - } - status = await session_pool.DropSchemaTemplateAsync(testTemplateName); - System.Diagnostics.Debug.Assert(status == 0); - status = await session_pool.DeleteDatabaseAsync(testDatabaseName); - await session_pool.Close(); - Console.WriteLine("TestCreateAndDropSchemaTemplate Passed!"); + status = await session_pool.CreateSchemaTemplateAsync(template); + System.Diagnostics.Debug.Assert(status == 0); + var templates = await session_pool.ShowAllTemplatesAsync(); + foreach (var t in templates) + { + Console.WriteLine("template name :\t{0}", t); + } + status = await session_pool.DropSchemaTemplateAsync(testTemplateName); + System.Diagnostics.Debug.Assert(status == 0); + status = await session_pool.DeleteDatabaseAsync(testDatabaseName); + await session_pool.Close(); + Console.WriteLine("TestCreateAndDropSchemaTemplate Passed!"); + } + + public async Task TestSetAndUnsetSchemaTemplate() + { + var session_pool = new SessionPool(host, port, poolSize); + await session_pool.Open(false); + if (debug) session_pool.OpenDebugMode(); + + System.Diagnostics.Debug.Assert(session_pool.IsOpen()); + var status = 0; + await session_pool.DeleteDatabaseAsync(testDatabaseName); + await session_pool.UnsetSchemaTemplateAsync(string.Format("{0}.{1}", testDatabaseName, testDevice), "template"); + await session_pool.DropSchemaTemplateAsync(testTemplateName); + + MeasurementNode node1 = new MeasurementNode(testMeasurements[1], TSDataType.INT32, TSEncoding.PLAIN, Compressor.SNAPPY); + MeasurementNode node2 = new MeasurementNode(testMeasurements[2], TSDataType.INT64, TSEncoding.PLAIN, Compressor.SNAPPY); + MeasurementNode node3 = new MeasurementNode(testMeasurements[3], TSDataType.DOUBLE, TSEncoding.PLAIN, Compressor.SNAPPY); + MeasurementNode node4 = new MeasurementNode(testMeasurements[4], TSDataType.FLOAT, TSEncoding.PLAIN, Compressor.SNAPPY); + + Template template = new Template(testTemplateName); + template.addToTemplate(node1); + template.addToTemplate(node2); + template.addToTemplate(node3); + template.addToTemplate(node4); + + status = await session_pool.CreateSchemaTemplateAsync(template); + System.Diagnostics.Debug.Assert(status == 0); + status = await session_pool.SetSchemaTemplateAsync(testTemplateName, string.Format("{0}.{1}", testDatabaseName, testDevice)); + var paths = await session_pool.ShowPathsTemplateSetOnAsync(testTemplateName); + foreach (var p in paths) + { + Console.WriteLine("path :\t{0}", p); + } + status = await session_pool.UnsetSchemaTemplateAsync(string.Format("{0}.{1}", testDatabaseName, testDevice), testTemplateName); + status = await session_pool.DropSchemaTemplateAsync(testTemplateName); + System.Diagnostics.Debug.Assert(status == 0); + status = await session_pool.DeleteDatabaseAsync(testDatabaseName); + await session_pool.Close(); + Console.WriteLine("TestSetAndUnsetSchemaTemplate Passed!"); + } } - public async Task TestSetAndUnsetSchemaTemplate() - { - var session_pool = new SessionPool(host, port, poolSize); - await session_pool.Open(false); - if (debug) session_pool.OpenDebugMode(); - - System.Diagnostics.Debug.Assert(session_pool.IsOpen()); - var status = 0; - await session_pool.DeleteDatabaseAsync(testDatabaseName); - await session_pool.UnsetSchemaTemplateAsync(string.Format("{0}.{1}", testDatabaseName, testDevice), "template"); - await session_pool.DropSchemaTemplateAsync(testTemplateName); - - MeasurementNode node1 = new MeasurementNode(testMeasurements[1], TSDataType.INT32, TSEncoding.PLAIN, Compressor.SNAPPY); - MeasurementNode node2 = new MeasurementNode(testMeasurements[2], TSDataType.INT64, TSEncoding.PLAIN, Compressor.SNAPPY); - MeasurementNode node3 = new MeasurementNode(testMeasurements[3], TSDataType.DOUBLE, TSEncoding.PLAIN, Compressor.SNAPPY); - MeasurementNode node4 = new MeasurementNode(testMeasurements[4], TSDataType.FLOAT, TSEncoding.PLAIN, Compressor.SNAPPY); - - Template template = new Template(testTemplateName); - template.addToTemplate(node1); - template.addToTemplate(node2); - template.addToTemplate(node3); - template.addToTemplate(node4); - - status = await session_pool.CreateSchemaTemplateAsync(template); - System.Diagnostics.Debug.Assert(status == 0); - status = await session_pool.SetSchemaTemplateAsync(testTemplateName, string.Format("{0}.{1}", testDatabaseName, testDevice)); - var paths = await session_pool.ShowPathsTemplateSetOnAsync(testTemplateName); - foreach (var p in paths) - { - Console.WriteLine("path :\t{0}", p); - } - status = await session_pool.UnsetSchemaTemplateAsync(string.Format("{0}.{1}", testDatabaseName, testDevice), testTemplateName); - status = await session_pool.DropSchemaTemplateAsync(testTemplateName); - System.Diagnostics.Debug.Assert(status == 0); - status = await session_pool.DeleteDatabaseAsync(testDatabaseName); - await session_pool.Close(); - Console.WriteLine("TestSetAndUnsetSchemaTemplate Passed!"); - } - } - }
diff --git a/samples/Apache.IoTDB.Samples/SessionPoolTest.TestNetwork.cs b/samples/Apache.IoTDB.Samples/SessionPoolTest.TestNetwork.cs index fad6591..6f0857f 100644 --- a/samples/Apache.IoTDB.Samples/SessionPoolTest.TestNetwork.cs +++ b/samples/Apache.IoTDB.Samples/SessionPoolTest.TestNetwork.cs
@@ -24,101 +24,102 @@ using Apache.IoTDB.DataStructure; namespace Apache.IoTDB.Samples { - public partial class SessionPoolTest - { - public async Task TestTestInsertRecord() + public partial class SessionPoolTest { - var session_pool = new SessionPool(host, port, poolSize); - int status; - await session_pool.Open(false); - if (debug) session_pool.OpenDebugMode(); + public async Task TestTestInsertRecord() + { + var session_pool = new SessionPool(host, port, poolSize); + int status; + await session_pool.Open(false); + if (debug) session_pool.OpenDebugMode(); - System.Diagnostics.Debug.Assert(session_pool.IsOpen()); - status = await session_pool.DeleteDatabaseAsync(testDatabaseName); + 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); + 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> + 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 }; - var tasks = new List<Task<int>>(); - var start_ms = DateTime.Now.Ticks / 10000; - for (var timestamp = 1; timestamp <= fetchSize * processedSize; timestamp++) - { - var rowRecord = new RowRecord(timestamp, values, measures); - var task = session_pool.TestInsertRecordAsync( - string.Format("{0}.{1}", testDatabaseName, testDevice), rowRecord); - tasks.Add(task); - } + var values = new List<object> { "test_text", true, (int)123 }; + var dataTypes = new List<string>() { "TEXT", "BOOLEAN", "INT32" }; + var tasks = new List<Task<int>>(); + var start_ms = DateTime.Now.Ticks / 10000; + for (var timestamp = 1; timestamp <= fetchSize * processedSize; timestamp++) + { + var rowRecord = new RowRecord(timestamp, values, measures, dataTypes); + var task = session_pool.TestInsertRecordAsync( + string.Format("{0}.{1}", testDatabaseName, testDevice), rowRecord); + tasks.Add(task); + } - Task.WaitAll(tasks.ToArray()); - var end_ms = DateTime.Now.Ticks / 10000; - Console.WriteLine(string.Format("total insert record time is {0}", end_ms - start_ms)); - status = await session_pool.DeleteDatabaseAsync(testDatabaseName); - await session_pool.Close(); - Console.WriteLine("TestTestInsertRecordAsync Passed"); - } + Task.WaitAll(tasks.ToArray()); + var end_ms = DateTime.Now.Ticks / 10000; + Console.WriteLine(string.Format("total insert record time is {0}", end_ms - start_ms)); + status = await session_pool.DeleteDatabaseAsync(testDatabaseName); + await session_pool.Close(); + Console.WriteLine("TestTestInsertRecordAsync Passed"); + } - public async Task TestTestInsertRecords() - { - var session_pool = new SessionPool(host, port, poolSize); - await session_pool.Open(false); - if (debug) session_pool.OpenDebugMode(); + public async Task TestTestInsertRecords() + { + var session_pool = new SessionPool(host, port, poolSize); + await session_pool.Open(false); + if (debug) session_pool.OpenDebugMode(); - System.Diagnostics.Debug.Assert(session_pool.IsOpen()); - var status = 0; - await session_pool.DeleteDatabaseAsync(testDatabaseName); - status = await session_pool.CreateTimeSeries( - string.Format("{0}.{1}.{2}", testDatabaseName, testDevice, testMeasurements[1]), - TSDataType.BOOLEAN, TSEncoding.PLAIN, Compressor.SNAPPY); - System.Diagnostics.Debug.Assert(status == 0); - status = await session_pool.CreateTimeSeries( - string.Format("{0}.{1}.{2}", testDatabaseName, testDevice, testMeasurements[2]), - TSDataType.INT32, TSEncoding.PLAIN, Compressor.SNAPPY); - System.Diagnostics.Debug.Assert(status == 0); - status = await session_pool.CreateTimeSeries( - string.Format("{0}.{1}.{2}", testDatabaseName, testDevice, testMeasurements[3]), - TSDataType.INT64, TSEncoding.PLAIN, Compressor.SNAPPY); - System.Diagnostics.Debug.Assert(status == 0); - status = await session_pool.CreateTimeSeries( - string.Format("{0}.{1}.{2}", testDatabaseName, testDevice, testMeasurements[4]), - TSDataType.DOUBLE, TSEncoding.PLAIN, Compressor.SNAPPY); - System.Diagnostics.Debug.Assert(status == 0); - status = await session_pool.CreateTimeSeries( - string.Format("{0}.{1}.{2}", testDatabaseName, testDevice, testMeasurements[5]), - TSDataType.FLOAT, TSEncoding.PLAIN, Compressor.SNAPPY); - System.Diagnostics.Debug.Assert(status == 0); - status = await session_pool.CreateTimeSeries( - string.Format("{0}.{1}.{2}", testDatabaseName, testDevice, testMeasurements[6]), TSDataType.TEXT, - TSEncoding.PLAIN, Compressor.SNAPPY); - System.Diagnostics.Debug.Assert(status == 0); + System.Diagnostics.Debug.Assert(session_pool.IsOpen()); + var status = 0; + await session_pool.DeleteDatabaseAsync(testDatabaseName); + status = await session_pool.CreateTimeSeries( + string.Format("{0}.{1}.{2}", testDatabaseName, testDevice, testMeasurements[1]), + TSDataType.BOOLEAN, TSEncoding.PLAIN, Compressor.SNAPPY); + System.Diagnostics.Debug.Assert(status == 0); + status = await session_pool.CreateTimeSeries( + string.Format("{0}.{1}.{2}", testDatabaseName, testDevice, testMeasurements[2]), + TSDataType.INT32, TSEncoding.PLAIN, Compressor.SNAPPY); + System.Diagnostics.Debug.Assert(status == 0); + status = await session_pool.CreateTimeSeries( + string.Format("{0}.{1}.{2}", testDatabaseName, testDevice, testMeasurements[3]), + TSDataType.INT64, TSEncoding.PLAIN, Compressor.SNAPPY); + System.Diagnostics.Debug.Assert(status == 0); + status = await session_pool.CreateTimeSeries( + string.Format("{0}.{1}.{2}", testDatabaseName, testDevice, testMeasurements[4]), + TSDataType.DOUBLE, TSEncoding.PLAIN, Compressor.SNAPPY); + System.Diagnostics.Debug.Assert(status == 0); + status = await session_pool.CreateTimeSeries( + string.Format("{0}.{1}.{2}", testDatabaseName, testDevice, testMeasurements[5]), + TSDataType.FLOAT, TSEncoding.PLAIN, Compressor.SNAPPY); + System.Diagnostics.Debug.Assert(status == 0); + status = await session_pool.CreateTimeSeries( + string.Format("{0}.{1}.{2}", testDatabaseName, testDevice, testMeasurements[6]), TSDataType.TEXT, + TSEncoding.PLAIN, Compressor.SNAPPY); + System.Diagnostics.Debug.Assert(status == 0); - var device_id = new List<string>() { }; - for (var i = 0; i < 3; i++) device_id.Add(string.Format("{0}.{1}", testDatabaseName, testDevice)); + var device_id = new List<string>() { }; + for (var i = 0; i < 3; i++) device_id.Add(string.Format("{0}.{1}", testDatabaseName, testDevice)); - var measurements_lst = new List<List<string>>() { }; - measurements_lst.Add(new List<string>() { testMeasurements[1], testMeasurements[2] }); - measurements_lst.Add(new List<string>() + var measurements_lst = new List<List<string>>() { }; + measurements_lst.Add(new List<string>() { testMeasurements[1], testMeasurements[2] }); + measurements_lst.Add(new List<string>() { testMeasurements[1], testMeasurements[2], testMeasurements[3], testMeasurements[4] }); - measurements_lst.Add(new List<string>() + measurements_lst.Add(new List<string>() { testMeasurements[1], testMeasurements[2], @@ -127,156 +128,161 @@ testMeasurements[5], testMeasurements[6] }); - var values_lst = new List<List<object>>() { }; - values_lst.Add(new List<object>() { true, (int)123 }); - values_lst.Add(new List<object>() { true, (int)123, (long)456, (double)1.1 }); - values_lst.Add(new List<object>() + var values_lst = new List<List<object>>() { }; + values_lst.Add(new List<object>() { true, (int)123 }); + values_lst.Add(new List<object>() { true, (int)123, (long)456, (double)1.1 }); + values_lst.Add(new List<object>() {true, (int) 123, (long) 456, (double) 1.1, (float) 10001.1, "test_record"}); - var timestamp_lst = new List<long>() { 1, 2, 3 }; - var rowRecords = new List<RowRecord>() { }; - for (var i = 0; i < 3; i++) - { - var rowRecord = new RowRecord(timestamp_lst[i], values_lst[i], measurements_lst[i]); - rowRecords.Add(rowRecord); - } + var timestamp_lst = new List<long>() { 1, 2, 3 }; + var dataTypes_lst = new List<List<string>>() { }; + dataTypes_lst.Add(new List<string>() { "BOOLEAN", "INT32" }); + dataTypes_lst.Add(new List<string>() { "BOOLEAN", "INT32", "INT64", "DOUBLE" }); + dataTypes_lst.Add(new List<string>() { "BOOLEAN", "INT32", "INT64", "DOUBLE", "FLOAT", "TEXT" }); + var rowRecords = new List<RowRecord>() { }; + for (var i = 0; i < 3; i++) + { + var rowRecord = new RowRecord(timestamp_lst[i], values_lst[i], measurements_lst[i], dataTypes_lst[i]); + rowRecords.Add(rowRecord); + } - status = await session_pool.TestInsertRecordsAsync(device_id, rowRecords); - // 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()); + status = await session_pool.TestInsertRecordsAsync(device_id, rowRecords); + // 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 (await res.HasNextAsync()) Console.WriteLine(res.Next()); - await res.Close(); + await res.Close(); - // large data test - device_id = new List<string>() { }; - rowRecords = new List<RowRecord>() { }; - var tasks = new List<Task<int>>(); - for (var timestamp = 4; timestamp <= fetchSize * processedSize; timestamp++) - { - device_id.Add(string.Format("{0}.{1}", testDatabaseName, testDevice)); - rowRecords.Add(new RowRecord(timestamp, new List<object>() { true, (int)123 }, - new List<string>() { testMeasurements[1], testMeasurements[2] })); - if (timestamp % fetchSize == 0) - { - tasks.Add(session_pool.TestInsertRecordsAsync(device_id, rowRecords)); - device_id = new List<string>() { }; - rowRecords = new List<RowRecord>() { }; + // large data test + device_id = new List<string>() { }; + rowRecords = new List<RowRecord>() { }; + var tasks = new List<Task<int>>(); + for (var timestamp = 4; timestamp <= fetchSize * processedSize; timestamp++) + { + device_id.Add(string.Format("{0}.{1}", testDatabaseName, testDevice)); + rowRecords.Add(new RowRecord(timestamp, new List<object>() { true, (int)123 }, + new List<string>() { testMeasurements[1], testMeasurements[2] }, + new List<string>() { "BOOLEAN", "INT32" })); + if (timestamp % fetchSize == 0) + { + tasks.Add(session_pool.TestInsertRecordsAsync(device_id, rowRecords)); + device_id = new List<string>() { }; + rowRecords = new List<RowRecord>() { }; + } + } + + Task.WaitAll(tasks.ToArray()); + res = await session_pool.ExecuteQueryStatementAsync( + "select * from " + string.Format("{0}.{1}", testDatabaseName, testDevice)); + res.ShowTableNames(); + var record_count = fetchSize * processedSize; + var res_count = 0; + while (await res.HasNextAsync()) + { + res.Next(); + res_count += 1; + } + + await res.Close(); + System.Diagnostics.Debug.Assert(res_count == 0); + System.Diagnostics.Debug.Assert(status == 0); + status = await session_pool.DeleteDatabaseAsync(testDatabaseName); + System.Diagnostics.Debug.Assert(status == 0); + await session_pool.Close(); + Console.WriteLine("TestTestInsertRecords Passed!"); } - } - Task.WaitAll(tasks.ToArray()); - res = await session_pool.ExecuteQueryStatementAsync( - "select * from " + string.Format("{0}.{1}", testDatabaseName, testDevice)); - res.ShowTableNames(); - var record_count = fetchSize * processedSize; - var res_count = 0; - while (res.HasNext()) - { - res.Next(); - res_count += 1; - } + public async Task TestTestInsertTablet() + { + var session_pool = new SessionPool(host, port, poolSize); + var status = 0; + await session_pool.Open(false); + if (debug) session_pool.OpenDebugMode(); - await res.Close(); - System.Diagnostics.Debug.Assert(res_count == 0); - System.Diagnostics.Debug.Assert(status == 0); - status = await session_pool.DeleteDatabaseAsync(testDatabaseName); - System.Diagnostics.Debug.Assert(status == 0); - await session_pool.Close(); - Console.WriteLine("TestTestInsertRecords Passed!"); - } - - public async Task TestTestInsertTablet() - { - 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()); - await session_pool.DeleteDatabaseAsync(testDatabaseName); - var device_id = string.Format("{0}.{1}", testDatabaseName, testDevice); - var measurement_lst = new List<string> + System.Diagnostics.Debug.Assert(session_pool.IsOpen()); + await session_pool.DeleteDatabaseAsync(testDatabaseName); + var device_id = string.Format("{0}.{1}", testDatabaseName, testDevice); + var measurement_lst = new List<string> { testMeasurements[1], testMeasurements[2], testMeasurements[3] }; - var value_lst = new List<List<object>> + var value_lst = new List<List<object>> { new() {"iotdb", true, (int) 12}, new() {"c#", false, (int) 13}, new() {"client", true, (int) 14} }; - var timestamp_lst = new List<long> { 2, 1, 3 }; - var datatype_lst = new List<TSDataType> { TSDataType.TEXT, TSDataType.BOOLEAN, TSDataType.INT32 }; - var tablet = new Tablet(device_id, measurement_lst, datatype_lst, value_lst, timestamp_lst); - status = await session_pool.TestInsertTabletAsync(tablet); - System.Diagnostics.Debug.Assert(status == 0); - var res = await session_pool.ExecuteQueryStatementAsync( - "select * from " + string.Format("{0}.{1}", testDatabaseName, testDevice) + " where time<15"); - res.ShowTableNames(); - while (res.HasNext()) Console.WriteLine(res.Next()); + var timestamp_lst = new List<long> { 2, 1, 3 }; + var datatype_lst = new List<TSDataType> { TSDataType.TEXT, TSDataType.BOOLEAN, TSDataType.INT32 }; + var tablet = new Tablet(device_id, measurement_lst, datatype_lst, value_lst, timestamp_lst); + status = await session_pool.TestInsertTabletAsync(tablet); + System.Diagnostics.Debug.Assert(status == 0); + var res = await session_pool.ExecuteQueryStatementAsync( + "select * from " + string.Format("{0}.{1}", testDatabaseName, testDevice) + " where time<15"); + res.ShowTableNames(); + while (await res.HasNextAsync()) Console.WriteLine(res.Next()); - await res.Close(); - // large data test - value_lst = new List<List<object>>() { }; - timestamp_lst = new List<long>() { }; - var tasks = new List<Task<int>>(); - var start_ms = DateTime.Now.Ticks / 10000; - for (var timestamp = 4; timestamp <= fetchSize * processedSize; timestamp++) - { - timestamp_lst.Add(timestamp); - value_lst.Add(new List<object>() { "iotdb", true, (int)timestamp }); - if (timestamp % (fetchSize / 32) == 0) - { - tablet = new Tablet(device_id, measurement_lst, datatype_lst, value_lst, timestamp_lst); - tasks.Add(session_pool.TestInsertTabletAsync(tablet)); - value_lst = new List<List<object>>() { }; - timestamp_lst = new List<long>() { }; + await res.Close(); + // large data test + value_lst = new List<List<object>>() { }; + timestamp_lst = new List<long>() { }; + var tasks = new List<Task<int>>(); + var start_ms = DateTime.Now.Ticks / 10000; + for (var timestamp = 4; timestamp <= fetchSize * processedSize; timestamp++) + { + timestamp_lst.Add(timestamp); + value_lst.Add(new List<object>() { "iotdb", true, (int)timestamp }); + if (timestamp % (fetchSize / 32) == 0) + { + tablet = new Tablet(device_id, measurement_lst, datatype_lst, value_lst, timestamp_lst); + tasks.Add(session_pool.TestInsertTabletAsync(tablet)); + value_lst = new List<List<object>>() { }; + timestamp_lst = new List<long>() { }; + } + } + + Task.WaitAll(tasks.ToArray()); + var end_ms = DateTime.Now.Ticks / 10000; + Console.WriteLine(string.Format("total tablet insert time is {0}", end_ms - start_ms)); + res = await session_pool.ExecuteQueryStatementAsync( + "select * from " + string.Format("{0}.{1}", testDatabaseName, testDevice)); + res.ShowTableNames(); + var res_count = 0; + while (await res.HasNextAsync()) + { + res.Next(); + res_count += 1; + } + + await res.Close(); + System.Diagnostics.Debug.Assert(res_count == 0); + status = await session_pool.DeleteDatabaseAsync(testDatabaseName); + System.Diagnostics.Debug.Assert(status == 0); + await session_pool.Close(); + Console.WriteLine("TestTestInsertTablet Passed!"); } - } - Task.WaitAll(tasks.ToArray()); - var end_ms = DateTime.Now.Ticks / 10000; - Console.WriteLine(string.Format("total tablet insert time is {0}", end_ms - start_ms)); - res = await session_pool.ExecuteQueryStatementAsync( - "select * from " + string.Format("{0}.{1}", testDatabaseName, testDevice)); - res.ShowTableNames(); - var res_count = 0; - while (res.HasNext()) - { - res.Next(); - res_count += 1; - } + public async Task TestTestInsertTablets() + { + var session_pool = new SessionPool(host, port, poolSize); + var status = 0; + await session_pool.Open(false); + if (debug) session_pool.OpenDebugMode(); - await res.Close(); - System.Diagnostics.Debug.Assert(res_count == 0); - status = await session_pool.DeleteDatabaseAsync(testDatabaseName); - System.Diagnostics.Debug.Assert(status == 0); - await session_pool.Close(); - Console.WriteLine("TestTestInsertTablet Passed!"); - } - - public async Task TestTestInsertTablets() - { - 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()); - await session_pool.DeleteDatabaseAsync(testDatabaseName); - var device_id = new List<string>() + System.Diagnostics.Debug.Assert(session_pool.IsOpen()); + await session_pool.DeleteDatabaseAsync(testDatabaseName); + var device_id = new List<string>() { string.Format("{0}.{1}", testDatabaseName, testDevices[1]), string.Format("{0}.{1}", testDatabaseName, testDevices[2]) }; - var measurements_lst = new List<List<string>>() + var measurements_lst = new List<List<string>>() { new() {testMeasurements[1], testMeasurements[2], testMeasurements[3]}, new() {testMeasurements[1], testMeasurements[2], testMeasurements[3]} }; - var values_lst = new List<List<List<object>>>() + var values_lst = new List<List<List<object>>>() { new() { @@ -289,66 +295,66 @@ new List<object>() {"client_2", true, (int) 3} } }; - var datatype_lst = new List<List<TSDataType>>() + var datatype_lst = new List<List<TSDataType>>() { new() {TSDataType.TEXT, TSDataType.BOOLEAN, TSDataType.INT32}, new() {TSDataType.TEXT, TSDataType.BOOLEAN, TSDataType.INT32} }; - var timestamp_lst = new List<List<long>>() + var timestamp_lst = new List<List<long>>() {new() {2, 1, 3}, new() {3, 1, 2}}; - var tablets = new List<Tablet>() { }; - for (var i = 0; i < device_id.Count; i++) - { - var tablet = new Tablet(device_id[i], measurements_lst[i], datatype_lst[i], values_lst[i], timestamp_lst[i]); - tablets.Add(tablet); - } + var tablets = new List<Tablet>() { }; + for (var i = 0; i < device_id.Count; i++) + { + var tablet = new Tablet(device_id[i], measurements_lst[i], datatype_lst[i], values_lst[i], timestamp_lst[i]); + tablets.Add(tablet); + } - status = await session_pool.TestInsertTabletsAsync(tablets); - // System.Diagnostics.Debug.Assert(status == 0); - var res = await session_pool.ExecuteQueryStatementAsync( - "select * from " + string.Format("{0}.{1}", testDatabaseName, testDevices[1]) + " where time<15"); - res.ShowTableNames(); - while (res.HasNext()) Console.WriteLine(res.Next()); - await res.Close(); + status = await session_pool.TestInsertTabletsAsync(tablets); + // System.Diagnostics.Debug.Assert(status == 0); + var res = await session_pool.ExecuteQueryStatementAsync( + "select * from " + string.Format("{0}.{1}", testDatabaseName, testDevices[1]) + " where time<15"); + res.ShowTableNames(); + while (await res.HasNextAsync()) Console.WriteLine(res.Next()); + await res.Close(); - // large data test + // large data test - var tasks = new List<Task<int>>(); - for (var timestamp = 4; timestamp <= processedSize * fetchSize; timestamp++) - { - var local_device_id = string.Format("{0}.{1}", testDatabaseName, testDevices[1]); - var local_measurements = new List<string>() + var tasks = new List<Task<int>>(); + for (var timestamp = 4; timestamp <= processedSize * fetchSize; timestamp++) + { + var local_device_id = string.Format("{0}.{1}", testDatabaseName, testDevices[1]); + var local_measurements = new List<string>() {testMeasurements[1], testMeasurements[2], testMeasurements[3]}; - var local_value = new List<List<object>>() { new() { "iotdb", true, (int)timestamp } }; - var local_timestamp = new List<long> { timestamp }; - var local_data_type = new List<TSDataType> { TSDataType.TEXT, TSDataType.BOOLEAN, TSDataType.INT32 }; - var tablet = new Tablet(local_device_id, local_measurements, local_data_type, local_value, local_timestamp); - tablets.Add(tablet); - if (timestamp % fetchSize == 0) - { - tasks.Add(session_pool.TestInsertTabletsAsync(tablets)); - tablets = new List<Tablet>() { }; + var local_value = new List<List<object>>() { new() { "iotdb", true, (int)timestamp } }; + var local_timestamp = new List<long> { timestamp }; + var local_data_type = new List<TSDataType> { TSDataType.TEXT, TSDataType.BOOLEAN, TSDataType.INT32 }; + var tablet = new Tablet(local_device_id, local_measurements, local_data_type, local_value, local_timestamp); + tablets.Add(tablet); + if (timestamp % fetchSize == 0) + { + tasks.Add(session_pool.TestInsertTabletsAsync(tablets)); + tablets = new List<Tablet>() { }; + } + } + + Task.WaitAll(tasks.ToArray()); + res = await session_pool.ExecuteQueryStatementAsync( + "select * from " + string.Format("{0}.{1}", testDatabaseName, testDevices[1])); + res.ShowTableNames(); + var res_count = 0; + while (await res.HasNextAsync()) + { + res.Next(); + res_count += 1; + } + + await res.Close(); + System.Diagnostics.Debug.Assert(res_count == 0); + status = await session_pool.DeleteDatabaseAsync(testDatabaseName); + System.Diagnostics.Debug.Assert(status == 0); + await session_pool.Close(); + Console.WriteLine("TestTestInsertTablets Passed!"); } - } - Task.WaitAll(tasks.ToArray()); - res = await session_pool.ExecuteQueryStatementAsync( - "select * from " + string.Format("{0}.{1}", testDatabaseName, testDevices[1])); - res.ShowTableNames(); - var res_count = 0; - while (res.HasNext()) - { - res.Next(); - res_count += 1; - } - - await res.Close(); - System.Diagnostics.Debug.Assert(res_count == 0); - status = await session_pool.DeleteDatabaseAsync(testDatabaseName); - System.Diagnostics.Debug.Assert(status == 0); - await session_pool.Close(); - Console.WriteLine("TestTestInsertTablets Passed!"); } - - } }
diff --git a/samples/Apache.IoTDB.Samples/SessionPoolTest.TimeSeries.cs b/samples/Apache.IoTDB.Samples/SessionPoolTest.TimeSeries.cs index ce98c60..049407c 100644 --- a/samples/Apache.IoTDB.Samples/SessionPoolTest.TimeSeries.cs +++ b/samples/Apache.IoTDB.Samples/SessionPoolTest.TimeSeries.cs
@@ -25,120 +25,120 @@ using Apache.IoTDB.DataStructure; namespace Apache.IoTDB.Samples { - public partial class SessionPoolTest - { - public async Task TestCreateMultiTimeSeries() + public partial class SessionPoolTest { - // by Luzhan - var session_pool = new SessionPool(host, port, username, password, poolSize); - await session_pool.Open(false); - var status = 0; - if (debug) session_pool.OpenDebugMode(); + public async Task TestCreateMultiTimeSeries() + { + // by Luzhan + var session_pool = new SessionPool(host, port, username, password, poolSize); + await session_pool.Open(false); + var status = 0; + if (debug) session_pool.OpenDebugMode(); - status = await session_pool.DeleteDatabaseAsync(testDatabaseName); - var measurement_lst = new List<int> { 1, 2, 3, 4, 5, 6 }; - var ts_path_lst = new List<string>(measurement_lst.ConvertAll( - (measurement) => string.Format("{0}.{1}.{2}{3}", testDatabaseName, testDevice, testMeasurement, measurement))); - var data_type_lst = new List<TSDataType>() + status = await session_pool.DeleteDatabaseAsync(testDatabaseName); + var measurement_lst = new List<int> { 1, 2, 3, 4, 5, 6 }; + var ts_path_lst = new List<string>(measurement_lst.ConvertAll( + (measurement) => string.Format("{0}.{1}.{2}{3}", testDatabaseName, testDevice, testMeasurement, measurement))); + var data_type_lst = new List<TSDataType>() { TSDataType.BOOLEAN, TSDataType.INT32, TSDataType.INT64, TSDataType.FLOAT, TSDataType.DOUBLE, TSDataType.TEXT }; - var encoding_lst = new List<TSEncoding>() + var encoding_lst = new List<TSEncoding>() { TSEncoding.PLAIN, TSEncoding.PLAIN, TSEncoding.PLAIN, TSEncoding.PLAIN, TSEncoding.PLAIN, TSEncoding.PLAIN }; - var compressor_lst = new List<Compressor>() + var compressor_lst = new List<Compressor>() { Compressor.SNAPPY, Compressor.SNAPPY, Compressor.SNAPPY, Compressor.SNAPPY, Compressor.SNAPPY, Compressor.SNAPPY }; - status = await session_pool.CreateMultiTimeSeriesAsync(ts_path_lst, data_type_lst, encoding_lst, - compressor_lst); - System.Diagnostics.Debug.Assert(status == 0); - status = await session_pool.DeleteDatabaseAsync(testDatabaseName); - System.Diagnostics.Debug.Assert(status == 0); - await session_pool.Close(); - Console.WriteLine("TestCreateMultiTimeSeries Passed!"); - } + status = await session_pool.CreateMultiTimeSeriesAsync(ts_path_lst, data_type_lst, encoding_lst, + compressor_lst); + System.Diagnostics.Debug.Assert(status == 0); + status = await session_pool.DeleteDatabaseAsync(testDatabaseName); + System.Diagnostics.Debug.Assert(status == 0); + await session_pool.Close(); + Console.WriteLine("TestCreateMultiTimeSeries Passed!"); + } - public async Task TestDeleteTimeSeries() - { - var session_pool = new SessionPool(host, port, username, password, poolSize); - await session_pool.Open(false); - var status = 0; - if (debug) session_pool.OpenDebugMode(); + public async Task TestDeleteTimeSeries() + { + var session_pool = new SessionPool(host, port, username, password, poolSize); + await session_pool.Open(false); + var status = 0; + if (debug) session_pool.OpenDebugMode(); - status = await session_pool.DeleteDatabaseAsync(testDatabaseName); - var measurement_lst = new List<int> { 1, 2, 3, 4, 5, 6 }; - var ts_path_lst = new List<string>(measurement_lst.ConvertAll( - (measurement) => string.Format("{0}.{1}.{2}{3}", testDatabaseName, testDevice, testMeasurement, measurement))); - var data_type_lst = new List<TSDataType>() + status = await session_pool.DeleteDatabaseAsync(testDatabaseName); + var measurement_lst = new List<int> { 1, 2, 3, 4, 5, 6 }; + var ts_path_lst = new List<string>(measurement_lst.ConvertAll( + (measurement) => string.Format("{0}.{1}.{2}{3}", testDatabaseName, testDevice, testMeasurement, measurement))); + var data_type_lst = new List<TSDataType>() { TSDataType.BOOLEAN, TSDataType.INT32, TSDataType.INT64, TSDataType.FLOAT, TSDataType.DOUBLE, TSDataType.TEXT }; - var encoding_lst = new List<TSEncoding>() + var encoding_lst = new List<TSEncoding>() { TSEncoding.PLAIN, TSEncoding.PLAIN, TSEncoding.PLAIN, TSEncoding.PLAIN, TSEncoding.PLAIN, TSEncoding.PLAIN }; - var compressor_lst = new List<Compressor>() + var compressor_lst = new List<Compressor>() { Compressor.SNAPPY, Compressor.SNAPPY, Compressor.SNAPPY, Compressor.SNAPPY, Compressor.SNAPPY, Compressor.SNAPPY }; - status = await session_pool.CreateMultiTimeSeriesAsync(ts_path_lst, data_type_lst, encoding_lst, - compressor_lst); - System.Diagnostics.Debug.Assert(status == 0); - status = await session_pool.DeleteTimeSeriesAsync(ts_path_lst); - System.Diagnostics.Debug.Assert(status == 0); - Console.WriteLine("TestDeleteTimeSeries Passed!"); - status = await session_pool.DeleteDatabaseAsync(testDatabaseName); - await session_pool.Close(); - } - public async Task TestCreateTimeSeries() - { - var session_pool = new SessionPool(host, port, poolSize); - await session_pool.Open(false); - if (debug) session_pool.OpenDebugMode(); + status = await session_pool.CreateMultiTimeSeriesAsync(ts_path_lst, data_type_lst, encoding_lst, + compressor_lst); + System.Diagnostics.Debug.Assert(status == 0); + status = await session_pool.DeleteTimeSeriesAsync(ts_path_lst); + System.Diagnostics.Debug.Assert(status == 0); + Console.WriteLine("TestDeleteTimeSeries Passed!"); + status = await session_pool.DeleteDatabaseAsync(testDatabaseName); + await session_pool.Close(); + } + public async Task TestCreateTimeSeries() + { + 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(await session_pool.CreateTimeSeries( - string.Format("{0}.{1}.{2}", testDatabaseName, testDevice, testMeasurements[1]), - TSDataType.BOOLEAN, TSEncoding.PLAIN, Compressor.SNAPPY) == 0); - System.Diagnostics.Debug.Assert(await session_pool.CreateTimeSeries( - string.Format("{0}.{1}.{2}", testDatabaseName, testDevice, testMeasurements[2]), - TSDataType.INT32, TSEncoding.PLAIN, Compressor.SNAPPY) == 0); - System.Diagnostics.Debug.Assert(await session_pool.CreateTimeSeries( - string.Format("{0}.{1}.{2}", testDatabaseName, testDevice, testMeasurements[3]), - TSDataType.INT64, TSEncoding.PLAIN, Compressor.SNAPPY) == 0); - System.Diagnostics.Debug.Assert(await session_pool.CreateTimeSeries( - string.Format("{0}.{1}.{2}", testDatabaseName, testDevice, testMeasurements[4]), - TSDataType.FLOAT, TSEncoding.PLAIN, Compressor.SNAPPY) == 0); - System.Diagnostics.Debug.Assert(await session_pool.CreateTimeSeries( - string.Format("{0}.{1}.{2}", testDatabaseName, testDevice, testMeasurements[5]), - TSDataType.DOUBLE, TSEncoding.PLAIN, Compressor.SNAPPY) == 0); - System.Diagnostics.Debug.Assert(await session_pool.CreateTimeSeries( - string.Format("{0}.{1}.{2}", testDatabaseName, testDevice, testMeasurements[6]), - TSDataType.TEXT, TSEncoding.PLAIN, Compressor.SNAPPY) == 0); - await session_pool.DeleteDatabaseAsync(testDatabaseName); - await session_pool.Close(); - Console.WriteLine("TestCreateTimeSeries Passed!"); - } + await session_pool.DeleteDatabaseAsync(testDatabaseName); + System.Diagnostics.Debug.Assert(await session_pool.CreateTimeSeries( + string.Format("{0}.{1}.{2}", testDatabaseName, testDevice, testMeasurements[1]), + TSDataType.BOOLEAN, TSEncoding.PLAIN, Compressor.SNAPPY) == 0); + System.Diagnostics.Debug.Assert(await session_pool.CreateTimeSeries( + string.Format("{0}.{1}.{2}", testDatabaseName, testDevice, testMeasurements[2]), + TSDataType.INT32, TSEncoding.PLAIN, Compressor.SNAPPY) == 0); + System.Diagnostics.Debug.Assert(await session_pool.CreateTimeSeries( + string.Format("{0}.{1}.{2}", testDatabaseName, testDevice, testMeasurements[3]), + TSDataType.INT64, TSEncoding.PLAIN, Compressor.SNAPPY) == 0); + System.Diagnostics.Debug.Assert(await session_pool.CreateTimeSeries( + string.Format("{0}.{1}.{2}", testDatabaseName, testDevice, testMeasurements[4]), + TSDataType.FLOAT, TSEncoding.PLAIN, Compressor.SNAPPY) == 0); + System.Diagnostics.Debug.Assert(await session_pool.CreateTimeSeries( + string.Format("{0}.{1}.{2}", testDatabaseName, testDevice, testMeasurements[5]), + TSDataType.DOUBLE, TSEncoding.PLAIN, Compressor.SNAPPY) == 0); + System.Diagnostics.Debug.Assert(await session_pool.CreateTimeSeries( + string.Format("{0}.{1}.{2}", testDatabaseName, testDevice, testMeasurements[6]), + TSDataType.TEXT, TSEncoding.PLAIN, Compressor.SNAPPY) == 0); + await session_pool.DeleteDatabaseAsync(testDatabaseName); + await session_pool.Close(); + Console.WriteLine("TestCreateTimeSeries Passed!"); + } - public async Task TestCreateAlignedTimeseries() - { - var session_pool = new SessionPool(host, port, username, password, poolSize); - await session_pool.Open(false); - var status = 0; - if (debug) session_pool.OpenDebugMode(); + public async Task TestCreateAlignedTimeseries() + { + var session_pool = new SessionPool(host, port, username, password, poolSize); + await session_pool.Open(false); + var status = 0; + if (debug) session_pool.OpenDebugMode(); - status = await session_pool.DeleteDatabaseAsync(testDatabaseName); + status = await session_pool.DeleteDatabaseAsync(testDatabaseName); - string prefixPath = string.Format("{0}.{1}", testDatabaseName, testDevice); - var measurement_lst = new List<string>() + string prefixPath = string.Format("{0}.{1}", testDatabaseName, testDevice); + var measurement_lst = new List<string>() { testMeasurements[1], testMeasurements[2], @@ -147,51 +147,51 @@ testMeasurements[5], testMeasurements[6] }; - var data_type_lst = new List<TSDataType>() + var data_type_lst = new List<TSDataType>() { TSDataType.BOOLEAN, TSDataType.INT32, TSDataType.INT64, TSDataType.FLOAT, TSDataType.DOUBLE, TSDataType.TEXT }; - var encoding_lst = new List<TSEncoding>() + var encoding_lst = new List<TSEncoding>() { TSEncoding.PLAIN, TSEncoding.PLAIN, TSEncoding.PLAIN, TSEncoding.PLAIN, TSEncoding.PLAIN, TSEncoding.PLAIN }; - var compressor_lst = new List<Compressor>() + var compressor_lst = new List<Compressor>() { Compressor.SNAPPY, Compressor.SNAPPY, Compressor.SNAPPY, Compressor.SNAPPY, Compressor.SNAPPY, Compressor.SNAPPY }; - status = await session_pool.CreateAlignedTimeseriesAsync(prefixPath, measurement_lst, data_type_lst, encoding_lst, - compressor_lst); - System.Diagnostics.Debug.Assert(status == 0); - status = await session_pool.DeleteDatabaseAsync(testDatabaseName); - System.Diagnostics.Debug.Assert(status == 0); - await session_pool.Close(); - Console.WriteLine("TestCreateAlignedTimeSeries Passed!"); - } - public async Task TestCheckTimeSeriesExists() - { - 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.CreateAlignedTimeseriesAsync(prefixPath, measurement_lst, data_type_lst, encoding_lst, + compressor_lst); + System.Diagnostics.Debug.Assert(status == 0); + status = await session_pool.DeleteDatabaseAsync(testDatabaseName); + System.Diagnostics.Debug.Assert(status == 0); + await session_pool.Close(); + Console.WriteLine("TestCreateAlignedTimeSeries Passed!"); + } + public async Task TestCheckTimeSeriesExists() + { + 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()); - await session_pool.DeleteDatabaseAsync(testDatabaseName); - await session_pool.CreateTimeSeries( - string.Format("{0}.{1}.{2}", testDatabaseName, testDevice, testMeasurements[1]), - TSDataType.BOOLEAN, TSEncoding.PLAIN, Compressor.SNAPPY); - var ifExist_1 = await session_pool.CheckTimeSeriesExistsAsync( - string.Format("{0}.{1}.{2}", testDatabaseName, testDevice, testMeasurements[1])); - var ifExist_2 = await session_pool.CheckTimeSeriesExistsAsync( - string.Format("{0}.{1}.{2}", testDatabaseName, testDevice, testMeasurements[2])); - System.Diagnostics.Debug.Assert(ifExist_1 == true && ifExist_2 == false); - status = await session_pool.DeleteDatabaseAsync(testDatabaseName); - System.Diagnostics.Debug.Assert(status == 0); - await session_pool.Close(); - Console.WriteLine("TestCheckTimeSeriesExists Passed!"); + System.Diagnostics.Debug.Assert(session_pool.IsOpen()); + await session_pool.DeleteDatabaseAsync(testDatabaseName); + await session_pool.CreateTimeSeries( + string.Format("{0}.{1}.{2}", testDatabaseName, testDevice, testMeasurements[1]), + TSDataType.BOOLEAN, TSEncoding.PLAIN, Compressor.SNAPPY); + var ifExist_1 = await session_pool.CheckTimeSeriesExistsAsync( + string.Format("{0}.{1}.{2}", testDatabaseName, testDevice, testMeasurements[1])); + var ifExist_2 = await session_pool.CheckTimeSeriesExistsAsync( + string.Format("{0}.{1}.{2}", testDatabaseName, testDevice, testMeasurements[2])); + System.Diagnostics.Debug.Assert(ifExist_1 == true && ifExist_2 == false); + status = await session_pool.DeleteDatabaseAsync(testDatabaseName); + System.Diagnostics.Debug.Assert(status == 0); + await session_pool.Close(); + Console.WriteLine("TestCheckTimeSeriesExists Passed!"); + } } - } }
diff --git a/samples/Apache.IoTDB.Samples/SessionPoolTest.cs b/samples/Apache.IoTDB.Samples/SessionPoolTest.cs index 8b27c97..298063e 100644 --- a/samples/Apache.IoTDB.Samples/SessionPoolTest.cs +++ b/samples/Apache.IoTDB.Samples/SessionPoolTest.cs
@@ -192,7 +192,7 @@ 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] }); + var rowRecord = new RowRecord(1668404120807, new() { "1111111", "22222", "333333" }, new() { testMeasurements[0], testMeasurements[1], testMeasurements[2] }, new List<string>() { "TEXT", "TEXT", "TEXT" }); 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!"); @@ -212,7 +212,7 @@ 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] }); + var rowRecord = new RowRecord(1668404120807, new() { "1111111", "22222", "333333" }, new() { testMeasurements[0], testMeasurements[1], testMeasurements[2] }, new List<string>() { "TEXT", "TEXT", "TEXT" }); status = await session_pool.InsertRecordsAsync(new List<string>() { string.Format("{0}.{1}", testDatabaseName, testDevice) }, new List<RowRecord>() { rowRecord }); } public async Task TestGetTimeZone() @@ -310,22 +310,23 @@ testMeasurements[1], testMeasurements[2], testMeasurements[3] }; var values = new List<object> { "test_text", true, (int)123 }; + var dataTypes = new List<string>() { "TEXT", "BOOLEAN", "INT32" }; status = await session_pool.InsertRecordAsync( - string.Format("{0}.{1}", testDatabaseName, testDevice), new RowRecord(1, values, measures)); + string.Format("{0}.{1}", testDatabaseName, testDevice), new RowRecord(1, values, measures, dataTypes)); System.Diagnostics.Debug.Assert(status == 0); status = await session_pool.InsertRecordAsync( - string.Format("{0}.{1}", testDatabaseName, testDevice), new RowRecord(2, values, measures)); + string.Format("{0}.{1}", testDatabaseName, testDevice), new RowRecord(2, values, measures, dataTypes)); System.Diagnostics.Debug.Assert(status == 0); status = await session_pool.InsertRecordAsync( - string.Format("{0}.{1}", testDatabaseName, testDevice), new RowRecord(3, values, measures)); + string.Format("{0}.{1}", testDatabaseName, testDevice), new RowRecord(3, values, measures, dataTypes)); System.Diagnostics.Debug.Assert(status == 0); status = await session_pool.InsertRecordAsync( - string.Format("{0}.{1}", testDatabaseName, testDevice), new RowRecord(4, values, measures)); + string.Format("{0}.{1}", testDatabaseName, testDevice), new RowRecord(4, values, measures, dataTypes)); 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()); + while (await res.HasNextAsync()) Console.WriteLine(res.Next()); await res.Close(); var ts_path_lst = new List<string>() @@ -337,7 +338,7 @@ 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()); + while (await res.HasNextAsync()) Console.WriteLine(res.Next()); await res.Close(); status = await session_pool.DeleteDatabaseAsync(testDatabaseName); @@ -375,7 +376,7 @@ 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()); + while (await res.HasNextAsync()) Console.WriteLine(res.Next()); await res.Close(); status = await session_pool.DeleteDatabaseAsync(testDatabaseName); @@ -453,32 +454,32 @@ var res = await session_pool.ExecuteQueryStatementAsync("show timeseries root"); res.ShowTableNames(); - while (res.HasNext()) Console.WriteLine(res.Next()); + while (await res.HasNextAsync()) 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()); + while (await res.HasNextAsync()) 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()); + while (await res.HasNextAsync()) 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()); + while (await res.HasNextAsync()) 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()); + while (await res.HasNextAsync()) Console.WriteLine(res.Next()); await res.Close(); status = await session_pool.DeleteDatabaseAsync(testDatabaseName); @@ -506,9 +507,10 @@ var records = new List<RowRecord>(); var values = new List<object>() { true, 20.0f }; var device_id_lst = new List<string>() { }; + var dataTypes = new List<string>() { "BOOLEAN", "FLOAT" }; for (int i = 1; i <= fetchSize * processedSize; i++) { - var record = new RowRecord(i, values, measurements); + var record = new RowRecord(i, values, measurements, dataTypes); records.Add(record); device_id_lst.Add(device_id); } @@ -519,7 +521,7 @@ var res = await session_pool.ExecuteRawDataQuery(paths, 10, fetchSize * processedSize); var count = 0; - while (res.HasNext()) + while (await res.HasNextAsync()) { var record = res.Next(); count++; @@ -553,9 +555,10 @@ var records = new List<RowRecord>(); var values = new List<object>() { true, 20.0f }; var device_id_lst = new List<string>() { }; + var dataTypes = new List<string>() { "BOOLEAN", "FLOAT" }; for (int i = 1; i <= fetchSize * processedSize; i++) { - var record = new RowRecord(i, values, measurements); + var record = new RowRecord(i, values, measurements, dataTypes); records.Add(record); device_id_lst.Add(device_id); } @@ -566,7 +569,7 @@ var res = await session_pool.ExecuteLastDataQueryAsync(paths, fetchSize * processedSize - 10); var count = 0; - while (res.HasNext()) + while (await res.HasNextAsync()) { var record = res.Next(); Console.WriteLine(record); @@ -608,9 +611,10 @@ var records = new List<RowRecord>(); var values = new List<object>() { true, 20.0f }; var device_id_lst = new List<string>() { }; + var dataTypes = new List<string>() { "BOOLEAN", "FLOAT" }; for (int i = 1; i <= fetchSize * processedSize * 4 + 783; i++) { - var record = new RowRecord(i, values, measurements); + var record = new RowRecord(i, values, measurements, dataTypes); records.Add(record); device_id_lst.Add(device_id); } @@ -623,7 +627,7 @@ var res = await session_pool.ExecuteQueryStatementAsync("select * from " + string.Format("{0}.{1}", testDatabaseName, testDevice)); res.ShowTableNames(); var count = 0; - while (res.HasNext()) + while (await res.HasNextAsync()) { var record = res.Next(); count++;
diff --git a/src/Apache.IoTDB.Data/IoTDBCommand.cs b/src/Apache.IoTDB.Data/IoTDBCommand.cs index 1d3c2b9..997c52a 100644 --- a/src/Apache.IoTDB.Data/IoTDBCommand.cs +++ b/src/Apache.IoTDB.Data/IoTDBCommand.cs
@@ -368,6 +368,7 @@ { var measures = new List<string>(); var values = new List<object>(); + var dataTypes = new List<TSDataType>(); for (int i = 0; i < pms.Count; i++) @@ -380,13 +381,16 @@ { case TypeCode.Boolean: values.Add((tp.Value as bool?).GetValueOrDefault()); + dataTypes.Add(TSDataType.BOOLEAN); break; case TypeCode.Char: values.Add(tp.Value as string); + dataTypes.Add(TSDataType.TEXT); break; case TypeCode.Byte: case TypeCode.SByte: values.Add((tp.Value as byte?).GetValueOrDefault()); + dataTypes.Add(TSDataType.INT32); break; case TypeCode.DateTime: var t0 = tp.Value as DateTime?; @@ -395,43 +399,53 @@ throw new ArgumentException($"InvalidArgumentOfDateTime{tp.Value}"); } values.Add(t0.GetValueOrDefault()); + dataTypes.Add(TSDataType.DATE); break; case TypeCode.DBNull: break; case TypeCode.Single: values.Add((tp.Value as float?).GetValueOrDefault()); + dataTypes.Add(TSDataType.FLOAT); break; case TypeCode.Decimal: case TypeCode.Double: values.Add((tp.Value as double?).GetValueOrDefault()); + dataTypes.Add(TSDataType.DOUBLE); break; case TypeCode.Int16: values.Add((tp.Value as short?).GetValueOrDefault()); + dataTypes.Add(TSDataType.INT32); break; case TypeCode.Int32: values.Add((tp.Value as int?).GetValueOrDefault()); + dataTypes.Add(TSDataType.INT32); break; case TypeCode.Int64: values.Add((tp.Value as long?).GetValueOrDefault()); + dataTypes.Add(TSDataType.INT64); break; case TypeCode.UInt16: values.Add((tp.Value as short?).GetValueOrDefault()); + dataTypes.Add(TSDataType.INT32); break; case TypeCode.UInt32: values.Add((tp.Value as uint?).GetValueOrDefault()); + dataTypes.Add(TSDataType.INT64); break; case TypeCode.UInt64: values.Add((tp.Value as ulong?).GetValueOrDefault()); + dataTypes.Add(TSDataType.INT64); break; case TypeCode.String: default: values.Add(tp.Value as string); + dataTypes.Add(TSDataType.TEXT); break; } } - return new RowRecord(DateTimeOffset.UtcNow.ToUnixTimeMilliseconds(), values, measures); + return new RowRecord(DateTimeOffset.UtcNow.ToUnixTimeMilliseconds(), values, measures, dataTypes); } /// <summary>
diff --git a/src/Apache.IoTDB.Data/IoTDBDataReader.cs b/src/Apache.IoTDB.Data/IoTDBDataReader.cs index 7076a68..be45542 100644 --- a/src/Apache.IoTDB.Data/IoTDBDataReader.cs +++ b/src/Apache.IoTDB.Data/IoTDBDataReader.cs
@@ -136,8 +136,10 @@ { throw new InvalidOperationException($"DataReaderClosed{nameof(Read)}"); } +#pragma warning disable CS0618 // Sync HasNext() required by ADO.NET DbDataReader.Read() interface if (_dataSet.HasNext()) { +#pragma warning restore CS0618 rowdata = _dataSet.Next(); } else @@ -454,7 +456,9 @@ /// <returns>A System.Data.DataTable that describes the column metadata.</returns> public override DataTable GetSchemaTable() { +#pragma warning disable CS0618 // Sync HasNext() required by ADO.NET interface if (_dataSet.HasNext()) +#pragma warning restore CS0618 { rowdata = _dataSet.GetRow(); }
diff --git a/src/Apache.IoTDB/DataStructure/SessionDataSet.cs b/src/Apache.IoTDB/DataStructure/SessionDataSet.cs index 75f47d0..2cc4aa9 100644 --- a/src/Apache.IoTDB/DataStructure/SessionDataSet.cs +++ b/src/Apache.IoTDB/DataStructure/SessionDataSet.cs
@@ -145,6 +145,7 @@ Console.WriteLine(str); } + [Obsolete("Use HasNextAsync() instead. This synchronous method blocks on async calls and can cause deadlocks.")] public bool HasNext() { if (_hasCatchedResult) @@ -166,11 +167,34 @@ return true; } + public async Task<bool> HasNextAsync() + { + if (_hasCatchedResult) + { + return true; + } + + // we have consumed all current data, fetch some more + if (!_timeBuffer.HasRemaining()) + { + if (!await FetchResultsAsync()) + { + return false; + } + } + + ConstructOneRow(); + _hasCatchedResult = true; + return true; + } + public RowRecord Next() { if (!_hasCatchedResult) { +#pragma warning disable CS0618 if (!HasNext()) +#pragma warning restore CS0618 { return null; } @@ -278,7 +302,13 @@ long timestamp = _timeBuffer.GetLong(); _rowIndex += 1; - _cachedRowRecord = new RowRecord(timestamp, fieldLst, _columnNames); + + var dataTypes = new List<TSDataType>(); + for (int j = 0; j < _columnSize; j++) + { + dataTypes.Add(GetDataTypeFromStr(_columnTypeLst[j])); + } + _cachedRowRecord = new RowRecord(timestamp, fieldLst, _columnNames, dataTypes); } private bool IsNull(int loc, int row_index) @@ -328,6 +358,44 @@ } } + private async Task<bool> FetchResultsAsync() + { + _rowIndex = 0; + var req = new TSFetchResultsReq(_client.SessionId, _sql, FetchSize, _queryId, true) + { + Timeout = DefaultTimeout + }; + try + { + var resp = await _client.ServiceClient.fetchResultsAsync(req); + + if (resp.HasResultSet) + { + _queryDataset = resp.QueryDataSet; + // reset buffer + _timeBuffer = new ByteBuffer(resp.QueryDataSet.Time); + _valueBufferLst = new List<ByteBuffer>(); + _bitmapBufferLst = new List<ByteBuffer>(); + for (int index = 0; index < _queryDataset.ValueList.Count; index++) + { + string columnName = _columnNames[index]; + int valueIndex = _columnNameIndexMap[columnName]; + _valueBufferLst.Add(new ByteBuffer(_queryDataset.ValueList[valueIndex])); + _bitmapBufferLst.Add(new ByteBuffer(_queryDataset.BitmapList[valueIndex])); + } + + // reset row index + _rowIndex = 0; + } + + return resp.HasResultSet; + } + catch (TException e) + { + throw new TException("Cannot fetch result from server, because of network connection", e); + } + } + public async Task Close() { if (!_isClosed)
diff --git a/src/Apache.IoTDB/Rpc/TSStatusCode.cs b/src/Apache.IoTDB/Rpc/TSStatusCode.cs index e2eaca8..3f6cee9 100644 --- a/src/Apache.IoTDB/Rpc/TSStatusCode.cs +++ b/src/Apache.IoTDB/Rpc/TSStatusCode.cs
@@ -285,4 +285,4 @@ } } -} \ No newline at end of file +}
diff --git a/src/Apache.IoTDB/SessionPool.cs b/src/Apache.IoTDB/SessionPool.cs index 2d3e4a1..ff202c8 100644 --- a/src/Apache.IoTDB/SessionPool.cs +++ b/src/Apache.IoTDB/SessionPool.cs
@@ -702,7 +702,7 @@ { var sql = "SHOW TIMESERIES " + tsPath; var sessionDataset = await ExecuteQueryStatementAsync(sql); - bool timeSeriesExists = sessionDataset.HasNext(); + bool timeSeriesExists = await sessionDataset.HasNextAsync(); await sessionDataset.Close(); // be sure to close the sessionDataset to put the client back to the pool return timeSeriesExists; }