Merge re/r1.4-gethi-finding: R1.1 ExecuteSqlCommand + R1.4 GetHistorianInfo (bounded)
# Conflicts: # docs/plans/hcal-roadmap.md # src/AVEVA.Historian.Client/HistorianClient.cs # src/AVEVA.Historian.Client/Protocol/Historian2020ProtocolDialect.cs # tests/AVEVA.Historian.Client.Tests/HistorianClientIntegrationTests.cs # tools/AVEVA.Historian.NativeTraceHarness/Program.cs
This commit is contained in:
Joseph Doherty
@@ -0,0 +1,221 @@
|
||||
using System.Formats.Nrbf;
|
||||
using System.Globalization;
|
||||
using System.Xml.Linq;
|
||||
using AVEVA.Historian.Client.Models;
|
||||
using AVEVA.Historian.Client.Protocol;
|
||||
|
||||
namespace AVEVA.Historian.Client.Wcf;
|
||||
|
||||
/// <summary>
|
||||
/// Parses the <c>GetR</c> (GetRecordSetByteStream) result buffer returned by the 2020 WCF
|
||||
/// <c>ExeC</c>/<c>GetR</c> SQL surface into a <see cref="HistorianSqlResult"/>.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// Captured from the native client (<c>scripts/Capture-ExecSql.ps1</c> +
|
||||
/// instrument-wcf-{write,read}message; golden-pinned in <c>WcfSqlResultProtocolTests</c>). The
|
||||
/// <c>GetR</c> <c>pResultBuff</c> is a <b>.NET Remoting Binary Format (NRBF)</b> stream wrapping a
|
||||
/// <c>System.Data.DataTable</c> serialized with <c>SerializationFormat.Xml</c>: its data lives in
|
||||
/// two string members — <c>XmlSchema</c> (an XSD describing the columns + types) and
|
||||
/// <c>XmlDiffGram</c> (an ADO.NET diffgram carrying the rows).
|
||||
///
|
||||
/// <para>BinaryFormatter is removed from .NET 10, so the stream is decoded read-only with
|
||||
/// <see cref="NrbfDecoder"/> (the sanctioned successor — it parses records without instantiating
|
||||
/// or executing any payload type). The two embedded XML strings are then parsed with
|
||||
/// <see cref="XDocument"/>.</para>
|
||||
///
|
||||
/// <para>Only the <c>SerializationFormat.Xml</c> DataTable shape is evidence-backed; a stream whose
|
||||
/// root is not a <c>DataTable</c> class record, or that lacks the two XML members, throws
|
||||
/// <see cref="ProtocolEvidenceMissingException"/>.</para>
|
||||
/// </remarks>
|
||||
internal static class HistorianSqlResultProtocol
|
||||
{
|
||||
private const string DataTableTypeName = "System.Data.DataTable";
|
||||
private const string XmlSchemaMember = "XmlSchema";
|
||||
private const string XmlDiffGramMember = "XmlDiffGram";
|
||||
|
||||
private static readonly XNamespace Xsd = "http://www.w3.org/2001/XMLSchema";
|
||||
private static readonly XNamespace MsData = "urn:schemas-microsoft-com:xml-msdata";
|
||||
private static readonly XNamespace DiffGr = "urn:schemas-microsoft-com:xml-diffgram-v1";
|
||||
|
||||
public static HistorianSqlResult Parse(ReadOnlyMemory<byte> resultBuffer, int returnValue)
|
||||
{
|
||||
(string? schemaXml, string? diffGramXml) = ExtractXmlMembers(resultBuffer);
|
||||
|
||||
if (string.IsNullOrEmpty(schemaXml))
|
||||
{
|
||||
// No record set (e.g. a non-query, or an empty stream): return columns/rows empty.
|
||||
return new HistorianSqlResult([], [], returnValue);
|
||||
}
|
||||
|
||||
(string? tableName, List<HistorianSqlColumn> columns) = ParseSchema(schemaXml);
|
||||
IReadOnlyList<IReadOnlyList<object?>> rows = ParseDiffGram(diffGramXml, tableName, columns);
|
||||
return new HistorianSqlResult(columns, rows, returnValue);
|
||||
}
|
||||
|
||||
private static (string? schemaXml, string? diffGramXml) ExtractXmlMembers(ReadOnlyMemory<byte> resultBuffer)
|
||||
{
|
||||
if (resultBuffer.Length == 0)
|
||||
{
|
||||
return (null, null);
|
||||
}
|
||||
|
||||
SerializationRecord root;
|
||||
try
|
||||
{
|
||||
using MemoryStream stream = new(resultBuffer.ToArray(), writable: false);
|
||||
root = NrbfDecoder.Decode(stream);
|
||||
}
|
||||
catch (Exception ex) when (ex is not ProtocolEvidenceMissingException)
|
||||
{
|
||||
throw new ProtocolEvidenceMissingException(
|
||||
$"GetR result buffer is not a decodable NRBF stream: {ex.Message}");
|
||||
}
|
||||
|
||||
if (root is not ClassRecord classRecord || !classRecord.TypeNameMatches(typeof(System.Data.DataTable)))
|
||||
{
|
||||
// TypeNameMatches needs the real type; fall back to the raw type name when the assembly
|
||||
// identity differs from the local one.
|
||||
if (root is not ClassRecord cr || !string.Equals(cr.TypeName.FullName, DataTableTypeName, StringComparison.Ordinal))
|
||||
{
|
||||
throw new ProtocolEvidenceMissingException(
|
||||
$"GetR result root record is not a {DataTableTypeName} (got '{(root as ClassRecord)?.TypeName.FullName ?? root.GetType().Name}').");
|
||||
}
|
||||
classRecord = cr;
|
||||
}
|
||||
|
||||
if (!classRecord.HasMember(XmlSchemaMember) || !classRecord.HasMember(XmlDiffGramMember))
|
||||
{
|
||||
throw new ProtocolEvidenceMissingException(
|
||||
$"GetR DataTable record is missing the {XmlSchemaMember}/{XmlDiffGramMember} members (non-Xml SerializationFormat not supported).");
|
||||
}
|
||||
|
||||
return (classRecord.GetString(XmlSchemaMember), classRecord.GetString(XmlDiffGramMember));
|
||||
}
|
||||
|
||||
private static (string? tableName, List<HistorianSqlColumn> columns) ParseSchema(string schemaXml)
|
||||
{
|
||||
XDocument doc = XDocument.Parse(schemaXml);
|
||||
XElement? schema = doc.Root;
|
||||
List<HistorianSqlColumn> columns = [];
|
||||
if (schema is null)
|
||||
{
|
||||
return (null, columns);
|
||||
}
|
||||
|
||||
// The dataset element advertises the main data table via msdata:MainDataTable.
|
||||
XElement? dataSet = schema.Elements(Xsd + "element")
|
||||
.FirstOrDefault(e => (string?)e.Attribute(MsData + "IsDataSet") == "true");
|
||||
string? mainTable = (string?)dataSet?.Attribute(MsData + "MainDataTable");
|
||||
|
||||
// The table element (named by MainDataTable, else the first non-dataset top-level element).
|
||||
XElement? tableElement = schema.Elements(Xsd + "element")
|
||||
.FirstOrDefault(e => mainTable is not null && (string?)e.Attribute("name") == mainTable)
|
||||
?? schema.Elements(Xsd + "element").FirstOrDefault(e => (string?)e.Attribute(MsData + "IsDataSet") != "true");
|
||||
|
||||
string? tableName = (string?)tableElement?.Attribute("name");
|
||||
|
||||
// Columns are the xs:element children of the table's complexType/sequence.
|
||||
IEnumerable<XElement> columnElements = tableElement
|
||||
?.Descendants(Xsd + "element")
|
||||
.Where(e => e.Attribute("name") is not null)
|
||||
?? [];
|
||||
foreach (XElement column in columnElements)
|
||||
{
|
||||
string name = (string)column.Attribute("name")!;
|
||||
string type = (string?)column.Attribute("type") ?? InlineSimpleType(column) ?? "xs:string";
|
||||
columns.Add(new HistorianSqlColumn(name, type));
|
||||
}
|
||||
|
||||
return (tableName, columns);
|
||||
}
|
||||
|
||||
private static string? InlineSimpleType(XElement column)
|
||||
{
|
||||
XElement? restriction = column.Element(Xsd + "simpleType")?.Element(Xsd + "restriction");
|
||||
return (string?)restriction?.Attribute("base");
|
||||
}
|
||||
|
||||
private static IReadOnlyList<IReadOnlyList<object?>> ParseDiffGram(
|
||||
string? diffGramXml,
|
||||
string? tableName,
|
||||
List<HistorianSqlColumn> columns)
|
||||
{
|
||||
if (string.IsNullOrEmpty(diffGramXml) || columns.Count == 0)
|
||||
{
|
||||
return [];
|
||||
}
|
||||
|
||||
XDocument doc = XDocument.Parse(diffGramXml);
|
||||
XElement? diffgram = doc.Root;
|
||||
if (diffgram is null)
|
||||
{
|
||||
return [];
|
||||
}
|
||||
|
||||
// The dataset element is the diffgram's child that is not in the diffgr: namespace
|
||||
// (diffgr:before / diffgr:errors are change-tracking sections we don't need for a SELECT).
|
||||
XElement? dataSet = diffgram.Elements().FirstOrDefault(e => e.Name.Namespace != DiffGr);
|
||||
if (dataSet is null)
|
||||
{
|
||||
return [];
|
||||
}
|
||||
|
||||
// Row elements are the dataset's children whose local name matches the table (or any child
|
||||
// when the table name is unknown).
|
||||
IEnumerable<XElement> rowElements = dataSet.Elements()
|
||||
.Where(e => tableName is null || e.Name.LocalName == tableName);
|
||||
|
||||
List<IReadOnlyList<object?>> rows = [];
|
||||
foreach (XElement rowElement in rowElements)
|
||||
{
|
||||
object?[] values = new object?[columns.Count];
|
||||
for (int i = 0; i < columns.Count; i++)
|
||||
{
|
||||
HistorianSqlColumn column = columns[i];
|
||||
// ADO.NET maps a column either as a child element or as an attribute on the row.
|
||||
XElement? cell = rowElement.Elements().FirstOrDefault(e => e.Name.LocalName == column.Name);
|
||||
string? raw = cell?.Value
|
||||
?? rowElement.Attributes().FirstOrDefault(a => a.Name.LocalName == column.Name)?.Value;
|
||||
values[i] = raw is null ? null : ConvertValue(raw, column.SchemaType);
|
||||
}
|
||||
rows.Add(values);
|
||||
}
|
||||
|
||||
return rows;
|
||||
}
|
||||
|
||||
private static object ConvertValue(string raw, string schemaType)
|
||||
{
|
||||
string type = schemaType.Contains(':') ? schemaType[(schemaType.IndexOf(':') + 1)..] : schemaType;
|
||||
try
|
||||
{
|
||||
return type switch
|
||||
{
|
||||
"int" or "integer" => int.Parse(raw, CultureInfo.InvariantCulture),
|
||||
"long" => long.Parse(raw, CultureInfo.InvariantCulture),
|
||||
"short" => short.Parse(raw, CultureInfo.InvariantCulture),
|
||||
"byte" => sbyte.Parse(raw, CultureInfo.InvariantCulture),
|
||||
"unsignedByte" => byte.Parse(raw, CultureInfo.InvariantCulture),
|
||||
"unsignedShort" => ushort.Parse(raw, CultureInfo.InvariantCulture),
|
||||
"unsignedInt" => uint.Parse(raw, CultureInfo.InvariantCulture),
|
||||
"unsignedLong" => ulong.Parse(raw, CultureInfo.InvariantCulture),
|
||||
"double" => double.Parse(raw, CultureInfo.InvariantCulture),
|
||||
"float" => float.Parse(raw, CultureInfo.InvariantCulture),
|
||||
"decimal" => decimal.Parse(raw, CultureInfo.InvariantCulture),
|
||||
"boolean" => ParseXsdBoolean(raw),
|
||||
"dateTime" => DateTime.Parse(raw, CultureInfo.InvariantCulture, DateTimeStyles.RoundtripKind),
|
||||
"base64Binary" => Convert.FromBase64String(raw),
|
||||
_ => raw,
|
||||
};
|
||||
}
|
||||
catch (Exception ex) when (ex is FormatException or OverflowException)
|
||||
{
|
||||
// Keep the raw string rather than throwing — the column type is informational and the
|
||||
// server's textual rendering is always available.
|
||||
return raw;
|
||||
}
|
||||
}
|
||||
|
||||
private static bool ParseXsdBoolean(string raw) =>
|
||||
raw is "true" or "1" || (raw is not ("false" or "0") && bool.Parse(raw));
|
||||
}
|
||||
@@ -0,0 +1,132 @@
|
||||
using System.ServiceModel;
|
||||
using System.ServiceModel.Channels;
|
||||
using AVEVA.Historian.Client.Models;
|
||||
using AVEVA.Historian.Client.Wcf.Contracts;
|
||||
|
||||
namespace AVEVA.Historian.Client.Wcf;
|
||||
|
||||
/// <summary>
|
||||
/// Executes SQL commands (HCAL R1.1) over the 2020 WCF <c>aa/Retr/ExeC</c> + <c>aa/Retr/GetR</c>
|
||||
/// ops. Both are string-handle ops reached with the Open2 storage-session GUID formatted UPPERCASE
|
||||
/// (the same handle format that unlocked GETRP/GETHI; see <c>wcf-string-handle-wall.md</c>). The
|
||||
/// record set comes back from <c>GetR</c> as an NRBF-serialized <c>DataTable</c>, parsed by
|
||||
/// <see cref="HistorianSqlResultProtocol"/>.
|
||||
/// </summary>
|
||||
internal static class HistorianWcfSqlClient
|
||||
{
|
||||
// GetR is byte-stream-paged. A small record set returns the whole stream in the first call;
|
||||
// larger ones chunk across calls accumulated here. The cap is a runaway guard.
|
||||
private const int MaxPages = 4096;
|
||||
|
||||
/// <summary>Diagnostic: the ExeC/GetR error from the last call (set only on server rejection).</summary>
|
||||
public static string? LastError { get; private set; }
|
||||
|
||||
public static Task<HistorianSqlResult> ExecuteSqlCommandAsync(
|
||||
HistorianClientOptions options,
|
||||
string command,
|
||||
HistorianSqlExecuteOption option,
|
||||
CancellationToken cancellationToken)
|
||||
{
|
||||
ArgumentException.ThrowIfNullOrWhiteSpace(command);
|
||||
return Task.Run(() => ExecuteSqlCommand(options, command, option), cancellationToken);
|
||||
}
|
||||
|
||||
private static HistorianSqlResult ExecuteSqlCommand(
|
||||
HistorianClientOptions options,
|
||||
string command,
|
||||
HistorianSqlExecuteOption option)
|
||||
{
|
||||
Guid contextKey = Guid.NewGuid();
|
||||
var (histBinding, histEndpoint, retrBinding, retrEndpoint) = HistorianWcfBindingFactory.CreateBindingPair(options);
|
||||
|
||||
LastError = null;
|
||||
HistorianSqlResult? result = null;
|
||||
HistorianWcfAuthChainHelper.OpenAuthenticatedConnection(
|
||||
options, histBinding, histEndpoint, contextKey, CancellationToken.None,
|
||||
additionalSetup: (_, context) =>
|
||||
result = RunSql(retrBinding, retrEndpoint, options, context.StorageSessionId, command, option));
|
||||
return result ?? new HistorianSqlResult([], [], 0);
|
||||
}
|
||||
|
||||
private static HistorianSqlResult RunSql(
|
||||
Binding retrBinding,
|
||||
EndpointAddress retrEndpoint,
|
||||
HistorianClientOptions options,
|
||||
Guid storageSessionId,
|
||||
string command,
|
||||
HistorianSqlExecuteOption option)
|
||||
{
|
||||
// ExeC/GetR take the storage-session GUID as a string handle, uppercase dash-no-braces.
|
||||
string handle = storageSessionId.ToString("D").ToUpperInvariant();
|
||||
|
||||
ChannelFactory<IRetrievalServiceContract3> factory = new(retrBinding, retrEndpoint);
|
||||
HistorianWcfClientCredentialsHelper.Configure(factory, options);
|
||||
IRetrievalServiceContract3 channel = factory.CreateChannel();
|
||||
ICommunicationObject co = (ICommunicationObject)channel;
|
||||
try
|
||||
{
|
||||
// Prime the Retrieval service version handshake (Retr.GetV).
|
||||
channel.GetInterfaceVersion(out _);
|
||||
|
||||
uint queryHandle = 0;
|
||||
bool execOk = channel.ExecuteSqlCommand(
|
||||
handle, command, (uint)option, ref queryHandle,
|
||||
out int returnValue, out uint errSize, out byte[] errBuf);
|
||||
if (!execOk)
|
||||
{
|
||||
LastError = $"ExeC returned false (retValue={returnValue}, errSize={errSize}, errLen={errBuf?.Length ?? 0}).";
|
||||
return new HistorianSqlResult([], [], returnValue);
|
||||
}
|
||||
|
||||
using MemoryStream accumulated = new();
|
||||
uint sequence = 0;
|
||||
for (int page = 0; page < MaxPages; page++)
|
||||
{
|
||||
bool getrOk = channel.GetRecordSetByteStream(
|
||||
handle, queryHandle, ref sequence,
|
||||
out uint resultSize, out byte[] resultBuffer, out uint gErrSize, out byte[] gErrBuf);
|
||||
|
||||
// GetRResult is false even on a successful single-page read — the byte stream is in
|
||||
// resultBuffer regardless, and false signals "this is the final page". So always
|
||||
// consume the buffer first, then stop on a false result or an empty page.
|
||||
if (resultBuffer is { Length: > 0 })
|
||||
{
|
||||
accumulated.Write(resultBuffer, 0, resultBuffer.Length);
|
||||
}
|
||||
|
||||
if (!getrOk || resultBuffer is null || resultBuffer.Length == 0)
|
||||
{
|
||||
if (page == 0 && accumulated.Length == 0 && gErrBuf is { Length: > 0 })
|
||||
{
|
||||
LastError = $"GetR error (resultSize={resultSize}, errSize={gErrSize}, errLen={gErrBuf.Length}).";
|
||||
}
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
byte[] stream = accumulated.ToArray();
|
||||
DumpStreamIfRequested(stream);
|
||||
return HistorianSqlResultProtocol.Parse(stream, returnValue);
|
||||
}
|
||||
finally
|
||||
{
|
||||
try { if (co.State == CommunicationState.Faulted) co.Abort(); else co.Close(); } catch { try { co.Abort(); } catch { } }
|
||||
try { if (factory.State == CommunicationState.Faulted) factory.Abort(); else factory.Close(); } catch { try { factory.Abort(); } catch { } }
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Reverse-engineering aid: when <c>AVEVA_HISTORIAN_SQL_DUMP</c> is set to a path, writes the
|
||||
/// base64 of the reassembled GetR NRBF stream there (the clean contract-level byte[], free of
|
||||
/// the MDAS transport chunk markers that corrupt a raw instrument-wcf capture). Used once to
|
||||
/// produce the golden fixture for <c>WcfSqlResultProtocolTests</c>.
|
||||
/// </summary>
|
||||
private static void DumpStreamIfRequested(byte[] stream)
|
||||
{
|
||||
string? path = Environment.GetEnvironmentVariable("AVEVA_HISTORIAN_SQL_DUMP");
|
||||
if (!string.IsNullOrWhiteSpace(path) && stream.Length > 0)
|
||||
{
|
||||
try { File.WriteAllText(path, Convert.ToBase64String(stream)); } catch { /* diagnostic only */ }
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user