Transactional Group Records Getting Stuck In Store & Forward

[chase.mason.se]

Background:

I’m making an API call using Ignitions scripting module, the response of which is a JSON structure with roughly 200 items. I’m splitting the structure up and inserting them as rows into a SQL server on the same SQL Instance as my Ignition DB.

=================================================================

Problem:

1. I initially used a sleep function to throttle the amount of outgoing calls to limit tractions to SQL. This sleep function caused a clock drift on the gateway. This caused a desync between Ignition and the backend SQL instance and everything started populating in Store & Forward.

a. This approach is discouraged.

b. Recovery was to disable / reenable SQL connection, and or bounce the gateway.

2. Removing the sleep function, script runs fine but Transactional Group records start getting stuck in Store & Forward.

a. In this case I suspect a performance spike SQL side since I cant throttle connection.

b. SQL server is still responsive, other services are not impact.

c. Checked query performance runs at .0386 ms roughly.

=================================================================

Problem Statement:

Gateway clock drift / SQL performance spikes seem to trap Transactional Group records in Store / Forwards.

1. Not sure how to notify the group when this happens.

2. Manual recovery required?

=================================================================

Test Case:

1. To try to get a bit more detail on the issue I tried to replicate the issue locally on my PC with a local instance of Ignition / SQL.

2. I set up a transactional group, and locked a DB on the SQL instance to see if records would get stuck in Store and Forward… No luck

o I ran the same script locally and no issue on the Ignition instance.

o Mind you the scale is a lot smaller.

=================================================================

Question:

1. For the use case of inserting (X) number of records from an API response, what is the best way to throttle input SQL side?

o I don’t believe invoke later works scripting side.

o I cant delay the active thread without causing clock drift on the gateway.

o I could cache results with one script and process the records with a secondary script. I’ve been actively discouraged to make files for this approach.

§ If a file was made, potentially could do a SQL merge to reduce burden SQL side.

o Potentially could limit the scope via the API call, but given the number of records preference would be to have the delay roughly around 50 ms per call not seconds.

2. For the Ignition Gateway, do I need to be concerned about long running scripts adding to the clock drift?

a. I thought the gateway was multithreaded, but it was acting as if it was executing scripts procedurally on a single thread.

[chase.mason.se]

From what I can find, it recommends a controlled merge into SQL via a batch insert into a temp SQL table. Alternatively I could delete records from the existing table and just batch insert.

[chase.mason.se]

Had some issues getting Ignition to run the bulk insert as a named query.

Ended up settling on exporting the data set to a CSV on a shared network location. From there a SQL Job performs the bulk insert using the file. Frequency set at once a week. As long as I space out when the data sets are made Ignition side, should greatly reduce the time it takes the script to operate and limit the number of requests SQL side to one per group.

Think this will work.

[Clancy_Cavanaugh]

Seems you found a solution, but I will share my solution as another option.

The issue on my end stemmed from each row insert being done as a separate query and the delay of waiting for the response between each query.

My only reason for using a separate table before merging was because I needed to upsert (update if the id already exists, and insert if it does not). Unfortunately Microsoft SQL Server does not have an upsert command. If you only need to insert rows, then I would modify this to insert directly to the correct table and avoid the stored procedure call.

I would not call it perfect, but it reduced the runtime of the script from 10-15 minutes to under a minute (probably faster, been a while since I tested).

Script Function:

def api_sql_write(table_name, api_data):
	# type: (str, list[dict]) -> None
	""" Inserts rows into Temp table, then calls stored procedure to merge with actual table.
		
		Args:
			table_name (str): Name of SQL table to be updated. Note: this should not be the Temp table.
			api_data (list[dict]): Array of dictionaries to be upserted. Each dictionary represents one row.
		
		Returns:
			None
	"""
	# Standardized logger format (ProjectName.function_name)
	logger = system.util.getLogger('{}.{}'.format(system.util.getProjectName(), sys._getframe().f_code.co_name))
	if api_data == []:
		logger.debug("No data to insert into {}.".format(table_name))
		return
	
	base_query = """ INSERT INTO Temp_{} """.format(table_name)
	
	# Get Column Names from SQL
	columns = system.db.runQuery(""" SELECT COLUMN_NAME FROM INFORMATION_SCHEMA.COLUMNS WHERE TABLE_NAME = '{}'""".format(table_name), database='My_Database').getColumnAsList(0)
	
	if columns == []:
		logger.warn('No columns found for table {}. Are you sure you provided the correct table name?'.format(table_name))
		return

	# Filters api_data to only include keys that are in the column list from sql.
	api_data = [{k: d[k] for k in columns if k in d} for d in api_data]
	
	# convert column list to sql string
	columns = '({})'.format(', '.join(['"{}"'.format(item) for item in api_data[0].keys()]))
	num_of_columns = len(api_data[0].keys())
	
	base_query += columns + ' VALUES '
	
	# Our query will be run as a prepared statement to avoid issues with special characters in strings.
	# Create a row in the form (?, ?, ..., ?, ?) based on the number of columns in the dataset.
	# We concatenate a number of these onto our query based on the number of rows in the dataset.
	row_template = '({})'.format(', '.join(['?' for x in range(num_of_columns)]))
	
	# This is a limit of either Microsoft SQL Server or the Java SQL Driver.  To get around it we will break the insert into chunks.
	sql_parameter_limit = 2100 - 1
	row_limit = sql_parameter_limit//num_of_columns
	
	# If dataset is small enough, process the simple way. Otherwise break it into chunks.
	if len(api_data) <= row_limit:
		rows = ', '.join([row_template for row in api_data])
		query = base_query + rows
		args = [value for row in api_data for value in row.values()]
		system.db.runPrepUpdate(query, args, database='My_Database')
	else:
		
		for index in range(0, len(api_data), row_limit):
			# Avoid overindexing.
			end_index = index + row_limit
			if end_index > len(api_data):
				end_index = len(api_data)
			
			# Create rows string in format (?, ?, ..., ?, ?), (?, ?, ..., ?, ?), ..., (?, ?, ..., ?, ?), (?, ?, ..., ?, ?)
			# This is concatenated on the end of our query.
			rows = ', '.join([row_template for row in api_data[index:end_index]])
			query = base_query + rows
			
			# Create arguments array
			args = [value for row in api_data[index:end_index] for value in row.values()]
			
			# Run Insert Query
			system.db.runPrepUpdate(query, args, database='My_Database')
	
	# The daily and shift summary tables do not merge on the id column.  For now, I just created separate stored procedures to handle them.
	if 'DailySummary' in table_name:
		sp = 'API_DailySummary_Merge_sp'
	elif 'Shift_Summary' in table_name:
		sp = 'API_ShiftSummary_Merge_sp'
	else:
		sp = 'API_ID_Merge_sp'
	
	# Run stored procedure to merge data from the temp table into the actual table.
	system.db.runUpdateQuery("EXEC	[dbo].[{}] @target_table = N'{}'".format(sp, table_name), database='My_Database')

Stored Procedure:

ALTER PROCEDURE [dbo].[API_ID_Merge_sp] 
	-- Add the parameters for the stored procedure here
	@target_table NVARCHAR(128) = ''
AS
BEGIN
	-- SET NOCOUNT ON added to prevent extra result sets from
	-- interfering with SELECT statements.
	SET NOCOUNT ON;

    -- Insert statements for procedure here
	DECLARE @sql NVARCHAR(MAX);
	DECLARE @columns NVARCHAR(MAX);
	DECLARE @columndefinitions NVARCHAR(MAX);
	DECLARE @updateColumns NVARCHAR(MAX);
	DECLARE @source_table NVARCHAR(128) = 'Temp_' + @target_table;
	
	-- Generate the column list dynamically
	SELECT  @columns = STRING_AGG('[' + c.name + ']', ', '),
			@columndefinitions = STRING_AGG('[' + c.name + ']' + ' ' + t.name, ', '),
			@updateColumns = STRING_AGG('target.' + '[' + c.name + ']' + ' = source.' + '[' + c.name + ']', ', ')
	FROM sys.columns c
	JOIN sys.types t ON c.user_type_id = t.user_type_id
	WHERE c.object_id = OBJECT_ID(@target_table) AND c.is_computed = 0


	-- Construct the dynamic SQL for the MERGE statement
	SET @sql = N'
	DECLARE @ProcessedRows TABLE (performed_action NVARCHAR(10), id bigint);

	MERGE ' + @target_table + ' AS target
	USING ' + @source_table + ' AS source
	ON target.id = source.id
	WHEN MATCHED THEN
		UPDATE SET ' + @updateColumns + '
	WHEN NOT MATCHED BY TARGET THEN
		INSERT (' + @columns + ')
		VALUES (' + @columns + ')
	OUTPUT $action, inserted.id
	INTO @ProcessedRows;

	DELETE FROM ' + @source_table + ' 
	WHERE id IN (SELECT id FROM @ProcessedRows);
	'
	-- Execute the dynamic SQL
	EXEC sp_executesql @sql
END

[MMaynard]

200 records to SQL Server is NOT a heavy load by any means.

I would script this and either do a gateway tag change script to trigger it, or schedule it with the API call and then use the bulk insert code from this thread utilizing the prepInsertQMarks specifically.

I gather around 4000± tag values every 15 minutes and write the values to rows in a database that are 4 columns wide, the entire thing is done in around 8 seconds, mind you this is also gathering those data points from 15 different gateways.

Tag History to CSV and Back how do i fix ignition crash?

Insert multiple rows at once. But, as @pturmel said, it may only support a few thousand question marks. I modified your script, shown below. Start with 500 rows at a time, then raise or lower depending on how it runs. def runAction(self, event): import csv def prepInsertQmarks(numOfGroups, groupSize): ''' Generate a string of grouped question marks for prepUpdate inserts example: Generate three groups of four question marks: prepInsertQmarks(3, 4) returns (?,?,?,?),(?,?,…