[FLINK-12457][table-planner-blink] Remove char support in blink planner

This closes apache#8379

flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/calcite/FlinkTypeFactory.scala

@@ -85,9 +85,6 @@ class FlinkTypeFactory(typeSystem: RelDataTypeSystem) extends JavaTypeFactoryImp
 
  case InternalTypes.BINARY => createSqlType(VARBINARY)
 
- case InternalTypes.CHAR =>
- throw new TableException("Character type is not supported.")
-
  case decimal: DecimalType =>
  createSqlType(DECIMAL, decimal.precision(), decimal.scale())
 

flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/codegen/CodeGenUtils.scala

@@ -34,7 +34,7 @@ import org.apache.flink.table.util.MurmurHashUtil
 import org.apache.flink.types.Row
 
 import java.lang.reflect.Method
-import java.lang.{Boolean => JBoolean, Byte => JByte, Character => JChar, Double => JDouble, Float => JFloat, Integer => JInt, Long => JLong, Short => JShort}
+import java.lang.{Boolean => JBoolean, Byte => JByte, Double => JDouble, Float => JFloat, Integer => JInt, Long => JLong, Short => JShort}
 import java.util.concurrent.atomic.AtomicInteger
 
 object CodeGenUtils {
@@ -118,7 +118,6 @@ object CodeGenUtils {
  case InternalTypes.FLOAT => "float"
  case InternalTypes.DOUBLE => "double"
  case InternalTypes.BOOLEAN => "boolean"
- case InternalTypes.CHAR => "char"
 
  case InternalTypes.DATE => "int"
  case InternalTypes.TIME => "int"
@@ -138,7 +137,6 @@ object CodeGenUtils {
  case InternalTypes.FLOAT => className[JFloat]
  case InternalTypes.DOUBLE => className[JDouble]
  case InternalTypes.BOOLEAN => className[JBoolean]
- case InternalTypes.CHAR => className[JChar]
 
  case InternalTypes.DATE => boxedTypeTermForType(InternalTypes.INT)
  case InternalTypes.TIME => boxedTypeTermForType(InternalTypes.INT)
@@ -171,7 +169,6 @@ object CodeGenUtils {
  case InternalTypes.DOUBLE => "-1.0d"
  case InternalTypes.BOOLEAN => "false"
  case InternalTypes.STRING => s"$BINARY_STRING.EMPTY_UTF8"
- case InternalTypes.CHAR => "'\\0'"
 
  case _: DateType | InternalTypes.TIME => "-1"
  case _: TimestampType => "-1L"
@@ -195,7 +192,6 @@ object CodeGenUtils {
  case InternalTypes.LONG => s"${className[JLong]}.hashCode($term)"
  case InternalTypes.FLOAT => s"${className[JFloat]}.hashCode($term)"
  case InternalTypes.DOUBLE => s"${className[JDouble]}.hashCode($term)"
- case InternalTypes.CHAR => s"${className[JChar]}.hashCode($term)"
  case InternalTypes.STRING => s"$term.hashCode()"
  case InternalTypes.BINARY => s"${className[MurmurHashUtil]}.hashUnsafeBytes(" +
  s"$term, $BYTE_ARRAY_BASE_OFFSET, $term.length)"
@@ -367,7 +363,6 @@ object CodeGenUtils {
  // primitive types
  case InternalTypes.BOOLEAN => s"$rowTerm.getBoolean($indexTerm)"
  case InternalTypes.BYTE => s"$rowTerm.getByte($indexTerm)"
- case InternalTypes.CHAR => s"$rowTerm.getChar($indexTerm)"
  case InternalTypes.SHORT => s"$rowTerm.getShort($indexTerm)"
  case InternalTypes.INT => s"$rowTerm.getInt($indexTerm)"
  case InternalTypes.LONG => s"$rowTerm.getLong($indexTerm)"
@@ -500,7 +495,6 @@ object CodeGenUtils {
  case InternalTypes.FLOAT => s"$binaryRowTerm.setFloat($index, $fieldValTerm)"
  case InternalTypes.DOUBLE => s"$binaryRowTerm.setDouble($index, $fieldValTerm)"
  case InternalTypes.BOOLEAN => s"$binaryRowTerm.setBoolean($index, $fieldValTerm)"
- case InternalTypes.CHAR => s"$binaryRowTerm.setChar($index, $fieldValTerm)"
  case _: DateType => s"$binaryRowTerm.setInt($index, $fieldValTerm)"
  case InternalTypes.TIME => s"$binaryRowTerm.setInt($index, $fieldValTerm)"
  case _: TimestampType => s"$binaryRowTerm.setLong($index, $fieldValTerm)"
@@ -526,7 +520,6 @@ object CodeGenUtils {
  case InternalTypes.FLOAT => s"$rowTerm.setFloat($indexTerm, $fieldTerm)"
  case InternalTypes.DOUBLE => s"$rowTerm.setDouble($indexTerm, $fieldTerm)"
  case InternalTypes.BOOLEAN => s"$rowTerm.setBoolean($indexTerm, $fieldTerm)"
- case InternalTypes.CHAR => s"$rowTerm.setChar($indexTerm, $fieldTerm)"
  case _: DateType => s"$rowTerm.setInt($indexTerm, $fieldTerm)"
  case InternalTypes.TIME => s"$rowTerm.setInt($indexTerm, $fieldTerm)"
  case _: TimestampType => s"$rowTerm.setLong($indexTerm, $fieldTerm)"
@@ -541,7 +534,6 @@ object CodeGenUtils {
  elementType: InternalType): String = elementType match {
  case InternalTypes.BOOLEAN => s"$arrayTerm.setNullBoolean($index)"
  case InternalTypes.BYTE => s"$arrayTerm.setNullByte($index)"
- case InternalTypes.CHAR => s"$arrayTerm.setNullChar($index)"
  case InternalTypes.SHORT => s"$arrayTerm.setNullShort($index)"
  case InternalTypes.INT => s"$arrayTerm.setNullInt($index)"
  case InternalTypes.LONG => s"$arrayTerm.setNullLong($index)"
@@ -593,7 +585,6 @@ object CodeGenUtils {
  case InternalTypes.STRING => s"$writerTerm.writeString($indexTerm, $fieldValTerm)"
  case d: DecimalType =>
  s"$writerTerm.writeDecimal($indexTerm, $fieldValTerm, ${d.precision()})"
- case InternalTypes.CHAR => s"$writerTerm.writeChar($indexTerm, $fieldValTerm)"
  case _: DateType => s"$writerTerm.writeInt($indexTerm, $fieldValTerm)"
  case InternalTypes.TIME => s"$writerTerm.writeInt($indexTerm, $fieldValTerm)"
  case _: TimestampType => s"$writerTerm.writeLong($indexTerm, $fieldValTerm)"

flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/codegen/calls/FunctionGenerator.scala

@@ -647,11 +647,6 @@ object FunctionGenerator {
  Seq(InternalTypes.STRING, InternalTypes.DOUBLE),
  new PrintCallGen())
 
- addSqlFunction(
- PRINT,
- Seq(InternalTypes.STRING, InternalTypes.CHAR),
- new PrintCallGen())
-
  addSqlFunction(
  PRINT,
  Seq(InternalTypes.STRING, InternalTypes.DATE),
@@ -734,11 +729,6 @@ object FunctionGenerator {
  }
  }
 
- addSqlFunction(
- IF,
- Seq(InternalTypes.BOOLEAN, InternalTypes.CHAR, InternalTypes.CHAR),
- new IfCallGen())
-
  addSqlFunction(
  IF,
  Seq(InternalTypes.BOOLEAN, InternalTypes.DATE, InternalTypes.DATE),
@@ -826,11 +816,6 @@ object FunctionGenerator {
  Seq(InternalTypes.DOUBLE),
  new HashCodeCallGen())
 
- addSqlFunction(
- HASH_CODE,
- Seq(InternalTypes.CHAR),
- new HashCodeCallGen())
-
  addSqlFunction(
  HASH_CODE,
  Seq(InternalTypes.DATE),

flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/codegen/calls/ScalarOperatorGens.scala

@@ -820,10 +820,6 @@ object ScalarOperatorGens {
  terms => s""" "" + ${terms.head}"""
  }
 
- // * -> Character
- case (_, InternalTypes.CHAR) =>
- throw new CodeGenException("Character type not supported.")
-
  // String -> Boolean
  case (InternalTypes.STRING, InternalTypes.BOOLEAN) =>
  generateUnaryOperatorIfNotNull(

flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/codegen/sort/SortCodeGenerator.scala

@@ -408,7 +408,6 @@ class SortCodeGenerator(
  case InternalTypes.FLOAT => "Float"
  case InternalTypes.DOUBLE => "Double"
  case InternalTypes.BOOLEAN => "Boolean"
- case InternalTypes.CHAR => "Char"
  case InternalTypes.STRING => "String"
  case InternalTypes.BINARY => "Binary"
  case _: DecimalType => "Decimal"
@@ -444,7 +443,6 @@ class SortCodeGenerator(
  case InternalTypes.BOOLEAN => 1
  case InternalTypes.BYTE => 1
  case InternalTypes.SHORT => 2
- case InternalTypes.CHAR => 2
  case InternalTypes.INT => 4
  case InternalTypes.FLOAT => 4
  case InternalTypes.DOUBLE => 8

flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/typeutils/TypeCheckUtils.scala

@@ -91,7 +91,6 @@ object TypeCheckUtils {
  | InternalTypes.FLOAT
  | InternalTypes.DOUBLE
  | InternalTypes.BOOLEAN
- | InternalTypes.CHAR
  | _: DateType
  | InternalTypes.TIME
  | _: TimestampType => false

flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/typeutils/TypeCoercion.scala

@@ -86,8 +86,6 @@ object TypeCoercion {
 
  case (_, InternalTypes.STRING) => true
 
- case (_, InternalTypes.CHAR) => false // Character type not supported.
-
  case (InternalTypes.STRING, b) if isNumeric(b) => true
  case (InternalTypes.STRING, InternalTypes.BOOLEAN) => true
  case (InternalTypes.STRING, _: DecimalType) => true

flink-table/flink-table-planner-blink/src/test/java/org/apache/flink/table/codegen/SortCodeGeneratorTest.java

@@ -83,7 +83,6 @@ public class SortCodeGeneratorTest {
  InternalTypes.LONG,
  InternalTypes.FLOAT,
  InternalTypes.DOUBLE,
- InternalTypes.CHAR,
  InternalTypes.STRING,
  new DecimalType(18, 2),
  new DecimalType(38, 18),
@@ -211,8 +210,6 @@ private Object[] generateValues(InternalType type) {
  seeds[i] = rnd.nextFloat() * rnd.nextLong();
  } else if (type.equals(InternalTypes.DOUBLE)) {
  seeds[i] = rnd.nextDouble() * rnd.nextLong();
- } else if (type.equals(InternalTypes.CHAR)) {
- seeds[i] = (char) rnd.nextInt();
  } else if (type.equals(InternalTypes.STRING)) {
  seeds[i] = BinaryString.fromString(RandomStringUtils.random(rnd.nextInt(20)));
  } else if (type instanceof DecimalType) {
@@ -263,8 +260,6 @@ private Object value1(InternalType type, Random rnd) {
  return Float.MIN_VALUE;
  } else if (type.equals(InternalTypes.DOUBLE)) {
  return Double.MIN_VALUE;
- } else if (type.equals(InternalTypes.CHAR)) {
- return '1';
  } else if (type.equals(InternalTypes.STRING)) {
  return BinaryString.fromString("");
  } else if (type instanceof DecimalType) {
@@ -307,8 +302,6 @@ private Object value2(InternalType type, Random rnd) {
  return 0f;
  } else if (type.equals(InternalTypes.DOUBLE)) {
  return 0d;
- } else if (type.equals(InternalTypes.CHAR)) {
- return '0';
  } else if (type.equals(InternalTypes.STRING)) {
  return BinaryString.fromString("0");
  } else if (type instanceof DecimalType) {
@@ -348,8 +341,6 @@ private Object value3(InternalType type, Random rnd) {
  return Float.MAX_VALUE;
  } else if (type.equals(InternalTypes.DOUBLE)) {
  return Double.MAX_VALUE;
- } else if (type.equals(InternalTypes.CHAR)) {
- return '鼎';
  } else if (type.equals(InternalTypes.STRING)) {
  return BinaryString.fromString(RandomStringUtils.random(100));
  } else if (type instanceof DecimalType) {

flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/dataformat/BinaryArray.java

@@ -42,7 +42,6 @@ public final class BinaryArray extends BinaryFormat implements TypeGetterSetters
  private static final int BYTE_ARRAY_BASE_OFFSET = UNSAFE.arrayBaseOffset(byte[].class);
  private static final int BOOLEAN_ARRAY_OFFSET = UNSAFE.arrayBaseOffset(boolean[].class);
  private static final int SHORT_ARRAY_OFFSET = UNSAFE.arrayBaseOffset(short[].class);
- private static final int CHAR_ARRAY_OFFSET = UNSAFE.arrayBaseOffset(char[].class);
  private static final int INT_ARRAY_OFFSET = UNSAFE.arrayBaseOffset(int[].class);
  private static final int LONG_ARRAY_OFFSET = UNSAFE.arrayBaseOffset(long[].class);
  private static final int FLOAT_ARRAY_OFFSET = UNSAFE.arrayBaseOffset(float[].class);
@@ -67,8 +66,6 @@ public static int calculateFixLengthPartSize(InternalType type) {
  return 4;
  } else if (type.equals(InternalTypes.FLOAT)) {
  return 4;
- } else if (type.equals(InternalTypes.CHAR)) {
- return 2;
  } else if (type.equals(InternalTypes.DATE)) {
  return 4;
  } else if (type.equals(InternalTypes.TIME)) {
@@ -325,19 +322,6 @@ public void setNullDouble(int pos) {
  SegmentsUtil.setDouble(segments, getElementOffset(pos, 8), 0.0);
  }
 
- @Override
- public char getChar(int pos) {
- assertIndexIsValid(pos);
- return SegmentsUtil.getChar(segments, getElementOffset(pos, 2));
- }
-
- @Override
- public void setChar(int pos, char value) {
- assertIndexIsValid(pos);
- setNotNullAt(pos);
- SegmentsUtil.setChar(segments, getElementOffset(pos, 2), value);
- }
-
  @Override
  public void setDecimal(int pos, Decimal value, int precision) {
  assertIndexIsValid(pos);
@@ -369,12 +353,6 @@ public void setDecimal(int pos, Decimal value, int precision) {
  }
  }
 
- public void setNullChar(int pos) {
- assertIndexIsValid(pos);
- SegmentsUtil.bitSet(segments, offset + 4, pos);
- SegmentsUtil.setChar(segments, getElementOffset(pos, 2), '\0');
- }
-
  public boolean anyNull() {
  for (int i = offset + 4; i < elementOffset; i += 4) {
  if (SegmentsUtil.getInt(segments, i) != 0) {
@@ -414,14 +392,6 @@ public short[] toShortArray() {
  return values;
  }
 
- public char[] toCharArray() {
- checkNoNull();
- char[] values = new char[numElements];
- SegmentsUtil.copyToUnsafe(
- segments, elementOffset, values, CHAR_ARRAY_OFFSET, numElements * 2);
- return values;
- }
-
  public int[] toIntArray() {
  checkNoNull();
  int[] values = new int[numElements];
@@ -505,10 +475,6 @@ public static BinaryArray fromPrimitiveArray(short[] arr) {
  return fromPrimitiveArray(arr, SHORT_ARRAY_OFFSET, arr.length, 2);
  }
 
- public static BinaryArray fromPrimitiveArray(char[] arr) {
- return fromPrimitiveArray(arr, CHAR_ARRAY_OFFSET, arr.length, 2);
- }
-
  public static BinaryArray fromPrimitiveArray(int[] arr) {
  return fromPrimitiveArray(arr, INT_ARRAY_OFFSET, arr.length, 4);
  }

flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/dataformat/BinaryArrayWriter.java

@@ -129,8 +129,6 @@ public void setNullAt(int pos, InternalType type) {
  setNullInt(pos);
  } else if (type.equals(InternalTypes.TIMESTAMP)) {
  setNullLong(pos);
- } else if (type.equals(InternalTypes.CHAR)) {
- setNullShort(pos);
  } else {
  setNullAt(pos);
  }
@@ -192,11 +190,6 @@ public void writeDouble(int pos, double value) {
  segment.putDouble(getElementOffset(pos, 8), value);
  }
 
- @Override
- public void writeChar(int pos, char value) {
- segment.putChar(getElementOffset(pos, 2), value);
- }
-
  @Override
  public void afterGrow() {
  array.pointTo(segment, 0, segment.size());

flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/dataformat/BinaryRow.java

@@ -73,7 +73,6 @@ public final class BinaryRow extends BinaryFormat implements BaseRow {
  InternalTypes.LONG,
  InternalTypes.FLOAT,
  InternalTypes.DOUBLE,
- InternalTypes.CHAR,
  InternalTypes.TIMESTAMP,
  InternalTypes.DATE,
  InternalTypes.TIME,
@@ -191,13 +190,6 @@ public void setDouble(int pos, double value) {
  segments[0].putDouble(getFieldOffset(pos), value);
  }
 
- @Override
- public void setChar(int pos, char value) {
- assertIndexIsValid(pos);
- setNotNullAt(pos);
- segments[0].putChar(getFieldOffset(pos), value);
- }
-
  @Override
  public void setDecimal(int pos, Decimal value, int precision) {
  assertIndexIsValid(pos);
@@ -299,12 +291,6 @@ public double getDouble(int pos) {
  return segments[0].getDouble(getFieldOffset(pos));
  }
 
- @Override
- public char getChar(int pos) {
- assertIndexIsValid(pos);
- return segments[0].getChar(getFieldOffset(pos));
- }
-
  @Override
  public BinaryString getString(int pos) {
  assertIndexIsValid(pos);

flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/dataformat/BinaryRowWriter.java

@@ -107,11 +107,6 @@ public void writeDouble(int pos, double value) {
  segment.putDouble(getFieldOffset(pos), value);
  }
 
- @Override
- public void writeChar(int pos, char value) {
- segment.putChar(getFieldOffset(pos), value);
- }
-
  @Override
  public void complete() {
  row.setTotalSize(cursor);

flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/dataformat/BinaryWriter.java

@@ -60,8 +60,6 @@ public interface BinaryWriter {
 
  void writeDouble(int pos, double value);
 
- void writeChar(int pos, char value);
-
  void writeString(int pos, BinaryString value);
 
  void writeBinary(int pos, byte[] bytes);
@@ -98,8 +96,6 @@ static void write(BinaryWriter writer, int pos, Object o, InternalType type) {
  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 instanceof DateType) {
  writer.writeInt(pos, (int) o);
  } else if (type.equals(InternalTypes.TIME)) {