Skip to content

Commit e6834e3

Browse files
vanshikaagupta22vanshikaagupta22
committed
Fix data reading errors
1 parent 231c6ee commit e6834e3

4 files changed

Lines changed: 129 additions & 170 deletions

File tree

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

Lines changed: 89 additions & 22 deletions
Original file line numberDiff line numberDiff line change
@@ -30,6 +30,8 @@
3030
import java.lang.reflect.InvocationTargetException;
3131
import java.math.BigDecimal;
3232
import java.nio.ByteBuffer;
33+
import java.sql.Blob;
34+
import java.sql.Clob;
3335
import java.sql.Connection;
3436
import java.sql.PreparedStatement;
3537
import java.sql.ResultSet;
@@ -258,6 +260,31 @@ private byte[] getBfileBytes(ResultSet resultSet, String columnName) throws SQLE
258260
}
259261
}
260262

263+
private byte[] getBfileBytes(Object bfile) throws SQLException {
264+
if (bfile == null) {
265+
return null;
266+
}
267+
try {
268+
ClassLoader classLoader = bfile.getClass().getClassLoader();
269+
Class<?> oracleBfileClass = classLoader.loadClass("oracle.jdbc.OracleBfile");
270+
boolean isFileExist = (boolean) oracleBfileClass.getMethod("fileExists").invoke(bfile);
271+
if (!isFileExist) {
272+
return null;
273+
}
274+
275+
oracleBfileClass.getMethod("openFile").invoke(bfile);
276+
InputStream binaryStream = (InputStream) oracleBfileClass.getMethod("getBinaryStream").invoke(bfile);
277+
byte[] bytes = ByteStreams.toByteArray(binaryStream);
278+
oracleBfileClass.getMethod("closeFile").invoke(bfile);
279+
return bytes;
280+
} catch (ClassNotFoundException | InvocationTargetException | NoSuchMethodException | IllegalAccessException e) {
281+
throw new InvalidStageException("Field is of type 'BFILE', which is not supported " +
282+
"with this version of the JDBC driver.", e);
283+
} catch (IOException e) {
284+
throw new InvalidStageException("Error reading the contents of the BFILE.", e);
285+
}
286+
}
287+
261288
private void handleOracleSpecificType(ResultSet resultSet, StructuredRecord.Builder recordBuilder, Schema.Field field,
262289
int columnIndex, int sqlType, int precision, int scale)
263290
throws SQLException {
@@ -343,10 +370,9 @@ private void handleOracleSpecificType(ResultSet resultSet, StructuredRecord.Buil
343370
recordBuilder.set(field.getName(), resultSet.getBytes(columnIndex));
344371
break;
345372
case Types.STRUCT:
346-
java.sql.Struct structValue = (java.sql.Struct) resultSet.getObject(columnIndex);
373+
Struct structValue = (Struct) resultSet.getObject(columnIndex);
347374
if (structValue != null) {
348-
recordBuilder.set(field.getName(), convertStructToRecord(structValue, nonNullSchema,
349-
resultSet.getStatement().getConnection()));
375+
recordBuilder.set(field.getName(), convertStructToRecord(structValue, nonNullSchema, resultSet));
350376
}
351377
break;
352378
case Types.DECIMAL:
@@ -379,39 +405,80 @@ private void handleOracleSpecificType(ResultSet resultSet, StructuredRecord.Buil
379405
}
380406
}
381407

382-
private StructuredRecord convertStructToRecord(java.sql.Struct struct, Schema schema,
383-
Connection connection) throws SQLException {
408+
private StructuredRecord convertStructToRecord(Struct struct, Schema schema, ResultSet resultSet)
409+
throws SQLException {
384410
Object[] attributes = struct.getAttributes();
385411
List<Schema.Field> fields = schema.getFields();
386412
StructuredRecord.Builder builder = StructuredRecord.builder(schema);
387413

388-
for (int i = 0; i < fields.size() && i < attributes.length; i++) {
389-
Schema.Field field = fields.get(i);
390-
Object attrValue = attributes[i];
414+
for (int index = 0; index < attributes.length; index++) {
415+
Schema.Field field = fields.get(index);
416+
Object attrValue = attributes[index];
391417

392418
if (attrValue == null) {
393419
builder.set(field.getName(), null);
394420
continue;
395421
}
396-
397-
Schema fieldSchema = field.getSchema().isNullable()
398-
? field.getSchema().getNonNullable() : field.getSchema();
399-
422+
// If it is an internal nested STRUCT, recurse down
400423
if (attrValue instanceof Struct) {
401-
builder.set(field.getName(), convertStructToRecord((Struct) attrValue, fieldSchema, connection));
402-
} else if (attrValue instanceof java.sql.Date) {
403-
builder.setDate(field.getName(), ((java.sql.Date) attrValue).toLocalDate());
404-
} else if (attrValue instanceof java.sql.Time) {
405-
builder.setTime(field.getName(), ((java.sql.Time) attrValue).toLocalTime());
424+
Schema fieldSchema = field.getSchema().isNullable() ? field.getSchema().getNonNullable() : field.getSchema();
425+
builder.set(field.getName(), convertStructToRecord((Struct) attrValue, fieldSchema, resultSet));
426+
continue;
427+
}
428+
429+
String attrClassName = attrValue.getClass().getName();
430+
Schema fieldSchema = field.getSchema().isNullable() ? field.getSchema().getNonNullable() : field.getSchema();
431+
if (attrValue instanceof BigDecimal) {
432+
if (Schema.LogicalType.DECIMAL.equals(fieldSchema.getLogicalType())) {
433+
builder.setDecimal(field.getName(), ((BigDecimal) attrValue).setScale(getScale(field.getSchema()),
434+
java.math.RoundingMode.HALF_UP));
435+
} else if (Schema.Type.DOUBLE.equals(fieldSchema.getType())) {
436+
builder.set(field.getName(), ((BigDecimal) attrValue).doubleValue());
437+
} else if (Schema.Type.FLOAT.equals(fieldSchema.getType())) {
438+
builder.set(field.getName(), ((BigDecimal) attrValue).floatValue());
439+
} else if (Schema.Type.INT.equals(fieldSchema.getType())) {
440+
builder.set(field.getName(), ((BigDecimal) attrValue).intValue());
441+
} else if (Schema.Type.LONG.equals(fieldSchema.getType())) {
442+
builder.set(field.getName(), ((BigDecimal) attrValue).longValue());
443+
} else {
444+
builder.set(field.getName(), attrValue.toString());
445+
}
406446
} else if (attrValue instanceof Timestamp) {
447+
Timestamp timestamp = (Timestamp) attrValue;
407448
if (Schema.LogicalType.DATETIME.equals(fieldSchema.getLogicalType())) {
408-
builder.setDateTime(field.getName(), ((Timestamp) attrValue).toLocalDateTime());
449+
builder.setDateTime(field.getName(), timestamp.toLocalDateTime());
450+
} else if (Schema.LogicalType.DATE.equals(fieldSchema.getLogicalType())) {
451+
builder.setDate(field.getName(), timestamp.toLocalDateTime().toLocalDate());
452+
} else {
453+
builder.set(field.getName(), attrValue.toString());
454+
}
455+
} else if (attrValue instanceof OffsetDateTime || attrValue instanceof ZonedDateTime) {
456+
ZonedDateTime zonedDateTime = (attrValue instanceof OffsetDateTime)
457+
? ((OffsetDateTime) attrValue).atZoneSameInstant(ZoneId.of("UTC"))
458+
: ((ZonedDateTime) attrValue).withZoneSameInstant(ZoneId.of("UTC"));
459+
if (fieldSchema.getLogicalType() != null &&
460+
(Schema.LogicalType.TIMESTAMP_MICROS.equals(fieldSchema.getLogicalType()) ||
461+
Schema.LogicalType.TIMESTAMP_MILLIS.equals(fieldSchema.getLogicalType()))) {
462+
builder.setTimestamp(field.getName(), zonedDateTime);
463+
} else if (Schema.Type.LONG.equals(fieldSchema.getType())) {
464+
builder.set(field.getName(), zonedDateTime.toInstant().toEpochMilli());
409465
} else {
410-
builder.setTimestamp(field.getName(),
411-
((Timestamp) attrValue).toInstant().atZone(java.time.ZoneId.of("UTC")));
466+
builder.set(field.getName(), zonedDateTime.toString());
412467
}
413-
} else if (attrValue instanceof BigDecimal) {
414-
builder.setDecimal(field.getName(), (BigDecimal) attrValue);
468+
} else if (attrValue instanceof Clob) {
469+
Clob clob = (Clob) attrValue;
470+
builder.set(field.getName(), clob.getSubString(1, (int) clob.length()));
471+
} else if (attrValue instanceof Blob) {
472+
Blob blob = (Blob) attrValue;
473+
builder.set(field.getName(), blob.getBytes(1, (int) blob.length()));
474+
} else if ("oracle.jdbc.OracleBfile".equals(attrClassName)) {
475+
builder.set(field.getName(), getBfileBytes(attrValue));
476+
} else if (attrValue instanceof byte[]) {
477+
byte[] bytesValue = (byte[]) attrValue;
478+
builder.set(field.getName(), bytesValue);
479+
} else if ("oracle.sql.INTERVALDS".equals(attrClassName)
480+
|| "oracle.sql.INTERVALYM".equals(attrClassName)) {
481+
builder.set(field.getName(), attrValue.toString());
415482
} else {
416483
builder.set(field.getName(), attrValue);
417484
}

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

Lines changed: 35 additions & 19 deletions
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,9 @@
1818

1919
import com.google.common.collect.ImmutableSet;
2020
import io.cdap.cdap.api.data.schema.Schema;
21+
import io.cdap.cdap.api.exception.ErrorCategory;
22+
import io.cdap.cdap.api.exception.ErrorType;
23+
import io.cdap.cdap.api.exception.ErrorUtils;
2124
import io.cdap.plugin.db.CommonSchemaReader;
2225
import org.jetbrains.annotations.NotNull;
2326
import org.slf4j.Logger;
@@ -148,8 +151,8 @@ public Schema getSchema(ResultSetMetaData metadata, int index) throws SQLExcepti
148151
+ "Use getSchemaFields(ResultSet) to enable STRUCT type resolution.");
149152
}
150153
String typeName = metadata.getColumnTypeName(index);
151-
String oracleSchemaName = metadata.getSchemaName(index);
152-
return getStructSchema(connection, oracleSchemaName, typeName);
154+
String owner = typeName.substring(0, typeName.lastIndexOf('.'));
155+
return getStructSchema(connection, typeName, owner);
153156
default:
154157
return super.getSchema(metadata, index);
155158
}
@@ -162,22 +165,23 @@ public List<Schema.Field> getSchemaFields(ResultSet resultSet) throws SQLExcepti
162165
}
163166

164167
/**
165-
* Builds a CDAP RECORD schema for an Oracle STRUCT type by querying the database metadata
168+
* Builds a CDAP RECORD schema for an Oracle STRUCT type by querying the
169+
* database metadata
166170
* for the type's attributes.
167171
*
168172
* @param connection the database connection
169-
* @param schemaName the Oracle schema owning the type
170173
* @param typeName the Oracle type name (e.g., "ADDRESS_TYPE")
171-
* @return a CDAP RECORD schema with fields corresponding to the STRUCT's attributes
174+
* @return a CDAP RECORD schema with fields corresponding to the STRUCT's
175+
* attributes
172176
*/
173-
private Schema getStructSchema(Connection connection, String schemaName,
174-
String typeName) throws SQLException {
177+
private Schema getStructSchema(Connection connection, String typeName, String owner) throws SQLException {
175178
List<Schema.Field> fields = new ArrayList<>();
176-
177-
String sql = "SELECT * FROM ALL_TYPE_ATTRS WHERE TYPE_NAME = ? ORDER BY ATTR_NO";
179+
String sql = "SELECT * FROM ALL_TYPE_ATTRS WHERE TYPE_NAME = ? AND OWNER = ? ORDER BY ATTR_NO";
178180

179181
try (PreparedStatement stmt = connection.prepareStatement(sql)) {
182+
180183
stmt.setString(1, typeName.substring(typeName.lastIndexOf('.') + 1));
184+
stmt.setString(2, owner);
181185

182186
try (ResultSet attrRs = stmt.executeQuery()) {
183187
while (attrRs.next()) {
@@ -186,43 +190,49 @@ private Schema getStructSchema(Connection connection, String schemaName,
186190
int attrSize = attrRs.getInt("PRECISION");
187191
int attrScale = attrRs.getInt("SCALE");
188192

189-
Schema attrSchema = mapPrimitiveOracleType(attrTypeName, attrSize, attrScale);
193+
Schema attrSchema = mapPrimitiveOracleType(attrTypeName, attrSize, attrScale, attrName);
190194
if (attrSchema != null) {
191195
fields.add(Schema.Field.of(attrName, attrSchema));
192196
} else {
193-
Schema nestedSchema = getStructSchema(connection, schemaName, attrTypeName);
197+
String nestedStructOwner = attrRs.getString("ATTR_TYPE_OWNER");
198+
Schema nestedSchema = getStructSchema(connection, attrTypeName, nestedStructOwner);
194199
fields.add(Schema.Field.of(attrName, nestedSchema));
195200
}
196201
}
197202
}
198203
}
199204
if (fields.isEmpty()) {
200205
throw new SQLException(String.format(
201-
"No attributes found for Oracle STRUCT type '%s.%s'. "
202-
+ "Ensure the type exists and is accessible.",
203-
schemaName, typeName));
206+
"No attributes found for Oracle STRUCT type '%s'. "
207+
+ "Ensure the type exists and is accessible.",
208+
typeName));
204209
}
205210

206211
return Schema.recordOf(typeName, fields);
207212
}
208213

209-
private Schema mapPrimitiveOracleType(String typeName, int precision, int scale) {
214+
private Schema mapPrimitiveOracleType(String typeName, int precision, int scale, String columnName) {
210215
switch (typeName) {
211216
case "TIMESTAMP WITH TZ":
212217
return isTimestampOldBehavior ? Schema.of(Schema.Type.STRING) : Schema.of(Schema.LogicalType.TIMESTAMP_MICROS);
213218
case "TIMESTAMP WITH LTZ":
214219
return getTimestampLtzSchema();
215220
case "TIMESTAMP":
216-
return Schema.of(Schema.LogicalType.DATETIME);
217-
case "DATE" :
221+
return isTimestampOldBehavior ?
222+
Schema.of(Schema.LogicalType.TIMESTAMP_MICROS) : Schema.of(Schema.LogicalType.DATETIME);
223+
case "DATE":
218224
return Schema.of(Schema.LogicalType.DATE);
225+
case "TIME":
226+
return Schema.of(Schema.LogicalType.TIME_MICROS);
219227
case "BINARY FLOAT":
228+
case "REAL":
220229
case "FLOAT":
221230
return Schema.of(Schema.Type.FLOAT);
222231
case "BINARY DOUBLE":
223232
case "DOUBLE":
224233
return Schema.of(Schema.Type.DOUBLE);
225234
case "BFILE":
235+
case "BLOB":
226236
case "RAW":
227237
case "LONG RAW":
228238
return Schema.of(Schema.Type.BYTES);
@@ -231,15 +241,15 @@ private Schema mapPrimitiveOracleType(String typeName, int precision, int scale)
231241
case "VARCHAR2":
232242
case "VARCHAR":
233243
case "CHAR":
244+
case "CHAR2":
234245
case "CLOB":
235-
case "BLOB":
246+
case "NCLOB":
236247
case "LONG":
237248
return Schema.of(Schema.Type.STRING);
238249
case "INTEGER":
239250
return Schema.of(Schema.Type.INT);
240251
case "NUMBER":
241252
case "DECIMAL":
242-
// FLOAT and REAL are returned as java.sql.Types.NUMERIC but with value that is a java.lang.Double
243253
if (Double.class.getTypeName().equals(typeName)) {
244254
return Schema.of(Schema.Type.DOUBLE);
245255
} else {
@@ -261,6 +271,12 @@ private Schema mapPrimitiveOracleType(String typeName, int precision, int scale)
261271
}
262272
return Schema.decimalOf(precision, scale);
263273
}
274+
case "ARRAY":
275+
case "OTHER":
276+
case "XML":
277+
String errorMessage = String.format("Column %s has unsupported SQL type of %s.", columnName, typeName);
278+
throw ErrorUtils.getProgramFailureException(new ErrorCategory(ErrorCategory.ErrorCategoryEnum.PLUGIN),
279+
errorMessage, errorMessage, ErrorType.SYSTEM, true, null);
264280
default:
265281
return null;
266282
}

‎oracle-plugin/src/test/java/io/cdap/plugin/oracle/OracleSchemaReaderTest.java‎

Lines changed: 5 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -20,9 +20,7 @@
2020
import io.cdap.cdap.api.data.schema.Schema;
2121
import org.junit.Assert;
2222
import org.junit.Test;
23-
import org.junit.runner.RunWith;
2423
import org.mockito.Mockito;
25-
import org.mockito.junit.MockitoJUnitRunner;
2624

2725
import java.sql.Connection;
2826
import java.sql.PreparedStatement;
@@ -108,28 +106,22 @@ public void getSchema_timestampLTZFieldFalse_returnDatetime() throws SQLExceptio
108106
@Test
109107
public void getSchemaFields_structType_returnRecord() throws SQLException {
110108
OracleSourceSchemaReader schemaReader = new OracleSourceSchemaReader();
111-
112109
ResultSet resultSet = Mockito.mock(ResultSet.class);
113110
ResultSetMetaData metadata = Mockito.mock(ResultSetMetaData.class);
114111
Statement statement = Mockito.mock(Statement.class);
115112
Connection connection = Mockito.mock(Connection.class);
116113
PreparedStatement stmt = Mockito.mock(PreparedStatement.class);
117114
ResultSet attrRs = Mockito.mock(ResultSet.class);
118-
119115
Mockito.when(resultSet.getMetaData()).thenReturn(metadata);
120116
Mockito.when(resultSet.getStatement()).thenReturn(statement);
121117
Mockito.when(statement.getConnection()).thenReturn(connection);
122118
Mockito.when(connection.prepareStatement(Mockito.anyString())).thenReturn(stmt);
123119
Mockito.when(stmt.executeQuery()).thenReturn(attrRs);
124-
125-
// One STRUCT column
126120
Mockito.when(metadata.getColumnCount()).thenReturn(1);
127121
Mockito.when(metadata.getColumnType(1)).thenReturn(Types.STRUCT);
128122
Mockito.when(metadata.getColumnName(1)).thenReturn("address");
129-
Mockito.when(metadata.getColumnTypeName(1)).thenReturn("ADDRESS_TYPE");
123+
Mockito.when(metadata.getColumnTypeName(1)).thenReturn("CS_ITN.ADDRESS_TYPE");
130124
Mockito.when(metadata.getSchemaName(1)).thenReturn("TEST_SCHEMA");
131-
132-
// Mock ALL_TYPE_ATTRS for ADDRESS_TYPE with two VARCHAR2 attributes
133125
Mockito.when(attrRs.next()).thenReturn(true, true, false);
134126
Mockito.when(attrRs.getString("ATTR_NAME")).thenReturn("STREET", "CITY");
135127
Mockito.when(attrRs.getString("ATTR_TYPE_NAME")).thenReturn("VARCHAR2", "VARCHAR2");
@@ -138,16 +130,14 @@ public void getSchemaFields_structType_returnRecord() throws SQLException {
138130

139131
List<Schema.Field> actualFields = schemaReader.getSchemaFields(resultSet);
140132

141-
Assert.assertEquals(1, actualFields.size());
142133
Schema.Field addressField = actualFields.get(0);
143-
Assert.assertEquals("address", addressField.getName());
144-
145134
Schema addressSchema = addressField.getSchema().isNullable()
146135
? addressField.getSchema().getNonNullable() : addressField.getSchema();
147-
Assert.assertEquals(Schema.Type.RECORD, addressSchema.getType());
148-
Assert.assertEquals("ADDRESS_TYPE", addressSchema.getRecordName());
149-
150136
List<Schema.Field> structFields = addressSchema.getFields();
137+
Assert.assertEquals(1, actualFields.size());
138+
Assert.assertEquals("address", addressField.getName());
139+
Assert.assertEquals(Schema.Type.RECORD, addressSchema.getType());
140+
Assert.assertEquals("CS_ITN.ADDRESS_TYPE", addressSchema.getRecordName());
151141
Assert.assertEquals(2, structFields.size());
152142
Assert.assertEquals("STREET", structFields.get(0).getName());
153143
Assert.assertEquals("CITY", structFields.get(1).getName());

0 commit comments

Comments
 (0)