Merge branch 'release/4.4.1'

Author: hopengfoong

docs/4.4.1-release-notes.md

@@ -0,0 +1,4 @@
+### Fixes
+- Prevent unnecessary warning messages when using sync mode with stream cache disabled
+- Add option to enable or disable escaping of non a-z,A-Z,0-9,_ characters
+- Add option to enable or disable writing of guids as strings

src/Connector.AzureDataLake/AzureDataLakeConnectorJobData.cs

@@ -19,10 +19,6 @@ public AzureDataLakeConnectorJobData(
     public string DirectoryName => GetConfigurationValue(AzureDataLakeConstants.DirectoryName) as string;
     public string FileSystemName => GetConfigurationValue(AzureDataLakeConstants.FileSystemName) as string;
 
-    public override bool ShouldEscapeVocabularyKeys => IsStreamCacheEnabled && DataLakeConstants.OutputFormats.Parquet.Equals(OutputFormat, StringComparison.OrdinalIgnoreCase);
-
-    public override bool ShouldWriteGuidAsString => ShouldEscapeVocabularyKeys;
-
     protected override void AddToHashCode(HashCode hash)
     {
         hash.Add(AccountName);

src/Connector.AzureDataLake/AzureDataLakeConstants.cs

@@ -66,6 +66,52 @@ private static AuthMethods GetAzureDataLakeAuthMethods(ApplicationContext applic
         };
 
         controls.AddRange(GetAuthMethods(applicationContext));
+        controls.Add(
+            new()
+            {
+                Name = ShouldEscapeVocabularyKeys,
+                DisplayName = "Replace Non-Alphanumeric Characters in Column Names",
+                Type = "checkbox",
+                IsRequired = false,
+                Help = """
+                    Replaces characters in the column names that are not in this list ('a-z', 'A-Z', '0-9' and '_') with the character '_'.
+                    Enable this if you plan to access the output file in Microsoft Purview.
+                    """,
+                DisplayDependencies = new[]
+                {
+                    new ControlDisplayDependency
+                    {
+                        Name = OutputFormat,
+                        Operator = ControlDependencyOperator.Equals,
+                        Value = OutputFormats.Parquet.ToLowerInvariant(),
+                        UnfulfilledAction = ControlDependencyUnfulfilledAction.Hidden,
+                    },
+                },
+            }
+        );
+        controls.Add(
+            new()
+            {
+                Name = ShouldWriteGuidAsString,
+                DisplayName = "Write Guid as string",
+                Type = "checkbox",
+                IsRequired = false,
+                Help = """
+                    Write Guid values as string instead of byte array.
+                    Enable this if you plan to access the output file in Microsoft Purview.
+                    """,
+                DisplayDependencies = new[]
+                {
+                    new ControlDisplayDependency
+                    {
+                        Name = OutputFormat,
+                        Operator = ControlDependencyOperator.Equals,
+                        Value = OutputFormats.Parquet.ToLowerInvariant(),
+                        UnfulfilledAction = ControlDependencyUnfulfilledAction.Hidden,
+                    },
+                },
+            }
+        );
 
         return new AuthMethods
         {

src/Connector.DataLake.Common/DataLakeScheduler.cs

@@ -48,15 +48,15 @@ async Task addStreamExportJobs(ExecutionContext executionContext, ProviderDefini
             if (stream.Mode != StreamMode.Sync)
             {
                 _logger.LogDebug("Stream {StreamId} is not in {Mode} mode. Removing it from scheduler '{SchedulerName}'.", stream.Id, StreamMode.Sync, _schedulerName);
-                removeExportJob(jobKey);
+                removeExportJob(jobKey, warnIfNotRemoved: false);
                 return;
             }
 
             var jobSchedule = await _dataLakeJobDataFactory.GetScheduleAsync(executionContext, stream);
             if (IsExportJobDisabled(jobSchedule))
             {
-                _logger.LogDebug("Stream {StreamId} is disabled. Removing it from scheduler '{SchedulerName}'.", stream.Id, _schedulerName);
-                removeExportJob(jobKey);
+                _logger.LogDebug("Stream {StreamId} export is disabled. Removing it from scheduler '{SchedulerName}'.", stream.Id, _schedulerName);
+                removeExportJob(jobKey, warnIfNotRemoved: false);
                 return;
             }
 
@@ -72,7 +72,7 @@ async Task addStreamExportJobs(ExecutionContext executionContext, ProviderDefini
         var removed = 0;
         foreach (var jobKey in jobKeysToRemove)
         {
-            removeExportJob(jobKey);
+            removeExportJob(jobKey, warnIfNotRemoved: true);
             removed++;
         }
 
@@ -81,15 +81,15 @@ async Task addStreamExportJobs(ExecutionContext executionContext, ProviderDefini
             _logger.LogInformation("Total of obsolete {RemovedCount} stream jobs removed from scheduler '{SchedulerName}'.", removed, _schedulerName);
         }
 
-        void removeExportJob(string jobKey)
+        void removeExportJob(string jobKey, bool warnIfNotRemoved)
         {
             if (jobQueue.TryRemove(jobKey, out var _))
             {
                 _logger.LogDebug("Stream export for stream {StreamId} removed from scheduler '{SchedulerName}'.", jobKey, _schedulerName);
             }
-            else
+            else if (warnIfNotRemoved)
             {
-                _logger.LogWarning("Failed to disable stream export for stream {StreamId} from scheduler '{SchedulerName}'..", jobKey, _schedulerName);
+                _logger.LogWarning("Failed to disable stream export for stream {StreamId} from scheduler '{SchedulerName}'.", jobKey, _schedulerName);
             }
         }
     }

test/integration/Connector.AzureDataLake.Tests.Integration/AzureDataLakeConnectorTests.cs

@@ -15,6 +15,7 @@
 using Castle.MicroKernel.Registration;
 using Castle.Windsor;
 
+using CluedIn.Connector.AzureDataLake;
 using CluedIn.Connector.AzureDataLake.Connector;
 using CluedIn.Connector.DataLake.Common;
 using CluedIn.Core;
@@ -32,6 +33,7 @@
 using CsvHelper.Configuration;
 
 using Microsoft.Data.SqlClient;
+using Microsoft.Data.SqlClient.Server;
 using Microsoft.Extensions.Internal;
 using Microsoft.Extensions.Logging;
 
@@ -626,9 +628,22 @@ public async Task VerifyStoreData_Sync_WithStreamCacheAndCsvFormat()
         }
 
         [Fact]
-        public async Task VerifyStoreData_Sync_WithStreamCacheAndParquetFormat()
+        public async Task VerifyStoreData_Sync_WithStreamCacheAndParquetFormatUnescaped()
         {
-            await VerifyStoreData_Sync_WithStreamCache("pArQuet", AssertParquetResult);
+            await VerifyStoreData_Sync_WithStreamCache("pArQuet", AssertParquetResultUnescaped);
+        }
+
+        [Fact]
+        public async Task VerifyStoreData_Sync_WithStreamCacheAndParquetFormatWithEscaped()
+        {
+            await VerifyStoreData_Sync_WithStreamCache(
+                "pArQuet",
+                AssertParquetResultEscaped,
+                configureAuthentication: (values) =>
+                {
+                    values.Add(nameof(DataLakeConstants.ShouldEscapeVocabularyKeys), true);
+                    values.Add(nameof(DataLakeConstants.ShouldWriteGuidAsString), true);
+                });
         }
 
         [Fact]
@@ -806,7 +821,8 @@ private async Task VerifyStoreData_Sync_WithStreamCache(
             string format,
             Func<DataLakeFileClient, Task> assertMethod,
             Func<ExecuteExportArg, Task<PathItem>> executeExport = null,
-            Action<Mock<IDateTimeOffsetProvider>> configureTimeProvider = null)
+            Action<Mock<IDateTimeOffsetProvider>> configureTimeProvider = null,
+            Action<Dictionary<string, object>> configureAuthentication = null)
         {
             var organizationId = Guid.NewGuid();
             var providerDefinitionId = Guid.Parse("c444cda8-d9b5-45cc-a82d-fef28e08d55c");
@@ -890,7 +906,7 @@ private async Task VerifyStoreData_Sync_WithStreamCache(
             var directoryName = $"xunit-{DateTime.Now.Ticks}";
 
             var connectorConnectionMock = new Mock<IConnectorConnectionV2>();
-            connectorConnectionMock.Setup(x => x.Authentication).Returns(new Dictionary<string, object>()
+            var authenticationValues = new Dictionary<string, object>()
             {
                 { nameof(AzureDataLakeConstants.AccountName), accountName },
                 { nameof(AzureDataLakeConstants.AccountKey), accountKey },
@@ -900,7 +916,10 @@ private async Task VerifyStoreData_Sync_WithStreamCache(
                 { nameof(DataLakeConstants.StreamCacheConnectionString), streamCacheConnectionString },
                 { nameof(DataLakeConstants.OutputFormat), format },
                 { nameof(DataLakeConstants.UseCurrentTimeForExport), true },
-            });
+            };
+            configureAuthentication?.Invoke(authenticationValues);
+
+            connectorConnectionMock.Setup(x => x.Authentication).Returns(authenticationValues);
 
             var azureDataLakeClient = new AzureDataLakeClient();
             var jobDataFactory = new Mock<AzureDataLakeJobDataFactory>();
@@ -1264,7 +1283,17 @@ private async Task AssertCsvResult(DataLakeFileClient fileClient)
             Assert.Equal(sb.ToString(), content);
         }
 
-        private async Task AssertParquetResult(DataLakeFileClient fileClient)
+        private Task AssertParquetResultUnescaped(DataLakeFileClient fileClient)
+        {
+            return AssertParquetResult(fileClient, ".");
+        }
+
+        private Task AssertParquetResultEscaped(DataLakeFileClient fileClient)
+        {
+            return AssertParquetResult(fileClient, "_");
+        }
+
+        private async Task AssertParquetResult(DataLakeFileClient fileClient, string separator)
         {
             using var memoryStream = new MemoryStream();
             await fileClient.ReadToAsync(memoryStream);
@@ -1300,32 +1329,30 @@ Name Jean Luc Picard
             PersistVersion 1
             ProviderDefinitionId c444cda8-d9b5-45cc-a82d-fef28e08d55c
             Timestamp 2024-08-21T03:16:00.0000000+05:00
-            user_age 123
-            user_dobInDateTime 2000-01-02T03:04:05
-            user_dobInDateTimeOffset 2000-01-02T03:04:05+12:34
-            user_lastName Picard
+            user{{{separator}}}age 123
+            user{{{separator}}}dobInDateTime 2000-01-02T03:04:05
+            user{{{separator}}}dobInDateTimeOffset 2000-01-02T03:04:05+12:34
+            user{{{separator}}}lastName Picard
 
             """, sb.ToString());
 
             object getValue(Parquet.Data.DataColumn dataColumn)
             {
-                if (dataColumn.Field.ClrType == typeof(Guid))
-                    return ((Guid[])dataColumn.Data)[0];
-                else if (dataColumn.Field.ClrType == typeof(int))
-                {
-                    if (dataColumn.Field.IsNullable)
-                        return ((int?[])dataColumn.Data)[0];
+                var type = dataColumn.Field.ClrType;
 
-                    return ((int[])dataColumn.Data)[0];
+                if (type == typeof(Guid))
+                {
+                    return getValueFromArray<Guid?, Guid>(dataColumn);
                 }
-                else if (dataColumn.Field.ClrType == typeof(long))
+                else if (type == typeof(int))
                 {
-                    if (dataColumn.Field.IsNullable)
-                        return ((long?[])dataColumn.Data)[0];
-
-                    return ((long[])dataColumn.Data)[0];
+                    return getValueFromArray<int?, int>(dataColumn);
                 }
-                else if (dataColumn.Field.ClrType == typeof(string))
+                else if (type == typeof(long))
+                {
+                    return getValueFromArray<long?, long>(dataColumn);
+                }
+                else if (type == typeof(string))
                 {
                     var value = ((string[])dataColumn.Data);
 
@@ -1337,9 +1364,15 @@ object getValue(Parquet.Data.DataColumn dataColumn)
 
                 throw new NotSupportedException($"Type {dataColumn.Field.ClrType} not supported.");
             }
-        }
 
+            object getValueFromArray<TNullable, TNonNullable>(Parquet.Data.DataColumn dataColumn)
+            {
+                if (dataColumn.Field.IsNullable)
+                    return ((TNullable[])dataColumn.Data)[0];
 
+                return ((TNonNullable[])dataColumn.Data)[0];
+            }
+        }
 
         private static async Task WaitForFileToBeDeleted(string fileSystemName, string directoryName, DataLakeServiceClient client, PathItem path)
         {