Skip to content

Fix RpcDataSet deadlock, timezone compatibility, column index issues, and deprecation cleanup #48

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

Open
CritasWang wants to merge 2 commits into
base: main
Choose a base branch
from
Open

Fix RpcDataSet deadlock, timezone compatibility, column index issues, and deprecation cleanup #48

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
6e2c057
Fix RpcDataSet deadlock, timezone compatibility, and column index issues
CritasWang Feb 26, 2026
9fb873a
Fix GetRow() type handling, GetInt overflow, and timezone fallback in…
CritasWang Feb 27, 2026
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
2 changes: 1 addition & 1 deletion Apache-IoTDB-Client-CSharp-UserCase/Program.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -99,7 +99,7 @@ static async Task ExecuteQueryStatement()
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());
}
Expand Down
74 changes: 54 additions & 20 deletions samples/Apache.IoTDB.Samples/SessionPoolTest.AlignedRecord.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,7 +28,11 @@ public partial class SessionPoolTest
{
public async Task TestInsertAlignedRecord()
{
var session_pool = new SessionPool(host, port, poolSize);
var session_pool = new SessionPool.Builder()
.SetHost(host)
.SetPort(port)
.SetPoolSize(poolSize)
.Build();
int status;
await session_pool.Open(false);
if (debug) session_pool.OpenDebugMode();
Expand All @@ -51,7 +55,7 @@ public async Task TestInsertAlignedRecord()
var start_ms = DateTime.Now.Ticks / 10000;
for (var timestamp = 1; timestamp <= fetchSize * processedSize; timestamp++)
{
var rowRecord = new RowRecord(timestamp, values, measures);
var rowRecord = new RowRecord(timestamp, values, measures, new List<TSDataType> { TSDataType.TEXT, TSDataType.BOOLEAN, TSDataType.INT32 });
var task = session_pool.InsertAlignedRecordAsync(
string.Format("{0}.{1}", testDatabaseName, testDevice), rowRecord);
tasks.Add(task);
Expand All @@ -65,7 +69,11 @@ public async Task TestInsertAlignedRecord()
}
public async Task TestInsertAlignedStringRecord()
{
var session_pool = new SessionPool(host, port, poolSize);
var session_pool = new SessionPool.Builder()
.SetHost(host)
.SetPort(port)
.SetPoolSize(poolSize)
.Build();
var status = 0;
await session_pool.Open(false);
if (debug) session_pool.OpenDebugMode();
Expand Down Expand Up @@ -98,7 +106,7 @@ public async Task TestInsertAlignedStringRecord()
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())
while (await res.HasNextAsync())
{
res_cnt++;
res.Next();
Expand All @@ -111,7 +119,11 @@ public async Task TestInsertAlignedStringRecord()
}
public async Task TestInsertAlignedRecords()
{
var session_pool = new SessionPool(host, port, poolSize);
var session_pool = new SessionPool.Builder()
.SetHost(host)
.SetPort(port)
.SetPoolSize(poolSize)
.Build();
await session_pool.Open(false);
if (debug) session_pool.OpenDebugMode();

Expand Down Expand Up @@ -168,6 +180,10 @@ public async Task TestInsertAlignedRecords()
testMeasurements[5],
testMeasurements[6]
});
var dataTypes_lst = new List<List<TSDataType>>() { };
dataTypes_lst.Add(new List<TSDataType>() { TSDataType.BOOLEAN, TSDataType.INT32 });
dataTypes_lst.Add(new List<TSDataType>() { TSDataType.BOOLEAN, TSDataType.INT32, TSDataType.INT64, TSDataType.DOUBLE });
dataTypes_lst.Add(new List<TSDataType>() { TSDataType.BOOLEAN, TSDataType.INT32, TSDataType.INT64, TSDataType.DOUBLE, TSDataType.FLOAT, TSDataType.TEXT });
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 });
Expand All @@ -177,15 +193,15 @@ public async Task TestInsertAlignedRecords()
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]);
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");
SessionPoolTest.PrintDataSetByString(res);
await SessionPoolTest.PrintDataSetByString(res);
Console.WriteLine(rowRecords);

System.Diagnostics.Debug.Assert(true);
Expand All @@ -201,7 +217,8 @@ public async Task TestInsertAlignedRecords()
{
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>() { testMeasurements[1], testMeasurements[2] },
new List<TSDataType>() { TSDataType.BOOLEAN, TSDataType.INT32 }));
if (timestamp % fetchSize == 0)
{
tasks.Add(session_pool.InsertAlignedRecordsAsync(device_id, rowRecords));
Expand All @@ -216,7 +233,7 @@ public async Task TestInsertAlignedRecords()
res.ShowTableNames();
var record_count = fetchSize * processedSize;
var res_count = 0;
while (res.HasNext())
while (await res.HasNextAsync())
{
res_count += 1;
Console.WriteLine(res.Next());
Expand All @@ -234,7 +251,11 @@ public async Task TestInsertAlignedRecords()
}
public async Task TestInsertAlignedStringRecords()
{
var session_pool = new SessionPool(host, port, poolSize);
var session_pool = new SessionPool.Builder()
.SetHost(host)
.SetPort(port)
.SetPoolSize(poolSize)
.Build();
await session_pool.Open(false);
if (debug) session_pool.OpenDebugMode();

Expand Down Expand Up @@ -267,7 +288,7 @@ public async Task TestInsertAlignedStringRecords()
System.Diagnostics.Debug.Assert(status == 0);
var res = await session_pool.ExecuteQueryStatementAsync(
"select * from " + string.Format("{0}.{1}", testDatabaseName, testDevice) + " where time<10");
SessionPoolTest.PrintDataSetByString(res);
await SessionPoolTest.PrintDataSetByString(res);

await res.Close();

Expand Down Expand Up @@ -298,7 +319,7 @@ public async Task TestInsertAlignedStringRecords()
"select * from " + string.Format("{0}.{1}", testDatabaseName, testDevice));
res.ShowTableNames();
var res_count = 0;
while (res.HasNext())
while (await res.HasNextAsync())
{
Console.WriteLine(res.Next());
res_count += 1;
Expand All @@ -314,7 +335,11 @@ public async Task TestInsertAlignedStringRecords()
}
public async Task TestInsertAlignedRecordsOfOneDevice()
{
var session_pool = new SessionPool(host, port, poolSize);
var session_pool = new SessionPool.Builder()
.SetHost(host)
.SetPort(port)
.SetPoolSize(poolSize)
.Build();
await session_pool.Open(false);
if (debug) session_pool.OpenDebugMode();

Expand Down Expand Up @@ -377,25 +402,30 @@ public async Task TestInsertAlignedRecordsOfOneDevice()
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 dataTypes_lst = new List<List<TSDataType>>() { };
dataTypes_lst.Add(new List<TSDataType>() { TSDataType.BOOLEAN, TSDataType.INT32 });
dataTypes_lst.Add(new List<TSDataType>() { TSDataType.BOOLEAN, TSDataType.INT32, TSDataType.INT64, TSDataType.DOUBLE });
dataTypes_lst.Add(new List<TSDataType>() { TSDataType.BOOLEAN, TSDataType.INT32, TSDataType.INT64, TSDataType.DOUBLE, TSDataType.FLOAT, TSDataType.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]);
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");
SessionPoolTest.PrintDataSetByString(res);
await SessionPoolTest.PrintDataSetByString(res);

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] }));
new List<string>() { testMeasurements[1], testMeasurements[2] },
new List<TSDataType>() { TSDataType.BOOLEAN, TSDataType.INT32 }));
if (timestamp % fetchSize == 0)
{
tasks.Add(session_pool.InsertAlignedRecordsOfOneDeviceAsync(device_id, rowRecords));
Expand All @@ -407,7 +437,7 @@ public async Task TestInsertAlignedRecordsOfOneDevice()
res = await session_pool.ExecuteQueryStatementAsync(
"select * from " + string.Format("{0}.{1}", testDatabaseName, testDevice));
var res_count = 0;
while (res.HasNext())
while (await res.HasNextAsync())
{
res_count += 1;
res.Next();
Expand All @@ -423,7 +453,11 @@ public async Task TestInsertAlignedRecordsOfOneDevice()
}
public async Task TestInsertAlignedStringRecordsOfOneDevice()
{
var session_pool = new SessionPool(host, port, poolSize);
var session_pool = new SessionPool.Builder()
.SetHost(host)
.SetPort(port)
.SetPoolSize(poolSize)
.Build();
await session_pool.Open(false);
if (debug) session_pool.OpenDebugMode();

Expand Down Expand Up @@ -454,7 +488,7 @@ public async Task TestInsertAlignedStringRecordsOfOneDevice()
System.Diagnostics.Debug.Assert(status == 0);
var res = await session_pool.ExecuteQueryStatementAsync(
"select * from " + string.Format("{0}.{1}", testDatabaseName, testDevice) + " where time<10");
SessionPoolTest.PrintDataSetByString(res);
await SessionPoolTest.PrintDataSetByString(res);

await res.Close();
// large data test
Expand All @@ -480,7 +514,7 @@ public async Task TestInsertAlignedStringRecordsOfOneDevice()
res = await session_pool.ExecuteQueryStatementAsync(
"select * from " + string.Format("{0}.{1}", testDatabaseName, testDevice));
var res_count = 0;
while (res.HasNext())
while (await res.HasNextAsync())
{
res_count += 1;
res.Next();
Expand Down
20 changes: 14 additions & 6 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,7 +29,11 @@ public partial class SessionPoolTest
{
public async Task TestInsertAlignedTablet()
{
var session_pool = new SessionPool(host, port, poolSize);
var session_pool = new SessionPool.Builder()
.SetHost(host)
.SetPort(port)
.SetPoolSize(poolSize)
.Build();
var status = 0;
await session_pool.Open(false);
if (debug) session_pool.OpenDebugMode();
Expand All @@ -54,7 +58,7 @@ public async Task TestInsertAlignedTablet()
System.Diagnostics.Debug.Assert(status == 0);
var res = await session_pool.ExecuteQueryStatementAsync(
"select * from " + string.Format("{0}.{1}", testDatabaseName, testDevice) + " where time<15");
SessionPoolTest.PrintDataSetByString(res);
await SessionPoolTest.PrintDataSetByString(res);

await res.Close();
// large data test
Expand Down Expand Up @@ -83,7 +87,7 @@ public async Task TestInsertAlignedTablet()
"select * from " + string.Format("{0}.{1}", testDatabaseName, testDevice));
res.ShowTableNames();
var res_count = 0;
while (res.HasNext())
while (await res.HasNextAsync())
{
res_count += 1;
res.Next();
Expand All @@ -100,7 +104,11 @@ public async Task TestInsertAlignedTablet()

public async Task TestInsertAlignedTablets()
{
var session_pool = new SessionPool(host, port, poolSize);
var session_pool = new SessionPool.Builder()
.SetHost(host)
.SetPort(port)
.SetPoolSize(poolSize)
.Build();
var status = 0;
await session_pool.Open(false);
if (debug) session_pool.OpenDebugMode();
Expand Down Expand Up @@ -148,7 +156,7 @@ public async Task TestInsertAlignedTablets()
System.Diagnostics.Debug.Assert(status == 0);
var res = await session_pool.ExecuteQueryStatementAsync(
"select * from " + string.Format("{0}.{1}", testDatabaseName, testDevices[1]) + " where time<15");
SessionPoolTest.PrintDataSetByString(res);
await SessionPoolTest.PrintDataSetByString(res);

// large data test
var tasks = new List<Task<int>>();
Expand All @@ -175,7 +183,7 @@ public async Task TestInsertAlignedTablets()
"select * from " + string.Format("{0}.{1}", testDatabaseName, testDevices[1]));
res.ShowTableNames();
var res_count = 0;
while (res.HasNext())
while (await res.HasNextAsync())
{
res_count += 1;
res.Next();
Expand Down
Loading
Loading