TableSchemaBuilder { public TableSchema create(String schemaPrefix, String envelopSchemaName, Table table, ColumnNameFilter filter, ColumnMappers mappers, KeyMapper keysMapper) { if (schemaPrefix == null) { schemaPrefix = ""; } final TableId tableId = table.id(); final String tableIdStr = tableSchemaName(tableId); final String schemaNamePrefix = schemaPrefix + tableIdStr; LOGGER.debug("Mapping table '{}' to schemas under '{}'", tableId, schemaNamePrefix); SchemaBuilder valSchemaBuilder = SchemaBuilder.struct().name(schemaNameAdjuster.adjust(schemaNamePrefix + ".Value")); SchemaBuilder keySchemaBuilder = SchemaBuilder.struct().name(schemaNameAdjuster.adjust(schemaNamePrefix + ".Key")); AtomicBoolean hasPrimaryKey = new AtomicBoolean(false); Key tableKey = new Key.Builder(table).customKeyMapper(keysMapper).build(); tableKey.keyColumns().forEach(column -> { addField(keySchemaBuilder, table, column, null); hasPrimaryKey.set(true); }); table.columns() .stream() .filter(column -> filter == null || filter.matches(tableId.catalog(), tableId.schema(), tableId.table(), column.name())) .forEach(column -> { ColumnMapper mapper = mappers == null ? null : mappers.mapperFor(tableId, column); addField(valSchemaBuilder, table, column, mapper); }); Schema valSchema = valSchemaBuilder.optional().build(); Schema keySchema = hasPrimaryKey.get() ? keySchemaBuilder.build() : null; if (LOGGER.isDebugEnabled()) { LOGGER.debug("Mapped primary key for table '{}' to schema: {}", tableId, SchemaUtil.asDetailedString(keySchema)); LOGGER.debug("Mapped columns for table '{}' to schema: {}", tableId, SchemaUtil.asDetailedString(valSchema)); } Envelope envelope = Envelope.defineSchema() .withName(schemaNameAdjuster.adjust(envelopSchemaName)) .withRecord(valSchema) .withSource(sourceInfoSchema) .build(); StructGenerator keyGenerator = createKeyGenerator(keySchema, tableId, tableKey.keyColumns()); StructGenerator valueGenerator = createValueGenerator(valSchema, tableId, table.columns(), filter, mappers); return new TableSchema(tableId, keySchema, keyGenerator, envelope, valSchema, valueGenerator); } TableSchemaBuilder(ValueConverterProvider valueConverterProvider, SchemaNameAdjuster schemaNameAdjuster, CustomConverterRegistry customConverterRegistry,
                              Schema sourceInfoSchema, boolean sanitizeFieldNames); TableSchema create(String schemaPrefix, String envelopSchemaName, Table table, ColumnNameFilter filter, ColumnMappers mappers, KeyMapper keysMapper);  }
@Test(expected = NullPointerException.class) public void shouldFailToBuildTableSchemaFromNullTable() { new TableSchemaBuilder(new JdbcValueConverters(), adjuster, customConverterRegistry, SchemaBuilder.struct().build(), false) .create(prefix, "sometopic", null, null, null, null); }
@Test public void shouldBuildTableSchemaFromTable() { schema = new TableSchemaBuilder(new JdbcValueConverters(), adjuster, customConverterRegistry, SchemaBuilder.struct().build(), false) .create(prefix, "sometopic", table, null, null, null); assertThat(schema).isNotNull(); }
@Test @FixFor("DBZ-1089") public void shouldBuildCorrectSchemaNames() { schema = new TableSchemaBuilder(new JdbcValueConverters(), adjuster, customConverterRegistry, SchemaBuilder.struct().build(), false) .create(prefix, "sometopic", table, null, null, null); assertThat(schema).isNotNull(); assertThat(schema.keySchema().name()).isEqualTo("schema.table.Key"); assertThat(schema.valueSchema().name()).isEqualTo("schema.table.Value"); table = table.edit() .tableId(new TableId("testDb", null, "testTable")) .create(); schema = new TableSchemaBuilder(new JdbcValueConverters(), adjuster, customConverterRegistry, SchemaBuilder.struct().build(), false) .create(prefix, "sometopic", table, null, null, null); assertThat(schema).isNotNull(); assertThat(schema.keySchema().name()).isEqualTo("testDb.testTable.Key"); assertThat(schema.valueSchema().name()).isEqualTo("testDb.testTable.Value"); table = table.edit() .tableId(new TableId(null, "testSchema", "testTable")) .create(); schema = new TableSchemaBuilder(new JdbcValueConverters(), adjuster, customConverterRegistry, SchemaBuilder.struct().build(), false) .create(prefix, "sometopic", table, null, null, null); assertThat(schema).isNotNull(); assertThat(schema.keySchema().name()).isEqualTo("testSchema.testTable.Key"); assertThat(schema.valueSchema().name()).isEqualTo("testSchema.testTable.Value"); table = table.edit() .tableId(new TableId(null, null, "testTable")) .create(); schema = new TableSchemaBuilder(new JdbcValueConverters(), adjuster, customConverterRegistry, SchemaBuilder.struct().build(), false) .create(prefix, "sometopic", table, null, null, null); assertThat(schema).isNotNull(); assertThat(schema.keySchema().name()).isEqualTo("testTable.Key"); assertThat(schema.valueSchema().name()).isEqualTo("testTable.Value"); }
@Test public void shouldBuildTableSchemaFromTableWithoutPrimaryKey() { table = table.edit().setPrimaryKeyNames().create(); schema = new TableSchemaBuilder(new JdbcValueConverters(), adjuster, customConverterRegistry, SchemaBuilder.struct().build(), false) .create(prefix, "sometopic", table, null, null, null); assertThat(schema).isNotNull(); assertThat(schema.keySchema()).isNull(); assertThat(schema.keyFromColumnData(data)).isNull(); Schema values = schema.valueSchema(); assertThat(values).isNotNull(); assertThat(values.field("C1").name()).isEqualTo("C1"); assertThat(values.field("C1").index()).isEqualTo(0); assertThat(values.field("C1").schema()).isEqualTo(SchemaBuilder.string().build()); assertThat(values.field("C2").name()).isEqualTo("C2"); assertThat(values.field("C2").index()).isEqualTo(1); assertThat(values.field("C2").schema()).isEqualTo(Decimal.builder(3).parameter("connect.decimal.precision", "5").optional().build()); assertThat(values.field("C3").name()).isEqualTo("C3"); assertThat(values.field("C3").index()).isEqualTo(2); assertThat(values.field("C3").schema()).isEqualTo(Date.builder().optional().build()); assertThat(values.field("C4").name()).isEqualTo("C4"); assertThat(values.field("C4").index()).isEqualTo(3); assertThat(values.field("C4").schema()).isEqualTo(SchemaBuilder.int32().optional().build()); assertThat(values.field("C5").index()).isEqualTo(4); assertThat(values.field("C5").schema()).isEqualTo(SchemaBuilder.bytes().build()); assertThat(values.field("C6").index()).isEqualTo(5); assertThat(values.field("C6").schema()).isEqualTo(SchemaBuilder.int16().build()); assertThat(values.field("7C7").name()).isEqualTo("7C7"); assertThat(values.field("7C7").index()).isEqualTo(6); assertThat(values.field("7C7").schema()).isEqualTo(SchemaBuilder.string().build()); assertThat(values.field("C-8").name()).isEqualTo("C-8"); assertThat(values.field("C-8").index()).isEqualTo(7); assertThat(values.field("C-8").schema()).isEqualTo(SchemaBuilder.string().build()); assertThat(values.field(AVRO_UNSUPPORTED_NAME).name()).isEqualTo(AVRO_UNSUPPORTED_NAME); assertThat(values.field(AVRO_UNSUPPORTED_NAME).index()).isEqualTo(8); assertThat(values.field(AVRO_UNSUPPORTED_NAME).schema()).isEqualTo(SchemaBuilder.string().build()); Struct value = schema.valueFromColumnData(data); assertThat(value).isNotNull(); assertThat(value.get("C1")).isEqualTo("c1value"); assertThat(value.get("C2")).isEqualTo(BigDecimal.valueOf(3.142d)); assertThat(value.get("C3")).isEqualTo(11626); assertThat(value.get("C4")).isEqualTo(4); assertThat(value.get("C5")).isEqualTo(ByteBuffer.wrap(new byte[]{ 71, 117, 110, 110, 97, 114 })); assertThat(value.get("C6")).isEqualTo(Short.valueOf((short) 0)); }
@Test @FixFor("DBZ-1044") public void shouldSanitizeFieldNamesAndBuildTableSchemaFromTableWithoutPrimaryKey() { table = table.edit().setPrimaryKeyNames().create(); schema = new TableSchemaBuilder(new JdbcValueConverters(), adjuster, customConverterRegistry, SchemaBuilder.struct().build(), true) .create(prefix, "sometopic", table, null, null, null); assertThat(schema).isNotNull(); assertThat(schema.keySchema()).isNull(); assertThat(schema.keyFromColumnData(data)).isNull(); Schema values = schema.valueSchema(); assertThat(values).isNotNull(); assertThat(values.field("C1").name()).isEqualTo("C1"); assertThat(values.field("C1").index()).isEqualTo(0); assertThat(values.field("C1").schema()).isEqualTo(SchemaBuilder.string().build()); assertThat(values.field("C2").name()).isEqualTo("C2"); assertThat(values.field("C2").index()).isEqualTo(1); assertThat(values.field("C2").schema()).isEqualTo(Decimal.builder(3).parameter("connect.decimal.precision", "5").optional().build()); assertThat(values.field("C3").name()).isEqualTo("C3"); assertThat(values.field("C3").index()).isEqualTo(2); assertThat(values.field("C3").schema()).isEqualTo(Date.builder().optional().build()); assertThat(values.field("C4").name()).isEqualTo("C4"); assertThat(values.field("C4").index()).isEqualTo(3); assertThat(values.field("C4").schema()).isEqualTo(SchemaBuilder.int32().optional().build()); assertThat(values.field("C5").index()).isEqualTo(4); assertThat(values.field("C5").schema()).isEqualTo(SchemaBuilder.bytes().build()); assertThat(values.field("C6").index()).isEqualTo(5); assertThat(values.field("C6").schema()).isEqualTo(SchemaBuilder.int16().build()); assertThat(values.field("7C7")).isNull(); assertThat(values.field("_7C7").name()).isEqualTo("_7C7"); assertThat(values.field("_7C7").index()).isEqualTo(6); assertThat(values.field("_7C7").schema()).isEqualTo(SchemaBuilder.string().build()); assertThat(values.field("C-8")).isNull(); assertThat(values.field("C_8").name()).isEqualTo("C_8"); assertThat(values.field("C_8").index()).isEqualTo(7); assertThat(values.field("C_8").schema()).isEqualTo(SchemaBuilder.string().build()); assertThat(values.field(AVRO_UNSUPPORTED_NAME_CONVERTED).name()).isEqualTo(AVRO_UNSUPPORTED_NAME_CONVERTED); assertThat(values.field(AVRO_UNSUPPORTED_NAME_CONVERTED).index()).isEqualTo(8); assertThat(values.field(AVRO_UNSUPPORTED_NAME_CONVERTED).schema()).isEqualTo(SchemaBuilder.string().build()); Struct value = schema.valueFromColumnData(data); assertThat(value).isNotNull(); assertThat(value.get("C1")).isEqualTo("c1value"); assertThat(value.get("C2")).isEqualTo(BigDecimal.valueOf(3.142d)); assertThat(value.get("C3")).isEqualTo(11626); assertThat(value.get("C4")).isEqualTo(4); assertThat(value.get("C5")).isEqualTo(ByteBuffer.wrap(new byte[]{ 71, 117, 110, 110, 97, 114 })); assertThat(value.get("C6")).isEqualTo(Short.valueOf((short) 0)); }
@Test @FixFor("DBZ-1044") public void shouldSanitizeFieldNamesAndValidateSerialization() { schema = new TableSchemaBuilder(new JdbcValueConverters(), adjuster, customConverterRegistry, SchemaBuilder.struct().build(), true) .create(prefix, "sometopic", table, null, null, null); Struct key = (Struct) schema.keyFromColumnData(keyData); Struct value = schema.valueFromColumnData(data); SourceRecord record = new SourceRecord(Collections.emptyMap(), Collections.emptyMap(), "sometopic", schema.keySchema(), key, schema.valueSchema(), value); VerifyRecord.isValid(record); }
@Test @FixFor("DBZ-1015") public void shouldBuildTableSchemaFromTableWithCustomKey() { table = table.edit().setPrimaryKeyNames().create(); schema = new TableSchemaBuilder(new JdbcValueConverters(), adjuster, customConverterRegistry, SchemaBuilder.struct().build(), false) .create(prefix, "sometopic", table, null, null, CustomKeyMapper.getInstance("(.*).table:C2,C3", null)); assertThat(schema).isNotNull(); Schema keys = schema.keySchema(); assertThat(keys).isNotNull(); assertThat(keys.fields()).hasSize(2); assertThat(keys.field("C2").name()).isEqualTo("C2"); assertThat(keys.field("C3").name()).isEqualTo("C3"); }
@Test @FixFor("DBZ-1015") public void shouldOverrideIdentityKey() { schema = new TableSchemaBuilder(new JdbcValueConverters(), adjuster, customConverterRegistry, SchemaBuilder.struct().build(), false) .create(prefix, "sometopic", table, null, null, CustomKeyMapper.getInstance("(.*).table:C2,C3", null)); assertThat(schema).isNotNull(); Schema keys = schema.keySchema(); assertThat(keys).isNotNull(); assertThat(keys.fields()).hasSize(2); assertThat(keys.field("C1")).isNull(); assertThat(keys.field("C2").name()).isEqualTo("C2"); assertThat(keys.field("C3").name()).isEqualTo("C3"); }
@Test @FixFor("DBZ-1015") public void shouldFallbackToIdentyKeyWhenCustomMapperIsNull() { schema = new TableSchemaBuilder(new JdbcValueConverters(), adjuster, customConverterRegistry, SchemaBuilder.struct().build(), false) .create(prefix, "sometopic", table, null, null, null); assertThat(schema).isNotNull(); Schema keys = schema.keySchema(); assertThat(keys).isNotNull(); assertThat(keys.fields()).hasSize(2); assertThat(keys.field("C1").name()).isEqualTo("C1"); assertThat(keys.field("C2").name()).isEqualTo("C2"); }
@Test @FixFor("DBZ-1015") public void customKeyMapperShouldMapMultipleTables() { TableId id2 = new TableId("catalog", "schema", "table2"); Table table2 = Table.editor() .tableId(id2) .addColumns(Column.editor().name("C1") .type("VARCHAR").jdbcType(Types.VARCHAR).length(10) .optional(false) .create(), Column.editor().name("C2") .type("NUMBER").jdbcType(Types.NUMERIC).length(5).scale(3) .create(), Column.editor().name("C3") .type("DATE").jdbcType(Types.DATE).length(4) .optional(true) .create()) .create(); KeyMapper keyMapper = CustomKeyMapper.getInstance("(.*).table:C2,C3;(.*).table2:C1", null); schema = new TableSchemaBuilder(new JdbcValueConverters(), adjuster, customConverterRegistry, SchemaBuilder.struct().build(), false) .create(prefix, "sometopic", table, null, null, keyMapper); assertThat(schema).isNotNull(); Schema keys = schema.keySchema(); assertThat(keys).isNotNull(); assertThat(keys.fields()).hasSize(2); assertThat(keys.field("C1")).isNull(); assertThat(keys.field("C2").name()).isEqualTo("C2"); assertThat(keys.field("C3").name()).isEqualTo("C3"); TableSchema schema2 = new TableSchemaBuilder(new JdbcValueConverters(), adjuster, customConverterRegistry, SchemaBuilder.struct().build(), false) .create(prefix, "sometopic", table2, null, null, keyMapper); assertThat(schema2).isNotNull(); Schema key2 = schema2.keySchema(); assertThat(key2).isNotNull(); assertThat(key2.fields()).hasSize(1); assertThat(key2.field("C1").name()).isEqualTo("C1"); }
@Test @FixFor("DBZ-1507") public void defaultKeyMapperShouldOrderKeyColumnsBasedOnPrimaryKeyColumnNamesOrder() { TableId id2 = new TableId("catalog", "schema", "info"); Table table2 = Table.editor() .tableId(id2) .addColumns(Column.editor().name("t1ID") .type("INT").jdbcType(Types.INTEGER) .optional(false) .create(), Column.editor().name("t2ID") .type("INT").jdbcType(Types.INTEGER) .optional(false) .create()) .setPrimaryKeyNames("t2ID", "t1ID") .create(); TableSchema schema2 = new TableSchemaBuilder(new JdbcValueConverters(), adjuster, customConverterRegistry, SchemaBuilder.struct().build(), false) .create(prefix, "sometopic", table2, null, null, null); Schema key2 = schema2.keySchema(); assertThat(key2).isNotNull(); assertThat(key2.fields()).hasSize(2); assertThat(key2.fields().get(0).name()).isEqualTo("t2ID"); assertThat(key2.fields().get(1).name()).isEqualTo("t1ID"); }