[FLINK-18325] [table-common] fix potential NPE when calling SqlDataTy…

…peSpec#getNullable.
apache · Oct 29, 2020 · 394a442 · 394a442
1 parent 69c3373
commit 394a442
Show file tree

Hide file tree

Showing 9 changed files with 22 additions and 14 deletions.
diff --git a/...link-sql-parser-hive/src/main/java/org/apache/flink/sql/parser/hive/ddl/HiveDDLUtils.java b/...link-sql-parser-hive/src/main/java/org/apache/flink/sql/parser/hive/ddl/HiveDDLUtils.java
@@ -181,7 +181,8 @@ private static SqlDataTypeSpec convertDataTypes(SqlDataTypeSpec typeSpec) throws
 		SqlTypeNameSpec nameSpec = typeSpec.getTypeNameSpec();
 		SqlTypeNameSpec convertedNameSpec = convertDataTypes(nameSpec);
 		if (nameSpec != convertedNameSpec) {
-			typeSpec = new SqlDataTypeSpec(convertedNameSpec, typeSpec.getTimeZone(), typeSpec.getNullable(),
+			boolean nullable = typeSpec.getNullable() == null ? true : typeSpec.getNullable();
+			typeSpec = new SqlDataTypeSpec(convertedNameSpec, typeSpec.getTimeZone(), nullable,
 					typeSpec.getParserPosition());
 		}
 		return typeSpec;

diff --git a/...e/src/main/java/org/apache/flink/sql/parser/hive/type/ExtendedHiveStructTypeNameSpec.java b/...e/src/main/java/org/apache/flink/sql/parser/hive/type/ExtendedHiveStructTypeNameSpec.java
@@ -55,7 +55,7 @@ public void unparse(SqlWriter writer, int leftPrec, int rightPrec) {
 			writer.sep(",", false);
 			p.left.unparse(writer, 0, 0);
 			p.right.unparse(writer, leftPrec, rightPrec);
-			if (!p.right.getNullable()) {
+			if (p.right.getNullable() != null && !p.right.getNullable()) {
 				writer.keyword("NOT NULL");
 			}
 			if (getComments().get(i) != null) {

diff --git a/...-table/flink-sql-parser/src/main/java/org/apache/flink/sql/parser/ddl/SqlTableColumn.java b/...-table/flink-sql-parser/src/main/java/org/apache/flink/sql/parser/ddl/SqlTableColumn.java
@@ -124,7 +124,7 @@ public Optional<SqlTableConstraint> getConstraint() {
 		@Override
 		protected void unparseColumn(SqlWriter writer, int leftPrec, int rightPrec) {
 			type.unparse(writer, leftPrec, rightPrec);
-			if (!type.getNullable()) {
+			if (this.type.getNullable() != null && !this.type.getNullable()) {
 				// Default is nullable.
 				writer.keyword("NOT NULL");
 			}
@@ -179,7 +179,7 @@ public boolean isVirtual() {
 		@Override
 		protected void unparseColumn(SqlWriter writer, int leftPrec, int rightPrec) {
 			type.unparse(writer, leftPrec, rightPrec);
-			if (!type.getNullable()) {
+			if (this.type.getNullable() != null && !this.type.getNullable()) {
 				// Default is nullable.
 				writer.keyword("NOT NULL");
 			}

diff --git a/...sql-parser/src/main/java/org/apache/flink/sql/parser/type/ExtendedSqlRowTypeNameSpec.java b/...sql-parser/src/main/java/org/apache/flink/sql/parser/type/ExtendedSqlRowTypeNameSpec.java
@@ -111,7 +111,7 @@ public void unparse(SqlWriter writer, int leftPrec, int rightPrec) {
 				writer.sep(",", false);
 				p.left.unparse(writer, 0, 0);
 				p.right.unparse(writer, leftPrec, rightPrec);
-				if (!p.right.getNullable()) {
+				if (p.right.getNullable() != null && !p.right.getNullable()) {
 					writer.keyword("NOT NULL");
 				}
 				if (comments.get(i) != null) {

diff --git a/...e/flink-sql-parser/src/main/java/org/apache/flink/sql/parser/type/SqlMapTypeNameSpec.java b/...e/flink-sql-parser/src/main/java/org/apache/flink/sql/parser/type/SqlMapTypeNameSpec.java
@@ -78,13 +78,13 @@ public void unparse(SqlWriter writer, int leftPrec, int rightPrec) {
 		writer.sep(","); // configures the writer
 		keyType.unparse(writer, leftPrec, rightPrec);
 		// Default is nullable.
-		if (!keyType.getNullable()) {
+		if (keyType.getNullable() != null && !keyType.getNullable()) {
 			writer.keyword("NOT NULL");
 		}
 		writer.sep(",");
 		valType.unparse(writer, leftPrec, rightPrec);
 		// Default is nullable.
-		if (!valType.getNullable()) {
+		if (valType.getNullable() != null && !valType.getNullable()) {
 			writer.keyword("NOT NULL");
 		}
 		writer.endList(frame);

diff --git a/...able/flink-sql-parser/src/test/java/org/apache/flink/sql/parser/FlinkDDLDataTypeTest.java b/...able/flink-sql-parser/src/test/java/org/apache/flink/sql/parser/FlinkDDLDataTypeTest.java
@@ -452,7 +452,7 @@ public void checkUnparsed(String sql, String expectedUnparsed) {
 			// SqlDataTypeSpec does not take care of the nullable attribute unparse,
 			// So we unparse nullable attribute specifically, this unparsing logic should
 			// keep sync with SqlTableColumn.
-			if (!dataTypeSpec.getNullable()) {
+			if (dataTypeSpec.getNullable() != null && !dataTypeSpec.getNullable()) {
 				sqlWriter.keyword("NOT NULL");
 			}
 			assertEquals(expectedUnparsed, sqlWriter.toSqlString().getSql());

diff --git a/...ner-blink/src/main/java/org/apache/flink/table/planner/operations/MergeTableLikeUtil.java b/...ner-blink/src/main/java/org/apache/flink/table/planner/operations/MergeTableLikeUtil.java
@@ -43,6 +43,7 @@
 
 import org.apache.calcite.rel.type.RelDataType;
 import org.apache.calcite.rel.type.RelDataTypeField;
+import org.apache.calcite.sql.SqlDataTypeSpec;
 import org.apache.calcite.sql.SqlIdentifier;
 import org.apache.calcite.sql.SqlNode;
 import org.apache.calcite.sql.validate.SqlValidator;
@@ -432,8 +433,9 @@ private void appendDerivedColumns(
 						}
 					}
 
-					final RelDataType relType = metadataColumn.getType()
-						.deriveType(sqlValidator, metadataColumn.getType().getNullable());
+					SqlDataTypeSpec type = metadataColumn.getType();
+					boolean nullable = type.getNullable() == null ? true : type.getNullable();
+					RelDataType relType = type.deriveType(sqlValidator, nullable);
 					column = TableColumn.metadata(
 						name,
 						fromLogicalToDataType(toLogicalType(relType)),
@@ -457,8 +459,9 @@ private void collectPhysicalFieldsTypes(List<SqlNode> derivedColumns) {
 							"A column named '%s' already exists in the base table.",
 							name));
 					}
-					RelDataType relType = regularColumn.getType()
-						.deriveType(sqlValidator, regularColumn.getType().getNullable());
+					SqlDataTypeSpec type = regularColumn.getType();
+					boolean nullable = type.getNullable() == null ? true : type.getNullable();
+					RelDataType relType = type.deriveType(sqlValidator, nullable);
 					// add field name and field type to physical field list
 					physicalFieldNamesToTypes.put(name, relType);
 				}

diff --git a/...ner-blink/src/main/java/org/apache/flink/table/planner/utils/OperationConverterUtils.java b/...ner-blink/src/main/java/org/apache/flink/table/planner/utils/OperationConverterUtils.java
@@ -173,8 +173,9 @@ private static TableColumn toTableColumn(SqlTableColumn tableColumn, SqlValidato
 		SqlRegularColumn regularColumn = (SqlRegularColumn) tableColumn;
 		String name = regularColumn.getName().getSimple();
 		SqlDataTypeSpec typeSpec = regularColumn.getType();
+		boolean nullable = typeSpec.getNullable() == null ? true : typeSpec.getNullable();
 		LogicalType logicalType = FlinkTypeFactory.toLogicalType(
-				typeSpec.deriveType(sqlValidator, typeSpec.getNullable()));
+				typeSpec.deriveType(sqlValidator, nullable));
 		DataType dataType = TypeConversions.fromLogicalToDataType(logicalType);
 		return TableColumn.physical(name, dataType);
 	}

diff --git a/...k-table-planner/src/main/java/org/apache/flink/table/sqlexec/SqlToOperationConverter.java b/...k-table-planner/src/main/java/org/apache/flink/table/sqlexec/SqlToOperationConverter.java
@@ -97,6 +97,7 @@
 
 import org.apache.calcite.rel.RelRoot;
 import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.sql.SqlDataTypeSpec;
 import org.apache.calcite.sql.SqlDialect;
 import org.apache.calcite.sql.SqlExplain;
 import org.apache.calcite.sql.SqlExplainFormat;
@@ -621,10 +622,12 @@ private TableSchema createTableSchema(SqlCreateTable sqlCreateTable) {
 			.map(SqlRegularColumn.class::cast)
 			.collect(Collectors.toList());
 		for (SqlRegularColumn regularColumn : physicalColumns) {
+			SqlDataTypeSpec type = regularColumn.getType();
+			boolean nullable = type.getNullable() == null ? true : type.getNullable();
 			final RelDataType relType = regularColumn.getType()
 				.deriveType(
 					flinkPlanner.getOrCreateSqlValidator(),
-					regularColumn.getType().getNullable());
+						nullable);
 			builder.field(regularColumn.getName().getSimple(),
 				TypeConversions.fromLegacyInfoToDataType(FlinkTypeFactory.toTypeInfo(relType)));
 		}