SchemaUtil { static Schema toBeamSchema(ResultSetMetaData md) throws SQLException { Schema.Builder schemaBuilder = Schema.builder(); for (int i = 1; i <= md.getColumnCount(); i++) { JDBCType jdbcType = valueOf(md.getColumnType(i)); BeamFieldConverter fieldConverter = jdbcTypeToBeamFieldConverter(jdbcType); schemaBuilder.addField(fieldConverter.create(i, md)); } return schemaBuilder.build(); }  static boolean compareSchemaField(Schema.Field a, Schema.Field b);  }
@Test public void testToBeamSchema() throws SQLException { ResultSetMetaData mockResultSetMetaData = mock(ResultSetMetaData.class); ImmutableList<JdbcFieldInfo> fieldInfo = ImmutableList.of( JdbcFieldInfo.of("int_array_col", Types.ARRAY, JDBCType.INTEGER.getName(), false), JdbcFieldInfo.of("bigint_col", Types.BIGINT), JdbcFieldInfo.of("binary_col", Types.BINARY, 255), JdbcFieldInfo.of("bit_col", Types.BIT), JdbcFieldInfo.of("boolean_col", Types.BOOLEAN), JdbcFieldInfo.of("char_col", Types.CHAR, 255), JdbcFieldInfo.of("date_col", Types.DATE), JdbcFieldInfo.of("decimal_col", Types.DECIMAL), JdbcFieldInfo.of("double_col", Types.DOUBLE), JdbcFieldInfo.of("float_col", Types.FLOAT), JdbcFieldInfo.of("integer_col", Types.INTEGER), JdbcFieldInfo.of("longnvarchar_col", Types.LONGNVARCHAR, 1024), JdbcFieldInfo.of("longvarchar_col", Types.LONGVARCHAR, 1024), JdbcFieldInfo.of("longvarbinary_col", Types.LONGVARBINARY, 1024), JdbcFieldInfo.of("nchar_col", Types.NCHAR, 255), JdbcFieldInfo.of("numeric_col", Types.NUMERIC, 12, 4), JdbcFieldInfo.of("nvarchar_col", Types.NVARCHAR, 255), JdbcFieldInfo.of("real_col", Types.REAL), JdbcFieldInfo.of("smallint_col", Types.SMALLINT), JdbcFieldInfo.of("time_col", Types.TIME), JdbcFieldInfo.of("timestamp_col", Types.TIMESTAMP), JdbcFieldInfo.of("timestamptz_col", Types.TIMESTAMP_WITH_TIMEZONE), JdbcFieldInfo.of("tinyint_col", Types.TINYINT), JdbcFieldInfo.of("varbinary_col", Types.VARBINARY, 255), JdbcFieldInfo.of("varchar_col", Types.VARCHAR, 255)); when(mockResultSetMetaData.getColumnCount()).thenReturn(fieldInfo.size()); for (int i = 0; i < fieldInfo.size(); i++) { JdbcFieldInfo f = fieldInfo.get(i); when(mockResultSetMetaData.getColumnLabel(eq(i + 1))).thenReturn(f.columnLabel); when(mockResultSetMetaData.getColumnType(eq(i + 1))).thenReturn(f.columnType); when(mockResultSetMetaData.getColumnTypeName(eq(i + 1))).thenReturn(f.columnTypeName); when(mockResultSetMetaData.getPrecision(eq(i + 1))).thenReturn(f.precision); when(mockResultSetMetaData.getScale(eq(i + 1))).thenReturn(f.scale); when(mockResultSetMetaData.isNullable(eq(i + 1))) .thenReturn( f.nullable ? ResultSetMetaData.columnNullable : ResultSetMetaData.columnNoNulls); } Schema wantBeamSchema = Schema.builder() .addArrayField("int_array_col", Schema.FieldType.INT32) .addField("bigint_col", Schema.FieldType.INT64) .addField("binary_col", LogicalTypes.fixedLengthBytes(JDBCType.BINARY, 255)) .addField("bit_col", LogicalTypes.JDBC_BIT_TYPE) .addField("boolean_col", Schema.FieldType.BOOLEAN) .addField("char_col", LogicalTypes.fixedLengthString(JDBCType.CHAR, 255)) .addField("date_col", LogicalTypes.JDBC_DATE_TYPE) .addField("decimal_col", Schema.FieldType.DECIMAL) .addField("double_col", Schema.FieldType.DOUBLE) .addField("float_col", LogicalTypes.JDBC_FLOAT_TYPE) .addField("integer_col", Schema.FieldType.INT32) .addField( "longnvarchar_col", LogicalTypes.variableLengthString(JDBCType.LONGNVARCHAR, 1024)) .addField( "longvarchar_col", LogicalTypes.variableLengthString(JDBCType.LONGVARCHAR, 1024)) .addField( "longvarbinary_col", LogicalTypes.variableLengthBytes(JDBCType.LONGVARBINARY, 1024)) .addField("nchar_col", LogicalTypes.fixedLengthString(JDBCType.NCHAR, 255)) .addField("numeric_col", LogicalTypes.numeric(12, 4)) .addField("nvarchar_col", LogicalTypes.variableLengthString(JDBCType.NVARCHAR, 255)) .addField("real_col", Schema.FieldType.FLOAT) .addField("smallint_col", Schema.FieldType.INT16) .addField("time_col", LogicalTypes.JDBC_TIME_TYPE) .addField("timestamp_col", Schema.FieldType.DATETIME) .addField("timestamptz_col", LogicalTypes.JDBC_TIMESTAMP_WITH_TIMEZONE_TYPE) .addField("tinyint_col", Schema.FieldType.BYTE) .addField("varbinary_col", LogicalTypes.variableLengthBytes(JDBCType.VARBINARY, 255)) .addField("varchar_col", LogicalTypes.variableLengthString(JDBCType.VARCHAR, 255)) .build(); Schema haveBeamSchema = SchemaUtil.toBeamSchema(mockResultSetMetaData); assertEquals(wantBeamSchema, haveBeamSchema); }