Skip to content

Handle common logical types that Copycat supports: Decimal, Date, Time, Timestamp. #11

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
Ewen Cheslack-Postava (ewencp) merged 3 commits into from
Oct 10, 2015
Merged

Handle common logical types that Copycat supports: Decimal, Date, Time, Timestamp. #11

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
b1421e6
Handle common logical types that Copycat supports: Decimal, Date, Tim…
ewencp Oct 7, 2015
c315319
Fix handling of timezones for timestamp type.
ewencp Oct 8, 2015
3271750
Use import instead of fully-qualified name in DataConverter.
ewencp Oct 10, 2015
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
56 changes: 30 additions & 26 deletions src/main/java/io/confluent/copycat/jdbc/DataConverter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,9 +16,13 @@

package io.confluent.copycat.jdbc;

import org.apache.kafka.copycat.data.Date;
import org.apache.kafka.copycat.data.Decimal;
import org.apache.kafka.copycat.data.Schema;
import org.apache.kafka.copycat.data.SchemaBuilder;
import org.apache.kafka.copycat.data.Struct;
import org.apache.kafka.copycat.data.Time;
import org.apache.kafka.copycat.data.Timestamp;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

Expand All @@ -30,8 +34,10 @@
import java.sql.ResultSetMetaData;
import java.sql.SQLException;
import java.sql.SQLXML;
import java.sql.Timestamp;
import java.sql.Types;
import java.util.Calendar;
import java.util.GregorianCalendar;
import java.util.TimeZone;


/**
Expand All @@ -41,6 +47,8 @@
public class DataConverter {
private static final Logger log = LoggerFactory.getLogger(JdbcSourceTask.class);

private static final Calendar UTC_CALENDAR = new GregorianCalendar(TimeZone.getTimeZone("UTC"));

public static Schema convertSchema(String tableName, ResultSetMetaData metadata)
throws SQLException {
// TODO: Detect changes to metadata, which will require schema updates
Expand Down Expand Up @@ -163,8 +171,7 @@ private static void addFieldSchema(ResultSetMetaData metadata, int col,

case Types.NUMERIC:
case Types.DECIMAL: {
// FIXME This should use Avro's decimal logical type
log.warn("JDBC type {} not currently supported", sqlType);
builder.field(fieldName, Decimal.schema(metadata.getScale(col)));
break;
}

Expand Down Expand Up @@ -204,27 +211,31 @@ private static void addFieldSchema(ResultSetMetaData metadata, int col,

// Date is day + moth + year
case Types.DATE: {
// FIXME Dates/times are hard
log.warn("JDBC type DATE not currently supported");
SchemaBuilder dateSchemaBuilder = Date.builder();
if (optional) {
dateSchemaBuilder.optional();
}
builder.field(fieldName, dateSchemaBuilder.build());
break;
}

// Time is a time of day -- hour, minute, seconds, nanoseconds
case Types.TIME: {
// FIXME Dates/times are hard
log.warn("JDBC type TIME not currently supported");
SchemaBuilder timeSchemaBuilder = Time.builder();
if (optional) {
timeSchemaBuilder.optional();
}
builder.field(fieldName, timeSchemaBuilder.build());
break;
}

// Timestamp is a date + time
case Types.TIMESTAMP: {
// Current approach uses UNIX time in milliseconds. May lose nanosecond precision
// available in source type
SchemaBuilder tsSchemaBuilder = Timestamp.builder();
if (optional) {
builder.field(fieldName, Schema.OPTIONAL_INT64_SCHEMA);
} else {
builder.field(fieldName, Schema.INT64_SCHEMA);
tsSchemaBuilder.optional();
}
builder.field(fieldName, tsSchemaBuilder.build());
break;
}

Expand Down Expand Up @@ -307,10 +318,8 @@ private static void convertFieldValue(ResultSet resultSet, int col, int colType,

case Types.NUMERIC:
case Types.DECIMAL: {
// FIXME This should use Avro's decimal logical type
// resultSet.getBigDecimal(col);
log.warn("Skipping NUMERIC or DECIMAL field");
return;
colValue = resultSet.getBigDecimal(col);
break;
}

case Types.CHAR:
Expand All @@ -337,24 +346,19 @@ private static void convertFieldValue(ResultSet resultSet, int col, int colType,

// Date is day + moth + year
case Types.DATE: {
// FIXME Dates/times are hard
log.warn("Skipping DATE field");
return;
colValue = resultSet.getDate(col, UTC_CALENDAR);
break;
}

// Time is a time of day -- hour, minute, seconds, nanoseconds
case Types.TIME: {
// FIXME Dates/times are hard
log.warn("Skipping TIME field");
return;
colValue = resultSet.getTime(col, UTC_CALENDAR);
break;
}

// Timestamp is a date + time
case Types.TIMESTAMP: {
// Current approach is to use UNIX timestamp in milliseconds. This loses potential
// nanosecond precision.
Timestamp ts = resultSet.getTimestamp(col);
colValue = ts.getTime();
colValue = resultSet.getTimestamp(col, UTC_CALENDAR);
break;
}

Expand Down
22 changes: 22 additions & 0 deletions src/main/java/io/confluent/copycat/jdbc/JdbcUtils.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,12 +23,15 @@
import java.sql.ResultSetMetaData;
import java.sql.SQLException;
import java.sql.Statement;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.Date;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.TimeZone;

/**
* Utilties for interacting with a JDBC database.
Expand All @@ -52,6 +55,16 @@ public class JdbcUtils {
private static final int GET_COLUMNS_COLUMN_NAME = 4;
private static final int GET_COLUMNS_IS_AUTOINCREMENT = 23;


private static ThreadLocal<SimpleDateFormat> DATE_FORMATTER = new ThreadLocal<SimpleDateFormat>() {
@Override
protected SimpleDateFormat initialValue() {
SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS");
sdf.setTimeZone(TimeZone.getTimeZone("UTC"));
return sdf;
}
};

/**
* Get a list of tables in the database. This uses the default filters, which only include
* user-defined tables.
Expand Down Expand Up @@ -130,5 +143,14 @@ public static String getAutoincrementColumn(Connection conn, String table) throw
}
return (matches == 1 ? result : null);
}

/**
* Format the given Date assuming UTC timezone in a format supported by SQL.
* @param date the date to convert to a String
* @return the formatted string
*/
public static String formatUTC(Date date) {
return DATE_FORMATTER.get().format(date);
}
}

20 changes: 14 additions & 6 deletions src/main/java/io/confluent/copycat/jdbc/TimestampIncreasingTableQuerier.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,9 +24,13 @@
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Timestamp;
import java.util.Calendar;
import java.util.Collections;
import java.util.Date;
import java.util.GregorianCalendar;
import java.util.HashMap;
import java.util.Map;
import java.util.TimeZone;

/**
* <p>
Expand All @@ -46,6 +50,8 @@
* </p>
*/
public class TimestampIncreasingTableQuerier extends TableQuerier {
private static final Calendar UTC_CALENDAR = new GregorianCalendar(TimeZone.getTimeZone("UTC"));

private String timestampColumn;
private Long timestampOffset;
private String increasingColumn;
Expand Down Expand Up @@ -121,13 +127,15 @@ protected void createPreparedStatement(Connection db) throws SQLException {
@Override
protected ResultSet executeQuery() throws SQLException {
if (increasingColumn != null && timestampColumn != null) {
stmt.setTimestamp(1, new Timestamp(timestampOffset == null ? 0 : timestampOffset));
Timestamp ts = new Timestamp(timestampOffset == null ? 0 : timestampOffset);
stmt.setTimestamp(1, ts, UTC_CALENDAR);
stmt.setLong(2, (increasingOffset == null ? -1 : increasingOffset));
stmt.setTimestamp(3, new Timestamp(timestampOffset == null ? 0 : timestampOffset));
stmt.setTimestamp(3, ts, UTC_CALENDAR);
} else if (increasingColumn != null) {
stmt.setLong(1, (increasingOffset == null ? -1 : increasingOffset));
} else if (timestampColumn != null) {
stmt.setTimestamp(1, new Timestamp(timestampOffset == null ? 0 : timestampOffset));
Timestamp ts = new Timestamp(timestampOffset == null ? 0 : timestampOffset);
stmt.setTimestamp(1, ts, UTC_CALENDAR);
}
return stmt.executeQuery();
}
Expand Down Expand Up @@ -160,9 +168,9 @@ public SourceRecord extractRecord() throws SQLException {


if (timestampColumn != null) {
Long timestamp = (Long)record.get(timestampColumn);
assert timestampOffset == null || timestamp >= timestampOffset;
timestampOffset = timestamp;
Date timestamp = (Date) record.get(timestampColumn);
assert timestampOffset == null || timestamp.getTime() >= timestampOffset;
timestampOffset = timestamp.getTime();
offset.put(JdbcSourceTask.TIMESTAMP_FIELD, timestampOffset);
}

Expand Down
53 changes: 30 additions & 23 deletions src/test/java/io/confluent/copycat/jdbc/JdbcSourceTaskUpdateTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -129,16 +129,16 @@ public void testTimestamp() throws Exception {
db.createTable(SINGLE_TABLE_NAME,
"modified", "TIMESTAMP NOT NULL",
"id", "INT");
db.insert(SINGLE_TABLE_NAME, "modified", new Timestamp(10L).toString(), "id", 1);
db.insert(SINGLE_TABLE_NAME, "modified", JdbcUtils.formatUTC(new Timestamp(10L)), "id", 1);

startTask("modified", null);
verifyTimestampFirstPoll();

// If there isn't enough resolution, this could miss some rows. In this case, we'll only see
// IDs 3 & 4.
db.insert(SINGLE_TABLE_NAME, "modified", new Timestamp(10L).toString(), "id", 2);
db.insert(SINGLE_TABLE_NAME, "modified", new Timestamp(11L).toString(), "id", 3);
db.insert(SINGLE_TABLE_NAME, "modified", new Timestamp(12L).toString(), "id", 4);
db.insert(SINGLE_TABLE_NAME, "modified", JdbcUtils.formatUTC(new Timestamp(10L)), "id", 2);
db.insert(SINGLE_TABLE_NAME, "modified", JdbcUtils.formatUTC(new Timestamp(11L)), "id", 3);
db.insert(SINGLE_TABLE_NAME, "modified", JdbcUtils.formatUTC(new Timestamp(12L)), "id", 4);

verifyPoll(2, "id", Arrays.asList(3, 4), true, false);

Expand All @@ -155,15 +155,15 @@ public void testTimestampAndIncreasing() throws Exception {
db.createTable(SINGLE_TABLE_NAME,
"modified", "TIMESTAMP NOT NULL",
"id", "INT NOT NULL");
db.insert(SINGLE_TABLE_NAME, "modified", new Timestamp(10L).toString(), "id", 1);
db.insert(SINGLE_TABLE_NAME, "modified", JdbcUtils.formatUTC(new Timestamp(10L)), "id", 1);

startTask("modified", "id");
verifyIncreasingAndTimestampFirstPoll();

// Should be able to pick up id 2 because of ID despite same timestamp.
// Note this is setup so we can reuse some validation code
db.insert(SINGLE_TABLE_NAME, "modified", new Timestamp(10L).toString(), "id", 3);
db.insert(SINGLE_TABLE_NAME, "modified", new Timestamp(11L).toString(), "id", 1);
db.insert(SINGLE_TABLE_NAME, "modified", JdbcUtils.formatUTC(new Timestamp(10L)), "id", 3);
db.insert(SINGLE_TABLE_NAME, "modified", JdbcUtils.formatUTC(new Timestamp(11L)), "id", 1);

verifyPoll(2, "id", Arrays.asList(3, 1), true, true);

Expand Down Expand Up @@ -234,9 +234,9 @@ public void testTimestampRestoreOffset() throws Exception {
"modified", "TIMESTAMP NOT NULL",
"id", "INT");
// id=2 will be ignored since it has the same timestamp as the initial offset.
db.insert(SINGLE_TABLE_NAME, "modified", new Timestamp(10L).toString(), "id", 2);
db.insert(SINGLE_TABLE_NAME, "modified", new Timestamp(11L).toString(), "id", 3);
db.insert(SINGLE_TABLE_NAME, "modified", new Timestamp(12L).toString(), "id", 4);
db.insert(SINGLE_TABLE_NAME, "modified", JdbcUtils.formatUTC(new Timestamp(10L)), "id", 2);
db.insert(SINGLE_TABLE_NAME, "modified", JdbcUtils.formatUTC(new Timestamp(11L)), "id", 3);
db.insert(SINGLE_TABLE_NAME, "modified", JdbcUtils.formatUTC(new Timestamp(12L)), "id", 4);

startTask("modified", null);

Expand All @@ -262,11 +262,11 @@ public void testTimestampAndIncreasingRestoreOffset() throws Exception {
"id", "INT NOT NULL");
// id=3 will be ignored since it has the same timestamp + id as the initial offset, rest
// should be included, including id=1 which is an old ID with newer timestamp
db.insert(SINGLE_TABLE_NAME, "modified", new Timestamp(9L).toString(), "id", 2);
db.insert(SINGLE_TABLE_NAME, "modified", new Timestamp(10L).toString(), "id", 3);
db.insert(SINGLE_TABLE_NAME, "modified", new Timestamp(11L).toString(), "id", 4);
db.insert(SINGLE_TABLE_NAME, "modified", new Timestamp(12L).toString(), "id", 5);
db.insert(SINGLE_TABLE_NAME, "modified", new Timestamp(13L).toString(), "id", 1);
db.insert(SINGLE_TABLE_NAME, "modified", JdbcUtils.formatUTC(new Timestamp(9L)), "id", 2);
db.insert(SINGLE_TABLE_NAME, "modified", JdbcUtils.formatUTC(new Timestamp(10L)), "id", 3);
db.insert(SINGLE_TABLE_NAME, "modified", JdbcUtils.formatUTC(new Timestamp(11L)), "id", 4);
db.insert(SINGLE_TABLE_NAME, "modified", JdbcUtils.formatUTC(new Timestamp(12L)), "id", 5);
db.insert(SINGLE_TABLE_NAME, "modified", JdbcUtils.formatUTC(new Timestamp(13L)), "id", 1);

startTask("modified", "id");

Expand All @@ -276,7 +276,6 @@ public void testTimestampAndIncreasingRestoreOffset() throws Exception {
}



private void startTask(String timestampColumn, String increasingColumn) {
String mode = null;
if (timestampColumn != null && increasingColumn != null) {
Expand Down Expand Up @@ -313,7 +312,7 @@ private List<SourceRecord> verifyTimestampFirstPoll() throws Exception {
List<SourceRecord> records = task.poll();
assertEquals(1, records.size());
assertEquals(Collections.singletonMap(1, 1), countIntValues(records, "id"));
assertEquals(Collections.singletonMap(10L, 1), countLongValues(records, "modified"));
assertEquals(Collections.singletonMap(10L, 1), countTimestampValues(records, "modified"));
assertTimestampOffsets(records);
return records;
}
Expand Down Expand Up @@ -346,6 +345,7 @@ private <T> void verifyPoll(int numRecords, String valueField, List<T> values,
private enum Field {
KEY,
VALUE,
TIMESTAMP_VALUE,
INCREASING_OFFSET,
TIMESTAMP_OFFSET
}
Expand All @@ -361,12 +361,20 @@ private <T> Map<T, Integer> countInts(List<SourceRecord> records, Field field, S
case VALUE:
extracted = (T)((Struct)record.value()).get(fieldName);
break;
case TIMESTAMP_VALUE: {
java.util.Date rawTimestamp = (java.util.Date) ((Struct)record.value()).get(fieldName);
extracted = (T) (Long) rawTimestamp.getTime();
break;
}
case INCREASING_OFFSET:
extracted = (T)(record.sourceOffset()).get(JdbcSourceTask.INCREASING_FIELD);
break;
case TIMESTAMP_OFFSET:
extracted = (T)(record.sourceOffset()).get(JdbcSourceTask.TIMESTAMP_FIELD);
case TIMESTAMP_OFFSET: {
java.util.Date rawTimestamp
= (java.util.Date) record.sourceOffset().get(JdbcSourceTask.TIMESTAMP_FIELD);
extracted = (T) (Long) rawTimestamp.getTime();
break;
}
default:
throw new RuntimeException("Invalid field");
}
Expand All @@ -381,8 +389,8 @@ private Map<Integer, Integer> countIntValues(List<SourceRecord> records, String
return countInts(records, Field.VALUE, fieldName);
}

private Map<Long, Integer> countLongValues(List<SourceRecord> records, String fieldName) {
return countInts(records, Field.VALUE, fieldName);
private Map<Long, Integer> countTimestampValues(List<SourceRecord> records, String fieldName) {
return countInts(records, Field.TIMESTAMP_VALUE, fieldName);
}

private Map<Long, Integer> countIntIncreasingOffsets(List<SourceRecord> records, String fieldName) {
Expand All @@ -405,11 +413,10 @@ private void assertIncreasingOffsets(List<SourceRecord> records) {
private void assertTimestampOffsets(List<SourceRecord> records) {
// Should use timestamps as offsets
for(SourceRecord record : records) {
long timestampValue = (Long)((Struct)record.value()).get("modified");
long timestampValue = ((java.util.Date) ((Struct)record.value()).get("modified")).getTime();
long offsetValue = (Long)(record.sourceOffset()
.get(JdbcSourceTask.TIMESTAMP_FIELD));
assertEquals(timestampValue, offsetValue);
}
}

}