You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by rh...@apache.org on 2019/10/03 20:53:46 UTC
[kafka] branch 1.1 updated (c2af356 -> 81db9de)
This is an automated email from the ASF dual-hosted git repository.
rhauch pushed a change to branch 1.1
in repository https://gitbox.apache.org/repos/asf/kafka.git.
from c2af356 KAFKA-8523 Enabling InsertField transform to be used with tombstone events (#6914)
new aaa71d7 KAFKA-6684: Support casting Connect values with bytes schema to string
new 81db9de KAFKA-6290: Support casting from logical types in cast transform (#7371)
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../java/org/apache/kafka/connect/data/Values.java | 2 +-
.../org/apache/kafka/connect/transforms/Cast.java | 86 +++++++++---
.../apache/kafka/connect/transforms/CastTest.java | 152 ++++++++++++++++++++-
3 files changed, 217 insertions(+), 23 deletions(-)
[kafka] 01/02: KAFKA-6684: Support casting Connect values with
bytes schema to string
Posted by rh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rhauch pushed a commit to branch 1.1
in repository https://gitbox.apache.org/repos/asf/kafka.git
commit aaa71d7e014fa611714c7b0471f8eaa923f6c4f1
Author: Amit Sela <am...@gmail.com>
AuthorDate: Sun Sep 30 22:24:09 2018 -0700
KAFKA-6684: Support casting Connect values with bytes schema to string
Allow to cast LogicalType to string by calling the serialized (Java) object's toString().
Added tests for `BigDecimal` and `Date` as whole record and as fields.
Author: Amit Sela <am...@gmail.com>
Reviewers: Randall Hauch <rh...@gmail.com>, Robert Yokota <ra...@gmail.com>, Ewen Cheslack-Postava <ew...@confluent.io>
Closes #4820 from amitsela/cast-transform-bytes
---
.../java/org/apache/kafka/connect/data/Values.java | 2 +-
.../org/apache/kafka/connect/transforms/Cast.java | 63 ++++++++++++++--------
.../apache/kafka/connect/transforms/CastTest.java | 55 ++++++++++++++++++-
3 files changed, 97 insertions(+), 23 deletions(-)
diff --git a/connect/api/src/main/java/org/apache/kafka/connect/data/Values.java b/connect/api/src/main/java/org/apache/kafka/connect/data/Values.java
index 042079b..87428b3 100644
--- a/connect/api/src/main/java/org/apache/kafka/connect/data/Values.java
+++ b/connect/api/src/main/java/org/apache/kafka/connect/data/Values.java
@@ -707,7 +707,7 @@ public class Values {
return value.replaceAll("\\\\", "\\\\\\\\").replaceAll("\"", "\\\\\"");
}
- protected static DateFormat dateFormatFor(java.util.Date value) {
+ public static DateFormat dateFormatFor(java.util.Date value) {
if (value.getTime() < MILLIS_PER_DAY) {
return new SimpleDateFormat(ISO_8601_TIME_FORMAT_PATTERN);
}
diff --git a/connect/transforms/src/main/java/org/apache/kafka/connect/transforms/Cast.java b/connect/transforms/src/main/java/org/apache/kafka/connect/transforms/Cast.java
index 22b1972..1463a68 100644
--- a/connect/transforms/src/main/java/org/apache/kafka/connect/transforms/Cast.java
+++ b/connect/transforms/src/main/java/org/apache/kafka/connect/transforms/Cast.java
@@ -28,6 +28,7 @@ import org.apache.kafka.connect.data.Field;
import org.apache.kafka.connect.data.Schema;
import org.apache.kafka.connect.data.SchemaBuilder;
import org.apache.kafka.connect.data.Struct;
+import org.apache.kafka.connect.data.Values;
import org.apache.kafka.connect.errors.DataException;
import org.apache.kafka.connect.transforms.util.SchemaUtil;
import org.apache.kafka.connect.transforms.util.SimpleConfig;
@@ -78,9 +79,16 @@ public abstract class Cast<R extends ConnectRecord<R>> implements Transformation
private static final String PURPOSE = "cast types";
- private static final Set<Schema.Type> SUPPORTED_CAST_TYPES = EnumSet.of(
+ private static final Set<Schema.Type> SUPPORTED_CAST_INPUT_TYPES = EnumSet.of(
Schema.Type.INT8, Schema.Type.INT16, Schema.Type.INT32, Schema.Type.INT64,
- Schema.Type.FLOAT32, Schema.Type.FLOAT64, Schema.Type.BOOLEAN, Schema.Type.STRING
+ Schema.Type.FLOAT32, Schema.Type.FLOAT64, Schema.Type.BOOLEAN,
+ Schema.Type.STRING, Schema.Type.BYTES
+ );
+
+ private static final Set<Schema.Type> SUPPORTED_CAST_OUTPUT_TYPES = EnumSet.of(
+ Schema.Type.INT8, Schema.Type.INT16, Schema.Type.INT32, Schema.Type.INT64,
+ Schema.Type.FLOAT32, Schema.Type.FLOAT64, Schema.Type.BOOLEAN,
+ Schema.Type.STRING
);
// As a special case for casting the entire value (e.g. the incoming key is a int64 but you know it could be an
@@ -120,14 +128,14 @@ public abstract class Cast<R extends ConnectRecord<R>> implements Transformation
private R applySchemaless(R record) {
if (wholeValueCastType != null) {
- return newRecord(record, null, castValueToType(operatingValue(record), wholeValueCastType));
+ return newRecord(record, null, castValueToType(null, operatingValue(record), wholeValueCastType));
}
final Map<String, Object> value = requireMap(operatingValue(record), PURPOSE);
final HashMap<String, Object> updatedValue = new HashMap<>(value);
for (Map.Entry<String, Schema.Type> fieldSpec : casts.entrySet()) {
String field = fieldSpec.getKey();
- updatedValue.put(field, castValueToType(value.get(field), fieldSpec.getValue()));
+ updatedValue.put(field, castValueToType(null, value.get(field), fieldSpec.getValue()));
}
return newRecord(record, null, updatedValue);
}
@@ -138,7 +146,7 @@ public abstract class Cast<R extends ConnectRecord<R>> implements Transformation
// Whole-record casting
if (wholeValueCastType != null)
- return newRecord(record, updatedSchema, castValueToType(operatingValue(record), wholeValueCastType));
+ return newRecord(record, updatedSchema, castValueToType(valueSchema, operatingValue(record), wholeValueCastType));
// Casting within a struct
final Struct value = requireStruct(operatingValue(record), PURPOSE);
@@ -147,7 +155,8 @@ public abstract class Cast<R extends ConnectRecord<R>> implements Transformation
for (Field field : value.schema().fields()) {
final Object origFieldValue = value.get(field);
final Schema.Type targetType = casts.get(field.name());
- final Object newFieldValue = targetType != null ? castValueToType(origFieldValue, targetType) : origFieldValue;
+ final Object newFieldValue = targetType != null ? castValueToType(field.schema(), origFieldValue, targetType) : origFieldValue;
+ log.trace("Cast field '{}' from '{}' to '{}'", field.name(), origFieldValue, newFieldValue);
updatedValue.put(updatedSchema.field(field.name()), newFieldValue);
}
return newRecord(record, updatedSchema, updatedValue);
@@ -168,8 +177,10 @@ public abstract class Cast<R extends ConnectRecord<R>> implements Transformation
SchemaBuilder fieldBuilder = convertFieldType(casts.get(field.name()));
if (field.schema().isOptional())
fieldBuilder.optional();
- if (field.schema().defaultValue() != null)
- fieldBuilder.defaultValue(castValueToType(field.schema().defaultValue(), fieldBuilder.type()));
+ if (field.schema().defaultValue() != null) {
+ Schema fieldSchema = field.schema();
+ fieldBuilder.defaultValue(castValueToType(fieldSchema, fieldSchema.defaultValue(), fieldBuilder.type()));
+ }
builder.field(field.name(), fieldBuilder.build());
} else {
builder.field(field.name(), field.schema());
@@ -181,7 +192,7 @@ public abstract class Cast<R extends ConnectRecord<R>> implements Transformation
if (valueSchema.isOptional())
builder.optional();
if (valueSchema.defaultValue() != null)
- builder.defaultValue(castValueToType(valueSchema.defaultValue(), builder.type()));
+ builder.defaultValue(castValueToType(valueSchema, valueSchema.defaultValue(), builder.type()));
updatedSchema = builder.build();
schemaUpdateCache.put(valueSchema, updatedSchema);
@@ -212,11 +223,12 @@ public abstract class Cast<R extends ConnectRecord<R>> implements Transformation
}
- private static Object castValueToType(Object value, Schema.Type targetType) {
+ private static Object castValueToType(Schema schema, Object value, Schema.Type targetType) {
try {
if (value == null) return null;
- Schema.Type inferredType = ConnectSchema.schemaType(value.getClass());
+ Schema.Type inferredType = schema == null ? ConnectSchema.schemaType(value.getClass()) :
+ schema.type();
if (inferredType == null) {
throw new DataException("Cast transformation was passed a value of type " + value.getClass()
+ " which is not supported by Connect's data API");
@@ -327,7 +339,12 @@ public abstract class Cast<R extends ConnectRecord<R>> implements Transformation
}
private static String castToString(Object value) {
- return value.toString();
+ if (value instanceof java.util.Date) {
+ java.util.Date dateValue = (java.util.Date) value;
+ return Values.dateFormatFor(dateValue).format(dateValue);
+ } else {
+ return value.toString();
+ }
}
protected abstract Schema operatingSchema(R record);
@@ -370,15 +387,19 @@ public abstract class Cast<R extends ConnectRecord<R>> implements Transformation
}
private static Schema.Type validCastType(Schema.Type type, FieldType fieldType) {
- if (!SUPPORTED_CAST_TYPES.contains(type)) {
- String message = "Cast transformation does not support casting to/from " + type
- + "; supported types are " + SUPPORTED_CAST_TYPES;
- switch (fieldType) {
- case INPUT:
- throw new DataException(message);
- case OUTPUT:
- throw new ConfigException(message);
- }
+ switch (fieldType) {
+ case INPUT:
+ if (!SUPPORTED_CAST_INPUT_TYPES.contains(type)) {
+ throw new DataException("Cast transformation does not support casting from " +
+ type + "; supported types are " + SUPPORTED_CAST_INPUT_TYPES);
+ }
+ break;
+ case OUTPUT:
+ if (!SUPPORTED_CAST_OUTPUT_TYPES.contains(type)) {
+ throw new ConfigException("Cast transformation does not support casting to " +
+ type + "; supported types are " + SUPPORTED_CAST_OUTPUT_TYPES);
+ }
+ break;
}
return type;
}
diff --git a/connect/transforms/src/test/java/org/apache/kafka/connect/transforms/CastTest.java b/connect/transforms/src/test/java/org/apache/kafka/connect/transforms/CastTest.java
index decd043..c568afb 100644
--- a/connect/transforms/src/test/java/org/apache/kafka/connect/transforms/CastTest.java
+++ b/connect/transforms/src/test/java/org/apache/kafka/connect/transforms/CastTest.java
@@ -18,15 +18,18 @@
package org.apache.kafka.connect.transforms;
import org.apache.kafka.common.config.ConfigException;
+import org.apache.kafka.connect.data.Decimal;
import org.apache.kafka.connect.data.Schema;
import org.apache.kafka.connect.data.SchemaBuilder;
import org.apache.kafka.connect.data.Struct;
import org.apache.kafka.connect.data.Timestamp;
+import org.apache.kafka.connect.data.Values;
import org.apache.kafka.connect.errors.DataException;
import org.apache.kafka.connect.source.SourceRecord;
import org.junit.After;
import org.junit.Test;
+import java.math.BigDecimal;
import java.util.Collections;
import java.util.Date;
import java.util.HashMap;
@@ -39,6 +42,7 @@ import static org.junit.Assert.assertTrue;
public class CastTest {
private final Cast<SourceRecord> xformKey = new Cast.Key<>();
private final Cast<SourceRecord> xformValue = new Cast.Value<>();
+ private static final long MILLIS_PER_DAY = 24 * 60 * 60 * 1000;
@After
public void teardown() {
@@ -62,6 +66,11 @@ public class CastTest {
}
@Test(expected = ConfigException.class)
+ public void testUnsupportedTargetType() {
+ xformKey.configure(Collections.singletonMap(Cast.SPEC_CONFIG, "foo:bytes"));
+ }
+
+ @Test(expected = ConfigException.class)
public void testConfigInvalidMap() {
xformKey.configure(Collections.singletonMap(Cast.SPEC_CONFIG, "foo:int8:extra"));
}
@@ -172,6 +181,28 @@ public class CastTest {
}
@Test
+ public void castWholeBigDecimalRecordValueWithSchemaString() {
+ BigDecimal bigDecimal = new BigDecimal(42);
+ xformValue.configure(Collections.singletonMap(Cast.SPEC_CONFIG, "string"));
+ SourceRecord transformed = xformValue.apply(new SourceRecord(null, null, "topic", 0,
+ Decimal.schema(bigDecimal.scale()), bigDecimal));
+
+ assertEquals(Schema.Type.STRING, transformed.valueSchema().type());
+ assertEquals("42", transformed.value());
+ }
+
+ @Test
+ public void castWholeDateRecordValueWithSchemaString() {
+ Date timestamp = new Date(MILLIS_PER_DAY + 1); // day + 1msec to get a timestamp formatting.
+ xformValue.configure(Collections.singletonMap(Cast.SPEC_CONFIG, "string"));
+ SourceRecord transformed = xformValue.apply(new SourceRecord(null, null, "topic", 0,
+ Timestamp.SCHEMA, timestamp));
+
+ assertEquals(Schema.Type.STRING, transformed.valueSchema().type());
+ assertEquals(Values.dateFormatFor(timestamp).format(timestamp), transformed.value());
+ }
+
+ @Test
public void castWholeRecordDefaultValue() {
// Validate default value in schema is correctly converted
xformValue.configure(Collections.singletonMap(Cast.SPEC_CONFIG, "int32"));
@@ -292,7 +323,8 @@ public class CastTest {
@Test
public void castFieldsWithSchema() {
- xformValue.configure(Collections.singletonMap(Cast.SPEC_CONFIG, "int8:int16,int16:int32,int32:int64,int64:boolean,float32:float64,float64:boolean,boolean:int8,string:int32,optional:int32"));
+ Date day = new Date(MILLIS_PER_DAY);
+ xformValue.configure(Collections.singletonMap(Cast.SPEC_CONFIG, "int8:int16,int16:int32,int32:int64,int64:boolean,float32:float64,float64:boolean,boolean:int8,string:int32,bigdecimal:string,date:string,optional:int32"));
// Include an optional fields and fields with defaults to validate their values are passed through properly
SchemaBuilder builder = SchemaBuilder.struct();
@@ -305,6 +337,8 @@ public class CastTest {
builder.field("float64", SchemaBuilder.float64().defaultValue(-1.125).build());
builder.field("boolean", Schema.BOOLEAN_SCHEMA);
builder.field("string", Schema.STRING_SCHEMA);
+ builder.field("bigdecimal", Decimal.schema(new BigDecimal(42).scale()));
+ builder.field("date", Timestamp.SCHEMA);
builder.field("optional", Schema.OPTIONAL_FLOAT32_SCHEMA);
builder.field("timestamp", Timestamp.SCHEMA);
Schema supportedTypesSchema = builder.build();
@@ -317,6 +351,8 @@ public class CastTest {
recordValue.put("float32", 32.f);
recordValue.put("float64", -64.);
recordValue.put("boolean", true);
+ recordValue.put("bigdecimal", new BigDecimal(42));
+ recordValue.put("date", day);
recordValue.put("string", "42");
recordValue.put("timestamp", new Date(0));
// optional field intentionally omitted
@@ -335,8 +371,25 @@ public class CastTest {
assertEquals(true, ((Struct) transformed.value()).schema().field("float64").schema().defaultValue());
assertEquals((byte) 1, ((Struct) transformed.value()).get("boolean"));
assertEquals(42, ((Struct) transformed.value()).get("string"));
+ assertEquals("42", ((Struct) transformed.value()).get("bigdecimal"));
+ assertEquals(Values.dateFormatFor(day).format(day), ((Struct) transformed.value()).get("date"));
assertEquals(new Date(0), ((Struct) transformed.value()).get("timestamp"));
assertNull(((Struct) transformed.value()).get("optional"));
+
+ Schema transformedSchema = ((Struct) transformed.value()).schema();
+ assertEquals(Schema.INT16_SCHEMA.type(), transformedSchema.field("int8").schema().type());
+ assertEquals(Schema.OPTIONAL_INT32_SCHEMA.type(), transformedSchema.field("int16").schema().type());
+ assertEquals(Schema.INT64_SCHEMA.type(), transformedSchema.field("int32").schema().type());
+ assertEquals(Schema.BOOLEAN_SCHEMA.type(), transformedSchema.field("int64").schema().type());
+ assertEquals(Schema.FLOAT64_SCHEMA.type(), transformedSchema.field("float32").schema().type());
+ assertEquals(Schema.BOOLEAN_SCHEMA.type(), transformedSchema.field("float64").schema().type());
+ assertEquals(Schema.INT8_SCHEMA.type(), transformedSchema.field("boolean").schema().type());
+ assertEquals(Schema.INT32_SCHEMA.type(), transformedSchema.field("string").schema().type());
+ assertEquals(Schema.STRING_SCHEMA.type(), transformedSchema.field("bigdecimal").schema().type());
+ assertEquals(Schema.STRING_SCHEMA.type(), transformedSchema.field("date").schema().type());
+ assertEquals(Schema.OPTIONAL_INT32_SCHEMA.type(), transformedSchema.field("optional").schema().type());
+ // The following fields are not changed
+ assertEquals(Timestamp.SCHEMA.type(), transformedSchema.field("timestamp").schema().type());
}
@SuppressWarnings("unchecked")
[kafka] 02/02: KAFKA-6290: Support casting from logical types in
cast transform (#7371)
Posted by rh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rhauch pushed a commit to branch 1.1
in repository https://gitbox.apache.org/repos/asf/kafka.git
commit 81db9de2d8c6bde8598e942c456f3950ddde2278
Author: Nigel Liang <ni...@nigelliang.com>
AuthorDate: Thu Oct 3 12:55:52 2019 -0700
KAFKA-6290: Support casting from logical types in cast transform (#7371)
Adds support for the Connect Cast transforms to cast from Connect logical types, such as DATE, TIME, TIMESTAMP, and DECIMAL. Casting to numeric types will produce the underlying numeric value represented in the desired type. For logical types represented by underlying Java Date class, this means the milliseconds since EPOCH. For Decimal, this means the underlying value. If the value does not fit in the desired target type, it may overflow.
Casting to String from Date, Time, and Timestamp types will produce their ISO 8601 representation. Casting to String from Decimal will result in the value represented as a string. e.g. 1234 -> "1234".
Author: Nigel Liang <ni...@nigelliang.com>
Reviewer: Randall Hauch <rh...@gmail.com>
---
.../org/apache/kafka/connect/transforms/Cast.java | 23 +++++
.../apache/kafka/connect/transforms/CastTest.java | 101 ++++++++++++++++++++-
2 files changed, 122 insertions(+), 2 deletions(-)
diff --git a/connect/transforms/src/main/java/org/apache/kafka/connect/transforms/Cast.java b/connect/transforms/src/main/java/org/apache/kafka/connect/transforms/Cast.java
index 1463a68..9dcec15 100644
--- a/connect/transforms/src/main/java/org/apache/kafka/connect/transforms/Cast.java
+++ b/connect/transforms/src/main/java/org/apache/kafka/connect/transforms/Cast.java
@@ -24,14 +24,20 @@ import org.apache.kafka.common.config.ConfigDef;
import org.apache.kafka.common.config.ConfigException;
import org.apache.kafka.connect.connector.ConnectRecord;
import org.apache.kafka.connect.data.ConnectSchema;
+import org.apache.kafka.connect.data.Date;
import org.apache.kafka.connect.data.Field;
import org.apache.kafka.connect.data.Schema;
+import org.apache.kafka.connect.data.Schema.Type;
import org.apache.kafka.connect.data.SchemaBuilder;
import org.apache.kafka.connect.data.Struct;
+import org.apache.kafka.connect.data.Time;
+import org.apache.kafka.connect.data.Timestamp;
import org.apache.kafka.connect.data.Values;
import org.apache.kafka.connect.errors.DataException;
import org.apache.kafka.connect.transforms.util.SchemaUtil;
import org.apache.kafka.connect.transforms.util.SimpleConfig;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import java.util.EnumSet;
import java.util.HashMap;
@@ -44,6 +50,7 @@ import static org.apache.kafka.connect.transforms.util.Requirements.requireMap;
import static org.apache.kafka.connect.transforms.util.Requirements.requireStruct;
public abstract class Cast<R extends ConnectRecord<R>> implements Transformation<R> {
+ private static final Logger log = LoggerFactory.getLogger(Cast.class);
// TODO: Currently we only support top-level field casting. Ideally we could use a dotted notation in the spec to
// allow casting nested fields.
@@ -220,7 +227,18 @@ public abstract class Cast<R extends ConnectRecord<R>> implements Transformation
default:
throw new DataException("Unexpected type in Cast transformation: " + type);
}
+ }
+ private static Object encodeLogicalType(Schema schema, Object value) {
+ switch (schema.name()) {
+ case Date.LOGICAL_NAME:
+ return Date.fromLogical(schema, (java.util.Date) value);
+ case Time.LOGICAL_NAME:
+ return Time.fromLogical(schema, (java.util.Date) value);
+ case Timestamp.LOGICAL_NAME:
+ return Timestamp.fromLogical(schema, (java.util.Date) value);
+ }
+ return value;
}
private static Object castValueToType(Schema schema, Object value, Schema.Type targetType) {
@@ -236,6 +254,11 @@ public abstract class Cast<R extends ConnectRecord<R>> implements Transformation
// Ensure the type we are trying to cast from is supported
validCastType(inferredType, FieldType.INPUT);
+ // Perform logical type encoding to their internal representation.
+ if (schema != null && schema.name() != null && targetType != Type.STRING) {
+ value = encodeLogicalType(schema, value);
+ }
+
switch (targetType) {
case INT8:
return castToInt8(value);
diff --git a/connect/transforms/src/test/java/org/apache/kafka/connect/transforms/CastTest.java b/connect/transforms/src/test/java/org/apache/kafka/connect/transforms/CastTest.java
index c568afb..a28aa28 100644
--- a/connect/transforms/src/test/java/org/apache/kafka/connect/transforms/CastTest.java
+++ b/connect/transforms/src/test/java/org/apache/kafka/connect/transforms/CastTest.java
@@ -17,11 +17,16 @@
package org.apache.kafka.connect.transforms;
+import java.util.Arrays;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
import org.apache.kafka.common.config.ConfigException;
import org.apache.kafka.connect.data.Decimal;
import org.apache.kafka.connect.data.Schema;
+import org.apache.kafka.connect.data.Schema.Type;
import org.apache.kafka.connect.data.SchemaBuilder;
import org.apache.kafka.connect.data.Struct;
+import org.apache.kafka.connect.data.Time;
import org.apache.kafka.connect.data.Timestamp;
import org.apache.kafka.connect.data.Values;
import org.apache.kafka.connect.errors.DataException;
@@ -42,7 +47,8 @@ import static org.junit.Assert.assertTrue;
public class CastTest {
private final Cast<SourceRecord> xformKey = new Cast.Key<>();
private final Cast<SourceRecord> xformValue = new Cast.Value<>();
- private static final long MILLIS_PER_DAY = 24 * 60 * 60 * 1000;
+ private static final long MILLIS_PER_HOUR = TimeUnit.HOURS.toMillis(1);
+ private static final long MILLIS_PER_DAY = TimeUnit.DAYS.toMillis(1);
@After
public void teardown() {
@@ -320,6 +326,97 @@ public class CastTest {
xformValue.apply(new SourceRecord(null, null, "topic", 0, null, Collections.singletonList("foo")));
}
+ @Test
+ public void castLogicalToPrimitive() {
+ List<String> specParts = Arrays.asList(
+ "date_to_int32:int32", // Cast to underlying representation
+ "timestamp_to_int64:int64", // Cast to underlying representation
+ "time_to_int64:int64", // Cast to wider datatype than underlying representation
+ "decimal_to_int32:int32", // Cast to narrower datatype with data loss
+ "timestamp_to_float64:float64", // loss of precision casting to double
+ "null_timestamp_to_int32:int32"
+ );
+
+ Date day = new Date(MILLIS_PER_DAY);
+ xformValue.configure(Collections.singletonMap(Cast.SPEC_CONFIG,
+ String.join(",", specParts)));
+
+ SchemaBuilder builder = SchemaBuilder.struct();
+ builder.field("date_to_int32", org.apache.kafka.connect.data.Date.SCHEMA);
+ builder.field("timestamp_to_int64", Timestamp.SCHEMA);
+ builder.field("time_to_int64", Time.SCHEMA);
+ builder.field("decimal_to_int32", Decimal.schema(new BigDecimal((long) Integer.MAX_VALUE + 1).scale()));
+ builder.field("timestamp_to_float64", Timestamp.SCHEMA);
+ builder.field("null_timestamp_to_int32", Timestamp.builder().optional().build());
+
+ Schema supportedTypesSchema = builder.build();
+
+ Struct recordValue = new Struct(supportedTypesSchema);
+ recordValue.put("date_to_int32", day);
+ recordValue.put("timestamp_to_int64", new Date(0));
+ recordValue.put("time_to_int64", new Date(1));
+ recordValue.put("decimal_to_int32", new BigDecimal((long) Integer.MAX_VALUE + 1));
+ recordValue.put("timestamp_to_float64", new Date(Long.MAX_VALUE));
+ recordValue.put("null_timestamp_to_int32", null);
+
+ SourceRecord transformed = xformValue.apply(
+ new SourceRecord(null, null, "topic", 0,
+ supportedTypesSchema, recordValue));
+
+ assertEquals(1, ((Struct) transformed.value()).get("date_to_int32"));
+ assertEquals(0L, ((Struct) transformed.value()).get("timestamp_to_int64"));
+ assertEquals(1L, ((Struct) transformed.value()).get("time_to_int64"));
+ assertEquals(Integer.MIN_VALUE, ((Struct) transformed.value()).get("decimal_to_int32"));
+ assertEquals(9.223372036854776E18, ((Struct) transformed.value()).get("timestamp_to_float64"));
+ assertNull(((Struct) transformed.value()).get("null_timestamp_to_int32"));
+
+ Schema transformedSchema = ((Struct) transformed.value()).schema();
+ assertEquals(Type.INT32, transformedSchema.field("date_to_int32").schema().type());
+ assertEquals(Type.INT64, transformedSchema.field("timestamp_to_int64").schema().type());
+ assertEquals(Type.INT64, transformedSchema.field("time_to_int64").schema().type());
+ assertEquals(Type.INT32, transformedSchema.field("decimal_to_int32").schema().type());
+ assertEquals(Type.FLOAT64, transformedSchema.field("timestamp_to_float64").schema().type());
+ assertEquals(Type.INT32, transformedSchema.field("null_timestamp_to_int32").schema().type());
+ }
+
+ @Test
+ public void castLogicalToString() {
+ Date date = new Date(MILLIS_PER_DAY);
+ Date time = new Date(MILLIS_PER_HOUR);
+ Date timestamp = new Date();
+
+ xformValue.configure(Collections.singletonMap(Cast.SPEC_CONFIG,
+ "date:string,decimal:string,time:string,timestamp:string"));
+
+ SchemaBuilder builder = SchemaBuilder.struct();
+ builder.field("date", org.apache.kafka.connect.data.Date.SCHEMA);
+ builder.field("decimal", Decimal.schema(new BigDecimal(1982).scale()));
+ builder.field("time", Time.SCHEMA);
+ builder.field("timestamp", Timestamp.SCHEMA);
+
+ Schema supportedTypesSchema = builder.build();
+
+ Struct recordValue = new Struct(supportedTypesSchema);
+ recordValue.put("date", date);
+ recordValue.put("decimal", new BigDecimal(1982));
+ recordValue.put("time", time);
+ recordValue.put("timestamp", timestamp);
+
+ SourceRecord transformed = xformValue.apply(
+ new SourceRecord(null, null, "topic", 0,
+ supportedTypesSchema, recordValue));
+
+ assertEquals(Values.dateFormatFor(date).format(date), ((Struct) transformed.value()).get("date"));
+ assertEquals("1982", ((Struct) transformed.value()).get("decimal"));
+ assertEquals(Values.dateFormatFor(time).format(time), ((Struct) transformed.value()).get("time"));
+ assertEquals(Values.dateFormatFor(timestamp).format(timestamp), ((Struct) transformed.value()).get("timestamp"));
+
+ Schema transformedSchema = ((Struct) transformed.value()).schema();
+ assertEquals(Type.STRING, transformedSchema.field("date").schema().type());
+ assertEquals(Type.STRING, transformedSchema.field("decimal").schema().type());
+ assertEquals(Type.STRING, transformedSchema.field("time").schema().type());
+ assertEquals(Type.STRING, transformedSchema.field("timestamp").schema().type());
+ }
@Test
public void castFieldsWithSchema() {
@@ -338,7 +435,7 @@ public class CastTest {
builder.field("boolean", Schema.BOOLEAN_SCHEMA);
builder.field("string", Schema.STRING_SCHEMA);
builder.field("bigdecimal", Decimal.schema(new BigDecimal(42).scale()));
- builder.field("date", Timestamp.SCHEMA);
+ builder.field("date", org.apache.kafka.connect.data.Date.SCHEMA);
builder.field("optional", Schema.OPTIONAL_FLOAT32_SCHEMA);
builder.field("timestamp", Timestamp.SCHEMA);
Schema supportedTypesSchema = builder.build();