private DataType fromJDBCType()

in flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/catalog/PostgresCatalog.java [285:368]
80 lines of code
37 McCabe index (conditional complexity)

	private DataType fromJDBCType(ResultSetMetaData metadata, int colIndex) throws SQLException {
		String pgType = metadata.getColumnTypeName(colIndex);

		int precision = metadata.getPrecision(colIndex);
		int scale = metadata.getScale(colIndex);

		switch (pgType) {
			case PG_BOOLEAN:
				return DataTypes.BOOLEAN();
			case PG_BOOLEAN_ARRAY:
				return DataTypes.ARRAY(DataTypes.BOOLEAN());
			case PG_BYTEA:
				return DataTypes.BYTES();
			case PG_BYTEA_ARRAY:
				return DataTypes.ARRAY(DataTypes.BYTES());
			case PG_SMALLINT:
				return DataTypes.SMALLINT();
			case PG_SMALLINT_ARRAY:
				return DataTypes.ARRAY(DataTypes.SMALLINT());
			case PG_INTEGER:
			case PG_SERIAL:
				return DataTypes.INT();
			case PG_INTEGER_ARRAY:
				return DataTypes.ARRAY(DataTypes.INT());
			case PG_BIGINT:
			case PG_BIGSERIAL:
				return DataTypes.BIGINT();
			case PG_BIGINT_ARRAY:
				return DataTypes.ARRAY(DataTypes.BIGINT());
			case PG_REAL:
				return DataTypes.FLOAT();
			case PG_REAL_ARRAY:
				return DataTypes.ARRAY(DataTypes.FLOAT());
			case PG_DOUBLE_PRECISION:
				return DataTypes.DOUBLE();
			case PG_DOUBLE_PRECISION_ARRAY:
				return DataTypes.ARRAY(DataTypes.DOUBLE());
			case PG_NUMERIC:
				// see SPARK-26538: handle numeric without explicit precision and scale.
				if (precision > 0) {
					return DataTypes.DECIMAL(precision, metadata.getScale(colIndex));
				}
				return DataTypes.DECIMAL(DecimalType.MAX_PRECISION, 18);
			case PG_NUMERIC_ARRAY:
				// see SPARK-26538: handle numeric without explicit precision and scale.
				if (precision > 0) {
					return DataTypes.ARRAY(DataTypes.DECIMAL(precision, metadata.getScale(colIndex)));
				}
				return DataTypes.ARRAY(DataTypes.DECIMAL(DecimalType.MAX_PRECISION, 18));
			case PG_CHAR:
			case PG_CHARACTER:
				return DataTypes.CHAR(precision);
			case PG_CHAR_ARRAY:
			case PG_CHARACTER_ARRAY:
				return DataTypes.ARRAY(DataTypes.CHAR(precision));
			case PG_CHARACTER_VARYING:
				return DataTypes.VARCHAR(precision);
			case PG_CHARACTER_VARYING_ARRAY:
				return DataTypes.ARRAY(DataTypes.VARCHAR(precision));
			case PG_TEXT:
				return DataTypes.STRING();
			case PG_TEXT_ARRAY:
				return DataTypes.ARRAY(DataTypes.STRING());
			case PG_TIMESTAMP:
				return DataTypes.TIMESTAMP(scale);
			case PG_TIMESTAMP_ARRAY:
				return DataTypes.ARRAY(DataTypes.TIMESTAMP(scale));
			case PG_TIMESTAMPTZ:
				return DataTypes.TIMESTAMP_WITH_LOCAL_TIME_ZONE(scale);
			case PG_TIMESTAMPTZ_ARRAY:
				return DataTypes.ARRAY(DataTypes.TIMESTAMP_WITH_LOCAL_TIME_ZONE(scale));
			case PG_TIME:
				return DataTypes.TIME(scale);
			case PG_TIME_ARRAY:
				return DataTypes.ARRAY(DataTypes.TIME(scale));
			case PG_DATE:
				return DataTypes.DATE();
			case PG_DATE_ARRAY:
				return DataTypes.ARRAY(DataTypes.DATE());
			default:
				throw new UnsupportedOperationException(
					String.format("Doesn't support Postgres type '%s' yet", pgType));
		}
	}