@Test public void testDateConversion() throws IOException, SchemaConversionException, SQLException { final String db = "db"; final String table = "users"; Map<String, JdbcType> dateColums = new HashMap<>(); dateColums.put("date_of_birth", JdbcType.DATE); dateColums.put("last_modified", JdbcType.TIME); dateColums.put("created", JdbcType.TIMESTAMP); JdbcWriterCommands mockWriterCommands = mock(JdbcWriterCommands.class); when(mockWriterCommands.retrieveDateColumns(db, table)).thenReturn(dateColums); JdbcWriterCommandsFactory factory = mock(JdbcWriterCommandsFactory.class); when(factory.newInstance(any(State.class), any(Connection.class))).thenReturn(mockWriterCommands); List<JdbcEntryMetaDatum> jdbcEntryMetaData = new ArrayList<>(); jdbcEntryMetaData.add(new JdbcEntryMetaDatum("name", JdbcType.VARCHAR)); jdbcEntryMetaData.add(new JdbcEntryMetaDatum("favorite_number", JdbcType.VARCHAR)); jdbcEntryMetaData.add(new JdbcEntryMetaDatum("favorite_color", JdbcType.VARCHAR)); jdbcEntryMetaData.add(new JdbcEntryMetaDatum("date_of_birth", JdbcType.DATE)); jdbcEntryMetaData.add(new JdbcEntryMetaDatum("last_modified", JdbcType.TIME)); jdbcEntryMetaData.add(new JdbcEntryMetaDatum("created", JdbcType.TIMESTAMP)); JdbcEntrySchema expected = new JdbcEntrySchema(jdbcEntryMetaData); Schema inputSchema = new Schema.Parser().parse(getClass().getResourceAsStream("/converter/fieldPickInput.avsc")); WorkUnitState workUnitState = new WorkUnitState(); workUnitState.appendToListProp(JdbcPublisher.JDBC_PUBLISHER_FINAL_TABLE_NAME, table); AvroToJdbcEntryConverter converter = new AvroToJdbcEntryConverter(workUnitState); Map<String, JdbcType> dateColumnMapping = Maps.newHashMap(); dateColumnMapping.put("date_of_birth", JdbcType.DATE); dateColumnMapping.put("last_modified", JdbcType.TIME); dateColumnMapping.put("created", JdbcType.TIMESTAMP); workUnitState.appendToListProp(AvroToJdbcEntryConverter.CONVERTER_AVRO_JDBC_DATE_FIELDS, new Gson().toJson(dateColumnMapping)); JdbcEntrySchema actual = converter.convertSchema(inputSchema, workUnitState); Assert.assertEquals(expected, actual); }
@Override public JdbcEntrySchema convertSchema(Schema inputSchema, WorkUnitState workUnit) throws SchemaConversionException { LOG.info("Converting schema " + inputSchema); Preconditions.checkArgument(Type.RECORD.equals(inputSchema.getType()), "%s is expected for the first level element in Avro schema %s", Type.RECORD, inputSchema); Map<String, Type> avroColumnType = flatten(inputSchema); String jsonStr = Preconditions.checkNotNull(workUnit.getProp(CONVERTER_AVRO_JDBC_DATE_FIELDS)); java.lang.reflect.Type typeOfMap = new TypeToken<Map<String, JdbcType>>() {}.getType(); Map<String, JdbcType> dateColumnMapping = new Gson().fromJson(jsonStr, typeOfMap); LOG.info("Date column mapping: " + dateColumnMapping); List<JdbcEntryMetaDatum> jdbcEntryMetaData = Lists.newArrayList(); for (Map.Entry<String, Type> avroEntry : avroColumnType.entrySet()) { String colName = tryConvertAvroColNameToJdbcColName(avroEntry.getKey()); JdbcType JdbcType = dateColumnMapping.get(colName); if (JdbcType == null) { JdbcType = AVRO_TYPE_JDBC_TYPE_MAPPING.get(avroEntry.getValue()); } Preconditions.checkNotNull(JdbcType, "Failed to convert " + avroEntry + " AVRO_TYPE_JDBC_TYPE_MAPPING: " + AVRO_TYPE_JDBC_TYPE_MAPPING + " , dateColumnMapping: " + dateColumnMapping); jdbcEntryMetaData.add(new JdbcEntryMetaDatum(colName, JdbcType)); } JdbcEntrySchema converted = new JdbcEntrySchema(jdbcEntryMetaData); LOG.info("Converted schema into " + converted); return converted; }
AvroToJdbcEntryConverter extends Converter<Schema, JdbcEntrySchema, GenericRecord, JdbcEntryData> { @Override public JdbcEntrySchema convertSchema(Schema inputSchema, WorkUnitState workUnit) throws SchemaConversionException { LOG.info("Converting schema " + inputSchema); Preconditions.checkArgument(Type.RECORD.equals(inputSchema.getType()), "%s is expected for the first level element in Avro schema %s", Type.RECORD, inputSchema); Map<String, Type> avroColumnType = flatten(inputSchema); String jsonStr = Preconditions.checkNotNull(workUnit.getProp(CONVERTER_AVRO_JDBC_DATE_FIELDS)); java.lang.reflect.Type typeOfMap = new TypeToken<Map<String, JdbcType>>() {}.getType(); Map<String, JdbcType> dateColumnMapping = new Gson().fromJson(jsonStr, typeOfMap); LOG.info("Date column mapping: " + dateColumnMapping); List<JdbcEntryMetaDatum> jdbcEntryMetaData = Lists.newArrayList(); for (Map.Entry<String, Type> avroEntry : avroColumnType.entrySet()) { String colName = tryConvertAvroColNameToJdbcColName(avroEntry.getKey()); JdbcType JdbcType = dateColumnMapping.get(colName); if (JdbcType == null) { JdbcType = AVRO_TYPE_JDBC_TYPE_MAPPING.get(avroEntry.getValue()); } Preconditions.checkNotNull(JdbcType, "Failed to convert " + avroEntry + " AVRO_TYPE_JDBC_TYPE_MAPPING: " + AVRO_TYPE_JDBC_TYPE_MAPPING + " , dateColumnMapping: " + dateColumnMapping); jdbcEntryMetaData.add(new JdbcEntryMetaDatum(colName, JdbcType)); } JdbcEntrySchema converted = new JdbcEntrySchema(jdbcEntryMetaData); LOG.info("Converted schema into " + converted); return converted; } }
AvroToJdbcEntryConverter extends Converter<Schema, JdbcEntrySchema, GenericRecord, JdbcEntryData> { @Override public JdbcEntrySchema convertSchema(Schema inputSchema, WorkUnitState workUnit) throws SchemaConversionException { LOG.info("Converting schema " + inputSchema); Preconditions.checkArgument(Type.RECORD.equals(inputSchema.getType()), "%s is expected for the first level element in Avro schema %s", Type.RECORD, inputSchema); Map<String, Type> avroColumnType = flatten(inputSchema); String jsonStr = Preconditions.checkNotNull(workUnit.getProp(CONVERTER_AVRO_JDBC_DATE_FIELDS)); java.lang.reflect.Type typeOfMap = new TypeToken<Map<String, JdbcType>>() {}.getType(); Map<String, JdbcType> dateColumnMapping = new Gson().fromJson(jsonStr, typeOfMap); LOG.info("Date column mapping: " + dateColumnMapping); List<JdbcEntryMetaDatum> jdbcEntryMetaData = Lists.newArrayList(); for (Map.Entry<String, Type> avroEntry : avroColumnType.entrySet()) { String colName = tryConvertAvroColNameToJdbcColName(avroEntry.getKey()); JdbcType JdbcType = dateColumnMapping.get(colName); if (JdbcType == null) { JdbcType = AVRO_TYPE_JDBC_TYPE_MAPPING.get(avroEntry.getValue()); } Preconditions.checkNotNull(JdbcType, "Failed to convert " + avroEntry + " AVRO_TYPE_JDBC_TYPE_MAPPING: " + AVRO_TYPE_JDBC_TYPE_MAPPING + " , dateColumnMapping: " + dateColumnMapping); jdbcEntryMetaData.add(new JdbcEntryMetaDatum(colName, JdbcType)); } JdbcEntrySchema converted = new JdbcEntrySchema(jdbcEntryMetaData); LOG.info("Converted schema into " + converted); return converted; } AvroToJdbcEntryConverter(); @VisibleForTesting  AvroToJdbcEntryConverter(WorkUnitState workUnit); }
AvroToJdbcEntryConverter extends Converter<Schema, JdbcEntrySchema, GenericRecord, JdbcEntryData> { @Override public JdbcEntrySchema convertSchema(Schema inputSchema, WorkUnitState workUnit) throws SchemaConversionException { LOG.info("Converting schema " + inputSchema); Preconditions.checkArgument(Type.RECORD.equals(inputSchema.getType()), "%s is expected for the first level element in Avro schema %s", Type.RECORD, inputSchema); Map<String, Type> avroColumnType = flatten(inputSchema); String jsonStr = Preconditions.checkNotNull(workUnit.getProp(CONVERTER_AVRO_JDBC_DATE_FIELDS)); java.lang.reflect.Type typeOfMap = new TypeToken<Map<String, JdbcType>>() {}.getType(); Map<String, JdbcType> dateColumnMapping = new Gson().fromJson(jsonStr, typeOfMap); LOG.info("Date column mapping: " + dateColumnMapping); List<JdbcEntryMetaDatum> jdbcEntryMetaData = Lists.newArrayList(); for (Map.Entry<String, Type> avroEntry : avroColumnType.entrySet()) { String colName = tryConvertAvroColNameToJdbcColName(avroEntry.getKey()); JdbcType JdbcType = dateColumnMapping.get(colName); if (JdbcType == null) { JdbcType = AVRO_TYPE_JDBC_TYPE_MAPPING.get(avroEntry.getValue()); } Preconditions.checkNotNull(JdbcType, "Failed to convert " + avroEntry + " AVRO_TYPE_JDBC_TYPE_MAPPING: " + AVRO_TYPE_JDBC_TYPE_MAPPING + " , dateColumnMapping: " + dateColumnMapping); jdbcEntryMetaData.add(new JdbcEntryMetaDatum(colName, JdbcType)); } JdbcEntrySchema converted = new JdbcEntrySchema(jdbcEntryMetaData); LOG.info("Converted schema into " + converted); return converted; } AvroToJdbcEntryConverter(); @VisibleForTesting  AvroToJdbcEntryConverter(WorkUnitState workUnit); @Override Converter<Schema, JdbcEntrySchema, GenericRecord, JdbcEntryData> init(WorkUnitState workUnit); @Override JdbcEntrySchema convertSchema(Schema inputSchema, WorkUnitState workUnit); @Override Iterable<JdbcEntryData> convertRecord(JdbcEntrySchema outputSchema, GenericRecord record,
      WorkUnitState workUnit); @Override ConverterInitializer getInitializer(State state, WorkUnitStream workUnits, int branches, int branchId); }
AvroToJdbcEntryConverter extends Converter<Schema, JdbcEntrySchema, GenericRecord, JdbcEntryData> { @Override public JdbcEntrySchema convertSchema(Schema inputSchema, WorkUnitState workUnit) throws SchemaConversionException { LOG.info("Converting schema " + inputSchema); Preconditions.checkArgument(Type.RECORD.equals(inputSchema.getType()), "%s is expected for the first level element in Avro schema %s", Type.RECORD, inputSchema); Map<String, Type> avroColumnType = flatten(inputSchema); String jsonStr = Preconditions.checkNotNull(workUnit.getProp(CONVERTER_AVRO_JDBC_DATE_FIELDS)); java.lang.reflect.Type typeOfMap = new TypeToken<Map<String, JdbcType>>() {}.getType(); Map<String, JdbcType> dateColumnMapping = new Gson().fromJson(jsonStr, typeOfMap); LOG.info("Date column mapping: " + dateColumnMapping); List<JdbcEntryMetaDatum> jdbcEntryMetaData = Lists.newArrayList(); for (Map.Entry<String, Type> avroEntry : avroColumnType.entrySet()) { String colName = tryConvertAvroColNameToJdbcColName(avroEntry.getKey()); JdbcType JdbcType = dateColumnMapping.get(colName); if (JdbcType == null) { JdbcType = AVRO_TYPE_JDBC_TYPE_MAPPING.get(avroEntry.getValue()); } Preconditions.checkNotNull(JdbcType, "Failed to convert " + avroEntry + " AVRO_TYPE_JDBC_TYPE_MAPPING: " + AVRO_TYPE_JDBC_TYPE_MAPPING + " , dateColumnMapping: " + dateColumnMapping); jdbcEntryMetaData.add(new JdbcEntryMetaDatum(colName, JdbcType)); } JdbcEntrySchema converted = new JdbcEntrySchema(jdbcEntryMetaData); LOG.info("Converted schema into " + converted); return converted; } AvroToJdbcEntryConverter(); @VisibleForTesting  AvroToJdbcEntryConverter(WorkUnitState workUnit); @Override Converter<Schema, JdbcEntrySchema, GenericRecord, JdbcEntryData> init(WorkUnitState workUnit); @Override JdbcEntrySchema convertSchema(Schema inputSchema, WorkUnitState workUnit); @Override Iterable<JdbcEntryData> convertRecord(JdbcEntrySchema outputSchema, GenericRecord record,
      WorkUnitState workUnit); @Override ConverterInitializer getInitializer(State state, WorkUnitStream workUnits, int branches, int branchId); static final String CONVERTER_AVRO_JDBC_DATE_FIELDS; }