CDAP-19389: Show correct SQL Datetime type for source table during Assessment (#218)

Author: sumitjnn

sqlserver-delta-plugins/src/main/java/io/cdap/delta/sqlserver/SqlServerTableAssessor.java

@@ -59,6 +59,7 @@ public TableAssessment assess(TableDetail tableDetail) {
   static ColumnEvaluation evaluateColumn(ColumnDetail detail) throws IllegalArgumentException {
     Schema schema;
     int sqlType = detail.getType().getVendorTypeNumber();
+    String sqlTypeName = detail.getType().getName();
     Map<String, String> properties = detail.getProperties();
     String upperCaseTypeName = properties.get(TYPE_NAME).toUpperCase();
     ColumnSupport support = ColumnSupport.YES;
@@ -118,6 +119,7 @@ static ColumnEvaluation evaluateColumn(ColumnDetail detail) throws IllegalArgume
         break;
       case Types.TIMESTAMP:
         schema = Schema.of(Schema.LogicalType.DATETIME);
+        sqlTypeName = upperCaseTypeName;
         if (DATETIME2.equals(upperCaseTypeName)) {
           scale = Integer.parseInt(properties.get(SCALE));
           if (scale > MAX_SUPPORTED_SCALE) {
@@ -161,7 +163,7 @@ static ColumnEvaluation evaluateColumn(ColumnDetail detail) throws IllegalArgume
 
     Schema.Field field = schema == null ? null :
       Schema.Field.of(detail.getName(), detail.isNullable() ? Schema.nullableOf(schema) : schema);
-    ColumnAssessment assessment = ColumnAssessment.builder(detail.getName(), detail.getType().getName())
+    ColumnAssessment assessment = ColumnAssessment.builder(detail.getName(), sqlTypeName)
       .setSupport(support)
       .setSuggestion(suggestion)
       .build();

sqlserver-delta-plugins/src/test/java/io.cdap.delta.sqlserver/SqlServerTableAssessorTest.java

@@ -16,6 +16,7 @@
 package io.cdap.delta.sqlserver;
 
 import io.cdap.cdap.api.data.schema.Schema;
+import io.cdap.delta.api.assessment.ColumnAssessment;
 import io.cdap.delta.api.assessment.ColumnDetail;
 import io.cdap.delta.api.assessment.ColumnSupport;
 import io.cdap.delta.api.assessment.TableAssessment;
@@ -56,10 +57,12 @@ public void testDateTime2WithNanoScaleMapping() {
     ColumnEvaluation columnEvaluation = SqlServerTableAssessor.evaluateColumn(columnDetail);
     TableAssessment assessment = tableAssessor.assess(tableDetail);
 
+    Assert.assertEquals(1, assessment.getColumns().size());
+    ColumnAssessment columnAssessment = assessment.getColumns().get(0);
+    Assert.assertEquals(SqlServerTableAssessor.DATETIME2, columnAssessment.getType());
+    Assert.assertEquals(ColumnSupport.PARTIAL, columnAssessment.getSupport());
     Assert.assertEquals(Schema.LogicalType.DATETIME,
                         columnEvaluation.getField().getSchema().getNonNullable().getLogicalType());
-    Assert.assertEquals(1, assessment.getColumns().size());
-    Assert.assertEquals(ColumnSupport.PARTIAL, assessment.getColumns().get(0).getSupport());
   }
 
   @Test
@@ -94,7 +97,9 @@ public void testDateTimeMapping() {
     Assert.assertEquals(Schema.LogicalType.DATETIME,
                         columnEvaluation.getField().getSchema().getNonNullable().getLogicalType());
     Assert.assertEquals(1, assessment.getColumns().size());
-    Assert.assertEquals(ColumnSupport.YES, assessment.getColumns().get(0).getSupport());
+    ColumnAssessment columnAssessment = assessment.getColumns().get(0);
+    Assert.assertEquals(DATETIME, columnAssessment.getType());
+    Assert.assertEquals(ColumnSupport.YES, columnAssessment.getSupport());
   }
 
   @Test