Skip to content

Commit d24127c

Browse files
vanshikaagupta22vanshikaagupta22
committed
Minor fixes
1 parent 98f58ff commit d24127c

5 files changed

Lines changed: 462 additions & 173 deletions

File tree

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

Lines changed: 15 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -36,6 +36,7 @@
3636
import java.math.BigDecimal;
3737
import java.math.BigInteger;
3838
import java.nio.ByteBuffer;
39+
import java.sql.Connection;
3940
import java.sql.Date;
4041
import java.sql.PreparedStatement;
4142
import java.sql.ResultSet;
@@ -187,15 +188,23 @@ protected void handleField(ResultSet resultSet, StructuredRecord.Builder recordB
187188

188189
protected void setField(ResultSet resultSet, StructuredRecord.Builder recordBuilder, Schema.Field field,
189190
int columnIndex, int sqlType, int sqlPrecision, int sqlScale) throws SQLException {
190-
Object o = DBUtils.transformValue(sqlType, sqlPrecision, sqlScale, resultSet, columnIndex);
191-
setFieldValue(recordBuilder, field, o);
191+
Object fieldValue = DBUtils.transformValue(sqlType, sqlPrecision, sqlScale, resultSet, columnIndex);
192+
populateRecordField(resultSet.getStatement().getConnection(), recordBuilder, field, fieldValue);
192193
}
193194

194-
protected void setFieldValue(StructuredRecord.Builder recordBuilder, Schema.Field field, Object o)
195+
/**
196+
* Populates the value of a field in the {@link StructuredRecord.Builder}.
197+
*
198+
* @param connection the SQL connection, provided for subclass overrides that require database connection
199+
* @param recordBuilder the builder for constructing the {@link StructuredRecord}
200+
* @param field the field to set in the record
201+
* @param o the object value read from the database
202+
* @throws SQLException if an error occurs while setting the field value
203+
*/
204+
protected void populateRecordField(Connection connection, StructuredRecord.Builder recordBuilder,
205+
Schema.Field field, Object o)
195206
throws SQLException {
196-
if (o == null) {
197-
recordBuilder.set(field.getName(), null);
198-
} else if (o instanceof Date) {
207+
if (o instanceof Date) {
199208
recordBuilder.setDate(field.getName(), ((Date) o).toLocalDate());
200209
} else if (o instanceof Time) {
201210
recordBuilder.setTime(field.getName(), ((Time) o).toLocalTime());

‎oracle-plugin/src/main/java/io/cdap/plugin/oracle/OracleSourceDBRecord.java‎

Lines changed: 79 additions & 51 deletions
Original file line numberDiff line numberDiff line change
@@ -113,15 +113,22 @@ record = recordBuilder.build();
113113
@Override
114114
protected void handleField(ResultSet resultSet, StructuredRecord.Builder recordBuilder, Schema.Field field,
115115
int columnIndex, int sqlType, int sqlPrecision, int sqlScale) throws SQLException {
116-
if (OracleSourceSchemaReader.ORACLE_TYPES.contains(sqlType) || sqlType == Types.NCLOB || sqlType == Types.STRUCT) {
116+
if (isOracleSpecificType(sqlType)) {
117117
handleOracleSpecificType(resultSet, recordBuilder, field, columnIndex, sqlType, sqlPrecision, sqlScale);
118118
} else {
119119
setField(resultSet, recordBuilder, field, columnIndex, sqlType, sqlPrecision, sqlScale);
120120
}
121121
}
122122

123+
protected boolean isOracleSpecificType(int sqlType) {
124+
return OracleSourceSchemaReader.ORACLE_TYPES.contains(sqlType)
125+
|| sqlType == Types.NCLOB
126+
|| sqlType == Types.STRUCT;
127+
}
128+
123129
@Override
124-
protected void setFieldValue(StructuredRecord.Builder recordBuilder, Schema.Field field, Object attrValue)
130+
protected void populateRecordField(Connection connection, StructuredRecord.Builder recordBuilder,
131+
Schema.Field field, Object attrValue)
125132
throws SQLException {
126133
if (attrValue == null) {
127134
recordBuilder.set(field.getName(), null);
@@ -132,7 +139,7 @@ protected void setFieldValue(StructuredRecord.Builder recordBuilder, Schema.Fiel
132139
: field.getSchema();
133140
String attrClassName = attrValue.getClass().getName();
134141
if (attrValue instanceof Struct) {
135-
recordBuilder.set(field.getName(), convertStructToRecord((Struct) attrValue, fieldSchema));
142+
recordBuilder.set(field.getName(), convertStructToRecord((Struct) attrValue, fieldSchema, connection));
136143
return;
137144
}
138145
if (attrValue instanceof Clob) {
@@ -154,42 +161,61 @@ protected void setFieldValue(StructuredRecord.Builder recordBuilder, Schema.Fiel
154161
return;
155162
}
156163
if (attrValue instanceof BigDecimal) {
157-
populateDecimalValue(attrValue, fieldSchema, recordBuilder, field);
164+
handleDecimalValue((BigDecimal) attrValue, fieldSchema, recordBuilder, field);
158165
return;
159166
}
160-
161167
if (attrValue instanceof Timestamp) {
162-
Timestamp timestamp = (Timestamp) attrValue;
163-
if (Schema.LogicalType.DATETIME.equals(fieldSchema.getLogicalType())) {
164-
recordBuilder.setDateTime(field.getName(), timestamp.toLocalDateTime());
165-
} else {
166-
super.setFieldValue(recordBuilder, field, attrValue);
167-
}
168+
handleTimestampValue((Timestamp) attrValue, fieldSchema, recordBuilder, field, connection);
168169
return;
169170
}
170171
if (attrValue instanceof OffsetDateTime) {
171-
ZonedDateTime zonedDateTime = ((OffsetDateTime) attrValue).atZoneSameInstant(ZoneId.of("UTC"));
172-
173-
if (fieldSchema.getLogicalType() != null &&
174-
(Schema.LogicalType.TIMESTAMP_MICROS.equals(fieldSchema.getLogicalType()) ||
175-
Schema.LogicalType.TIMESTAMP_MILLIS.equals(fieldSchema.getLogicalType()))) {
176-
recordBuilder.setTimestamp(field.getName(), zonedDateTime);
177-
} else {
178-
recordBuilder.set(field.getName(), zonedDateTime.toString());
179-
}
172+
handleOffsetDateTimeValue((OffsetDateTime) attrValue, fieldSchema, recordBuilder, field);
173+
return;
174+
}
175+
if (isBfileValue(attrValue, field.getName())) {
176+
recordBuilder.set(field.getName(), getBfileBytes(attrValue, field.getName()));
180177
return;
181178
}
182179

180+
super.populateRecordField(connection, recordBuilder, field, attrValue);
181+
}
182+
183+
private void handleTimestampValue(Timestamp timestamp, Schema fieldSchema,
184+
StructuredRecord.Builder recordBuilder, Schema.Field field,
185+
Connection connection) throws SQLException {
186+
if (Schema.LogicalType.DATETIME.equals(fieldSchema.getLogicalType())) {
187+
recordBuilder.setDateTime(field.getName(), timestamp.toLocalDateTime());
188+
} else {
189+
super.populateRecordField(connection, recordBuilder, field, timestamp);
190+
}
191+
}
192+
193+
private void handleOffsetDateTimeValue(OffsetDateTime offsetDateTime, Schema fieldSchema,
194+
StructuredRecord.Builder recordBuilder, Schema.Field field) {
195+
ZonedDateTime zonedDateTime = offsetDateTime.atZoneSameInstant(ZoneId.of("UTC"));
196+
if (fieldSchema.getLogicalType() != null &&
197+
(Schema.LogicalType.TIMESTAMP_MICROS.equals(fieldSchema.getLogicalType()) ||
198+
Schema.LogicalType.TIMESTAMP_MILLIS.equals(fieldSchema.getLogicalType()))) {
199+
recordBuilder.setTimestamp(field.getName(), zonedDateTime);
200+
} else if (Schema.LogicalType.DATETIME.equals(fieldSchema.getLogicalType())) {
201+
LocalDateTime systemLocalDateTime = offsetDateTime.atZoneSameInstant(ZoneId.systemDefault()).toLocalDateTime();
202+
recordBuilder.setDateTime(field.getName(), systemLocalDateTime);
203+
} else {
204+
recordBuilder.set(field.getName(), zonedDateTime.toString());
205+
}
206+
}
207+
208+
private boolean isBfileValue(Object attrValue, String fieldName) throws SQLException {
183209
ClassLoader oracleLoader = attrValue.getClass().getClassLoader();
184-
try {
185-
if (oracleLoader != null && oracleLoader.loadClass("oracle.jdbc.OracleBfile").isInstance(attrValue)) {
186-
recordBuilder.set(field.getName(), getBfileBytes(attrValue, field.getName()));
187-
return;
188-
}
189-
} catch (ClassNotFoundException e) {
190-
throw new RuntimeException(e);
210+
try {
211+
if (oracleLoader != null && oracleLoader.loadClass("oracle.jdbc.OracleBfile").isInstance(attrValue)) {
212+
return true;
191213
}
192-
super.setFieldValue(recordBuilder, field, attrValue);
214+
} catch (ClassNotFoundException e) {
215+
throw new SQLException(String.format("Column '%s' is of type 'BFILE', which is not supported with " +
216+
"this version of the JDBC driver.", fieldName), e);
217+
}
218+
return false;
193219
}
194220

195221
@Override
@@ -248,9 +274,8 @@ protected void writeNonNullToDB(PreparedStatement stmt, Schema fieldSchema,
248274
}
249275
}
250276

251-
private void populateDecimalValue(Object attrValue, Schema fieldSchema,
277+
private void handleDecimalValue(BigDecimal bigDecimal, Schema fieldSchema,
252278
StructuredRecord.Builder recordBuilder, Schema.Field field) {
253-
BigDecimal bigDecimal = (BigDecimal) attrValue;
254279
if (Schema.LogicalType.DECIMAL.equals(fieldSchema.getLogicalType())) {
255280
recordBuilder.setDecimal(field.getName(), bigDecimal.setScale(fieldSchema.getScale(), RoundingMode.HALF_UP));
256281
return;
@@ -269,7 +294,7 @@ private void populateDecimalValue(Object attrValue, Schema fieldSchema,
269294
recordBuilder.set(field.getName(), bigDecimal.longValue());
270295
break;
271296
case STRING:
272-
recordBuilder.set(field.getName(), bigDecimal.toString());
297+
recordBuilder.set(field.getName(), bigDecimal.toPlainString());
273298
break;
274299
default:
275300
recordBuilder.set(field.getName(), bigDecimal);
@@ -455,7 +480,10 @@ private void handleOracleSpecificType(ResultSet resultSet, StructuredRecord.Buil
455480
case Types.STRUCT:
456481
Struct structValue = (Struct) resultSet.getObject(columnIndex);
457482
if (structValue != null) {
458-
recordBuilder.set(field.getName(), convertStructToRecord(structValue, nonNullSchema));
483+
recordBuilder.set(field.getName(), convertStructToRecord(structValue, nonNullSchema,
484+
resultSet.getStatement().getConnection()));
485+
} else {
486+
recordBuilder.set(field.getName(), null);
459487
}
460488
break;
461489
case Types.DECIMAL:
@@ -493,17 +521,18 @@ private void handleOracleSpecificType(ResultSet resultSet, StructuredRecord.Buil
493521
*
494522
* @param struct the SQL structured type containing the source data attributes
495523
* @param schema the target record schema defining the fields to map
524+
* @param connection the database connection
496525
* @return a populated {@code StructuredRecord} instance
497526
* @throws SQLException if an error occurs reading the struct attributes or metadata
498527
*/
499-
protected StructuredRecord convertStructToRecord(Struct struct, Schema schema)
528+
protected StructuredRecord convertStructToRecord(Struct struct, Schema schema, Connection connection)
500529
throws SQLException {
501-
Map<String, Object> attributeMap = getAttributeMap(struct, schema);
530+
Map<String, Object> attributeMap = getAttributeMap(struct, schema, connection);
502531
StructuredRecord.Builder builder = StructuredRecord.builder(schema);
503532

504533
for (Schema.Field field : schema.getFields()) {
505534
Object attrValue = attributeMap.get(field.getName());
506-
setFieldValue(builder, field, attrValue);
535+
populateRecordField(connection, builder, field, attrValue);
507536
}
508537
return builder.build();
509538
}
@@ -514,29 +543,28 @@ protected StructuredRecord convertStructToRecord(Struct struct, Schema schema)
514543
*
515544
* @param struct the source SQL structured type
516545
* @param schema the target schema used for context in error messages
546+
* @param connection the database connection
517547
* @return a case-insensitive {@code Map} linking column names to their attribute values
518548
* @throws SQLException if metadata extraction fails or driver-specific methods are inaccessible
519549
*/
520-
private Map<String, Object> getAttributeMap(Struct struct, Schema schema) throws SQLException {
550+
protected Map<String, Object> getAttributeMap(Struct struct, Schema schema, Connection connection)
551+
throws SQLException {
521552
Map<String, Object> attributeMap = new TreeMap<>(String.CASE_INSENSITIVE_ORDER);
522553
Object[] attributes = struct.getAttributes();
523-
524-
try {
525-
Object descriptor = struct.getClass().getMethod("getDescriptor").invoke(struct);
526-
ResultSetMetaData metaData =
527-
(ResultSetMetaData) descriptor.getClass().getMethod("getMetaData").invoke(descriptor);
528-
for (int i = 1; i <= metaData.getColumnCount() && (i - 1) < attributes.length; i++) {
529-
attributeMap.put(metaData.getColumnName(i), attributes[i - 1]);
554+
if (attributes != null) {
555+
try {
556+
Object descriptor = struct.getClass().getMethod("getDescriptor").invoke(struct);
557+
ResultSetMetaData metaData =
558+
(ResultSetMetaData) descriptor.getClass().getMethod("getMetaData").invoke(descriptor);
559+
for (int i = 1; i <= metaData.getColumnCount() && (i - 1) < attributes.length; i++) {
560+
attributeMap.put(metaData.getColumnName(i), attributes[i - 1]);
561+
}
562+
} catch (Exception e) {
563+
throw new SQLException(String.format("Failed to retrieve attribute metadata for Oracle STRUCT schema '%s': %s",
564+
schema.getRecordName(), e.getMessage()));
530565
}
531-
} catch (SQLException | NoSuchMethodException e) {
532-
throw new SQLException(String.format("Failed to retrieve attribute metadata for Oracle STRUCT schema '%s': %s",
533-
schema.getRecordName(), e.getMessage()), e);
534-
} catch (InvocationTargetException | IllegalAccessException e) {
535-
throw new SQLException(String.format("Unable to retrieve attribute metadata for Oracle STRUCT schema '%s'. "
536-
+ "Ensure the Oracle JDBC driver supports JDBC StructDescriptor metadata.",
537-
schema.getRecordName()), e);
538566
}
539-
return attributeMap;
567+
return attributeMap;
540568
}
541569

542570
/**

0 commit comments

Comments
 (0)