From e68103226139ba971f85897d514179ab116c9cbd Mon Sep 17 00:00:00 2001
From: Sumit Jain <sumitjn@google.com>
Date: Wed, 30 Nov 2022 18:49:57 +0530
Subject: [PATCH] CDAP-19389: Show correct SQL Datetime type for source table
 during Assessment

---
 .../cdap/delta/sqlserver/SqlServerTableAssessor.java  |  4 +++-
 .../SqlServerTableAssessorTest.java                   | 11 ++++++++---
 2 files changed, 11 insertions(+), 4 deletions(-)

diff --git a/sqlserver-delta-plugins/src/main/java/io/cdap/delta/sqlserver/SqlServerTableAssessor.java b/sqlserver-delta-plugins/src/main/java/io/cdap/delta/sqlserver/SqlServerTableAssessor.java
index 6014dedd..6b65b054 100644
--- a/sqlserver-delta-plugins/src/main/java/io/cdap/delta/sqlserver/SqlServerTableAssessor.java
+++ b/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();
diff --git a/sqlserver-delta-plugins/src/test/java/io.cdap.delta.sqlserver/SqlServerTableAssessorTest.java b/sqlserver-delta-plugins/src/test/java/io.cdap.delta.sqlserver/SqlServerTableAssessorTest.java
index 27578d1e..79b4b5ca 100644
--- a/sqlserver-delta-plugins/src/test/java/io.cdap.delta.sqlserver/SqlServerTableAssessorTest.java
+++ b/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