diff --git a/src/lib/sttp.core/DataSubscriber.cs b/src/lib/sttp.core/DataSubscriber.cs index d1e3175a..a2ee15ae 100644 --- a/src/lib/sttp.core/DataSubscriber.cs +++ b/src/lib/sttp.core/DataSubscriber.cs @@ -444,6 +444,7 @@ public UserCommandArgs(ServerCommand command, ServerResponse response, byte[] bu private long m_syncProgressTotalActions; private long m_syncProgressActionsCount; private long m_syncProgressLastMessage; + private long m_syncStatementCount; private bool m_disposed; @@ -921,6 +922,27 @@ public override int ProcessingInterval /// public bool UseTransactionForMetadata { get; set; } + /// + /// Gets or sets the number of records combined into each database command during meta-data synchronization. + /// + /// + /// A value of zero, the default, selects a batch size appropriate for the configured database type. A value + /// of one disables batching so that each record is written with its own statement, which is useful when + /// isolating a suspected batching issue. Values are further constrained by database limits on parameters + /// per command and rows per statement. + /// + public int MetadataSyncBatchSize { get; set; } + + /// + /// Gets or sets flag that determines if bulk copy should be used to insert measurement records during + /// meta-data synchronization when connected to SQL Server. + /// + /// + /// Enabled by default. The bulk path is declined automatically, with a status message, when the optional + /// audit log schema is installed - those triggers only record correct history for single row writes. + /// + public bool UseBulkMetadataSync { get; set; } = true; + /// /// Gets or sets flag that determines whether to use the local clock when calculating statistics. /// @@ -1412,6 +1434,14 @@ public override void Initialize() if (settings.TryGetValue(nameof(UseTransactionForMetadata), out setting)) UseTransactionForMetadata = setting.ParseBoolean(); + // Check if user has defined a batch size for meta-data synchronization + if (settings.TryGetValue(nameof(MetadataSyncBatchSize), out setting) && int.TryParse(setting, out int metadataSyncBatchSize)) + MetadataSyncBatchSize = metadataSyncBatchSize; + + // Check if user has defined a flag for using bulk copy during meta-data synchronization + if (settings.TryGetValue(nameof(UseBulkMetadataSync), out setting)) + UseBulkMetadataSync = setting.ParseBoolean(); + // Check if user has defined a flag for using identity inserts during meta-data synchronization if (settings.TryGetValue(nameof(UseIdentityInsertsForMetadata), out setting)) UseIdentityInsertsForMetadata = setting.ParseBoolean(); @@ -3248,8 +3278,8 @@ protected virtual void SynchronizeMetadata() { bool dataMonitoringEnabled = false; - // TODO: This function is complex and very closely tied to the current time-series data schema - perhaps it should be moved outside this class and referenced - // TODO: as a delegate that can be assigned and called to allow other schemas as well. DataPublisher is already very flexible in what data it can deliver. + // The schema-specific synchronization logic lives in the per-table operations invoked by MetadataSynchronizer. + // This method is responsible only for connection, transaction and progress reporting concerns. try { DataSet? metadata = m_receivedMetadata; @@ -3271,10 +3301,14 @@ protected virtual void SynchronizeMetadata() dataMonitoringEnabled = true; } - // Track total meta-data synchronization process time + // Track total meta-data synchronization process time, as well as time spent in each + // table synchronization phase, so slow phases can be identified from the status message Ticks startTime = DateTime.UtcNow.Ticks; + Ticks deviceSyncTime = 0L, measurementSyncTime = 0L, phasorSyncTime = 0L; DateTime latestUpdateTime = DateTime.MinValue; + m_syncStatementCount = 0L; + // Open the configuration database using settings found in the config file #if NET using (AdoDataConnection database = new(ConfigSettings.Default)) @@ -3295,695 +3329,45 @@ protected virtual void SynchronizeMetadata() if (transaction is not null) command.Transaction = transaction; - // Query the actual record ID based on the known run-time ID for this subscriber device - object? sourceID = ExecuteScalar(command, $"SELECT SourceID FROM Runtime WHERE ID = {ID} AND SourceTable='Device'"); - - if (sourceID is null || sourceID == DBNull.Value) - return; - - int parentID = Convert.ToInt32(sourceID); - - // Validate that the subscriber device is marked as a concentrator (we are about to associate children devices with it) - if (!(ExecuteScalar(command, $"SELECT IsConcentrator FROM Device WHERE ID = {parentID}")?.ToString() ?? "false").ParseBoolean()) - ExecuteNonQuery(command, $"UPDATE Device SET IsConcentrator = 1 WHERE ID = {parentID}"); - - // Get any historian associated with the subscriber device - object? historianID = ExecuteScalar(command, $"SELECT HistorianID FROM Device WHERE ID = {parentID}"); + MetadataSyncContext context = new(database, command, MetadataSynchronizationTimeout, InitSyncProgress, UpdateSyncProgress) + { + SourcePrefix = UseSourcePrefixNames ? $"{Name}!" : "", + Internal = Internal, + MutualSubscription = MutualSubscription, + SyncIndependentDevices = SyncIndependentDevices, + AutoEnableSyncedDevices = AutoEnableSyncedDevices, + UseIdentityInserts = UseIdentityInsertsForMetadata, + AutoDeleteCalculatedMeasurements = AutoDeleteCalculatedMeasurements, + AutoDeleteAlarmMeasurements = AutoDeleteAlarmMeasurements, + ReceiveInternalMetadata = ReceiveInternalMetadata, + ReceiveExternalMetadata = ReceiveExternalMetadata, + LastMetadataRefreshTime = m_lastMetaDataRefreshTime, + BatchSize = MetadataSyncBatchSize, + UseBulkLoad = UseBulkMetadataSync + }; #if !NET - // Determine the active node ID - we cache this since this value won't change for the lifetime of this class - if (m_nodeID == Guid.Empty) - m_nodeID = Guid.Parse(ExecuteScalar(command, $"SELECT NodeID FROM IaonInputAdapter WHERE ID = {(int)ID}")?.ToString() ?? Guid.Empty.ToString()); - - // Determine the protocol record auto-inc ID value for STTP - this value is also cached since it shouldn't change for the lifetime of this class - if (m_sttpProtocolID == 0) - m_sttpProtocolID = int.Parse(ExecuteScalar(command, "SELECT ID FROM Protocol WHERE Acronym='STTP'")?.ToString() ?? "0"); + // Node and protocol IDs are cached for the lifetime of this class since they will not change + context.NodeID = m_nodeID; + context.ProtocolID = m_sttpProtocolID; #endif - // Ascertain total number of actions required for all meta-data synchronization so some level feed back can be provided on progress - InitSyncProgress(metadata.Tables.Cast().Select(dataTable => (long)dataTable.Rows.Count).Sum() + 3); - - // Prefix all children devices with the name of the parent since the same device names could appear in different connections (helps keep device names unique) - string sourcePrefix = UseSourcePrefixNames ? $"{Name}!" : ""; - Dictionary deviceIDs = new(StringComparer.OrdinalIgnoreCase); - DateTime updateTime; - string deviceAcronym; - - // Check to see if data for the "DeviceDetail" table was included in the meta-data - if (metadata.Tables.Contains("DeviceDetail")) - { - DataTable deviceDetail = metadata.Tables["DeviceDetail"]!; - DataRow[] deviceRows; - - // Define SQL statement to query if this device is already defined (this should always be based on the unique guid-based device ID) - string deviceExistsSql = database.ParameterizedQueryString("SELECT COUNT(*) FROM Device WHERE UniqueID = {0}", "uniqueID"); - - #if NET - // Define SQL statement to insert new device record - string insertDeviceSql = database.ParameterizedQueryString("INSERT INTO Device(ParentID, HistorianID, Acronym, Name, OriginalSource, AccessID, Longitude, Latitude, ContactList, ConnectionString, IsConcentrator, Internal, Enabled) " + - "VALUES ({0}, {1}, {2}, {3}, {4}, {5}, {6}, {7}, {8}, {9}, 0, {10}, " + (AutoEnableSyncedDevices ? "1" : "0") + ")", - "parentID", "historianID", "acronym", "name", "originalSource", "accessID", "longitude", "latitude", "contactList", "connectionString", "internal"); - - // Define SQL statement to update existing device record - string updateDeviceSql = database.ParameterizedQueryString("UPDATE Device SET Acronym = {0}, Name = {1}, OriginalSource = {2}, HistorianID = {3}, AccessID = {4}, Longitude = {5}, Latitude = {6}, ContactList = {7}, Internal = {8} WHERE UniqueID = {9}", - "acronym", "name", "originalSource", "historianID", "accessID", "longitude", "latitude", "contactList", "internal", "uniqueID"); - - string updateDeviceWithConnectionStringSql = database.ParameterizedQueryString("UPDATE Device SET Acronym = {0}, Name = {1}, OriginalSource = {2}, HistorianID = {3}, AccessID = {4}, Longitude = {5}, Latitude = {6}, ContactList = {7}, ConnectionString = {8}, Internal = {9} WHERE UniqueID = {10}", - "acronym", "name", "originalSource", "historianID", "accessID", "longitude", "latitude", "contactList", "connectionString", "internal", "uniqueID"); - #else - // Define SQL statement to insert new device record - string insertDeviceSql = database.ParameterizedQueryString("INSERT INTO Device(NodeID, ParentID, HistorianID, Acronym, Name, ProtocolID, FramesPerSecond, OriginalSource, AccessID, Longitude, Latitude, ContactList, ConnectionString, IsConcentrator, Enabled) " + - "VALUES ({0}, {1}, {2}, {3}, {4}, {5}, {6}, {7}, {8}, {9}, {10}, {11}, {12}, 0, " + (AutoEnableSyncedDevices ? "1" : "0") + ")", - "nodeID", "parentID", "historianID", "acronym", "name", "protocolID", "framesPerSecond", "originalSource", "accessID", "longitude", "latitude", "contactList", "connectionString"); - - // Define SQL statement to update existing device record - string updateDeviceSql = database.ParameterizedQueryString("UPDATE Device SET Acronym = {0}, Name = {1}, OriginalSource = {2}, ProtocolID = {3}, FramesPerSecond = {4}, HistorianID = {5}, AccessID = {6}, Longitude = {7}, Latitude = {8}, ContactList = {9} WHERE UniqueID = {10}", - "acronym", "name", "originalSource", "protocolID", "framesPerSecond", "historianID", "accessID", "longitude", "latitude", "contactList", "uniqueID"); - - string updateDeviceWithConnectionStringSql = database.ParameterizedQueryString("UPDATE Device SET Acronym = {0}, Name = {1}, OriginalSource = {2}, ProtocolID = {3}, FramesPerSecond = {4}, HistorianID = {5}, AccessID = {6}, Longitude = {7}, Latitude = {8}, ContactList = {9}, ConnectionString = {10} WHERE UniqueID = {11}", - "acronym", "name", "originalSource", "protocolID", "framesPerSecond", "historianID", "accessID", "longitude", "latitude", "contactList", "connectionString", "uniqueID"); - #endif - - // Define SQL statement to update device's guid-based unique ID after insert - string updateDeviceUniqueIDSql = database.ParameterizedQueryString("UPDATE Device SET UniqueID = {0} WHERE Acronym = {1}", "uniqueID", "acronym"); - - // Define SQL statement to query if a device can be safely updated - string deviceParentRestriction = SyncIndependentDevices ? "OriginalSource <> {1}" : "(ParentID <> {1} OR ParentID IS NULL)"; - string deviceIsUpdateableSql = database.ParameterizedQueryString("SELECT COUNT(*) FROM Device WHERE UniqueID = {0} AND " + deviceParentRestriction, "uniqueID", "parentID"); - - // Define SQL statement to retrieve device's auto-inc ID based on its unique guid-based ID - string queryDeviceIDSql = database.ParameterizedQueryString("SELECT ID FROM Device WHERE UniqueID = {0}", "uniqueID"); - - // Define SQL statement to retrieve all unique device ID's for the current parent to check for mismatches - string queryUniqueDeviceIDsSql = database.ParameterizedQueryString($"SELECT UniqueID FROM Device WHERE {(SyncIndependentDevices ? "OriginalSource" : "ParentID")} = {{0}}", "parentID"); - - // Define SQL statement to remove device records that no longer exist in the meta-data - string deleteDeviceSql = database.ParameterizedQueryString("DELETE FROM Device WHERE UniqueID = {0}", "uniqueID"); - - // Determine which device rows should be synchronized based on operational mode flags - if (ReceiveInternalMetadata && ReceiveExternalMetadata || MutualSubscription) - deviceRows = deviceDetail.Select(); - else if (ReceiveInternalMetadata) - deviceRows = deviceDetail.Select("OriginalSource IS NULL"); - else if (ReceiveExternalMetadata) - deviceRows = deviceDetail.Select("OriginalSource IS NOT NULL"); - else - deviceRows = []; - - // Check existence of optional meta-data fields - DataColumnCollection deviceDetailColumns = deviceDetail.Columns; - bool accessIDFieldExists = deviceDetailColumns.Contains("AccessID"); - bool longitudeFieldExists = deviceDetailColumns.Contains("Longitude"); - bool latitudeFieldExists = deviceDetailColumns.Contains("Latitude"); - bool companyAcronymFieldExists = deviceDetailColumns.Contains("CompanyAcronym"); - bool protocolNameFieldExists = deviceDetailColumns.Contains("ProtocolName"); - bool vendorAcronymFieldExists = deviceDetailColumns.Contains("VendorAcronym"); - bool vendorDeviceNameFieldExists = deviceDetailColumns.Contains("VendorDeviceName"); - bool interconnectionNameFieldExists = deviceDetailColumns.Contains("InterconnectionName"); - bool updatedOnFieldExists = deviceDetailColumns.Contains("UpdatedOn"); - bool connectionStringFieldExists = deviceDetailColumns.Contains("ConnectionString"); - bool framesPerSecondFieldExists = deviceDetailColumns.Contains("FramesPerSecond"); - object parentIDValue = SyncIndependentDevices ? parentID.ToString() : parentID; - int accessID = 0; - - List uniqueIDs = deviceRows - .Select(deviceRow => deviceRow.ConvertGuidField("UniqueID")) - .ToList(); - - // Remove any device records associated with this subscriber that no longer exist in the meta-data - if (uniqueIDs.Count > 0) - { - // ReSharper disable once AccessToDisposedClosure - IEnumerable retiredUniqueIDs = RetrieveData(database, command, queryUniqueDeviceIDsSql, parentIDValue) - .Select() - .Select(deviceRow => deviceRow.ConvertGuidField("UniqueID")) - .Except(uniqueIDs); - - foreach (Guid retiredUniqueID in retiredUniqueIDs) - ExecuteNonQuery(command, deleteDeviceSql, database.Guid(retiredUniqueID)); - - UpdateSyncProgress(); - } - - foreach (DataRow row in deviceRows) - { - Guid uniqueID = row.ConvertGuidField("UniqueID"); - bool recordNeedsUpdating; - - // Determine if record has changed since last synchronization - if (updatedOnFieldExists) - { - try - { - updateTime = Convert.ToDateTime(row["UpdatedOn"]); - recordNeedsUpdating = updateTime > m_lastMetaDataRefreshTime; - - if (updateTime > latestUpdateTime) - latestUpdateTime = updateTime; - } - catch - { - recordNeedsUpdating = true; - } - } - else - { - recordNeedsUpdating = true; - } - - // We will synchronize meta-data only if the source owns this device, and it's not defined as a concentrator (these should normally be filtered by publisher - but we check just in case). - if (!row["IsConcentrator"].ToNonNullString("0").ParseBoolean()) - { - if (accessIDFieldExists) - accessID = row.ConvertField("AccessID"); - - // Get longitude and latitude values if they are defined - decimal longitude = 0M; - decimal latitude = 0M; - decimal? location; - string protocolName = string.Empty; - string connectionString = string.Empty; - - if (longitudeFieldExists) - { - location = row.ConvertNullableField("Longitude"); - - if (location.HasValue) - longitude = location.Value; - } - - if (latitudeFieldExists) - { - location = row.ConvertNullableField("Latitude"); - - if (location.HasValue) - latitude = location.Value; - } - - if (protocolNameFieldExists) - protocolName = row.Field("ProtocolName") ?? string.Empty; - - if (connectionStringFieldExists) - connectionString = row.Field("ConnectionString") ?? string.Empty; - - // Save any reported extraneous values from device meta-data in connection string formatted contact list - all fields are considered optional - Dictionary contactList = new(); - - if (companyAcronymFieldExists) - contactList["companyAcronym"] = row.Field("CompanyAcronym") ?? string.Empty; - - if (protocolNameFieldExists) - contactList["protocolName"] = protocolName; - - if (vendorAcronymFieldExists) - contactList["vendorAcronym"] = row.Field("VendorAcronym") ?? string.Empty; - - if (vendorDeviceNameFieldExists) - contactList["vendorDeviceName"] = row.Field("VendorDeviceName") ?? string.Empty; - - if (interconnectionNameFieldExists) - contactList["interconnectionName"] = row.Field("InterconnectionName") ?? string.Empty; - - #if !NET - int protocolID = m_sttpProtocolID; - #endif - - // If we are synchronizing independent devices, we need to determine the protocol ID for the device - // based on the protocol name defined in the meta-data - if (SyncIndependentDevices && !string.IsNullOrWhiteSpace(protocolName)) - { - #if NET - Dictionary settings = connectionString.ParseKeyValuePairs(); - settings["phasorProtocol"] = protocolName; - connectionString = settings.JoinKeyValuePairs(); - #else - string queryProtocolIDSql = database.ParameterizedQueryString("SELECT ID FROM Protocol WHERE Name = {0}", "protocolName"); - object? protocolIDValue = ExecuteScalar(command, queryProtocolIDSql, protocolName); - - if (protocolIDValue is not null && protocolIDValue is not DBNull) - protocolID = Convert.ToInt32(protocolIDValue); - - if (protocolID == 0) - protocolID = m_sttpProtocolID; - #endif - } - - // For mutual subscriptions where this subscription is owner (i.e., internal is true), we only sync devices that we did not provide - if (!MutualSubscription || !Internal || string.IsNullOrEmpty(row.Field("OriginalSource"))) - { - // Gateway is assuming ownership of the device records when the "internal" flag is true - this means the device's measurements can be forwarded to another party. From a device record perspective, - // ownership is inferred by setting 'OriginalSource' to null. When gateway doesn't own device records (i.e., the "internal" flag is false), this means the device's measurements can only be consumed - // locally - from a device record perspective this means the 'OriginalSource' field is set to the acronym of the PDC or PMU that generated the source measurements. This field allows a mirrored source - // restriction to be implemented later to ensure all devices in an output protocol came from the same original source connection, if desired. - object originalSource = SyncIndependentDevices ? parentID.ToString() : Internal ? DBNull.Value : - string.IsNullOrEmpty(row.Field("ParentAcronym")) ? - sourcePrefix + row.Field("Acronym") : - sourcePrefix + row.Field("ParentAcronym"); - - // Determine if device record already exists - if (Convert.ToInt32(ExecuteScalar(command, deviceExistsSql, database.Guid(uniqueID))) == 0) - { - #if NET - // Insert new device record - ExecuteNonQuery(command, insertDeviceSql, SyncIndependentDevices ? DBNull.Value : parentID, - historianID, sourcePrefix + row.Field("Acronym"), row.Field("Name"), originalSource, - accessID, longitude, latitude, contactList.JoinKeyValuePairs(), connectionString, database.Bool(Internal)); - #else - // Insert new device record - ExecuteNonQuery(command, insertDeviceSql, database.Guid(m_nodeID), SyncIndependentDevices ? DBNull.Value : parentID, - historianID, sourcePrefix + row.Field("Acronym"), row.Field("Name"), protocolID, - framesPerSecondFieldExists ? row.ConvertField("FramesPerSecond") : 30, originalSource, accessID, - longitude, latitude, contactList.JoinKeyValuePairs(), connectionString); - #endif - - // Guids are normally auto-generated during insert - after insertion update the Guid so that it matches the source data. Most of the database - // scripts have triggers that support properly assigning the Guid during an insert, but this code ensures the Guid will always get assigned. - ExecuteNonQuery(command, updateDeviceUniqueIDSql, database.Guid(uniqueID), sourcePrefix + row.Field("Acronym")); - } - else if (recordNeedsUpdating) - { - // Perform safety check to preserve device records which are not safe to overwrite (e.g., device already exists locally as part of another connection) - if (Convert.ToInt32(ExecuteScalar(command, deviceIsUpdateableSql, database.Guid(uniqueID), parentIDValue)) > 0) - continue; - - #if NET - // Update existing device record - if (connectionStringFieldExists) - ExecuteNonQuery(command, updateDeviceWithConnectionStringSql, sourcePrefix + row.Field("Acronym"), row.Field("Name"), - originalSource, historianID, accessID, longitude, latitude, contactList.JoinKeyValuePairs(), connectionString, database.Bool(Internal), database.Guid(uniqueID)); - else - ExecuteNonQuery(command, updateDeviceSql, sourcePrefix + row.Field("Acronym"), row.Field("Name"), - originalSource, historianID, accessID, longitude, latitude, contactList.JoinKeyValuePairs(), database.Bool(Internal), database.Guid(uniqueID)); - #else - // Update existing device record - if (connectionStringFieldExists) - ExecuteNonQuery(command, updateDeviceWithConnectionStringSql, sourcePrefix + row.Field("Acronym"), row.Field("Name"), - originalSource, protocolID, framesPerSecondFieldExists ? row.ConvertField("FramesPerSecond") : 30, historianID, accessID, longitude, latitude, contactList.JoinKeyValuePairs(), connectionString, database.Guid(uniqueID)); - else - ExecuteNonQuery(command, updateDeviceSql, sourcePrefix + row.Field("Acronym"), row.Field("Name"), - originalSource, protocolID, framesPerSecondFieldExists ? row.ConvertField("FramesPerSecond") : 30, historianID, accessID, longitude, latitude, contactList.JoinKeyValuePairs(), database.Guid(uniqueID)); - #endif - } - } - } - - // Capture local device ID auto-inc value for measurement association - deviceIDs[row.Field("Acronym")!] = Convert.ToInt32(ExecuteScalar(command, queryDeviceIDSql, database.Guid(uniqueID))); - - // Periodically notify user about synchronization progress - UpdateSyncProgress(); - } - } - - // Check to see if data for the "MeasurementDetail" table was included in the meta-data - if (metadata.Tables.Contains("MeasurementDetail")) - { - DataTable measurementDetail = metadata.Tables["MeasurementDetail"]!; - List signalIDs = []; - DataRow[] measurementRows; - - // Define SQL statement to query if this measurement is already defined (this should always be based on the unique signal ID Guid) - string measurementExistsSql = database.ParameterizedQueryString("SELECT COUNT(*) FROM Measurement WHERE SignalID = {0}", "signalID"); - - // Define SQL statement to query if this measurement is already defined (this will be used before identity insert) - string identityMeasurementExistsSql = database.ParameterizedQueryString("SELECT COUNT(*) FROM Measurement WHERE PointID = {0}", "pointID"); - - // Define SQL statement to insert new measurement record - string insertMeasurementSql = database.ParameterizedQueryString("INSERT INTO Measurement(DeviceID, HistorianID, PointTag, AlternateTag, SignalTypeID, PhasorSourceIndex, SignalReference, Description, Internal, Subscribed, Enabled) " + - "VALUES ({0}, {1}, {2}, {3}, {4}, {5}, {6}, {7}, {8}, 0, 1)", "deviceID", "historianID", "pointTag", "tempAlternateTagID", "signalTypeID", "phasorSourceIndex", "signalReference", "description", "internal"); - - // Define SQL statement to insert new measurement record - string identityInsertMeasurementSql = database.ParameterizedQueryString("INSERT INTO Measurement(PointID, DeviceID, HistorianID, PointTag, AlternateTag, SignalTypeID, PhasorSourceIndex, SignalReference, Description, Internal, Subscribed, Enabled) " + - "VALUES ({0}, {1}, {2}, {3}, {4}, {5}, {6}, {7}, {8}, {9}, 0, 1)", "pointID", "deviceID", "historianID", "pointTag", "tempAlternateTagID", "signalTypeID", "phasorSourceIndex", "signalReference", "description", "internal"); - - // Define SQL statement to update measurement's signal ID after insert, restoring original signal ID and alternate tag from meta-data - string updateMeasurementSignalIDSql = database.ParameterizedQueryString("UPDATE Measurement SET SignalID = {0}, AlternateTag = {1} WHERE AlternateTag = {2}", "signalID", "alternateTag", "tempAlternateTagID"); - - // Define SQL statement to update existing measurement record - string updateMeasurementSql = database.ParameterizedQueryString("UPDATE Measurement SET HistorianID = {0}, PointTag = {1}, AlternateTag = {2}, SignalTypeID = {3}, PhasorSourceIndex = {4}, SignalReference = {5}, Description = {6}, Internal = {7} WHERE SignalID = {8}", - "historianID", "pointTag", "alternateTag", "signalTypeID", "phasorSourceIndex", "signalReference", "description", "internal", "signalID"); - - // Define SQL statement to update existing measurement record - string identityUpdateMeasurementSql = database.ParameterizedQueryString("UPDATE Measurement SET DeviceID = {0}, HistorianID = {1}, PointTag = {2}, AlternateTag = {3}, SignalTypeID = {4}, PhasorSourceIndex = {5}, SignalReference = {6}, Description = {7}, Internal = {8}, Subscribed = 0, Enabled = 1, SignalID = {9} WHERE PointID = {10}", - "deviceID", "historianID", "pointTag", "tempAlternateTagID", "signalTypeID", "phasorSourceIndex", "signalReference", "description", "internal", "signalID", "pointID"); - - // Define SQL statement to retrieve all measurement signal ID's for the current parent to check for mismatches - note that we use the ActiveMeasurements view - // since it associates measurements with their top-most parent runtime device ID, this allows us to easily query all measurements for the parent device - string queryMeasurementSignalIDsSql = database.ParameterizedQueryString("SELECT SignalID FROM ActiveMeasurement WHERE DeviceID = {0}", "deviceID"); - - // Define SQL statement to retrieve measurement's associated device ID, i.e., actual record ID, based on measurement's signal ID - string queryMeasurementDeviceIDSql = database.ParameterizedQueryString("SELECT DeviceID FROM Measurement WHERE SignalID = {0}", "signalID"); - - // Load signal type ID's from local database associated with their acronym for proper signal type translation - Dictionary signalTypeIDs = new(StringComparer.OrdinalIgnoreCase); - - string? signalTypeAcronym; - - foreach (DataRow row in RetrieveData(database, command, "SELECT ID, Acronym FROM SignalType").Rows) - { - signalTypeAcronym = row.Field("Acronym"); - - if (!string.IsNullOrWhiteSpace(signalTypeAcronym)) - signalTypeIDs[signalTypeAcronym] = row.ConvertField("ID"); - } - - // Define local signal type ID deletion exclusion set - string deleteCondition = ""; - - if (MutualSubscription && !Internal) - { - // For mutual subscriptions where this subscription is renter (i.e., internal is false), do not delete measurements that are locally owned - deleteCondition = " AND Internal == 0"; - } - else - { - List excludedSignalTypeIDs = []; - - // We are intentionally ignoring CALC and ALRM signals during measurement deletion since if you have subscribed to a device and subsequently created local - // calculations and alarms associated with this device, these signals are locally owned and not part of the publisher subscription stream. As a result any - // CALC or ALRM measurements that are created at source and then removed could be orphaned in subscriber. The best fix would be to have a simple flag that - // clearly designates that a measurement was created locally and is not part of the remote synchronization set. - if (!AutoDeleteCalculatedMeasurements && signalTypeIDs.TryGetValue("CALC", out int signalTypeID)) - excludedSignalTypeIDs.Add(signalTypeID); - - if (!AutoDeleteAlarmMeasurements && signalTypeIDs.TryGetValue("ALRM", out signalTypeID)) - excludedSignalTypeIDs.Add(signalTypeID); - - if (excludedSignalTypeIDs.Count > 0) - deleteCondition = $" AND NOT SignalTypeID IN ({excludedSignalTypeIDs.ToDelimitedString(',')})"; - } - - // Define SQL statement to remove device records that no longer exist in the meta-data - string deleteMeasurementSql = database.ParameterizedQueryString($"DELETE FROM Measurement WHERE SignalID = {{0}}{deleteCondition}", "signalID"); - - // Determine which measurement rows should be synchronized based on operational mode flags - if (ReceiveInternalMetadata && ReceiveExternalMetadata) - measurementRows = measurementDetail.Select(); - else if (ReceiveInternalMetadata) - measurementRows = measurementDetail.Select("Internal <> 0"); - else if (ReceiveExternalMetadata) - measurementRows = measurementDetail.Select("Internal = 0"); - else - measurementRows = []; - - // Check existence of optional meta-data fields - DataColumnCollection measurementDetailColumns = measurementDetail.Columns; - bool phasorSourceIndexFieldExists = measurementDetailColumns.Contains("PhasorSourceIndex"); - bool updatedOnFieldExists = measurementDetailColumns.Contains("UpdatedOn"); - bool alternateTagFieldExists = measurementDetailColumns.Contains("AlternateTag"); - - object phasorSourceIndex = DBNull.Value; - object alternateTag = DBNull.Value; - - if (UseIdentityInsertsForMetadata && database.IsSQLServer) - ExecuteNonQuery(command, "SET IDENTITY_INSERT Measurement ON"); - - try - { - foreach (DataRow row in measurementRows) - { - bool recordNeedsUpdating; - - // Determine if record has changed since last synchronization - if (updatedOnFieldExists) - { - try - { - updateTime = Convert.ToDateTime(row["UpdatedOn"]); - recordNeedsUpdating = updateTime > m_lastMetaDataRefreshTime; - - if (updateTime > latestUpdateTime) - latestUpdateTime = updateTime; - } - catch - { - recordNeedsUpdating = true; - } - } - else - { - recordNeedsUpdating = true; - } - - // Get device and signal type acronyms - deviceAcronym = row.Field("DeviceAcronym") ?? string.Empty; - signalTypeAcronym = row.Field("SignalAcronym") ?? string.Empty; - - // Get phasor source index if field is defined - if (phasorSourceIndexFieldExists) - { - // Using ConvertNullableField extension since publisher could use SQLite database in which case - // all integers would arrive in data set as longs and need to be converted back to integers - int? index = row.ConvertNullableField("PhasorSourceIndex"); - phasorSourceIndex = index ?? (object)DBNull.Value; - } - - // Get alternate tag if field is defined - if (alternateTagFieldExists) - alternateTag = row.Field("AlternateTag") ?? (object)DBNull.Value; - - // Make sure we have an associated device and signal type already defined for the measurement - if (!string.IsNullOrWhiteSpace(deviceAcronym) && deviceIDs.ContainsKey(deviceAcronym) && !string.IsNullOrWhiteSpace(signalTypeAcronym) && signalTypeIDs.ContainsKey(signalTypeAcronym)) - { - Guid signalID = row.ConvertGuidField("SignalID"); - - // Track unique measurement signal Guids in this meta-data session, we'll need to remove any old associated measurements that no longer exist - signalIDs.Add(signalID); - - // Prefix the tag name with the "updated" device name - string pointTag = sourcePrefix + row.Field("PointTag"); - - // Look up associated device ID (local DB auto-inc) - int deviceID = deviceIDs[deviceAcronym]; - - // Determine if measurement record already exists - if (Convert.ToInt32(ExecuteScalar(command, measurementExistsSql, database.Guid(signalID))) == 0) - { - string temporaryAlternateTagID = Guid.NewGuid().ToString(); - - // Insert new measurement record - if (UseIdentityInsertsForMetadata && MeasurementKey.TryParse(row.Field("ID")!, out MeasurementKey measurementKey)) - { - long pointID = (long)measurementKey.ID; - - if (Convert.ToInt32(ExecuteScalar(command, identityMeasurementExistsSql, pointID)) == 0) - ExecuteNonQuery(command, identityInsertMeasurementSql, pointID, deviceID, historianID, pointTag, temporaryAlternateTagID, signalTypeIDs[signalTypeAcronym], phasorSourceIndex, sourcePrefix + row.Field("SignalReference"), row.Field("Description") ?? string.Empty, database.Bool(Internal)); - else - ExecuteNonQuery(command, identityUpdateMeasurementSql, deviceID, historianID, pointTag, temporaryAlternateTagID, signalTypeIDs[signalTypeAcronym], phasorSourceIndex, sourcePrefix + row.Field("SignalReference"), row.Field("Description") ?? string.Empty, database.Bool(Internal), database.Guid(signalID), pointID); - } - else - { - ExecuteNonQuery(command, insertMeasurementSql, deviceID, historianID, pointTag, temporaryAlternateTagID, signalTypeIDs[signalTypeAcronym], phasorSourceIndex, sourcePrefix + row.Field("SignalReference"), row.Field("Description") ?? string.Empty, database.Bool(Internal)); - } - - // Guids are normally auto-generated during insert - after insertion update the Guid so that it matches the source data. Most of the database - // scripts have triggers that support properly assigning the Guid during an insert, but this code ensures the Guid will always get assigned. - // TODO: Ensure database schemas define an index on the AlternateTag field to optimize this update - ExecuteNonQuery(command, updateMeasurementSignalIDSql, database.Guid(signalID), alternateTag, temporaryAlternateTagID); - } - else if (recordNeedsUpdating) - { - // Update existing measurement record. Note that this update assumes that measurements will remain associated with a static source device. - ExecuteNonQuery(command, updateMeasurementSql, historianID, pointTag, alternateTag, signalTypeIDs[signalTypeAcronym], phasorSourceIndex, sourcePrefix + row.Field("SignalReference"), row.Field("Description") ?? string.Empty, database.Bool(Internal), database.Guid(signalID)); - } - } - - // Periodically notify user about synchronization progress - UpdateSyncProgress(); - } - } - finally - { - if (UseIdentityInsertsForMetadata && database.IsSQLServer) - ExecuteNonQuery(command, "SET IDENTITY_INSERT Measurement OFF"); - } - - // Remove any measurement records associated with existing devices in this session but no longer exist in the meta-data - if (signalIDs.Count > 0) - { - // Sort signal ID list so that binary search can be used for quick lookups - signalIDs.Sort(); - - // Query all the guid-based signal ID's for all measurement records associated with the parent device using run-time ID - DataTable measurementSignalIDs = RetrieveData(database, command, queryMeasurementSignalIDsSql, (int)ID); - - // Walk through each database record and see if the measurement exists in the provided meta-data - foreach (DataRow measurementRow in measurementSignalIDs.Rows) - { - Guid signalID = measurementRow.ConvertGuidField("SignalID"); - - // Remove any measurements in the database that are associated with received devices and do not exist in the meta-data - if (signalIDs.BinarySearch(signalID) >= 0) - continue; - - // Measurement was not in the meta-data, get the measurement's actual record based ID for its associated device - object? measurementDeviceID = ExecuteScalar(command, queryMeasurementDeviceIDSql, database.Guid(signalID)); - - // If the unknown measurement is directly associated with a device that exists in the meta-data it is assumed that this measurement - // was removed from the publishing system and no longer exists therefore we remove it from the local measurement cache. If the user - // needs custom local measurements associated with a remote device, they should be associated with the parent device only. - if (measurementDeviceID is not null && measurementDeviceID is not DBNull && deviceIDs.ContainsValue(Convert.ToInt32(measurementDeviceID))) - ExecuteNonQuery(command, deleteMeasurementSql, database.Guid(signalID)); - } - - UpdateSyncProgress(); - } - } - - // Check to see if data for the "PhasorDetail" table was included in the meta-data - if (metadata.Tables.Contains("PhasorDetail")) - { - #if NET - const string PrimaryVoltageID = "PrimaryVoltageID"; - const string DestinationPhasorID = "DestinationPhasorID"; - #else - const string PrimaryVoltageID = "DestinationPhasorID"; - const string DestinationPhasorID = "PrimaryVoltageID"; - #endif - - DataTable phasorDetail = metadata.Tables["PhasorDetail"]!; - Dictionary> definedSourceIndices = new(); - Dictionary metadataToDatabaseIDMap = new(); - Dictionary sourceToDestinationIDMap = new(); - - // Phasor data is normally only needed so that the user can properly generate a mirrored IEEE C37.118 output stream from the source data. - // This is necessary since, in this protocol, the phasors are described (i.e., labeled) as a unit (i.e., as a complex number) instead of - // as two distinct angle and magnitude measurements. - - // Define SQL statement to query if phasor record is already defined (no Guid is defined for these simple label records) - string phasorExistsSql = database.ParameterizedQueryString("SELECT COUNT(*) FROM Phasor WHERE DeviceID = {0} AND SourceIndex = {1}", "deviceID", "sourceIndex"); - - #if NET - // Define SQL statement to insert new phasor record - string insertPhasorSql = database.ParameterizedQueryString("INSERT INTO Phasor(DeviceID, Label, Type, Phase, SourceIndex, Internal) VALUES ({0}, {1}, {2}, {3}, {4}, {5})", "deviceID", "label", "type", "phase", "sourceIndex", "internal"); - - // Define SQL statement to update existing phasor record - string updatePhasorSql = database.ParameterizedQueryString("UPDATE Phasor SET Label = {0}, Type = {1}, Phase = {2}, Internal = {3} WHERE DeviceID = {4} AND SourceIndex = {5}", "label", "type", "phase", "internal", "deviceID", "sourceIndex"); - #else - // Define SQL statement to insert new phasor record - string insertPhasorSql = database.ParameterizedQueryString("INSERT INTO Phasor(DeviceID, Label, Type, Phase, SourceIndex) VALUES ({0}, {1}, {2}, {3}, {4})", "deviceID", "label", "type", "phase", "sourceIndex"); - - // Define SQL statement to update existing phasor record - string updatePhasorSql = database.ParameterizedQueryString("UPDATE Phasor SET Label = {0}, Type = {1}, Phase = {2} WHERE DeviceID = {3} AND SourceIndex = {4}", "label", "type", "phase", "deviceID", "sourceIndex"); - #endif - - // Define SQL statement to delete a phasor record - string deletePhasorSql = database.ParameterizedQueryString("DELETE FROM Phasor WHERE DeviceID = {0}", "deviceID"); - - // Define SQL statement to query phasor record ID - string queryPhasorIDSql = database.ParameterizedQueryString("SELECT ID FROM Phasor WHERE DeviceID = {0} AND SourceIndex = {1}", "deviceID", "sourceIndex"); - - // Define SQL statement to update destinationPhasorID field of existing phasor record - string updatePrimaryVoltageIDSql = database.ParameterizedQueryString($"UPDATE Phasor SET {PrimaryVoltageID} = {{0}} WHERE ID = {{1}}", "primaryVoltageID", "id"); - - // Define SQL statement to update phasor BaseKV - string updatePhasorBaseKVSql = database.ParameterizedQueryString("UPDATE Phasor SET BaseKV = {0} WHERE DeviceID = {1} AND SourceIndex = {2}", "baseKV", "deviceID", "sourceIndex"); - - // Check existence of optional meta-data fields - DataColumnCollection phasorDetailColumns = phasorDetail.Columns; - bool phasorIDFieldExists = phasorDetailColumns.Contains("ID"); - bool primaryVoltageIDFieldExists = phasorDetailColumns.Contains(PrimaryVoltageID) || phasorDetailColumns.Contains(DestinationPhasorID); - bool baseKVFieldExists = phasorDetailColumns.Contains("BaseKV"); - - foreach (DataRow row in phasorDetail.Rows) - { - // Get device acronym - deviceAcronym = row.Field("DeviceAcronym") ?? string.Empty; - - // Make sure we have an associated device already defined for the phasor record - // ReSharper disable once CanSimplifyDictionaryLookupWithTryGetValue - if (!string.IsNullOrWhiteSpace(deviceAcronym) && deviceIDs.TryGetValue(deviceAcronym, out int deviceID)) - { - bool recordNeedsUpdating; - - // Determine if record has changed since last synchronization - try - { - updateTime = Convert.ToDateTime(row["UpdatedOn"]); - recordNeedsUpdating = updateTime > m_lastMetaDataRefreshTime; - - if (updateTime > latestUpdateTime) - latestUpdateTime = updateTime; - } - catch - { - recordNeedsUpdating = true; - } - - int sourceIndex = row.ConvertField("SourceIndex"); - bool updateRecord = false; - - // Determine if phasor record already exists - if (Convert.ToInt32(ExecuteScalar(command, phasorExistsSql, deviceID, sourceIndex)) == 0) - { - #if NET - // Insert new phasor record - ExecuteNonQuery(command, insertPhasorSql, deviceID, row.Field("Label") ?? "undefined", (row.Field("Type") ?? "V").TruncateLeft(1), (row.Field("Phase") ?? "+").TruncateLeft(1), sourceIndex, database.Bool(Internal)); - #else - // Insert new phasor record - ExecuteNonQuery(command, insertPhasorSql, deviceID, row.Field("Label") ?? "undefined", (row.Field("Type") ?? "V").TruncateLeft(1), (row.Field("Phase") ?? "+").TruncateLeft(1), sourceIndex); - #endif - updateRecord = true; - } - else if (recordNeedsUpdating) - { - #if NET - // Update existing phasor record - ExecuteNonQuery(command, updatePhasorSql, row.Field("Label") ?? "undefined", (row.Field("Type") ?? "V").TruncateLeft(1), (row.Field("Phase") ?? "+").TruncateLeft(1), database.Bool(Internal), deviceID, sourceIndex); - #else - // Update existing phasor record - ExecuteNonQuery(command, updatePhasorSql, row.Field("Label") ?? "undefined", (row.Field("Type") ?? "V").TruncateLeft(1), (row.Field("Phase") ?? "+").TruncateLeft(1), deviceID, sourceIndex); - #endif - updateRecord = true; - } - - if (updateRecord && baseKVFieldExists) - ExecuteNonQuery(command, updatePhasorBaseKVSql, row.ConvertField("BaseKV"), deviceID, sourceIndex); - - if (phasorIDFieldExists && primaryVoltageIDFieldExists) - { - int sourcePhasorID = row.ConvertField("ID"); - - // Using ConvertNullableField extension since publisher could use SQLite database in which case - // all integers would arrive in data set as longs and need to be converted back to integers - int? destinationPhasorID = row.ConvertNullableField(phasorDetailColumns.Contains(PrimaryVoltageID) ? - PrimaryVoltageID : - DestinationPhasorID); - - if (destinationPhasorID.HasValue) - sourceToDestinationIDMap[sourcePhasorID] = destinationPhasorID.Value; - - // Map all metadata phasor IDs to associated local database phasor IDs - metadataToDatabaseIDMap[sourcePhasorID] = Convert.ToInt32(ExecuteScalar(command, queryPhasorIDSql, deviceID, sourceIndex)); - } - - // Track defined phasors for each device - definedSourceIndices.GetOrAdd(deviceID, _ => []).Add(sourceIndex); - } - - // Periodically notify user about synchronization progress - UpdateSyncProgress(); - } + if (!MetadataSynchronizer.Synchronize(context, metadata, (int)ID)) + return; - // Once all phasor records have been processed, handle updating of destination phasor IDs - foreach (KeyValuePair item in sourceToDestinationIDMap) - { - if (metadataToDatabaseIDMap.TryGetValue(item.Key, out int sourcePhasorID) && metadataToDatabaseIDMap.TryGetValue(item.Value, out int destinationPhasorID)) - ExecuteNonQuery(command, updatePrimaryVoltageIDSql, destinationPhasorID, sourcePhasorID); - } + if (context.BulkLoadStatus is not null) + OnStatusMessage(MessageLevel.Warning, context.BulkLoadStatus); - // For mutual subscriptions where this subscription is owner (i.e., internal is true), do not delete any phasor data - it will be managed by owner only - if (!MutualSubscription || !Internal) - { - // Remove any phasor records associated with existing devices in this session but no longer exist in the meta-data - foreach (int id in deviceIDs.Values) - { - string deleteSql = definedSourceIndices.TryGetValue(id, out List? sourceIndices) ? $"{deletePhasorSql} AND SourceIndex NOT IN ({string.Join(",", sourceIndices)})" : deletePhasorSql; + #if !NET + m_nodeID = context.NodeID; + m_sttpProtocolID = context.ProtocolID; + #endif - ExecuteNonQuery(command, deleteSql, id); - } - } - } + latestUpdateTime = context.LatestUpdateTime; + deviceSyncTime = context.DeviceSyncTime; + measurementSyncTime = context.MeasurementSyncTime; + phasorSyncTime = context.PhasorSyncTime; + m_syncStatementCount = context.StatementCount; transaction?.Commit(); @@ -4042,7 +3426,8 @@ protected virtual void SynchronizeMetadata() m_lastMetaDataRefreshTime = latestUpdateTime > DateTime.MinValue ? latestUpdateTime : DateTime.UtcNow; - OnStatusMessage(MessageLevel.Info, $"Meta-data synchronization completed successfully in {(DateTime.UtcNow.Ticks - startTime).ToElapsedTimeString(2)}"); + OnStatusMessage(MessageLevel.Info, $"Meta-data synchronization completed successfully in {(DateTime.UtcNow.Ticks - startTime).ToElapsedTimeString(2)} using {m_syncStatementCount:N0} database statements " + + $"[devices: {deviceSyncTime.ToElapsedTimeString(2)}, measurements: {measurementSyncTime.ToElapsedTimeString(2)}, phasors: {phasorSyncTime.ToElapsedTimeString(2)}]"); // Send notification that system configuration has changed OnConfigurationChanged(); @@ -4059,40 +3444,6 @@ protected virtual void SynchronizeMetadata() } } - // Since Gemstone data extensions moved position of timeout parameter to a more logical ordinal location so that it will not - // conflict with parameters, we establish overloads for .NET core / framework versions of needed data extension methods -#if NET - private DataTable RetrieveData(AdoDataConnection _, DbCommand command, string sql, params object?[] parameters) - { - return command.RetrieveData(MetadataSynchronizationTimeout, sql, parameters); - } - - private void ExecuteNonQuery(DbCommand command, string sql, params object?[] parameters) - { - command.ExecuteNonQuery(MetadataSynchronizationTimeout, sql, parameters); - } - - private object? ExecuteScalar(DbCommand command, string sql, params object?[] parameters) - { - return command.ExecuteScalar(MetadataSynchronizationTimeout, sql, parameters); - } -#else - private DataTable RetrieveData(AdoDataConnection database, IDbCommand command, string sql, params object?[] parameters) - { - return command.RetrieveData(database.AdapterType, sql, MetadataSynchronizationTimeout, parameters); - } - - private void ExecuteNonQuery(IDbCommand command, string sql, params object?[] parameters) - { - command.ExecuteNonQuery(sql, MetadataSynchronizationTimeout, parameters); - } - - private object? ExecuteScalar(IDbCommand command, string sql, params object?[] parameters) - { - return command.ExecuteScalar(sql, MetadataSynchronizationTimeout, parameters); - } -#endif - private void InitSyncProgress(long totalActions) { m_syncProgressTotalActions = totalActions; diff --git a/src/lib/sttp.core/Metadata/DeviceMetadataSync.cs b/src/lib/sttp.core/Metadata/DeviceMetadataSync.cs new file mode 100644 index 00000000..604448e3 --- /dev/null +++ b/src/lib/sttp.core/Metadata/DeviceMetadataSync.cs @@ -0,0 +1,397 @@ +//****************************************************************************************************** +// DeviceMetadataSync.cs - Gbtc +// +// Copyright © 2026, Grid Protection Alliance. All Rights Reserved. +// +// Licensed to the Grid Protection Alliance (GPA) under one or more contributor license agreements. See +// the NOTICE file distributed with this work for additional information regarding copyright ownership. +// The GPA licenses this file to you under the MIT License (MIT), the "License"; you may +// not use this file except in compliance with the License. You may obtain a copy of the License at: +// +// http://www.opensource.org/licenses/MIT +// +// Unless agreed to in writing, the subject software distributed under the License is distributed on an +// "AS-IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. Refer to the +// License for the specific language governing permissions and limitations. +// +// Code Modification History: +// ---------------------------------------------------------------------------------------------------- +// 08/13/2026 - J. Ritchie Carroll +// Generated original version of source code. +// +//****************************************************************************************************** + +namespace sttp; + +/// +/// Synchronizes the DeviceDetail meta-data table into the local Device table. +/// +internal static class DeviceMetadataSync +{ + /// + /// Describes an existing local device record, as loaded by the pre-synchronization snapshot. + /// + private sealed class DeviceSnapshot + { + public int ID; + public int? ParentID; + public string? OriginalSource; + } + + /// + /// Synchronizes device records and populates , which + /// subsequent measurement and phasor synchronization depend upon. + /// + public static void Synchronize(MetadataSyncContext context, DataTable deviceDetail) + { + AdoDataConnection database = context.Database; + + // Determine which device rows should be synchronized based on operational mode flags + DataRow[] deviceRows; + + if (context.ReceiveInternalMetadata && context.ReceiveExternalMetadata || context.MutualSubscription) + deviceRows = deviceDetail.Select(); + else if (context.ReceiveInternalMetadata) + deviceRows = deviceDetail.Select("OriginalSource IS NULL"); + else if (context.ReceiveExternalMetadata) + deviceRows = deviceDetail.Select("OriginalSource IS NOT NULL"); + else + deviceRows = []; + + // Check existence of optional meta-data fields + DataColumnCollection columns = deviceDetail.Columns; + bool accessIDFieldExists = columns.Contains("AccessID"); + bool longitudeFieldExists = columns.Contains("Longitude"); + bool latitudeFieldExists = columns.Contains("Latitude"); + bool companyAcronymFieldExists = columns.Contains("CompanyAcronym"); + bool protocolNameFieldExists = columns.Contains("ProtocolName"); + bool vendorAcronymFieldExists = columns.Contains("VendorAcronym"); + bool vendorDeviceNameFieldExists = columns.Contains("VendorDeviceName"); + bool interconnectionNameFieldExists = columns.Contains("InterconnectionName"); + bool updatedOnFieldExists = columns.Contains("UpdatedOn"); + bool connectionStringFieldExists = columns.Contains("ConnectionString"); + #if !NET + bool framesPerSecondFieldExists = columns.Contains("FramesPerSecond"); + #endif + + List uniqueIDs = deviceRows + .Select(deviceRow => deviceRow.ConvertGuidField("UniqueID")) + .ToList(); + + // Load a snapshot of all local device records that are relevant to this synchronization pass. This + // replaces the per-row existence, ownership and record ID lookups that were previously issued for + // every single device row. + Dictionary snapshot = LoadSnapshot(context, uniqueIDs, out List ownedUniqueIDs); + + // Remove any device records associated with this subscriber that no longer exist in the meta-data + if (uniqueIDs.Count > 0) + { + HashSet retainedUniqueIDs = new(uniqueIDs); + + List retiredUniqueIDs = ownedUniqueIDs + .Where(uniqueID => !retainedUniqueIDs.Contains(uniqueID)) + .ToList(); + + if (retiredUniqueIDs.Count > 0) + { + foreach (Guid[] chunk in MetadataSyncContext.Chunk(retiredUniqueIDs)) + { + string deleteDeviceSql = context.BuildInListQuery("DELETE FROM Device WHERE UniqueID IN (", chunk.Length, ")", "uniqueID"); + context.ExecuteNonQuery(deleteDeviceSql, chunk.Select(uniqueID => database.Guid(uniqueID)).ToArray()); + } + } + + context.UpdateProgress(); + } + + // Define the batched statements used while applying device changes + string enabledLiteral = context.AutoEnableSyncedDevices ? "1" : "0"; + + #if NET + InsertBatch insertDevices = new(context, + "INSERT INTO Device(UniqueID, ParentID, HistorianID, Acronym, Name, OriginalSource, AccessID, Longitude, Latitude, ContactList, ConnectionString, IsConcentrator, Internal, Enabled)", + $"?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 0, ?, {enabledLiteral}"); + + StatementBatch updateDevices = new(context, + "UPDATE Device SET Acronym = ?, Name = ?, OriginalSource = ?, HistorianID = ?, AccessID = ?, Longitude = ?, Latitude = ?, ContactList = ?, Internal = ? WHERE UniqueID = ?"); + + StatementBatch updateDevicesWithConnectionString = new(context, + "UPDATE Device SET Acronym = ?, Name = ?, OriginalSource = ?, HistorianID = ?, AccessID = ?, Longitude = ?, Latitude = ?, ContactList = ?, ConnectionString = ?, Internal = ? WHERE UniqueID = ?"); + #else + InsertBatch insertDevices = new(context, + "INSERT INTO Device(NodeID, UniqueID, ParentID, HistorianID, Acronym, Name, ProtocolID, FramesPerSecond, OriginalSource, AccessID, Longitude, Latitude, ContactList, ConnectionString, IsConcentrator, Enabled)", + $"?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 0, {enabledLiteral}"); + + StatementBatch updateDevices = new(context, + "UPDATE Device SET Acronym = ?, Name = ?, OriginalSource = ?, ProtocolID = ?, FramesPerSecond = ?, HistorianID = ?, AccessID = ?, Longitude = ?, Latitude = ?, ContactList = ? WHERE UniqueID = ?"); + + StatementBatch updateDevicesWithConnectionString = new(context, + "UPDATE Device SET Acronym = ?, Name = ?, OriginalSource = ?, ProtocolID = ?, FramesPerSecond = ?, HistorianID = ?, AccessID = ?, Longitude = ?, Latitude = ?, ContactList = ?, ConnectionString = ? WHERE UniqueID = ?"); + + // Define SQL statement to look up a protocol record ID by name - only used when synchronizing independent devices + string queryProtocolIDSql = database.ParameterizedQueryString("SELECT ID FROM Protocol WHERE Name = {0}", "protocolName"); + #endif + + // Devices that exist locally but belong to another connection are left untouched - and so are all of + // their child records, since measurement and phasor synchronization both key off 'DeviceIDs'. + HashSet preservedAcronyms = new(StringComparer.OrdinalIgnoreCase); + int accessID = 0; + + foreach (DataRow row in deviceRows) + { + Guid uniqueID = row.ConvertGuidField("UniqueID"); + bool recordNeedsUpdating = context.RecordNeedsUpdating(row, updatedOnFieldExists); + + // We will synchronize meta-data only if the source owns this device, and it's not defined as a concentrator (these should normally be filtered by publisher - but we check just in case). + if (!row["IsConcentrator"].ToNonNullString("0").ParseBoolean()) + { + if (accessIDFieldExists) + accessID = row.ConvertField("AccessID"); + + // Get longitude and latitude values if they are defined + decimal longitude = 0M; + decimal latitude = 0M; + decimal? location; + string protocolName = string.Empty; + string connectionString = string.Empty; + + if (longitudeFieldExists) + { + location = row.ConvertNullableField("Longitude"); + + if (location.HasValue) + longitude = location.Value; + } + + if (latitudeFieldExists) + { + location = row.ConvertNullableField("Latitude"); + + if (location.HasValue) + latitude = location.Value; + } + + if (protocolNameFieldExists) + protocolName = row.Field("ProtocolName") ?? string.Empty; + + if (connectionStringFieldExists) + connectionString = row.Field("ConnectionString") ?? string.Empty; + + // Save any reported extraneous values from device meta-data in connection string formatted contact list - all fields are considered optional + Dictionary contactList = new(); + + if (companyAcronymFieldExists) + contactList["companyAcronym"] = row.Field("CompanyAcronym") ?? string.Empty; + + if (protocolNameFieldExists) + contactList["protocolName"] = protocolName; + + if (vendorAcronymFieldExists) + contactList["vendorAcronym"] = row.Field("VendorAcronym") ?? string.Empty; + + if (vendorDeviceNameFieldExists) + contactList["vendorDeviceName"] = row.Field("VendorDeviceName") ?? string.Empty; + + if (interconnectionNameFieldExists) + contactList["interconnectionName"] = row.Field("InterconnectionName") ?? string.Empty; + + #if !NET + int protocolID = context.ProtocolID; + #endif + + // If we are synchronizing independent devices, we need to determine the protocol ID for the device + // based on the protocol name defined in the meta-data + if (context.SyncIndependentDevices && !string.IsNullOrWhiteSpace(protocolName)) + { + #if NET + Dictionary settings = connectionString.ParseKeyValuePairs(); + settings["phasorProtocol"] = protocolName; + connectionString = settings.JoinKeyValuePairs(); + #else + object? protocolIDValue = context.ExecuteScalar(queryProtocolIDSql, protocolName); + + if (protocolIDValue is not null && protocolIDValue is not DBNull) + protocolID = Convert.ToInt32(protocolIDValue); + + if (protocolID == 0) + protocolID = context.ProtocolID; + #endif + } + + // For mutual subscriptions where this subscription is owner (i.e., internal is true), we only sync devices that we did not provide + if (!context.MutualSubscription || !context.Internal || string.IsNullOrEmpty(row.Field("OriginalSource"))) + { + // Gateway is assuming ownership of the device records when the "internal" flag is true - this means the device's measurements can be forwarded to another party. From a device record perspective, + // ownership is inferred by setting 'OriginalSource' to null. When gateway doesn't own device records (i.e., the "internal" flag is false), this means the device's measurements can only be consumed + // locally - from a device record perspective this means the 'OriginalSource' field is set to the acronym of the PDC or PMU that generated the source measurements. This field allows a mirrored source + // restriction to be implemented later to ensure all devices in an output protocol came from the same original source connection, if desired. + object originalSource = context.SyncIndependentDevices ? context.ParentID.ToString() : context.Internal ? DBNull.Value : + string.IsNullOrEmpty(row.Field("ParentAcronym")) ? + context.SourcePrefix + row.Field("Acronym") : + context.SourcePrefix + row.Field("ParentAcronym"); + + if (!snapshot.TryGetValue(uniqueID, out DeviceSnapshot? existing)) + { + // Insert new device record. Note that the guid-based unique ID is supplied directly rather than + // relying on a database default followed by a corrective update - the previous approach cost an + // extra write per device and could not seek an index, since the unique acronym index is keyed on + // node and acronym together rather than on acronym alone. + #if NET + insertDevices.Add(database.Guid(uniqueID), context.SyncIndependentDevices ? DBNull.Value : context.ParentID, + context.HistorianID, context.SourcePrefix + row.Field("Acronym"), row.Field("Name"), originalSource, + accessID, longitude, latitude, contactList.JoinKeyValuePairs(), connectionString, database.Bool(context.Internal)); + #else + insertDevices.Add(database.Guid(context.NodeID), database.Guid(uniqueID), context.SyncIndependentDevices ? DBNull.Value : context.ParentID, + context.HistorianID, context.SourcePrefix + row.Field("Acronym"), row.Field("Name"), protocolID, + framesPerSecondFieldExists ? row.ConvertField("FramesPerSecond") : 30, originalSource, accessID, + longitude, latitude, contactList.JoinKeyValuePairs(), connectionString); + #endif + } + else + { + // Perform safety check to preserve device records which are not safe to overwrite (e.g., device already exists locally as part of another connection). + // Skipping the record here also keeps it out of the 'DeviceIDs' lookup, which in turn causes all of its measurements and phasors to be skipped as well. + // This is the desired behavior: those local records belong to another connection. + // + // Note that this check is evaluated for every existing record, not only for records that have changed since the last synchronization. It was + // previously nested inside the change check, which meant the protection only held while the source kept updating the device record: once a + // device stopped changing, the check was skipped, the device entered the 'DeviceIDs' lookup, and its measurements and phasors became subject + // to this connection's inserts, updates and - most importantly - the retirement passes, which delete records this connection did not report. + if (!IsOwnedByThisConnection(context, existing)) + { + preservedAcronyms.Add(row.Field("Acronym")!); + context.UpdateProgress(); + continue; + } + + if (recordNeedsUpdating) + { + #if NET + // Update existing device record + if (connectionStringFieldExists) + updateDevicesWithConnectionString.Add(context.SourcePrefix + row.Field("Acronym"), row.Field("Name"), + originalSource, context.HistorianID, accessID, longitude, latitude, contactList.JoinKeyValuePairs(), connectionString, database.Bool(context.Internal), database.Guid(uniqueID)); + else + updateDevices.Add(context.SourcePrefix + row.Field("Acronym"), row.Field("Name"), + originalSource, context.HistorianID, accessID, longitude, latitude, contactList.JoinKeyValuePairs(), database.Bool(context.Internal), database.Guid(uniqueID)); + #else + // Update existing device record + if (connectionStringFieldExists) + updateDevicesWithConnectionString.Add(context.SourcePrefix + row.Field("Acronym"), row.Field("Name"), + originalSource, protocolID, framesPerSecondFieldExists ? row.ConvertField("FramesPerSecond") : 30, context.HistorianID, accessID, longitude, latitude, contactList.JoinKeyValuePairs(), connectionString, database.Guid(uniqueID)); + else + updateDevices.Add(context.SourcePrefix + row.Field("Acronym"), row.Field("Name"), + originalSource, protocolID, framesPerSecondFieldExists ? row.ConvertField("FramesPerSecond") : 30, context.HistorianID, accessID, longitude, latitude, contactList.JoinKeyValuePairs(), database.Guid(uniqueID)); + #endif + } + } + } + } + + // Periodically notify user about synchronization progress + context.UpdateProgress(); + } + + // Pending rows must reach the database before record IDs are read back below + insertDevices.Flush(); + updateDevices.Flush(); + updateDevicesWithConnectionString.Flush(); + + // Capture local device ID auto-inc values for measurement and phasor association. Records inserted above + // do not appear in the snapshot, so record IDs are resolved in bulk here rather than one query per device. + ResolveDeviceIDs(context, deviceRows, uniqueIDs, preservedAcronyms); + } + + /// + /// Determines whether an existing local device record belongs to this subscriber connection and may + /// therefore be safely overwritten. + /// + /// + /// This mirrors the SQL predicate that was previously evaluated per device row. Note the null handling: + /// the original OriginalSource <> {parentID} comparison never matches a null value, since + /// SQL comparisons against null yield null rather than true, so a null original source is treated as + /// owned. The ParentID form explicitly tested for null and treated it as not owned. + /// + private static bool IsOwnedByThisConnection(MetadataSyncContext context, DeviceSnapshot existing) + { + if (context.SyncIndependentDevices) + return existing.OriginalSource is null || existing.OriginalSource.Equals(context.ParentID.ToString(), StringComparison.Ordinal); + + return existing.ParentID.HasValue && existing.ParentID.Value == context.ParentID; + } + + /// + /// Loads existing local device records relevant to this synchronization pass. + /// + /// Current synchronization context. + /// Unique IDs of all devices present in the received meta-data. + /// Unique IDs of all local devices currently owned by this connection. + private static Dictionary LoadSnapshot(MetadataSyncContext context, List uniqueIDs, out List ownedUniqueIDs) + { + Dictionary snapshot = new(); + ownedUniqueIDs = []; + + // Load all devices currently owned by this connection - used both to detect retired records and as part + // of the existence check + string queryOwnedDevicesSql = context.Database.ParameterizedQueryString($"SELECT ID, UniqueID, ParentID, OriginalSource FROM Device WHERE {context.ParentColumn} = {{0}}", "parentID"); + + foreach (DataRow row in context.RetrieveData(queryOwnedDevicesSql, context.ParentIDValue).Rows) + { + Guid uniqueID = row.ConvertGuidField("UniqueID"); + ownedUniqueIDs.Add(uniqueID); + snapshot[uniqueID] = CreateSnapshot(row); + } + + // Devices present in the meta-data may already exist locally under a different owner, so they are looked + // up by unique ID as well - the unique ID column is uniquely indexed, so these lookups seek + foreach (Guid[] chunk in MetadataSyncContext.Chunk(uniqueIDs)) + { + string querySql = context.BuildInListQuery("SELECT ID, UniqueID, ParentID, OriginalSource FROM Device WHERE UniqueID IN (", chunk.Length, ")", "uniqueID"); + + foreach (DataRow row in context.RetrieveData(querySql, chunk.Select(uniqueID => context.Database.Guid(uniqueID)).ToArray()).Rows) + snapshot[row.ConvertGuidField("UniqueID")] = CreateSnapshot(row); + } + + return snapshot; + } + + private static DeviceSnapshot CreateSnapshot(DataRow row) + { + return new DeviceSnapshot + { + ID = row.ConvertField("ID"), + ParentID = row.ConvertNullableField("ParentID"), + OriginalSource = row["OriginalSource"] == DBNull.Value ? null : row.Field("OriginalSource") + }; + } + + /// + /// Populates with the local record ID for every synchronized device. + /// + /// + /// Devices absent from the local database resolve to zero, matching the behavior of the per-row lookup this + /// replaces. Devices preserved because they belong to another connection are deliberately excluded. + /// + private static void ResolveDeviceIDs(MetadataSyncContext context, DataRow[] deviceRows, List uniqueIDs, HashSet preservedAcronyms) + { + Dictionary deviceIDsByUniqueID = new(); + + foreach (Guid[] chunk in MetadataSyncContext.Chunk(uniqueIDs)) + { + string querySql = context.BuildInListQuery("SELECT ID, UniqueID FROM Device WHERE UniqueID IN (", chunk.Length, ")", "uniqueID"); + + foreach (DataRow row in context.RetrieveData(querySql, chunk.Select(uniqueID => context.Database.Guid(uniqueID)).ToArray()).Rows) + deviceIDsByUniqueID[row.ConvertGuidField("UniqueID")] = row.ConvertField("ID"); + } + + foreach (DataRow row in deviceRows) + { + string acronym = row.Field("Acronym")!; + + if (preservedAcronyms.Contains(acronym)) + continue; + + context.DeviceIDs[acronym] = deviceIDsByUniqueID.TryGetValue(row.ConvertGuidField("UniqueID"), out int deviceID) ? deviceID : 0; + } + } +} diff --git a/src/lib/sttp.core/Metadata/MeasurementMetadataSync.cs b/src/lib/sttp.core/Metadata/MeasurementMetadataSync.cs new file mode 100644 index 00000000..a698bf91 --- /dev/null +++ b/src/lib/sttp.core/Metadata/MeasurementMetadataSync.cs @@ -0,0 +1,367 @@ +//****************************************************************************************************** +// MeasurementMetadataSync.cs - Gbtc +// +// Copyright © 2026, Grid Protection Alliance. All Rights Reserved. +// +// Licensed to the Grid Protection Alliance (GPA) under one or more contributor license agreements. See +// the NOTICE file distributed with this work for additional information regarding copyright ownership. +// The GPA licenses this file to you under the MIT License (MIT), the "License"; you may +// not use this file except in compliance with the License. You may obtain a copy of the License at: +// +// http://www.opensource.org/licenses/MIT +// +// Unless agreed to in writing, the subject software distributed under the License is distributed on an +// "AS-IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. Refer to the +// License for the specific language governing permissions and limitations. +// +// Code Modification History: +// ---------------------------------------------------------------------------------------------------- +// 08/13/2026 - J. Ritchie Carroll +// Generated original version of source code. +// +//****************************************************************************************************** + +namespace sttp; + +/// +/// Synchronizes the MeasurementDetail meta-data table into the local Measurement table. +/// +internal static class MeasurementMetadataSync +{ + /// + /// Synchronizes measurement records for all devices captured in . + /// + public static void Synchronize(MetadataSyncContext context, DataTable measurementDetail) + { + AdoDataConnection database = context.Database; + + // Load signal type ID's from local database associated with their acronym for proper signal type translation + foreach (DataRow row in context.RetrieveData("SELECT ID, Acronym FROM SignalType").Rows) + { + string? signalTypeAcronym = row.Field("Acronym"); + + if (!string.IsNullOrWhiteSpace(signalTypeAcronym)) + context.SignalTypeIDs[signalTypeAcronym] = row.ConvertField("ID"); + } + + // Determine which measurement rows should be synchronized based on operational mode flags + DataRow[] measurementRows; + + if (context.ReceiveInternalMetadata && context.ReceiveExternalMetadata) + measurementRows = measurementDetail.Select(); + else if (context.ReceiveInternalMetadata) + measurementRows = measurementDetail.Select("Internal <> 0"); + else if (context.ReceiveExternalMetadata) + measurementRows = measurementDetail.Select("Internal = 0"); + else + measurementRows = []; + + // Check existence of optional meta-data fields + DataColumnCollection columns = measurementDetail.Columns; + bool phasorSourceIndexFieldExists = columns.Contains("PhasorSourceIndex"); + bool updatedOnFieldExists = columns.Contains("UpdatedOn"); + bool alternateTagFieldExists = columns.Contains("AlternateTag"); + + // Define the batched statements used while applying measurement changes. Note that the guid-based signal ID + // is supplied directly on insert. The previous implementation inserted a temporary value into the alternate + // tag field and then issued a corrective 'UPDATE ... WHERE AlternateTag = ' - that column is + // large-object typed and cannot be indexed, so every new measurement cost a full table scan. + InsertBatch insertMeasurements = new(context, + "INSERT INTO Measurement(SignalID, DeviceID, HistorianID, PointTag, AlternateTag, SignalTypeID, PhasorSourceIndex, SignalReference, Description, Internal, Subscribed, Enabled)", + "?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 0, 1"); + + InsertBatch identityInsertMeasurements = new(context, + "INSERT INTO Measurement(PointID, SignalID, DeviceID, HistorianID, PointTag, AlternateTag, SignalTypeID, PhasorSourceIndex, SignalReference, Description, Internal, Subscribed, Enabled)", + "?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 0, 1"); + + StatementBatch updateMeasurements = new(context, + "UPDATE Measurement SET HistorianID = ?, PointTag = ?, AlternateTag = ?, SignalTypeID = ?, PhasorSourceIndex = ?, SignalReference = ?, Description = ?, Internal = ? WHERE SignalID = ?"); + + StatementBatch identityUpdateMeasurements = new(context, + "UPDATE Measurement SET DeviceID = ?, HistorianID = ?, PointTag = ?, AlternateTag = ?, SignalTypeID = ?, PhasorSourceIndex = ?, SignalReference = ?, Description = ?, Internal = ?, Subscribed = 0, Enabled = 1, SignalID = ? WHERE PointID = ?"); + + // Collect the signal IDs, and optionally point IDs, that this synchronization pass will touch + List metadataSignalIDs = []; + List metadataPointIDs = []; + + foreach (DataRow row in measurementRows) + { + string deviceAcronym = row.Field("DeviceAcronym") ?? string.Empty; + string signalTypeAcronym = row.Field("SignalAcronym") ?? string.Empty; + + if (string.IsNullOrWhiteSpace(deviceAcronym) || !context.DeviceIDs.ContainsKey(deviceAcronym) || string.IsNullOrWhiteSpace(signalTypeAcronym) || !context.SignalTypeIDs.ContainsKey(signalTypeAcronym)) + continue; + + metadataSignalIDs.Add(row.ConvertGuidField("SignalID")); + + if (context.UseIdentityInserts && MeasurementKey.TryParse(row.Field("ID")!, out MeasurementKey measurementKey)) + metadataPointIDs.Add((long)measurementKey.ID); + } + + // Load existing records in bulk, replacing the per-row existence probes + HashSet existingSignalIDs = LoadExistingSignalIDs(context, metadataSignalIDs); + HashSet existingPointIDs = context.UseIdentityInserts ? LoadExistingPointIDs(context, metadataPointIDs) : []; + + // On SQL Server, new measurement records are written with SqlBulkCopy rather than batched statements. This + // matters more than raw throughput on the .NET schema, where an insert trigger issues an unscoped + // 'UPDATE Measurement SET SignalID = NEWID() WHERE SignalID IS NULL' - a full table scan for every insert + // statement. One bulk copy fires that trigger once instead of once per batch, and because signal IDs are + // supplied by this code the trigger has nothing left to assign. + SqlServerBulkInsert? bulkInsert = null; + SqlServerBulkInsert? bulkIdentityInsert = null; + SqlServerBulkUpdate? bulkUpdate = null; + + if (context.BulkLoadEnabled) + { + // Updates are staged and applied set based for the same reason. Combining update statements into one + // command reduces round trips but not trigger cost, since each statement in the batch still fires the + // change tracking trigger separately; joining to a staging table collapses a whole batch into one. + bulkUpdate = new SqlServerBulkUpdate(context, "Measurement", "SignalID", + ["HistorianID", "PointTag", "AlternateTag", "SignalTypeID", "PhasorSourceIndex", "SignalReference", "Description", "Internal"]); + + bulkInsert = new SqlServerBulkInsert(context, "Measurement", + ["SignalID", "DeviceID", "HistorianID", "PointTag", "AlternateTag", "SignalTypeID", "PhasorSourceIndex", "SignalReference", "Description", "Internal", "Subscribed", "Enabled"], false); + + if (context.UseIdentityInserts) + { + bulkIdentityInsert = new SqlServerBulkInsert(context, "Measurement", + ["PointID", "SignalID", "DeviceID", "HistorianID", "PointTag", "AlternateTag", "SignalTypeID", "PhasorSourceIndex", "SignalReference", "Description", "Internal", "Subscribed", "Enabled"], true); + } + } + + object phasorSourceIndex = DBNull.Value; + object alternateTag = DBNull.Value; + List signalIDs = []; + + if (context.UseIdentityInserts && database.IsSQLServer) + context.ExecuteNonQuery("SET IDENTITY_INSERT Measurement ON"); + + try + { + foreach (DataRow row in measurementRows) + { + bool recordNeedsUpdating = context.RecordNeedsUpdating(row, updatedOnFieldExists); + + // Get device and signal type acronyms + string deviceAcronym = row.Field("DeviceAcronym") ?? string.Empty; + string signalTypeAcronym = row.Field("SignalAcronym") ?? string.Empty; + + // Get phasor source index if field is defined + if (phasorSourceIndexFieldExists) + { + // Using ConvertNullableField extension since publisher could use SQLite database in which case + // all integers would arrive in data set as longs and need to be converted back to integers + int? index = row.ConvertNullableField("PhasorSourceIndex"); + phasorSourceIndex = index ?? (object)DBNull.Value; + } + + // Get alternate tag if field is defined + if (alternateTagFieldExists) + alternateTag = row.Field("AlternateTag") ?? (object)DBNull.Value; + + // Make sure we have an associated device and signal type already defined for the measurement + if (!string.IsNullOrWhiteSpace(deviceAcronym) && context.DeviceIDs.ContainsKey(deviceAcronym) && !string.IsNullOrWhiteSpace(signalTypeAcronym) && context.SignalTypeIDs.ContainsKey(signalTypeAcronym)) + { + Guid signalID = row.ConvertGuidField("SignalID"); + + // Track unique measurement signal Guids in this meta-data session, we'll need to remove any old associated measurements that no longer exist + signalIDs.Add(signalID); + + // Prefix the tag name with the "updated" device name + string pointTag = context.SourcePrefix + row.Field("PointTag"); + + // Look up associated device ID (local DB auto-inc) + int deviceID = context.DeviceIDs[deviceAcronym]; + int signalTypeID = context.SignalTypeIDs[signalTypeAcronym]; + string signalReference = context.SourcePrefix + row.Field("SignalReference"); + string description = row.Field("Description") ?? string.Empty; + + // Determine if measurement record already exists + if (!existingSignalIDs.Contains(signalID)) + { + // Insert new measurement record + if (context.UseIdentityInserts && MeasurementKey.TryParse(row.Field("ID")!, out MeasurementKey measurementKey)) + { + long pointID = (long)measurementKey.ID; + + if (!existingPointIDs.Contains(pointID)) + { + if (bulkIdentityInsert is not null) + bulkIdentityInsert.Add(pointID, signalID, deviceID, context.HistorianID, pointTag, alternateTag, signalTypeID, phasorSourceIndex, signalReference, description, context.Internal, false, true); + else + identityInsertMeasurements.Add(pointID, database.Guid(signalID), deviceID, context.HistorianID, pointTag, alternateTag, signalTypeID, phasorSourceIndex, signalReference, description, database.Bool(context.Internal)); + } + else + { + identityUpdateMeasurements.Add(deviceID, context.HistorianID, pointTag, alternateTag, signalTypeID, phasorSourceIndex, signalReference, description, database.Bool(context.Internal), database.Guid(signalID), pointID); + } + } + else if (bulkInsert is not null) + { + bulkInsert.Add(signalID, deviceID, context.HistorianID, pointTag, alternateTag, signalTypeID, phasorSourceIndex, signalReference, description, context.Internal, false, true); + } + else + { + insertMeasurements.Add(database.Guid(signalID), deviceID, context.HistorianID, pointTag, alternateTag, signalTypeID, phasorSourceIndex, signalReference, description, database.Bool(context.Internal)); + } + } + else if (recordNeedsUpdating) + { + // Update existing measurement record. Note that this update assumes that measurements will remain associated with a static source device. + if (bulkUpdate is not null) + bulkUpdate.Add(signalID, context.HistorianID, pointTag, alternateTag, signalTypeID, phasorSourceIndex, signalReference, description, context.Internal); + else + updateMeasurements.Add(context.HistorianID, pointTag, alternateTag, signalTypeID, phasorSourceIndex, signalReference, description, database.Bool(context.Internal), database.Guid(signalID)); + } + } + + // Periodically notify user about synchronization progress + context.UpdateProgress(); + } + + // Pending rows must reach the database before identity inserts are disabled below, and before the + // retirement pass reads back the current measurement set + bulkInsert?.Flush(); + bulkIdentityInsert?.Flush(); + bulkUpdate?.Flush(); + insertMeasurements.Flush(); + identityInsertMeasurements.Flush(); + updateMeasurements.Flush(); + identityUpdateMeasurements.Flush(); + } + finally + { + bulkInsert?.Dispose(); + bulkIdentityInsert?.Dispose(); + bulkUpdate?.Dispose(); + + if (context.UseIdentityInserts && database.IsSQLServer) + context.ExecuteNonQuery("SET IDENTITY_INSERT Measurement OFF"); + } + + // Remove any measurement records associated with existing devices in this session but no longer exist in the meta-data + if (signalIDs.Count > 0) + { + RemoveRetiredMeasurements(context, signalIDs); + context.UpdateProgress(); + } + } + + /// + /// Loads the set of signal IDs that already exist in the local database. + /// + private static HashSet LoadExistingSignalIDs(MetadataSyncContext context, List signalIDs) + { + HashSet existing = []; + + foreach (Guid[] chunk in MetadataSyncContext.Chunk(signalIDs)) + { + string querySql = context.BuildInListQuery("SELECT SignalID FROM Measurement WHERE SignalID IN (", chunk.Length, ")", "signalID"); + + foreach (DataRow row in context.RetrieveData(querySql, chunk.Select(signalID => context.Database.Guid(signalID)).ToArray()).Rows) + existing.Add(row.ConvertGuidField("SignalID")); + } + + return existing; + } + + /// + /// Loads the set of point IDs that already exist in the local database, used by the identity insert path. + /// + private static HashSet LoadExistingPointIDs(MetadataSyncContext context, List pointIDs) + { + HashSet existing = []; + + foreach (long[] chunk in MetadataSyncContext.Chunk(pointIDs)) + { + string querySql = context.BuildInListQuery("SELECT PointID FROM Measurement WHERE PointID IN (", chunk.Length, ")", "pointID"); + + foreach (DataRow row in context.RetrieveData(querySql, chunk.Select(pointID => (object)pointID).ToArray()).Rows) + existing.Add(Convert.ToInt64(row["PointID"])); + } + + return existing; + } + + /// + /// Removes measurement records that are associated with synchronized devices but no longer appear in the meta-data. + /// + /// + /// + /// The previous implementation queried the ActiveMeasurement view - an eleven table join that includes a + /// cross join - and then issued a separate query per candidate row to discover its associated device. Both are + /// replaced here by a single restricted query against the Measurement table. + /// + /// + /// The enabled-state restrictions reproduce the filtering that the ActiveMeasurement view applied, so that + /// measurements belonging to a disabled device continue to be left alone. + /// + /// + private static void RemoveRetiredMeasurements(MetadataSyncContext context, List signalIDs) + { + // Independently synchronized devices are not parented to the subscriber device, so the ActiveMeasurement + // view resolved each device to its own run-time ID rather than the subscriber's. The lookup this replaces + // was restricted to the subscriber's run-time ID, which means it never matched anything in this mode and + // retired measurements were silently left in place. That behavior is preserved deliberately: a direct + // query would begin deleting records that have never been deleted before. Whether independently + // synchronized devices should participate in measurement retirement is a separate decision. + if (context.SyncIndependentDevices) + return; + + HashSet retainedSignalIDs = new(signalIDs); + List deviceIDs = context.DeviceIDs.Values.Where(deviceID => deviceID > 0).Distinct().ToList(); + List retiredSignalIDs = []; + + foreach (int[] chunk in MetadataSyncContext.Chunk(deviceIDs)) + { + string querySql = context.BuildInListQuery("SELECT M.SignalID FROM Measurement M INNER JOIN Device D ON M.DeviceID = D.ID WHERE M.DeviceID IN (", chunk.Length, ") AND M.Enabled <> 0 AND D.Enabled <> 0", "deviceID"); + + foreach (DataRow row in context.RetrieveData(querySql, chunk.Select(deviceID => (object)deviceID).ToArray()).Rows) + { + Guid signalID = row.ConvertGuidField("SignalID"); + + if (!retainedSignalIDs.Contains(signalID)) + retiredSignalIDs.Add(signalID); + } + } + + if (retiredSignalIDs.Count == 0) + return; + + // Define local signal type ID deletion exclusion set + string deleteCondition = ""; + + if (context.MutualSubscription && !context.Internal) + { + // For mutual subscriptions where this subscription is renter (i.e., internal is false), do not delete measurements that are locally owned. + // Note that "=" is used here, not "==": the latter is only accepted by SQLite and is a syntax error on all other supported database types. + deleteCondition = " AND Internal = 0"; + } + else + { + List excludedSignalTypeIDs = []; + + // We are intentionally ignoring CALC and ALRM signals during measurement deletion since if you have subscribed to a device and subsequently created local + // calculations and alarms associated with this device, these signals are locally owned and not part of the publisher subscription stream. As a result any + // CALC or ALRM measurements that are created at source and then removed could be orphaned in subscriber. The best fix would be to have a simple flag that + // clearly designates that a measurement was created locally and is not part of the remote synchronization set. + if (!context.AutoDeleteCalculatedMeasurements && context.SignalTypeIDs.TryGetValue("CALC", out int signalTypeID)) + excludedSignalTypeIDs.Add(signalTypeID); + + if (!context.AutoDeleteAlarmMeasurements && context.SignalTypeIDs.TryGetValue("ALRM", out signalTypeID)) + excludedSignalTypeIDs.Add(signalTypeID); + + if (excludedSignalTypeIDs.Count > 0) + deleteCondition = $" AND NOT SignalTypeID IN ({excludedSignalTypeIDs.ToDelimitedString(',')})"; + } + + // Deleting in batches matters disproportionately here: the SQL Server schema defines an INSTEAD OF DELETE + // trigger on Measurement that issues seven dependent delete statements, and that cost was previously paid + // once per retired measurement rather than once per batch. + foreach (Guid[] chunk in MetadataSyncContext.Chunk(retiredSignalIDs)) + { + string deleteMeasurementSql = context.BuildInListQuery("DELETE FROM Measurement WHERE SignalID IN (", chunk.Length, $"){deleteCondition}", "signalID"); + context.ExecuteNonQuery(deleteMeasurementSql, chunk.Select(signalID => context.Database.Guid(signalID)).ToArray()); + } + } +} diff --git a/src/lib/sttp.core/Metadata/MetadataSyncBatch.cs b/src/lib/sttp.core/Metadata/MetadataSyncBatch.cs new file mode 100644 index 00000000..f363a02a --- /dev/null +++ b/src/lib/sttp.core/Metadata/MetadataSyncBatch.cs @@ -0,0 +1,186 @@ +//****************************************************************************************************** +// MetadataSyncBatch.cs - Gbtc +// +// Copyright © 2026, Grid Protection Alliance. All Rights Reserved. +// +// Licensed to the Grid Protection Alliance (GPA) under one or more contributor license agreements. See +// the NOTICE file distributed with this work for additional information regarding copyright ownership. +// The GPA licenses this file to you under the MIT License (MIT), the "License"; you may +// not use this file except in compliance with the License. You may obtain a copy of the License at: +// +// http://www.opensource.org/licenses/MIT +// +// Unless agreed to in writing, the subject software distributed under the License is distributed on an +// "AS-IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. Refer to the +// License for the specific language governing permissions and limitations. +// +// Code Modification History: +// ---------------------------------------------------------------------------------------------------- +// 08/13/2026 - J. Ritchie Carroll +// Generated original version of source code. +// +//****************************************************************************************************** + +namespace sttp; + +/// +/// Accumulates rows for a multi-row INSERT ... VALUES statement, flushing automatically once the +/// configured batch size is reached. +/// +/// +/// +/// The row template uses ? to mark each parameterized value, allowing literal values to be expressed +/// inline, e.g., "?, ?, 0, 1". Placeholders are substituted with generated parameter names at flush +/// time. +/// +/// +/// Batched statements deliberately bypass the framework parameter helpers. Both frameworks re-parse the +/// statement text on every call to infer parameter names, and the .NET Framework tokenizer recognizes only +/// space, parenthesis, comma and equals as delimiters - a parameter adjacent to a semicolon or line break is +/// silently dropped, which would fail with a parameter count mismatch. Parameters are therefore constructed +/// directly against the command. See . +/// +/// +internal sealed class InsertBatch +{ + private readonly MetadataSyncContext m_context; + private readonly string m_insertPrefix; + private readonly string m_rowTemplate; + private readonly int m_parametersPerRow; + private readonly int m_batchSize; + private readonly List m_values; + private int m_rowCount; + + /// + /// Creates a new . + /// + /// Current synchronization context. + /// Statement text up to and including the column list, e.g., "INSERT INTO Phasor(DeviceID, Label)". + /// Single row value list using ? for each parameterized value, e.g., "?, ?, 0". + public InsertBatch(MetadataSyncContext context, string insertPrefix, string rowTemplate) + { + m_context = context; + m_insertPrefix = insertPrefix; + m_rowTemplate = rowTemplate; + m_parametersPerRow = rowTemplate.Count(character => character == '?'); + m_batchSize = context.Dialect.GetBatchSize(m_parametersPerRow, context.Dialect.MaxRowsPerValuesClause, context.BatchSize); + m_values = new List(m_parametersPerRow * m_batchSize); + } + + /// + /// Adds a row to the batch, flushing if the batch is now full. + /// + public void Add(params object?[] rowValues) + { + if (rowValues.Length != m_parametersPerRow) + throw new ArgumentException($"Expected {m_parametersPerRow} values to match insert row template, received {rowValues.Length}.", nameof(rowValues)); + + m_values.AddRange(rowValues); + m_rowCount++; + + if (m_rowCount >= m_batchSize) + Flush(); + } + + /// + /// Writes any accumulated rows to the database. + /// + public void Flush() + { + if (m_rowCount == 0) + return; + + StringBuilder sql = new(m_insertPrefix); + sql.Append(" VALUES "); + + int parameterIndex = 0; + + for (int row = 0; row < m_rowCount; row++) + { + if (row > 0) + sql.Append(", "); + + sql.Append('('); + MetadataSyncContext.AppendRowTemplate(sql, m_rowTemplate, ref parameterIndex); + sql.Append(')'); + } + + m_context.ExecuteBatch(sql.ToString(), m_values); + + m_values.Clear(); + m_rowCount = 0; + } +} + +/// +/// Accumulates repeated executions of a single statement, combining them into semicolon separated commands +/// where the database supports it. +/// +/// +/// Used for the update and delete statement forms, which cannot be expressed as a multi-row value list. On +/// databases that do not support multi-statement commands the batch size collapses to one and the resulting +/// statement stream is identical to issuing each statement individually. +/// +internal sealed class StatementBatch +{ + private readonly MetadataSyncContext m_context; + private readonly string m_statementTemplate; + private readonly int m_parametersPerStatement; + private readonly int m_batchSize; + private readonly List m_values; + private int m_statementCount; + + /// + /// Creates a new . + /// + /// Current synchronization context. + /// Statement text using ? for each parameterized value. + public StatementBatch(MetadataSyncContext context, string statementTemplate) + { + m_context = context; + m_statementTemplate = statementTemplate; + m_parametersPerStatement = statementTemplate.Count(character => character == '?'); + m_batchSize = context.Dialect.GetBatchSize(m_parametersPerStatement, context.Dialect.MaxStatementsPerCommand, context.BatchSize); + m_values = new List(m_parametersPerStatement * m_batchSize); + } + + /// + /// Adds an execution of the statement to the batch, flushing if the batch is now full. + /// + public void Add(params object?[] values) + { + if (values.Length != m_parametersPerStatement) + throw new ArgumentException($"Expected {m_parametersPerStatement} values to match statement template, received {values.Length}.", nameof(values)); + + m_values.AddRange(values); + m_statementCount++; + + if (m_statementCount >= m_batchSize) + Flush(); + } + + /// + /// Writes any accumulated statements to the database. + /// + public void Flush() + { + if (m_statementCount == 0) + return; + + StringBuilder sql = new(); + int parameterIndex = 0; + + for (int statement = 0; statement < m_statementCount; statement++) + { + if (statement > 0) + sql.Append("; "); + + MetadataSyncContext.AppendRowTemplate(sql, m_statementTemplate, ref parameterIndex); + } + + m_context.ExecuteBatch(sql.ToString(), m_values); + + m_values.Clear(); + m_statementCount = 0; + } +} diff --git a/src/lib/sttp.core/Metadata/MetadataSyncContext.cs b/src/lib/sttp.core/Metadata/MetadataSyncContext.cs new file mode 100644 index 00000000..e1ef3ec0 --- /dev/null +++ b/src/lib/sttp.core/Metadata/MetadataSyncContext.cs @@ -0,0 +1,395 @@ +//****************************************************************************************************** +// MetadataSyncContext.cs - Gbtc +// +// Copyright © 2026, Grid Protection Alliance. All Rights Reserved. +// +// Licensed to the Grid Protection Alliance (GPA) under one or more contributor license agreements. See +// the NOTICE file distributed with this work for additional information regarding copyright ownership. +// The GPA licenses this file to you under the MIT License (MIT), the "License"; you may +// not use this file except in compliance with the License. You may obtain a copy of the License at: +// +// http://www.opensource.org/licenses/MIT +// +// Unless agreed to in writing, the subject software distributed under the License is distributed on an +// "AS-IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. Refer to the +// License for the specific language governing permissions and limitations. +// +// Code Modification History: +// ---------------------------------------------------------------------------------------------------- +// 08/13/2026 - J. Ritchie Carroll +// Generated original version of source code. +// +//****************************************************************************************************** + +namespace sttp; + +/// +/// Represents the shared state and database primitives used while synchronizing received meta-data +/// into the local configuration database. +/// +/// +/// +/// A single instance of this class is created per meta-data synchronization pass and handed to each of +/// the per-table synchronization operations, i.e., , +/// and . It owns the reused +/// IDbCommand, the query helpers and the values that flow between the table operations, most +/// notably . +/// +/// +/// The query helpers deliberately route every statement through a single place so that statement counts +/// can be tracked for the completion status message. +/// +/// +internal sealed class MetadataSyncContext +{ + #region [ Members ] + + // Constants + + /// + /// Maximum number of values placed in a single generated IN (...) clause. + /// + /// + /// SQL Server allows a hard maximum of 2,100 parameters per command and large IN lists get + /// progressively more expensive for the query optimizer to compile, so lists are chunked well below + /// the limit. SQLite's default limit is lower still on older builds, hence the conservative value. + /// + public const int MaxInListSize = 500; + + // Fields + + /// Active database connection. + public readonly AdoDataConnection Database; + +#if NET + /// Reused command, already associated with any active transaction. + public readonly DbCommand Command; +#else + /// Reused command, already associated with any active transaction. + public readonly IDbCommand Command; +#endif + + /// Timeout, in seconds, applied to each meta-data synchronization query. + public readonly int Timeout; + + /// Statement batching capabilities and limits for the active database type. + public readonly MetadataSyncDialect Dialect; + + /// User configured batch size, where zero selects the per-database default and one disables batching. + public int BatchSize; + + /// Determines if the SQL Server bulk insert path was requested. + public bool UseBulkLoad; + + /// Determines if the SQL Server bulk insert path is available and permitted. + public bool BulkLoadEnabled; + + /// Explains why the requested bulk insert path was declined, when applicable. + public string? BulkLoadStatus; + + /// Record ID of the local device record that represents this subscriber connection. + public int ParentID; + + /// Historian associated with the subscriber device, if any. + public object? HistorianID; + + /// Prefix applied to synchronized device and point tag names to keep them unique. + public string SourcePrefix = ""; + +#if !NET + /// Active node ID - only defined for the .NET Framework schema. + public Guid NodeID; + + /// Record ID of the STTP protocol - only defined for the .NET Framework schema. + public int ProtocolID; +#endif + + /// Value used to restrict device records to those owned by this subscriber connection. + /// + /// This is the parent device record ID, expressed as a string when + /// is enabled since independently synchronized devices track ownership through the text-based + /// OriginalSource field instead of the integer ParentID field. + /// + public object ParentIDValue = null!; + + /// Name of the device field that identifies ownership by this subscriber connection. + public string ParentColumn = "ParentID"; + + // Synchronization option flags, captured from the owning data subscriber + public bool Internal; + public bool MutualSubscription; + public bool SyncIndependentDevices; + public bool AutoEnableSyncedDevices; + public bool UseIdentityInserts; + public bool AutoDeleteCalculatedMeasurements; + public bool AutoDeleteAlarmMeasurements; + public bool ReceiveInternalMetadata; + public bool ReceiveExternalMetadata; + + /// Time of the last successful meta-data refresh, used to skip unchanged records. + public DateTime LastMetadataRefreshTime; + + /// Latest UpdatedOn value encountered across all synchronized records. + public DateTime LatestUpdateTime = DateTime.MinValue; + + /// Maps a source device acronym to its local device record ID. + public readonly Dictionary DeviceIDs = new(StringComparer.OrdinalIgnoreCase); + + /// Maps a signal type acronym to its local signal type record ID. + public readonly Dictionary SignalTypeIDs = new(StringComparer.OrdinalIgnoreCase); + + /// Total number of database statements issued during this synchronization pass. + public long StatementCount; + + /// Time spent synchronizing device records. + public Ticks DeviceSyncTime; + + /// Time spent synchronizing measurement records. + public Ticks MeasurementSyncTime; + + /// Time spent synchronizing phasor records. + public Ticks PhasorSyncTime; + + private readonly Action m_updateProgress; + private readonly Action m_initProgress; + + #endregion + + #region [ Constructors ] + + /// + /// Creates a new . + /// +#if NET + public MetadataSyncContext(AdoDataConnection database, DbCommand command, int timeout, Action initProgress, Action updateProgress) +#else + public MetadataSyncContext(AdoDataConnection database, IDbCommand command, int timeout, Action initProgress, Action updateProgress) +#endif + { + Database = database; + Command = command; + Timeout = timeout; + Dialect = MetadataSyncDialect.Create(database); + m_initProgress = initProgress; + m_updateProgress = updateProgress; + } + + #endregion + + #region [ Methods ] + + /// + /// Establishes the total number of actions expected during this synchronization pass. + /// + public void InitProgress(long totalActions) + { + m_initProgress(totalActions); + } + + /// + /// Reports incremental synchronization progress to the user. + /// + public void UpdateProgress() + { + m_updateProgress(); + } + + /// + /// Executes a query that returns a result set. + /// + public DataTable RetrieveData(string sql, params object?[] parameters) + { + StatementCount++; + #if NET + return Command.RetrieveData(Timeout, sql, parameters); + #else + return Command.RetrieveData(Database.AdapterType, sql, Timeout, parameters); + #endif + } + + /// + /// Executes a statement that returns no result set. + /// + public void ExecuteNonQuery(string sql, params object?[] parameters) + { + StatementCount++; + #if NET + Command.ExecuteNonQuery(Timeout, sql, parameters); + #else + Command.ExecuteNonQuery(sql, Timeout, parameters); + #endif + } + + /// + /// Executes a query that returns a single value. + /// + public object? ExecuteScalar(string sql, params object?[] parameters) + { + StatementCount++; + #if NET + return Command.ExecuteScalar(Timeout, sql, parameters); + #else + return Command.ExecuteScalar(sql, Timeout, parameters); + #endif + } + + /// + /// Tracks the latest record update time and determines whether a record has changed since the last + /// synchronization pass. + /// + /// Source meta-data row. + /// Flag that determines if the optional UpdatedOn field is defined. + /// true if the record should be updated; otherwise, false. + /// + /// When the UpdatedOn field is missing or cannot be parsed, records are always considered + /// changed - this matches long-standing behavior and errs toward synchronizing too much rather than + /// too little. + /// + public bool RecordNeedsUpdating(DataRow row, bool updatedOnFieldExists) + { + if (!updatedOnFieldExists) + return true; + + try + { + DateTime updateTime = Convert.ToDateTime(row["UpdatedOn"]); + + if (updateTime > LatestUpdateTime) + LatestUpdateTime = updateTime; + + return updateTime > LastMetadataRefreshTime; + } + catch + { + return true; + } + } + + /// + /// Executes a batched statement, binding parameters directly rather than through the framework helpers. + /// + /// Statement text with generated @pN parameter references. + /// Parameter values, in the order the references appear. + /// + /// + /// Both frameworks normally infer parameters by re-parsing the statement text on every execution, which is + /// quadratic in the parameter count and therefore unsuitable for batches of several hundred values. The + /// .NET Framework tokenizer additionally recognizes only space, parenthesis, comma and equals as + /// delimiters, so a parameter adjacent to a semicolon would be dropped and the call would fail. Building + /// the parameter collection here avoids both problems. + /// + /// + /// Values are expected to have already passed through or + /// where the database requires a substitute representation. + /// + /// + public void ExecuteBatch(string sql, IReadOnlyList parameters) + { + StatementCount++; + + Command.CommandText = sql; + Command.CommandTimeout = Timeout; + Command.Parameters.Clear(); + + for (int i = 0; i < parameters.Count; i++) + { + #if NET + DbParameter parameter = Command.CreateParameter(); + #else + IDbDataParameter parameter = Command.CreateParameter(); + #endif + object? value = parameters[i]; + + parameter.ParameterName = ParameterName(i); + parameter.Value = value ?? DBNull.Value; + + #if !NET + // Match the string handling applied by the framework helpers, which default to ANSI strings so that + // comparisons against non-Unicode columns do not incur an implicit conversion + if (value is string && Database.DefaultStringType.HasValue) + parameter.DbType = Database.DefaultStringType.Value; + #endif + + Command.Parameters.Add(parameter); + } + + Command.ExecuteNonQuery(); + } + + /// + /// Appends a row or statement template to , replacing each ? placeholder with + /// a generated parameter reference. + /// + public static void AppendRowTemplate(StringBuilder sql, string template, ref int parameterIndex) + { + foreach (char character in template) + { + if (character == '?') + sql.Append(ParameterName(parameterIndex++)); + else + sql.Append(character); + } + } + + /// + /// Gets the generated name for the parameter at the given ordinal. + /// + /// + /// The @ prefix is accepted by SQL Server, PostgreSQL, MySQL and SQLite. Oracle requires a colon + /// prefix, but its dialect reports no batching support so this path is never reached for Oracle. + /// + private static string ParameterName(int ordinal) + { + return $"@p{ordinal}"; + } + + /// + /// Builds a parameterized statement containing a generated IN (...) value list. + /// + /// SQL text appearing immediately before the value list, e.g., "DELETE FROM Device WHERE UniqueID IN (". + /// Number of values in the list. + /// SQL text appearing immediately after the value list, e.g., ")". + /// Base name used for the generated parameters. + /// Parameterized SQL statement. + /// + /// The generated text is deliberately kept on a single line with values separated by ", ". + /// The .NET Framework parameter tokenizer only treats space, parenthesis, comma and equals as + /// delimiters, so a parameter placed adjacent to any other character - a line break in particular - + /// would not be recognized and the statement would fail with a parameter count mismatch. + /// + public string BuildInListQuery(string prefixSql, int valueCount, string suffixSql, string parameterName) + { + StringBuilder placeholders = new(); + string[] parameterNames = new string[valueCount]; + + for (int i = 0; i < valueCount; i++) + { + if (i > 0) + placeholders.Append(", "); + + placeholders.Append('{').Append(i).Append('}'); + parameterNames[i] = parameterName + i; + } + + return Database.ParameterizedQueryString($"{prefixSql}{placeholders}{suffixSql}", parameterNames); + } + + /// + /// Splits a sequence into chunks no larger than . + /// + public static IEnumerable Chunk(IReadOnlyList items) + { + for (int index = 0; index < items.Count; index += MaxInListSize) + { + int length = Math.Min(MaxInListSize, items.Count - index); + T[] chunk = new T[length]; + + for (int i = 0; i < length; i++) + chunk[i] = items[index + i]; + + yield return chunk; + } + } + + #endregion +} diff --git a/src/lib/sttp.core/Metadata/MetadataSyncDialect.cs b/src/lib/sttp.core/Metadata/MetadataSyncDialect.cs new file mode 100644 index 00000000..870adb70 --- /dev/null +++ b/src/lib/sttp.core/Metadata/MetadataSyncDialect.cs @@ -0,0 +1,195 @@ +//****************************************************************************************************** +// MetadataSyncDialect.cs - Gbtc +// +// Copyright © 2026, Grid Protection Alliance. All Rights Reserved. +// +// Licensed to the Grid Protection Alliance (GPA) under one or more contributor license agreements. See +// the NOTICE file distributed with this work for additional information regarding copyright ownership. +// The GPA licenses this file to you under the MIT License (MIT), the "License"; you may +// not use this file except in compliance with the License. You may obtain a copy of the License at: +// +// http://www.opensource.org/licenses/MIT +// +// Unless agreed to in writing, the subject software distributed under the License is distributed on an +// "AS-IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. Refer to the +// License for the specific language governing permissions and limitations. +// +// Code Modification History: +// ---------------------------------------------------------------------------------------------------- +// 08/13/2026 - J. Ritchie Carroll +// Generated original version of source code. +// +//****************************************************************************************************** + +namespace sttp; + +/// +/// Describes the statement batching capabilities and limits of a particular database type. +/// +/// +/// +/// Meta-data synchronization issues large numbers of very similar statements. Combining them into fewer +/// commands is the single largest remaining cost reduction, but the safe combining strategy and its limits +/// vary by database. This type isolates that variation so the per-table synchronization operations can stay +/// database agnostic. +/// +/// +/// Databases that support neither multi-row value lists nor multi-statement commands - Oracle in particular - +/// simply report a batch size of one, which produces exactly the same statement stream as before. +/// +/// +internal class MetadataSyncDialect +{ + #region [ Properties ] + + /// + /// Gets the maximum number of parameters permitted in a single command. + /// + public virtual int MaxParametersPerCommand => 999; + + /// + /// Gets the maximum number of rows permitted in a single INSERT ... VALUES statement. + /// + public virtual int MaxRowsPerValuesClause => 1; + + /// + /// Gets the maximum number of statements that may be combined into a single command. + /// + public virtual int MaxStatementsPerCommand => 1; + + /// + /// Gets flag that determines if the database supports multi-row INSERT ... VALUES syntax. + /// + public bool SupportsMultiRowValues => MaxRowsPerValuesClause > 1; + + /// + /// Gets flag that determines if the database supports semicolon separated multi-statement commands. + /// + public bool SupportsMultiStatementCommands => MaxStatementsPerCommand > 1; + + #endregion + + #region [ Methods ] + + /// + /// Determines how many rows of a given width may be combined into a single command. + /// + /// Number of parameters required by a single row. + /// Batching limit reported by this dialect for the statement form in use. + /// User configured batch size, where zero selects the dialect default. + public int GetBatchSize(int parametersPerRow, int maxRows, int requestedBatchSize) + { + if (parametersPerRow < 1) + parametersPerRow = 1; + + int batchSize = Math.Min(maxRows, MaxParametersPerCommand / parametersPerRow); + + if (requestedBatchSize > 0) + batchSize = Math.Min(batchSize, requestedBatchSize); + + return Math.Max(1, batchSize); + } + + #endregion + + #region [ Static ] + + /// + /// Creates the appropriate for the supplied database connection. + /// + public static MetadataSyncDialect Create(AdoDataConnection database) + { + return database.DatabaseType switch + { + DatabaseType.SQLServer => new SqlServerMetadataSyncDialect(), + DatabaseType.PostgreSQL => new PostgreSqlMetadataSyncDialect(), + DatabaseType.SQLite => new SqliteMetadataSyncDialect(), + DatabaseType.MySQL => new MySqlMetadataSyncDialect(), + + // Oracle supports neither multi-row value lists nor semicolon separated commands without wrapping + // statements in an anonymous PL/SQL block, and Access supports neither at all - both fall back to + // the unbatched behavior provided by this base implementation + _ => new MetadataSyncDialect() + }; + } + + #endregion +} + +/// +/// Statement batching limits for SQL Server. +/// +internal sealed class SqlServerMetadataSyncDialect : MetadataSyncDialect +{ + /// + /// SQL Server permits a hard maximum of 2,100 parameters per command. + /// + public override int MaxParametersPerCommand => 2000; + + /// + /// SQL Server permits a hard maximum of 1,000 rows per INSERT ... VALUES statement. + /// + public override int MaxRowsPerValuesClause => 1000; + + /// + public override int MaxStatementsPerCommand => 250; +} + +/// +/// Statement batching limits for PostgreSQL. +/// +internal sealed class PostgreSqlMetadataSyncDialect : MetadataSyncDialect +{ + /// + /// PostgreSQL permits 65,535 parameters per command; a lower value is used to keep individual + /// commands small enough to parse and plan quickly. + /// + public override int MaxParametersPerCommand => 8000; + + /// + public override int MaxRowsPerValuesClause => 1000; + + /// + public override int MaxStatementsPerCommand => 250; +} + +/// +/// Statement batching limits for SQLite. +/// +/// +/// SQLite benefits from batching more than any other supported database. Its per-row insert triggers issue +/// follow-up update statements against the same table, and without an enclosing transaction every statement +/// is separately committed to disk. +/// +internal sealed class SqliteMetadataSyncDialect : MetadataSyncDialect +{ + /// + /// Recent SQLite builds permit 32,766 parameters, but builds compiled with the older default permit only + /// 999, so the conservative limit is used. + /// + public override int MaxParametersPerCommand => 900; + + /// + public override int MaxRowsPerValuesClause => 500; + + /// + public override int MaxStatementsPerCommand => 250; +} + +/// +/// Statement batching limits for MySQL and MariaDB. +/// +internal sealed class MySqlMetadataSyncDialect : MetadataSyncDialect +{ + /// + public override int MaxParametersPerCommand => 8000; + + /// + public override int MaxRowsPerValuesClause => 1000; + + /// + /// Multi-statement commands are rejected unless the connection was opened with the option enabled, so + /// only value list batching is used. + /// + public override int MaxStatementsPerCommand => 1; +} diff --git a/src/lib/sttp.core/Metadata/MetadataSynchronizer.cs b/src/lib/sttp.core/Metadata/MetadataSynchronizer.cs new file mode 100644 index 00000000..2ff6d315 --- /dev/null +++ b/src/lib/sttp.core/Metadata/MetadataSynchronizer.cs @@ -0,0 +1,127 @@ +//****************************************************************************************************** +// MetadataSynchronizer.cs - Gbtc +// +// Copyright © 2026, Grid Protection Alliance. All Rights Reserved. +// +// Licensed to the Grid Protection Alliance (GPA) under one or more contributor license agreements. See +// the NOTICE file distributed with this work for additional information regarding copyright ownership. +// The GPA licenses this file to you under the MIT License (MIT), the "License"; you may +// not use this file except in compliance with the License. You may obtain a copy of the License at: +// +// http://www.opensource.org/licenses/MIT +// +// Unless agreed to in writing, the subject software distributed under the License is distributed on an +// "AS-IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. Refer to the +// License for the specific language governing permissions and limitations. +// +// Code Modification History: +// ---------------------------------------------------------------------------------------------------- +// 08/13/2026 - J. Ritchie Carroll +// Generated original version of source code. +// +//****************************************************************************************************** + +namespace sttp; + +/// +/// Orchestrates synchronization of a received meta-data into the local +/// configuration database. +/// +/// +/// This type owns the ordering of the individual table synchronization operations - devices must be +/// synchronized first since both measurements and phasors are associated with them through +/// . +/// +internal static class MetadataSynchronizer +{ + /// + /// Synchronizes the supplied meta-data into the local configuration database. + /// + /// Synchronization context, already associated with an open connection and command. + /// Received meta-data to synchronize. + /// Run-time ID of the owning data subscriber adapter. + /// true if synchronization was performed; otherwise, false. + public static bool Synchronize(MetadataSyncContext context, DataSet metadata, int runtimeID) + { + if (!LoadSubscriberDeviceInfo(context, runtimeID)) + return false; + + // Determine whether the SQL Server bulk insert path may be used. Databases other than SQL Server simply + // do not qualify and that is not worth reporting, but a SQL Server connection that declines the path has + // a specific reason the operator should see. + if (context.UseBulkLoad && context.Database.IsSQLServer) + { + context.BulkLoadEnabled = SqlServerBulkInsert.IsSupported(context, out string? declineReason); + + if (!context.BulkLoadEnabled) + context.BulkLoadStatus = $"Bulk meta-data loading was requested but is not being used: {declineReason}."; + } + + // Ascertain total number of actions required for all meta-data synchronization so some level feed back can be provided on progress + context.InitProgress(metadata.Tables.Cast().Select(dataTable => (long)dataTable.Rows.Count).Sum() + 3); + + Ticks phaseStartTime = DateTime.UtcNow.Ticks; + + // Check to see if data for the "DeviceDetail" table was included in the meta-data + if (metadata.Tables.Contains("DeviceDetail")) + DeviceMetadataSync.Synchronize(context, metadata.Tables["DeviceDetail"]!); + + context.DeviceSyncTime = DateTime.UtcNow.Ticks - phaseStartTime; + phaseStartTime = DateTime.UtcNow.Ticks; + + // Check to see if data for the "MeasurementDetail" table was included in the meta-data + if (metadata.Tables.Contains("MeasurementDetail")) + MeasurementMetadataSync.Synchronize(context, metadata.Tables["MeasurementDetail"]!); + + context.MeasurementSyncTime = DateTime.UtcNow.Ticks - phaseStartTime; + phaseStartTime = DateTime.UtcNow.Ticks; + + // Check to see if data for the "PhasorDetail" table was included in the meta-data + if (metadata.Tables.Contains("PhasorDetail")) + PhasorMetadataSync.Synchronize(context, metadata.Tables["PhasorDetail"]!); + + context.PhasorSyncTime = DateTime.UtcNow.Ticks - phaseStartTime; + + return true; + } + + /// + /// Resolves the local device record that represents this subscriber connection, along with the values + /// derived from it that the table synchronization operations depend upon. + /// + /// false when no subscriber device record could be resolved, in which case synchronization is skipped. + private static bool LoadSubscriberDeviceInfo(MetadataSyncContext context, int runtimeID) + { + // Query the actual record ID based on the known run-time ID for this subscriber device + object? sourceID = context.ExecuteScalar($"SELECT SourceID FROM Runtime WHERE ID = {runtimeID} AND SourceTable='Device'"); + + if (sourceID is null || sourceID == DBNull.Value) + return false; + + context.ParentID = Convert.ToInt32(sourceID); + + // Validate that the subscriber device is marked as a concentrator (we are about to associate children devices with it) + if (!(context.ExecuteScalar($"SELECT IsConcentrator FROM Device WHERE ID = {context.ParentID}")?.ToString() ?? "false").ParseBoolean()) + context.ExecuteNonQuery($"UPDATE Device SET IsConcentrator = 1 WHERE ID = {context.ParentID}"); + + // Get any historian associated with the subscriber device + context.HistorianID = context.ExecuteScalar($"SELECT HistorianID FROM Device WHERE ID = {context.ParentID}"); + + #if !NET + // Determine the active node ID - we cache this since this value won't change for the lifetime of the owning class + if (context.NodeID == Guid.Empty) + context.NodeID = Guid.Parse(context.ExecuteScalar($"SELECT NodeID FROM IaonInputAdapter WHERE ID = {runtimeID}")?.ToString() ?? Guid.Empty.ToString()); + + // Determine the protocol record auto-inc ID value for STTP - this value is also cached since it shouldn't change for the lifetime of the owning class + if (context.ProtocolID == 0) + context.ProtocolID = int.Parse(context.ExecuteScalar("SELECT ID FROM Protocol WHERE Acronym='STTP'")?.ToString() ?? "0"); + #endif + + // Devices synchronized independently track ownership through the text-based 'OriginalSource' field rather + // than the integer 'ParentID' field, since they are not parented to the subscriber device record + context.ParentColumn = context.SyncIndependentDevices ? "OriginalSource" : "ParentID"; + context.ParentIDValue = context.SyncIndependentDevices ? context.ParentID.ToString() : context.ParentID; + + return true; + } +} diff --git a/src/lib/sttp.core/Metadata/PhasorMetadataSync.cs b/src/lib/sttp.core/Metadata/PhasorMetadataSync.cs new file mode 100644 index 00000000..c88c7084 --- /dev/null +++ b/src/lib/sttp.core/Metadata/PhasorMetadataSync.cs @@ -0,0 +1,242 @@ +//****************************************************************************************************** +// PhasorMetadataSync.cs - Gbtc +// +// Copyright © 2026, Grid Protection Alliance. All Rights Reserved. +// +// Licensed to the Grid Protection Alliance (GPA) under one or more contributor license agreements. See +// the NOTICE file distributed with this work for additional information regarding copyright ownership. +// The GPA licenses this file to you under the MIT License (MIT), the "License"; you may +// not use this file except in compliance with the License. You may obtain a copy of the License at: +// +// http://www.opensource.org/licenses/MIT +// +// Unless agreed to in writing, the subject software distributed under the License is distributed on an +// "AS-IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. Refer to the +// License for the specific language governing permissions and limitations. +// +// Code Modification History: +// ---------------------------------------------------------------------------------------------------- +// 08/13/2026 - J. Ritchie Carroll +// Generated original version of source code. +// +//****************************************************************************************************** + +namespace sttp; + +/// +/// Synchronizes the PhasorDetail meta-data table into the local Phasor table. +/// +/// +/// Phasor data is normally only needed so that the user can properly generate a mirrored IEEE C37.118 +/// output stream from the source data. This is necessary since, in this protocol, the phasors are +/// described (i.e., labeled) as a unit (i.e., as a complex number) instead of as two distinct angle and +/// magnitude measurements. +/// +internal static class PhasorMetadataSync +{ + /// + /// Synchronizes phasor records for all devices captured in . + /// + public static void Synchronize(MetadataSyncContext context, DataTable phasorDetail) + { + #if NET + const string PrimaryVoltageID = "PrimaryVoltageID"; + const string DestinationPhasorID = "DestinationPhasorID"; + #else + const string PrimaryVoltageID = "DestinationPhasorID"; + const string DestinationPhasorID = "PrimaryVoltageID"; + #endif + + AdoDataConnection database = context.Database; + + // Check existence of optional meta-data fields + DataColumnCollection columns = phasorDetail.Columns; + bool phasorIDFieldExists = columns.Contains("ID"); + bool primaryVoltageIDFieldExists = columns.Contains(PrimaryVoltageID) || columns.Contains(DestinationPhasorID); + bool baseKVFieldExists = columns.Contains("BaseKV"); + + // Define SQL statements used while applying phasor changes. The BaseKV assignment is folded into the primary + // insert and update statements rather than being issued as a separate follow-up statement: every phasor update + // fires a trigger that joins the eleven table ActiveMeasurement view, so halving the number of write statements + // against this table halves that cost. + #if NET + const string InternalColumn = ", Internal"; + const string InternalValue = ", ?"; + const string InternalAssignment = ", Internal = ?"; + #else + const string InternalColumn = ""; + const string InternalValue = ""; + const string InternalAssignment = ""; + #endif + + string baseKVColumn = baseKVFieldExists ? ", BaseKV" : ""; + string baseKVValue = baseKVFieldExists ? ", ?" : ""; + string baseKVAssignment = baseKVFieldExists ? ", BaseKV = ?" : ""; + + InsertBatch insertPhasors = new(context, + $"INSERT INTO Phasor(DeviceID, Label, Type, Phase, SourceIndex{InternalColumn}{baseKVColumn})", + $"?, ?, ?, ?, ?{InternalValue}{baseKVValue}"); + + StatementBatch updatePhasors = new(context, + $"UPDATE Phasor SET Label = ?, Type = ?, Phase = ?{InternalAssignment}{baseKVAssignment} WHERE DeviceID = ? AND SourceIndex = ?"); + + // Define SQL statement to update destination phasor ID field of existing phasor record + string updatePrimaryVoltageIDSql = database.ParameterizedQueryString($"UPDATE Phasor SET {PrimaryVoltageID} = {{0}} WHERE ID = {{1}}", "primaryVoltageID", "id"); + + // Load a snapshot of existing phasor records, replacing the per-row existence and record ID lookups + Dictionary<(int DeviceID, int SourceIndex), int> snapshot = LoadSnapshot(context); + + Dictionary> definedSourceIndices = new(); + Dictionary sourceToDestinationIDMap = new(); + List<(int SourcePhasorID, int DeviceID, int SourceIndex)> phasorIDLookups = []; + + foreach (DataRow row in phasorDetail.Rows) + { + // Get device acronym + string deviceAcronym = row.Field("DeviceAcronym") ?? string.Empty; + + // Make sure we have an associated device already defined for the phasor record + if (!string.IsNullOrWhiteSpace(deviceAcronym) && context.DeviceIDs.TryGetValue(deviceAcronym, out int deviceID)) + { + bool recordNeedsUpdating = context.RecordNeedsUpdating(row, true); + + int sourceIndex = row.ConvertField("SourceIndex"); + string label = row.Field("Label") ?? "undefined"; + string type = (row.Field("Type") ?? "V").TruncateLeft(1); + string phase = (row.Field("Phase") ?? "+").TruncateLeft(1); + + // Determine if phasor record already exists + if (!snapshot.ContainsKey((deviceID, sourceIndex))) + { + // Insert new phasor record + List values = [deviceID, label, type, phase, sourceIndex]; + + #if NET + values.Add(database.Bool(context.Internal)); + #endif + + if (baseKVFieldExists) + values.Add(row.ConvertField("BaseKV")); + + insertPhasors.Add(values.ToArray()); + } + else if (recordNeedsUpdating) + { + // Update existing phasor record + List values = [label, type, phase]; + + #if NET + values.Add(database.Bool(context.Internal)); + #endif + + if (baseKVFieldExists) + values.Add(row.ConvertField("BaseKV")); + + values.Add(deviceID); + values.Add(sourceIndex); + + updatePhasors.Add(values.ToArray()); + } + + if (phasorIDFieldExists && primaryVoltageIDFieldExists) + { + int sourcePhasorID = row.ConvertField("ID"); + + // Using ConvertNullableField extension since publisher could use SQLite database in which case + // all integers would arrive in data set as longs and need to be converted back to integers + int? destinationPhasorID = row.ConvertNullableField(columns.Contains(PrimaryVoltageID) ? + PrimaryVoltageID : + DestinationPhasorID); + + if (destinationPhasorID.HasValue) + sourceToDestinationIDMap[sourcePhasorID] = destinationPhasorID.Value; + + // Map all metadata phasor IDs to associated local database phasor IDs - resolved in bulk below + phasorIDLookups.Add((sourcePhasorID, deviceID, sourceIndex)); + } + + // Track defined phasors for each device + definedSourceIndices.GetOrAdd(deviceID, _ => []).Add(sourceIndex); + } + + // Periodically notify user about synchronization progress + context.UpdateProgress(); + } + + // Pending rows must reach the database before record IDs are read back below + insertPhasors.Flush(); + updatePhasors.Flush(); + + // Once all phasor records have been processed, handle updating of destination phasor IDs + if (phasorIDLookups.Count > 0) + { + // Reload the snapshot so that records inserted above are included, then resolve every metadata phasor + // ID to its local record ID in one pass rather than one query per phasor + Dictionary<(int DeviceID, int SourceIndex), int> resolved = LoadSnapshot(context); + Dictionary metadataToDatabaseIDMap = new(); + + foreach ((int sourcePhasorID, int deviceID, int sourceIndex) in phasorIDLookups) + metadataToDatabaseIDMap[sourcePhasorID] = resolved.TryGetValue((deviceID, sourceIndex), out int phasorID) ? phasorID : 0; + + foreach (KeyValuePair item in sourceToDestinationIDMap) + { + if (metadataToDatabaseIDMap.TryGetValue(item.Key, out int sourcePhasorID) && metadataToDatabaseIDMap.TryGetValue(item.Value, out int destinationPhasorID)) + context.ExecuteNonQuery(updatePrimaryVoltageIDSql, destinationPhasorID, sourcePhasorID); + } + } + + // For mutual subscriptions where this subscription is owner (i.e., internal is true), do not delete any phasor data - it will be managed by owner only + if (context.MutualSubscription && context.Internal) + return; + + RemoveRetiredPhasors(context, definedSourceIndices); + } + + /// + /// Loads existing phasor records for all synchronized devices, keyed by device and source index. + /// + private static Dictionary<(int DeviceID, int SourceIndex), int> LoadSnapshot(MetadataSyncContext context) + { + Dictionary<(int, int), int> snapshot = new(); + List deviceIDs = context.DeviceIDs.Values.Where(deviceID => deviceID > 0).Distinct().ToList(); + + foreach (int[] chunk in MetadataSyncContext.Chunk(deviceIDs)) + { + string querySql = context.BuildInListQuery("SELECT ID, DeviceID, SourceIndex FROM Phasor WHERE DeviceID IN (", chunk.Length, ")", "deviceID"); + + foreach (DataRow row in context.RetrieveData(querySql, chunk.Select(deviceID => (object)deviceID).ToArray()).Rows) + snapshot[(row.ConvertField("DeviceID"), row.ConvertField("SourceIndex"))] = row.ConvertField("ID"); + } + + return snapshot; + } + + /// + /// Removes phasor records associated with synchronized devices that no longer appear in the meta-data. + /// + private static void RemoveRetiredPhasors(MetadataSyncContext context, Dictionary> definedSourceIndices) + { + // Devices that reported no phasors at all can be cleared in batches; devices that reported some phasors need + // a per-device statement since each carries its own retained source index list + List devicesWithoutPhasors = []; + + foreach (int deviceID in context.DeviceIDs.Values.Where(deviceID => deviceID > 0).Distinct()) + { + if (definedSourceIndices.TryGetValue(deviceID, out List? sourceIndices)) + { + string deletePhasorSql = context.Database.ParameterizedQueryString($"DELETE FROM Phasor WHERE DeviceID = {{0}} AND SourceIndex NOT IN ({string.Join(",", sourceIndices)})", "deviceID"); + context.ExecuteNonQuery(deletePhasorSql, deviceID); + } + else + { + devicesWithoutPhasors.Add(deviceID); + } + } + + foreach (int[] chunk in MetadataSyncContext.Chunk(devicesWithoutPhasors)) + { + string deletePhasorSql = context.BuildInListQuery("DELETE FROM Phasor WHERE DeviceID IN (", chunk.Length, ")", "deviceID"); + context.ExecuteNonQuery(deletePhasorSql, chunk.Select(deviceID => (object)deviceID).ToArray()); + } + } +} diff --git a/src/lib/sttp.core/Metadata/SqlServerBulkInsert.cs b/src/lib/sttp.core/Metadata/SqlServerBulkInsert.cs new file mode 100644 index 00000000..7ed71a98 --- /dev/null +++ b/src/lib/sttp.core/Metadata/SqlServerBulkInsert.cs @@ -0,0 +1,219 @@ +//****************************************************************************************************** +// SqlServerBulkInsert.cs - Gbtc +// +// Copyright © 2026, Grid Protection Alliance. All Rights Reserved. +// +// Licensed to the Grid Protection Alliance (GPA) under one or more contributor license agreements. See +// the NOTICE file distributed with this work for additional information regarding copyright ownership. +// The GPA licenses this file to you under the MIT License (MIT), the "License"; you may +// not use this file except in compliance with the License. You may obtain a copy of the License at: +// +// http://www.opensource.org/licenses/MIT +// +// Unless agreed to in writing, the subject software distributed under the License is distributed on an +// "AS-IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. Refer to the +// License for the specific language governing permissions and limitations. +// +// Code Modification History: +// ---------------------------------------------------------------------------------------------------- +// 08/13/2026 - J. Ritchie Carroll +// Generated original version of source code. +// +//****************************************************************************************************** + +#if NET +using Microsoft.Data.SqlClient; +#else +using System.Data.SqlClient; +#endif + +namespace sttp; + +/// +/// Writes rows to a SQL Server table using . +/// +/// +/// +/// This is the fastest available insert path for SQL Server and is used for measurement records, which +/// dominate meta-data volume. Device and phasor records continue to use ordinary batched statements: their +/// volumes are orders of magnitude lower, and device inserts drive a trigger that maintains the +/// Runtime table, which the run-time views depend upon. +/// +/// +/// Triggers are explicitly enabled. This costs some throughput but is required for correctness - the +/// .NET Framework schema maintains change tracking through an insert trigger on Measurement, and +/// silently skipping it would leave the rest of the system unaware that configuration had changed. Because +/// the trigger is statement level, a single bulk copy fires it once rather than once per batch. +/// +/// +/// The staging takes its schema from the destination table itself rather than +/// assuming column types. The two supported schemas differ here: the .NET Framework schema stores signal IDs +/// as uniqueidentifier while the .NET schema stores them as nvarchar(36), and +/// is far less forgiving about type mismatches than a parameterized statement. +/// +/// +internal sealed class SqlServerBulkInsert : IDisposable +{ + private readonly MetadataSyncContext m_context; + private readonly string m_tableName; + private readonly DataTable m_stagingTable; + private readonly int m_batchSize; + private readonly bool m_keepIdentity; + private bool m_disposed; + + /// + /// Creates a new for the given table and column set. + /// + /// Current synchronization context. + /// Destination table name. + /// Columns that will be supplied, in the order values are added. + /// Determines if supplied identity column values should be preserved. + public SqlServerBulkInsert(MetadataSyncContext context, string tableName, string[] columns, bool keepIdentity) + { + m_context = context; + m_tableName = tableName; + m_keepIdentity = keepIdentity; + m_batchSize = context.BatchSize > 0 ? context.BatchSize : 10000; + + // Take the staging schema from the destination table so that column types match exactly + m_stagingTable = context.RetrieveData($"SELECT TOP 0 {string.Join(", ", columns)} FROM {tableName}"); + m_stagingTable.TableName = tableName; + } + + /// + /// Gets the number of rows currently awaiting a write. + /// + public int PendingRows => m_stagingTable.Rows.Count; + + /// + /// Adds a row to the batch, flushing if the batch is now full. + /// + /// + /// Values are coerced to the destination column type, which allows callers to supply a + /// without knowing whether the target column is a native unique identifier or text. + /// + public void Add(params object?[] values) + { + if (values.Length != m_stagingTable.Columns.Count) + throw new ArgumentException($"Expected {m_stagingTable.Columns.Count} values to match bulk insert column list, received {values.Length}.", nameof(values)); + + DataRow row = m_stagingTable.NewRow(); + + for (int i = 0; i < values.Length; i++) + row[i] = Coerce(values[i], m_stagingTable.Columns[i].DataType); + + m_stagingTable.Rows.Add(row); + + if (m_stagingTable.Rows.Count >= m_batchSize) + Flush(); + } + + /// + /// Writes any accumulated rows to the database. + /// + public void Flush() + { + if (m_stagingTable.Rows.Count == 0) + return; + + SqlBulkCopyOptions options = SqlBulkCopyOptions.FireTriggers; + + if (m_keepIdentity) + options |= SqlBulkCopyOptions.KeepIdentity; + + using (SqlBulkCopy bulkCopy = new((SqlConnection)m_context.Database.Connection, options, (SqlTransaction?)m_context.Command.Transaction)) + { + bulkCopy.DestinationTableName = m_tableName; + bulkCopy.BulkCopyTimeout = m_context.Timeout; + bulkCopy.BatchSize = m_batchSize; + + // Map by name so that column order in the staging table is irrelevant + foreach (DataColumn column in m_stagingTable.Columns) + bulkCopy.ColumnMappings.Add(column.ColumnName, column.ColumnName); + + bulkCopy.WriteToServer(m_stagingTable); + } + + m_context.StatementCount++; + m_stagingTable.Rows.Clear(); + } + + /// + /// Converts a value supplied by the caller to the type expected by the destination column. + /// + internal static object Coerce(object? value, Type columnType) + { + if (value is null || value == DBNull.Value) + return DBNull.Value; + + if (columnType == typeof(string) && value is Guid guidValue) + return guidValue.ToString(); + + if (columnType == typeof(Guid) && value is string stringValue) + return Guid.Parse(stringValue); + + if (value.GetType() == columnType) + return value; + + return Convert.ChangeType(value, columnType); + } + + /// + /// Releases resources held by this . + /// + public void Dispose() + { + if (m_disposed) + return; + + m_stagingTable.Dispose(); + m_disposed = true; + } + + /// + /// Determines whether the bulk insert path may be used against the connected database. + /// + /// + /// The optional audit log schema shipped with the .NET Framework applications defines triggers whose + /// bodies assign from an arbitrary single row of the inserted pseudo-table, e.g., + /// SELECT @id = CONVERT(NVARCHAR(MAX), SignalID) FROM #inserted. Those triggers only produce + /// correct output when exactly one row is affected, so multi-row writes are declined outright rather + /// than silently recording misleading audit history. + /// + public static bool IsSupported(MetadataSyncContext context, out string? declineReason) + { + declineReason = null; + + if (!context.Database.IsSQLServer) + { + declineReason = "database is not SQL Server"; + return false; + } + + if (context.Database.Connection is not SqlConnection) + { + declineReason = "connection is not a SQL Server client connection"; + return false; + } + + try + { + object? auditTriggerCount = context.ExecuteScalar( + "SELECT COUNT(*) FROM sys.triggers t INNER JOIN sys.tables tb ON t.parent_id = tb.object_id " + + "WHERE tb.name IN ('Device', 'Measurement', 'Phasor') AND t.is_disabled = 0 AND t.name LIKE '%Audit%'"); + + if (auditTriggerCount is not null && auditTriggerCount != DBNull.Value && Convert.ToInt32(auditTriggerCount) > 0) + { + declineReason = "audit log triggers are installed, and they only record correct history for single row writes"; + return false; + } + } + catch (Exception ex) + { + declineReason = $"audit trigger check failed: {ex.Message}"; + return false; + } + + return true; + } +} diff --git a/src/lib/sttp.core/Metadata/SqlServerBulkUpdate.cs b/src/lib/sttp.core/Metadata/SqlServerBulkUpdate.cs new file mode 100644 index 00000000..08fba69d --- /dev/null +++ b/src/lib/sttp.core/Metadata/SqlServerBulkUpdate.cs @@ -0,0 +1,172 @@ +//****************************************************************************************************** +// SqlServerBulkUpdate.cs - Gbtc +// +// Copyright © 2026, Grid Protection Alliance. All Rights Reserved. +// +// Licensed to the Grid Protection Alliance (GPA) under one or more contributor license agreements. See +// the NOTICE file distributed with this work for additional information regarding copyright ownership. +// The GPA licenses this file to you under the MIT License (MIT), the "License"; you may +// not use this file except in compliance with the License. You may obtain a copy of the License at: +// +// http://www.opensource.org/licenses/MIT +// +// Unless agreed to in writing, the subject software distributed under the License is distributed on an +// "AS-IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. Refer to the +// License for the specific language governing permissions and limitations. +// +// Code Modification History: +// ---------------------------------------------------------------------------------------------------- +// 08/16/2026 - J. Ritchie Carroll +// Generated original version of source code. +// +//****************************************************************************************************** + +#if NET +using Microsoft.Data.SqlClient; +#else +using System.Data.SqlClient; +#endif + +namespace sttp; + +/// +/// Applies bulk updates to a SQL Server table by staging rows in a temporary table and issuing a single +/// set based UPDATE ... FROM per batch. +/// +/// +/// +/// Combining update statements into one command reduces round trips but not trigger cost: semicolon +/// separated statements are still separate statements, so a statement level trigger fires once for each of +/// them. Measured against a 193,000 measurement re-synchronization, that left the update path dominated by +/// trigger work - the .NET Framework schema's change tracking trigger creates and drops two temporary +/// tables every time it fires. +/// +/// +/// Staging the rows and joining to them collapses an entire batch into one statement, so the trigger fires +/// once per batch rather than once per row, while still recording the same change tracking rows. +/// +/// +internal sealed class SqlServerBulkUpdate : IDisposable +{ + private readonly MetadataSyncContext m_context; + private readonly string m_targetTable; + private readonly string m_stagingTableName; + private readonly string m_updateSql; + private readonly DataTable m_stagingTable; + private readonly int m_batchSize; + private bool m_stagingTableCreated; + private bool m_disposed; + + /// + /// Creates a new . + /// + /// Current synchronization context. + /// Table to update. + /// Column used to match staged rows to target rows. + /// Columns to assign, in the order values are added after the key. + public SqlServerBulkUpdate(MetadataSyncContext context, string targetTable, string keyColumn, string[] updateColumns) + { + m_context = context; + m_targetTable = targetTable; + m_stagingTableName = $"#{targetTable}BulkUpdate"; + m_batchSize = context.BatchSize > 0 ? context.BatchSize : 10000; + + string columnList = string.Join(", ", new[] { keyColumn }.Concat(updateColumns)); + + // Take the staging schema from the destination table so that column types match exactly + m_stagingTable = context.RetrieveData($"SELECT TOP 0 {columnList} FROM {targetTable}"); + m_stagingTable.TableName = m_stagingTableName; + + string assignments = string.Join(", ", updateColumns.Select(column => $"target.{column} = staged.{column}")); + m_updateSql = $"UPDATE target SET {assignments} FROM {targetTable} AS target INNER JOIN {m_stagingTableName} AS staged ON target.{keyColumn} = staged.{keyColumn}"; + } + + /// + /// Adds a row to the batch, flushing if the batch is now full. + /// + public void Add(params object?[] values) + { + if (values.Length != m_stagingTable.Columns.Count) + throw new ArgumentException($"Expected {m_stagingTable.Columns.Count} values to match bulk update column list, received {values.Length}.", nameof(values)); + + DataRow row = m_stagingTable.NewRow(); + + for (int i = 0; i < values.Length; i++) + row[i] = SqlServerBulkInsert.Coerce(values[i], m_stagingTable.Columns[i].DataType); + + m_stagingTable.Rows.Add(row); + + if (m_stagingTable.Rows.Count >= m_batchSize) + Flush(); + } + + /// + /// Applies any accumulated rows to the target table. + /// + public void Flush() + { + if (m_stagingTable.Rows.Count == 0) + return; + + EnsureStagingTable(); + + // Triggers are irrelevant on the staging table itself, so the fastest copy options apply here; the + // subsequent set based update fires the target table's triggers normally + using (SqlBulkCopy bulkCopy = new((SqlConnection)m_context.Database.Connection, SqlBulkCopyOptions.TableLock, (SqlTransaction?)m_context.Command.Transaction)) + { + bulkCopy.DestinationTableName = m_stagingTableName; + bulkCopy.BulkCopyTimeout = m_context.Timeout; + bulkCopy.BatchSize = m_batchSize; + + foreach (DataColumn column in m_stagingTable.Columns) + bulkCopy.ColumnMappings.Add(column.ColumnName, column.ColumnName); + + bulkCopy.WriteToServer(m_stagingTable); + } + + m_context.StatementCount++; + m_context.ExecuteNonQuery(m_updateSql); + m_context.ExecuteNonQuery($"TRUNCATE TABLE {m_stagingTableName}"); + + m_stagingTable.Rows.Clear(); + } + + /// + /// Creates the session scoped staging table on first use. + /// + private void EnsureStagingTable() + { + if (m_stagingTableCreated) + return; + + string columnList = string.Join(", ", m_stagingTable.Columns.Cast().Select(column => column.ColumnName)); + + // Deriving the staging table from the target guarantees identical column types without hard coding them + m_context.ExecuteNonQuery($"SELECT TOP 0 {columnList} INTO {m_stagingTableName} FROM {m_targetTable}"); + m_stagingTableCreated = true; + } + + /// + /// Releases resources held by this . + /// + public void Dispose() + { + if (m_disposed) + return; + + if (m_stagingTableCreated) + { + try + { + m_context.ExecuteNonQuery($"DROP TABLE {m_stagingTableName}"); + } + catch + { + // Staging table is session scoped and will be reclaimed when the connection closes + } + } + + m_stagingTable.Dispose(); + m_disposed = true; + } +} diff --git a/src/lib/sttp.core/sttp.core.projitems b/src/lib/sttp.core/sttp.core.projitems index da2dae77..7a32daf2 100644 --- a/src/lib/sttp.core/sttp.core.projitems +++ b/src/lib/sttp.core/sttp.core.projitems @@ -15,6 +15,15 @@ + + + + + + + + +