Changed the logic to handle datetime datatype (#375)

* Changed the logic to handle datetime datatype

* Changed the logic to handle datetime datatype

* Changed the logic to handle datetime datatype

Author: Nikitapaliwal123

database-commons/src/main/java/io/cdap/plugin/db/DBRecord.java

@@ -311,6 +311,9 @@ protected void writeToDB(PreparedStatement stmt, @Nullable Schema.Field field, i
         case DECIMAL:
           stmt.setBigDecimal(sqlIndex, record.getDecimal(fieldName));
           break;
+        case DATETIME:
+          stmt.setTimestamp(sqlIndex, Timestamp.valueOf(record.getDateTime(fieldName)));
+          break;
       }
       return;
     }

mssql-plugin/src/main/java/io/cdap/plugin/mssql/SqlFieldsValidator.java

@@ -38,6 +38,14 @@ public boolean isFieldCompatible(Schema.Field field, ResultSetMetaData metadata,
     boolean isSigned = metadata.isSigned(index);
     int precision = metadata.getPrecision(index);
 
+    // Handles datetime datatypes
+    // Case when Timestamp maps to datetime
+    // Case when Datetime2 maps to datetime
+    // Case when DatetimeOffset maps to datetime
+    if ((sqlType == Types.TIMESTAMP || sqlType == SqlServerSourceSchemaReader.DATETIME_OFFSET_TYPE)
+            && fieldLogicalType.equals(Schema.LogicalType.DATETIME)) {
+      return true;
+    }
     // Handle logical types first
     if (fieldLogicalType != null) {
       return super.isFieldCompatible(fieldType, fieldLogicalType, sqlType, precision, isSigned);