[flink] branch master updated: [FLINK-12253][table-common] Add a NULL type

[tw...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

twalthr 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 1a703e7  [FLINK-12253][table-common] Add a NULL type
1a703e7 is described below

commit 1a703e7a2d9bdc9b3a084f901ef4fe8f6c271597
Author: Timo Walther <tw...@apache.org>
AuthorDate: Thu May 2 14:25:19 2019 +0200

    [FLINK-12253][table-common] Add a NULL type
---
 .../table/types/logical/LogicalTypeVisitor.java    |  2 +
 .../apache/flink/table/types/logical/NullType.java | 90 ++++++++++++++++++++++
 .../apache/flink/table/types/LogicalTypesTest.java | 22 ++++++
 3 files changed, 114 insertions(+)

diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/logical/LogicalTypeVisitor.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/logical/LogicalTypeVisitor.java
index c574f4b..bab6a96 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/logical/LogicalTypeVisitor.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/logical/LogicalTypeVisitor.java
@@ -79,5 +79,7 @@ public interface LogicalTypeVisitor<R> {
 
 	R visit(StructuredType structuredType);
 
+	R visit(NullType nullType);
+
 	R visit(LogicalType other);
 }
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/logical/NullType.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/logical/NullType.java
new file mode 100644
index 0000000..827fc28
--- /dev/null
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/logical/NullType.java
@@ -0,0 +1,90 @@
+/*
+ * 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.types.logical;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.table.api.TableException;
+
+import java.util.Collections;
+import java.util.List;
+
+/**
+ * Logical type for representing untyped {@code NULL} values. The null type is an extension to the
+ * SQL standard. A null type has no other value except {@code NULL}, thus, it can be cast to any
+ * nullable type similar to JVM semantics.
+ *
+ * <p>This type helps in representing unknown types in API calls that use a {@code NULL} literal as
+ * well as bridging to formats such as JSON or Avro that define such a type as well.
+ *
+ * <p>The serialized string representation is {@code NULL}.
+ */
+@PublicEvolving
+public final class NullType extends LogicalType {
+
+	private static final String DEFAULT_FORMAT = "NULL";
+
+	private static final Class<?> INPUT_CONVERSION = Object.class;
+
+	private static final Class<?> DEFAULT_CONVERSION = Object.class;
+
+	public NullType() {
+		super(true, LogicalTypeRoot.NULL);
+	}
+
+	@Override
+	public LogicalType copy(boolean isNullable) {
+		if (!isNullable) {
+			throw new TableException(
+				"The nullability of a NULL type cannot be disabled because the type must always " +
+					"be able to contain a null value.");
+		}
+		return new NullType();
+	}
+
+	@Override
+	public String asSerializableString() {
+		return DEFAULT_FORMAT;
+	}
+
+	@Override
+	public boolean supportsInputConversion(Class<?> clazz) {
+		return INPUT_CONVERSION.equals(clazz);
+	}
+
+	@Override
+	public boolean supportsOutputConversion(Class<?> clazz) {
+		// any nullable class is supported
+		return !clazz.isPrimitive();
+	}
+
+	@Override
+	public Class<?> getDefaultConversion() {
+		return DEFAULT_CONVERSION;
+	}
+
+	@Override
+	public List<LogicalType> getChildren() {
+		return Collections.emptyList();
+	}
+
+	@Override
+	public <R> R accept(LogicalTypeVisitor<R> visitor) {
+		return visitor.visit(this);
+	}
+}
diff --git a/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/LogicalTypesTest.java b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/LogicalTypesTest.java
index 8d0e169..ec452a8 100644
--- a/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/LogicalTypesTest.java
+++ b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/LogicalTypesTest.java
@@ -34,6 +34,7 @@ import org.apache.flink.table.types.logical.LocalZonedTimestampType;
 import org.apache.flink.table.types.logical.LogicalType;
 import org.apache.flink.table.types.logical.MapType;
 import org.apache.flink.table.types.logical.MultisetType;
+import org.apache.flink.table.types.logical.NullType;
 import org.apache.flink.table.types.logical.RowType;
 import org.apache.flink.table.types.logical.SmallIntType;
 import org.apache.flink.table.types.logical.StructuredType;
@@ -432,6 +433,27 @@ public class LogicalTypesTest {
 		assertFalse(createHumanType(true).supportsInputConversion(User.class));
 	}
 
+	@Test
+	public void testNullType() {
+		final NullType nullType = new NullType();
+
+		testEquality(nullType, new TimeType());
+
+		testJavaSerializability(nullType);
+
+		testStringSerializability(nullType, "NULL");
+
+		testStringSummary(nullType, "NULL");
+
+		assertTrue(nullType.supportsInputConversion(Object.class));
+
+		assertTrue(nullType.supportsOutputConversion(Object.class));
+
+		assertTrue(nullType.supportsOutputConversion(Integer.class));
+
+		assertFalse(nullType.supportsOutputConversion(int.class));
+	}
+
 	// --------------------------------------------------------------------------------------------
 
 	private static void testAll(