Mssql destination: enable DAT tests, use nvarchar and datetime2 by default

airbyte-integrations/bases/base-normalization/Dockerfile

@@ -28,5 +28,5 @@ WORKDIR /airbyte
 ENV AIRBYTE_ENTRYPOINT "/airbyte/entrypoint.sh"
 ENTRYPOINT ["/airbyte/entrypoint.sh"]
 
-LABEL io.airbyte.version=0.1.77
+LABEL io.airbyte.version=0.1.78
 LABEL io.airbyte.name=airbyte/normalization

airbyte-integrations/bases/base-normalization/dbt-project-template/macros/cross_db_utils/datatypes.sql

@@ -33,7 +33,7 @@
 {%- endmacro -%}
 
 {%- macro sqlserver__type_json() -%}
-    VARCHAR(max)
+    NVARCHAR(max)
 {%- endmacro -%}
 
 {% macro clickhouse__type_json() %}
@@ -52,7 +52,7 @@
 {%- endmacro -%}
 
 {% macro sqlserver__type_string() %}
-    VARCHAR(max)
+    NVARCHAR(max)
 {%- endmacro -%}
 
 {%- macro clickhouse__type_string() -%}
@@ -154,7 +154,7 @@
 {%- macro sqlserver__type_timestamp_with_timezone() -%}
     {#-- in TSQL timestamp is really datetime or datetime2 --#}
     {#-- https://docs.microsoft.com/en-us/sql/t-sql/functions/date-and-time-data-types-and-functions-transact-sql?view=sql-server-ver15#DateandTimeDataTypes --#}
-    datetime
+    datetime2
 {%- endmacro -%}
 
 {% macro clickhouse__type_timestamp_with_timezone() %}

airbyte-integrations/bases/base-normalization/integration_tests/resources/test_simple_streams/data_input/messages.txt

@@ -51,3 +51,4 @@
 {"type": "RECORD", "record": {"stream": "1_prefix_startwith_number", "emitted_at": 1602637990800, "data": { "id": 2, "date": "2020-09-01", "text": "hi 4"}}}
 
 {"type":"RECORD","record":{"stream":"multiple_column_names_conflicts","data":{"id":1,"User Id":"chris","user_id":42,"User id":300,"user id": 102,"UserId":101},"emitted_at":1623959926}}
+{"type":"RECORD","record":{"stream":"special_characters","data":{"id":1,"character_1":"some random special characters: ࠈൡሗ","user_id":42,"character_2":"some random special characters: žõ⥁🌞"},"emitted_at":1623959932}}

airbyte-integrations/bases/standard-destination-test/src/main/java/io/airbyte/integrations/standardtest/destination/comparator/AdvancedTestDataComparator.java

@@ -8,6 +8,7 @@
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
 import com.fasterxml.jackson.databind.JsonNode;
+import io.airbyte.commons.json.Jsons;
 import java.time.ZoneOffset;
 import java.time.ZonedDateTime;
 import java.time.format.DateTimeFormatter;
@@ -91,6 +92,8 @@ protected boolean compareJsonNodes(final JsonNode expectedValue, final JsonNode
       return compareDateValues(expectedValue.asText(), actualValue.asText());
     } else if (expectedValue.isArray() && actualValue.isArray()) {
       return compareArrays(expectedValue, actualValue);
+    } else if (expectedValue.isArray() && isQuotedArray(actualValue)) {
+      return compareArrays(expectedValue, Jsons.deserialize(actualValue.asText()));
     } else if (expectedValue.isObject() && actualValue.isObject()) {
       compareObjects(expectedValue, actualValue);
       return true;
@@ -99,6 +102,10 @@ protected boolean compareJsonNodes(final JsonNode expectedValue, final JsonNode
     }
   }
 
+  protected boolean isQuotedArray(JsonNode actualValue) {
+    return actualValue.isTextual() && actualValue.asText().matches("^\\[.*\\]$");
+  }
+
   protected boolean compareString(final JsonNode expectedValue, final JsonNode actualValue) {
     return expectedValue.asText().equals(actualValue.asText());
   }

airbyte-integrations/connectors/destination-mssql/src/test-integration/java/io/airbyte/integrations/destination/mssql/MSSQLDestinationAcceptanceTest.java

@@ -15,6 +15,7 @@
 import io.airbyte.integrations.base.JavaBaseConstants;
 import io.airbyte.integrations.destination.ExtendedNameTransformer;
 import io.airbyte.integrations.standardtest.destination.DestinationAcceptanceTest;
+import io.airbyte.integrations.standardtest.destination.comparator.TestDataComparator;
 import java.sql.SQLException;
 import java.util.ArrayList;
 import java.util.List;
@@ -167,6 +168,26 @@ protected void setup(final TestDestinationEnv testEnv) throws SQLException {
   @Override
   protected void tearDown(final TestDestinationEnv testEnv) {}
 
+  @Override
+  protected TestDataComparator getTestDataComparator() {
+    return new MSSQLTestDataComparator();
+  }
+
+  @Override
+  protected boolean supportBasicDataTypeTest() {
+    return true;
+  }
+
+  @Override
+  protected boolean supportArrayDataTypeTest() {
+    return true;
+  }
+
+  @Override
+  protected boolean supportObjectDataTypeTest() {
+    return true;
+  }
+
   @AfterAll
   static void cleanUp() {
     db.stop();

airbyte-integrations/connectors/destination-mssql/src/test-integration/java/io/airbyte/integrations/destination/mssql/MSSQLTestDataComparator.java

@@ -0,0 +1,45 @@
+package io.airbyte.integrations.destination.mssql;
+
+import io.airbyte.integrations.destination.ExtendedNameTransformer;
+import io.airbyte.integrations.standardtest.destination.comparator.AdvancedTestDataComparator;
+import java.time.LocalDateTime;
+import java.time.ZoneOffset;
+import java.time.ZonedDateTime;
+import java.time.format.DateTimeFormatter;
+import java.util.ArrayList;
+import java.util.List;
+
+public class MSSQLTestDataComparator extends AdvancedTestDataComparator {
+
+  public static final String ACTUAL_MSSQL_AIRBYTE_DATETIME_FORMAT = "yyyy-MM-dd HH:mm:ss.S";
+  private final ExtendedNameTransformer namingResolver = new ExtendedNameTransformer();
+
+
+  @Override
+  protected boolean compareDateTimeValues(String airbyteMessageValue, String destinationValue) {
+    if(!isDateTimeValue(destinationValue)){
+      destinationValue = LocalDateTime.parse(destinationValue, DateTimeFormatter.ofPattern(ACTUAL_MSSQL_AIRBYTE_DATETIME_FORMAT)).toString();
+    }
+    return super.compareDateTimeValues(airbyteMessageValue, destinationValue);
+  }
+
+  @Override
+  protected ZonedDateTime parseDestinationDateWithTz(String destinationValue) {
+    LocalDateTime parsedDateTime = LocalDateTime.parse(destinationValue, DateTimeFormatter.ofPattern(ACTUAL_MSSQL_AIRBYTE_DATETIME_FORMAT));
+    return ZonedDateTime.of(parsedDateTime, ZoneOffset.UTC);
+  }
+
+  @Override
+  protected List<String> resolveIdentifier(final String identifier) {
+    final List<String> result = new ArrayList<>();
+    final String resolved = namingResolver.getIdentifier(identifier);
+    result.add(identifier);
+    result.add(resolved);
+    if (!resolved.startsWith("\"")) {
+      result.add(resolved.toLowerCase());
+      result.add(resolved.toUpperCase());
+    }
+    return result;
+  }
+
+}

airbyte-workers/src/main/java/io/airbyte/workers/normalization/NormalizationRunnerFactory.java

@@ -14,7 +14,7 @@
 public class NormalizationRunnerFactory {
 
   public static final String BASE_NORMALIZATION_IMAGE_NAME = "airbyte/normalization";
-  public static final String NORMALIZATION_VERSION = "0.1.77";
+  public static final String NORMALIZATION_VERSION = "0.1.78";
 
   static final Map<String, ImmutablePair<String, DefaultNormalizationRunner.DestinationType>> NORMALIZATION_MAPPING =
       ImmutableMap.<String, ImmutablePair<String, DefaultNormalizationRunner.DestinationType>>builder()