Merge branch 'release/4.5.1'

Author: nicholasbarlow

docs/4.4.5-release-notes.md

@@ -0,0 +1,2 @@
+### Fix
+- Fixed issue where the buffer would not wait after a flush

docs/4.4.6-release-notes.md

@@ -0,0 +1,5 @@
+### Features
+- Add _partnerEvents.json when creating _metadata.json in Open Mirroring
+
+### Fixes
+- Fix issue where rows are not updated in Open Mirroring when entities are updated

docs/4.5.1-release-notes.md

@@ -0,0 +1,5 @@
+### Features
+- Add _partnerEvents.json when creating _metadata.json in Open Mirroring
+
+### Fixes
+- Fix issue where rows are not updated in Open Mirroring when entities are updated

src/Connector.DataLake.Common/Connector/DataLakeConnector.cs

@@ -359,7 +359,9 @@ IF EXISTS (
                         BEGIN
                             UPDATE [{tableName}]
                             SET
-                                [{DataLakeConstants.ChangeTypeKey}] = {syncItem.ChangeType.ToString()}
+                                [{DataLakeConstants.ChangeTypeKey}] = @ChangeType,
+                                [Timestamp] = @Timestamp,
+                                [Epoch] = @Epoch
                             WHERE {DataLakeConstants.IdKey} = @{DataLakeConstants.IdKey};
                         END
                         """;
@@ -368,6 +370,9 @@ UPDATE [{tableName}]
                     CommandType = CommandType.Text
                 };
                 command.Parameters.Add(new SqlParameter($"@{DataLakeConstants.IdKey}", syncItem.EntityId));
+                command.Parameters.Add(new SqlParameter($"@ChangeType", syncItem.ChangeType.ToString()));
+                command.Parameters.Add(new SqlParameter($"@Timestamp", syncItem.Data["Timestamp"]));
+                command.Parameters.Add(new SqlParameter($"@Epoch", syncItem.Data["Epoch"]));
 
                 var rowsAffected = await command.ExecuteNonQueryAsync();
                 if (rowsAffected != 1)

src/Connector.DataLake.Common/Connector/DataLakeExportEntitiesJobBase.cs

@@ -7,8 +7,6 @@
 using System.Threading.Tasks;
 using System.Transactions;
 
-using Apache.Arrow;
-
 using Azure.Storage.Files.DataLake;
 
 using CluedIn.Connector.DataLake.Common.Connector.SqlDataWriter;
@@ -20,8 +18,6 @@
 using Microsoft.Data.SqlClient;
 using Microsoft.Extensions.Logging;
 
-using Parquet;
-
 namespace CluedIn.Connector.DataLake.Common.Connector;
 
 internal abstract class DataLakeExportEntitiesJobBase : DataLakeJobBase
@@ -117,7 +113,7 @@ public override async Task DoRunAsync(ExecutionContext context, IDataLakeJobArgs
 
         var subDirectory = await GetSubDirectory(configuration, streamId);
         var directoryClient = await _dataLakeClient.EnsureDataLakeDirectoryExist(configuration, subDirectory);
-        await InitializeDirectoryAsync(configuration, streamId, directoryClient);
+        await InitializeDirectoryAsync(context, connection, configuration, exportJobData, directoryClient);
         var startExportTime = _dateTimeOffsetProvider.GetCurrentUtcTime();
         var exportHistory = new ExportHistory(
             streamId,
@@ -152,10 +148,6 @@ public override async Task DoRunAsync(ExecutionContext context, IDataLakeJobArgs
             .Select(reader.GetName)
             .ToList();
 
-        var fieldNamesToUse = configuration.IsDeltaMode
-            ? fieldNames
-            : fieldNames.Where(fieldName => fieldName != DataLakeConstants.ChangeTypeKey).ToList();
-
         var temporaryOutputFileName = outputFileName + TemporaryFileSuffix;
         using var loggingScope = context.Log.BeginScope(new Dictionary<string, object>
         {
@@ -214,10 +206,11 @@ public override async Task DoRunAsync(ExecutionContext context, IDataLakeJobArgs
 
         async Task<long> writeFileContentsAsync()
         {
+            var fieldNamesToUse = await GetFieldNamesAsync(context, exportJobData, configuration, fieldNames);
             var sqlDataWriter = GetSqlDataWriter(outputFormat);
             await using var outputStream = await temporaryFileClient.OpenWriteAsync(configuration.IsOverwriteEnabled);
             using var bufferedStream = new DataLakeBufferedWriteStream(outputStream);
-            return await sqlDataWriter?.WriteAsync(context, configuration, bufferedStream, fieldNamesToUse, reader);
+            return await sqlDataWriter?.WriteAsync(context, configuration, bufferedStream, fieldNamesToUse, IsInitialExport, reader);
         }
 
         async Task setFilePropertiesAsync()
@@ -268,8 +261,22 @@ async Task deleteFileIfExistsAsync(string file)
             await targetFileClient.DeleteIfExistsAsync();
         }
     }
+
+    private protected virtual async Task<List<string>> GetFieldNamesAsync(
+        ExecutionContext context,
+        ExportJobData exportJobData,
+        IDataLakeJobData configuration,
+        List<string> fieldNames)
+    {
+        return configuration.IsDeltaMode
+                ? fieldNames
+                : fieldNames.Where(fieldName => fieldName != DataLakeConstants.ChangeTypeKey).ToList();
+    }
+
     private protected ExportHistory LastExport { get; private set; }
 
+    protected virtual bool IsInitialExport => LastExport == null;
+
     private protected virtual bool GetIsEmptyFileAllowed(ExportJobData exportJobData) => true;
 
     private protected virtual Task PostExportAsync(ExecutionContext context, ExportJobData exportJobData)
@@ -282,7 +289,7 @@ protected virtual Task<string> GetSubDirectory(IDataLakeJobData configuration, G
         return Task.FromResult(string.Empty);
     }
 
-    protected virtual Task InitializeDirectoryAsync(IDataLakeJobData configuration, Guid streamId, DataLakeDirectoryClient client)
+    private protected virtual Task InitializeDirectoryAsync(ExecutionContext context, SqlConnection connection, IDataLakeJobData configuration, ExportJobData exportJobData, DataLakeDirectoryClient client)
     {
         return Task.CompletedTask;
     }

src/Connector.DataLake.Common/Connector/SqlDataWriter/CsvSqlDataWriter.cs

@@ -18,6 +18,7 @@ public override async Task<long> WriteOutputAsync(
         IDataLakeJobData configuration,
         Stream outputStream,
         ICollection<string> fieldNames,
+        bool isInitialExport,
         SqlDataReader reader)
     {
         context.Log.LogInformation("Begin writing output.");
@@ -27,14 +28,20 @@ public override async Task<long> WriteOutputAsync(
         await using var csv = new CsvWriter(writer, csvConfig);
         foreach (var fieldName in fieldNames)
         {
-            var fieldNameToUse = configuration.ShouldEscapeVocabularyKeys ? EscapeVocabularyKey(fieldName) : fieldName;
+            var fieldNameToUse = GetFieldName(configuration, fieldName);
             csv.WriteField(fieldNameToUse);
         }
+
         await csv.NextRecordAsync();
 
         var totalProcessed = 0L;
         while (await reader.ReadAsync())
         {
+            if (ShouldSkip(configuration, isInitialExport, reader))
+            {
+                continue;
+            }
+
             var fieldValues = fieldNames.Select(name => GetValue(name, reader, configuration));
             foreach (var field in fieldValues)
             {
@@ -52,4 +59,9 @@ public override async Task<long> WriteOutputAsync(
 
         return totalProcessed;
     }
+
+    protected virtual string GetFieldName(IDataLakeJobData configuration, string fieldName)
+    {
+        return configuration.ShouldEscapeVocabularyKeys ? EscapeVocabularyKey(fieldName) : fieldName;
+    }
 }

src/Connector.DataLake.Common/Connector/SqlDataWriter/ISqlDataWriter.cs

@@ -13,5 +13,6 @@ Task<long> WriteAsync(
         IDataLakeJobData configuration,
         Stream outputStream,
         ICollection<string> fieldNames,
+        bool isInitialExport,
         SqlDataReader reader);
 }

src/Connector.DataLake.Common/Connector/SqlDataWriter/JsonSqlDataWriter.cs

@@ -16,6 +16,7 @@ public override async Task<long> WriteOutputAsync(
         IDataLakeJobData configuration,
         Stream outputStream,
         ICollection<string> fieldNames,
+        bool isInitialExport,
         SqlDataReader reader)
     {
         await using var stringWriter = new StreamWriter(outputStream);
@@ -26,10 +27,15 @@ public override async Task<long> WriteOutputAsync(
         await writer.WriteStartArrayAsync();
         while (await reader.ReadAsync())
         {
+            if (ShouldSkip(configuration, isInitialExport, reader))
+            {
+                continue;
+            }
+
             await writer.WriteStartObjectAsync();
 
             foreach(var field in fieldNames)
-            { 
+            {
                 await writer.WritePropertyNameAsync(field);
                 var value = GetValue(field, reader, configuration);
                 if (value is JArray jArray)

src/Connector.DataLake.Common/Connector/SqlDataWriter/ParquetSqlDataWriter.cs

@@ -34,6 +34,7 @@ public override async Task<long> WriteOutputAsync(
         IDataLakeJobData configuration,
         Stream outputStream,
         ICollection<string> fieldNames,
+        bool isInitialExport,
         SqlDataReader reader)
     {
         var fields = new List<Field>();
@@ -51,6 +52,11 @@ public override async Task<long> WriteOutputAsync(
 
         while (await reader.ReadAsync())
         {
+            if (ShouldSkip(configuration, isInitialExport, reader))
+            {
+                continue;
+            }
+
             var fieldValues = fieldNames.Select(key => {
                 var value = GetValue(key, reader, configuration);
                 return value;

src/Connector.DataLake.Common/Connector/SqlDataWriter/SqlDataWriterBase.cs

@@ -1,10 +1,12 @@
-using System;
+using System;
 using System.Collections.Generic;
 using System.Data;
 using System.IO;
 using System.Text.RegularExpressions;
 using System.Threading.Tasks;
 using CluedIn.Core;
+using CluedIn.Core.Data.Parts;
+
 using Microsoft.Data.SqlClient;
 using Microsoft.Extensions.Logging;
 
@@ -15,6 +17,11 @@ internal abstract class SqlDataWriterBase : ISqlDataWriter
     protected const int LoggingThreshold = 1000;
     private static readonly Regex NonAlphaNumericRegex = new("[^a-zA-Z0-9_]");
     protected virtual object GetValue(string key, SqlDataReader reader, IDataLakeJobData configuration)
+    {
+        return GetValueInternal(key, reader);
+    }
+
+    private static object GetValueInternal(string key, SqlDataReader reader)
     {
         var value = reader.GetValue(key);
 
@@ -31,11 +38,12 @@ public async Task<long> WriteAsync(
         IDataLakeJobData configuration,
         Stream outputStream,
         ICollection<string> fieldNames,
+        bool isInitialExport,
         SqlDataReader reader)
     {
         context.Log.LogInformation("Begin writing output.");
 
-        var totalProcessed = await WriteOutputAsync(context, configuration, outputStream, fieldNames, reader);
+        var totalProcessed = await WriteOutputAsync(context, configuration, outputStream, fieldNames, isInitialExport, reader);
         context.Log.LogInformation("End writing output. Total processed: {TotalProcessed}.", totalProcessed);
         return totalProcessed;
     }
@@ -45,10 +53,18 @@ public abstract Task<long> WriteOutputAsync(
         IDataLakeJobData configuration,
         Stream outputStream,
         ICollection<string> fieldNames,
+        bool isInitialExport,
         SqlDataReader reader);
 
     protected string EscapeVocabularyKey(string fieldName)
     {
         return NonAlphaNumericRegex.Replace(fieldName, "_");
     }
+
+    protected virtual bool ShouldSkip(IDataLakeJobData configuration, bool isInitialExport, SqlDataReader reader)
+    {
+        return configuration.IsDeltaMode
+            && isInitialExport
+            && GetValueInternal(DataLakeConstants.ChangeTypeKey, reader).Equals(VersionChangeType.Removed.ToString());
+    }
 }