Skip to content

Commit cb4d025

Browse files
MrRahulSharmaMrRahulSharma
authored
MediumInt datatype changes (#390)
1 parent f717eb2 commit cb4d025

File tree

13 files changed

+252
-192
lines changed

13 files changed

+252
-192
lines changed

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

Lines changed: 49 additions & 30 deletions
Original file line numberDiff line numberDiff line change
@@ -212,10 +212,10 @@ public void write(DataOutput out) throws IOException {
212212
* @param stmt the {@link PreparedStatement} to write the {@link StructuredRecord} to
213213
*/
214214
public void write(PreparedStatement stmt) throws SQLException {
215-
for (int i = 0; i < columnTypes.size(); i++) {
216-
ColumnType columnType = columnTypes.get(i);
215+
for (int fieldIndex = 0; fieldIndex < columnTypes.size(); fieldIndex++) {
216+
ColumnType columnType = columnTypes.get(fieldIndex);
217217
Schema.Field field = record.getSchema().getField(columnType.getName());
218-
writeToDB(stmt, field, i);
218+
writeToDB(stmt, field, fieldIndex);
219219
}
220220
}
221221

@@ -274,26 +274,54 @@ private void writeToDataOut(DataOutput out, Schema.Field field) throws IOExcepti
274274
}
275275
}
276276

277-
protected void writeToDB(PreparedStatement stmt, @Nullable Schema.Field field, int fieldIndex) throws SQLException {
277+
private void writeToDB(PreparedStatement stmt, @Nullable Schema.Field field, int fieldIndex) throws SQLException {
278+
if (shouldWriteNullField(field)) {
279+
writeNullToDB(stmt, fieldIndex);
280+
} else {
281+
Schema nonNullableSchema = getNonNullableSchema(field);
282+
writeNonNullToDB(stmt, nonNullableSchema, field.getName(), fieldIndex);
283+
}
284+
}
278285

286+
/**
287+
* This method returns true in case a field can support writeNullToDB for the current field.
288+
* By default, this method returns true when field or field value is set to null.
289+
*
290+
* @param field Field
291+
* @return true if null value of the field can be written to DB
292+
*/
293+
protected boolean shouldWriteNullField(Schema.Field field) {
294+
return (field == null || record.get(field.getName()) == null);
295+
}
296+
297+
/**
298+
* This method handle the null field and null field values, by internally using the PreparedStatement.setNull
299+
* method. Any class requiring a custom handling to write null value for any type should override this method.
300+
*
301+
* @param stmt PreparedStatement object for writing to db
302+
* @param fieldIndex Field index in the columnTypes
303+
* @throws SQLException Exception while calling PreparedStatement.setNull
304+
*/
305+
protected void writeNullToDB(PreparedStatement stmt, int fieldIndex) throws SQLException {
279306
int sqlIndex = fieldIndex + 1;
280307
int sqlType = columnTypes.get(fieldIndex).getType();
281-
if (field == null) {
282-
// Some of the fields can be absent in the record
283-
stmt.setNull(sqlIndex, sqlType);
284-
return;
285-
}
308+
stmt.setNull(sqlIndex, sqlType);
309+
}
286310

287-
String fieldName = field.getName();
288-
Schema fieldSchema = getNonNullableSchema(field);
289-
Schema.Type fieldType = fieldSchema.getType();
290-
Schema.LogicalType fieldLogicalType = fieldSchema.getLogicalType();
291-
Object fieldValue = record.get(fieldName);
311+
/**
312+
* Write Non null values to DB.
313+
*
314+
* @param stmt PreparedStatement object for writing to db
315+
* @param fieldSchema Non-Nullable schema of the field
316+
* @param fieldName Current Field from record's schema
317+
* @param fieldIndex Field index in the columnTypes
318+
* @throws SQLException Exception while calling PreparedStatement.set... calls
319+
*/
320+
protected void writeNonNullToDB(PreparedStatement stmt, Schema fieldSchema,
321+
String fieldName, int fieldIndex) throws SQLException {
292322

293-
if (fieldValue == null) {
294-
stmt.setNull(sqlIndex, columnTypes.get(fieldIndex).getType());
295-
return;
296-
}
323+
int sqlIndex = fieldIndex + 1;
324+
Schema.LogicalType fieldLogicalType = fieldSchema.getLogicalType();
297325

298326
if (fieldLogicalType != null) {
299327
switch (fieldLogicalType) {
@@ -318,6 +346,8 @@ protected void writeToDB(PreparedStatement stmt, @Nullable Schema.Field field, i
318346
return;
319347
}
320348

349+
Schema.Type fieldType = fieldSchema.getType();
350+
Object fieldValue = record.get(fieldName);
321351
switch (fieldType) {
322352
case NULL:
323353
stmt.setNull(sqlIndex, columnTypes.get(fieldIndex).getType());
@@ -330,8 +360,7 @@ protected void writeToDB(PreparedStatement stmt, @Nullable Schema.Field field, i
330360
stmt.setBoolean(sqlIndex, (Boolean) fieldValue);
331361
break;
332362
case INT:
333-
// write short or int appropriately
334-
writeInt(stmt, fieldIndex, sqlIndex, fieldValue);
363+
stmt.setInt(sqlIndex, (Integer) fieldValue);
335364
break;
336365
case LONG:
337366
long fieldValueLong = ((Number) fieldValue).longValue();
@@ -364,16 +393,6 @@ protected void writeBytes(PreparedStatement stmt, int fieldIndex, int sqlIndex,
364393
stmt.setBytes(sqlIndex, byteValue);
365394
}
366395

367-
protected void writeInt(PreparedStatement stmt, int fieldIndex, int sqlIndex, Object fieldValue) throws SQLException {
368-
Integer intValue = (Integer) fieldValue;
369-
int parameterType = columnTypes.get(fieldIndex).getType();
370-
if (Types.TINYINT == parameterType || Types.SMALLINT == parameterType) {
371-
stmt.setShort(sqlIndex, intValue.shortValue());
372-
return;
373-
}
374-
stmt.setInt(sqlIndex, intValue);
375-
}
376-
377396
@Override
378397
public void setConf(Configuration conf) {
379398
this.conf = conf;

‎mssql-plugin/src/main/java/io/cdap/plugin/mssql/SqlServerSinkDBRecord.java‎

Lines changed: 20 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -25,7 +25,6 @@
2525
import java.sql.SQLException;
2626
import java.sql.Types;
2727
import java.util.List;
28-
import javax.annotation.Nullable;
2928

3029
/**
3130
* SQL Server Sink implementation {@link org.apache.hadoop.mapreduce.lib.db.DBWritable} and
@@ -38,18 +37,29 @@ public SqlServerSinkDBRecord(StructuredRecord record, List<ColumnType> columnTyp
3837
}
3938

4039
@Override
41-
protected void writeToDB(PreparedStatement stmt, @Nullable Schema.Field field, int fieldIndex) throws SQLException {
42-
Object fieldValue = (field != null) ? record.get(field.getName()) : null;
40+
protected void writeNullToDB(PreparedStatement stmt, int fieldIndex) throws SQLException {
41+
int sqlType = columnTypes.get(fieldIndex).getType();
42+
int sqlIndex = fieldIndex + 1;
43+
if (sqlType == SqlServerSourceSchemaReader.GEOGRAPHY_TYPE
44+
|| sqlType == SqlServerSourceSchemaReader.GEOMETRY_TYPE) {
45+
// Handle setting GEOGRAPHY and GEOMETRY 'null' values. Using 'stmt.setNull(sqlIndex, GEOMETRY_TYPE)' leads
46+
// to com.microsoft.sqlserver.jdbc.SQLServerException: The conversion from OBJECT to GEOMETRY is unsupported
47+
stmt.setString(sqlIndex, "Null");
48+
return;
49+
}
50+
super.writeNullToDB(stmt, fieldIndex);
51+
}
52+
53+
@Override
54+
protected void writeNonNullToDB(PreparedStatement stmt, Schema fieldSchema,
55+
String fieldName, int fieldIndex) throws SQLException {
56+
Object fieldValue = (fieldName != null) ? record.get(fieldName) : null;
4357
int sqlType = columnTypes.get(fieldIndex).getType();
4458
int sqlIndex = fieldIndex + 1;
4559
switch (sqlType) {
4660
case SqlServerSourceSchemaReader.GEOGRAPHY_TYPE:
4761
case SqlServerSourceSchemaReader.GEOMETRY_TYPE:
48-
if (fieldValue == null) {
49-
// Handle setting GEOGRAPHY and GEOMETRY 'null' values. Using 'stmt.setNull(sqlIndex, GEOMETRY_TYPE)' leads
50-
// to com.microsoft.sqlserver.jdbc.SQLServerException: The conversion from OBJECT to GEOMETRY is unsupported
51-
stmt.setString(sqlIndex, "Null");
52-
} else if (fieldValue instanceof String) {
62+
if (fieldValue instanceof String) {
5363
// Handle setting GEOGRAPHY and GEOMETRY values from Well Known Text.
5464
// For example, "POINT(3 40 5 6)"
5565
stmt.setString(sqlIndex, (String) fieldValue);
@@ -60,15 +70,10 @@ protected void writeToDB(PreparedStatement stmt, @Nullable Schema.Field field, i
6070
case Types.TIME:
6171
// Handle setting SQL Server 'TIME' data type as string to avoid accuracy loss. 'TIME' data type has the
6272
// accuracy of 100 nanoseconds(1 millisecond in Informatica) but 'java.sql.Time' will round value to second.
63-
if (fieldValue != null) {
64-
String fieldName = field.getName();
65-
stmt.setString(sqlIndex, record.getTime(fieldName).toString());
66-
} else {
67-
stmt.setNull(sqlIndex, sqlType);
68-
}
73+
stmt.setString(sqlIndex, record.getTime(fieldName).toString());
6974
break;
7075
default:
71-
super.writeToDB(stmt, field, fieldIndex);
76+
super.writeNonNullToDB(stmt, fieldSchema, fieldName, fieldIndex);
7277
}
7378
}
7479

‎mysql-plugin/docs/Mysql-batchsink.md‎

Lines changed: 34 additions & 31 deletions
Original file line numberDiff line numberDiff line change
@@ -66,37 +66,40 @@ connections.
6666

6767
Data Types Mapping
6868
----------
69-
| MySQL Data Type | CDAP Schema Data Type | Comment |
70-
| ------------------------------ |-----------------------|---------------------------------------------------------|
71-
| BIT | boolean | |
72-
| TINYINT | int | |
73-
| BOOL, BOOLEAN | boolean | |
74-
| SMALLINT | int | |
75-
| MEDIUMINT | double | |
76-
| INT,INTEGER | int | |
77-
| BIGINT | long | |
78-
| FLOAT | float | |
79-
| DOUBLE | double | |
80-
| DECIMAL | decimal | |
81-
| DATE | date | |
82-
| DATETIME | timestamp | |
83-
| TIMESTAMP | timestamp | |
84-
| TIME | time | |
85-
| YEAR | int | Users can manually set output schema to map it to Date. |
86-
| CHAR | string | |
87-
| VARCHAR | string | |
88-
| BINARY | bytes | |
89-
| VARBINARY | bytes | |
90-
| TINYBLOB | bytes | |
91-
| TINYTEXT | string | |
92-
| BLOB | bytes | |
93-
| TEXT | string | |
94-
| MEDIUMBLOB | bytes | |
95-
| MEDIUMTEXT | string | |
96-
| LONGBLOB | bytes | |
97-
| LONGTEXT | string | |
98-
| ENUM | string | |
99-
| SET | string | |
69+
| MySQL Data Type | CDAP Schema Data Type | Comment |
70+
|--------------------|-----------------------|---------------------------------------------------------|
71+
| BIT | boolean | |
72+
| TINYINT | int | |
73+
| BOOL, BOOLEAN | boolean | |
74+
| SMALLINT | int | |
75+
| MEDIUMINT | int | |
76+
| MEDIUMINT UNSIGNED | int | Users can manually set output schema to map it to Long. |
77+
| INT,INTEGER | int | |
78+
| INT UNSIGNED | long | |
79+
| BIGINT | long | |
80+
| BIGINT UNSIGNED | decimal | |
81+
| FLOAT | float | |
82+
| DOUBLE | double | |
83+
| DECIMAL | decimal | |
84+
| DATE | date | |
85+
| DATETIME | timestamp | |
86+
| TIMESTAMP | timestamp | |
87+
| TIME | time | |
88+
| YEAR | int | Users can manually set output schema to map it to Date. |
89+
| CHAR | string | |
90+
| VARCHAR | string | |
91+
| BINARY | bytes | |
92+
| VARBINARY | bytes | |
93+
| TINYBLOB | bytes | |
94+
| TINYTEXT | string | |
95+
| BLOB | bytes | |
96+
| TEXT | string | |
97+
| MEDIUMBLOB | bytes | |
98+
| MEDIUMTEXT | string | |
99+
| LONGBLOB | bytes | |
100+
| LONGTEXT | string | |
101+
| ENUM | string | |
102+
| SET | string | |
100103

101104

102105
Example

0 commit comments

Comments
 (0)