From ac515b8064715738b4ab7141f708c99df2685073 Mon Sep 17 00:00:00 2001
From: "orr.ganani" <orr.ganani@equalum.io>
Date: Thu, 26 Apr 2018 15:43:01 +0300
Subject: [PATCH] DBZ-644 change Original Data Type mapper and add a test

---
 .../mapping/AddOriginalDataType.java          |  3 +-
 .../mapping/OriginalDataTypeTest.java         | 37 +++++++++++++++++++
 2 files changed, 39 insertions(+), 1 deletion(-)
 create mode 100644 debezium-core/src/test/java/io/debezium/relational/mapping/OriginalDataTypeTest.java

diff --git a/debezium-core/src/main/java/io/debezium/relational/mapping/AddOriginalDataType.java b/debezium-core/src/main/java/io/debezium/relational/mapping/AddOriginalDataType.java
index 15639aca6..d8b24cc58 100644
--- a/debezium-core/src/main/java/io/debezium/relational/mapping/AddOriginalDataType.java
+++ b/debezium-core/src/main/java/io/debezium/relational/mapping/AddOriginalDataType.java
@@ -17,6 +17,7 @@ public ValueConverter create(Column column) {
 
     @Override
     public void alterFieldSchema(Column column, SchemaBuilder schemaBuilder) {
-       schemaBuilder.parameter("originalType", column.typeName().toUpperCase());
+       schemaBuilder.parameter("originalType", String.valueOf(column.jdbcType()));
+       schemaBuilder.parameter("columnSize", String.valueOf(column.length()));
     }
 }
diff --git a/debezium-core/src/test/java/io/debezium/relational/mapping/OriginalDataTypeTest.java b/debezium-core/src/test/java/io/debezium/relational/mapping/OriginalDataTypeTest.java
new file mode 100644
index 000000000..16f9c1b55
--- /dev/null
+++ b/debezium-core/src/test/java/io/debezium/relational/mapping/OriginalDataTypeTest.java
@@ -0,0 +1,37 @@
+/*
+ * Copyright Debezium Authors.
+ *
+ * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+ */
+package io.debezium.relational.mapping;
+
+import io.debezium.relational.Column;
+import org.apache.kafka.connect.data.SchemaBuilder;
+import org.junit.Test;
+
+import java.sql.Types;
+
+import static org.fest.assertions.Assertions.assertThat;
+
+/**
+ * @author Orr Ganani
+ *
+ */
+public class OriginalDataTypeTest {
+
+    private final int length = 255;
+    private final Column column = Column.editor()
+        .name("col")
+        .type("VARCHAR")
+        .jdbcType(Types.VARCHAR)
+        .length(length)
+        .create();
+
+    @Test
+    public void shouldTruncateStrings() {
+        SchemaBuilder schemaBuilder = SchemaBuilder.string();
+        new AddOriginalDataType().alterFieldSchema(column, schemaBuilder);
+        assertThat(schemaBuilder.parameters().get("originalType")).isEqualTo(String.valueOf(Types.VARCHAR));
+        assertThat(schemaBuilder.parameters().get("columnSize")).isEqualTo(String.valueOf(length));
+    }
+}