JdbcTypeConverter { public static TypeInformation<?> getFlinkType(int typeIndex) { TypeInformation<?> typeInformation = MAP_INDEX_TO_FLINK_TYPE.get(typeIndex); Preconditions.checkArgument(typeInformation != null, "Unsupported type: %s", typeIndex); return typeInformation; }  static int getIntegerSqlType(TypeInformation<?> type); static TypeInformation<?> getFlinkType(int typeIndex);  }
@Test public void testSpecialStringType() { int[] types = new int[]{ Types.LONGVARCHAR, Types.NULL}; for (int type : types) { TypeInformation<?> flinkType = JdbcTypeConverter.getFlinkType(type); Assert.assertEquals(flinkType, BasicTypeInfo.STRING_TYPE_INFO); } }
@Test(expected = IllegalArgumentException.class) public void testUnsupportedSqlType() { JdbcTypeConverter.getFlinkType(Types.TIME_WITH_TIMEZONE); }