Skip to content

[V2.0.1]Wx table session #26

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 3 commits into from
Feb 28, 2025
Merged

[V2.0.1]Wx table session #26

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
c967c96
update Rpc to IoTDB 2.0.1
CritasWang Feb 12, 2025
40d9afd
feat: add TableSessionPool
CritasWang Feb 13, 2025
5b70f93
update iotdb-version
CritasWang Feb 18, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions docker-compose.yml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,7 @@ services:
ipv4_address: 172.18.0.2

iotdb:
image: apache/iotdb:1.3.3-datanode
image: apache/iotdb:2.0.1-beta-datanode
restart: always
container_name: iotdb-dn-1
depends_on:
Expand All @@ -39,7 +39,7 @@ services:
- dn_seed_config_node=iotdb-confignode-1:22277

iotdb-confignode-1:
image: apache/iotdb:1.3.3-confignode
image: apache/iotdb:2.0.1-beta-confignode
restart: always
container_name: iotdb-cn-1
healthcheck:
Expand Down
10 changes: 10 additions & 0 deletions samples/Apache.IoTDB.Samples/Program.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,8 @@ public static async Task Main(string[] args)
utilsTest.TestParseEndPoint();
var sessionPoolTest = new SessionPoolTest("iotdb");
await sessionPoolTest.Test();
var tableSessionPoolTest = new TableSessionPoolTest(sessionPoolTest);
await tableSessionPoolTest.Test();
}

public static void OpenDebugMode(this SessionPool session)
Expand All @@ -42,5 +44,13 @@ public static void OpenDebugMode(this SessionPool session)
builder.AddNLog();
});
}
public static void OpenDebugMode(this TableSessionPool session)
{
session.OpenDebugMode(builder =>
{
builder.AddConsole();
builder.AddNLog();
});
}
}
}
220 changes: 110 additions & 110 deletions samples/Apache.IoTDB.Samples/SessionPoolTest.AlignedRecord.cs
View file
Open in desktop

Large diffs are not rendered by default.

56 changes: 28 additions & 28 deletions samples/Apache.IoTDB.Samples/SessionPoolTest.AlignedTablet.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,18 +29,18 @@ public partial class SessionPoolTest
{
public async Task TestInsertAlignedTablet()
{
var session_pool = new SessionPool(host, port, pool_size);
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(test_database_name);
var device_id = string.Format("{0}.{1}", test_database_name, test_device);
await session_pool.DeleteDatabaseAsync(testDatabaseName);
var device_id = string.Format("{0}.{1}", testDatabaseName, testDevice);
var measurement_lst = new List<string>
{ test_measurements[1],
test_measurements[2],
test_measurements[3]
{ testMeasurements[1],
testMeasurements[2],
testMeasurements[3]
};
var value_lst = new List<List<object>>
{
Expand All @@ -53,7 +53,7 @@ public async Task TestInsertAlignedTablet()
status = await session_pool.InsertAlignedTabletAsync(tablet);
System.Diagnostics.Debug.Assert(status == 0);
var res = await session_pool.ExecuteQueryStatementAsync(
"select * from " + string.Format("{0}.{1}", test_database_name, test_device) + " where time<15");
"select * from " + string.Format("{0}.{1}", testDatabaseName, testDevice) + " where time<15");
res.ShowTableNames();
while (res.HasNext()) Console.WriteLine(res.Next());

Expand All @@ -63,11 +63,11 @@ public async Task TestInsertAlignedTablet()
timestamp_lst = new List<long>() { };
var tasks = new List<Task<int>>();
var start_ms = DateTime.Now.Ticks / 10000;
for (var timestamp = 4; timestamp <= fetch_size * processed_size; timestamp++)
for (var timestamp = 4; timestamp <= fetchSize * processedSize; timestamp++)
{
timestamp_lst.Add(timestamp);
value_lst.Add(new List<object>() { "iotdb", true, (int)timestamp });
if (timestamp % fetch_size == 0)
if (timestamp % fetchSize == 0)
{
tablet = new Tablet(device_id, measurement_lst, datatype_lst, value_lst, timestamp_lst);
tasks.Add(session_pool.InsertAlignedTabletAsync(tablet));
Expand All @@ -81,7 +81,7 @@ public async Task TestInsertAlignedTablet()
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}", test_database_name, test_device));
"select * from " + string.Format("{0}.{1}", testDatabaseName, testDevice));
res.ShowTableNames();
var res_count = 0;
while (res.HasNext())
Expand All @@ -91,32 +91,32 @@ public async Task TestInsertAlignedTablet()
}

await res.Close();
Console.WriteLine(res_count + " " + fetch_size * processed_size);
System.Diagnostics.Debug.Assert(res_count == fetch_size * processed_size);
status = await session_pool.DeleteDatabaseAsync(test_database_name);
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, pool_size);
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(test_database_name);
await session_pool.DeleteDatabaseAsync(testDatabaseName);
var device_id = new List<string>()
{
string.Format("{0}.{1}", test_database_name, test_devices[1]),
string.Format("{0}.{1}", test_database_name, test_devices[2])
string.Format("{0}.{1}", testDatabaseName, testDevices[1]),
string.Format("{0}.{1}", testDatabaseName, testDevices[2])
};
var measurements_lst = new List<List<string>>()
{
new() {test_measurements[1], test_measurements[2], test_measurements[3] },
new() {test_measurements[1], test_measurements[2], test_measurements[3] }
new() {testMeasurements[1], testMeasurements[2], testMeasurements[3] },
new() {testMeasurements[1], testMeasurements[2], testMeasurements[3] }
};
var values_lst = new List<List<List<object>>>()
{
Expand Down Expand Up @@ -148,24 +148,24 @@ public async Task TestInsertAlignedTablets()
status = await session_pool.InsertAlignedTabletsAsync(tablets);
System.Diagnostics.Debug.Assert(status == 0);
var res = await session_pool.ExecuteQueryStatementAsync(
"select * from " + string.Format("{0}.{1}", test_database_name, test_devices[1]) + " where time<15");
"select * from " + string.Format("{0}.{1}", testDatabaseName, testDevices[1]) + " where time<15");
res.ShowTableNames();
while (res.HasNext()) Console.WriteLine(res.Next());

// large data test
var tasks = new List<Task<int>>();
// tablets = new List<Tablet>() { };
for (var timestamp = 4; timestamp <= processed_size * fetch_size; timestamp++)
for (var timestamp = 4; timestamp <= processedSize * fetchSize; timestamp++)
{
var local_device_id = string.Format("{0}.{1}", test_database_name, test_devices[1]);
var local_device_id = string.Format("{0}.{1}", testDatabaseName, testDevices[1]);
var local_measurements = new List<string>()
{test_measurements[1], test_measurements[2], test_measurements[3]};
{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 % fetch_size == 0)
if (timestamp % fetchSize == 0)
{
tasks.Add(session_pool.InsertAlignedTabletsAsync(tablets));
tablets = new List<Tablet>() { };
Expand All @@ -174,7 +174,7 @@ public async Task TestInsertAlignedTablets()

Task.WaitAll(tasks.ToArray());
res = await session_pool.ExecuteQueryStatementAsync(
"select * from " + string.Format("{0}.{1}", test_database_name, test_devices[1]));
"select * from " + string.Format("{0}.{1}", testDatabaseName, testDevices[1]));
res.ShowTableNames();
var res_count = 0;
while (res.HasNext())
Expand All @@ -184,9 +184,9 @@ public async Task TestInsertAlignedTablets()
}

await res.Close();
Console.WriteLine(res_count + " " + fetch_size * processed_size);
System.Diagnostics.Debug.Assert(res_count == fetch_size * processed_size);
status = await session_pool.DeleteDatabaseAsync(test_database_name);
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!");
Expand Down
Loading
Loading