[FLINK-9557] [formats] Parse 'integer' type as BigDecimal

This closes apache#6153.
fhueske · Jun 26, 2018 · 25fbcf8 · 25fbcf8
1 parent 8674b69
commit 25fbcf8
Show file tree

Hide file tree

Showing 5 changed files with 12 additions and 12 deletions.
diff --git a/...java/org/apache/flink/streaming/connectors/kafka/KafkaJsonTableSourceFactoryTestBase.java b/...java/org/apache/flink/streaming/connectors/kafka/KafkaJsonTableSourceFactoryTestBase.java
@@ -117,7 +117,7 @@ private void testTableSource(FormatDescriptor format) {
  .withSchema(
  TableSchema.builder()
  .field("fruit-name", Types.STRING)
- .field("count", Types.BIG_INT)
+ .field("count", Types.BIG_DEC)
  .field("event-time", Types.SQL_TIMESTAMP)
  .field("proc-time", Types.SQL_TIMESTAMP)
  .build())
@@ -141,7 +141,7 @@ private void testTableSource(FormatDescriptor format) {
  .addSchema(
  new Schema()
  .field("fruit-name", Types.STRING).from("name")
- .field("count", Types.BIG_INT) // no from so it must match with the input
+ .field("count", Types.BIG_DEC) // no from so it must match with the input
  .field("event-time", Types.SQL_TIMESTAMP).rowtime(
  new Rowtime().timestampsFromField("time").watermarksFromSource())
  .field("proc-time", Types.SQL_TIMESTAMP).proctime());

diff --git a/...k-formats/flink-json/src/main/java/org/apache/flink/formats/json/JsonSchemaConverter.java b/...k-formats/flink-json/src/main/java/org/apache/flink/formats/json/JsonSchemaConverter.java
@@ -165,7 +165,9 @@ else if (typeNode.isTextual()) {
  typeSet.add(Types.BIG_DEC);
  break;
  case TYPE_INTEGER:
- typeSet.add(Types.BIG_INT);
+ // use BigDecimal for easier interoperability
+ // without affecting the correctness of the result
+ typeSet.add(Types.BIG_DEC);
  break;
  case TYPE_OBJECT:
  typeSet.add(convertObject(location, node, root));

diff --git a/...nk-json/src/test/java/org/apache/flink/formats/json/JsonRowDeserializationSchemaTest.java b/...nk-json/src/test/java/org/apache/flink/formats/json/JsonRowDeserializationSchemaTest.java
@@ -29,7 +29,6 @@
 
 import java.io.IOException;
 import java.math.BigDecimal;
-import java.math.BigInteger;
 import java.sql.Date;
 import java.sql.Time;
 import java.sql.Timestamp;
@@ -79,12 +78,12 @@ public void testTypeInfoDeserialization() throws Exception {
 
  @Test
  public void testSchemaDeserialization() throws Exception {
- final BigInteger id = BigInteger.valueOf(1238123899121L);
+ final BigDecimal id = BigDecimal.valueOf(1238123899121L);
  final String name = "asdlkjasjkdla998y1122";
  final byte[] bytes = new byte[1024];
  ThreadLocalRandom.current().nextBytes(bytes);
- final BigInteger[] numbers = new BigInteger[] {
- BigInteger.valueOf(1), BigInteger.valueOf(2), BigInteger.valueOf(3)};
+ final BigDecimal[] numbers = new BigDecimal[] {
+ BigDecimal.valueOf(1), BigDecimal.valueOf(2), BigDecimal.valueOf(3)};
  final String[] strings = new String[] {"one", "two", "three"};
 
  final ObjectMapper objectMapper = new ObjectMapper();

diff --git a/...link-json/src/test/java/org/apache/flink/formats/json/JsonRowSerializationSchemaTest.java b/...link-json/src/test/java/org/apache/flink/formats/json/JsonRowSerializationSchemaTest.java
@@ -25,7 +25,6 @@
 
 import java.io.IOException;
 import java.math.BigDecimal;
-import java.math.BigInteger;
 import java.sql.Date;
 import java.sql.Time;
 import java.sql.Timestamp;
@@ -137,7 +136,7 @@ public void testSchema() throws IOException {
  "}");
 
  final Row row = new Row(11);
- row.setField(0, BigInteger.valueOf(-333));
+ row.setField(0, BigDecimal.valueOf(-333));
  row.setField(1, BigDecimal.valueOf(12.2222));
  row.setField(2, null);
  row.setField(3, "");
@@ -148,8 +147,8 @@ public void testSchema() throws IOException {
  final byte[] bytes = new byte[1024];
  ThreadLocalRandom.current().nextBytes(bytes);
  row.setField(7, bytes);
- final BigInteger[] numbers = new BigInteger[] {
- BigInteger.valueOf(1), BigInteger.valueOf(2), BigInteger.valueOf(3)};
+ final BigDecimal[] numbers = new BigDecimal[] {
+ BigDecimal.valueOf(1), BigDecimal.valueOf(2), BigDecimal.valueOf(3)};
  row.setField(8, numbers);
  final String[] strings = new String[] {"one", "two", "three"};
  row.setField(9, strings);

diff --git a/...rmats/flink-json/src/test/java/org/apache/flink/formats/json/JsonSchemaConverterTest.java b/...rmats/flink-json/src/test/java/org/apache/flink/formats/json/JsonSchemaConverterTest.java
@@ -47,7 +47,7 @@ public void testComplexSchema() throws Exception {
  "email", "tel", "sound", "org"},
  Types.STRING, Types.STRING, Types.BOOLEAN, Types.ROW(Types.BIG_DEC, Types.STRING, Types.STRING, Types.STRING),
  Types.OBJECT_ARRAY(Types.STRING), Types.STRING, Types.ROW_NAMED(new String[] {"type", "value"}, Types.STRING, Types.STRING),
- Types.ROW_NAMED(new String[] {"type", "value"}, Types.BIG_INT, Types.STRING), Types.VOID,
+ Types.ROW_NAMED(new String[] {"type", "value"}, Types.BIG_DEC, Types.STRING), Types.VOID,
  Types.ROW_NAMED(new String[] {"organizationUnit"}, Types.ROW()));
 
  assertEquals(expected, result);