Improve STRUCT support and expand data type mappings

Author: vishwasvaidya-cloudsufi

oracle-plugin/src/main/java/io/cdap/plugin/oracle/OracleSourceSchemaReader.java

@@ -19,7 +19,6 @@
 import com.google.common.collect.ImmutableSet;
 import io.cdap.cdap.api.data.schema.Schema;
 import io.cdap.plugin.db.CommonSchemaReader;
-import org.jetbrains.annotations.NotNull;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -198,6 +197,10 @@ private Schema getStructSchema(Connection connection, String typeName, String ow
             fields.add(Schema.Field.of(attrName, attrSchema));
           } else {
             String nestedStructOwner = attrRs.getString("ATTR_TYPE_OWNER");
+            if (nestedStructOwner == null || nestedStructOwner.isEmpty()) {
+              throw new SQLException(String.format("Attribute '%s' is not a primitive type, but it lacks a type " +
+                      "owner. Therefore, it cannot be resolved as a STRUCT type. ", attrName));
+            }
             Schema nestedSchema = getStructSchema(connection, attrTypeName, nestedStructOwner);
             fields.add(Schema.Field.of(attrName, nestedSchema));
           }
@@ -215,11 +218,11 @@ private Schema getStructSchema(Connection connection, String typeName, String ow
   }
 
   private Schema mapPrimitiveOracleType(String typeName, int precision, int scale, String columnName) {
-    return OracleUserTypeSchemaMapping.mapPrimitiveOracleType(isTimestampOldBehavior, getTimestampLtzSchema(),
+    return OracleStructTypeSchemaMapping.mapPrimitiveOracleType(isTimestampOldBehavior, getTimestampLtzSchema(),
             isPrecisionlessNumAsDecimal, typeName, precision, scale, columnName);
   }
 
-  private @NotNull Schema getTimestampLtzSchema() {
+  private Schema getTimestampLtzSchema() {
     return isTimestampOldBehavior || isTimestampLtzFieldTimestamp
       ? Schema.of(Schema.LogicalType.TIMESTAMP_MICROS)
       : Schema.of(Schema.LogicalType.DATETIME);

oracle-plugin/src/main/java/io/cdap/plugin/oracle/OracleStructAttributeConverters.java

@@ -23,6 +23,7 @@
 import java.sql.Blob;
 import java.sql.Clob;
 import java.sql.SQLException;
+import java.sql.SQLXML;
 import java.sql.Timestamp;
 import java.time.OffsetDateTime;
 import java.time.ZoneId;
@@ -154,7 +155,13 @@ public void convert(StructuredRecord.Builder builder, Schema.Field field, Schema
   private static class OracleBfileConverter implements AttributeConverter {
     @Override
     public boolean canConvert(Object attrValue, String attrClassName) {
-      return "oracle.jdbc.OracleBfile".equals(attrClassName);
+      try {
+        ClassLoader oracleLoader = attrValue.getClass().getClassLoader();
+        Class<?> bfileInterface = oracleLoader.loadClass("oracle.jdbc.OracleBfile");
+        return bfileInterface.isInstance(attrValue);
+      } catch (Exception e) {
+        return false;
+      }
     }
 
     @Override
@@ -190,6 +197,26 @@ public void convert(StructuredRecord.Builder builder, Schema.Field field, Schema
     }
   }
 
+  private static class SqlXmlConverter implements AttributeConverter {
+
+    @Override
+    public boolean canConvert(Object attrValue, String attrClassName) {
+      return attrValue instanceof SQLXML;
+    }
+
+    @Override
+    public void convert(
+            StructuredRecord.Builder builder,
+            Schema.Field field,
+            Schema fieldSchema,
+            Object attrValue,
+            BfileBytesResolver resolver) throws SQLException {
+
+      SQLXML xml = (SQLXML) attrValue;
+      builder.set(field.getName(), xml.getString());
+    }
+  }
+
   private static class DefaultConverter implements AttributeConverter {
     @Override
     public boolean canConvert(Object attrValue, String attrClassName) {
@@ -212,6 +239,7 @@ public void convert(StructuredRecord.Builder builder, Schema.Field field, Schema
       new OracleBfileConverter(),
       new ByteArrayConverter(),
       new OracleIntervalConverter(),
+      new SqlXmlConverter(),
       new DefaultConverter()
   );
 

oracle-plugin/src/main/java/io/cdap/plugin/oracle/OracleStructTypeSchemaMapping.java

@@ -29,8 +29,8 @@
 /**
  * Registry containing schema type mappers for Oracle specific datatypes.
  */
-public final class OracleUserTypeSchemaMapping {
-  private static final Logger LOG = LoggerFactory.getLogger(OracleUserTypeSchemaMapping.class);
+public final class OracleStructTypeSchemaMapping {
+  private static final Logger LOG = LoggerFactory.getLogger(OracleStructTypeSchemaMapping.class);
 
   private interface TypeMapper {
     Schema map(boolean isTimestampOldBehavior, Schema timestampLtzSchema,
@@ -41,12 +41,12 @@ Schema map(boolean isTimestampOldBehavior, Schema timestampLtzSchema,
 
   static {
     TypeMapper floatMapper = (isOld, ltzS, precD, typeName, p, s, col) -> Schema.of(Schema.Type.FLOAT);
-    TYPE_MAPPERS.put("BINARY FLOAT", floatMapper);
+    TYPE_MAPPERS.put("BINARY_FLOAT", floatMapper);
     TYPE_MAPPERS.put("REAL", floatMapper);
     TYPE_MAPPERS.put("FLOAT", floatMapper);
 
     TypeMapper doubleMapper = (isOld, ltzS, precD, typeName, p, s, col) -> Schema.of(Schema.Type.DOUBLE);
-    TYPE_MAPPERS.put("BINARY DOUBLE", doubleMapper);
+    TYPE_MAPPERS.put("BINARY_DOUBLE", doubleMapper);
     TYPE_MAPPERS.put("DOUBLE", doubleMapper);
 
     // Bytes types
@@ -64,44 +64,52 @@ Schema map(boolean isTimestampOldBehavior, Schema timestampLtzSchema,
     TYPE_MAPPERS.put("VARCHAR", stringMapper);
     TYPE_MAPPERS.put("CHAR", stringMapper);
     TYPE_MAPPERS.put("CHAR2", stringMapper);
+    TYPE_MAPPERS.put("NCHAR", stringMapper);
+    TYPE_MAPPERS.put("NVARCHAR2", stringMapper);
     TYPE_MAPPERS.put("CLOB", stringMapper);
     TYPE_MAPPERS.put("NCLOB", stringMapper);
     TYPE_MAPPERS.put("LONG", stringMapper);
+    TYPE_MAPPERS.put("ROWID", stringMapper);
+    TYPE_MAPPERS.put("UROWID", stringMapper);
 
-    // Specific types
+    // Date and Time types
     TYPE_MAPPERS.put("TIMESTAMP WITH TZ", (isOld, ltzS, precD, typeName, p, s, col) ->
             isOld ? Schema.of(Schema.Type.STRING) : Schema.of(Schema.LogicalType.TIMESTAMP_MICROS)
     );
-    TYPE_MAPPERS.put("TIMESTAMP WITH LTZ", (isOld, ltzS, precD, typeName, p, s, col) -> ltzS);
+    TYPE_MAPPERS.put("TIMESTAMP WITH LOCAL TZ", (isOld, ltzS, precD, typeName, p, s, col) -> ltzS);
     TYPE_MAPPERS.put("TIMESTAMP", (isOld, ltzS, precD, typeName, p, s, col) ->
             isOld ? Schema.of(Schema.LogicalType.TIMESTAMP_MICROS) : Schema.of(Schema.LogicalType.DATETIME)
     );
     TYPE_MAPPERS.put("DATE", (isOld, ltzS, precD, typeName, p, s, col) -> Schema.of(Schema.LogicalType.DATE));
     TYPE_MAPPERS.put("TIME", (isOld, ltzS, precD, typeName, p, s, col) -> Schema.of(Schema.LogicalType.TIME_MICROS));
+
+    // Numeric types
     TYPE_MAPPERS.put("INTEGER", (isOld, ltzS, precD, typeName, p, s, col) -> Schema.of(Schema.Type.INT));
+    TYPE_MAPPERS.put("NUMBER", OracleStructTypeSchemaMapping::mapNumberOrDecimal);
+    TYPE_MAPPERS.put("DECIMAL", OracleStructTypeSchemaMapping::mapNumberOrDecimal);
 
-    TYPE_MAPPERS.put("NUMBER", OracleUserTypeSchemaMapping::mapNumberOrDecimal);
-    TYPE_MAPPERS.put("DECIMAL", OracleUserTypeSchemaMapping::mapNumberOrDecimal);
+    // XML type
+    TYPE_MAPPERS.put("XMLTYPE", (isOld, ltzS, precD, typeName, p, s, col) -> Schema.of(Schema.Type.STRING));
 
     // Unsupported types that throw error
     TYPE_MAPPERS.put("ARRAY", (isOld, ltzS, precD, typeName, p, s, col) -> {
       String errorMessage = String.format("Column %s has unsupported SQL type of %s.", col, typeName);
       throw ErrorUtils.getProgramFailureException(new ErrorCategory(ErrorCategory.ErrorCategoryEnum.PLUGIN),
               errorMessage, errorMessage, ErrorType.SYSTEM, true, null);
     });
-    TYPE_MAPPERS.put("OTHER", (isOld, ltzS, precD, typeName, p, s, col) -> {
+    TYPE_MAPPERS.put("ANYDATA", (isOld, ltzS, precD, typeName, p, s, col) -> {
       String errorMessage = String.format("Column %s has unsupported SQL type of %s.", col, typeName);
       throw ErrorUtils.getProgramFailureException(new ErrorCategory(ErrorCategory.ErrorCategoryEnum.PLUGIN),
               errorMessage, errorMessage, ErrorType.SYSTEM, true, null);
     });
-    TYPE_MAPPERS.put("XML", (isOld, ltzS, precD, typeName, p, s, col) -> {
+    TYPE_MAPPERS.put("OTHER", (isOld, ltzS, precD, typeName, p, s, col) -> {
       String errorMessage = String.format("Column %s has unsupported SQL type of %s.", col, typeName);
       throw ErrorUtils.getProgramFailureException(new ErrorCategory(ErrorCategory.ErrorCategoryEnum.PLUGIN),
               errorMessage, errorMessage, ErrorType.SYSTEM, true, null);
     });
   }
 
-  private OracleUserTypeSchemaMapping() {
+  private OracleStructTypeSchemaMapping() {
     // Private constructor to prevent instantiation of utility class.
   }