[FLINK-18425][table] Convert object arrays to primitive arrays in Gen…

…ericArrayData This closes apache#12762.
spotify · Jun 24, 2020 · 7def95b · 7def95b
1 parent 00863a2
commit 7def95b
Show file tree

Hide file tree

Showing 3 changed files with 62 additions and 9 deletions.
diff --git a/...-table/flink-table-common/src/main/java/org/apache/flink/table/data/GenericArrayData.java b/...-table/flink-table-common/src/main/java/org/apache/flink/table/data/GenericArrayData.java
@@ -244,39 +244,82 @@ private Object getObject(int pos) {
  // Conversion Utilities
  // ------------------------------------------------------------------------------------------
 
+ private boolean anyNull() {
+ for (Object element : (Object[]) array) {
+ if (element == null) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ private void checkNoNull() {
+ if (anyNull()) {
+ throw new RuntimeException("Primitive array must not contain a null value.");
+ }
+ }
+
  @Override
  public boolean[] toBooleanArray() {
- return (boolean[]) array;
+ if (isPrimitiveArray) {
+ return (boolean[]) array;
+ }
+ checkNoNull();
+ return ArrayUtils.toPrimitive((Boolean[]) array);
  }
 
  @Override
  public byte[] toByteArray() {
- return (byte[]) array;
+ if (isPrimitiveArray) {
+ return (byte[]) array;
+ }
+ checkNoNull();
+ return ArrayUtils.toPrimitive((Byte[]) array);
  }
 
  @Override
  public short[] toShortArray() {
- return (short[]) array;
+ if (isPrimitiveArray) {
+ return (short[]) array;
+ }
+ checkNoNull();
+ return ArrayUtils.toPrimitive((Short[]) array);
  }
 
  @Override
  public int[] toIntArray() {
- return (int[]) array;
+ if (isPrimitiveArray) {
+ return (int[]) array;
+ }
+ checkNoNull();
+ return ArrayUtils.toPrimitive((Integer[]) array);
  }
 
  @Override
  public long[] toLongArray() {
- return (long[]) array;
+ if (isPrimitiveArray) {
+ return (long[]) array;
+ }
+ checkNoNull();
+ return ArrayUtils.toPrimitive((Long[]) array);
  }
 
  @Override
  public float[] toFloatArray() {
- return (float[]) array;
+ if (isPrimitiveArray) {
+ return (float[]) array;
+ }
+ checkNoNull();
+ return ArrayUtils.toPrimitive((Float[]) array);
  }
 
  @Override
  public double[] toDoubleArray() {
- return (double[]) array;
+ if (isPrimitiveArray) {
+ return (double[]) array;
+ }
+ checkNoNull();
+ return ArrayUtils.toPrimitive((Double[]) array);
  }
 }
 
diff --git a/.../flink-table-common/src/main/java/org/apache/flink/table/data/binary/BinaryArrayData.java b/.../flink-table-common/src/main/java/org/apache/flink/table/data/binary/BinaryArrayData.java
@@ -434,7 +434,7 @@ public boolean anyNull() {
 
  private void checkNoNull() {
  if (anyNull()) {
- throw new RuntimeException("Array can not have null value!");
+ throw new RuntimeException("Primitive array must not contain a null value.");
  }
  }
 

diff --git a/...-runtime-blink/src/test/java/org/apache/flink/table/data/DataStructureConvertersTest.java b/...-runtime-blink/src/test/java/org/apache/flink/table/data/DataStructureConvertersTest.java
@@ -198,10 +198,20 @@ public static List<TestSpec> testData() {
  .convertedTo(Long.class, 123L),
 
  TestSpec
- .forDataType(ARRAY(BOOLEAN()))
+ .forDataType(ARRAY(BOOLEAN().notNull()))
  .convertedTo(boolean[].class, new boolean[]{true, false, true, true})
  .convertedTo(ArrayData.class, new GenericArrayData(new boolean[]{true, false, true, true})),
 
+ TestSpec
+ .forDataType(ARRAY(BOOLEAN()))
+ .convertedTo(Boolean[].class, new Boolean[]{true, null, true, true})
+ .convertedTo(ArrayData.class, new GenericArrayData(new Boolean[]{true, null, true, true})),
+
+ TestSpec
+ .forDataType(ARRAY(INT().notNull()))
+ .convertedTo(int[].class, new int[]{1, 2, 3, 4})
+ .convertedTo(Integer[].class, new Integer[]{1, 2, 3, 4}),
+
  // arrays of TINYINT, SMALLINT, INTEGER, BIGINT, FLOAT, DOUBLE are skipped for simplicity
 
  TestSpec