From f0a1e03849eefe0c57010b87757dd71b60b8517b Mon Sep 17 00:00:00 2001 From: Krzysztof Dusko Date: Wed, 23 Sep 2026 20:08:01 +0200 Subject: [PATCH 1/2] Add streaming row API and structured diagnostics --- README.md | 17 + .../FieldAccessBench.cs | 64 ++++ .../README.md | 30 +- .../AsyncTests.cs | 44 +++ .../BackendDiagnosticResponseParserTests.cs | 49 +++ .../BasicTests.cs | 303 +----------------- .../CommandAbortTest.cs | 3 +- .../InvalidSqlTests.cs | 29 ++ .../JustyBase.NetezzaDriver.Tests.csproj | 4 +- .../NoticeTests.cs | 9 +- .../NzpyExtendedReference.cs | 84 +++++ .../NzpyExtendedResultCompatibilityTests.cs | 88 +++++ src/JustyBase.NetezzaDriver.Tests/README.md | 17 +- src/JustyBase.NetezzaDriver.Tests/SslTest.cs | 12 +- .../TemporaryCertificateFile.cs | 41 +++ .../nzpy-extended-reference.py | 58 ++++ .../BackendDiagnosticResponse.cs | 86 +++++ .../JustyBase.NetezzaDriver.csproj | 2 +- src/JustyBase.NetezzaDriver/NzCommand.cs | 37 ++- src/JustyBase.NetezzaDriver/NzConnection.cs | 93 ++++-- 20 files changed, 730 insertions(+), 340 deletions(-) create mode 100644 src/JustyBase.NetezzaDriver.Benchmarks/FieldAccessBench.cs create mode 100644 src/JustyBase.NetezzaDriver.Tests/BackendDiagnosticResponseParserTests.cs create mode 100644 src/JustyBase.NetezzaDriver.Tests/NzpyExtendedReference.cs create mode 100644 src/JustyBase.NetezzaDriver.Tests/NzpyExtendedResultCompatibilityTests.cs create mode 100644 src/JustyBase.NetezzaDriver.Tests/TemporaryCertificateFile.cs create mode 100644 src/JustyBase.NetezzaDriver.Tests/nzpy-extended-reference.py create mode 100644 src/JustyBase.NetezzaDriver/BackendDiagnosticResponse.cs diff --git a/README.md b/README.md index dae76f1..1bab5c4 100644 --- a/README.md +++ b/README.md @@ -247,6 +247,23 @@ Sample results (net10.0, BenchmarkDotNet, Windows 11): Async overhead is negligible (~1–3% time, 0–5% allocations). +## Streaming mapped rows and server diagnostics + +`NzCommand.ExecuteRowsAsync` maps each row while the reader is active, so callers can process large result sets without materializing them first: + +```csharp +await using var command = connection.CreateCommand("SELECT ID, NAME FROM MY_TABLE"); +await foreach (var item in command.ExecuteRowsAsync( + reader => new Item(reader.GetInt64(0), reader.GetString(1)), cancellationToken)) +{ + await ProcessAsync(item, cancellationToken); +} +``` + +The mapper is synchronous and the returned `IAsyncEnumerable` streams rows; cancellation is passed to database reads, and disposing the enumeration closes the reader. `command.Notices` contains the notices from that command's most recent execution. The existing connection-wide `NoticeReceived` event remains available. + +For server failures, `NetezzaException.Message` contains the primary backend message. `SqlState`, `Severity`, `Detail`, and `Hint` expose structured fields when supplied. `RawResponse` keeps the complete decoded backend payload, and `Diagnostics` provides all fields by protocol code, including fields unknown to this driver. + ## Testing ```bash diff --git a/src/JustyBase.NetezzaDriver.Benchmarks/FieldAccessBench.cs b/src/JustyBase.NetezzaDriver.Benchmarks/FieldAccessBench.cs new file mode 100644 index 0000000..c22e20c --- /dev/null +++ b/src/JustyBase.NetezzaDriver.Benchmarks/FieldAccessBench.cs @@ -0,0 +1,64 @@ +using BenchmarkDotNet.Attributes; + +namespace JustyBase.NetezzaDriver.Benchmarks; + +/// +/// Measures field-access choices while streaming the same deterministic rows +/// from a live Netezza table. The returned checksum guards against accidentally +/// benchmarking a path that does not consume the values. +/// +[MemoryDiagnoser] +public class FieldAccessBench +{ + private const string Query = "SELECT ROWID::BIGINT AS ID64 FROM JUST_DATA..FACTPRODUCTINVENTORY ORDER BY ROWID LIMIT 100000"; + private NzConnection _connection = null!; + + [GlobalSetup] + public void Setup() + { + _connection = new NzConnection(Config.UserName, Config.Password, Config.Host, Config.DbName, Config.Port); + _connection.Open(); + } + + [GlobalCleanup] + public void Cleanup() => _connection.Dispose(); + + [Benchmark(Baseline = true)] + public long GetValueAndCast() + { + using var command = _connection.CreateCommand(Query); + using var reader = command.ExecuteReader(); + long checksum = 0; + while (reader.Read()) + { + checksum = unchecked(checksum + (long)reader.GetValue(0)); + } + return checksum; + } + + [Benchmark] + public long GenericGetFieldValue() + { + using var command = _connection.CreateCommand(Query); + using var reader = command.ExecuteReader(); + long checksum = 0; + while (reader.Read()) + { + checksum = unchecked(checksum + reader.GetFieldValue(0)); + } + return checksum; + } + + [Benchmark] + public long TypedGetInt64() + { + using var command = _connection.CreateCommand(Query); + using var reader = command.ExecuteReader(); + long checksum = 0; + while (reader.Read()) + { + checksum = unchecked(checksum + reader.GetInt64(0)); + } + return checksum; + } +} diff --git a/src/JustyBase.NetezzaDriver.Benchmarks/README.md b/src/JustyBase.NetezzaDriver.Benchmarks/README.md index 4150d23..ad95894 100644 --- a/src/JustyBase.NetezzaDriver.Benchmarks/README.md +++ b/src/JustyBase.NetezzaDriver.Benchmarks/README.md @@ -16,4 +16,32 @@ | Method | Mean | Error | StdDev | Allocated | |---------------------------------- |--------:|---------:|---------:|----------:| | ExternalUnloadAndLoadNz | 1.690 s | 0.0165 s | 0.0138 s | 28.6 KB | -| ExternalUnloadAndLoadOriginalOdbc | 1.683 s | 0.0058 s | 0.0054 s | 7.9 KB | \ No newline at end of file +| ExternalUnloadAndLoadOriginalOdbc | 1.683 s | 0.0058 s | 0.0054 s | 7.9 KB | + +### Generic field access experiment + +`FieldAccessBench` reads the same 100,000 `BIGINT` values from +`JUST_DATA..FACTPRODUCTINVENTORY` and checks a checksum. The following +before/after runs evaluated a direct `GetFieldValue` fast path; the +candidate was removed because the improvement was not consistent across JIT +and NativeAOT and allocation savings did not meet the 10% acceptance threshold. + +| Runtime | Version | Method | Mean | Allocated | +|---------|---------|--------|-----:|----------:| +| .NET 10 JIT, before | existing cast | 474.5 ms | 60.28 MB | +| .NET 10 JIT, before | `GetFieldValue` | 507.6 ms | 60.28 MB | +| .NET 10 JIT, candidate | `GetFieldValue` | 490.8 ms | 57.99 MB | +| .NET 10 NativeAOT, before | existing cast | 328.5 ms | 60.28 MB | +| .NET 10 NativeAOT, before | `GetFieldValue` | 347.5 ms | 60.28 MB | +| .NET 10 NativeAOT, candidate | `GetFieldValue` | 358.7 ms | 58.00 MB | + +Each run used three measured iterations and one warmup. The candidate saved +about 24 bytes per row (3.8% of total measured allocations), while its generic +access time improved 3.3% in JIT and regressed 3.2% in NativeAOT. The timing +spread and opposite direction do not establish a throughput gain. The benchmark +can be rerun with: + +```bash +dotnet run -c Release --project src/JustyBase.NetezzaDriver.Benchmarks --framework net10.0 -- \ + --filter '*FieldAccessBench*' --runtimes net10.0 --iterationCount 3 --warmupCount 1 +``` diff --git a/src/JustyBase.NetezzaDriver.Tests/AsyncTests.cs b/src/JustyBase.NetezzaDriver.Tests/AsyncTests.cs index 286f748..325c548 100644 --- a/src/JustyBase.NetezzaDriver.Tests/AsyncTests.cs +++ b/src/JustyBase.NetezzaDriver.Tests/AsyncTests.cs @@ -64,6 +64,50 @@ public async Task ExecuteReaderAsync_ShouldReadLargeResultSet() Assert.True(rows >= 1000); } + [Fact] + public async Task ExecuteRowsAsync_ShouldMapRowsIncrementallyAndCloseReaderOnEarlyExit() + { + var ct = TestContext.Current.CancellationToken; + await using NzConnection connection = new NzConnection(Config.UserName, Config.Password, Config.Host, Config.DbName, Config.Port); + await connection.OpenAsync(ct); + await using var command = connection.CreateCommand("SELECT 1 AS VALUE UNION ALL SELECT 2 UNION ALL SELECT 3"); + + List values = []; + await foreach (int value in command.ExecuteRowsAsync(reader => reader.GetInt32(0), ct)) + { + values.Add(value); + } + + Assert.Equal([1, 2, 3], values); + + await foreach (int _ in command.ExecuteRowsAsync(reader => reader.GetInt32(0), ct)) + { + break; + } + + await using var followUp = connection.CreateCommand("SELECT 42 AS VALUE"); + Assert.Equal(42, await followUp.ExecuteScalarAsync(ct)); + } + + [Fact] + public async Task ExecuteRowsAsync_ShouldDisposeReaderWhenMapperThrows() + { + var ct = TestContext.Current.CancellationToken; + await using NzConnection connection = new NzConnection(Config.UserName, Config.Password, Config.Host, Config.DbName, Config.Port); + await connection.OpenAsync(ct); + await using var command = connection.CreateCommand("SELECT 1 AS VALUE UNION ALL SELECT 2"); + + await Assert.ThrowsAsync(async () => + { + await foreach (int _ in command.ExecuteRowsAsync(_ => throw new InvalidOperationException("mapping failed"), ct)) + { + } + }); + + await using var followUp = connection.CreateCommand("SELECT 42 AS VALUE"); + Assert.Equal(42, await followUp.ExecuteScalarAsync(ct)); + } + [Fact] public async Task Reader_GetBytesAndGetChars_ShouldWorkForTextValues() { diff --git a/src/JustyBase.NetezzaDriver.Tests/BackendDiagnosticResponseParserTests.cs b/src/JustyBase.NetezzaDriver.Tests/BackendDiagnosticResponseParserTests.cs new file mode 100644 index 0000000..ba877c4 --- /dev/null +++ b/src/JustyBase.NetezzaDriver.Tests/BackendDiagnosticResponseParserTests.cs @@ -0,0 +1,49 @@ +using System.Text; + +namespace JustyBase.NetezzaDriver.Tests; + +[Trait("Category", "Unit")] +public sealed class BackendDiagnosticResponseParserTests +{ + [Fact] + public void Parse_ReadsStructuredErrorAndPreservesUnknownDiagnosticFields() + { + const string payload = "SERROR\0VERROR\0C42601\0Msyntax error\0Dunexpected token\0Hcheck the query\0Zserver extension\0\0"; + var response = BackendDiagnosticResponseParser.Parse(Encoding.UTF8.GetBytes(payload)); + + Assert.Equal("syntax error", response.Message); + Assert.Equal("ERROR", response.Severity); + Assert.Equal("42601", response.SqlState); + Assert.Equal("unexpected token", response.Detail); + Assert.Equal("check the query", response.Hint); + Assert.Equal("server extension", response.Diagnostics['Z']); + Assert.Equal(payload, response.RawResponse); + + var exception = new NetezzaException(response); + Assert.Equal("syntax error", exception.Message); + Assert.Equal("42601", exception.SqlState); + Assert.Equal(payload, exception.RawResponse); + } + + [Fact] + public void Parse_PrefersNonLocalizedSeverity() + { + var response = BackendDiagnosticResponseParser.Parse( + Encoding.UTF8.GetBytes("Slocalized\0Vnonlocalized\0Mmessage\0\0")); + + Assert.Equal("nonlocalized", response.Severity); + } + + [Theory] + [InlineData("permission denied\0", "permission denied")] + [InlineData("\0\0", "Netezza backend returned an empty error response")] + public void Parse_UsesPlainTextFallback(string payload, string expectedMessage) + { + var response = BackendDiagnosticResponseParser.Parse(Encoding.UTF8.GetBytes(payload)); + + Assert.Equal(expectedMessage, response.Message); + Assert.Equal(payload, response.RawResponse); + Assert.Empty(response.Diagnostics); + Assert.Null(response.SqlState); + } +} diff --git a/src/JustyBase.NetezzaDriver.Tests/BasicTests.cs b/src/JustyBase.NetezzaDriver.Tests/BasicTests.cs index 0766691..8264c04 100644 --- a/src/JustyBase.NetezzaDriver.Tests/BasicTests.cs +++ b/src/JustyBase.NetezzaDriver.Tests/BasicTests.cs @@ -1,23 +1,13 @@ -using System.Data.Odbc; -using Xunit.Sdk; - -namespace JustyBase.NetezzaDriver.Tests; +namespace JustyBase.NetezzaDriver.Tests; [Collection("Sequential")] [Trait("Category", "Integration")] public class BasicTests : IDisposable { - private readonly ITestOutputHelper _output; - - OdbcConnection _odbcConnection; NzConnection _nzNewConnection; - public BasicTests(ITestOutputHelper output) + public BasicTests() { - _output = output; - _odbcConnection = new OdbcConnection($"Driver={{NetezzaSQL}};servername={Config.Host};port={Config.Port};database={Config.DbName};username={Config.UserName};password={Config.Password}"); - _odbcConnection.Open(); - _nzNewConnection = new NzConnection(Config.UserName, Config.Password, Config.Host, Config.DbName, Config.Port); _nzNewConnection.Open(); } @@ -45,148 +35,6 @@ public void aaaa() //from SYSTEM.._V_TABLE; //select 'SELECT * FROM SYSTEM.ADMIN.' || VIEWNAME //from SYSTEM.._V_VIEW; - private const string queryManyTypes = """ - SELECT - 10::bigint - , null ::bigint - , true::Boolean -- ?? - , false::Boolean -- ?? - , null::Boolean - , 5::Byteint - , null::Byteint - , 'a' :: Char - , null :: Char - , current_date::Date - , null::Date - , 0.5::float - , null::float - , 10::integer - , null::integer - , 'next should be 02:00:00 time' - , '02:00:00'::TIME - , 'abc' ::nchar(10) - , null ::nchar(10) - , 1.54::numeric(30, 6) - , null::numeric(30, 6) - , 'abc'::Nvarchar(10) - , null::Nvarchar(10) - , 1.54::real - , null::real - , 5::smallint - , null::smallint - --, current_time::time - , '10:12:13'::TIME - , null::time - --, null::Timewithzone - , DATE_TRUNC('hour',current_timestamp)::Timestamp - , null:: Timestamp - , 'abc' ::varchar(10) - , null ::varchar(10) - ,* - FROM JUST_DATA..FACTPRODUCTINVENTORY - order by rowid asc - LIMIT 1 - """; - - private const string queryManyTypes2 = """ - SELECT - 10::bigint - , null ::bigint - , true::Boolean -- ?? - , false::Boolean -- ?? - , null::Boolean - , 5::Byteint - , null::Byteint - , 'a' :: Char - , null :: Char - , current_date::Date - , null::Date - , 0.5::float - , null::float - , 10::integer - , null::integer - , 'next should be 02:00:00 time' - , '02:00:00'::TIME - , 'abc' ::nchar(10) - , null ::nchar(10) - , 1.54::numeric(30, 6) - , null::numeric(30, 6) - , 'abc'::Nvarchar(10) - , null::Nvarchar(10) - , 1.54::real - , null::real - , 5::smallint - , null::smallint - --, current_time::time - , '10:12:13'::TIME - , null::time - --, null::Timewithzone - , DATE_TRUNC('hour',current_timestamp)::Timestamp - , null:: Timestamp - , 'abc' ::varchar(10) - , null ::varchar(10) - FROM JUST_DATA.._V_RELATION_COLUMN - LIMIT 10 - """; - - private readonly string[] _queryListBasic = - [ - "SELECT '12:00:00'::TIME, '12:00:00'::TIMETZ,'14:13:12.4321+11:15'::TIMETZ", - "SELECT NOW()", - "SELECT * FROM JUST_DATA.ADMIN.DIMDATE ORDER BY ROWID LIMIT 1000", - "SELECT false::BOOLEAN FROM JUST_DATA.ADMIN.DIMDATE LIMIT 1", - "SELECT 15::BYTEINT FROM JUST_DATA.ADMIN.DIMDATE LIMIT 1", - "SELECT 'ABC'::VARCHAR(10) FROM JUST_DATA.ADMIN.DIMDATE LIMIT 1", - "SELECT '2024-12-12'::DATE FROM JUST_DATA.ADMIN.DIMDATE LIMIT 1", - "SELECT '2024-12-12'::TIMESTAMP FROM JUST_DATA.ADMIN.DIMDATE LIMIT 1", - "SELECT 3.14::NUMERIC(10,4) FROM JUST_DATA.ADMIN.DIMDATE LIMIT 1", - "SELECT 3.14::NUMERIC(38,8) FROM JUST_DATA.ADMIN.DIMDATE LIMIT 1", - "SELECT 123456789::NUMERIC(38,0) FROM JUST_DATA.ADMIN.DIMDATE LIMIT 1", - "SELECT 3.14::REAL FROM JUST_DATA.ADMIN.DIMDATE LIMIT 1", - "SELECT 3.14::DOUBLE FROM JUST_DATA.ADMIN.DIMDATE LIMIT 1", - "SELECT 12345678::INTEGER FROM JUST_DATA.ADMIN.DIMDATE LIMIT 1", - "SELECT -9223372036854775808::BIGINT FROM JUST_DATA.ADMIN.DIMDATE LIMIT 1", - "SELECT 9223372036854775807::BIGINT FROM JUST_DATA.ADMIN.DIMDATE LIMIT 1", - "SELECT 25000::SMALLINT FROM JUST_DATA.ADMIN.DIMDATE LIMIT 1", - "SELECT false::BOOLEAN", - "SELECT 15::BYTEINT", - "SELECT '2024-12-12'::DATE", - "SELECT 3.14::NUMERIC(38,8)", - "SELECT * FROM JUST_DATA.ADMIN.DIMACCOUNT ORDER BY ROWID LIMIT 1000", - "SELECT * FROM JUST_DATA.ADMIN.DIMDATE ORDER BY ROWID LIMIT 1000", - "SELECT * FROM JUST_DATA.ADMIN.DIMPRODUCT ORDER BY ROWID LIMIT 1000", - "SELECT * FROM JUST_DATA.ADMIN.FACTPRODUCTINVENTORY ORDER BY ROWID LIMIT 1000", - "SELECT * FROM JUST_DATA..NUMERIC_TEST ORDER BY ROWID LIMIT 1000" - ]; - - private readonly string[] _queriesFromSystemTables = - [ - "SELECT * FROM SYSTEM.ADMIN._T_OBJECT ORDER BY ROWID LIMIT 1000", - "SELECT * FROM SYSTEM.ADMIN._T_DATABASE ORDER BY ROWID LIMIT 1000", - "SELECT * FROM SYSTEM.ADMIN._T_USER ORDER BY ROWID LIMIT 1000", - "SELECT * FROM SYSTEM.ADMIN._T_GROUP ORDER BY ROWID LIMIT 1000", - "SELECT * FROM SYSTEM.ADMIN._T_SCHEMA ORDER BY ROWID LIMIT 1000", - "SELECT * FROM SYSTEM.ADMIN._VT_DUAL ORDER BY ROWID LIMIT 1000", - "SELECT * FROM SYSTEM.ADMIN._V_DUAL", - "SELECT * FROM SYSTEM.ADMIN._V_DATABASE ORDER BY OBJID LIMIT 1000", - "SELECT * FROM SYSTEM.ADMIN._V_USER LIMIT 1000", - "SELECT * FROM SYSTEM.ADMIN._V_GROUP LIMIT 1000", - "SELECT * FROM SYSTEM.ADMIN._V_SCHEMA LIMIT 1000", - "SELECT * FROM SYSTEM.ADMIN._V_TABLE LIMIT 1000", - "SELECT * FROM SYSTEM.ADMIN._V_VIEW ORDER BY OBJID LIMIT 1000", - "SELECT * FROM SYSTEM.ADMIN._V_INDEX LIMIT 1000", - "SELECT * FROM SYSTEM.ADMIN._V_RELATION_COLUMN LIMIT 1000", - "SELECT * FROM SYSTEM.ADMIN._V_DATATYPE LIMIT 1000", - "SELECT * FROM SYSTEM.ADMIN._V_SEQUENCE LIMIT 1000", - "SELECT * FROM SYSTEM.ADMIN._V_FUNCTION LIMIT 1000", - "SELECT * FROM SYSTEM.ADMIN._V_OBJECT LIMIT 1000", - "SELECT * FROM SYSTEM.ADMIN._V_SYS_DATABASE LIMIT 1000", - "SELECT * FROM SYSTEM.ADMIN._V_SYSTEM_INFO LIMIT 1000", - "SELECT * FROM SYSTEM.ADMIN._V_CONNECTION LIMIT 1000", - "SELECT * FROM SYSTEM.ADMIN._V_ODBC_FEATURE LIMIT 1000", - "SELECT * FROM SYSTEM.ADMIN._V_DOTNET_FEATURE LIMIT 1000" - ]; - [Theory] [InlineData("SELECT '2 years 5 hours 11 months 41 minutes 15 sec'::interval FROM JUST_DATA..DIMDATE LIMIT 1", "2 years 11 mons 05:41:15 String")] [InlineData("SELECT '5 hours 41 minutes 15 sec'::interval FROM JUST_DATA..DIMDATE LIMIT 1", "05:41:15 String")] @@ -413,42 +261,6 @@ public void CheckDateTimeConsistency() } - [Fact(Timeout = 20000)] - public void OdbcAndNzResultsShouldMatch() - { - foreach (var query in _queryListBasic) - { - _output.WriteLine($"Query {query}"); - ValidateTypedQueryResultsByGetValue( query); - } - } - [Fact(Timeout = 20000)] - public void OdbcAndNzResultsShouldMatchSystem() - { - foreach (var query in _queriesFromSystemTables) - { - _output.WriteLine($"Query {query}"); - ValidateTypedQueryResultsByGetValue(query); - } - } - - private readonly string[] _queriesShouldMatchFast = - [ - "SELECT * FROM JUST_DATA..DIMDATE ORDER BY DATEKEY LIMIT 1500", - "SELECT NULL FROM ONE_ROW_TABLE UNION ALL SELECT 'XXXX' FROM ONE_ROW_TABLE", - "SELECT NULL UNION ALL SELECT 'XXXX'" - ]; - - [Fact] - public void OdbcAndNzResultsShouldMatchFastGetValue() - { - foreach (var query in _queriesShouldMatchFast) - { - _output.WriteLine($"Query {query}"); - ValidateTypedQueryResultsByGetValue(query); - } - } - [Fact] public void GetString_OnNullValue_ThrowsException() { @@ -528,119 +340,8 @@ public void GetString_OnMixedNullAndNonNullValues_HandlesCorrectly() Assert.Equal("abc", reader.GetString(1)); Assert.Equal("def", reader.GetString(3)); } - - - - private void ValidateTypedQueryResultsByGetValue(string query) - { - //Stopwatch stopwatch = Stopwatch.StartNew(); - using var cmd1 = _odbcConnection.CreateCommand(); - cmd1.CommandText = query; - using var readerOdbc = cmd1.ExecuteReader(); - - using var cmd2 = _nzNewConnection.CreateCommand(); - cmd2.CommandText = query; - using var readerNz = cmd2.ExecuteReader(); - - bool r1 = readerOdbc.Read(); - bool r2 = readerNz.Read(); - int num = 0; - - while (r1 && r2) - { - num++; - Assert.True(num < 2000, $"Too many rows returned {query}"); - Assert.Equal(readerOdbc.FieldCount, readerNz.FieldCount); - for (int i = 0; i < readerOdbc.FieldCount; i++) - { - Assert.Equal(readerOdbc.IsDBNull(i), readerNz.IsDBNull(i)); - if (readerOdbc.IsDBNull(i)) - { - continue; - } - - Assert.Equal(readerOdbc.GetFieldType(i), readerNz.GetFieldType(i)); - - var odbcObjValue = readerOdbc.GetValue(i); - var nzObjValue = readerNz.GetValue(i); - - if (odbcObjValue is string strOdbc && nzObjValue is string strNz) - { - if (strOdbc.Length > 4000) - { - strOdbc = strOdbc[0..4000]; - } - if (strNz.Length > 4000) - { - strNz = strNz[0..4000]; - } - Assert.Equal(strOdbc, strNz); - } - else if (odbcObjValue is DateTime datetimeOdbc && nzObjValue is DateTime datetimeNz) - { - Assert.Equal(datetimeOdbc, datetimeNz, TimeSpan.FromSeconds(15)); - } - else - { - Assert.Equal(odbcObjValue, nzObjValue); - } - - if (readerNz.GetFieldType(i) == typeof(byte)) - { - var o1 = readerNz.GetByte(i); - Assert.Equal(o1, nzObjValue); - } - else if (readerNz.GetFieldType(i) == typeof(Int16)) - { - var o1 = readerNz.GetInt16(i); - Assert.Equal(o1, nzObjValue); - } - else if (readerNz.GetFieldType(i) == typeof(int)) - { - var o1 = readerNz.GetInt32(i); - Assert.Equal(o1, nzObjValue); - } - else if (readerNz.GetFieldType(i) == typeof(long)) - { - var o1 = readerNz.GetInt64(i); - Assert.Equal(o1, nzObjValue); - } - else if (readerNz.GetFieldType(i) == typeof(DateTime)) - { - var o1 = readerNz.GetDateTime(i); - Assert.Equal(o1, (DateTime)nzObjValue, precision: TimeSpan.FromSeconds(15)); - } - else if (readerNz.GetFieldType(i) == typeof(decimal)) - { - var o1 = readerNz.GetDecimal(i); - Assert.Equal(o1, nzObjValue); - } - else if (readerNz.GetFieldType(i) == typeof(float)) - { - var o1 = readerNz.GetFloat(i); - Assert.Equal(o1, nzObjValue); - } - else if (readerNz.GetFieldType(i) == typeof(double)) - { - var o1 = readerNz.GetDouble(i); - Assert.Equal(o1, nzObjValue); - } - else if (readerNz.GetFieldType(i) == typeof(string)) - { - var o1 = readerNz.GetString(i); - Assert.Equal(o1, nzObjValue); - } - } - - r1 = readerOdbc.Read(); - r2 = readerNz.Read(); - } - Assert.Equal(r1,r2);//same number of rows - } - public void Dispose() { - _odbcConnection.Dispose(); _nzNewConnection.Dispose(); } diff --git a/src/JustyBase.NetezzaDriver.Tests/CommandAbortTest.cs b/src/JustyBase.NetezzaDriver.Tests/CommandAbortTest.cs index bdfbfc7..cbcff8c 100644 --- a/src/JustyBase.NetezzaDriver.Tests/CommandAbortTest.cs +++ b/src/JustyBase.NetezzaDriver.Tests/CommandAbortTest.cs @@ -26,7 +26,8 @@ public async Task AbortTest1() [Fact] public void AbortTestWithSSL() { - using NzConnection connection = new NzConnection(Config.UserName, Config.Password, Config.Host, Config.DbName, securityLevel: SecurityLevelCode.OnlySecuredSession, sslCerFilePath: @"C:\DEV\DEV\Others\keys\server-cert.pem", loggerFactory: new NullLoggerFactory()); + using var untrustedCertificate = TemporaryCertificateFile.Create(); + using NzConnection connection = new NzConnection(Config.UserName, Config.Password, Config.Host, Config.DbName, securityLevel: SecurityLevelCode.OnlySecuredSession, sslCerFilePath: untrustedCertificate.Path, loggerFactory: new NullLoggerFactory()); Assert.Throws(() => connection.Open()); } diff --git a/src/JustyBase.NetezzaDriver.Tests/InvalidSqlTests.cs b/src/JustyBase.NetezzaDriver.Tests/InvalidSqlTests.cs index 108b365..ce742e3 100644 --- a/src/JustyBase.NetezzaDriver.Tests/InvalidSqlTests.cs +++ b/src/JustyBase.NetezzaDriver.Tests/InvalidSqlTests.cs @@ -15,6 +15,35 @@ public void ReaderShouldThrow() Assert.Throws(() => command.ExecuteReader()); } + [Fact] + public void BackendErrorShouldExposeReadableMessageAndRawResponse() + { + using NzConnection connection = new NzConnection(Config.UserName, Config.Password, Config.Host, Config.DbName, Config.Port); + connection.Open(); + using var command = connection.CreateCommand("SELECT 1,,2"); + + var exception = Assert.Throws(() => command.ExecuteReader()); + + Assert.False(string.IsNullOrWhiteSpace(exception.Message)); + Assert.DoesNotContain('\0', exception.Message); + Assert.False(string.IsNullOrEmpty(exception.RawResponse)); + } + + [Fact] + public async Task BackendErrorShouldExposeReadableMessageAndRawResponseAsync() + { + await using NzConnection connection = new NzConnection(Config.UserName, Config.Password, Config.Host, Config.DbName, Config.Port); + await connection.OpenAsync(TestContext.Current.CancellationToken); + await using var command = connection.CreateCommand("SELECT 1,,2"); + + var exception = await Assert.ThrowsAsync( + async () => await command.ExecuteReaderAsync(TestContext.Current.CancellationToken)); + + Assert.False(string.IsNullOrWhiteSpace(exception.Message)); + Assert.DoesNotContain('\0', exception.Message); + Assert.False(string.IsNullOrEmpty(exception.RawResponse)); + } + [Fact] public void ExecuteNonQueryShouldThrow() { diff --git a/src/JustyBase.NetezzaDriver.Tests/JustyBase.NetezzaDriver.Tests.csproj b/src/JustyBase.NetezzaDriver.Tests/JustyBase.NetezzaDriver.Tests.csproj index ab65483..c16b306 100644 --- a/src/JustyBase.NetezzaDriver.Tests/JustyBase.NetezzaDriver.Tests.csproj +++ b/src/JustyBase.NetezzaDriver.Tests/JustyBase.NetezzaDriver.Tests.csproj @@ -15,7 +15,6 @@ runtime; build; native; contentfiles; analyzers; buildtransitive - all runtime; build; native; contentfiles; analyzers; buildtransitive @@ -35,6 +34,9 @@ PreserveNewest + + PreserveNewest + diff --git a/src/JustyBase.NetezzaDriver.Tests/NoticeTests.cs b/src/JustyBase.NetezzaDriver.Tests/NoticeTests.cs index be8175e..90afc6c 100644 --- a/src/JustyBase.NetezzaDriver.Tests/NoticeTests.cs +++ b/src/JustyBase.NetezzaDriver.Tests/NoticeTests.cs @@ -18,15 +18,18 @@ public void BasicNoticeTests() { using NzConnection connection = new NzConnection(Config.UserName, Config.Password, Config.Host, Config.DbName, Config.Port); connection.Open(); - using var command = connection.CreateCommand(); + using var command = connection.CreateCommand("CALL CUSTOMER_DOTNET();"); List notices = new List(); connection.NoticeReceived += (o,e) => { notices.Add(e.Message); }; - command.CommandText = "CALL CUSTOMER_DOTNET();"; command.ExecuteNonQuery(); - var expected = new List() { "The customer name is alpha\n", "The customer location is beta\n" }; + var expected = new List() { "The customer name is alpha", "The customer location is beta" }; Assert.Equal(expected, notices); + Assert.Equal(expected, command.Notices); + + command.ExecuteNonQuery(); + Assert.Equal(expected, command.Notices); } } diff --git a/src/JustyBase.NetezzaDriver.Tests/NzpyExtendedReference.cs b/src/JustyBase.NetezzaDriver.Tests/NzpyExtendedReference.cs new file mode 100644 index 0000000..c55be9e --- /dev/null +++ b/src/JustyBase.NetezzaDriver.Tests/NzpyExtendedReference.cs @@ -0,0 +1,84 @@ +using System.Diagnostics; +using System.Text.Json; + +namespace JustyBase.NetezzaDriver.Tests; + +internal static class NzpyExtendedReference +{ + private sealed record ReferenceResponse(ReferenceResultSet[] ResultSets); + + internal sealed record ReferenceResultSet(string[] Columns, string?[][] Rows); + + internal static async Task ExecuteAsync( + IReadOnlyCollection queries, + CancellationToken cancellationToken = default) + { + string python = Environment.GetEnvironmentVariable("NZPY_EXTENDED_PYTHON") + ?? (OperatingSystem.IsWindows() ? "python" : "python3"); + string scriptPath = Path.Combine(AppContext.BaseDirectory, "nzpy-extended-reference.py"); + + if (!File.Exists(scriptPath)) + { + throw new FileNotFoundException("The nzpy-extended reference script was not copied to the test output directory.", scriptPath); + } + + var startInfo = new ProcessStartInfo + { + FileName = python, + UseShellExecute = false, + RedirectStandardInput = true, + RedirectStandardOutput = true, + RedirectStandardError = true, + CreateNoWindow = true + }; + startInfo.ArgumentList.Add(scriptPath); + + using var process = new Process { StartInfo = startInfo }; + try + { + process.Start(); + } + catch (Exception exception) when (exception is System.ComponentModel.Win32Exception or InvalidOperationException) + { + throw new InvalidOperationException( + $"Could not start Python executable '{python}'. Install nzpy-extended and set NZPY_EXTENDED_PYTHON if needed.", + exception); + } + + Task standardOutputTask = process.StandardOutput.ReadToEndAsync(); + Task standardErrorTask = process.StandardError.ReadToEndAsync(); + try + { + await process.StandardInput.WriteAsync(JsonSerializer.Serialize(new { queries }).AsMemory(), cancellationToken); + process.StandardInput.Close(); + await process.WaitForExitAsync(cancellationToken); + } + catch + { + if (!process.HasExited) + { + process.Kill(entireProcessTree: true); + await process.WaitForExitAsync(CancellationToken.None); + } + + throw; + } + + string standardOutput = await standardOutputTask; + string standardError = await standardErrorTask; + + if (process.ExitCode != 0) + { + throw new InvalidOperationException( + $"nzpy-extended reference execution failed with exit code {process.ExitCode}: {standardError}"); + } + + var response = JsonSerializer.Deserialize(standardOutput, new JsonSerializerOptions + { + PropertyNameCaseInsensitive = true + }); + + return response?.ResultSets + ?? throw new InvalidOperationException("nzpy-extended returned an empty or invalid result document."); + } +} diff --git a/src/JustyBase.NetezzaDriver.Tests/NzpyExtendedResultCompatibilityTests.cs b/src/JustyBase.NetezzaDriver.Tests/NzpyExtendedResultCompatibilityTests.cs new file mode 100644 index 0000000..743868e --- /dev/null +++ b/src/JustyBase.NetezzaDriver.Tests/NzpyExtendedResultCompatibilityTests.cs @@ -0,0 +1,88 @@ +using System.Globalization; + +namespace JustyBase.NetezzaDriver.Tests; + +[Collection("Sequential")] +[Trait("Category", "Integration")] +public sealed class NzpyExtendedResultCompatibilityTests +{ + private const string RepresentativeQuery = """ + SELECT + 10::BIGINT AS I64_VALUE, + 10::INTEGER AS I32_VALUE, + 10::SMALLINT AS I16_VALUE, + 5::BYTEINT AS I8_VALUE, + true::BOOLEAN AS BOOL_TRUE, + false::BOOLEAN AS BOOL_FALSE, + null::BIGINT AS NULL_VALUE, + 'abc'::VARCHAR(10) AS TEXT_VALUE, + 1.54::NUMERIC(10,2) AS DECIMAL_VALUE, + 1.5::REAL AS FLOAT_VALUE, + 3.5::DOUBLE AS DOUBLE_VALUE, + '2026-09-23'::DATE AS DATE_VALUE, + '2026-09-23 10:12:13'::TIMESTAMP AS TIMESTAMP_VALUE, + '10:12:13'::TIME AS TIME_VALUE + """; + + [Fact(Timeout = 60000)] + public async Task Representative_scalar_results_match_nzpy_extended() + { + CancellationToken cancellationToken = TestContext.Current.CancellationToken; + NzpyExtendedReference.ReferenceResultSet[] referenceResults = + await NzpyExtendedReference.ExecuteAsync([RepresentativeQuery], cancellationToken); + var reference = Assert.Single(referenceResults); + + await using var connection = new NzConnection( + Config.UserName, + Config.Password, + Config.Host, + Config.DbName, + Config.Port); + await connection.OpenAsync(cancellationToken); + + await using var command = connection.CreateCommand(); + command.CommandText = RepresentativeQuery; + await using var reader = await command.ExecuteReaderAsync(cancellationToken); + + Assert.Equal(reference.Columns, Enumerable.Range(0, reader.FieldCount).Select(reader.GetName)); + + int rowIndex = 0; + while (await reader.ReadAsync(cancellationToken)) + { + Assert.True(rowIndex < reference.Rows.Length, "The .NET driver returned extra rows."); + string?[] expectedRow = reference.Rows[rowIndex]; + Assert.Equal(reader.FieldCount, expectedRow.Length); + + for (int ordinal = 0; ordinal < reader.FieldCount; ordinal++) + { + Assert.Equal( + expectedRow[ordinal], + NormalizeValue(reader.GetValue(ordinal), reader.GetName(ordinal))); + } + + rowIndex++; + } + + Assert.Equal(reference.Rows.Length, rowIndex); + } + + private static string? NormalizeValue(object value, string columnName) + { + if (value is DBNull) + { + return null; + } + + return value switch + { + bool boolean => boolean ? "true" : "false", + DateTime dateTime when columnName == "DATE_VALUE" => dateTime.ToString("yyyy-MM-dd", CultureInfo.InvariantCulture), + DateTime dateTime => TrimFraction(dateTime.ToString("yyyy-MM-dd HH:mm:ss.fffffff", CultureInfo.InvariantCulture)), + TimeSpan time => TrimFraction(time.ToString(@"hh\:mm\:ss\.fffffff", CultureInfo.InvariantCulture)), + IFormattable formatted => formatted.ToString(null, CultureInfo.InvariantCulture), + _ => value.ToString() + }; + } + + private static string TrimFraction(string value) => value.TrimEnd('0').TrimEnd('.'); +} diff --git a/src/JustyBase.NetezzaDriver.Tests/README.md b/src/JustyBase.NetezzaDriver.Tests/README.md index 82fc941..3036450 100644 --- a/src/JustyBase.NetezzaDriver.Tests/README.md +++ b/src/JustyBase.NetezzaDriver.Tests/README.md @@ -62,7 +62,16 @@ dotnet test .\src\JustyBase.NetezzaDriver.Tests\JustyBase.NetezzaDriver.Tests.cs --filter "Category=Stress" ``` -The .NET tests use `NZ_DEV_DB` for the database name. The Node and Python -drivers are not test dependencies and are not used as a protocol oracle; -their existing scenarios only informed the read-only workload and boundary -matrix. +The .NET tests use `NZ_DEV_DB` for the database name and do not require ODBC. +One integration test compares representative scalar result values and column +names against `nzpy-extended` as a lightweight reference. It needs Python 3.12 +or newer and `nzpy-extended` installed in the selected interpreter: + +```bash +python3 -m pip install nzpy-extended==0.0.1 +NZPY_EXTENDED_PYTHON=python3 dotnet test src/JustyBase.NetezzaDriver.Tests/JustyBase.NetezzaDriver.Tests.csproj --framework net10.0 --filter "FullyQualifiedName~NzpyExtendedResultCompatibilityTests" +``` + +The test opens read-only connections and compares fixed literals across +integer, boolean, null, text, decimal, floating-point, date, timestamp, and +time values. It does not need an ODBC installation. diff --git a/src/JustyBase.NetezzaDriver.Tests/SslTest.cs b/src/JustyBase.NetezzaDriver.Tests/SslTest.cs index af8f1ea..2625369 100644 --- a/src/JustyBase.NetezzaDriver.Tests/SslTest.cs +++ b/src/JustyBase.NetezzaDriver.Tests/SslTest.cs @@ -7,17 +7,12 @@ namespace JustyBase.NetezzaDriver.Tests; [Trait("Category", "Integration")] public class SslTest { - private readonly ITestOutputHelper _output; - public SslTest(ITestOutputHelper output) - { - _output = output; - } [Fact] public void BasicTests() { + using var untrustedCertificate = TemporaryCertificateFile.Create(); using NzConnection connection = new NzConnection(Config.UserName, Config.Password, Config.Host, Config.DbName, - securityLevel: SecurityLevelCode.OnlySecuredSession, sslCerFilePath: @"C:\DEV\DEV\Others\keys\server-cert.pem"); - //this cert file is invalid + securityLevel: SecurityLevelCode.OnlySecuredSession, sslCerFilePath: untrustedCertificate.Path); Assert.Throws(() => { connection.Open(); @@ -28,8 +23,9 @@ public void BasicTests() [Fact] public void BasicTests2() { + using var untrustedCertificate = TemporaryCertificateFile.Create(); using NzConnection connection = new NzConnection(Config.UserName, Config.Password, Config.Host, Config.DbName, - securityLevel: SecurityLevelCode.OnlySecuredSession, sslCerFilePath: @"C:\DEV\DEV\Others\keys\server-cert.pem", loggerFactory: new NullLoggerFactory()); + securityLevel: SecurityLevelCode.OnlySecuredSession, sslCerFilePath: untrustedCertificate.Path, loggerFactory: new NullLoggerFactory()); // Logger presence must not bypass TLS certificate validation. Assert.Throws(() => { diff --git a/src/JustyBase.NetezzaDriver.Tests/TemporaryCertificateFile.cs b/src/JustyBase.NetezzaDriver.Tests/TemporaryCertificateFile.cs new file mode 100644 index 0000000..c6826a0 --- /dev/null +++ b/src/JustyBase.NetezzaDriver.Tests/TemporaryCertificateFile.cs @@ -0,0 +1,41 @@ +using System.Security.Cryptography; +using System.Security.Cryptography.X509Certificates; + +namespace JustyBase.NetezzaDriver.Tests; + +internal sealed class TemporaryCertificateFile : IDisposable +{ + private TemporaryCertificateFile(string path) => Path = path; + + internal string Path { get; } + + internal static TemporaryCertificateFile Create() + { + using RSA key = RSA.Create(2048); + var request = new CertificateRequest( + "CN=JustyBase Netezza driver test certificate", + key, + HashAlgorithmName.SHA256, + RSASignaturePadding.Pkcs1); + using X509Certificate2 certificate = request.CreateSelfSigned( + DateTimeOffset.UtcNow.AddMinutes(-1), + DateTimeOffset.UtcNow.AddDays(1)); + + string path = System.IO.Path.Combine( + System.IO.Path.GetTempPath(), + $"justybase-test-{Guid.NewGuid():N}.pem"); + File.WriteAllText(path, certificate.ExportCertificatePem()); + return new TemporaryCertificateFile(path); + } + + public void Dispose() + { + try + { + File.Delete(Path); + } + catch (FileNotFoundException) + { + } + } +} diff --git a/src/JustyBase.NetezzaDriver.Tests/nzpy-extended-reference.py b/src/JustyBase.NetezzaDriver.Tests/nzpy-extended-reference.py new file mode 100644 index 0000000..b863102 --- /dev/null +++ b/src/JustyBase.NetezzaDriver.Tests/nzpy-extended-reference.py @@ -0,0 +1,58 @@ +import asyncio +import datetime +import json +import os +import sys +from decimal import Decimal + +import nzpy_extended + + +def normalize(value): + if value is None: + return None + if isinstance(value, bool): + return "true" if value else "false" + if isinstance(value, datetime.datetime): + return value.isoformat(sep=" ") + if isinstance(value, datetime.date): + return value.isoformat() + if isinstance(value, datetime.time): + return value.isoformat() + if isinstance(value, Decimal): + return str(value) + if isinstance(value, (int, float)): + return str(value) + if isinstance(value, str): + return value + raise TypeError(f"Unsupported nzpy-extended result type: {type(value).__name__}") + + +async def main(): + request = json.load(sys.stdin) + connection = await nzpy_extended.connect( + user=os.getenv("NZ_DEV_USER", "admin"), + password=os.getenv("NZ_DEV_PASSWORD", "password"), + host=os.getenv("NZ_DEV_HOST", "192.168.0.144"), + port=int(os.getenv("NZ_DEV_PORT", "5480")), + database=os.getenv("NZ_DEV_DB", "JUST_DATA"), + ) + + result_sets = [] + try: + for query in request["queries"]: + cursor = connection.cursor() + await cursor.execute(query) + columns = [column[0] for column in cursor.description] + rows = await cursor.fetchall() + result_sets.append({ + "columns": columns, + "rows": [[normalize(value) for value in row] for row in rows], + }) + finally: + await connection.close() + + json.dump({"resultSets": result_sets}, sys.stdout, ensure_ascii=False) + + +asyncio.run(main()) diff --git a/src/JustyBase.NetezzaDriver/BackendDiagnosticResponse.cs b/src/JustyBase.NetezzaDriver/BackendDiagnosticResponse.cs new file mode 100644 index 0000000..450518f --- /dev/null +++ b/src/JustyBase.NetezzaDriver/BackendDiagnosticResponse.cs @@ -0,0 +1,86 @@ +using System.Collections.ObjectModel; +using System.Text; + +namespace JustyBase.NetezzaDriver; + +internal sealed record BackendDiagnosticResponse( + string Message, + string RawResponse, + string? Severity, + string? SqlState, + string? Detail, + string? Hint, + IReadOnlyDictionary Diagnostics); + +/// +/// Parses PostgreSQL-style ErrorResponse and NoticeResponse payloads while +/// retaining a plain-text fallback for older Netezza servers. +/// +internal static class BackendDiagnosticResponseParser +{ + private const string EmptyMessage = "Netezza backend returned an empty error response"; + + public static BackendDiagnosticResponse Parse(ReadOnlySpan payload) + { + string rawResponse = Encoding.UTF8.GetString(payload); + string fallbackMessage = rawResponse.Replace("\0", string.Empty).Trim(); + if (fallbackMessage.Length == 0) + { + fallbackMessage = EmptyMessage; + } + + int nullCount = 0; + foreach (byte value in payload) + { + if (value == 0) + { + nullCount++; + } + } + + var fields = new Dictionary(); + if (nullCount >= 2) + { + int offset = 0; + while (offset < payload.Length) + { + byte fieldCode = payload[offset++]; + if (fieldCode == 0) + { + break; + } + + int valueStart = offset; + while (offset < payload.Length && payload[offset] != 0) + { + offset++; + } + + string value = Encoding.UTF8.GetString(payload[valueStart..offset]); + fields[(char)fieldCode] = value; + + if (offset < payload.Length) + { + offset++; + } + } + } + + fields.TryGetValue('M', out string? message); + fields.TryGetValue('V', out string? nonLocalizedSeverity); + fields.TryGetValue('S', out string? localizedSeverity); + fields.TryGetValue('C', out string? sqlState); + fields.TryGetValue('D', out string? detail); + fields.TryGetValue('H', out string? hint); + + var diagnostics = new ReadOnlyDictionary(fields); + return new BackendDiagnosticResponse( + string.IsNullOrEmpty(message) ? fallbackMessage : message, + rawResponse, + nonLocalizedSeverity ?? localizedSeverity, + sqlState, + detail, + hint, + diagnostics); + } +} diff --git a/src/JustyBase.NetezzaDriver/JustyBase.NetezzaDriver.csproj b/src/JustyBase.NetezzaDriver/JustyBase.NetezzaDriver.csproj index eab3018..3849d37 100644 --- a/src/JustyBase.NetezzaDriver/JustyBase.NetezzaDriver.csproj +++ b/src/JustyBase.NetezzaDriver/JustyBase.NetezzaDriver.csproj @@ -6,7 +6,7 @@ true True Pure C# driver for IBM Netezza database - 1.7.2 + 1.8.0 JustyBase.NetezzaDriver is a .NET library for interacting with IBM Netezza Performance Server databases. It provides a set of classes and methods to facilitate database connections, command execution, and data retrieval. Code is is based on nzpy and npgsql https://github.com/justybase/JustyBase.NetezzaDriver README.md diff --git a/src/JustyBase.NetezzaDriver/NzCommand.cs b/src/JustyBase.NetezzaDriver/NzCommand.cs index e9758b7..7fcd309 100644 --- a/src/JustyBase.NetezzaDriver/NzCommand.cs +++ b/src/JustyBase.NetezzaDriver/NzCommand.cs @@ -1,7 +1,9 @@ using JustyBase.NetezzaDriver.StringPool; +using System.Collections.ObjectModel; using System.Data; using System.Data.Common; using System.Diagnostics.CodeAnalysis; +using System.Runtime.CompilerServices; namespace JustyBase.NetezzaDriver; @@ -15,6 +17,16 @@ public sealed class NzCommand : DbCommand } private RowValue[] _row = null!; + private readonly List _notices = []; + private readonly ReadOnlyCollection _noticesView; + + /// + /// Notices returned during the most recent execution of this command. + /// + public IReadOnlyList Notices => _noticesView; + + internal void AddNotice(string notice) => _notices.Add(notice); + public void AddRow(RowValue[] row) { _row = row; @@ -26,11 +38,13 @@ public ref RowValue GetValue(int ordinal) public NzCommand(NzConnection connection) { + _noticesView = _notices.AsReadOnly(); _connection = connection; connection.SetNzCommand(this); } public NzCommand(string sql, NzConnection connection) { + _noticesView = _notices.AsReadOnly(); _connection = connection; CommandText = sql; } @@ -202,6 +216,7 @@ private void Clear() _prevReader = null!; NewPreparedStatement = null; _recordsAffected = -1; + _notices.Clear(); } public override void Cancel() @@ -306,9 +321,29 @@ public override async Task ExecuteNonQueryAsync(CancellationToken cancellat return null; } + /// + /// Executes the command and maps each row as it is read, without buffering + /// the full result set in memory. + /// + public async IAsyncEnumerable ExecuteRowsAsync( + Func map, + [EnumeratorCancellation] CancellationToken cancellationToken = default) + { + ArgumentNullException.ThrowIfNull(map); + + await using var reader = await ExecuteDbDataReaderAsync( + CommandBehavior.Default, + cancellationToken).ConfigureAwait(false); + + while (await reader.ReadAsync(cancellationToken).ConfigureAwait(false)) + { + cancellationToken.ThrowIfCancellationRequested(); + yield return map(reader); + } + } + public override void Prepare() { // no-op: server-side prepared statements are not exposed via ADO.NET parameters } } - diff --git a/src/JustyBase.NetezzaDriver/NzConnection.cs b/src/JustyBase.NetezzaDriver/NzConnection.cs index 3ff336f..bae8653 100644 --- a/src/JustyBase.NetezzaDriver/NzConnection.cs +++ b/src/JustyBase.NetezzaDriver/NzConnection.cs @@ -16,6 +16,7 @@ namespace JustyBase.NetezzaDriver; public sealed class NzConnection : DbConnection { private string? _error; + private NetezzaException? _backendException; private int _commandNumber = -1; @@ -971,6 +972,7 @@ private void PreExecution(NzCommand nzCommand, string query) { ThrowIfProtocolFaulted(); _error = null; + _backendException = null; nzCommand._recordsAffected = -1; nzCommand.NewPreparedStatement = new PreparedStatement(); nzCommand.NewPreparedStatement.Sql = query; @@ -1020,6 +1022,7 @@ private async Task PreExecutionAsync(NzCommand nzCommand, string query, Cancella { ThrowIfProtocolFaulted(); _error = null; + _backendException = null; nzCommand._recordsAffected = -1; nzCommand.NewPreparedStatement = new PreparedStatement(); nzCommand.NewPreparedStatement.Sql = query; @@ -1128,16 +1131,22 @@ private async ValueTask SkipBytesAsync(int count, CancellationToken cancellation public delegate void NzNoticeEventHandler(object sender, NzNoticeEventArgs e); public event NzNoticeEventHandler? NoticeReceived; - private void OnNoticeReceived(string notice) + private void OnNoticeReceived(string notice, NzCommand nzCommand) { if (notice.StartsWith("NOTICE:")) { notice = notice["NOTICE:".Length..]; } notice = notice.Trim().TrimEnd('\x00'); + nzCommand.AddNotice(notice); NoticeReceived?.Invoke(this, new NzNoticeEventArgs(notice)); } + private NetezzaException CreateCurrentException() + { + return _backendException ?? new NetezzaException(_error ?? "Netezza backend returned an unspecified error."); + } + private TimeSpan _defaultCommandTimeout = TimeSpan.FromSeconds(60); public TimeSpan DefaultCommandTimeout { @@ -1262,7 +1271,7 @@ public bool Execute(NzCommand nzCommand, string query) if (_error != null) { - throw new NetezzaException(_error); + throw CreateCurrentException(); } return response; @@ -1284,7 +1293,7 @@ public async Task ExecuteAsync(NzCommand nzCommand, string query, Cancella if (_error != null) { - throw new NetezzaException(_error); + throw CreateCurrentException(); } return response; @@ -1293,6 +1302,7 @@ public async Task ExecuteAsync(NzCommand nzCommand, string query, Cancella { CancelQuery(); _error = "Command timeout"; + _backendException = null; throw new NetezzaException(_error); } } @@ -1305,7 +1315,7 @@ public NzDataReader ExecuteReader(NzCommand nzCommand, string query) var rdr = new NzDataReader(nzCommand); if (_error != null) { - throw new NetezzaException(_error); + throw CreateCurrentException(); } return rdr; } @@ -1323,7 +1333,7 @@ public async Task ExecuteReaderAsync(NzCommand nzCommand, string q var rdr = await NzDataReader.CreateAsync(nzCommand, effectiveCancellationToken).ConfigureAwait(false); if (_error != null) { - throw new NetezzaException(_error); + throw CreateCurrentException(); } return rdr; @@ -1332,6 +1342,7 @@ public async Task ExecuteReaderAsync(NzCommand nzCommand, string q { CancelQuery(); _error = "Command timeout"; + _backendException = null; throw new NetezzaException(_error); } } @@ -1373,7 +1384,7 @@ internal bool DoNextStep(NzCommand nzCommand) ReadNextResponseByte(); res = IntepretReturnedByte(nzCommand); } - throw new NetezzaException(_error); + throw CreateCurrentException(); } return res; } @@ -1394,7 +1405,7 @@ internal async ValueTask DoNextStepAsync(NzCommand nzCommand, Cancellation await ReadNextResponseByteAsync(cancellationToken).ConfigureAwait(false); res = await IntepretReturnedByteAsync(nzCommand, cancellationToken).ConfigureAwait(false); } - throw new NetezzaException(_error); + throw CreateCurrentException(); } return res; } @@ -1765,7 +1776,8 @@ private bool IntepretReturnedByte(NzCommand nzCommand) int length = ReadProtocolLength("errorPayloadLength"); RegenerateBuffer(length); var data = Read(length, _tmp_buffer); - _error = Encoding.UTF8.GetString(data,0,length); + _backendException = new NetezzaException(BackendDiagnosticResponseParser.Parse(data.AsSpan(0, length))); + _error = _backendException.Message; _logger?.LogDebug("Response received from backend: {_error}", _error); //doContinue = true; } @@ -1842,8 +1854,8 @@ private bool IntepretReturnedByte(NzCommand nzCommand) int length = ReadProtocolLength("noticePayloadLength"); RegenerateBuffer(length); var data = Read(length, _tmp_buffer); - string notice = Encoding.UTF8.GetString(data[0..length]); - OnNoticeReceived(notice); + string notice = BackendDiagnosticResponseParser.Parse(data.AsSpan(0, length)).Message; + OnNoticeReceived(notice, nzCommand); _logger?.LogDebug("Response received from backend: {Notice}", notice); } else if (_lastResponse == (byte)'I') @@ -1852,7 +1864,7 @@ private bool IntepretReturnedByte(NzCommand nzCommand) RegenerateBuffer(length); var data = Read(length, _tmp_buffer); string notice = Encoding.UTF8.GetString(data[0..length]); - OnNoticeReceived(notice); + OnNoticeReceived(notice, nzCommand); _logger?.LogDebug("Response received from backend: {Notice}", notice); nzCommand.AddRow([]); } @@ -1911,7 +1923,8 @@ await ReadProtocolInt32Async("frameHeaderValue", cancellationToken) cancellationToken: cancellationToken).ConfigureAwait(false); RegenerateBuffer(length); var data = await ReadAsync(length, _tmp_buffer, cancellationToken).ConfigureAwait(false); - _error = Encoding.UTF8.GetString(data, 0, length); + _backendException = new NetezzaException(BackendDiagnosticResponseParser.Parse(data.AsSpan(0, length))); + _error = _backendException.Message; _logger?.LogDebug("Response received from backend: {_error}", _error); } else if (_lastResponse == (byte)BackendMessageCode.RowDescription) @@ -1993,8 +2006,8 @@ await ReadProtocolInt32Async("frameHeaderValue", cancellationToken) cancellationToken: cancellationToken).ConfigureAwait(false); RegenerateBuffer(length); var data = await ReadAsync(length, _tmp_buffer, cancellationToken).ConfigureAwait(false); - string notice = Encoding.UTF8.GetString(data, 0, length); - OnNoticeReceived(notice); + string notice = BackendDiagnosticResponseParser.Parse(data.AsSpan(0, length)).Message; + OnNoticeReceived(notice, nzCommand); _logger?.LogDebug("Response received from backend: {Notice}", notice); } else if (_lastResponse == (byte)'I') @@ -2005,7 +2018,7 @@ await ReadProtocolInt32Async("frameHeaderValue", cancellationToken) RegenerateBuffer(length); var data = await ReadAsync(length, _tmp_buffer, cancellationToken).ConfigureAwait(false); string notice = Encoding.UTF8.GetString(data, 0, length); - OnNoticeReceived(notice); + OnNoticeReceived(notice, nzCommand); _logger?.LogDebug("Response received from backend: {Notice}", notice); nzCommand.AddRow([]); } @@ -3538,10 +3551,52 @@ public async Task OpenAsync(ClientTypeId clientVersionId = ClientTypeId.SqlDotne public sealed class NetezzaException : DbException { - public NetezzaException() : base() { } - public NetezzaException(string msg) : base(msg) { } - public NetezzaException(string msg, Exception exception) : base(msg, exception) { } - public NetezzaException(Exception exception) : base("", exception) { } + private static readonly IReadOnlyDictionary EmptyDiagnostics = + new System.Collections.ObjectModel.ReadOnlyDictionary(new Dictionary()); + + public NetezzaException() : this(string.Empty) { } + + public NetezzaException(string msg) : base(msg) + { + RawResponse = msg; + Diagnostics = EmptyDiagnostics; + } + + public NetezzaException(string msg, Exception exception) : base(msg, exception) + { + RawResponse = msg; + Diagnostics = EmptyDiagnostics; + } + + public NetezzaException(Exception exception) : this(string.Empty, exception) { } + + internal NetezzaException(BackendDiagnosticResponse response) : base(response.Message) + { + RawResponse = response.RawResponse; + Severity = response.Severity; + SqlState = response.SqlState; + Detail = response.Detail; + Hint = response.Hint; + Diagnostics = response.Diagnostics; + } + + /// The severity reported by Netezza, when included in the response. + public string? Severity { get; } + + /// The five-character SQLSTATE reported by Netezza, when included. + public override string? SqlState { get; } + + /// Additional server detail, when included in the response. + public string? Detail { get; } + + /// A server-provided hint, when included in the response. + public string? Hint { get; } + + /// The complete decoded backend response, including its diagnostic fields. + public string? RawResponse { get; } + + /// All diagnostic fields supplied by the backend, keyed by protocol field code. + public IReadOnlyDictionary Diagnostics { get; } } public sealed class InterfaceException : DbException { From 54aeff66e5f9d18a187da1c086eeb760afd60162 Mon Sep 17 00:00:00 2001 From: Krzysztof Dusko Date: Wed, 23 Sep 2026 20:15:35 +0200 Subject: [PATCH 2/2] Update examples for streaming and diagnostics --- README.md | 2 ++ .../Examples/AsyncOperations.cs | 12 ++++++++++++ .../Examples/ErrorHandling.cs | 14 ++++++++++++++ .../JustyBase.NetezzaDriver.Examples/Program.cs | 4 ++-- 4 files changed, 30 insertions(+), 2 deletions(-) diff --git a/README.md b/README.md index 1bab5c4..d139f71 100644 --- a/README.md +++ b/README.md @@ -264,6 +264,8 @@ The mapper is synchronous and the returned `IAsyncEnumerable` streams rows; c For server failures, `NetezzaException.Message` contains the primary backend message. `SqlState`, `Severity`, `Detail`, and `Hint` expose structured fields when supplied. `RawResponse` keeps the complete decoded backend payload, and `Diagnostics` provides all fields by protocol code, including fields unknown to this driver. +The runnable [examples project](src/examples/JustyBase.NetezzaDriver.Examples) includes mapped streaming in its async example and server diagnostic fields in its error-handling example. + ## Testing ```bash diff --git a/src/examples/JustyBase.NetezzaDriver.Examples/Examples/AsyncOperations.cs b/src/examples/JustyBase.NetezzaDriver.Examples/Examples/AsyncOperations.cs index dcae6da..52071ed 100644 --- a/src/examples/JustyBase.NetezzaDriver.Examples/Examples/AsyncOperations.cs +++ b/src/examples/JustyBase.NetezzaDriver.Examples/Examples/AsyncOperations.cs @@ -17,6 +17,18 @@ public static async Task RunAsync() Console.WriteLine($" Result: {reader.GetInt32(0)}"); } + // ── Stream mapped rows without buffering the complete result ── + Console.WriteLine("\nMapped row streaming:"); + await using var streamConnection = await ConnectionHelper.OpenAsync(); + await using var streamCommand = streamConnection.CreateCommand( + "SELECT 1 AS value UNION ALL SELECT 2 UNION ALL SELECT 3"); + await foreach (int value in streamCommand.ExecuteRowsAsync( + row => row.GetInt32(0), + cts.Token)) + { + Console.WriteLine($" Streamed: {value}"); + } + // ── Concurrent async queries ── Console.WriteLine("\nConcurrent async queries:"); var queries = new[] { diff --git a/src/examples/JustyBase.NetezzaDriver.Examples/Examples/ErrorHandling.cs b/src/examples/JustyBase.NetezzaDriver.Examples/Examples/ErrorHandling.cs index 64e08d7..b4e3f96 100644 --- a/src/examples/JustyBase.NetezzaDriver.Examples/Examples/ErrorHandling.cs +++ b/src/examples/JustyBase.NetezzaDriver.Examples/Examples/ErrorHandling.cs @@ -18,6 +18,7 @@ public static async Task RunAsync() catch (NetezzaException ex) { Console.WriteLine($" NetezzaException: {ex.Message}"); + PrintDiagnostics(ex); } // ── Missing table ── @@ -30,6 +31,7 @@ public static async Task RunAsync() catch (NetezzaException ex) { Console.WriteLine($" NetezzaException: {ex.Message}"); + PrintDiagnostics(ex); } // ── DBNull handling ── @@ -70,4 +72,16 @@ public static async Task RunAsync() Console.WriteLine("ErrorHandling completed."); } + + private static void PrintDiagnostics(NetezzaException exception) + { + if (exception.SqlState is not null) + Console.WriteLine($" SQLSTATE: {exception.SqlState}"); + if (exception.Severity is not null) + Console.WriteLine($" Severity: {exception.Severity}"); + if (exception.Detail is not null) + Console.WriteLine($" Detail: {exception.Detail}"); + if (exception.Hint is not null) + Console.WriteLine($" Hint: {exception.Hint}"); + } } diff --git a/src/examples/JustyBase.NetezzaDriver.Examples/Program.cs b/src/examples/JustyBase.NetezzaDriver.Examples/Program.cs index 0abee42..9cd2d9d 100644 --- a/src/examples/JustyBase.NetezzaDriver.Examples/Program.cs +++ b/src/examples/JustyBase.NetezzaDriver.Examples/Program.cs @@ -11,9 +11,9 @@ ["3"] = ("Transactions (commit / rollback)", Transactions.RunAsync), ["4"] = ("Connection Pooling", ConnectionPooling.RunAsync), ["5"] = ("Metadata / Catalog Introspection", MetadataIntrospection.RunAsync), - ["6"] = ("Async Operations + Cancellation", AsyncOperations.RunAsync), + ["6"] = ("Async Operations, Streaming + Cancellation", AsyncOperations.RunAsync), ["7"] = ("Timeout and Query Cancel", TimeoutAndCancel.RunAsync), - ["8"] = ("Error Handling & Exception Types", ErrorHandling.RunAsync), + ["8"] = ("Error Handling & Server Diagnostics", ErrorHandling.RunAsync), }; Console.WriteLine("JustyBase.NetezzaDriver — Examples");