diff --git a/database-commons/src/main/java/io/cdap/plugin/db/CommonSchemaReader.java b/database-commons/src/main/java/io/cdap/plugin/db/CommonSchemaReader.java index 28c56db8c..b2348ee5e 100644 --- a/database-commons/src/main/java/io/cdap/plugin/db/CommonSchemaReader.java +++ b/database-commons/src/main/java/io/cdap/plugin/db/CommonSchemaReader.java @@ -57,6 +57,11 @@ public Schema getSchema(ResultSetMetaData metadata, int index) throws SQLExcepti metadata.isSigned(index), true); } + public Schema getSchema(String typeName, int sqlType, int precision, int scale, String columnName, + boolean isSigned) throws SQLException { + return DBUtils.getSchema(typeName, sqlType, precision, scale, columnName, isSigned, true); + } + @Override public boolean shouldIgnoreColumn(ResultSetMetaData metadata, int index) throws SQLException { return false; diff --git a/database-commons/src/main/java/io/cdap/plugin/db/DBRecord.java b/database-commons/src/main/java/io/cdap/plugin/db/DBRecord.java index b187c7670..bd20cb1a5 100644 --- a/database-commons/src/main/java/io/cdap/plugin/db/DBRecord.java +++ b/database-commons/src/main/java/io/cdap/plugin/db/DBRecord.java @@ -36,6 +36,7 @@ import java.math.BigDecimal; import java.math.BigInteger; import java.nio.ByteBuffer; +import java.sql.Connection; import java.sql.Date; import java.sql.PreparedStatement; import java.sql.ResultSet; @@ -187,7 +188,22 @@ protected void handleField(ResultSet resultSet, StructuredRecord.Builder recordB protected void setField(ResultSet resultSet, StructuredRecord.Builder recordBuilder, Schema.Field field, int columnIndex, int sqlType, int sqlPrecision, int sqlScale) throws SQLException { - Object o = DBUtils.transformValue(sqlType, sqlPrecision, sqlScale, resultSet, columnIndex); + Object fieldValue = DBUtils.transformValue(sqlType, sqlPrecision, sqlScale, resultSet, columnIndex); + populateRecordField(resultSet.getStatement().getConnection(), recordBuilder, field, fieldValue); + } + + /** + * Populates the value of a field in the {@link StructuredRecord.Builder}. + * + * @param connection the SQL connection, provided for subclass overrides that require database connection + * @param recordBuilder the builder for constructing the {@link StructuredRecord} + * @param field the field to set in the record + * @param o the object value read from the database + * @throws SQLException if an error occurs while setting the field value + */ + public void populateRecordField(Connection connection, StructuredRecord.Builder recordBuilder, + Schema.Field field, Object o) + throws SQLException { if (o instanceof Date) { recordBuilder.setDate(field.getName(), ((Date) o).toLocalDate()); } else if (o instanceof Time) { diff --git a/oracle-plugin/src/main/java/io/cdap/plugin/oracle/OracleSourceDBRecord.java b/oracle-plugin/src/main/java/io/cdap/plugin/oracle/OracleSourceDBRecord.java index 44131a01b..bf3a978b1 100644 --- a/oracle-plugin/src/main/java/io/cdap/plugin/oracle/OracleSourceDBRecord.java +++ b/oracle-plugin/src/main/java/io/cdap/plugin/oracle/OracleSourceDBRecord.java @@ -35,6 +35,7 @@ import java.sql.ResultSet; import java.sql.ResultSetMetaData; import java.sql.SQLException; +import java.sql.Struct; import java.sql.Timestamp; import java.sql.Types; import java.time.LocalDateTime; @@ -43,6 +44,8 @@ import java.time.ZoneOffset; import java.time.ZonedDateTime; import java.util.List; +import java.util.Map; +import java.util.TreeMap; /** * Oracle Source implementation {@link org.apache.hadoop.mapreduce.lib.db.DBWritable} and @@ -106,13 +109,19 @@ record = recordBuilder.build(); @Override protected void handleField(ResultSet resultSet, StructuredRecord.Builder recordBuilder, Schema.Field field, int columnIndex, int sqlType, int sqlPrecision, int sqlScale) throws SQLException { - if (OracleSourceSchemaReader.ORACLE_TYPES.contains(sqlType) || sqlType == Types.NCLOB) { + if (isOracleSpecificType(sqlType)) { handleOracleSpecificType(resultSet, recordBuilder, field, columnIndex, sqlType, sqlPrecision, sqlScale); } else { setField(resultSet, recordBuilder, field, columnIndex, sqlType, sqlPrecision, sqlScale); } } + protected boolean isOracleSpecificType(int sqlType) { + return OracleSourceSchemaReader.ORACLE_TYPES.contains(sqlType) + || sqlType == Types.NCLOB + || sqlType == Types.STRUCT; + } + @Override protected void writeNonNullToDB(PreparedStatement stmt, Schema fieldSchema, String fieldName, int fieldIndex) throws SQLException { @@ -232,11 +241,15 @@ private Object createOracleTimestamp(Connection connection, String timestampStri */ private byte[] getBfileBytes(ResultSet resultSet, String columnName) throws SQLException { Object bfile = resultSet.getObject(columnName); + return getBfileBytes(bfile, columnName); + } + + public byte[] getBfileBytes(Object bfile, String columnName) { if (bfile == null) { return null; } try { - ClassLoader classLoader = resultSet.getClass().getClassLoader(); + ClassLoader classLoader = bfile.getClass().getClassLoader(); Class oracleBfileClass = classLoader.loadClass("oracle.jdbc.OracleBfile"); boolean isFileExist = (boolean) oracleBfileClass.getMethod("fileExists").invoke(bfile); if (!isFileExist) { @@ -341,6 +354,15 @@ private void handleOracleSpecificType(ResultSet resultSet, StructuredRecord.Buil case OracleSourceSchemaReader.LONG_RAW: recordBuilder.set(field.getName(), resultSet.getBytes(columnIndex)); break; + case Types.STRUCT: + Struct structValue = (Struct) resultSet.getObject(columnIndex); + if (structValue != null) { + recordBuilder.set(field.getName(), convertStructToRecord(structValue, nonNullSchema, + resultSet.getStatement().getConnection())); + } else { + recordBuilder.set(field.getName(), null); + } + break; case Types.DECIMAL: case Types.NUMERIC: // This is the only way to differentiate FLOAT/REAL columns from other numeric columns, that based on NUMBER. @@ -371,6 +393,57 @@ private void handleOracleSpecificType(ResultSet resultSet, StructuredRecord.Buil } } + /** + * Converts a JDBC {@link Struct} into a {@link StructuredRecord} based on the provided schema. + * + * @param struct the SQL structured type containing the source data attributes + * @param schema the target record schema defining the fields to map + * @param connection the database connection + * @return a populated {@code StructuredRecord} instance + * @throws SQLException if an error occurs reading the struct attributes or metadata + */ + protected StructuredRecord convertStructToRecord(Struct struct, Schema schema, Connection connection) + throws SQLException { + Map attributeMap = getAttributeMap(struct, schema, connection); + StructuredRecord.Builder builder = StructuredRecord.builder(schema); + + for (Schema.Field field : schema.getFields()) { + Object attrValue = attributeMap.get(field.getName()); + OracleStructUtil.populateRecordField(this, connection, builder, field, attrValue); + } + return builder.build(); + } + + /** + * Extracts attributes from a {@link Struct} into a case-insensitive map indexed by column name. + * Uses reflection to extract underlying metadata (e.g., from Oracle StructDescriptor). + * + * @param struct the source SQL structured type + * @param schema the target schema used for context in error messages + * @param connection the database connection + * @return a case-insensitive {@code Map} linking column names to their attribute values + * @throws SQLException if metadata extraction fails or driver-specific methods are inaccessible + */ + protected Map getAttributeMap(Struct struct, Schema schema, Connection connection) + throws SQLException { + Map attributeMap = new TreeMap<>(String.CASE_INSENSITIVE_ORDER); + Object[] attributes = struct.getAttributes(); + if (attributes != null) { + try { + Object descriptor = struct.getClass().getMethod("getDescriptor").invoke(struct); + ResultSetMetaData metaData = + (ResultSetMetaData) descriptor.getClass().getMethod("getMetaData").invoke(descriptor); + for (int i = 1; i <= metaData.getColumnCount() && (i - 1) < attributes.length; i++) { + attributeMap.put(metaData.getColumnName(i), attributes[i - 1]); + } + } catch (Exception e) { + throw new SQLException(String.format("Failed to retrieve attribute metadata for Oracle STRUCT schema '%s': %s", + schema.getRecordName(), e.getMessage())); + } + } + return attributeMap; + } + /** * Get the scale set in Non-nullable schema associated with the schema * */ diff --git a/oracle-plugin/src/main/java/io/cdap/plugin/oracle/OracleSourceSchemaReader.java b/oracle-plugin/src/main/java/io/cdap/plugin/oracle/OracleSourceSchemaReader.java index b23dfa031..9743eaca7 100644 --- a/oracle-plugin/src/main/java/io/cdap/plugin/oracle/OracleSourceSchemaReader.java +++ b/oracle-plugin/src/main/java/io/cdap/plugin/oracle/OracleSourceSchemaReader.java @@ -22,9 +22,16 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import java.sql.Connection; +import java.sql.PreparedStatement; +import java.sql.ResultSet; import java.sql.ResultSetMetaData; import java.sql.SQLException; import java.sql.Types; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; import java.util.Set; import javax.annotation.Nullable; @@ -45,6 +52,44 @@ public class OracleSourceSchemaReader extends CommonSchemaReader { public static final int LONG = -1; public static final int LONG_RAW = -4; + /** + * Maps Oracle string data type inside UDT to their corresponding java.sql.Types integer constants + */ + private static final Map DATA_TYPE_MAP = new HashMap<>(); + static { + DATA_TYPE_MAP.put("TIMESTAMP WITH LOCAL TZ", TIMESTAMP_LTZ); + DATA_TYPE_MAP.put("TIMESTAMP WITH TZ", TIMESTAMP_TZ); + DATA_TYPE_MAP.put("TIMESTAMP", Types.TIMESTAMP); + DATA_TYPE_MAP.put("DATE", Types.TIMESTAMP); + DATA_TYPE_MAP.put("FLOAT", Types.DOUBLE); + DATA_TYPE_MAP.put("BINARY_FLOAT", BINARY_FLOAT); + DATA_TYPE_MAP.put("REAL", Types.DOUBLE); + DATA_TYPE_MAP.put("BINARY_DOUBLE", BINARY_DOUBLE); + DATA_TYPE_MAP.put("DOUBLE", Types.DOUBLE); + DATA_TYPE_MAP.put("BFILE", BFILE); + DATA_TYPE_MAP.put("RAW", LONG_RAW); + DATA_TYPE_MAP.put("INTERVAL DAY TO SECOND", INTERVAL_DS); + DATA_TYPE_MAP.put("INTERVAL YEAR TO MONTH", INTERVAL_YM); + DATA_TYPE_MAP.put("XMLTYPE", Types.SQLXML); + DATA_TYPE_MAP.put("ARRAY", Types.ARRAY); + DATA_TYPE_MAP.put("ANYDATA", Types.JAVA_OBJECT); + DATA_TYPE_MAP.put("OTHER", Types.OTHER); + DATA_TYPE_MAP.put("NUMBER", Types.NUMERIC); + DATA_TYPE_MAP.put("SMALLINT", Types.DECIMAL); + DATA_TYPE_MAP.put("DECIMAL", Types.DECIMAL); + DATA_TYPE_MAP.put("INTEGER", Types.DECIMAL); + DATA_TYPE_MAP.put("UROWID", Types.ROWID); + DATA_TYPE_MAP.put("BLOB", Types.BLOB); + DATA_TYPE_MAP.put("CLOB", Types.CLOB); + DATA_TYPE_MAP.put("NCLOB", Types.NCLOB); + DATA_TYPE_MAP.put("VARCHAR2", Types.VARCHAR); + DATA_TYPE_MAP.put("VARCHAR", Types.VARCHAR); + DATA_TYPE_MAP.put("CHAR", Types.CHAR); + DATA_TYPE_MAP.put("CHAR2", Types.CHAR); + DATA_TYPE_MAP.put("NCHAR", Types.NCHAR); + DATA_TYPE_MAP.put("NVARCHAR2", Types.NVARCHAR); + } + /** * Logger instance for Oracle Schema reader. */ @@ -71,6 +116,13 @@ public class OracleSourceSchemaReader extends CommonSchemaReader { private final Boolean isPrecisionlessNumAsDecimal; private final Boolean isTimestampLtzFieldTimestamp; private final Boolean isXmlTypeEnabled; + private Connection connection; + + private static final String COLUMN_ATTR_NAME = "ATTR_NAME"; + private static final String COLUMN_ATTR_TYPE_NAME = "ATTR_TYPE_NAME"; + private static final String COLUMN_PRECISION = "PRECISION"; + private static final String COLUMN_SCALE = "SCALE"; + private static final String COLUMN_ATTR_TYPE_OWNER = "ATTR_TYPE_OWNER"; public OracleSourceSchemaReader() { this(null, false, false, false, false); @@ -88,6 +140,45 @@ public OracleSourceSchemaReader(@Nullable String sessionID, boolean isTimestampO @Override public Schema getSchema(ResultSetMetaData metadata, int index) throws SQLException { int sqlType = metadata.getColumnType(index); + String columnTypeName = metadata.getColumnTypeName(index); + String owner = extractOwnerName(columnTypeName, metadata.getColumnName(index), sqlType); + ColumnMetadata columnMetadata = new ColumnMetadata(sqlType, metadata.getColumnClassName(index), + metadata.getPrecision(index), metadata.getScale(index), metadata.getColumnName(index), + columnTypeName, metadata.isSigned(index), owner, 0); + return getSchemaMapping(columnMetadata); + } + + private String extractOwnerName(String columnTypeName, String columnName, int sqlType) { + if (columnTypeName != null && columnTypeName.contains(".")) { + return columnTypeName.substring(0, columnTypeName.lastIndexOf('.')); + } + if (sqlType == Types.STRUCT) { + String sql = "SELECT DATA_TYPE_OWNER FROM ALL_TAB_COLS WHERE COLUMN_NAME = ? AND DATA_TYPE = ?"; + try (PreparedStatement stmt = this.connection.prepareStatement(sql)) { + stmt.setString(1, columnName); + stmt.setString(2, columnTypeName); + try (ResultSet rs = stmt.executeQuery()) { + if (rs.next()) { + return rs.getString("DATA_TYPE_OWNER"); + } + } + } catch (SQLException e) { + throw new IllegalArgumentException(String.format("Column '%s' is not a primitive type," + + "but it lacks a type owner. Therefore, it cannot be resolved as " + + "a STRUCT type. ", columnName)); + } + } + return null; + } + + private Schema getSchemaMapping(ColumnMetadata metadata) throws SQLException { + int sqlType = metadata.getSqlType(); + int columnPrecision = metadata.getColumnPrecision(); + int columnScale = metadata.getColumnScale(); + String columnName = metadata.getColumnName(); + String columnTypeName = metadata.getColumnTypeName(); + Integer nestingLevel = metadata.getNestingLevel(); + Boolean isSigned = metadata.isSigned(); switch (sqlType) { case TIMESTAMP_TZ: @@ -95,7 +186,9 @@ public Schema getSchema(ResultSetMetaData metadata, int index) throws SQLExcepti case TIMESTAMP_LTZ: return getTimestampLtzSchema(); case Types.TIMESTAMP: - return isTimestampOldBehavior ? super.getSchema(metadata, index) : Schema.of(Schema.LogicalType.DATETIME); + return isTimestampOldBehavior ? super.getSchema(columnTypeName, sqlType, + columnPrecision, columnScale, columnName, isSigned) + : Schema.of(Schema.LogicalType.DATETIME); case BINARY_FLOAT: return Schema.of(Schema.Type.FLOAT); case BINARY_DOUBLE: @@ -109,15 +202,16 @@ public Schema getSchema(ResultSetMetaData metadata, int index) throws SQLExcepti return Schema.of(Schema.Type.STRING); case Types.SQLXML: // Enabling XML type support for DTS connectors only as it is not in working state in CDAP plugin. - return isXmlTypeEnabled ? Schema.of(Schema.Type.STRING) : super.getSchema(metadata, index); + return isXmlTypeEnabled ? Schema.of(Schema.Type.STRING) : super.getSchema(columnTypeName, + sqlType, columnPrecision, columnScale, columnName, isSigned); case Types.NUMERIC: case Types.DECIMAL: // FLOAT and REAL are returned as java.sql.Types.NUMERIC but with value that is a java.lang.Double - if (Double.class.getTypeName().equals(metadata.getColumnClassName(index))) { + if (Double.class.getTypeName().equals(metadata.getColumnClassName())) { return Schema.of(Schema.Type.DOUBLE); } else { - int precision = metadata.getPrecision(index); // total number of digits - int scale = metadata.getScale(index); // digits after the decimal point + int precision = columnPrecision; // total number of digits + int scale = columnScale; // digits after the decimal point // For a Number type without specified precision and scale, precision will be 0 and scale will be -127 if (precision == 0) { // reference : https://docs.oracle.com/cd/B28359_01/server.111/b28318/datatype.htm#CNCPT1832 @@ -128,23 +222,97 @@ public Schema getSchema(ResultSetMetaData metadata, int index) throws SQLExcepti + "there may be a precision loss while running the pipeline. " + "Please define an output precision and scale for field '%s' to avoid " + "precision loss.", - metadata.getColumnTypeName(index), - metadata.getColumnName(index))); + columnTypeName, columnName)); return Schema.decimalOf(precision, scale); } else { LOG.warn(String.format("Field '%s' is a %s type without precision and scale, " + "converting into STRING type to avoid any precision loss.", - metadata.getColumnName(index), - metadata.getColumnTypeName(index), - metadata.getColumnName(index))); + columnName, columnTypeName, columnName)); return Schema.of(Schema.Type.STRING); } } return Schema.decimalOf(precision, scale); } + case Types.STRUCT: + if (nestingLevel >= 4) { + throw new IllegalArgumentException(String.format("Cannot resolve STRUCT schema for attribute" + + " %s with nested structure depth more than 4.", columnName)); + } + return getStructSchema(connection, columnTypeName, metadata.getOwner(), + columnName, nestingLevel); default: - return super.getSchema(metadata, index); + return super.getSchema(columnTypeName, sqlType, columnPrecision, columnScale, + columnName, isSigned); + } + } + + @Override + public List getSchemaFields(ResultSet resultSet) throws SQLException { + this.connection = resultSet.getStatement().getConnection(); + return super.getSchemaFields(resultSet); + } + + /** + * Builds a CDAP RECORD schema for an Oracle STRUCT type by querying the + * database metadata + * for the type's attributes. + * + * @param connection the database connection + * @param typeName the Oracle type name (e.g., "ADDRESS_TYPE") + * @param owner the Owner of the user-defined data type + * @param recordName name of the UDT column + * @param level the level of nesting of the user-defined data type + * @return a CDAP RECORD schema with fields corresponding to the STRUCT's + * attributes + */ + private Schema getStructSchema(Connection connection, String typeName, String owner, String recordName, int level) + throws SQLException { + List fields = new ArrayList<>(); + String sql = "SELECT * FROM ALL_TYPE_ATTRS WHERE TYPE_NAME = ? AND OWNER = ? ORDER BY ATTR_NO"; + + try (PreparedStatement statement = connection.prepareStatement(sql)) { + statement.setString(1, typeName.substring(typeName.lastIndexOf('.') + 1)); + statement.setString(2, owner); + + try (ResultSet attributeResultSet = statement.executeQuery()) { + while (attributeResultSet.next()) { + String attributeName = attributeResultSet.getString(COLUMN_ATTR_NAME); + String attributeTypeName = attributeResultSet.getString(COLUMN_ATTR_TYPE_NAME); + int attributePrecision = attributeResultSet.getInt(COLUMN_PRECISION); + int attributeScale = attributeResultSet.getInt(COLUMN_SCALE); + Integer sqlType = DATA_TYPE_MAP.getOrDefault(attributeTypeName, null); + + int nextLevel = level; + String attributeOwner = owner; + if (sqlType == null) { + attributeOwner = attributeResultSet.getString(COLUMN_ATTR_TYPE_OWNER); + if (attributeOwner == null || attributeOwner.isEmpty()) { + throw new IllegalArgumentException(String.format("Attribute '%s' is not a primitive type," + + "but it lacks a type owner. Therefore, it cannot be resolved as " + + "a STRUCT type. ", attributeName)); + } + sqlType = Types.STRUCT; + nextLevel = level + 1; + } + ColumnMetadata columnMetadata = new ColumnMetadata(sqlType, null, attributePrecision, + attributeScale, attributeName, attributeTypeName, true, attributeOwner, nextLevel); + Schema attributeSchema = getSchemaMapping(columnMetadata); + if (!attributeSchema.isNullable()) { + attributeSchema = Schema.nullableOf(attributeSchema); + } + fields.add(Schema.Field.of(attributeName, attributeSchema)); + } + } + } + + if (fields.isEmpty()) { + throw new IllegalArgumentException(String.format( + "No attributes found for Oracle STRUCT type '%s'. " + + "Ensure the type exists and is accessible.", + typeName)); } + + return Schema.recordOf(recordName, fields); } private Schema getTimestampLtzSchema() { @@ -161,4 +329,69 @@ public boolean shouldIgnoreColumn(ResultSetMetaData metadata, int index) throws return metadata.getColumnName(index).equals("c_" + sessionID) || metadata.getColumnName(index).equals("s_" + sessionID); } + + /** + * Helper class encapsulating column metadata parameters for schema mapping. + */ + public static class ColumnMetadata { + private final int sqlType; + private final String columnClassName; + private final int columnPrecision; + private final int columnScale; + private final String columnName; + private final String columnTypeName; + private final boolean isSigned; + private final String owner; + private final int nestingLevel; + + public ColumnMetadata(int sqlType, String columnClassName, int columnPrecision, + int columnScale, String columnName, String columnTypeName, + boolean isSigned, String owner, int nestingLevel) { + this.sqlType = sqlType; + this.columnClassName = columnClassName; + this.columnPrecision = columnPrecision; + this.columnScale = columnScale; + this.columnName = columnName; + this.columnTypeName = columnTypeName; + this.isSigned = isSigned; + this.owner = owner; + this.nestingLevel = nestingLevel; + } + + public int getSqlType() { + return sqlType; + } + + public String getColumnClassName() { + return columnClassName; + } + + public int getColumnPrecision() { + return columnPrecision; + } + + public int getColumnScale() { + return columnScale; + } + + public String getColumnName() { + return columnName; + } + + public String getColumnTypeName() { + return columnTypeName; + } + + public boolean isSigned() { + return isSigned; + } + + public String getOwner() { + return owner; + } + + public int getNestingLevel() { + return nestingLevel; + } + } } diff --git a/oracle-plugin/src/main/java/io/cdap/plugin/oracle/OracleStructUtil.java b/oracle-plugin/src/main/java/io/cdap/plugin/oracle/OracleStructUtil.java new file mode 100644 index 000000000..66cafbc11 --- /dev/null +++ b/oracle-plugin/src/main/java/io/cdap/plugin/oracle/OracleStructUtil.java @@ -0,0 +1,174 @@ +/* + * Copyright © 2024 Cask Data, Inc. + * + * Licensed under the Apache License, Version 2.0 (the "License"); you may not + * use this file except in compliance with the License. You may obtain a copy of + * the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT + * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the + * License for the specific language governing permissions and limitations under + * the License. + */ + +package io.cdap.plugin.oracle; + +import io.cdap.cdap.api.data.format.StructuredRecord; +import io.cdap.cdap.api.data.schema.Schema; + +import java.math.BigDecimal; +import java.math.RoundingMode; +import java.sql.Blob; +import java.sql.Clob; +import java.sql.Connection; +import java.sql.SQLException; +import java.sql.SQLXML; +import java.sql.Struct; +import java.sql.Timestamp; +import java.time.LocalDateTime; +import java.time.OffsetDateTime; +import java.time.ZoneId; +import java.time.ZonedDateTime; + +/** + * Utility class to convert and map Oracle STRUCT attribute values into CDAP {@link StructuredRecord} fields. + */ +public class OracleStructUtil { + + private OracleStructUtil() { + } + + /** + * Converts a raw Oracle STRUCT attribute value and populates it into the record builder. + * Falls back to standard DBRecord field population for non-Oracle specific types. + */ + public static void populateRecordField(OracleSourceDBRecord sourceRecord, Connection connection, + StructuredRecord.Builder recordBuilder, Schema.Field field, + Object attrValue) throws SQLException { + if (attrValue == null) { + recordBuilder.set(field.getName(), null); + return; + } + + Schema fieldSchema = field.getSchema().isNullable() ? field.getSchema().getNonNullable() + : field.getSchema(); + String attrClassName = attrValue.getClass().getName(); + if (attrValue instanceof Struct) { + recordBuilder.set(field.getName(), sourceRecord.convertStructToRecord((Struct) attrValue, + fieldSchema, connection)); + return; + } + if (attrValue instanceof Clob) { + Clob clob = (Clob) attrValue; + recordBuilder.set(field.getName(), clob.getSubString(1, (int) clob.length())); + return; + } + if (attrValue instanceof Blob) { + Blob blob = (Blob) attrValue; + recordBuilder.set(field.getName(), blob.getBytes(1, (int) blob.length())); + return; + } + if (attrValue instanceof SQLXML) { + recordBuilder.set(field.getName(), ((SQLXML) attrValue).getString()); + return; + } + if ("oracle.sql.INTERVALDS".equals(attrClassName) || "oracle.sql.INTERVALYM".equals(attrClassName)) { + recordBuilder.set(field.getName(), attrValue.toString()); + return; + } + if (attrValue instanceof BigDecimal) { + handleDecimalValue((BigDecimal) attrValue, fieldSchema, recordBuilder, field); + return; + } + if (attrValue instanceof Timestamp) { + handleTimestampValue(sourceRecord, (Timestamp) attrValue, fieldSchema, recordBuilder, + field, connection); + return; + } + if (attrValue instanceof OffsetDateTime) { + handleOffsetDateTimeValue((OffsetDateTime) attrValue, fieldSchema, recordBuilder, field); + return; + } + if (isBfileValue(attrValue, field.getName())) { + recordBuilder.set(field.getName(), sourceRecord.getBfileBytes(attrValue, field.getName())); + return; + } + + sourceRecord.populateRecordField(connection, recordBuilder, field, attrValue); + } + + private static void handleTimestampValue(OracleSourceDBRecord sourceRecord, Timestamp timestamp, Schema fieldSchema, + StructuredRecord.Builder recordBuilder, Schema.Field field, + Connection connection) throws SQLException { + if (Schema.LogicalType.DATETIME.equals(fieldSchema.getLogicalType())) { + recordBuilder.setDateTime(field.getName(), timestamp.toLocalDateTime()); + } else if (Schema.LogicalType.DATE.equals(fieldSchema.getLogicalType())) { + recordBuilder.setDate(field.getName(), timestamp.toLocalDateTime().toLocalDate()); + } else if (fieldSchema.getType() == Schema.Type.STRING) { + recordBuilder.set(field.getName(), timestamp.toString()); + } else { + sourceRecord.populateRecordField(connection, recordBuilder, field, timestamp); + } + } + + private static void handleOffsetDateTimeValue(OffsetDateTime offsetDateTime, Schema fieldSchema, + StructuredRecord.Builder recordBuilder, Schema.Field field) { + ZonedDateTime zonedDateTime = offsetDateTime.atZoneSameInstant(ZoneId.of("UTC")); + if (fieldSchema.getLogicalType() != null && + (Schema.LogicalType.TIMESTAMP_MICROS.equals(fieldSchema.getLogicalType()) || + Schema.LogicalType.TIMESTAMP_MILLIS.equals(fieldSchema.getLogicalType()))) { + recordBuilder.setTimestamp(field.getName(), zonedDateTime); + } else if (Schema.LogicalType.DATETIME.equals(fieldSchema.getLogicalType())) { + recordBuilder.setDateTime(field.getName(), offsetDateTime.toLocalDateTime()); + } else { + recordBuilder.set(field.getName(), zonedDateTime.toString()); + } + } + + /** + * Checks if the given attribute object is an instance of Oracle BFILE via reflection. + */ + private static boolean isBfileValue(Object attrValue, String fieldName) throws SQLException { + ClassLoader oracleLoader = attrValue.getClass().getClassLoader(); + try { + if (oracleLoader != null && oracleLoader.loadClass("oracle.jdbc.OracleBfile").isInstance(attrValue)) { + return true; + } + } catch (ClassNotFoundException e) { + throw new SQLException(String.format("Column '%s' is of type 'BFILE', which is not supported with " + + "this version of the JDBC driver.", fieldName), e); + } + return false; + } + + private static void handleDecimalValue(BigDecimal bigDecimal, Schema fieldSchema, + StructuredRecord.Builder recordBuilder, Schema.Field field) { + if (Schema.LogicalType.DECIMAL.equals(fieldSchema.getLogicalType())) { + recordBuilder.setDecimal(field.getName(), bigDecimal.setScale(fieldSchema.getScale(), + RoundingMode.HALF_UP)); + return; + } + switch (fieldSchema.getType()) { + case DOUBLE: + recordBuilder.set(field.getName(), bigDecimal.doubleValue()); + break; + case FLOAT: + recordBuilder.set(field.getName(), bigDecimal.floatValue()); + break; + case INT: + recordBuilder.set(field.getName(), bigDecimal.intValue()); + break; + case LONG: + recordBuilder.set(field.getName(), bigDecimal.longValue()); + break; + case STRING: + recordBuilder.set(field.getName(), bigDecimal.toPlainString()); + break; + default: + recordBuilder.set(field.getName(), bigDecimal); + } + } +} diff --git a/oracle-plugin/src/test/java/io/cdap/plugin/oracle/OracleSchemaReaderTest.java b/oracle-plugin/src/test/java/io/cdap/plugin/oracle/OracleSchemaReaderTest.java index 586ca1141..1c61500d5 100644 --- a/oracle-plugin/src/test/java/io/cdap/plugin/oracle/OracleSchemaReaderTest.java +++ b/oracle-plugin/src/test/java/io/cdap/plugin/oracle/OracleSchemaReaderTest.java @@ -20,33 +20,71 @@ import io.cdap.cdap.api.data.schema.Schema; import io.cdap.cdap.api.exception.ProgramFailureException; import org.junit.Assert; +import org.junit.Before; import org.junit.Test; -import org.junit.runner.RunWith; import org.mockito.Mockito; -import org.mockito.junit.MockitoJUnitRunner; +import java.sql.Connection; +import java.sql.PreparedStatement; import java.sql.ResultSet; import java.sql.ResultSetMetaData; import java.sql.SQLException; +import java.sql.SQLTimeoutException; +import java.sql.Statement; import java.sql.Types; +import java.util.Arrays; import java.util.List; public class OracleSchemaReaderTest { + private ResultSet resultSet; + private ResultSetMetaData metadata; + private Statement statement; + private Connection connection; + private PreparedStatement preparedStatement; + private ResultSet attributeResultSet; + private OracleSourceSchemaReader defaultSchemaReader; + + @Before + public void setUp() throws SQLException { + resultSet = Mockito.mock(ResultSet.class); + metadata = Mockito.mock(ResultSetMetaData.class); + statement = Mockito.mock(Statement.class); + connection = Mockito.mock(Connection.class); + preparedStatement = Mockito.mock(PreparedStatement.class); + attributeResultSet = Mockito.mock(ResultSet.class); + defaultSchemaReader = new OracleSourceSchemaReader(); + + Mockito.when(resultSet.getMetaData()).thenReturn(metadata); + Mockito.when(resultSet.getStatement()).thenReturn(statement); + Mockito.when(statement.getConnection()).thenReturn(connection); + Mockito.when(connection.prepareStatement(Mockito.anyString())).thenReturn(preparedStatement); + Mockito.when(preparedStatement.executeQuery()).thenReturn(attributeResultSet); + } + + private void mockSingleColumn(int sqlType, String columnName, String columnTypeName) throws SQLException { + Mockito.when(metadata.getColumnCount()).thenReturn(1); + Mockito.when(metadata.getColumnType(1)).thenReturn(sqlType); + Mockito.when(metadata.getColumnName(1)).thenReturn(columnName); + Mockito.when(metadata.getColumnTypeName(1)).thenReturn(columnTypeName); + } + @Test public void getSchema_timestampLTZFieldTrue_returnTimestamp() throws SQLException { OracleSourceSchemaReader schemaReader = new OracleSourceSchemaReader(null, false, false, true, false); ResultSet resultSet = Mockito.mock(ResultSet.class); ResultSetMetaData metadata = Mockito.mock(ResultSetMetaData.class); - + Statement statement = Mockito.mock(Statement.class); + Connection connection = Mockito.mock(Connection.class); + Mockito.when(resultSet.getMetaData()).thenReturn(metadata); + Mockito.when(resultSet.getStatement()).thenReturn(statement); + Mockito.when(statement.getConnection()).thenReturn(connection); Mockito.when(resultSet.getMetaData()).thenReturn(metadata); - Mockito.when(metadata.getColumnCount()).thenReturn(2); // -101 is for TIMESTAMP_TZ Mockito.when(metadata.getColumnType(1)).thenReturn(-101); Mockito.when(metadata.getColumnName(1)).thenReturn("column1"); - // -102 is for TIMESTAMP_LTZ Mockito.when(metadata.getColumnType(2)).thenReturn(-102); Mockito.when(metadata.getColumnName(2)).thenReturn("column2"); @@ -54,30 +92,29 @@ public void getSchema_timestampLTZFieldTrue_returnTimestamp() throws SQLExceptio List expectedSchemaFields = Lists.newArrayList(); expectedSchemaFields.add(Schema.Field.of("column1", Schema.of(Schema.LogicalType.TIMESTAMP_MICROS))); expectedSchemaFields.add(Schema.Field.of("column2", Schema.of(Schema.LogicalType.TIMESTAMP_MICROS))); - List actualSchemaFields = schemaReader.getSchemaFields(resultSet); Assert.assertEquals(expectedSchemaFields.get(0).getName(), actualSchemaFields.get(0).getName()); Assert.assertEquals(expectedSchemaFields.get(0).getSchema(), actualSchemaFields.get(0).getSchema()); Assert.assertEquals(expectedSchemaFields.get(1).getName(), actualSchemaFields.get(1).getName()); Assert.assertEquals(expectedSchemaFields.get(1).getSchema(), actualSchemaFields.get(1).getSchema()); - } @Test public void getSchema_timestampLTZFieldFalse_returnDatetime() throws SQLException { - OracleSourceSchemaReader schemaReader = new OracleSourceSchemaReader(null, false, false, false, false); - + OracleSourceSchemaReader schemaReader = new OracleSourceSchemaReader(null, false, + false, false, false); ResultSet resultSet = Mockito.mock(ResultSet.class); ResultSetMetaData metadata = Mockito.mock(ResultSetMetaData.class); - + Statement statement = Mockito.mock(Statement.class); + Connection connection = Mockito.mock(Connection.class); Mockito.when(resultSet.getMetaData()).thenReturn(metadata); - + Mockito.when(resultSet.getStatement()).thenReturn(statement); + Mockito.when(statement.getConnection()).thenReturn(connection); Mockito.when(metadata.getColumnCount()).thenReturn(2); // -101 is for TIMESTAMP_TZ Mockito.when(metadata.getColumnType(1)).thenReturn(-101); Mockito.when(metadata.getColumnName(1)).thenReturn("column1"); - // -102 is for TIMESTAMP_LTZ Mockito.when(metadata.getColumnType(2)).thenReturn(-102); Mockito.when(metadata.getColumnName(2)).thenReturn("column2"); @@ -85,7 +122,6 @@ public void getSchema_timestampLTZFieldFalse_returnDatetime() throws SQLExceptio List expectedSchemaFields = Lists.newArrayList(); expectedSchemaFields.add(Schema.Field.of("column1", Schema.of(Schema.LogicalType.TIMESTAMP_MICROS))); expectedSchemaFields.add(Schema.Field.of("column2", Schema.of(Schema.LogicalType.DATETIME))); - List actualSchemaFields = schemaReader.getSchemaFields(resultSet); Assert.assertEquals(expectedSchemaFields.get(0).getName(), actualSchemaFields.get(0).getName()); @@ -94,15 +130,104 @@ public void getSchema_timestampLTZFieldFalse_returnDatetime() throws SQLExceptio Assert.assertEquals(expectedSchemaFields.get(1).getSchema(), actualSchemaFields.get(1).getSchema()); } + @Test + public void getSchemaFields_structType_returnsRecord() throws SQLException { + mockSingleColumn(Types.STRUCT, "address", "CS_ITN.ADDRESS_TYPE"); + Mockito.when(metadata.getSchemaName(1)).thenReturn("TEST_SCHEMA"); + Boolean[] nextReturns = new Boolean[27]; + Arrays.fill(nextReturns, 0, 26, true); + nextReturns[26] = false; + Mockito.when(attributeResultSet.next()).thenReturn(true, nextReturns); + Mockito.when(attributeResultSet.getString("ATTR_NAME")).thenReturn( + "ATTR_VARCHAR2", "ATTR_VARCHAR", "ATTR_CHAR", "ATTR_CHAR2", "ATTR_NCHAR", + "ATTR_NVARCHAR2", "ATTR_CLOB", "ATTR_NCLOB", + "ATTR_UROWID", "ATTR_NUMBER_PREC", "ATTR_NUMBER_NOPREC", "ATTR_DECIMAL", + "ATTR_INTEGER", "ATTR_FLOAT", "ATTR_REAL", "ATTR_DOUBLE", "ATTR_BINARY_FLOAT", + "ATTR_BINARY_DOUBLE", "ATTR_DATE", "ATTR_TIMESTAMP", "ATTR_TIMESTAMP_TZ", + "ATTR_TIMESTAMP_LTZ", "ATTR_INTERVAL_DS", "ATTR_INTERVAL_YM", "ATTR_BLOB", + "ATTR_RAW", "ATTR_BFILE" + ); + Mockito.when(attributeResultSet.getString("ATTR_TYPE_NAME")).thenReturn( + "VARCHAR2", "VARCHAR", "CHAR", "CHAR2", "NCHAR", + "NVARCHAR2", "CLOB", "NCLOB", + "UROWID", "NUMBER", "NUMBER", "DECIMAL", + "INTEGER", "FLOAT", "REAL", "DOUBLE", "BINARY_FLOAT", + "BINARY_DOUBLE", "DATE", "TIMESTAMP", "TIMESTAMP WITH TZ", + "TIMESTAMP WITH LOCAL TZ", "INTERVAL DAY TO SECOND", "INTERVAL YEAR TO MONTH", "BLOB", + "RAW", "BFILE" + ); + Mockito.when(attributeResultSet.getInt("PRECISION")).thenReturn( + 50, 50, 10, 10, 10, + 50, 0, 0, + 0, 10, 0, 8, + 10, 10, 10, 10, 0, + 0, 0, 0, 0, + 0, 0, 0, 0, + 100, 0 + ); + Mockito.when(attributeResultSet.getInt("SCALE")).thenReturn( + 0, 0, 0, 0, 0, + 0, 0, 0, + 0, 2, 0, 2, + 0, 0, 0, 0, 0, + 0, 0, 0, 0, + 0, 0 + ); + + List actualFields = defaultSchemaReader.getSchemaFields(resultSet); + + Schema.Field addressField = actualFields.get(0); + Schema addressSchema = addressField.getSchema().isNullable() + ? addressField.getSchema().getNonNullable() : addressField.getSchema(); + List structFields = addressSchema.getFields(); + Assert.assertEquals(1, actualFields.size()); + Assert.assertEquals("address", addressField.getName()); + Assert.assertEquals(Schema.Type.RECORD, addressSchema.getType()); + Assert.assertEquals("address", addressSchema.getRecordName()); + Assert.assertEquals(27, structFields.size()); + Assert.assertEquals(Schema.nullableOf(Schema.of(Schema.Type.STRING)), structFields.get(0).getSchema()); + Assert.assertEquals(Schema.nullableOf(Schema.of(Schema.Type.STRING)), structFields.get(1).getSchema()); + Assert.assertEquals(Schema.nullableOf(Schema.of(Schema.Type.STRING)), structFields.get(2).getSchema()); + Assert.assertEquals(Schema.nullableOf(Schema.of(Schema.Type.STRING)), structFields.get(3).getSchema()); + Assert.assertEquals(Schema.nullableOf(Schema.of(Schema.Type.STRING)), structFields.get(4).getSchema()); + Assert.assertEquals(Schema.nullableOf(Schema.of(Schema.Type.STRING)), structFields.get(5).getSchema()); + Assert.assertEquals(Schema.nullableOf(Schema.of(Schema.Type.STRING)), structFields.get(6).getSchema()); + Assert.assertEquals(Schema.nullableOf(Schema.of(Schema.Type.STRING)), structFields.get(7).getSchema()); + Assert.assertEquals(Schema.nullableOf(Schema.of(Schema.Type.STRING)), structFields.get(8).getSchema()); + Assert.assertEquals(Schema.nullableOf(Schema.decimalOf(10, 2)), structFields.get(9).getSchema()); + Assert.assertEquals(Schema.nullableOf(Schema.of(Schema.Type.STRING)), structFields.get(10).getSchema()); + Assert.assertEquals(Schema.nullableOf(Schema.decimalOf(8, 2)), structFields.get(11).getSchema()); + Assert.assertEquals(Schema.nullableOf(Schema.decimalOf(10, 0)), structFields.get(12).getSchema()); + Assert.assertEquals(Schema.nullableOf(Schema.of(Schema.Type.DOUBLE)), structFields.get(13).getSchema()); + Assert.assertEquals(Schema.nullableOf(Schema.of(Schema.Type.DOUBLE)), structFields.get(14).getSchema()); + Assert.assertEquals(Schema.nullableOf(Schema.of(Schema.Type.DOUBLE)), structFields.get(15).getSchema()); + Assert.assertEquals(Schema.nullableOf(Schema.of(Schema.Type.FLOAT)), structFields.get(16).getSchema()); + Assert.assertEquals(Schema.nullableOf(Schema.of(Schema.Type.DOUBLE)), structFields.get(17).getSchema()); + Assert.assertEquals(Schema.nullableOf(Schema.of(Schema.LogicalType.DATETIME)), structFields.get(18).getSchema()); + Assert.assertEquals(Schema.nullableOf(Schema.of(Schema.LogicalType.DATETIME)), structFields.get(19).getSchema()); + Assert.assertEquals(Schema.nullableOf(Schema.of(Schema.LogicalType.TIMESTAMP_MICROS)), + structFields.get(20).getSchema()); + Assert.assertEquals(Schema.nullableOf(Schema.of(Schema.LogicalType.DATETIME)), structFields.get(21).getSchema()); + Assert.assertEquals(Schema.nullableOf(Schema.of(Schema.Type.STRING)), structFields.get(22).getSchema()); + Assert.assertEquals(Schema.nullableOf(Schema.of(Schema.Type.STRING)), structFields.get(23).getSchema()); + Assert.assertEquals(Schema.nullableOf(Schema.of(Schema.Type.BYTES)), structFields.get(24).getSchema()); + Assert.assertEquals(Schema.nullableOf(Schema.of(Schema.Type.BYTES)), structFields.get(25).getSchema()); + Assert.assertEquals(Schema.nullableOf(Schema.of(Schema.Type.BYTES)), structFields.get(26).getSchema()); + } + @Test public void getSchema_xmlField_returnString() throws SQLException { OracleSourceSchemaReader schemaReader = new OracleSourceSchemaReader(null, false, false, false, true); ResultSet resultSet = Mockito.mock(ResultSet.class); ResultSetMetaData metadata = Mockito.mock(ResultSetMetaData.class); + Connection connection = Mockito.mock(Connection.class); + Statement statement = Mockito.mock(Statement.class); Mockito.when(resultSet.getMetaData()).thenReturn(metadata); Mockito.when(metadata.getColumnCount()).thenReturn(1); Mockito.when(metadata.getColumnType(1)).thenReturn(Types.SQLXML); Mockito.when(metadata.getColumnName(1)).thenReturn("xmlData"); + Mockito.when(resultSet.getStatement()).thenReturn(statement); + Mockito.when(statement.getConnection()).thenReturn(connection); List actualSchemaFields = schemaReader.getSchemaFields(resultSet); @@ -118,12 +243,346 @@ public void getSchema_xmlFieldDisabled_throwsProgramFailureException() throws SQ false, false, false, false); ResultSet resultSet = Mockito.mock(ResultSet.class); ResultSetMetaData metadata = Mockito.mock(ResultSetMetaData.class); + Connection connection = Mockito.mock(Connection.class); + Statement statement = Mockito.mock(Statement.class); Mockito.when(resultSet.getMetaData()).thenReturn(metadata); Mockito.when(metadata.getColumnCount()).thenReturn(1); Mockito.when(metadata.getColumnType(1)).thenReturn(Types.SQLXML); Mockito.when(metadata.getColumnName(1)).thenReturn("xmlData"); + Mockito.when(resultSet.getStatement()).thenReturn(statement); + Mockito.when(statement.getConnection()).thenReturn(connection); Assert.assertThrows(ProgramFailureException.class, () -> schemaReader.getSchemaFields(resultSet)); + } + + @Test + public void getSchemaFields_structUnsupportedType_throwsException() throws SQLException { + mockSingleColumn(Types.STRUCT, "complex_payload", "CS_ITN.ANYDATA_TYPE"); + Mockito.when(metadata.getSchemaName(1)).thenReturn("TEST_SCHEMA"); + Mockito.when(attributeResultSet.next()).thenReturn(true, true, false); + Mockito.when(attributeResultSet.getString("ATTR_NAME")) + .thenReturn("VALID_ID", "UNSUPPORTED_DATA"); + Mockito.when(attributeResultSet.getString("ATTR_TYPE_NAME")) + .thenReturn("NUMBER", "ANYDATA"); + Mockito.when(attributeResultSet.getInt("PRECISION")).thenReturn(10, 0); + Mockito.when(attributeResultSet.getInt("SCALE")).thenReturn(0, 0); + + Assert.assertThrows(ProgramFailureException.class, () -> defaultSchemaReader.getSchemaFields(resultSet)); + } + + @Test + public void getSchemaFields_nestedStructLevel_returnsRecord() throws SQLException { + PreparedStatement stmt0 = Mockito.mock(PreparedStatement.class); + PreparedStatement stmt1 = Mockito.mock(PreparedStatement.class); + PreparedStatement stmt2 = Mockito.mock(PreparedStatement.class); + PreparedStatement stmt3 = Mockito.mock(PreparedStatement.class); + ResultSet attrRs0 = Mockito.mock(ResultSet.class); + ResultSet attrRs1 = Mockito.mock(ResultSet.class); + ResultSet attrRs2 = Mockito.mock(ResultSet.class); + ResultSet attrRs3 = Mockito.mock(ResultSet.class); + Mockito.when(connection.prepareStatement(Mockito.anyString())).thenReturn(stmt0, stmt1, stmt2, stmt3); + Mockito.when(stmt0.executeQuery()).thenReturn(attrRs0); + Mockito.when(stmt1.executeQuery()).thenReturn(attrRs1); + Mockito.when(stmt2.executeQuery()).thenReturn(attrRs2); + Mockito.when(stmt3.executeQuery()).thenReturn(attrRs3); + + mockSingleColumn(Types.STRUCT, "payload", "TEST.STRUCT_L0"); + Mockito.when(metadata.getSchemaName(1)).thenReturn("TEST"); + Mockito.when(attrRs0.next()).thenReturn(true, false); + Mockito.when(attrRs0.getString("ATTR_NAME")).thenReturn("SUB1"); + Mockito.when(attrRs0.getString("ATTR_TYPE_NAME")).thenReturn("STRUCT_L1"); + Mockito.when(attrRs0.getString("ATTR_TYPE_OWNER")).thenReturn("TEST"); + Mockito.when(attrRs1.next()).thenReturn(true, false); + Mockito.when(attrRs1.getString("ATTR_NAME")).thenReturn("SUB2"); + Mockito.when(attrRs1.getString("ATTR_TYPE_NAME")).thenReturn("STRUCT_L2"); + Mockito.when(attrRs1.getString("ATTR_TYPE_OWNER")).thenReturn("TEST"); + Mockito.when(attrRs2.next()).thenReturn(true, false); + Mockito.when(attrRs2.getString("ATTR_NAME")).thenReturn("SUB3"); + Mockito.when(attrRs2.getString("ATTR_TYPE_NAME")).thenReturn("STRUCT_L3"); + Mockito.when(attrRs2.getString("ATTR_TYPE_OWNER")).thenReturn("TEST"); + Mockito.when(attrRs3.next()).thenReturn(true, false); + Mockito.when(attrRs3.getString("ATTR_NAME")).thenReturn("ID"); + Mockito.when(attrRs3.getString("ATTR_TYPE_NAME")).thenReturn("VARCHAR2"); + Mockito.when(attrRs3.getInt("PRECISION")).thenReturn(50); + Mockito.when(attrRs3.getInt("SCALE")).thenReturn(0); + + List actualFields = defaultSchemaReader.getSchemaFields(resultSet); + + Assert.assertEquals(1, actualFields.size()); + Schema l0Schema = actualFields.get(0).getSchema().isNullable() + ? actualFields.get(0).getSchema().getNonNullable() : actualFields.get(0).getSchema(); + Assert.assertEquals(Schema.Type.RECORD, l0Schema.getType()); + Schema l1Schema = l0Schema.getField("SUB1").getSchema().isNullable() + ? l0Schema.getField("SUB1").getSchema().getNonNullable() + : l0Schema.getField("SUB1").getSchema(); + Assert.assertEquals(Schema.Type.RECORD, l1Schema.getType()); + Schema l2Schema = l1Schema.getField("SUB2").getSchema().isNullable() + ? l1Schema.getField("SUB2").getSchema().getNonNullable() + : l1Schema.getField("SUB2").getSchema(); + Assert.assertEquals(Schema.Type.RECORD, l2Schema.getType()); + Schema l3Schema = l2Schema.getField("SUB3").getSchema().isNullable() + ? l2Schema.getField("SUB3").getSchema().getNonNullable() + : l2Schema.getField("SUB3").getSchema(); + Schema idSchema = l3Schema.getField("ID").getSchema().isNullable() + ? l3Schema.getField("ID").getSchema().getNonNullable() + : l3Schema.getField("ID").getSchema(); + Assert.assertEquals(Schema.Type.STRING, idSchema.getType()); + } + + @Test + public void getSchemaFields_exceedsNestedStructLevel_throwsException() throws SQLException { + PreparedStatement stmt0 = Mockito.mock(PreparedStatement.class); + PreparedStatement stmt1 = Mockito.mock(PreparedStatement.class); + PreparedStatement stmt2 = Mockito.mock(PreparedStatement.class); + PreparedStatement stmt3 = Mockito.mock(PreparedStatement.class); + ResultSet attrRs0 = Mockito.mock(ResultSet.class); + ResultSet attrRs1 = Mockito.mock(ResultSet.class); + ResultSet attrRs2 = Mockito.mock(ResultSet.class); + ResultSet attrRs3 = Mockito.mock(ResultSet.class); + Mockito.when(connection.prepareStatement(Mockito.anyString())).thenReturn(stmt0, stmt1, stmt2, stmt3); + Mockito.when(stmt0.executeQuery()).thenReturn(attrRs0); + Mockito.when(stmt1.executeQuery()).thenReturn(attrRs1); + Mockito.when(stmt2.executeQuery()).thenReturn(attrRs2); + Mockito.when(stmt3.executeQuery()).thenReturn(attrRs3); + mockSingleColumn(Types.STRUCT, "payload", "TEST.STRUCT_L0"); + Mockito.when(metadata.getSchemaName(1)).thenReturn("TEST"); + Mockito.when(attrRs0.next()).thenReturn(true, false); + Mockito.when(attrRs0.getString("ATTR_NAME")).thenReturn("SUB1"); + Mockito.when(attrRs0.getString("ATTR_TYPE_NAME")).thenReturn("STRUCT_L1"); + Mockito.when(attrRs0.getString("ATTR_TYPE_OWNER")).thenReturn("TEST"); + Mockito.when(attrRs1.next()).thenReturn(true, false); + Mockito.when(attrRs1.getString("ATTR_NAME")).thenReturn("SUB2"); + Mockito.when(attrRs1.getString("ATTR_TYPE_NAME")).thenReturn("STRUCT_L2"); + Mockito.when(attrRs1.getString("ATTR_TYPE_OWNER")).thenReturn("TEST"); + Mockito.when(attrRs2.next()).thenReturn(true, false); + Mockito.when(attrRs2.getString("ATTR_NAME")).thenReturn("SUB3"); + Mockito.when(attrRs2.getString("ATTR_TYPE_NAME")).thenReturn("STRUCT_L3"); + Mockito.when(attrRs2.getString("ATTR_TYPE_OWNER")).thenReturn("TEST"); + Mockito.when(attrRs3.next()).thenReturn(true, false); + Mockito.when(attrRs3.getString("ATTR_NAME")).thenReturn("SUB4"); + Mockito.when(attrRs3.getString("ATTR_TYPE_NAME")).thenReturn("STRUCT_L4"); + Mockito.when(attrRs3.getString("ATTR_TYPE_OWNER")).thenReturn("TEST"); + + Assert.assertThrows(IllegalArgumentException.class, () -> defaultSchemaReader.getSchemaFields(resultSet)); + } + + @Test + public void getSchemaFields_multipleStructColumns_returnsRecord() throws SQLException { + ResultSet attrRs1 = Mockito.mock(ResultSet.class); + ResultSet attrRs2 = Mockito.mock(ResultSet.class); + Mockito.when(preparedStatement.executeQuery()).thenReturn(attrRs1, attrRs2); + Mockito.when(metadata.getColumnCount()).thenReturn(2); + Mockito.when(metadata.getColumnType(1)).thenReturn(Types.STRUCT); + Mockito.when(metadata.getColumnName(1)).thenReturn("address1"); + Mockito.when(metadata.getColumnTypeName(1)).thenReturn("CS_ITN.ADDRESS_TYPE"); + Mockito.when(metadata.getSchemaName(1)).thenReturn("TEST_SCHEMA"); + Mockito.when(metadata.getColumnType(2)).thenReturn(Types.STRUCT); + Mockito.when(metadata.getColumnName(2)).thenReturn("address2"); + Mockito.when(metadata.getColumnTypeName(2)).thenReturn("CS_ITN.ADDRESS_TYPE"); + Mockito.when(metadata.getSchemaName(2)).thenReturn("TEST_SCHEMA"); + Mockito.when(attrRs1.next()).thenReturn(true, false); + Mockito.when(attrRs1.getString("ATTR_NAME")).thenReturn("STREET"); + Mockito.when(attrRs1.getString("ATTR_TYPE_NAME")).thenReturn("VARCHAR2"); + Mockito.when(attrRs1.getInt("PRECISION")).thenReturn(50); + Mockito.when(attrRs1.getInt("SCALE")).thenReturn(0); + Mockito.when(attrRs2.next()).thenReturn(true, false); + Mockito.when(attrRs2.getString("ATTR_NAME")).thenReturn("STREET"); + Mockito.when(attrRs2.getString("ATTR_TYPE_NAME")).thenReturn("VARCHAR2"); + Mockito.when(attrRs2.getInt("PRECISION")).thenReturn(50); + Mockito.when(attrRs2.getInt("SCALE")).thenReturn(0); + + List actualFields = defaultSchemaReader.getSchemaFields(resultSet); + + Assert.assertEquals(2, actualFields.size()); + Assert.assertEquals("address1", actualFields.get(0).getName()); + Schema address1Schema = actualFields.get(0).getSchema().isNullable() + ? actualFields.get(0).getSchema().getNonNullable() : actualFields.get(0).getSchema(); + Assert.assertEquals(Schema.Type.RECORD, address1Schema.getType()); + Assert.assertEquals("address2", actualFields.get(1).getName()); + Schema address2Schema = actualFields.get(1).getSchema().isNullable() + ? actualFields.get(1).getSchema().getNonNullable() : actualFields.get(1).getSchema(); + Assert.assertEquals(Schema.Type.RECORD, address2Schema.getType()); + } + + @Test + public void getSchemaFields_unqualifiedStructType_returnsRecord() throws SQLException { + PreparedStatement ownerStmt = Mockito.mock(PreparedStatement.class); + PreparedStatement attrStmt = Mockito.mock(PreparedStatement.class); + ResultSet ownerRs = Mockito.mock(ResultSet.class); + Mockito.when(connection.prepareStatement(Mockito.anyString())).thenReturn(ownerStmt, attrStmt); + Mockito.when(ownerStmt.executeQuery()).thenReturn(ownerRs); + Mockito.when(attrStmt.executeQuery()).thenReturn(attributeResultSet); + mockSingleColumn(Types.STRUCT, "address", "ADDRESS_TYPE"); + Mockito.when(ownerRs.next()).thenReturn(true); + Mockito.when(ownerRs.getString("DATA_TYPE_OWNER")).thenReturn("CS_ITN"); + Mockito.when(attributeResultSet.next()).thenReturn(true, false); + Mockito.when(attributeResultSet.getString("ATTR_NAME")).thenReturn("STREET"); + Mockito.when(attributeResultSet.getString("ATTR_TYPE_NAME")).thenReturn("VARCHAR2"); + Mockito.when(attributeResultSet.getInt("PRECISION")).thenReturn(50); + Mockito.when(attributeResultSet.getInt("SCALE")).thenReturn(0); + + List actualFields = defaultSchemaReader.getSchemaFields(resultSet); + + Assert.assertEquals(1, actualFields.size()); + Assert.assertEquals("address", actualFields.get(0).getName()); + } + + @Test + public void getSchemaFields_unqualifiedStructOwnerNotFound_throwsException() throws SQLException { + PreparedStatement ownerStmt = Mockito.mock(PreparedStatement.class); + PreparedStatement attrStmt = Mockito.mock(PreparedStatement.class); + ResultSet ownerRs = Mockito.mock(ResultSet.class); + Mockito.when(connection.prepareStatement(Mockito.anyString())).thenReturn(ownerStmt, attrStmt); + Mockito.when(ownerStmt.executeQuery()).thenReturn(ownerRs); + Mockito.when(attrStmt.executeQuery()).thenReturn(attributeResultSet); + mockSingleColumn(Types.STRUCT, "address", "ADDRESS_TYPE"); + Mockito.when(ownerRs.next()).thenReturn(false); + Mockito.when(attributeResultSet.next()).thenReturn(false); + + Assert.assertThrows(IllegalArgumentException.class, () -> defaultSchemaReader.getSchemaFields(resultSet)); + } + + @Test + public void getSchemaFields_unqualifiedStructOwnerQueryTimeout_throwsException() throws SQLException { + mockSingleColumn(Types.STRUCT, "address", "ADDRESS_TYPE"); + Mockito.when(connection.prepareStatement(Mockito.anyString())) + .thenThrow(new SQLTimeoutException("Query timed out")); + + Assert.assertThrows(IllegalArgumentException.class, () -> defaultSchemaReader.getSchemaFields(resultSet)); + } + + @Test + public void getSchemaFields_structWithNoAttributes_throwsException() throws SQLException { + mockSingleColumn(Types.STRUCT, "address", "CS_ITN.ADDRESS_TYPE"); + Mockito.when(attributeResultSet.next()).thenReturn(false); + + Assert.assertThrows(IllegalArgumentException.class, () -> defaultSchemaReader.getSchemaFields(resultSet)); + } + + @Test + public void getSchemaFields_nestedStructEmptyTypeOwner_throwsException() throws SQLException { + mockSingleColumn(Types.STRUCT, "address", "CS_ITN.ADDRESS_TYPE"); + Mockito.when(attributeResultSet.next()).thenReturn(true, false); + Mockito.when(attributeResultSet.getString("ATTR_NAME")).thenReturn("SUB_STRUCT"); + Mockito.when(attributeResultSet.getString("ATTR_TYPE_NAME")).thenReturn("NESTED_TYPE"); + Mockito.when(attributeResultSet.getString("ATTR_TYPE_OWNER")).thenReturn(""); + + Assert.assertThrows(IllegalArgumentException.class, () -> defaultSchemaReader.getSchemaFields(resultSet)); + } + + @Test + public void getSchemaFields_nestedStructNullTypeOwner_throwsException() throws SQLException { + mockSingleColumn(Types.STRUCT, "address", "CS_ITN.ADDRESS_TYPE"); + Mockito.when(attributeResultSet.next()).thenReturn(true, false); + Mockito.when(attributeResultSet.getString("ATTR_NAME")).thenReturn("SUB_STRUCT"); + Mockito.when(attributeResultSet.getString("ATTR_TYPE_NAME")).thenReturn("NESTED_TYPE"); + Mockito.when(attributeResultSet.getString("ATTR_TYPE_OWNER")).thenReturn(null); + + Assert.assertThrows(IllegalArgumentException.class, () -> defaultSchemaReader.getSchemaFields(resultSet)); + } + + @Test + public void getSchema_precisionlessNumericAndOldTimestampBehavior_returnsExpectedSchema() throws SQLException { + OracleSourceSchemaReader schemaReader = new OracleSourceSchemaReader("sess1", true, + true, false, false); + Mockito.when(metadata.getColumnCount()).thenReturn(9); + Mockito.when(metadata.getColumnType(1)).thenReturn(OracleSourceSchemaReader.TIMESTAMP_TZ); + Mockito.when(metadata.getColumnName(1)).thenReturn("tzCol"); + Mockito.when(metadata.getColumnType(2)).thenReturn(Types.NUMERIC); + Mockito.when(metadata.getColumnName(2)).thenReturn("doubleNumCol"); + Mockito.when(metadata.getColumnClassName(2)).thenReturn(Double.class.getTypeName()); + Mockito.when(metadata.getColumnType(3)).thenReturn(Types.NUMERIC); + Mockito.when(metadata.getColumnName(3)).thenReturn("precisionlessDecCol"); + Mockito.when(metadata.getPrecision(3)).thenReturn(0); + Mockito.when(metadata.getColumnType(4)).thenReturn(Types.VARCHAR); + Mockito.when(metadata.getColumnName(4)).thenReturn("varcharCol"); + Mockito.when(metadata.getColumnType(5)).thenReturn(Types.TIMESTAMP); + Mockito.when(metadata.getColumnName(5)).thenReturn("oldTsCol"); + Mockito.when(metadata.getColumnType(6)).thenReturn(OracleSourceSchemaReader.TIMESTAMP_LTZ); + Mockito.when(metadata.getColumnName(6)).thenReturn("oldLtzCol"); + Mockito.when(metadata.getColumnType(7)).thenReturn(OracleSourceSchemaReader.LONG); + Mockito.when(metadata.getColumnName(7)).thenReturn("longCol"); + Mockito.when(metadata.getColumnType(8)).thenReturn(Types.VARCHAR); + Mockito.when(metadata.getColumnName(8)).thenReturn("c_sess1"); + Mockito.when(metadata.getColumnType(9)).thenReturn(Types.VARCHAR); + Mockito.when(metadata.getColumnName(9)).thenReturn("s_sess1"); + + List actualFields = schemaReader.getSchemaFields(resultSet); + + Assert.assertEquals(7, actualFields.size()); + Assert.assertEquals(Schema.of(Schema.Type.STRING), actualFields.get(0).getSchema()); + Assert.assertEquals(Schema.of(Schema.Type.DOUBLE), actualFields.get(1).getSchema()); + Assert.assertEquals(Schema.decimalOf(38, 0), actualFields.get(2).getSchema()); + Assert.assertEquals(Schema.of(Schema.Type.STRING), actualFields.get(3).getSchema()); + Assert.assertEquals(Schema.of(Schema.LogicalType.TIMESTAMP_MICROS), actualFields.get(4).getSchema()); + Assert.assertEquals(Schema.of(Schema.LogicalType.TIMESTAMP_MICROS), actualFields.get(5).getSchema()); + Assert.assertEquals(Schema.of(Schema.Type.STRING), actualFields.get(6).getSchema()); + } + + @Test + public void getSchemaFields_nullStatement_throwsNullPointerException() throws SQLException { + Mockito.when(resultSet.getStatement()).thenReturn(null); + + Assert.assertThrows(NullPointerException.class, () -> defaultSchemaReader.getSchemaFields(resultSet)); + } + + @Test + public void getSchemaFields_nullConnectionForStruct_throwsNullPointerException() throws SQLException { + Mockito.when(statement.getConnection()).thenReturn(null); + mockSingleColumn(Types.STRUCT, "address", "CS_ITN.ADDRESS_TYPE"); + + Assert.assertThrows(NullPointerException.class, () -> defaultSchemaReader.getSchemaFields(resultSet)); + } + + @Test + public void getSchemaFields_structNullColumnTypeName_throwsNullPointerException() throws SQLException { + PreparedStatement ownerStmt = Mockito.mock(PreparedStatement.class); + ResultSet ownerRs = Mockito.mock(ResultSet.class); + Mockito.when(connection.prepareStatement(Mockito.anyString())).thenReturn(ownerStmt, preparedStatement); + Mockito.when(ownerStmt.executeQuery()).thenReturn(ownerRs); + Mockito.when(ownerRs.next()).thenReturn(true); + Mockito.when(ownerRs.getString("DATA_TYPE_OWNER")).thenReturn("CS_ITN"); + mockSingleColumn(Types.STRUCT, "address", null); + + Assert.assertThrows(NullPointerException.class, () -> defaultSchemaReader.getSchemaFields(resultSet)); + } + + @Test + public void getSchemaFields_numericNegativePrecision_throwsIllegalArgumentException() throws SQLException { + mockSingleColumn(Types.NUMERIC, "negPrecCol", "NUMBER"); + Mockito.when(metadata.getPrecision(1)).thenReturn(-1); + Mockito.when(metadata.getScale(1)).thenReturn(0); + + Assert.assertThrows(IllegalArgumentException.class, () -> defaultSchemaReader.getSchemaFields(resultSet)); + } + + @Test + public void getSchemaFields_numericMinIntegerPrecision_throwsIllegalArgumentException() throws SQLException { + mockSingleColumn(Types.NUMERIC, "minPrecCol", "NUMBER"); + Mockito.when(metadata.getPrecision(1)).thenReturn(Integer.MIN_VALUE); + Mockito.when(metadata.getScale(1)).thenReturn(0); + + Assert.assertThrows(IllegalArgumentException.class, () -> defaultSchemaReader.getSchemaFields(resultSet)); + } + + @Test + public void getSchemaFields_structAttributeNegativePrecision_throwsIllegalArgumentException() throws SQLException { + mockSingleColumn(Types.STRUCT, "address", "CS_ITN.ADDRESS_TYPE"); + Mockito.when(attributeResultSet.next()).thenReturn(true, false); + Mockito.when(attributeResultSet.getString("ATTR_NAME")).thenReturn("BAD_NUM"); + Mockito.when(attributeResultSet.getString("ATTR_TYPE_NAME")).thenReturn("NUMBER"); + Mockito.when(attributeResultSet.getInt("PRECISION")).thenReturn(-5); + Mockito.when(attributeResultSet.getInt("SCALE")).thenReturn(10); + + Assert.assertThrows(IllegalArgumentException.class, () -> defaultSchemaReader.getSchemaFields(resultSet)); + } + + @Test + public void getSchemaFields_structAttributeQueryTimeout_throwsSQLTimeoutException() throws SQLException { + mockSingleColumn(Types.STRUCT, "address", "CS_ITN.ADDRESS_TYPE"); + Mockito.when(preparedStatement.executeQuery()) + .thenThrow(new SQLTimeoutException("Timeout fetching struct attributes")); + Assert.assertThrows(SQLTimeoutException.class, () -> defaultSchemaReader.getSchemaFields(resultSet)); } } diff --git a/oracle-plugin/src/test/java/io/cdap/plugin/oracle/OracleSourceDBRecordUnitTest.java b/oracle-plugin/src/test/java/io/cdap/plugin/oracle/OracleSourceDBRecordUnitTest.java index 77136e841..c3fc6c9af 100644 --- a/oracle-plugin/src/test/java/io/cdap/plugin/oracle/OracleSourceDBRecordUnitTest.java +++ b/oracle-plugin/src/test/java/io/cdap/plugin/oracle/OracleSourceDBRecordUnitTest.java @@ -18,18 +18,42 @@ import io.cdap.cdap.api.data.format.StructuredRecord; import io.cdap.cdap.api.data.schema.Schema; +import io.cdap.cdap.etl.api.validation.InvalidStageException; +import io.cdap.plugin.db.ColumnType; +import io.cdap.plugin.db.ConnectionConfigAccessor; +import oracle.jdbc.OracleBfile; +import oracle.sql.TIMESTAMPTZ; +import org.apache.hadoop.conf.Configuration; import org.junit.Assert; +import org.junit.Before; import org.junit.Test; import org.junit.runner.RunWith; import org.mockito.Mock; +import org.mockito.Mockito; import org.mockito.junit.MockitoJUnitRunner; +import java.io.ByteArrayInputStream; +import java.io.PipedInputStream; +import java.lang.reflect.Proxy; import java.math.BigDecimal; +import java.net.URL; +import java.net.URLClassLoader; +import java.nio.ByteBuffer; +import java.sql.Connection; +import java.sql.PreparedStatement; import java.sql.ResultSet; import java.sql.ResultSetMetaData; +import java.sql.SQLException; +import java.sql.Statement; +import java.sql.Struct; import java.sql.Timestamp; import java.sql.Types; -import java.time.ZonedDateTime; +import java.time.LocalDateTime; +import java.time.OffsetDateTime; +import java.time.ZoneId; +import java.time.ZoneOffset; +import java.util.Collections; +import java.util.Map; import static org.mockito.ArgumentMatchers.eq; import static org.mockito.Mockito.when; @@ -48,6 +72,53 @@ public class OracleSourceDBRecordUnitTest { @Mock ResultSetMetaData resultSetMetaData; + @Mock + Statement statement; + + @Mock + Connection connection; + + @Mock + PreparedStatement preparedStatement; + + private OracleSourceDBRecord dbRecord; + + /** + * Mock interface for Oracle Struct containing getDescriptor method. + */ + public interface MockOracleStruct extends Struct { + Object getDescriptor() throws Exception; + } + + /** + * Mock interface for Oracle StructDescriptor containing getMetaData method. + */ + public interface MockStructDescriptor { + ResultSetMetaData getMetaData() throws Exception; + } + + @Before + public void setUp() throws SQLException { + dbRecord = new OracleSourceDBRecord(null, null); + when(resultSet.getMetaData()).thenReturn(resultSetMetaData); + when(resultSet.getStatement()).thenReturn(statement); + when(statement.getConnection()).thenReturn(connection); + } + + private MockOracleStruct createMockStruct(String[] columnNames, Object[] attributes) throws Exception { + MockOracleStruct structMock = Mockito.mock(MockOracleStruct.class); + MockStructDescriptor descriptorMock = Mockito.mock(MockStructDescriptor.class); + ResultSetMetaData structMetaData = Mockito.mock(ResultSetMetaData.class); + when(structMock.getAttributes()).thenReturn(attributes); + when(structMock.getDescriptor()).thenReturn(descriptorMock); + when(descriptorMock.getMetaData()).thenReturn(structMetaData); + when(structMetaData.getColumnCount()).thenReturn(columnNames.length); + for (int i = 0; i < columnNames.length; i++) { + when(structMetaData.getColumnName(eq(i + 1))).thenReturn(columnNames[i]); + } + return structMock; + } + /** * Validate the precision less Numbers handling against following use cases. * 1. Ensure that for Number(0,-127) non nullable type a String type is returned if output schema is String. @@ -234,4 +305,359 @@ public void validateTimestampTZTypeNullHandling() throws Exception { StructuredRecord record = builder.build(); Assert.assertNull(record.get("field1")); } + + @Test + public void populateStructField_nullValue_setsFieldToNull() throws Exception { + Schema addressSchema = Schema.recordOf("ADDRESS_TYPE", + Schema.Field.of("STREET", Schema.of(Schema.Type.STRING)), + Schema.Field.of("CITY", Schema.of(Schema.Type.STRING))); + Schema.Field addressField = Schema.Field.of("address", Schema.nullableOf(addressSchema)); + Schema schema = Schema.recordOf("dbRecord", addressField); + when(resultSet.getObject(eq(1))).thenReturn(null); + StructuredRecord.Builder builder = StructuredRecord.builder(schema); + + dbRecord.handleField(resultSet, builder, addressField, 1, Types.STRUCT, + DEFAULT_PRECISION, 0); + + StructuredRecord record = builder.build(); + Assert.assertNull(record.get("address")); + } + + @Test + public void handleStructField_nonNullValue_setsFieldCorrectly() throws Exception { + Schema addressSchema = Schema.recordOf("ADDRESS_TYPE", + Schema.Field.of("STREET", Schema.of(Schema.Type.STRING)), + Schema.Field.of("CITY", Schema.of(Schema.Type.STRING)), + Schema.Field.of("ZIPCODE", Schema.of(Schema.Type.INT))); + Schema.Field addressField = Schema.Field.of("address", addressSchema); + Schema schema = Schema.recordOf("dbRecord", addressField); + MockOracleStruct structMock = createMockStruct(new String[]{ "STREET", "CITY", "ZIPCODE" }, + new Object[]{ "123 Main St", "San Francisco", 94105 }); + when(resultSet.getObject(eq(1))).thenReturn(structMock); + StructuredRecord.Builder builder = StructuredRecord.builder(schema); + + dbRecord.handleField(resultSet, builder, addressField, 1, Types.STRUCT, + DEFAULT_PRECISION, 0); + + StructuredRecord record = builder.build(); + StructuredRecord addressRecord = record.get("address"); + Assert.assertNotNull(addressRecord); + Assert.assertEquals("123 Main St", addressRecord.get("STREET")); + Assert.assertEquals("San Francisco", addressRecord.get("CITY")); + Assert.assertEquals(Integer.valueOf(94105), addressRecord.get("ZIPCODE")); + } + + @Test + public void handleStructField_nestedStructure_setsFieldCorrectly() throws Exception { + Schema locationSchema = Schema.recordOf("LOCATION_TYPE", + Schema.Field.of("LATITUDE", Schema.of(Schema.Type.DOUBLE)), + Schema.Field.of("LONGITUDE", Schema.of(Schema.Type.DOUBLE))); + Schema addressSchema = Schema.recordOf("ADDRESS_TYPE", + Schema.Field.of("STREET", Schema.of(Schema.Type.STRING)), + Schema.Field.of("LOCATION", locationSchema)); + Schema.Field addressField = Schema.Field.of("address", addressSchema); + Schema schema = Schema.recordOf("dbRecord", addressField); + MockOracleStruct locationStructMock = createMockStruct(new String[]{ "LATITUDE", "LONGITUDE" }, + new Object[]{ Double.valueOf(37.7749), Double.valueOf(-122.4194) }); + MockOracleStruct addressStructMock = createMockStruct(new String[]{ "STREET", "LOCATION" }, + new Object[]{ "123 Main St", locationStructMock }); + when(resultSet.getObject(eq(1))).thenReturn(addressStructMock); + StructuredRecord.Builder builder = StructuredRecord.builder(schema); + + dbRecord.handleField(resultSet, builder, addressField, 1, Types.STRUCT, + DEFAULT_PRECISION, 0); + + StructuredRecord record = builder.build(); + StructuredRecord addressRecord = record.get("address"); + Assert.assertNotNull(addressRecord); + Assert.assertEquals("123 Main St", addressRecord.get("STREET")); + StructuredRecord locationRecord = addressRecord.get("LOCATION"); + Assert.assertNotNull(locationRecord); + Assert.assertEquals(Double.valueOf(37.7749), locationRecord.get("LATITUDE")); + Assert.assertEquals(Double.valueOf(-122.4194), locationRecord.get("LONGITUDE")); + } + + @Test + public void getAttributeMap_nullAttributes_returnsEmptyMap() throws Exception { + Schema addressSchema = Schema.recordOf("ADDRESS_TYPE", + Schema.Field.of("STREET", Schema.of(Schema.Type.STRING))); + MockOracleStruct structMock = Mockito.mock(MockOracleStruct.class); + when(structMock.getAttributes()).thenReturn(null); + + Map attributeMap = dbRecord.getAttributeMap(structMock, addressSchema, connection); + + Assert.assertTrue(attributeMap.isEmpty()); + } + + @Test + public void getAttributeMap_fewerAttributesThanColumns_mapsAvailableAttributes() throws Exception { + Schema addressSchema = Schema.recordOf("ADDRESS_TYPE", + Schema.Field.of("STREET", Schema.of(Schema.Type.STRING)), + Schema.Field.of("CITY", Schema.nullableOf(Schema.of(Schema.Type.STRING)))); + MockOracleStruct structMock = createMockStruct(new String[]{ "STREET", "CITY" }, + new Object[]{ "123 Main St" }); + + Map attributeMap = dbRecord.getAttributeMap(structMock, addressSchema, connection); + + Assert.assertEquals(1, attributeMap.size()); + Assert.assertEquals("123 Main St", attributeMap.get("STREET")); + } + + @Test + public void getAttributeMap_nullDescriptor_throwsSQLException() throws Exception { + Schema addressSchema = Schema.recordOf("ADDRESS_TYPE", + Schema.Field.of("STREET", Schema.of(Schema.Type.STRING))); + MockOracleStruct structMock = Mockito.mock(MockOracleStruct.class); + when(structMock.getAttributes()).thenReturn(new Object[]{ "123 Main St" }); + when(structMock.getDescriptor()).thenReturn(null); + + Assert.assertThrows(SQLException.class, () -> dbRecord.getAttributeMap(structMock, addressSchema, connection)); + } + + @Test + public void getBfileBytes_nullOrNonExistentFile_returnsNull() throws Exception { + OracleBfile bfileMock = Mockito.mock(OracleBfile.class); + when(bfileMock.fileExists()).thenReturn(false); + + Assert.assertNull(dbRecord.getBfileBytes((Object) null, "BFILE_COL")); + Assert.assertNull(dbRecord.getBfileBytes(bfileMock, "BFILE_COL")); + } + + @Test + public void getBfileBytes_uninitializedBfileOrUnconnectedStream_throwsInvalidStageException() throws Exception { + Object uninitializedBfileProxy = Proxy.newProxyInstance(OracleBfile.class.getClassLoader(), + new Class[]{ OracleBfile.class }, (proxy, method, args) -> null); + OracleBfile bfileWithUnconnectedPipe = Mockito.mock(OracleBfile.class); + when(bfileWithUnconnectedPipe.fileExists()).thenReturn(true); + when(bfileWithUnconnectedPipe.getBinaryStream()).thenReturn(new PipedInputStream()); + + Assert.assertThrows(InvalidStageException.class, + () -> dbRecord.getBfileBytes(uninitializedBfileProxy, "BFILE_COL")); + Assert.assertThrows(InvalidStageException.class, + () -> dbRecord.getBfileBytes(bfileWithUnconnectedPipe, "BFILE_COL")); + } + + @Test + public void handleField_bfileColumnWithExistingFile_setsBytesCorrectly() throws Exception { + Schema.Field bfileField = Schema.Field.of("BFILE_COL", Schema.of(Schema.Type.BYTES)); + Schema schema = Schema.recordOf("dbRecord", bfileField); + StructuredRecord.Builder builder = StructuredRecord.builder(schema); + OracleBfile bfileMock = Mockito.mock(OracleBfile.class); + byte[] expectedBytes = new byte[]{ 10, 20, 30 }; + when(resultSetMetaData.getColumnName(eq(1))).thenReturn("BFILE_COL"); + when(resultSet.getObject(eq("BFILE_COL"))).thenReturn(bfileMock); + when(bfileMock.fileExists()).thenReturn(true); + when(bfileMock.getBinaryStream()).thenReturn(new ByteArrayInputStream(expectedBytes)); + + dbRecord.handleField(resultSet, builder, bfileField, 1, OracleSourceSchemaReader.BFILE, + 0, 0); + + StructuredRecord record = builder.build(); + Assert.assertArrayEquals(expectedBytes, record.get("BFILE_COL")); + } + + @Test + public void handleField_numericAndBinaryTypes_setsFieldsCorrectly() throws Exception { + Schema.Field nclobField = Schema.Field.of("NCLOB_COL", Schema.of(Schema.Type.STRING)); + Schema.Field bfloatField = Schema.Field.of("BFLOAT_COL", Schema.of(Schema.Type.FLOAT)); + Schema.Field bdoubleField = Schema.Field.of("BDOUBLE_COL", Schema.of(Schema.Type.DOUBLE)); + Schema.Field numDoubleField = Schema.Field.of("NUM_DOUBLE_COL", Schema.of(Schema.Type.DOUBLE)); + Schema schema = Schema.recordOf("dbRecord", nclobField, bfloatField, bdoubleField, numDoubleField); + StructuredRecord.Builder builder = StructuredRecord.builder(schema); + when(resultSet.getString(eq(1))).thenReturn("nclobVal"); + when(resultSet.getFloat(eq(2))).thenReturn(1.5f); + when(resultSet.getDouble(eq(3))).thenReturn(2.5d); + when(resultSetMetaData.getColumnClassName(eq(4))).thenReturn(Double.class.getTypeName()); + when(resultSet.getDouble(eq(4))).thenReturn(3.5d); + + dbRecord.handleField(resultSet, builder, nclobField, 1, Types.NCLOB, 0, 0); + dbRecord.handleField(resultSet, builder, bfloatField, 2, OracleSourceSchemaReader.BINARY_FLOAT, + 0, 0); + dbRecord.handleField(resultSet, builder, bdoubleField, 3, OracleSourceSchemaReader.BINARY_DOUBLE, + 0, 0); + dbRecord.handleField(resultSet, builder, numDoubleField, 4, Types.NUMERIC, 10, 2); + + StructuredRecord record = builder.build(); + Assert.assertEquals("nclobVal", record.get("NCLOB_COL")); + Assert.assertEquals(1.5f, record.get("BFLOAT_COL"), 0.001f); + Assert.assertEquals(2.5d, record.get("BDOUBLE_COL"), 0.001d); + Assert.assertEquals(3.5d, record.get("NUM_DOUBLE_COL"), 0.001d); + } + + @Test + public void handleField_temporalTypes_setsFieldsCorrectly() throws Exception { + Schema.Field tzStringField = Schema.Field.of("TZ_STR_COL", Schema.of(Schema.Type.STRING)); + Schema.Field tzMicrosField = Schema.Field.of("TZ_MICROS_COL", + Schema.of(Schema.LogicalType.TIMESTAMP_MICROS)); + Schema.Field tsDateTimeField = Schema.Field.of("TS_DT_COL", Schema.of(Schema.LogicalType.DATETIME)); + Schema.Field tsNullDateTimeField = Schema.Field.of("TS_NULL_DT_COL", + Schema.nullableOf(Schema.of(Schema.LogicalType.DATETIME))); + Schema.Field tsMicrosField = Schema.Field.of("TS_MICROS_COL", + Schema.of(Schema.LogicalType.TIMESTAMP_MICROS)); + Schema.Field ltzDateTimeField = Schema.Field.of("LTZ_DT_COL", Schema.of(Schema.LogicalType.DATETIME)); + Schema.Field ltzNullDateTimeField = Schema.Field.of("LTZ_NULL_DT_COL", + Schema.nullableOf(Schema.of(Schema.LogicalType.DATETIME))); + Schema.Field ltzMicrosField = Schema.Field.of("LTZ_MICROS_COL", + Schema.of(Schema.LogicalType.TIMESTAMP_MICROS)); + Schema schema = Schema.recordOf("dbRecord", tzStringField, tzMicrosField, tsDateTimeField, + tsNullDateTimeField, tsMicrosField, ltzDateTimeField, ltzNullDateTimeField, ltzMicrosField); + StructuredRecord.Builder builder = StructuredRecord.builder(schema); + Timestamp nowTs = Timestamp.valueOf("2024-05-15 10:30:00"); + OffsetDateTime expectedOdt = OffsetDateTime.of(2024, 5, 15, 10, + 30, 0, 0, ZoneOffset.UTC); + TIMESTAMPTZ tzMock = Mockito.mock(TIMESTAMPTZ.class); + when(tzMock.offsetDateTimeValue(eq(connection))).thenReturn(expectedOdt); + when(resultSet.getString(eq(1))).thenReturn("2024-05-15 10:30:00 GMT"); + when(resultSet.getObject(eq(2))).thenReturn(tzMock); + when(resultSet.getTimestamp(eq(3))).thenReturn(nowTs); + when(resultSet.getTimestamp(eq(4))).thenReturn(null); + when(resultSet.getObject(eq(5))).thenReturn(nowTs); + when(resultSet.getTimestamp(eq(5), Mockito.any())).thenReturn(nowTs); + when(resultSet.getTimestamp(eq(6))).thenReturn(nowTs); + when(resultSet.getTimestamp(eq(7))).thenReturn(null); + when(resultSet.getTimestamp(eq(8))).thenReturn(nowTs); + + dbRecord.handleField(resultSet, builder, tzStringField, 1, + OracleSourceSchemaReader.TIMESTAMP_TZ, 0, 0); + dbRecord.handleField(resultSet, builder, tzMicrosField, 2, + OracleSourceSchemaReader.TIMESTAMP_TZ, 0, 0); + dbRecord.handleField(resultSet, builder, tsDateTimeField, 3, + Types.TIMESTAMP, 0, 0); + dbRecord.handleField(resultSet, builder, tsNullDateTimeField, 4, + Types.TIMESTAMP, 0, 0); + dbRecord.handleField(resultSet, builder, tsMicrosField, 5, + Types.TIMESTAMP, 0, 0); + dbRecord.handleField(resultSet, builder, ltzDateTimeField, 6, + OracleSourceSchemaReader.TIMESTAMP_LTZ, 0, 0); + dbRecord.handleField(resultSet, builder, ltzNullDateTimeField, 7, + OracleSourceSchemaReader.TIMESTAMP_LTZ, 0, 0); + dbRecord.handleField(resultSet, builder, ltzMicrosField, 8, + OracleSourceSchemaReader.TIMESTAMP_LTZ, 0, 0); + + StructuredRecord record = builder.build(); + Assert.assertEquals("2024-05-15 10:30:00 GMT", record.get("TZ_STR_COL")); + Assert.assertEquals(expectedOdt.atZoneSameInstant(ZoneId.of("UTC")), + record.getTimestamp("TZ_MICROS_COL")); + Assert.assertEquals(nowTs.toLocalDateTime(), record.getDateTime("TS_DT_COL")); + Assert.assertNull(record.getDateTime("TS_NULL_DT_COL")); + Assert.assertNotNull(record.getTimestamp("TS_MICROS_COL")); + Assert.assertNotNull(record.getDateTime("LTZ_DT_COL")); + Assert.assertNull(record.getDateTime("LTZ_NULL_DT_COL")); + Assert.assertNotNull(record.getTimestamp("LTZ_MICROS_COL")); + } + + @Test + public void handleField_uninitializedTimestampTzObject_throwsRuntimeException() throws Exception { + Schema.Field tzMicrosField = Schema.Field.of("TZ_MICROS_COL", + Schema.of(Schema.LogicalType.TIMESTAMP_MICROS)); + Schema schema = Schema.recordOf("dbRecord", tzMicrosField); + StructuredRecord.Builder builder = StructuredRecord.builder(schema); + when(resultSet.getObject(eq(1))).thenReturn(new TIMESTAMPTZ()); + when(statement.getConnection()).thenReturn(null); + + Assert.assertThrows(RuntimeException.class, () -> + dbRecord.handleField(resultSet, builder, tzMicrosField, 1, + OracleSourceSchemaReader.TIMESTAMP_TZ, 0, 0)); + } + + @Test + public void readFields_longRawAndStandardColumns_readsInExpectedOrder() throws Exception { + OracleSourceDBRecord defaultRecord = new OracleSourceDBRecord(); + Schema.Field varcharField = Schema.Field.of("VARCHAR_COL", Schema.of(Schema.Type.STRING)); + Schema.Field longField = Schema.Field.of("LONG_COL", Schema.of(Schema.Type.STRING)); + Schema.Field longRawField = Schema.Field.of("LONG_RAW_COL", Schema.of(Schema.Type.BYTES)); + Schema schema = Schema.recordOf("dbRecord", varcharField, longField, longRawField); + Configuration conf = new Configuration(); + conf.set(ConnectionConfigAccessor.OVERRIDE_SCHEMA, schema.toString()); + defaultRecord.setConf(conf); + byte[] rawBytes = new byte[]{ 1, 2, 3 }; + when(resultSet.findColumn(eq("VARCHAR_COL"))).thenReturn(1); + when(resultSet.findColumn(eq("LONG_COL"))).thenReturn(2); + when(resultSet.findColumn(eq("LONG_RAW_COL"))).thenReturn(3); + when(resultSetMetaData.getColumnType(eq(1))).thenReturn(Types.VARCHAR); + when(resultSetMetaData.getColumnType(eq(2))).thenReturn(OracleSourceSchemaReader.LONG); + when(resultSetMetaData.getColumnType(eq(3))).thenReturn(OracleSourceSchemaReader.LONG_RAW); + when(resultSet.getObject(eq(1))).thenReturn("standardText"); + when(resultSet.getString(eq(2))).thenReturn("longText"); + when(resultSet.getBytes(eq(3))).thenReturn(rawBytes); + + defaultRecord.readFields(resultSet); + + StructuredRecord builtRecord = defaultRecord.getRecord(); + Assert.assertEquals("standardText", builtRecord.get("VARCHAR_COL")); + Assert.assertEquals("longText", builtRecord.get("LONG_COL")); + Assert.assertArrayEquals(rawBytes, builtRecord.get("LONG_RAW_COL")); + } + + @Test + public void write_invalidTimestampTzFormat_throwsInvalidStageException() throws Exception { + when(preparedStatement.getConnection()).thenReturn(connection); + Schema schema = Schema.recordOf("tzRecord", + Schema.Field.of("TZ_STR", Schema.of(Schema.Type.STRING))); + StructuredRecord record = StructuredRecord.builder(schema).set("TZ_STR", "invalid").build(); + OracleSourceDBRecord dbRecord = new OracleSourceDBRecord(record, + Collections.singletonList(new ColumnType("TZ_STR", "TIMESTAMPTZ", + OracleSourceSchemaReader.TIMESTAMP_TZ))); + + Assert.assertThrows(InvalidStageException.class, () -> dbRecord.write(preparedStatement)); + } + + @Test + public void write_invalidTimestampLtzFormat_throwsInvalidStageException() throws Exception { + when(preparedStatement.getConnection()).thenReturn(connection); + Schema schema = Schema.recordOf("ltzRecord", Schema.Field.of("LTZ_DT", + Schema.of(Schema.LogicalType.DATETIME))); + StructuredRecord record = StructuredRecord.builder(schema).setDateTime("LTZ_DT", + LocalDateTime.of(2024, 5, 15, 10, 30)).build(); + OracleSourceDBRecord dbRecord = new OracleSourceDBRecord(record, + Collections.singletonList(new ColumnType("LTZ_DT", + "TIMESTAMPLTZ", OracleSourceSchemaReader.TIMESTAMP_LTZ))); + + Assert.assertThrows(InvalidStageException.class, () -> dbRecord.write(preparedStatement)); + } + + @Test + public void write_isolatedConnectionWithStandardTimestamp_throwsInvalidStageException() throws Exception { + ClassLoader isolatedLoader = new URLClassLoader(new URL[0], Connection.class.getClassLoader()); + Connection isolatedConnection = (Connection) Proxy.newProxyInstance(isolatedLoader, + new Class[]{ Connection.class }, + (proxy, method, args) -> null); + when(preparedStatement.getConnection()).thenReturn(isolatedConnection); + Schema schema = Schema.recordOf("tsRecord", Schema.Field.of("TS_DT", + Schema.of(Schema.LogicalType.DATETIME))); + StructuredRecord record = StructuredRecord.builder(schema).setDateTime("TS_DT", + LocalDateTime.of(2024, 5, 15, 10, 30)).build(); + OracleSourceDBRecord dbRecord = new OracleSourceDBRecord(record, + Collections.singletonList(new ColumnType("TS_DT", "TIMESTAMP", Types.TIMESTAMP))); + + Assert.assertThrows(InvalidStageException.class, () -> dbRecord.write(preparedStatement)); + } + + @Test + public void write_isolatedConnectionWithTimestampLtz_throwsInvalidStageException() throws Exception { + ClassLoader isolatedLoader = new URLClassLoader(new URL[0], Connection.class.getClassLoader()); + Connection isolatedConnection = (Connection) Proxy.newProxyInstance(isolatedLoader, + new Class[]{ Connection.class }, + (proxy, method, args) -> null); + when(preparedStatement.getConnection()).thenReturn(isolatedConnection); + Schema schema = Schema.recordOf("ltzRecord", Schema.Field.of("LTZ_DT", + Schema.of(Schema.LogicalType.DATETIME))); + StructuredRecord record = StructuredRecord.builder(schema).setDateTime("LTZ_DT", + LocalDateTime.of(2024, 5, 15, 10, 30)).build(); + OracleSourceDBRecord dbRecord = new OracleSourceDBRecord(record, + Collections.singletonList(new ColumnType("LTZ_DT", "TIMESTAMPLTZ", + OracleSourceSchemaReader.TIMESTAMP_LTZ))); + + Assert.assertThrows(InvalidStageException.class, () -> dbRecord.write(preparedStatement)); + } + + @Test + public void writeBytes_byteBufferAndByteArray_setsBytesOnStatement() throws Exception { + byte[] rawBytes = new byte[]{ 5, 6, 7 }; + + dbRecord.writeBytes(preparedStatement, 0, 1, ByteBuffer.wrap(rawBytes)); + dbRecord.writeBytes(preparedStatement, 1, 2, rawBytes); + + Assert.assertNotNull(preparedStatement); + } } diff --git a/oracle-plugin/src/test/java/io/cdap/plugin/oracle/OracleStructUtilTest.java b/oracle-plugin/src/test/java/io/cdap/plugin/oracle/OracleStructUtilTest.java new file mode 100644 index 000000000..5054e6684 --- /dev/null +++ b/oracle-plugin/src/test/java/io/cdap/plugin/oracle/OracleStructUtilTest.java @@ -0,0 +1,227 @@ +/* + * Copyright © 2024 Cask Data, Inc. + * + * Licensed under the Apache License, Version 2.0 (the "License"); you may not + * use this file except in compliance with the License. You may obtain a copy of + * the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT + * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the + * License for the specific language governing permissions and limitations under + * the License. + */ + +package io.cdap.plugin.oracle; + +import io.cdap.cdap.api.data.format.StructuredRecord; +import io.cdap.cdap.api.data.schema.Schema; +import oracle.jdbc.internal.OracleBfile; +import oracle.sql.INTERVALDS; +import oracle.sql.INTERVALYM; +import org.junit.Assert; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.mockito.Mock; +import org.mockito.Mockito; +import org.mockito.junit.MockitoJUnitRunner; + +import java.math.BigDecimal; +import java.sql.Blob; +import java.sql.Clob; +import java.sql.Connection; +import java.sql.SQLXML; +import java.sql.Struct; +import java.sql.Timestamp; +import java.time.LocalDate; +import java.time.LocalDateTime; +import java.time.OffsetDateTime; +import java.time.ZoneId; +import java.time.ZoneOffset; + +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.when; + +/** + * Unit Test class for the OracleStructUtil + */ +@RunWith(MockitoJUnitRunner.class) +public class OracleStructUtilTest { + + @Mock + private OracleSourceDBRecord sourceRecord; + + @Mock + private Connection connection; + + @Test + public void populateRecordField_nullValue_setsNullValue() throws Exception { + Schema.Field field = Schema.Field.of("ID", Schema.nullableOf(Schema.of(Schema.Type.STRING))); + Schema schema = Schema.recordOf("testRecord", field); + StructuredRecord.Builder builder = StructuredRecord.builder(schema); + + OracleStructUtil.populateRecordField(sourceRecord, connection, builder, field, null); + + StructuredRecord record = builder.build(); + Assert.assertNull(record.get("ID")); + } + + @Test + public void populateRecordField_complexTypes_setsFieldsCorrectly() throws Exception { + Schema innerSchema = Schema.recordOf("inner", Schema.Field.of("INNER_COL", + Schema.of(Schema.Type.STRING))); + Schema.Field structField = Schema.Field.of("STRUCT_FIELD", innerSchema); + Schema.Field clobField = Schema.Field.of("CLOB_FIELD", Schema.of(Schema.Type.STRING)); + Schema.Field blobField = Schema.Field.of("BLOB_FIELD", Schema.of(Schema.Type.BYTES)); + Schema schema = Schema.recordOf("complexRecord", structField, clobField, blobField); + StructuredRecord.Builder builder = StructuredRecord.builder(schema); + Struct structMock = Mockito.mock(Struct.class); + StructuredRecord innerRecord = StructuredRecord.builder(innerSchema).set("INNER_COL", "val").build(); + when(sourceRecord.convertStructToRecord(eq(structMock), eq(innerSchema), eq(connection))) + .thenReturn(innerRecord); + Clob clobMock = Mockito.mock(Clob.class); + when(clobMock.length()).thenReturn(4L); + when(clobMock.getSubString(1, 4)).thenReturn("text"); + Blob blobMock = Mockito.mock(Blob.class); + byte[] blobBytes = new byte[]{1, 2, 3}; + when(blobMock.length()).thenReturn(3L); + when(blobMock.getBytes(1, 3)).thenReturn(blobBytes); + + OracleStructUtil.populateRecordField(sourceRecord, connection, builder, structField, structMock); + OracleStructUtil.populateRecordField(sourceRecord, connection, builder, clobField, clobMock); + OracleStructUtil.populateRecordField(sourceRecord, connection, builder, blobField, blobMock); + + StructuredRecord record = builder.build(); + Assert.assertEquals(innerRecord, record.get("STRUCT_FIELD")); + Assert.assertEquals("text", record.get("CLOB_FIELD")); + Assert.assertArrayEquals(blobBytes, record.get("BLOB_FIELD")); + } + + @Test + public void populateRecordField_numericConversions_setsFieldsCorrectly() throws Exception { + Schema.Field decimalField = Schema.Field.of("DEC_FIELD", Schema.decimalOf(10, 2)); + Schema.Field doubleField = Schema.Field.of("DOUBLE_FIELD", Schema.of(Schema.Type.DOUBLE)); + Schema.Field intField = Schema.Field.of("INT_FIELD", Schema.of(Schema.Type.INT)); + Schema.Field stringField = Schema.Field.of("STRING_FIELD", Schema.of(Schema.Type.STRING)); + Schema schema = Schema.recordOf("numericRecord", decimalField, doubleField, + intField, stringField); + StructuredRecord.Builder builder = StructuredRecord.builder(schema); + BigDecimal bigDecimalVal = new BigDecimal("123.456"); + + OracleStructUtil.populateRecordField(sourceRecord, connection, builder, decimalField, bigDecimalVal); + OracleStructUtil.populateRecordField(sourceRecord, connection, builder, doubleField, bigDecimalVal); + OracleStructUtil.populateRecordField(sourceRecord, connection, builder, intField, bigDecimalVal); + OracleStructUtil.populateRecordField(sourceRecord, connection, builder, stringField, bigDecimalVal); + + StructuredRecord record = builder.build(); + Assert.assertEquals(new BigDecimal("123.46"), record.getDecimal("DEC_FIELD")); + Assert.assertEquals(123.456, (Double) record.get("DOUBLE_FIELD"), 0.0001); + Assert.assertEquals(Integer.valueOf(123), record.get("INT_FIELD")); + Assert.assertEquals("123.456", record.get("STRING_FIELD")); + } + + @Test + public void populateRecordField_temporalTypes_setsFieldsCorrectly() throws Exception { + Schema.Field datetimeTsField = Schema.Field.of("DT_TS_FIELD", + Schema.of(Schema.LogicalType.DATETIME)); + Schema.Field microsOdtField = Schema.Field.of("MICROS_ODT_FIELD", + Schema.of(Schema.LogicalType.TIMESTAMP_MICROS)); + Schema.Field datetimeOdtField = Schema.Field.of("DT_ODT_FIELD", + Schema.of(Schema.LogicalType.DATETIME)); + Schema.Field stringOdtField = Schema.Field.of("STR_ODT_FIELD", Schema.of(Schema.Type.STRING)); + Schema schema = Schema.recordOf("temporalRecord", datetimeTsField, + microsOdtField, datetimeOdtField, stringOdtField); + StructuredRecord.Builder builder = StructuredRecord.builder(schema); + Timestamp timestamp = Timestamp.valueOf("2024-05-15 10:30:00"); + OffsetDateTime offsetDateTime = OffsetDateTime.of(2024, 5, 15, + 10, 30, 0, 0, ZoneOffset.ofHours(5)); + + OracleStructUtil.populateRecordField(sourceRecord, connection, builder, datetimeTsField, timestamp); + OracleStructUtil.populateRecordField(sourceRecord, connection, builder, microsOdtField, offsetDateTime); + OracleStructUtil.populateRecordField(sourceRecord, connection, builder, datetimeOdtField, offsetDateTime); + OracleStructUtil.populateRecordField(sourceRecord, connection, builder, stringOdtField, offsetDateTime); + + StructuredRecord record = builder.build(); + Assert.assertEquals(LocalDateTime.parse("2024-05-15T10:30:00"), + record.getDateTime("DT_TS_FIELD")); + Assert.assertEquals(offsetDateTime.atZoneSameInstant(ZoneId.of("UTC")), + record.getTimestamp("MICROS_ODT_FIELD")); + Assert.assertEquals(offsetDateTime.toLocalDateTime(), + record.getDateTime("DT_ODT_FIELD")); + Assert.assertEquals(offsetDateTime.atZoneSameInstant(ZoneId.of("UTC")).toString(), + record.get("STR_ODT_FIELD")); + } + + @Test + public void populateRecordField_xmlAndIntervalAndBfileTypes_setsFieldsCorrectly() throws Exception { + Schema.Field xmlField = Schema.Field.of("XML_FIELD", Schema.nullableOf(Schema.of(Schema.Type.STRING))); + Schema.Field intervalDsField = Schema.Field.of("IDS_FIELD", Schema.of(Schema.Type.STRING)); + Schema.Field intervalYmField = Schema.Field.of("IYM_FIELD", Schema.of(Schema.Type.STRING)); + Schema.Field bfileField = Schema.Field.of("BFILE_FIELD", Schema.of(Schema.Type.BYTES)); + Schema.Field defaultField = Schema.Field.of("STR_FIELD", Schema.of(Schema.Type.STRING)); + Schema schema = Schema.recordOf("oracleTypesRecord", xmlField, intervalDsField, + intervalYmField, bfileField, defaultField); + StructuredRecord.Builder builder = StructuredRecord.builder(schema); + OracleSourceDBRecord realSourceRecord = new OracleSourceDBRecord(null, null); + SQLXML sqlXmlMock = Mockito.mock(SQLXML.class); + when(sqlXmlMock.getString()).thenReturn(""); + INTERVALDS intervalDs = new INTERVALDS("23 3:2:10.0"); + INTERVALYM intervalYm = new INTERVALYM("300-5"); + OracleBfile bfileMock = Mockito.mock(OracleBfile.class); + byte[] bfileBytes = new byte[]{4, 5, 6}; + when(sourceRecord.getBfileBytes(eq(bfileMock), eq("BFILE_FIELD"))).thenReturn(bfileBytes); + + OracleStructUtil.populateRecordField(sourceRecord, connection, builder, xmlField, sqlXmlMock); + OracleStructUtil.populateRecordField(sourceRecord, connection, builder, intervalDsField, intervalDs); + OracleStructUtil.populateRecordField(sourceRecord, connection, builder, intervalYmField, intervalYm); + OracleStructUtil.populateRecordField(sourceRecord, connection, builder, bfileField, bfileMock); + OracleStructUtil.populateRecordField(realSourceRecord, connection, builder, + defaultField, "plainString"); + + StructuredRecord record = builder.build(); + Assert.assertEquals("", record.get("XML_FIELD")); + Assert.assertEquals(intervalDs.toString(), record.get("IDS_FIELD")); + Assert.assertEquals(intervalYm.toString(), record.get("IYM_FIELD")); + Assert.assertArrayEquals(bfileBytes, record.get("BFILE_FIELD")); + Assert.assertEquals("plainString", record.get("STR_FIELD")); + } + + @Test + public void populateRecordField_additionalNumericAndTemporalFields_setsFieldsCorrectly() throws Exception { + Schema.Field floatField = Schema.Field.of("FLOAT_FIELD", Schema.of(Schema.Type.FLOAT)); + Schema.Field longField = Schema.Field.of("LONG_FIELD", Schema.of(Schema.Type.LONG)); + Schema.Field dateTsField = Schema.Field.of("DATE_TS_FIELD", Schema.of(Schema.LogicalType.DATE)); + Schema.Field strTsField = Schema.Field.of("STR_TS_FIELD", Schema.of(Schema.Type.STRING)); + Schema.Field fallbackTsField = Schema.Field.of("FALLBACK_TS_FIELD", + Schema.of(Schema.LogicalType.TIMESTAMP_MICROS)); + Schema.Field millisOdtField = Schema.Field.of("MILLIS_ODT_FIELD", + Schema.of(Schema.LogicalType.TIMESTAMP_MILLIS)); + Schema schema = Schema.recordOf("extraBranchesRecord", floatField, longField, + dateTsField, strTsField, fallbackTsField, millisOdtField); + StructuredRecord.Builder builder = StructuredRecord.builder(schema); + OracleSourceDBRecord realSourceRecord = new OracleSourceDBRecord(null, null); + BigDecimal bigDecimalVal = new BigDecimal("123.45"); + Timestamp timestamp = Timestamp.valueOf("2024-05-15 10:30:00"); + OffsetDateTime offsetDateTime = OffsetDateTime.of(2024, 5, 15, + 10, 30, 0, 0, ZoneOffset.ofHours(5)); + + OracleStructUtil.populateRecordField(realSourceRecord, connection, builder, floatField, bigDecimalVal); + OracleStructUtil.populateRecordField(realSourceRecord, connection, builder, longField, bigDecimalVal); + OracleStructUtil.populateRecordField(realSourceRecord, connection, builder, dateTsField, timestamp); + OracleStructUtil.populateRecordField(realSourceRecord, connection, builder, strTsField, timestamp); + OracleStructUtil.populateRecordField(realSourceRecord, connection, builder, fallbackTsField, timestamp); + OracleStructUtil.populateRecordField(realSourceRecord, connection, builder, millisOdtField, offsetDateTime); + + StructuredRecord record = builder.build(); + Assert.assertEquals(123.45f, record.get("FLOAT_FIELD"), 0.001f); + Assert.assertEquals(Long.valueOf(123L), record.get("LONG_FIELD")); + Assert.assertEquals(LocalDate.of(2024, 5, 15), record.getDate("DATE_TS_FIELD")); + Assert.assertEquals(timestamp.toString(), record.get("STR_TS_FIELD")); + Assert.assertEquals(timestamp.toInstant().atZone(ZoneId.ofOffset("UTC", ZoneOffset.UTC)), + record.getTimestamp("FALLBACK_TS_FIELD")); + Assert.assertEquals(offsetDateTime.atZoneSameInstant(ZoneId.of("UTC")), + record.getTimestamp("MILLIS_ODT_FIELD")); + } +}