Refactored outputting log messages from DML nodes

MarkMpn.Sql4Cds.Engine/Ado/Sql4CdsDataReader.cs

@@ -8,6 +8,7 @@
 using System.Linq;
 using System.Linq.Expressions;
 using System.Text;
+using System.Xml;
 using MarkMpn.Sql4Cds.Engine.ExecutionPlan;
 using Microsoft.SqlServer.TransactSql.ScriptDom;
 using Microsoft.Xrm.Sdk;
@@ -62,13 +63,15 @@ public Sql4CdsDataReader(Sql4CdsCommand command, IQueryExecutionOptions options,
 
         private bool Execute(Dictionary<string, DataTypeReference> parameterTypes, Dictionary<string, object> parameterValues)
         {
-            var context = new NodeExecutionContext(_connection.DataSources, _options, parameterTypes, parameterValues);
+            IRootExecutionPlanNode logNode = null;
+            var context = new NodeExecutionContext(_connection.DataSources, _options, parameterTypes, parameterValues, msg => _connection.OnInfoMessage(logNode, msg));
 
             try
             {
                 while (_instructionPointer < _command.Plan.Length && !_options.CancellationToken.IsCancellationRequested)
                 {
                     var node = _command.Plan[_instructionPointer];
+                    logNode = node;
 
                     if (node is IDataReaderExecutionPlanNode dataSetNode)
                     {
@@ -90,10 +93,7 @@ private bool Execute(Dictionary<string, DataTypeReference> parameterTypes, Dicti
                     else if (node is IDmlQueryExecutionPlanNode dmlNode)
                     {
                         dmlNode = (IDmlQueryExecutionPlanNode)dmlNode.Clone();
-                        var msg = dmlNode.Execute(context, out var recordsAffected);
-
-                        if (!String.IsNullOrEmpty(msg))
-                            _connection.OnInfoMessage(dmlNode, msg);
+                        dmlNode.Execute(context, out var recordsAffected);
 
                         _command.OnStatementCompleted(dmlNode, recordsAffected);
 

MarkMpn.Sql4Cds.Engine/ExecutionPlan/AssignVariablesNode.cs

@@ -40,7 +40,7 @@ class AssignVariablesNode : BaseDmlNode
         [Browsable(false)]
         public override bool ContinueOnError { get; set; }
 
-        public override string Execute(NodeExecutionContext context, out int recordsAffected)
+        public override void Execute(NodeExecutionContext context, out int recordsAffected)
         {
             _executionCount++;
 
@@ -63,7 +63,6 @@ public override string Execute(NodeExecutionContext context, out int recordsAffe
             }
 
             recordsAffected = -1;
-            return null;
         }
 
         /// <summary>

MarkMpn.Sql4Cds.Engine/ExecutionPlan/BaseDmlNode.cs

@@ -129,8 +129,7 @@ public void Dispose()
         /// </summary>
         /// <param name="context">The context in which the node is being executed</param>
         /// <param name="recordsAffected">The number of records that were affected by the query</param>
-        /// <returns>A log message to display</returns>
-        public abstract string Execute(NodeExecutionContext context, out int recordsAffected);
+        public abstract void Execute(NodeExecutionContext context, out int recordsAffected);
 
         /// <summary>
         /// Indicates if some errors returned by the server can be silently ignored
@@ -539,8 +538,8 @@ protected class OperationNames
         /// <param name="meta">The metadata of the entity that will be affected</param>
         /// <param name="requestGenerator">A function to generate a DML request from a data source entity</param>
         /// <param name="operationNames">The constant strings to use in log messages</param>
-        /// <returns>The final log message</returns>
-        protected string ExecuteDmlOperation(DataSource dataSource, IQueryExecutionOptions options, List<Entity> entities, EntityMetadata meta, Func<Entity,OrganizationRequest> requestGenerator, OperationNames operationNames, out int recordsAffected, IDictionary<string, object> parameterValues, Action<OrganizationResponse> responseHandler = null)
+        /// <param name="log">A callback function to be executed when a log message is generated</param>
+        protected void ExecuteDmlOperation(DataSource dataSource, IQueryExecutionOptions options, List<Entity> entities, EntityMetadata meta, Func<Entity,OrganizationRequest> requestGenerator, OperationNames operationNames, NodeExecutionContext context, out int recordsAffected, Action<OrganizationResponse> responseHandler = null)
         {
             var inProgressCount = 0;
             var count = 0;
@@ -618,13 +617,15 @@ protected string ExecuteDmlOperation(DataSource dataSource, IQueryExecutionOptio
                                 }
                                 catch (FaultException<OrganizationServiceFault> ex)
                                 {
-                                    if (FilterErrors(ex.Detail))
+                                    if (FilterErrors(context, request, ex.Detail))
                                     {
                                         if (ContinueOnError)
                                             fault = fault ?? ex.Detail;
                                         else
                                             throw;
                                     }
+
+                                    Interlocked.Increment(ref errorCount);
                                 }
                             }
                             else
@@ -650,7 +651,7 @@ protected string ExecuteDmlOperation(DataSource dataSource, IQueryExecutionOptio
 
                                 if (threadLocalState.EMR.Requests.Count == BatchSize)
                                 {
-                                    ProcessBatch(threadLocalState.EMR, threadCount, ref count, ref inProgressCount, ref errorCount, entities, operationNames, meta, options, dataSource, threadLocalState.Service, responseHandler, ref fault);
+                                    ProcessBatch(threadLocalState.EMR, threadCount, ref count, ref inProgressCount, ref errorCount, entities, operationNames, meta, options, dataSource, threadLocalState.Service, context, responseHandler, ref fault);
 
                                     threadLocalState = new { threadLocalState.Service, EMR = default(ExecuteMultipleRequest) };
                                 }
@@ -661,7 +662,7 @@ protected string ExecuteDmlOperation(DataSource dataSource, IQueryExecutionOptio
                         (threadLocalState) =>
                         {
                             if (threadLocalState.EMR != null)
-                                ProcessBatch(threadLocalState.EMR, threadCount, ref count, ref inProgressCount, ref errorCount, entities, operationNames, meta, options, dataSource, threadLocalState.Service, responseHandler, ref fault);
+                                ProcessBatch(threadLocalState.EMR, threadCount, ref count, ref inProgressCount, ref errorCount, entities, operationNames, meta, options, dataSource, threadLocalState.Service, context, responseHandler, ref fault);
 
                             Interlocked.Decrement(ref threadCount);
 
@@ -692,8 +693,8 @@ protected string ExecuteDmlOperation(DataSource dataSource, IQueryExecutionOptio
             }
 
             recordsAffected = count;
-            parameterValues["@@ROWCOUNT"] = (SqlInt32)count;
-            return $"{count:N0} {GetDisplayName(count, meta)} {operationNames.CompletedLowercase}";
+            context.ParameterValues["@@ROWCOUNT"] = (SqlInt32)count;
+            context.Log($"{count:N0} {GetDisplayName(count, meta)} {operationNames.CompletedLowercase}");
         }
 
         protected class BulkApiErrorDetail
@@ -703,7 +704,7 @@ protected class BulkApiErrorDetail
             public int StatusCode { get; set; }
         }
 
-        private void ProcessBatch(ExecuteMultipleRequest req, int threadCount, ref int count, ref int inProgressCount, ref int errorCount, List<Entity> entities, OperationNames operationNames, EntityMetadata meta, IQueryExecutionOptions options, DataSource dataSource, IOrganizationService org, Action<OrganizationResponse> responseHandler, ref OrganizationServiceFault fault)
+        private void ProcessBatch(ExecuteMultipleRequest req, int threadCount, ref int count, ref int inProgressCount, ref int errorCount, List<Entity> entities, OperationNames operationNames, EntityMetadata meta, IQueryExecutionOptions options, DataSource dataSource, IOrganizationService org, NodeExecutionContext context, Action<OrganizationResponse> responseHandler, ref OrganizationServiceFault fault)
         {
             var newCount = Interlocked.Add(ref inProgressCount, req.Requests.Count);
             var progress = (double)newCount / entities.Count;
@@ -727,7 +728,7 @@ private void ProcessBatch(ExecuteMultipleRequest req, int threadCount, ref int c
             Interlocked.Add(ref count, req.Requests.Count - errorResponses.Count);
             Interlocked.Add(ref errorCount, errorResponses.Count);
 
-            var error = errorResponses.FirstOrDefault(item => FilterErrors(item.Fault));
+            var error = errorResponses.FirstOrDefault(item => FilterErrors(context, req.Requests[item.RequestIndex], item.Fault));
 
             if (error != null)
             {
@@ -738,7 +739,7 @@ private void ProcessBatch(ExecuteMultipleRequest req, int threadCount, ref int c
             }
         }
 
-        protected virtual bool FilterErrors(OrganizationServiceFault fault)
+        protected virtual bool FilterErrors(NodeExecutionContext context, OrganizationRequest request, OrganizationServiceFault fault)
         {
             return true;
         }

MarkMpn.Sql4Cds.Engine/ExecutionPlan/BulkDeleteJobNode.cs

@@ -43,7 +43,7 @@ public override void AddRequiredColumns(NodeCompilationContext context, IList<st
         {
         }
 
-        public string Execute(NodeExecutionContext context, out int recordsAffected)
+        public void Execute(NodeExecutionContext context, out int recordsAffected)
         {
             _executionCount++;
 
@@ -73,7 +73,7 @@ public string Execute(NodeExecutionContext context, out int recordsAffected)
 
                     recordsAffected = 1;
                     context.ParameterValues["@@IDENTITY"] = new SqlEntityReference(DataSource, "asyncoperation", resp.JobId);
-                    return $"Bulk delete job started";
+                    context.Log("Bulk delete job started");
                 }
             }
             catch (QueryExecutionException ex)

MarkMpn.Sql4Cds.Engine/ExecutionPlan/DeclareVariablesNode.cs

@@ -43,7 +43,7 @@ public override void AddRequiredColumns(NodeCompilationContext context, IList<st
         {
         }
 
-        public string Execute(NodeExecutionContext context, out int recordsAffected)
+        public void Execute(NodeExecutionContext context, out int recordsAffected)
         {
             _executionCount++;
 
@@ -57,7 +57,6 @@ public string Execute(NodeExecutionContext context, out int recordsAffected)
             }
 
             recordsAffected = -1;
-            return null;
         }
 
         public IRootExecutionPlanNodeInternal[] FoldQuery(NodeCompilationContext context, IList<OptimizerHint> hints)

MarkMpn.Sql4Cds.Engine/ExecutionPlan/DeleteNode.cs

@@ -106,7 +106,7 @@ protected override void RenameSourceColumns(IDictionary<string, string> columnRe
                 SecondaryIdSource = secondaryIdSourceRenamed;
         }
 
-        public override string Execute(NodeExecutionContext context, out int recordsAffected)
+        public override void Execute(NodeExecutionContext context, out int recordsAffected)
         {
             _executionCount++;
 
@@ -172,7 +172,7 @@ public override string Execute(NodeExecutionContext context, out int recordsAffe
 
                 using (_timer.Run())
                 {
-                    return ExecuteDmlOperation(
+                    ExecuteDmlOperation(
                         dataSource,
                         context.Options,
                         entities,
@@ -184,8 +184,8 @@ public override string Execute(NodeExecutionContext context, out int recordsAffe
                             InProgressLowercase = "deleting",
                             CompletedLowercase = "deleted"
                         },
-                        out recordsAffected,
-                        context.ParameterValues);
+                        context,
+                        out recordsAffected);
                 }
             }
             catch (QueryExecutionException ex)
@@ -250,7 +250,7 @@ private OrganizationRequest CreateDeleteRequest(EntityMetadata meta, Entity enti
             return req;
         }
 
-        protected override bool FilterErrors(OrganizationServiceFault fault)
+        protected override bool FilterErrors(NodeExecutionContext context, OrganizationRequest request, OrganizationServiceFault fault)
         {
             // Ignore errors trying to delete records that don't exist - record may have been deleted by another
             // process in parallel.

MarkMpn.Sql4Cds.Engine/ExecutionPlan/ExecuteAsNode.cs

@@ -51,7 +51,7 @@ public override void AddRequiredColumns(NodeCompilationContext context, IList<st
             Source.AddRequiredColumns(context, requiredColumns);
         }
 
-        public override string Execute(NodeExecutionContext context, out int recordsAffected)
+        public override void Execute(NodeExecutionContext context, out int recordsAffected)
         {
             _executionCount++;
 
@@ -91,7 +91,7 @@ public override string Execute(NodeExecutionContext context, out int recordsAffe
                         throw new QueryExecutionException("Unexpected organization service type");
 
                     recordsAffected = -1;
-                    return $"Impersonated user {userId}";
+                    context.Log($"Impersonated user {userId}");
                 }
             }
             catch (QueryExecutionException ex)

MarkMpn.Sql4Cds.Engine/ExecutionPlan/ExecuteMessageNode.cs

@@ -693,10 +693,10 @@ public static ExecuteMessageNode FromMessage(ExecutableProcedureReference sproc,
             return node;
         }
 
-        public string Execute(NodeExecutionContext context, out int recordsAffected)
+        public void Execute(NodeExecutionContext context, out int recordsAffected)
         {
             recordsAffected = Execute(context).Count();
-            return "Executed " + MessageName;
+            context.Log("Executed " + MessageName);
         }
 
         IRootExecutionPlanNodeInternal[] IRootExecutionPlanNodeInternal.FoldQuery(NodeCompilationContext context, IList<OptimizerHint> hints)

MarkMpn.Sql4Cds.Engine/ExecutionPlan/FetchXmlScan.cs

@@ -1185,7 +1185,7 @@ private void AddSchemaAttribute(DataSource dataSource, ColumnList schema, Dictio
 
             if (attrMetadata is LookupAttributeMetadata lookup)
             {
-                AddSchemaAttribute(schema, aliases, fullName + "name", attrMetadata.LogicalName + "name", DataTypeHelpers.NVarChar(lookup.Targets == null || lookup.Targets.Length == 0 ? 100 : lookup.Targets.Select(e => ((StringAttributeMetadata)dataSource.Metadata[e].Attributes.SingleOrDefault(a => a.LogicalName == dataSource.Metadata[e].PrimaryNameAttribute))?.MaxLength ?? 100).Max(), dataSource.DefaultCollation, CollationLabel.Implicit), notNull);
+                AddSchemaAttribute(schema, aliases, fullName + "name", attrMetadata.LogicalName + "name", DataTypeHelpers.NVarChar(lookup.Targets == null || lookup.Targets.Length == 0 ? 100 : lookup.Targets.Select(e => (dataSource.Metadata[e].Attributes.SingleOrDefault(a => a.LogicalName == dataSource.Metadata[e].PrimaryNameAttribute) as StringAttributeMetadata)?.MaxLength ?? 100).Max(), dataSource.DefaultCollation, CollationLabel.Implicit), notNull);
 
                 if (lookup.Targets?.Length != 1 && lookup.AttributeType != AttributeTypeCode.PartyList)
                     AddSchemaAttribute(schema, aliases, fullName + "type", attrMetadata.LogicalName + "type", DataTypeHelpers.NVarChar(MetadataExtensions.EntityLogicalNameMaxLength, dataSource.DefaultCollation, CollationLabel.Implicit), notNull);

MarkMpn.Sql4Cds.Engine/ExecutionPlan/FilterNode.cs

@@ -1511,7 +1511,7 @@ protected bool TranslateMetadataCriteria(NodeCompilationContext context, Boolean
             relationshipFilter = null;
 
             var expressionCompilationContext = new ExpressionCompilationContext(context, null, null);
-            var expressionExecutionContext = new ExpressionExecutionContext(new NodeExecutionContext(context.DataSources, context.Options, context.ParameterTypes, null));
+            var expressionExecutionContext = new ExpressionExecutionContext(new NodeExecutionContext(context.DataSources, context.Options, context.ParameterTypes, null, null));
 
             if (criteria is BooleanBinaryExpression binary)
             {

MarkMpn.Sql4Cds.Engine/ExecutionPlan/IDmlQueryExecutionPlanNode.cs

@@ -18,7 +18,6 @@ internal interface IDmlQueryExecutionPlanNode : IRootExecutionPlanNodeInternal
         /// </summary>
         /// <param name="context">The context in which the node is being executed</param>
         /// <param name="recordsAffected">The number of records that were affected by the query</param>
-        /// <returns>A status message for the results of the query</returns>
-        string Execute(NodeExecutionContext context, out int recordsAffected);
+        void Execute(NodeExecutionContext context, out int recordsAffected);
     }
 }

MarkMpn.Sql4Cds.Engine/ExecutionPlan/InsertNode.cs

@@ -65,6 +65,7 @@ public override void AddRequiredColumns(NodeCompilationContext context, IList<st
         }
 
         public override string Execute(NodeExecutionContext context, out int recordsAffected)
+        public override void Execute(NodeExecutionContext context, out int recordsAffected)
         {
             _executionCount++;
 
@@ -101,7 +102,7 @@ public override string Execute(NodeExecutionContext context, out int recordsAffe
 
                 using (_timer.Run())
                 {
-                    return ExecuteDmlOperation(
+                    ExecuteDmlOperation(
                         dataSource,
                         context.Options,
                         entities,
@@ -113,8 +114,8 @@ public override string Execute(NodeExecutionContext context, out int recordsAffe
                             InProgressLowercase = "inserting",
                             CompletedLowercase = "inserted"
                         },
+                        context,
                         out recordsAffected,
-                        context.ParameterValues,
                         LogicalName == "listmember" || meta.IsIntersect == true ? null : (Action<OrganizationResponse>) ((r) => SetIdentity(r, context.ParameterValues))
                         );
                 }

MarkMpn.Sql4Cds.Engine/ExecutionPlan/NestedLoopNode.cs

@@ -81,7 +81,7 @@ protected override IEnumerable<Entity> ExecuteInternal(NodeExecutionContext cont
 
                 var hasRight = false;
 
-                foreach (var right in RightSource.Execute(new NodeExecutionContext(context.DataSources, context.Options, innerParameterTypes, innerParameters)))
+                foreach (var right in RightSource.Execute(new NodeExecutionContext(context.DataSources, context.Options, innerParameterTypes, innerParameters, context.Log)))
                 {
                     if (rightSchema == null)
                     {

MarkMpn.Sql4Cds.Engine/ExecutionPlan/PartitionedAggregateNode.cs

@@ -202,7 +202,7 @@ protected override IEnumerable<Entity> ExecuteInternal(NodeExecutionContext cont
                                 partitionParameterValues[kvp.Key] = kvp.Value;
                         }
 
-                        var partitionContext = new NodeExecutionContext(context.DataSources, context.Options, context.ParameterTypes, partitionParameterValues);
+                        var partitionContext = new NodeExecutionContext(context.DataSources, context.Options, context.ParameterTypes, partitionParameterValues, context.Log);
 
                         return new { Context = partitionContext, Fetch = fetch };
                     },

MarkMpn.Sql4Cds.Engine/ExecutionPlan/PrintNode.cs

@@ -47,7 +47,7 @@ public object Clone()
             };
         }
 
-        public string Execute(NodeExecutionContext context, out int recordsAffected)
+        public void Execute(NodeExecutionContext context, out int recordsAffected)
         {
             _executionCount++;
             recordsAffected = -1;
@@ -56,10 +56,8 @@ public string Execute(NodeExecutionContext context, out int recordsAffected)
             {
                 var value = (SqlString)_expression(new ExpressionExecutionContext(context));
 
-                if (value.IsNull)
-                    return null;
-
-                return value.Value;
+                if (!value.IsNull)
+                    context.Log(value.Value);
             }
         }