Skip to content

Commit d8f92c3

Browse files
vanshikaagupta22vanshikaagupta22
committed
Fixed data type mapping
1 parent a33a1d3 commit d8f92c3

2 files changed

Lines changed: 38 additions & 59 deletions

File tree

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

Lines changed: 35 additions & 55 deletions
Original file line numberDiff line numberDiff line change
@@ -29,6 +29,7 @@
2929
import java.io.InputStream;
3030
import java.lang.reflect.InvocationTargetException;
3131
import java.math.BigDecimal;
32+
import java.math.RoundingMode;
3233
import java.nio.ByteBuffer;
3334
import java.sql.Blob;
3435
import java.sql.Clob;
@@ -37,6 +38,7 @@
3738
import java.sql.ResultSet;
3839
import java.sql.ResultSetMetaData;
3940
import java.sql.SQLException;
41+
import java.sql.SQLXML;
4042
import java.sql.Struct;
4143
import java.sql.Timestamp;
4244
import java.sql.Types;
@@ -124,105 +126,84 @@ protected void setFieldValue(StructuredRecord.Builder recordBuilder, Schema.Fiel
124126
return;
125127
}
126128

127-
Schema fieldSchema = field.getSchema().isNullable() ? field.getSchema().getNonNullable() : field.getSchema();
129+
Schema fieldSchema = field.getSchema().isNullable() ? field.getSchema().getNonNullable()
130+
: field.getSchema();
128131
String attrClassName = attrValue.getClass().getName();
129-
130-
// Handle Nested Structs Recursively
131132
if (attrValue instanceof Struct) {
132-
recordBuilder.set(field.getName(), convertStructToRecord((Struct) attrValue, fieldSchema, null));
133+
recordBuilder.set(field.getName(), convertStructToRecord((Struct) attrValue, fieldSchema));
133134
return;
134135
}
135-
136-
// Handle Oracle UDTs (Clobs, Blobs, SQLXML, INTERVALS)
137136
if (attrValue instanceof Clob) {
138137
Clob clob = (Clob) attrValue;
139138
recordBuilder.set(field.getName(), clob.getSubString(1, (int) clob.length()));
140139
return;
141140
}
142-
143141
if (attrValue instanceof Blob) {
144142
Blob blob = (Blob) attrValue;
145143
recordBuilder.set(field.getName(), blob.getBytes(1, (int) blob.length()));
146144
return;
147145
}
148-
149-
if (attrValue instanceof java.sql.SQLXML) {
150-
recordBuilder.set(field.getName(), ((java.sql.SQLXML) attrValue).getString());
146+
if (attrValue instanceof SQLXML) {
147+
recordBuilder.set(field.getName(), ((SQLXML) attrValue).getString());
151148
return;
152149
}
153-
154150
if ("oracle.sql.INTERVALDS".equals(attrClassName) || "oracle.sql.INTERVALYM".equals(attrClassName)) {
155151
recordBuilder.set(field.getName(), attrValue.toString());
156152
return;
157153
}
158-
159-
// Handle Oracle's lazy BigDecimals and downcast them
154+
if ("oracle.sql.NCLOB".equals(attrClassName)) {
155+
recordBuilder.set(field.getName(), attrValue.toString());
156+
return;
157+
}
160158
if (attrValue instanceof BigDecimal) {
161159
BigDecimal bigDecimal = (BigDecimal) attrValue;
162160
if (Schema.LogicalType.DECIMAL.equals(fieldSchema.getLogicalType())) {
163-
recordBuilder.setDecimal(field.getName(), bigDecimal.setScale(fieldSchema.getScale(),
164-
java.math.RoundingMode.HALF_UP));
165-
} else {
166-
switch (fieldSchema.getType()) {
167-
case DOUBLE: recordBuilder.set(field.getName(), bigDecimal.doubleValue()); break;
168-
case FLOAT: recordBuilder.set(field.getName(), bigDecimal.floatValue()); break;
169-
case INT: recordBuilder.set(field.getName(), bigDecimal.intValue()); break;
170-
case LONG: recordBuilder.set(field.getName(), bigDecimal.longValue()); break;
171-
case STRING: recordBuilder.set(field.getName(), bigDecimal.toString()); break;
172-
default: recordBuilder.set(field.getName(), bigDecimal);
173-
}
161+
recordBuilder.setDecimal(field.getName(), bigDecimal.setScale(fieldSchema.getScale(), RoundingMode.HALF_UP));
162+
return;
163+
}
164+
switch (fieldSchema.getType()) {
165+
case DOUBLE: recordBuilder.set(field.getName(), bigDecimal.doubleValue()); break;
166+
case FLOAT: recordBuilder.set(field.getName(), bigDecimal.floatValue()); break;
167+
case INT: recordBuilder.set(field.getName(), bigDecimal.intValue()); break;
168+
case LONG: recordBuilder.set(field.getName(), bigDecimal.longValue()); break;
169+
case STRING: recordBuilder.set(field.getName(), bigDecimal.toString()); break;
170+
default: recordBuilder.set(field.getName(), bigDecimal);
174171
}
175172
return;
176173
}
177174

178-
// 4. Handle Oracle Timestamps to ensure DATETIME schema compatibility
179175
if (attrValue instanceof Timestamp) {
180176
Timestamp timestamp = (Timestamp) attrValue;
181177
if (Schema.LogicalType.DATETIME.equals(fieldSchema.getLogicalType())) {
182178
recordBuilder.setDateTime(field.getName(), timestamp.toLocalDateTime());
183-
} else if (Schema.LogicalType.DATE.equals(fieldSchema.getLogicalType())) {
184-
recordBuilder.setDate(field.getName(), timestamp.toLocalDateTime().toLocalDate());
185-
} else if (fieldSchema.getLogicalType() == null && Schema.Type.STRING.equals(fieldSchema.getType())) {
186-
// Only stringify if the CDAP schema strictly demands a String
187-
recordBuilder.set(field.getName(), attrValue.toString());
188179
} else {
189-
// HAND IT BACK TO THE PARENT! This restores the exact behavior of your old build.
190180
super.setFieldValue(recordBuilder, field, attrValue);
191181
}
192182
return;
193183
}
194-
195-
// Handle Timezone shifting
196-
if (attrValue instanceof OffsetDateTime || attrValue instanceof ZonedDateTime) {
197-
ZonedDateTime zonedDateTime = (attrValue instanceof OffsetDateTime)
198-
? ((OffsetDateTime) attrValue).atZoneSameInstant(ZoneId.of("UTC"))
199-
: ((ZonedDateTime) attrValue).withZoneSameInstant(ZoneId.of("UTC"));
184+
if (attrValue instanceof OffsetDateTime) {
185+
ZonedDateTime zonedDateTime = ((OffsetDateTime) attrValue).atZoneSameInstant(ZoneId.of("UTC"));
200186

201187
if (fieldSchema.getLogicalType() != null &&
202188
(Schema.LogicalType.TIMESTAMP_MICROS.equals(fieldSchema.getLogicalType()) ||
203189
Schema.LogicalType.TIMESTAMP_MILLIS.equals(fieldSchema.getLogicalType()))) {
204190
recordBuilder.setTimestamp(field.getName(), zonedDateTime);
205-
} else if (Schema.Type.LONG.equals(fieldSchema.getType())) {
206-
recordBuilder.set(field.getName(), zonedDateTime.toInstant().toEpochMilli());
207191
} else {
208192
recordBuilder.set(field.getName(), zonedDateTime.toString());
209193
}
210194
return;
211195
}
212196

213-
// Handle BFILE
214-
try {
215-
ClassLoader oracleLoader = attrValue.getClass().getClassLoader();
216-
if (oracleLoader != null && oracleLoader.loadClass("oracle.jdbc.OracleBfile").isInstance(attrValue)) {
217-
recordBuilder.set(field.getName(), getBfileBytes(attrValue, field.getName()));
218-
return;
197+
ClassLoader oracleLoader = attrValue.getClass().getClassLoader();
198+
try {
199+
if (oracleLoader != null && oracleLoader.loadClass("oracle.jdbc.OracleBfile").isInstance(attrValue)) {
200+
recordBuilder.set(field.getName(), getBfileBytes(attrValue, field.getName()));
201+
return;
202+
}
203+
} catch (ClassNotFoundException e) {
204+
throw new RuntimeException(e);
219205
}
220-
} catch (Exception e) {
221-
// Not a BFile, let it fall through
222-
}
223-
224-
// Parent DBRecord handles the standard types
225-
super.setFieldValue(recordBuilder, field, attrValue);
206+
super.setFieldValue(recordBuilder, field, attrValue);
226207
}
227208

228209
@Override
@@ -347,7 +328,7 @@ private byte[] getBfileBytes(ResultSet resultSet, String columnName) throws SQLE
347328
return getBfileBytes(bfile, columnName);
348329
}
349330

350-
public static byte[] getBfileBytes(Object bfile, String columnName) {
331+
public byte[] getBfileBytes(Object bfile, String columnName) {
351332
if (bfile == null) {
352333
return null;
353334
}
@@ -460,7 +441,7 @@ private void handleOracleSpecificType(ResultSet resultSet, StructuredRecord.Buil
460441
case Types.STRUCT:
461442
Struct structValue = (Struct) resultSet.getObject(columnIndex);
462443
if (structValue != null) {
463-
recordBuilder.set(field.getName(), convertStructToRecord(structValue, nonNullSchema, resultSet));
444+
recordBuilder.set(field.getName(), convertStructToRecord(structValue, nonNullSchema));
464445
}
465446
break;
466447
case Types.DECIMAL:
@@ -493,7 +474,7 @@ private void handleOracleSpecificType(ResultSet resultSet, StructuredRecord.Buil
493474
}
494475
}
495476

496-
private StructuredRecord convertStructToRecord(Struct struct, Schema schema, ResultSet resultSet)
477+
protected StructuredRecord convertStructToRecord(Struct struct, Schema schema)
497478
throws SQLException {
498479
Object[] attributes = struct.getAttributes();
499480
List<Schema.Field> fields = schema.getFields();
@@ -502,7 +483,6 @@ private StructuredRecord convertStructToRecord(Struct struct, Schema schema, Res
502483
for (int index = 0; index < attributes.length; index++) {
503484
Schema.Field field = fields.get(index);
504485
Object attrValue = attributes[index];
505-
506486
setFieldValue(builder, field, attrValue);
507487
}
508488
return builder.build();

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

Lines changed: 3 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -60,11 +60,10 @@ public class OracleSourceSchemaReader extends CommonSchemaReader {
6060
DATA_TYPE_MAP.put("TIMESTAMP WITH LOCAL TZ", TIMESTAMP_LTZ);
6161
DATA_TYPE_MAP.put("TIMESTAMP WITH TZ", TIMESTAMP_TZ);
6262
DATA_TYPE_MAP.put("TIMESTAMP", Types.TIMESTAMP);
63-
DATA_TYPE_MAP.put("DATE", Types.DATE);
64-
DATA_TYPE_MAP.put("TIME", Types.TIME);
65-
DATA_TYPE_MAP.put("FLOAT", Types.FLOAT);
63+
DATA_TYPE_MAP.put("DATE", Types.TIMESTAMP);
64+
DATA_TYPE_MAP.put("FLOAT", Types.DOUBLE);
6665
DATA_TYPE_MAP.put("BINARY_FLOAT", BINARY_FLOAT);
67-
DATA_TYPE_MAP.put("REAL", Types.REAL);
66+
DATA_TYPE_MAP.put("REAL", Types.DOUBLE);
6867
DATA_TYPE_MAP.put("BINARY_DOUBLE", BINARY_DOUBLE);
6968
DATA_TYPE_MAP.put("DOUBLE", Types.DOUBLE);
7069
DATA_TYPE_MAP.put("BFILE", BFILE);

0 commit comments

Comments
 (0)