This is an automated email from the ASF dual-hosted git repository.
kurt pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 76eb039 [FLINK-11802][table-runtime-blink] Create TypeInfo and
TypeSerializer for blink data format
76eb039 is described below
commit 76eb0397041915c867df57b7e2478caca3afd7bd
Author: JingsongLi <lzljs3620...@aliyun.com>
AuthorDate: Mon Mar 4 10:01:58 2019 +0800
[FLINK-11802][table-runtime-blink] Create TypeInfo and TypeSerializer for
blink data format
This closes #7887
---
flink-table/flink-table-runtime-blink/pom.xml | 17 ++
.../apache/flink/table/dataformat/BinaryArray.java | 10 +
.../apache/flink/table/dataformat/BinaryMap.java | 10 +
.../apache/flink/table/dataformat/BinaryRow.java | 8 +-
.../flink/table/dataformat/BinaryString.java | 15 ++
.../flink/table/dataformat/BinaryWriter.java | 39 +++
.../flink/table/dataformat/TypeGetterSetters.java | 39 +++
.../org/apache/flink/table/type/ArrayType.java | 0
.../org/apache/flink/table/type/AtomicType.java | 0
.../org/apache/flink/table/type/BooleanType.java | 0
.../java/org/apache/flink/table/type/ByteType.java | 0
.../java/org/apache/flink/table/type/CharType.java | 0
.../java/org/apache/flink/table/type/DateType.java | 0
.../org/apache/flink/table/type/DecimalType.java | 0
.../org/apache/flink/table/type/DoubleType.java | 0
.../org/apache/flink/table/type/FloatType.java | 0
.../org/apache/flink/table/type/GenericType.java | 0
.../java/org/apache/flink/table/type/IntType.java | 0
.../org/apache/flink/table/type/InternalType.java | 0
.../org/apache/flink/table/type/InternalTypes.java | 0
.../java/org/apache/flink/table/type/LongType.java | 0
.../java/org/apache/flink/table/type/MapType.java | 0
.../org/apache/flink/table/type/PrimitiveType.java | 0
.../java/org/apache/flink/table/type/RowType.java | 2 +-
.../org/apache/flink/table/type/ShortType.java | 0
.../org/apache/flink/table/type/StringType.java | 0
.../java/org/apache/flink/table/type/TimeType.java | 0
.../org/apache/flink/table/type/TimestampType.java | 0
.../apache/flink/table/type/TypeConverters.java | 60 ++++-
.../flink/table/typeutils/BaseRowSerializer.java | 282 +++++++++++++++++++++
.../flink/table/typeutils/BaseRowTypeInfo.java | 145 +++++++++++
.../table/typeutils/BinaryArraySerializer.java | 107 ++++++++
.../flink/table/typeutils/BinaryArrayTypeInfo.java | 105 ++++++++
.../flink/table/typeutils/BinaryMapSerializer.java | 107 ++++++++
.../flink/table/typeutils/BinaryMapTypeInfo.java | 113 +++++++++
.../flink/table/typeutils/BinaryRowSerializer.java | 184 ++++++++++++++
.../table/typeutils/BinaryStringSerializer.java | 114 +++++++++
.../table/typeutils/BinaryStringTypeInfo.java | 89 +++++++
.../org/apache/flink/table/util/SegmentsUtil.java | 40 +++
.../apache/flink/table/type/InternalTypeTest.java | 0
.../table/typeutils/BaseRowSerializerTest.java | 229 +++++++++++++++++
.../table/typeutils/BinaryArraySerializerTest.java | 66 +++++
.../table/typeutils/BinaryMapSerializerTest.java | 66 +++++
.../table/typeutils/BinaryRowSerializerTest.java | 64 +++++
.../table/typeutils/BinaryRowTypeInfoTest.java | 77 ++++++
.../typeutils/BinaryStringSerializerTest.java} | 45 ++--
.../flink/table/typeutils/BinaryTypeInfoTest.java} | 35 ++-
47 files changed, 2017 insertions(+), 51 deletions(-)
diff --git a/flink-table/flink-table-runtime-blink/pom.xml
b/flink-table/flink-table-runtime-blink/pom.xml
index 29fb8a7..e8b1623 100644
--- a/flink-table/flink-table-runtime-blink/pom.xml
+++ b/flink-table/flink-table-runtime-blink/pom.xml
@@ -60,5 +60,22 @@ under the License.
<artifactId>janino</artifactId>
<version>${janino.version}</version>
</dependency>
+
+ <!-- test dependencies -->
+
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-core</artifactId>
+ <version>${project.version}</version>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+
<artifactId>flink-test-utils_${scala.binary.version}</artifactId>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ </dependency>
</dependencies>
</project>
diff --git
a/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/dataformat/BinaryArray.java
b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/dataformat/BinaryArray.java
index e708710..57ce04b 100644
---
a/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/dataformat/BinaryArray.java
+++
b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/dataformat/BinaryArray.java
@@ -347,6 +347,16 @@ public class BinaryArray extends BinaryFormat implements
TypeGetterSetters {
return values;
}
+ public BinaryArray copy() {
+ return copy(new BinaryArray());
+ }
+
+ public BinaryArray copy(BinaryArray reuse) {
+ byte[] bytes = SegmentsUtil.copyToBytes(segments, offset,
sizeInBytes);
+ reuse.pointTo(MemorySegmentFactory.wrap(bytes), 0, sizeInBytes);
+ return reuse;
+ }
+
private static BinaryArray fromPrimitiveArray(
Object arr, int offset, int length, int elementSize) {
final long headerInBytes = calculateHeaderInBytes(length);
diff --git
a/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/dataformat/BinaryMap.java
b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/dataformat/BinaryMap.java
index fefdc27..7c10193 100644
---
a/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/dataformat/BinaryMap.java
+++
b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/dataformat/BinaryMap.java
@@ -72,6 +72,16 @@ public class BinaryMap extends BinaryFormat {
return values;
}
+ public BinaryMap copy() {
+ return copy(new BinaryMap());
+ }
+
+ public BinaryMap copy(BinaryMap reuse) {
+ byte[] bytes = SegmentsUtil.copyToBytes(segments, offset,
sizeInBytes);
+ reuse.pointTo(MemorySegmentFactory.wrap(bytes), 0, sizeInBytes);
+ return reuse;
+ }
+
public static BinaryMap valueOf(BinaryArray key, BinaryArray value) {
checkArgument(key.getSegments().length == 1 &&
value.getSegments().length == 1);
byte[] bytes = new byte[4 + key.getSizeInBytes() +
value.getSizeInBytes()];
diff --git
a/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/dataformat/BinaryRow.java
b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/dataformat/BinaryRow.java
index 472f7b22..b88739e 100644
---
a/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/dataformat/BinaryRow.java
+++
b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/dataformat/BinaryRow.java
@@ -59,6 +59,10 @@ public final class BinaryRow extends BinaryFormat<Object>
implements BaseRow {
return ((arity + 63 + HEADER_SIZE_IN_BITS) / 64) * 8;
}
+ public static int calculateFixPartSizeInBytes(int arity) {
+ return calculateBitSetWidthInBytes(arity) + 8 * arity;
+ }
+
private final int arity;
private final int nullBitsSizeInBytes;
@@ -287,8 +291,8 @@ public final class BinaryRow extends BinaryFormat<Object>
implements BaseRow {
return copy(new BinaryRow(arity));
}
- public BinaryRow copy(BaseRow reuse) {
- return copyInternal((BinaryRow) reuse);
+ public BinaryRow copy(BinaryRow reuse) {
+ return copyInternal(reuse);
}
private BinaryRow copyInternal(BinaryRow reuse) {
diff --git
a/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/dataformat/BinaryString.java
b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/dataformat/BinaryString.java
index 4b388f6..152907e 100644
---
a/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/dataformat/BinaryString.java
+++
b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/dataformat/BinaryString.java
@@ -49,6 +49,16 @@ public class BinaryString extends BinaryFormat<String> {
}
}
+ public static BinaryString fromBytes(byte[] bytes) {
+ return new BinaryString(
+ new MemorySegment[]
{MemorySegmentFactory.wrap(bytes)}, 0, bytes.length);
+ }
+
+ public static BinaryString fromBytes(byte[] bytes, int offset, int
numBytes) {
+ return new BinaryString(
+ new MemorySegment[]
{MemorySegmentFactory.wrap(bytes)}, offset, numBytes);
+ }
+
@Override
public String toString() {
byte[] bytes = BinaryStringUtil.allocateReuseBytes(sizeInBytes);
@@ -56,6 +66,11 @@ public class BinaryString extends BinaryFormat<String> {
return new String(bytes, 0, sizeInBytes);
}
+ public BinaryString copy() {
+ byte[] copy = SegmentsUtil.copyToBytes(segments, offset,
sizeInBytes);
+ return BinaryString.fromBytes(copy, 0, copy.length);
+ }
+
/**
* Get binary string, if len less than 8, will be include in
variablePartOffsetAndLen.
*
diff --git
a/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/dataformat/BinaryWriter.java
b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/dataformat/BinaryWriter.java
index 435b064..08ba9eb 100644
---
a/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/dataformat/BinaryWriter.java
+++
b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/dataformat/BinaryWriter.java
@@ -17,6 +17,11 @@
package org.apache.flink.table.dataformat;
+import org.apache.flink.table.type.ArrayType;
+import org.apache.flink.table.type.InternalType;
+import org.apache.flink.table.type.InternalTypes;
+import org.apache.flink.table.type.MapType;
+
/**
* Writer to write a composite data format, like row, array.
* 1. Invoke {@link #reset()}.
@@ -61,4 +66,38 @@ public interface BinaryWriter {
* Finally, complete write to set real size to binary.
*/
void complete();
+
+ static void write(BinaryWriter writer, int pos, Object o, InternalType
type) {
+ if (type.equals(InternalTypes.BOOLEAN)) {
+ writer.writeBoolean(pos, (boolean) o);
+ } else if (type.equals(InternalTypes.BYTE)) {
+ writer.writeByte(pos, (byte) o);
+ } else if (type.equals(InternalTypes.SHORT)) {
+ writer.writeShort(pos, (short) o);
+ } else if (type.equals(InternalTypes.INT)) {
+ writer.writeInt(pos, (int) o);
+ } else if (type.equals(InternalTypes.LONG)) {
+ writer.writeLong(pos, (long) o);
+ } else if (type.equals(InternalTypes.FLOAT)) {
+ writer.writeFloat(pos, (float) o);
+ } else if (type.equals(InternalTypes.DOUBLE)) {
+ writer.writeDouble(pos, (double) o);
+ } else if (type.equals(InternalTypes.STRING)) {
+ writer.writeString(pos, (BinaryString) o);
+ } else if (type.equals(InternalTypes.CHAR)) {
+ writer.writeChar(pos, (char) o);
+ } else if (type.equals(InternalTypes.DATE)) {
+ writer.writeInt(pos, (int) o);
+ } else if (type.equals(InternalTypes.TIME)) {
+ writer.writeInt(pos, (int) o);
+ } else if (type.equals(InternalTypes.TIMESTAMP)) {
+ writer.writeLong(pos, (long) o);
+ } else if (type instanceof ArrayType) {
+ writer.writeArray(pos, (BinaryArray) o);
+ } else if (type instanceof MapType) {
+ writer.writeMap(pos, (BinaryMap) o);
+ } else {
+ throw new RuntimeException("Not support type: " + type);
+ }
+ }
}
diff --git
a/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/dataformat/TypeGetterSetters.java
b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/dataformat/TypeGetterSetters.java
index e567add..73d3d73 100644
---
a/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/dataformat/TypeGetterSetters.java
+++
b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/dataformat/TypeGetterSetters.java
@@ -17,6 +17,11 @@
package org.apache.flink.table.dataformat;
+import org.apache.flink.table.type.ArrayType;
+import org.apache.flink.table.type.InternalType;
+import org.apache.flink.table.type.InternalTypes;
+import org.apache.flink.table.type.MapType;
+
/**
* Provide type specialized getters and setters to reduce if/else and
eliminate box and unbox.
*
@@ -135,4 +140,38 @@ public interface TypeGetterSetters {
* Set char value.
*/
void setChar(int ordinal, char value);
+
+ static Object get(TypeGetterSetters row, int ordinal, InternalType
type) {
+ if (type.equals(InternalTypes.BOOLEAN)) {
+ return row.getBoolean(ordinal);
+ } else if (type.equals(InternalTypes.BYTE)) {
+ return row.getByte(ordinal);
+ } else if (type.equals(InternalTypes.SHORT)) {
+ return row.getShort(ordinal);
+ } else if (type.equals(InternalTypes.INT)) {
+ return row.getInt(ordinal);
+ } else if (type.equals(InternalTypes.LONG)) {
+ return row.getLong(ordinal);
+ } else if (type.equals(InternalTypes.FLOAT)) {
+ return row.getFloat(ordinal);
+ } else if (type.equals(InternalTypes.DOUBLE)) {
+ return row.getDouble(ordinal);
+ } else if (type.equals(InternalTypes.STRING)) {
+ return row.getString(ordinal);
+ } else if (type.equals(InternalTypes.CHAR)) {
+ return row.getChar(ordinal);
+ } else if (type.equals(InternalTypes.DATE)) {
+ return row.getInt(ordinal);
+ } else if (type.equals(InternalTypes.TIME)) {
+ return row.getInt(ordinal);
+ } else if (type.equals(InternalTypes.TIMESTAMP)) {
+ return row.getLong(ordinal);
+ } else if (type instanceof ArrayType) {
+ return row.getArray(ordinal);
+ } else if (type instanceof MapType) {
+ return row.getMap(ordinal);
+ } else {
+ throw new RuntimeException("Not support type: " + type);
+ }
+ }
}
diff --git
a/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/type/ArrayType.java
b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/type/ArrayType.java
similarity index 100%
copy from
flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/type/ArrayType.java
copy to
flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/type/ArrayType.java
diff --git
a/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/type/AtomicType.java
b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/type/AtomicType.java
similarity index 100%
rename from
flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/type/AtomicType.java
rename to
flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/type/AtomicType.java
diff --git
a/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/type/BooleanType.java
b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/type/BooleanType.java
similarity index 100%
rename from
flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/type/BooleanType.java
rename to
flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/type/BooleanType.java
diff --git
a/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/type/ByteType.java
b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/type/ByteType.java
similarity index 100%
rename from
flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/type/ByteType.java
rename to
flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/type/ByteType.java
diff --git
a/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/type/CharType.java
b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/type/CharType.java
similarity index 100%
rename from
flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/type/CharType.java
rename to
flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/type/CharType.java
diff --git
a/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/type/DateType.java
b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/type/DateType.java
similarity index 100%
rename from
flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/type/DateType.java
rename to
flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/type/DateType.java
diff --git
a/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/type/DecimalType.java
b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/type/DecimalType.java
similarity index 100%
rename from
flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/type/DecimalType.java
rename to
flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/type/DecimalType.java
diff --git
a/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/type/DoubleType.java
b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/type/DoubleType.java
similarity index 100%
rename from
flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/type/DoubleType.java
rename to
flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/type/DoubleType.java
diff --git
a/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/type/FloatType.java
b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/type/FloatType.java
similarity index 100%
rename from
flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/type/FloatType.java
rename to
flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/type/FloatType.java
diff --git
a/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/type/GenericType.java
b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/type/GenericType.java
similarity index 100%
rename from
flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/type/GenericType.java
rename to
flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/type/GenericType.java
diff --git
a/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/type/IntType.java
b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/type/IntType.java
similarity index 100%
rename from
flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/type/IntType.java
rename to
flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/type/IntType.java
diff --git
a/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/type/InternalType.java
b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/type/InternalType.java
similarity index 100%
rename from
flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/type/InternalType.java
rename to
flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/type/InternalType.java
diff --git
a/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/type/InternalTypes.java
b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/type/InternalTypes.java
similarity index 100%
rename from
flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/type/InternalTypes.java
rename to
flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/type/InternalTypes.java
diff --git
a/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/type/LongType.java
b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/type/LongType.java
similarity index 100%
rename from
flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/type/LongType.java
rename to
flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/type/LongType.java
diff --git
a/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/type/MapType.java
b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/type/MapType.java
similarity index 100%
rename from
flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/type/MapType.java
rename to
flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/type/MapType.java
diff --git
a/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/type/PrimitiveType.java
b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/type/PrimitiveType.java
similarity index 100%
rename from
flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/type/PrimitiveType.java
rename to
flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/type/PrimitiveType.java
diff --git
a/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/type/RowType.java
b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/type/RowType.java
similarity index 97%
rename from
flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/type/RowType.java
rename to
flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/type/RowType.java
index 6c02996..3fee7c2 100644
---
a/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/type/RowType.java
+++
b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/type/RowType.java
@@ -46,7 +46,7 @@ public class RowType implements InternalType {
}
}
- private static String[] generateDefaultFieldNames(int length) {
+ public static String[] generateDefaultFieldNames(int length) {
String[] fieldNames = new String[length];
for (int i = 0; i < length; i++) {
fieldNames[i] = "f" + i;
diff --git
a/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/type/ShortType.java
b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/type/ShortType.java
similarity index 100%
rename from
flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/type/ShortType.java
rename to
flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/type/ShortType.java
diff --git
a/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/type/StringType.java
b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/type/StringType.java
similarity index 100%
rename from
flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/type/StringType.java
rename to
flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/type/StringType.java
diff --git
a/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/type/TimeType.java
b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/type/TimeType.java
similarity index 100%
copy from
flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/type/TimeType.java
copy to
flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/type/TimeType.java
diff --git
a/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/type/TimestampType.java
b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/type/TimestampType.java
similarity index 100%
rename from
flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/type/TimestampType.java
rename to
flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/type/TimestampType.java
diff --git
a/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/type/TypeConverters.java
b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/type/TypeConverters.java
similarity index 64%
rename from
flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/type/TypeConverters.java
rename to
flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/type/TypeConverters.java
index 5bcf99d..9f9d055 100644
---
a/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/type/TypeConverters.java
+++
b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/type/TypeConverters.java
@@ -29,6 +29,10 @@ import
org.apache.flink.api.java.typeutils.ObjectArrayTypeInfo;
import org.apache.flink.api.java.typeutils.PojoTypeInfo;
import org.apache.flink.api.java.typeutils.RowTypeInfo;
import org.apache.flink.api.java.typeutils.TupleTypeInfo;
+import org.apache.flink.table.typeutils.BaseRowTypeInfo;
+import org.apache.flink.table.typeutils.BinaryArrayTypeInfo;
+import org.apache.flink.table.typeutils.BinaryMapTypeInfo;
+import org.apache.flink.table.typeutils.BinaryStringTypeInfo;
import java.util.Collections;
import java.util.HashMap;
@@ -41,7 +45,8 @@ import java.util.stream.Stream;
*/
public class TypeConverters {
- public static final Map<TypeInformation, InternalType>
TYPE_INFO_TO_INTERNAL_TYPE;
+ private static final Map<TypeInformation, InternalType>
TYPE_INFO_TO_INTERNAL_TYPE;
+ private static final Map<InternalType, TypeInformation>
INTERNAL_TYPE_TO_INTERNAL_TYPE_INFO;
static {
Map<TypeInformation, InternalType> tiToType = new HashMap<>();
tiToType.put(BasicTypeInfo.STRING_TYPE_INFO,
InternalTypes.STRING);
@@ -62,7 +67,26 @@ public class TypeConverters {
// Decimal(38, 18). If the user's BigDecimal is more precision
than this, we will
// throw Exception to remind user to use GenericType in real
data conversion.
tiToType.put(BasicTypeInfo.BIG_DEC_TYPE_INFO,
InternalTypes.SYSTEM_DEFAULT_DECIMAL);
+
+ // Internal type info
+ tiToType.put(BinaryStringTypeInfo.INSTANCE,
InternalTypes.STRING);
+
TYPE_INFO_TO_INTERNAL_TYPE =
Collections.unmodifiableMap(tiToType);
+
+ Map<InternalType, TypeInformation> internalTypeToInfo = new
HashMap<>();
+ internalTypeToInfo.put(InternalTypes.STRING,
BinaryStringTypeInfo.INSTANCE);
+ internalTypeToInfo.put(InternalTypes.BOOLEAN,
BasicTypeInfo.BOOLEAN_TYPE_INFO);
+ internalTypeToInfo.put(InternalTypes.DOUBLE,
BasicTypeInfo.DOUBLE_TYPE_INFO);
+ internalTypeToInfo.put(InternalTypes.FLOAT,
BasicTypeInfo.FLOAT_TYPE_INFO);
+ internalTypeToInfo.put(InternalTypes.BYTE,
BasicTypeInfo.BYTE_TYPE_INFO);
+ internalTypeToInfo.put(InternalTypes.INT,
BasicTypeInfo.INT_TYPE_INFO);
+ internalTypeToInfo.put(InternalTypes.LONG,
BasicTypeInfo.LONG_TYPE_INFO);
+ internalTypeToInfo.put(InternalTypes.SHORT,
BasicTypeInfo.SHORT_TYPE_INFO);
+ internalTypeToInfo.put(InternalTypes.CHAR,
BasicTypeInfo.CHAR_TYPE_INFO);
+ internalTypeToInfo.put(InternalTypes.DATE,
BasicTypeInfo.INT_TYPE_INFO);
+ internalTypeToInfo.put(InternalTypes.TIMESTAMP,
BasicTypeInfo.LONG_TYPE_INFO);
+ internalTypeToInfo.put(InternalTypes.TIME,
BasicTypeInfo.INT_TYPE_INFO);
+ INTERNAL_TYPE_TO_INTERNAL_TYPE_INFO =
Collections.unmodifiableMap(internalTypeToInfo);
}
/**
@@ -110,8 +134,42 @@ public class TypeConverters {
return InternalTypes.createMapType(
createInternalTypeFromTypeInfo(mapType.getKeyTypeInfo()),
createInternalTypeFromTypeInfo(mapType.getValueTypeInfo()));
+ } else if (typeInfo instanceof BinaryMapTypeInfo) {
+ BinaryMapTypeInfo mapType = (BinaryMapTypeInfo)
typeInfo;
+ return InternalTypes.createMapType(
+ mapType.getKeyType(),
mapType.getValueType());
+ } else if (typeInfo instanceof BinaryArrayTypeInfo) {
+ BinaryArrayTypeInfo arrayType = (BinaryArrayTypeInfo)
typeInfo;
+ return
InternalTypes.createArrayType(arrayType.getElementType());
} else {
return InternalTypes.createGenericType(typeInfo);
}
}
+
+ /**
+ * Create a internal {@link TypeInformation} from a {@link
InternalType}.
+ *
+ * <p>eg:
+ * {@link InternalTypes#STRING} => {@link BinaryStringTypeInfo}.
+ * {@link RowType} => {@link BaseRowTypeInfo}.
+ */
+ public static TypeInformation
createInternalTypeInfoFromInternalType(InternalType type) {
+ TypeInformation typeInfo =
INTERNAL_TYPE_TO_INTERNAL_TYPE_INFO.get(type);
+ if (typeInfo != null) {
+ return typeInfo;
+ }
+
+ if (type instanceof RowType) {
+ RowType rowType = (RowType) type;
+ return new BaseRowTypeInfo(rowType.getFieldTypes(),
rowType.getFieldNames());
+ } else if (type instanceof ArrayType) {
+ return new BinaryArrayTypeInfo(((ArrayType)
type).getElementType());
+ } else if (type instanceof MapType) {
+ MapType mapType = (MapType) type;
+ return new BinaryMapTypeInfo(mapType.getKeyType(),
mapType.getValueType());
+ } else {
+ // TODO support decimal and generic type.
+ throw new UnsupportedOperationException("Not support
yet!");
+ }
+ }
}
diff --git
a/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/typeutils/BaseRowSerializer.java
b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/typeutils/BaseRowSerializer.java
new file mode 100644
index 0000000..55ecb6f
--- /dev/null
+++
b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/typeutils/BaseRowSerializer.java
@@ -0,0 +1,282 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.typeutils;
+
+import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.typeutils.CompositeTypeSerializerUtil;
+import org.apache.flink.api.common.typeutils.NestedSerializersSnapshotDelegate;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.common.typeutils.TypeSerializerSchemaCompatibility;
+import org.apache.flink.api.common.typeutils.TypeSerializerSnapshot;
+import org.apache.flink.api.java.typeutils.runtime.DataInputViewStream;
+import org.apache.flink.api.java.typeutils.runtime.DataOutputViewStream;
+import org.apache.flink.core.memory.DataInputView;
+import org.apache.flink.core.memory.DataOutputView;
+import org.apache.flink.table.dataformat.BaseRow;
+import org.apache.flink.table.dataformat.BinaryRow;
+import org.apache.flink.table.dataformat.BinaryRowWriter;
+import org.apache.flink.table.dataformat.BinaryWriter;
+import org.apache.flink.table.dataformat.GenericRow;
+import org.apache.flink.table.dataformat.TypeGetterSetters;
+import org.apache.flink.table.type.InternalType;
+import org.apache.flink.table.type.TypeConverters;
+import org.apache.flink.util.InstantiationUtil;
+
+import java.io.IOException;
+import java.util.Arrays;
+
+/**
+ * Serializer for BaseRow.
+ */
+public class BaseRowSerializer extends TypeSerializer<BaseRow> {
+
+ private BinaryRowSerializer binarySerializer;
+ private final InternalType[] types;
+ private final TypeSerializer[] fieldSerializers;
+
+ public BaseRowSerializer(InternalType[] types, ExecutionConfig config) {
+ this(types, Arrays.stream(types)
+
.map(TypeConverters::createInternalTypeInfoFromInternalType)
+ .map(ti -> ti.createSerializer(config))
+ .toArray(TypeSerializer[]::new));
+ }
+
+ public BaseRowSerializer(InternalType[] types, TypeSerializer[]
fieldSerializers) {
+ this.types = types;
+ this.fieldSerializers = fieldSerializers;
+ this.binarySerializer = new BinaryRowSerializer(types.length);
+ }
+
+ public int getArity() {
+ return types.length;
+ }
+
+ @Override
+ public TypeSerializer<BaseRow> duplicate() {
+ return new BaseRowSerializer(types, fieldSerializers);
+ }
+
+ @Override
+ public BaseRow createInstance() {
+ // default use binary row to deserializer
+ return new BinaryRow(types.length);
+ }
+
+ @Override
+ public BaseRow copy(BaseRow from) {
+ if (from.getArity() != types.length) {
+ throw new IllegalArgumentException("Row arity: " +
from.getArity() +
+ ", but serializer arity: " +
types.length);
+ }
+ if (from instanceof BinaryRow) {
+ return ((BinaryRow) from).copy();
+ } else {
+ return copyBaseRow(from, new
GenericRow(from.getArity()));
+ }
+ }
+
+ @Override
+ public BaseRow copy(BaseRow from, BaseRow reuse) {
+ if (from.getArity() != types.length || reuse.getArity() !=
types.length) {
+ throw new IllegalArgumentException("Row arity: " +
from.getArity() +
+ ", Ruese Row arity: " +
reuse.getArity() +
+ ", but serializer arity: " +
types.length);
+ }
+ if (from instanceof BinaryRow) {
+ return reuse instanceof BinaryRow ?
+ ((BinaryRow) from).copy((BinaryRow)
reuse) :
+ ((BinaryRow) from).copy();
+ } else {
+ return copyBaseRow(from, reuse);
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ private BaseRow copyBaseRow(BaseRow from, BaseRow reuse) {
+ GenericRow ret;
+ if (reuse instanceof GenericRow) {
+ ret = (GenericRow) reuse;
+ } else {
+ ret = new GenericRow(from.getArity());
+ }
+ ret.setHeader(from.getHeader());
+ for (int i = 0; i < from.getArity(); i++) {
+ if (!from.isNullAt(i)) {
+ ret.setField(i,
fieldSerializers[i].copy((TypeGetterSetters.get(from, i, types[i]))));
+ } else {
+ ret.setNullAt(i);
+ }
+ }
+ return ret;
+ }
+
+ @Override
+ public void copy(DataInputView source, DataOutputView target) throws
IOException {
+ binarySerializer.copy(source, target);
+ }
+
+ /**
+ * Convert base row to binary row.
+ * TODO modify it to code gen, and reuse BinaryRow&BinaryRowWriter.
+ */
+ public BinaryRow baseRowToBinary(BaseRow row) {
+ if (row instanceof BinaryRow) {
+ return (BinaryRow) row;
+ }
+ BinaryRow binaryRow = new BinaryRow(types.length);
+ BinaryRowWriter writer = new BinaryRowWriter(binaryRow);
+ for (int i = 0; i < types.length; i++) {
+ if (row.isNullAt(i)) {
+ writer.setNullAt(i);
+ } else {
+ BinaryWriter.write(writer, i,
TypeGetterSetters.get(row, i, types[i]), types[i]);
+ }
+ }
+ writer.complete();
+ return binaryRow;
+ }
+
+ @Override
+ public void serialize(BaseRow row, DataOutputView target) throws
IOException {
+ binarySerializer.serialize(baseRowToBinary(row), target);
+ }
+
+ @Override
+ public BaseRow deserialize(DataInputView source) throws IOException {
+ return binarySerializer.deserialize(source);
+ }
+
+ @Override
+ public BaseRow deserialize(BaseRow reuse, DataInputView source) throws
IOException {
+ if (reuse instanceof BinaryRow) {
+ return binarySerializer.deserialize((BinaryRow) reuse,
source);
+ } else {
+ return binarySerializer.deserialize(source);
+ }
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ if (obj instanceof BaseRowSerializer) {
+ BaseRowSerializer other = (BaseRowSerializer) obj;
+ return Arrays.equals(types, other.types);
+ }
+
+ return false;
+ }
+
+ @Override
+ public int hashCode() {
+ return Arrays.hashCode(types);
+ }
+
+ @Override
+ public boolean isImmutableType() {
+ return false;
+ }
+
+ @Override
+ public int getLength() {
+ return -1;
+ }
+
+ @Override
+ public TypeSerializerSnapshot<BaseRow> snapshotConfiguration() {
+ return new BaseRowSerializerSnapshot(types, fieldSerializers);
+ }
+
+ /**
+ * {@link TypeSerializerSnapshot} for {@link BinaryRowSerializer}.
+ */
+ public static final class BaseRowSerializerSnapshot implements
TypeSerializerSnapshot<BaseRow> {
+ private static final int CURRENT_VERSION = 3;
+
+ private InternalType[] previousTypes;
+ private NestedSerializersSnapshotDelegate
nestedSerializersSnapshotDelegate;
+
+ @SuppressWarnings("unused")
+ public BaseRowSerializerSnapshot() {
+ // this constructor is used when restoring from a
checkpoint/savepoint.
+ }
+
+ BaseRowSerializerSnapshot(InternalType[] types,
TypeSerializer[] serializers) {
+ this.previousTypes = types;
+ this.nestedSerializersSnapshotDelegate = new
NestedSerializersSnapshotDelegate(serializers);
+ }
+
+ @Override
+ public int getCurrentVersion() {
+ return CURRENT_VERSION;
+ }
+
+ @Override
+ public void writeSnapshot(DataOutputView out) throws
IOException {
+ out.writeInt(previousTypes.length);
+ DataOutputViewStream stream = new
DataOutputViewStream(out);
+ for (InternalType previousType : previousTypes) {
+ InstantiationUtil.serializeObject(stream,
previousType);
+ }
+
nestedSerializersSnapshotDelegate.writeNestedSerializerSnapshots(out);
+ }
+
+ @Override
+ public void readSnapshot(int readVersion, DataInputView in,
ClassLoader userCodeClassLoader) throws IOException {
+ int length = in.readInt();
+ DataInputViewStream stream = new
DataInputViewStream(in);
+ previousTypes = new InternalType[length];
+ for (int i = 0; i < length; i++) {
+ try {
+ previousTypes[i] =
InstantiationUtil.deserializeObject(stream, userCodeClassLoader);
+ } catch (ClassNotFoundException e) {
+ throw new IOException(e);
+ }
+ }
+ this.nestedSerializersSnapshotDelegate =
NestedSerializersSnapshotDelegate.readNestedSerializerSnapshots(in,
userCodeClassLoader);
+ }
+
+ @Override
+ public BaseRowSerializer restoreSerializer() {
+ return new BaseRowSerializer(previousTypes,
nestedSerializersSnapshotDelegate.getRestoredNestedSerializers());
+ }
+
+ @Override
+ public TypeSerializerSchemaCompatibility<BaseRow>
resolveSchemaCompatibility(TypeSerializer<BaseRow> newSerializer) {
+ if (!(newSerializer instanceof BaseRowSerializer)) {
+ return
TypeSerializerSchemaCompatibility.incompatible();
+ }
+
+ BaseRowSerializer newRowSerializer =
(BaseRowSerializer) newSerializer;
+ if (!Arrays.equals(previousTypes,
newRowSerializer.types)) {
+ return
TypeSerializerSchemaCompatibility.incompatible();
+ }
+
+
CompositeTypeSerializerUtil.IntermediateCompatibilityResult<BaseRow>
intermediateResult =
+
CompositeTypeSerializerUtil.constructIntermediateCompatibilityResult(
+
newRowSerializer.fieldSerializers,
+
nestedSerializersSnapshotDelegate.getNestedSerializerSnapshots());
+
+ if
(intermediateResult.isCompatibleWithReconfiguredSerializer()) {
+ BaseRowSerializer
reconfiguredCompositeSerializer = restoreSerializer();
+ return
TypeSerializerSchemaCompatibility.compatibleWithReconfiguredSerializer(reconfiguredCompositeSerializer);
+ }
+
+ return intermediateResult.getFinalResult();
+ }
+ }
+}
diff --git
a/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/typeutils/BaseRowTypeInfo.java
b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/typeutils/BaseRowTypeInfo.java
new file mode 100644
index 0000000..b942c6a
--- /dev/null
+++
b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/typeutils/BaseRowTypeInfo.java
@@ -0,0 +1,145 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.typeutils;
+
+import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeutils.CompositeType;
+import org.apache.flink.api.common.typeutils.TypeComparator;
+import org.apache.flink.api.java.typeutils.TupleTypeInfoBase;
+import org.apache.flink.table.dataformat.BaseRow;
+import org.apache.flink.table.type.InternalType;
+import org.apache.flink.table.type.RowType;
+import org.apache.flink.table.type.TypeConverters;
+
+import java.util.Arrays;
+import java.util.HashSet;
+
+import static org.apache.flink.util.Preconditions.checkArgument;
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * Base row type info.
+ */
+public class BaseRowTypeInfo extends TupleTypeInfoBase<BaseRow> {
+
+ private static final long serialVersionUID = 1L;
+
+ private final String[] fieldNames;
+ private final InternalType[] internalTypes;
+
+ public BaseRowTypeInfo(InternalType... internalTypes) {
+ this(internalTypes,
RowType.generateDefaultFieldNames(internalTypes.length));
+ }
+
+ public BaseRowTypeInfo(InternalType[] internalTypes, String[]
fieldNames) {
+ super(BaseRow.class, Arrays.stream(internalTypes)
+
.map(TypeConverters::createInternalTypeInfoFromInternalType)
+ .toArray(TypeInformation[]::new));
+ this.internalTypes = internalTypes;
+ checkNotNull(fieldNames, "FieldNames should not be null.");
+ checkArgument(internalTypes.length == fieldNames.length,
+ "Number of field types and names is
different.");
+ checkArgument(!hasDuplicateFieldNames(fieldNames),
+ "Field names are not unique.");
+ this.fieldNames = Arrays.copyOf(fieldNames, fieldNames.length);
+ }
+
+ @Override
+ public <X> TypeInformation<X> getTypeAt(String fieldExpression) {
+ throw new UnsupportedOperationException("Not support!");
+ }
+
+ @Override
+ public TypeComparator<BaseRow> createComparator(
+ int[] logicalKeyFields,
+ boolean[] orders,
+ int logicalFieldOffset,
+ ExecutionConfig config) {
+ // TODO support it.
+ throw new UnsupportedOperationException("Not support yet!");
+ }
+
+ @Override
+ public String[] getFieldNames() {
+ return fieldNames;
+ }
+
+ @Override
+ public int getFieldIndex(String fieldName) {
+ for (int i = 0; i < fieldNames.length; i++) {
+ if (fieldNames[i].equals(fieldName)) {
+ return i;
+ }
+ }
+ return -1;
+ }
+
+ @Override
+ public boolean canEqual(Object obj) {
+ return obj instanceof BaseRowTypeInfo;
+ }
+
+ @Override
+ public int hashCode() {
+ return 31 * super.hashCode() + Arrays.hashCode(fieldNames);
+ }
+
+ @Override
+ public String toString() {
+ StringBuilder bld = new StringBuilder("BaseRow");
+ if (types.length > 0) {
+ bld.append('(').append(fieldNames[0]).append(":
").append(types[0]);
+
+ for (int i = 1; i < types.length; i++) {
+ bld.append(",
").append(fieldNames[i]).append(": ").append(types[i]);
+ }
+
+ bld.append(')');
+ }
+ return bld.toString();
+ }
+
+ /**
+ * Returns the field types of the row. The order matches the order of
the field names.
+ */
+ public TypeInformation<?>[] getFieldTypes() {
+ return types;
+ }
+
+ private boolean hasDuplicateFieldNames(String[] fieldNames) {
+ HashSet<String> names = new HashSet<>();
+ for (String field : fieldNames) {
+ if (!names.add(field)) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ @Override
+ public CompositeType.TypeComparatorBuilder<BaseRow>
createTypeComparatorBuilder() {
+ throw new UnsupportedOperationException("Not support!");
+ }
+
+ @Override
+ public BaseRowSerializer createSerializer(ExecutionConfig config) {
+ return new BaseRowSerializer(internalTypes, config);
+ }
+}
diff --git
a/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/typeutils/BinaryArraySerializer.java
b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/typeutils/BinaryArraySerializer.java
new file mode 100644
index 0000000..9526e30
--- /dev/null
+++
b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/typeutils/BinaryArraySerializer.java
@@ -0,0 +1,107 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.typeutils;
+
+import org.apache.flink.api.common.typeutils.SimpleTypeSerializerSnapshot;
+import org.apache.flink.api.common.typeutils.TypeSerializerSnapshot;
+import org.apache.flink.api.common.typeutils.base.TypeSerializerSingleton;
+import org.apache.flink.core.memory.DataInputView;
+import org.apache.flink.core.memory.DataOutputView;
+import org.apache.flink.core.memory.MemorySegmentFactory;
+import org.apache.flink.table.dataformat.BinaryArray;
+import org.apache.flink.table.util.SegmentsUtil;
+
+import java.io.IOException;
+
+/**
+ * Serializer for {@link BinaryArray}.
+ */
+public class BinaryArraySerializer extends
TypeSerializerSingleton<BinaryArray> {
+
+ public static final BinaryArraySerializer INSTANCE = new
BinaryArraySerializer();
+
+ private BinaryArraySerializer() {}
+
+ @Override
+ public boolean isImmutableType() {
+ return false;
+ }
+
+ @Override
+ public BinaryArray createInstance() {
+ return new BinaryArray();
+ }
+
+ @Override
+ public BinaryArray copy(BinaryArray from) {
+ return from.copy();
+ }
+
+ @Override
+ public BinaryArray copy(BinaryArray from, BinaryArray reuse) {
+ return from.copy(reuse);
+ }
+
+ @Override
+ public int getLength() {
+ return -1;
+ }
+
+ @Override
+ public void serialize(BinaryArray record, DataOutputView target) throws
IOException {
+ target.writeInt(record.getSizeInBytes());
+ SegmentsUtil.serializeToView(record.getSegments(),
record.getOffset(), record.getSizeInBytes(), target);
+ }
+
+ @Override
+ public BinaryArray deserialize(DataInputView source) throws IOException
{
+ return deserialize(new BinaryArray(), source);
+ }
+
+ @Override
+ public BinaryArray deserialize(BinaryArray reuse, DataInputView source)
throws IOException {
+ int length = source.readInt();
+ byte[] bytes = new byte[length];
+ source.readFully(bytes);
+ reuse.pointTo(MemorySegmentFactory.wrap(bytes), 0,
bytes.length);
+ return reuse;
+ }
+
+ @Override
+ public void copy(DataInputView source, DataOutputView target) throws
IOException {
+ int length = source.readInt();
+ target.writeInt(length);
+ target.write(source, length);
+ }
+
+ @Override
+ public TypeSerializerSnapshot<BinaryArray> snapshotConfiguration() {
+ return new BinaryArraySerializerSnapshot();
+ }
+
+ /**
+ * Serializer configuration snapshot for compatibility and format
evolution.
+ */
+ @SuppressWarnings("WeakerAccess")
+ public static final class BinaryArraySerializerSnapshot extends
SimpleTypeSerializerSnapshot<BinaryArray> {
+
+ public BinaryArraySerializerSnapshot() {
+ super(() -> INSTANCE);
+ }
+ }
+}
diff --git
a/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/typeutils/BinaryArrayTypeInfo.java
b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/typeutils/BinaryArrayTypeInfo.java
new file mode 100644
index 0000000..524b807
--- /dev/null
+++
b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/typeutils/BinaryArrayTypeInfo.java
@@ -0,0 +1,105 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.typeutils;
+
+import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.table.dataformat.BinaryArray;
+import org.apache.flink.table.type.InternalType;
+
+/**
+ * TypeInfo for BaseArray.
+ */
+public class BinaryArrayTypeInfo extends TypeInformation<BinaryArray> {
+
+ private static final long serialVersionUID = 1L;
+
+ private final InternalType elementType;
+
+ public BinaryArrayTypeInfo(InternalType elementType) {
+ this.elementType = elementType;
+ }
+
+ public InternalType getElementType() {
+ return elementType;
+ }
+
+ @Override
+ public boolean isBasicType() {
+ return false;
+ }
+
+ @Override
+ public boolean isTupleType() {
+ return false;
+ }
+
+ @Override
+ public int getArity() {
+ return 1;
+ }
+
+ @Override
+ public int getTotalFields() {
+ return 1;
+ }
+
+ @Override
+ public Class<BinaryArray> getTypeClass() {
+ return BinaryArray.class;
+ }
+
+ @Override
+ public boolean isKeyType() {
+ return false;
+ }
+
+ @Override
+ public TypeSerializer<BinaryArray> createSerializer(ExecutionConfig
executionConfig) {
+ return BinaryArraySerializer.INSTANCE;
+ }
+
+ @Override
+ public String toString() {
+ return this.getClass().getSimpleName() + "<" + this.elementType
+ ">";
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ if (obj instanceof BinaryArrayTypeInfo) {
+ BinaryArrayTypeInfo typeInfo = (BinaryArrayTypeInfo)
obj;
+
+ return typeInfo.canEqual(this) &&
+
elementType.equals(typeInfo.elementType);
+ } else {
+ return false;
+ }
+ }
+
+ @Override
+ public boolean canEqual(Object obj) {
+ return obj instanceof BinaryArrayTypeInfo;
+ }
+
+ @Override
+ public int hashCode() {
+ return this.elementType.hashCode();
+ }
+}
diff --git
a/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/typeutils/BinaryMapSerializer.java
b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/typeutils/BinaryMapSerializer.java
new file mode 100644
index 0000000..6030026
--- /dev/null
+++
b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/typeutils/BinaryMapSerializer.java
@@ -0,0 +1,107 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.typeutils;
+
+import org.apache.flink.api.common.typeutils.SimpleTypeSerializerSnapshot;
+import org.apache.flink.api.common.typeutils.TypeSerializerSnapshot;
+import org.apache.flink.api.common.typeutils.base.TypeSerializerSingleton;
+import org.apache.flink.core.memory.DataInputView;
+import org.apache.flink.core.memory.DataOutputView;
+import org.apache.flink.core.memory.MemorySegmentFactory;
+import org.apache.flink.table.dataformat.BinaryMap;
+import org.apache.flink.table.util.SegmentsUtil;
+
+import java.io.IOException;
+
+/**
+ * Serializer for {@link BinaryMap}.
+ */
+public class BinaryMapSerializer extends TypeSerializerSingleton<BinaryMap> {
+
+ public static final BinaryMapSerializer INSTANCE = new
BinaryMapSerializer();
+
+ private BinaryMapSerializer() {}
+
+ @Override
+ public boolean isImmutableType() {
+ return false;
+ }
+
+ @Override
+ public BinaryMap createInstance() {
+ return new BinaryMap();
+ }
+
+ @Override
+ public BinaryMap copy(BinaryMap from) {
+ return from.copy();
+ }
+
+ @Override
+ public BinaryMap copy(BinaryMap from, BinaryMap reuse) {
+ return from.copy(reuse);
+ }
+
+ @Override
+ public int getLength() {
+ return -1;
+ }
+
+ @Override
+ public void serialize(BinaryMap record, DataOutputView target) throws
IOException {
+ target.writeInt(record.getSizeInBytes());
+ SegmentsUtil.serializeToView(record.getSegments(),
record.getOffset(), record.getSizeInBytes(), target);
+ }
+
+ @Override
+ public BinaryMap deserialize(DataInputView source) throws IOException {
+ return deserialize(new BinaryMap(), source);
+ }
+
+ @Override
+ public BinaryMap deserialize(BinaryMap reuse, DataInputView source)
throws IOException {
+ int length = source.readInt();
+ byte[] bytes = new byte[length];
+ source.readFully(bytes);
+ reuse.pointTo(MemorySegmentFactory.wrap(bytes), 0,
bytes.length);
+ return reuse;
+ }
+
+ @Override
+ public void copy(DataInputView source, DataOutputView target) throws
IOException {
+ int length = source.readInt();
+ target.writeInt(length);
+ target.write(source, length);
+ }
+
+ @Override
+ public TypeSerializerSnapshot<BinaryMap> snapshotConfiguration() {
+ return new BinaryMapSerializerSnapshot();
+ }
+
+ /**
+ * Serializer configuration snapshot for compatibility and format
evolution.
+ */
+ @SuppressWarnings("WeakerAccess")
+ public static final class BinaryMapSerializerSnapshot extends
SimpleTypeSerializerSnapshot<BinaryMap> {
+
+ public BinaryMapSerializerSnapshot() {
+ super(() -> INSTANCE);
+ }
+ }
+}
diff --git
a/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/typeutils/BinaryMapTypeInfo.java
b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/typeutils/BinaryMapTypeInfo.java
new file mode 100644
index 0000000..ca6dbed
--- /dev/null
+++
b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/typeutils/BinaryMapTypeInfo.java
@@ -0,0 +1,113 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.typeutils;
+
+import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.table.dataformat.BinaryMap;
+import org.apache.flink.table.type.InternalType;
+
+/**
+ * TypeInfo for BaseMap.
+ */
+public class BinaryMapTypeInfo extends TypeInformation<BinaryMap> {
+
+ private static final long serialVersionUID = 1L;
+
+ private final InternalType keyType;
+ private final InternalType valueType;
+
+ public BinaryMapTypeInfo(InternalType keyType, InternalType valueType) {
+ this.keyType = keyType;
+ this.valueType = valueType;
+ }
+
+ //
--------------------------------------------------------------------------------------------
+
+ public InternalType getKeyType() {
+ return keyType;
+ }
+
+ public InternalType getValueType() {
+ return valueType;
+ }
+
+ @Override
+ public boolean isBasicType() {
+ return false;
+ }
+
+ @Override
+ public boolean isTupleType() {
+ return false;
+ }
+
+ @Override
+ public int getArity() {
+ return 1;
+ }
+
+ @Override
+ public int getTotalFields() {
+ return 1;
+ }
+
+ @Override
+ public Class<BinaryMap> getTypeClass() {
+ return BinaryMap.class;
+ }
+
+ @Override
+ public boolean isKeyType() {
+ return false;
+ }
+
+ @Override
+ public TypeSerializer<BinaryMap> createSerializer(ExecutionConfig
executionConfig) {
+ return BinaryMapSerializer.INSTANCE;
+ }
+
+ @Override
+ public String toString() {
+ return this.getClass().getSimpleName() +
+ "<" + keyType + ", " + valueType + ">";
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ if (obj instanceof BinaryMapTypeInfo) {
+ BinaryMapTypeInfo typeInfo = (BinaryMapTypeInfo) obj;
+
+ return keyType.equals(typeInfo.keyType) &&
valueType.equals(typeInfo.valueType);
+ } else {
+ return false;
+ }
+ }
+
+ @Override
+ public boolean canEqual(Object obj) {
+ return obj instanceof BinaryMapTypeInfo;
+ }
+
+ @Override
+ public int hashCode() {
+ return 31 * keyType.hashCode() + valueType.hashCode();
+ }
+}
diff --git
a/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/typeutils/BinaryRowSerializer.java
b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/typeutils/BinaryRowSerializer.java
new file mode 100644
index 0000000..078b948
--- /dev/null
+++
b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/typeutils/BinaryRowSerializer.java
@@ -0,0 +1,184 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.typeutils;
+
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.common.typeutils.TypeSerializerSchemaCompatibility;
+import org.apache.flink.api.common.typeutils.TypeSerializerSnapshot;
+import org.apache.flink.core.memory.DataInputView;
+import org.apache.flink.core.memory.DataOutputView;
+import org.apache.flink.core.memory.MemorySegment;
+import org.apache.flink.core.memory.MemorySegmentFactory;
+import org.apache.flink.table.dataformat.BinaryRow;
+import org.apache.flink.table.util.SegmentsUtil;
+
+import java.io.IOException;
+
+import static org.apache.flink.util.Preconditions.checkArgument;
+
+/**
+ * Serializer for {@link BinaryRow}.
+ */
+public class BinaryRowSerializer extends TypeSerializer<BinaryRow> {
+
+ private static final long serialVersionUID = 1L;
+
+ private final int numFields;
+
+ public BinaryRowSerializer(int numFields) {
+ this.numFields = numFields;
+ }
+
+ @Override
+ public boolean isImmutableType() {
+ return false;
+ }
+
+ @Override
+ public TypeSerializer<BinaryRow> duplicate() {
+ return new BinaryRowSerializer(numFields);
+ }
+
+ @Override
+ public BinaryRow createInstance() {
+ return new BinaryRow(numFields);
+ }
+
+ @Override
+ public BinaryRow copy(BinaryRow from) {
+ return copy(from, new BinaryRow(numFields));
+ }
+
+ @Override
+ public BinaryRow copy(BinaryRow from, BinaryRow reuse) {
+ return from.copy(reuse);
+ }
+
+ @Override
+ public int getLength() {
+ return -1;
+ }
+
+ @Override
+ public void serialize(BinaryRow record, DataOutputView target) throws
IOException {
+ target.writeInt(record.getSizeInBytes());
+ SegmentsUtil.serializeToView(record.getSegments(),
record.getOffset(), record.getSizeInBytes(), target);
+ }
+
+ @Override
+ public BinaryRow deserialize(DataInputView source) throws IOException {
+ BinaryRow row = new BinaryRow(numFields);
+ int length = source.readInt();
+ byte[] bytes = new byte[length];
+ source.readFully(bytes);
+ row.pointTo(MemorySegmentFactory.wrap(bytes), 0, length);
+ return row;
+ }
+
+ @Override
+ public BinaryRow deserialize(BinaryRow reuse, DataInputView source)
throws IOException {
+ MemorySegment[] segments = reuse.getSegments();
+ checkArgument(segments == null || (segments.length == 1 &&
reuse.getOffset() == 0),
+ "Reuse BinaryRow should have no segments or
only one segment and offset start at 0.");
+
+ int length = source.readInt();
+ if (segments == null || segments[0].size() < length) {
+ segments = new MemorySegment[]
{MemorySegmentFactory.wrap(new byte[length])};
+ }
+ source.readFully(segments[0].getArray(), 0, length);
+ reuse.pointTo(segments, 0, length);
+ return reuse;
+ }
+
+ @Override
+ public void copy(DataInputView source, DataOutputView target) throws
IOException {
+ int length = source.readInt();
+ target.writeInt(length);
+ target.write(source, length);
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ return obj instanceof BinaryRowSerializer
+ && numFields == ((BinaryRowSerializer)
obj).numFields;
+ }
+
+ @Override
+ public int hashCode() {
+ return Integer.hashCode(numFields);
+ }
+
+ @Override
+ public TypeSerializerSnapshot<BinaryRow> snapshotConfiguration() {
+ return new BinaryRowSerializerSnapshot(numFields);
+ }
+
+ /**
+ * {@link TypeSerializerSnapshot} for {@link BinaryRowSerializer}.
+ */
+ public static final class BinaryRowSerializerSnapshot implements
TypeSerializerSnapshot<BinaryRow> {
+ private static final int CURRENT_VERSION = 3;
+
+ private int previousNumFields;
+
+ @SuppressWarnings("unused")
+ public BinaryRowSerializerSnapshot() {
+ // this constructor is used when restoring from a
checkpoint/savepoint.
+ }
+
+ BinaryRowSerializerSnapshot(int numFields) {
+ this.previousNumFields = numFields;
+ }
+
+ @Override
+ public int getCurrentVersion() {
+ return CURRENT_VERSION;
+ }
+
+ @Override
+ public void writeSnapshot(DataOutputView out) throws
IOException {
+ out.writeInt(previousNumFields);
+ }
+
+ @Override
+ public void readSnapshot(int readVersion, DataInputView in,
ClassLoader userCodeClassLoader) throws IOException {
+ this.previousNumFields = in.readInt();
+ }
+
+ @Override
+ public TypeSerializer<BinaryRow> restoreSerializer() {
+ return new BinaryRowSerializer(previousNumFields);
+ }
+
+ @Override
+ public TypeSerializerSchemaCompatibility<BinaryRow>
resolveSchemaCompatibility(TypeSerializer<BinaryRow> newSerializer) {
+ if (!(newSerializer instanceof BinaryRowSerializer)) {
+ return
TypeSerializerSchemaCompatibility.incompatible();
+ }
+
+ BinaryRowSerializer newBinaryRowSerializer =
(BinaryRowSerializer) newSerializer;
+ if (previousNumFields !=
newBinaryRowSerializer.numFields) {
+ return
TypeSerializerSchemaCompatibility.incompatible();
+ } else {
+ return
TypeSerializerSchemaCompatibility.compatibleAsIs();
+ }
+ }
+ }
+
+}
diff --git
a/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/typeutils/BinaryStringSerializer.java
b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/typeutils/BinaryStringSerializer.java
new file mode 100644
index 0000000..0a45c8a
--- /dev/null
+++
b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/typeutils/BinaryStringSerializer.java
@@ -0,0 +1,114 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.typeutils;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.typeutils.SimpleTypeSerializerSnapshot;
+import org.apache.flink.api.common.typeutils.TypeSerializerSnapshot;
+import org.apache.flink.api.common.typeutils.base.TypeSerializerSingleton;
+import org.apache.flink.core.memory.DataInputView;
+import org.apache.flink.core.memory.DataOutputView;
+import org.apache.flink.table.dataformat.BinaryString;
+import org.apache.flink.table.util.SegmentsUtil;
+
+import java.io.IOException;
+
+/**
+ * Serializer for {@link BinaryString}.
+ */
+@Internal
+public final class BinaryStringSerializer extends
TypeSerializerSingleton<BinaryString> {
+
+ private static final long serialVersionUID = 1L;
+
+ public static final BinaryStringSerializer INSTANCE = new
BinaryStringSerializer();
+
+ private BinaryStringSerializer() {}
+
+ @Override
+ public boolean isImmutableType() {
+ return true;
+ }
+
+ @Override
+ public BinaryString createInstance() {
+ return BinaryString.fromString("");
+ }
+
+ @Override
+ public BinaryString copy(BinaryString from) {
+ return from.copy();
+ }
+
+ @Override
+ public BinaryString copy(BinaryString from, BinaryString reuse) {
+ return from.copy();
+ }
+
+ @Override
+ public int getLength() {
+ return -1;
+ }
+
+ @Override
+ public void serialize(BinaryString record, DataOutputView target)
throws IOException {
+ target.writeInt(record.getSizeInBytes());
+ SegmentsUtil.serializeToView(record.getSegments(),
record.getOffset(), record.getSizeInBytes(), target);
+ }
+
+ @Override
+ public BinaryString deserialize(DataInputView source) throws
IOException {
+ return deserializeInternal(source);
+ }
+
+ public static BinaryString deserializeInternal(DataInputView source)
throws IOException {
+ int length = source.readInt();
+ byte[] bytes = new byte[length];
+ source.readFully(bytes);
+ return BinaryString.fromBytes(bytes);
+ }
+
+ @Override
+ public BinaryString deserialize(BinaryString record, DataInputView
source) throws IOException {
+ return deserialize(source);
+ }
+
+ @Override
+ public void copy(DataInputView source, DataOutputView target) throws
IOException {
+ int length = source.readInt();
+ target.writeInt(length);
+ target.write(source, length);
+ }
+
+ @Override
+ public TypeSerializerSnapshot<BinaryString> snapshotConfiguration() {
+ return new BinaryStringSerializerSnapshot();
+ }
+
+ /**
+ * Serializer configuration snapshot for compatibility and format
evolution.
+ */
+ @SuppressWarnings("WeakerAccess")
+ public static final class BinaryStringSerializerSnapshot extends
SimpleTypeSerializerSnapshot<BinaryString> {
+
+ public BinaryStringSerializerSnapshot() {
+ super(() -> INSTANCE);
+ }
+ }
+}
diff --git
a/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/typeutils/BinaryStringTypeInfo.java
b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/typeutils/BinaryStringTypeInfo.java
new file mode 100644
index 0000000..f4a2a19
--- /dev/null
+++
b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/typeutils/BinaryStringTypeInfo.java
@@ -0,0 +1,89 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.typeutils;
+
+import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.table.dataformat.BinaryString;
+
+/**
+ * TypeInfo for BinaryString.
+ */
+public class BinaryStringTypeInfo extends TypeInformation<BinaryString> {
+
+ public static final BinaryStringTypeInfo INSTANCE = new
BinaryStringTypeInfo();
+
+ private BinaryStringTypeInfo() {}
+
+ @Override
+ public boolean isBasicType() {
+ return true;
+ }
+
+ @Override
+ public boolean isTupleType() {
+ return false;
+ }
+
+ @Override
+ public int getArity() {
+ return 1;
+ }
+
+ @Override
+ public int getTotalFields() {
+ return 1;
+ }
+
+ @Override
+ public Class<BinaryString> getTypeClass() {
+ return BinaryString.class;
+ }
+
+ @Override
+ public boolean isKeyType() {
+ return true;
+ }
+
+ @Override
+ public TypeSerializer<BinaryString> createSerializer(ExecutionConfig
config) {
+ return BinaryStringSerializer.INSTANCE;
+ }
+
+ @Override
+ public String toString() {
+ return "BinaryString";
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ return obj instanceof BinaryStringTypeInfo;
+ }
+
+ @Override
+ public int hashCode() {
+ return 0;
+ }
+
+ @Override
+ public boolean canEqual(Object obj) {
+ return obj instanceof BinaryStringTypeInfo;
+ }
+}
diff --git
a/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/util/SegmentsUtil.java
b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/util/SegmentsUtil.java
index 4ebb401..bc1148f 100644
---
a/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/util/SegmentsUtil.java
+++
b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/util/SegmentsUtil.java
@@ -18,8 +18,10 @@
package org.apache.flink.table.util;
import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.core.memory.DataOutputView;
import org.apache.flink.core.memory.MemorySegment;
+import java.io.IOException;
import java.nio.ByteOrder;
/**
@@ -808,4 +810,42 @@ public class SegmentsUtil {
}
segment.put(segOffset, (byte) (LITTLE_ENDIAN ? b2 : b1));
}
+
+ /**
+ * Serialize segments to output view with offset and size.
+ * Note: It just copies the data in, not include the length.
+ *
+ * @param segments source segments
+ * @param offset offset for segments
+ * @param sizeInBytes size in bytes
+ * @param target target output view
+ */
+ public static void serializeToView(MemorySegment[] segments, int offset,
+ int sizeInBytes, DataOutputView target) throws
IOException {
+ for (MemorySegment sourceSegment : segments) {
+ int curSegRemain = sourceSegment.size() - offset;
+ if (curSegRemain > 0) {
+ int copySize = Math.min(curSegRemain,
sizeInBytes);
+
+ // TODO after FLINK-11724
+ byte[] bytes = new byte[copySize];
+ sourceSegment.get(offset, bytes);
+ target.write(bytes);
+
+ sizeInBytes -= copySize;
+ offset = 0;
+ } else {
+ offset -= sourceSegment.size();
+ }
+
+ if (sizeInBytes == 0) {
+ return;
+ }
+ }
+
+ if (sizeInBytes != 0) {
+ throw new RuntimeException("No copy finished, this
should be a bug, " +
+ "The remaining length is: " +
sizeInBytes);
+ }
+ }
}
diff --git
a/flink-table/flink-table-planner-blink/src/test/java/org/apache/flink/table/type/InternalTypeTest.java
b/flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/type/InternalTypeTest.java
similarity index 100%
rename from
flink-table/flink-table-planner-blink/src/test/java/org/apache/flink/table/type/InternalTypeTest.java
rename to
flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/type/InternalTypeTest.java
diff --git
a/flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/typeutils/BaseRowSerializerTest.java
b/flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/typeutils/BaseRowSerializerTest.java
new file mode 100644
index 0000000..7cc8cde
--- /dev/null
+++
b/flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/typeutils/BaseRowSerializerTest.java
@@ -0,0 +1,229 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.typeutils;
+
+import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.typeutils.SerializerTestInstance;
+import org.apache.flink.table.dataformat.BaseRow;
+import org.apache.flink.table.dataformat.BinaryArray;
+import org.apache.flink.table.dataformat.BinaryArrayWriter;
+import org.apache.flink.table.dataformat.BinaryMap;
+import org.apache.flink.table.dataformat.BinaryRow;
+import org.apache.flink.table.dataformat.GenericRow;
+import org.apache.flink.table.type.InternalTypes;
+
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.ExpectedException;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+import java.util.Arrays;
+import java.util.Collection;
+
+import static org.apache.flink.table.dataformat.BinaryString.fromString;
+import static org.junit.Assert.assertEquals;
+
+/**
+ * Test for {@link BaseRowSerializer}.
+ */
+@RunWith(Parameterized.class)
+public class BaseRowSerializerTest extends SerializerTestInstance<BaseRow> {
+
+ @Rule
+ public ExpectedException thrown = ExpectedException.none();
+
+ private final BaseRowSerializer serializer;
+ private final BaseRow[] testData;
+
+ public BaseRowSerializerTest(BaseRowSerializer serializer, BaseRow[]
testData) {
+ super(serializer, BaseRow.class, -1, testData);
+ this.serializer = serializer;
+ this.testData = testData;
+ }
+
+ @Parameterized.Parameters
+ public static Collection<Object[]> parameters() {
+ return Arrays.asList(
+ testBaseRowSerializer(),
+ testLargeBaseRowSerializer(),
+ testBaseRowSerializerWithComplexTypes());
+ }
+
+ private static Object[] testBaseRowSerializer() {
+ BaseRowTypeInfo typeInfo = new
BaseRowTypeInfo(InternalTypes.INT, InternalTypes.STRING);
+ GenericRow row1 = new GenericRow(2);
+ row1.setField(0, 1);
+ row1.setField(1, fromString("a"));
+
+ GenericRow row2 = new GenericRow(2);
+ row2.setField(0, 2);
+ row2.setField(1, null);
+
+ BaseRowSerializer serializer = typeInfo.createSerializer(new
ExecutionConfig());
+ return new Object[] {serializer, new BaseRow[]{row1, row2}};
+ }
+
+ private static Object[] testLargeBaseRowSerializer() {
+ BaseRowTypeInfo typeInfo = new BaseRowTypeInfo(
+ InternalTypes.INT,
+ InternalTypes.INT,
+ InternalTypes.INT,
+ InternalTypes.INT,
+ InternalTypes.INT,
+ InternalTypes.INT,
+ InternalTypes.INT,
+ InternalTypes.INT,
+ InternalTypes.INT,
+ InternalTypes.INT,
+ InternalTypes.INT,
+ InternalTypes.INT,
+ InternalTypes.STRING);
+
+ GenericRow row = new GenericRow(13);
+ row.setField(0, 2);
+ row.setField(1, null);
+ row.setField(3, null);
+ row.setField(4, null);
+ row.setField(5, null);
+ row.setField(6, null);
+ row.setField(7, null);
+ row.setField(8, null);
+ row.setField(9, null);
+ row.setField(10, null);
+ row.setField(11, null);
+ row.setField(12, fromString("Test"));
+
+ BaseRowSerializer serializer = typeInfo.createSerializer(new
ExecutionConfig());
+ return new Object[] {serializer, new BaseRow[]{row}};
+ }
+
+ private static Object[] testBaseRowSerializerWithComplexTypes() {
+ BaseRowTypeInfo typeInfo = new BaseRowTypeInfo(
+ InternalTypes.INT,
+ InternalTypes.DOUBLE,
+ InternalTypes.STRING,
+ InternalTypes.createArrayType(InternalTypes.INT),
+ InternalTypes.createMapType(InternalTypes.INT,
InternalTypes.INT));
+
+ GenericRow[] data = new GenericRow[]{
+ createRow(null, null, null, null, null),
+ createRow(0, null, null, null, null),
+ createRow(0, 0.0, null, null, null),
+ createRow(0, 0.0, fromString("a"), null, null),
+ createRow(1, 0.0, fromString("a"), null, null),
+ createRow(1, 1.0, fromString("a"), null, null),
+ createRow(1, 1.0, fromString("b"), null, null),
+ createRow(1, 1.0, fromString("b"), createArray(1),
createMap(new int[]{1}, new int[]{1})),
+ createRow(1, 1.0, fromString("b"), createArray(1, 2),
createMap(new int[]{1, 4}, new int[]{1, 2})),
+ createRow(1, 1.0, fromString("b"), createArray(1, 2,
3), createMap(new int[]{1, 5}, new int[]{1, 3})),
+ createRow(1, 1.0, fromString("b"), createArray(1, 2, 3,
4), createMap(new int[]{1, 6}, new int[]{1, 4})),
+ createRow(1, 1.0, fromString("b"), createArray(1, 2, 3,
4, 5), createMap(new int[]{1, 7}, new int[]{1, 5})),
+ createRow(1, 1.0, fromString("b"), createArray(1, 2, 3,
4, 5, 6), createMap(new int[]{1, 8}, new int[]{1, 6}))
+ };
+
+ BaseRowSerializer serializer = typeInfo.createSerializer(new
ExecutionConfig());
+ return new Object[] {serializer, data};
+ }
+
+ //
----------------------------------------------------------------------------------------------
+
+ private static BinaryArray createArray(int... ints) {
+ BinaryArray array = new BinaryArray();
+ BinaryArrayWriter writer = new BinaryArrayWriter(array,
ints.length, 4);
+ for (int i = 0; i < ints.length; i++) {
+ writer.writeInt(i, ints[i]);
+ }
+ writer.complete();
+ return array;
+ }
+
+ private static BinaryMap createMap(int[] keys, int[] values) {
+ return BinaryMap.valueOf(createArray(keys),
createArray(values));
+ }
+
+ private static GenericRow createRow(Object f0, Object f1, Object f2,
Object f3, Object f4) {
+ GenericRow row = new GenericRow(5);
+ row.setField(0, f0);
+ row.setField(1, f1);
+ row.setField(2, f2);
+ row.setField(3, f3);
+ row.setField(4, f4);
+ return row;
+ }
+
+ @Override
+ protected void deepEquals(String message, BaseRow should, BaseRow is) {
+ int arity = should.getArity();
+ assertEquals(message, arity, is.getArity());
+ assertEquals(serializer.baseRowToBinary(should),
serializer.baseRowToBinary(is));
+ }
+
+ private void deepEquals(BaseRow should, BaseRow is) {
+ deepEquals("", should, is);
+ }
+
+ /**
+ * Override testDuplicate, Because it uses Object equals, deserialize
BaseRow to BinaryRow,
+ * which cannot be directly equals.
+ * See {@link BaseRowSerializer#deserialize}.
+ */
+ @Test
+ @Override
+ public void testDuplicate() throws Exception {}
+
+ @Test
+ public void testCopy() {
+ for (BaseRow row : testData) {
+ deepEquals(row, serializer.copy(row));
+ }
+
+ for (BaseRow row : testData) {
+ deepEquals(row, serializer.copy(row, new
GenericRow(row.getArity())));
+ }
+
+ for (BaseRow row : testData) {
+ deepEquals(row,
serializer.copy(serializer.baseRowToBinary(row),
+ new GenericRow(row.getArity())));
+ }
+
+ for (BaseRow row : testData) {
+ deepEquals(row,
serializer.copy(serializer.baseRowToBinary(row)));
+ }
+
+ for (BaseRow row : testData) {
+ deepEquals(row,
serializer.copy(serializer.baseRowToBinary(row),
+ new BinaryRow(row.getArity())));
+ }
+ }
+
+ @Test
+ public void testWrongCopy() {
+ thrown.expect(IllegalArgumentException.class);
+ serializer.copy(new GenericRow(serializer.getArity() + 1));
+ }
+
+ @Test
+ public void testWrongCopyReuse() {
+ thrown.expect(IllegalArgumentException.class);
+ for (BaseRow row : testData) {
+ deepEquals(row, serializer.copy(row, new
GenericRow(row.getArity() + 1)));
+ }
+ }
+}
diff --git
a/flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/typeutils/BinaryArraySerializerTest.java
b/flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/typeutils/BinaryArraySerializerTest.java
new file mode 100644
index 0000000..d1e5b35
--- /dev/null
+++
b/flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/typeutils/BinaryArraySerializerTest.java
@@ -0,0 +1,66 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.typeutils;
+
+import org.apache.flink.api.common.typeutils.SerializerTestBase;
+import org.apache.flink.table.dataformat.BinaryArray;
+import org.apache.flink.table.dataformat.BinaryArrayWriter;
+import org.apache.flink.table.dataformat.BinaryString;
+
+/**
+ * A test for the {@link BinaryArraySerializer}.
+ */
+public class BinaryArraySerializerTest extends SerializerTestBase<BinaryArray>
{
+
+ @Override
+ protected BinaryArraySerializer createSerializer() {
+ return BinaryArraySerializer.INSTANCE;
+ }
+
+ @Override
+ protected int getLength() {
+ return -1;
+ }
+
+ @Override
+ protected Class<BinaryArray> getTypeClass() {
+ return BinaryArray.class;
+ }
+
+ @Override
+ protected BinaryArray[] getTestData() {
+ return new BinaryArray[] {
+ createArray("11"),
+ createArray("11", "haa"),
+ createArray("11", "haa", "ke"),
+ createArray("11", "haa", "ke"),
+ createArray("11", "lele", "haa", "ke"),
+ };
+ }
+
+ static BinaryArray createArray(String... vs) {
+ BinaryArray array = new BinaryArray();
+ BinaryArrayWriter writer = new BinaryArrayWriter(array,
vs.length, 8);
+ for (int i = 0; i < vs.length; i++) {
+ writer.writeString(i, BinaryString.fromString(vs[i]));
+ }
+ writer.complete();
+ return array;
+ }
+}
diff --git
a/flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/typeutils/BinaryMapSerializerTest.java
b/flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/typeutils/BinaryMapSerializerTest.java
new file mode 100644
index 0000000..fe6b4e5
--- /dev/null
+++
b/flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/typeutils/BinaryMapSerializerTest.java
@@ -0,0 +1,66 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.typeutils;
+
+import org.apache.flink.api.common.typeutils.SerializerTestBase;
+import org.apache.flink.table.dataformat.BinaryArray;
+import org.apache.flink.table.dataformat.BinaryArrayWriter;
+import org.apache.flink.table.dataformat.BinaryMap;
+
+/**
+ * A test for the {@link BinaryArraySerializer}.
+ */
+public class BinaryMapSerializerTest extends SerializerTestBase<BinaryMap> {
+
+ @Override
+ protected BinaryMapSerializer createSerializer() {
+ return BinaryMapSerializer.INSTANCE;
+ }
+
+ @Override
+ protected int getLength() {
+ return -1;
+ }
+
+ @Override
+ protected Class<BinaryMap> getTypeClass() {
+ return BinaryMap.class;
+ }
+
+ @Override
+ protected BinaryMap[] getTestData() {
+ return new BinaryMap[] {
+ BinaryMap.valueOf(createArray(1),
BinaryArraySerializerTest.createArray("11")),
+ BinaryMap.valueOf(createArray(1, 2),
BinaryArraySerializerTest.createArray("11", "haa")),
+ BinaryMap.valueOf(createArray(1, 3, 4),
BinaryArraySerializerTest.createArray("11", "haa", "ke")),
+ BinaryMap.valueOf(createArray(1, 4, 2),
BinaryArraySerializerTest.createArray("11", "haa", "ke")),
+ BinaryMap.valueOf(createArray(1, 5, 6, 7),
BinaryArraySerializerTest.createArray("11", "lele", "haa", "ke"))
+ };
+ }
+
+ private static BinaryArray createArray(int... vs) {
+ BinaryArray array = new BinaryArray();
+ BinaryArrayWriter writer = new BinaryArrayWriter(array,
vs.length, 8);
+ for (int i = 0; i < vs.length; i++) {
+ writer.writeInt(i, vs[i]);
+ }
+ writer.complete();
+ return array;
+ }
+}
diff --git
a/flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/typeutils/BinaryRowSerializerTest.java
b/flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/typeutils/BinaryRowSerializerTest.java
new file mode 100644
index 0000000..11f3ea3
--- /dev/null
+++
b/flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/typeutils/BinaryRowSerializerTest.java
@@ -0,0 +1,64 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.typeutils;
+
+import org.apache.flink.api.common.typeutils.SerializerTestBase;
+import org.apache.flink.table.dataformat.BinaryRow;
+import org.apache.flink.table.dataformat.BinaryRowWriter;
+import org.apache.flink.table.dataformat.BinaryString;
+
+/**
+ * A test for the {@link BinaryArraySerializer}.
+ */
+public class BinaryRowSerializerTest extends SerializerTestBase<BinaryRow> {
+
+ @Override
+ protected BinaryRowSerializer createSerializer() {
+ return new BinaryRowSerializer(2);
+ }
+
+ @Override
+ protected int getLength() {
+ return -1;
+ }
+
+ @Override
+ protected Class<BinaryRow> getTypeClass() {
+ return BinaryRow.class;
+ }
+
+ @Override
+ protected BinaryRow[] getTestData() {
+ return new BinaryRow[] {
+ createRow("11", 1),
+ createRow("12", 2),
+ createRow("132", 3),
+ createRow("13", 4)
+ };
+ }
+
+ private static BinaryRow createRow(String f0, int f1) {
+ BinaryRow row = new BinaryRow(2);
+ BinaryRowWriter writer = new BinaryRowWriter(row);
+ writer.writeString(0, BinaryString.fromString(f0));
+ writer.writeInt(1, f1);
+ writer.complete();
+ return row;
+ }
+}
diff --git
a/flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/typeutils/BinaryRowTypeInfoTest.java
b/flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/typeutils/BinaryRowTypeInfoTest.java
new file mode 100644
index 0000000..5aa9b44
--- /dev/null
+++
b/flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/typeutils/BinaryRowTypeInfoTest.java
@@ -0,0 +1,77 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.typeutils;
+
+import org.apache.flink.table.type.InternalType;
+import org.apache.flink.table.type.InternalTypes;
+
+import org.junit.Test;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotEquals;
+
+/**
+ * Test for {@link BaseRowTypeInfo}.
+ */
+public class BinaryRowTypeInfoTest {
+ private static InternalType[] typeList = new InternalType[]{
+ InternalTypes.INT,
+ InternalTypes.STRING
+ };
+
+ @Test(expected = IllegalArgumentException.class)
+ public void testWrongNumberOfFieldNames() {
+ new BaseRowTypeInfo(typeList, new String[]{"int", "string",
"int"});
+ // number of field names should be equal to number of types, go
fail
+ }
+
+ @Test(expected = IllegalArgumentException.class)
+ public void testDuplicateCustomFieldNames() {
+ new BaseRowTypeInfo(typeList, new String[]{"int", "int"});
+ // field names should not be the same, go fail
+ }
+
+ @Test
+ public void testBinaryRowTypeInfoEquality() {
+ BaseRowTypeInfo typeInfo1 = new BaseRowTypeInfo(
+ InternalTypes.INT,
+ InternalTypes.STRING);
+
+ BaseRowTypeInfo typeInfo2 = new BaseRowTypeInfo(
+ InternalTypes.INT,
+ InternalTypes.STRING);
+
+ assertEquals(typeInfo1, typeInfo2);
+ assertEquals(typeInfo1.hashCode(), typeInfo2.hashCode());
+ }
+
+ @Test
+ public void testBinaryRowTypeInfoInequality() {
+ BaseRowTypeInfo typeInfo1 = new BaseRowTypeInfo(
+ InternalTypes.INT,
+ InternalTypes.STRING);
+
+ BaseRowTypeInfo typeInfo2 = new BaseRowTypeInfo(
+ InternalTypes.INT,
+ InternalTypes.BOOLEAN);
+
+ assertNotEquals(typeInfo1, typeInfo2);
+ assertNotEquals(typeInfo1.hashCode(), typeInfo2.hashCode());
+ }
+}
diff --git
a/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/type/ArrayType.java
b/flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/typeutils/BinaryStringSerializerTest.java
similarity index 53%
rename from
flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/type/ArrayType.java
rename to
flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/typeutils/BinaryStringSerializerTest.java
index f825b22..2c87e77 100644
---
a/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/type/ArrayType.java
+++
b/flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/typeutils/BinaryStringSerializerTest.java
@@ -16,46 +16,37 @@
* limitations under the License.
*/
-package org.apache.flink.table.type;
+package org.apache.flink.table.typeutils;
-import org.apache.flink.util.Preconditions;
+import org.apache.flink.api.common.typeutils.SerializerTestBase;
+import org.apache.flink.table.dataformat.BinaryString;
+
+import java.util.Arrays;
/**
- * Type for Array.
+ * A test for the {@link BinaryStringSerializer}.
*/
-public class ArrayType implements InternalType {
-
- private final InternalType elementType;
-
- public ArrayType(InternalType elementType) {
- this.elementType = Preconditions.checkNotNull(elementType);
- }
+public class BinaryStringSerializerTest extends
SerializerTestBase<BinaryString> {
- public InternalType getElementType() {
- return elementType;
+ @Override
+ protected BinaryStringSerializer createSerializer() {
+ return BinaryStringSerializer.INSTANCE;
}
@Override
- public boolean equals(Object o) {
- if (this == o) {
- return true;
- }
- if (o == null || getClass() != o.getClass()) {
- return false;
- }
-
- ArrayType arrayType = (ArrayType) o;
-
- return elementType.equals(arrayType.elementType);
+ protected int getLength() {
+ return -1;
}
@Override
- public int hashCode() {
- return elementType.hashCode();
+ protected Class<BinaryString> getTypeClass() {
+ return BinaryString.class;
}
@Override
- public String toString() {
- return "ArrayType{elementType=" + elementType + '}';
+ protected BinaryString[] getTestData() {
+ return Arrays.stream(
+ new String[] {"a", "", "bcd", "jbmbmner8 jhk hj
\n \t üäßß@µ", "", "non-empty"})
+
.map(BinaryString::fromString).toArray(BinaryString[]::new);
}
}
diff --git
a/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/type/TimeType.java
b/flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/typeutils/BinaryTypeInfoTest.java
similarity index 52%
rename from
flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/type/TimeType.java
rename to
flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/typeutils/BinaryTypeInfoTest.java
index b3cdf5b..a3741ef 100644
---
a/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/type/TimeType.java
+++
b/flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/typeutils/BinaryTypeInfoTest.java
@@ -16,30 +16,25 @@
* limitations under the License.
*/
-package org.apache.flink.table.type;
+package org.apache.flink.table.typeutils;
+
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeutils.TypeInformationTestBase;
+import org.apache.flink.table.type.InternalTypes;
/**
- * Sql time type.
+ * Test for {@link BinaryStringTypeInfo}, {@link BinaryArrayTypeInfo}, {@link
BinaryMapTypeInfo}.
*/
-public class TimeType implements AtomicType {
-
- public static final TimeType INSTANCE = new TimeType();
-
- private TimeType() {}
-
- @Override
- public boolean equals(Object o) {
- return this == o || o != null && getClass() == o.getClass();
- }
+public class BinaryTypeInfoTest extends
TypeInformationTestBase<TypeInformation<?>> {
@Override
- public int hashCode() {
- return getClass().hashCode();
+ protected TypeInformation[] getTestData() {
+ return new TypeInformation[] {
+ BinaryStringTypeInfo.INSTANCE,
+ new BinaryArrayTypeInfo(InternalTypes.INT),
+ new BinaryArrayTypeInfo(InternalTypes.STRING),
+ new BinaryMapTypeInfo(InternalTypes.STRING,
InternalTypes.INT),
+ new BinaryMapTypeInfo(InternalTypes.DOUBLE,
InternalTypes.INT)
+ };
}
-
- @Override
- public String toString() {
- return getClass().getSimpleName();
- }
-
}
