Skip to content

Commit 3cbde92

Browse files
zilderzilder
committed
Table destination mapper
1 parent 297bb11 commit 3cbde92

8 files changed

Lines changed: 135 additions & 16 deletions

File tree

‎debezium-server-iceberg-sink/src/main/java/io/debezium/server/iceberg/IcebergChangeConsumer.java‎

Lines changed: 8 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -15,6 +15,7 @@
1515
import io.debezium.server.iceberg.converter.EventConverter;
1616
import io.debezium.server.iceberg.converter.JsonEventConverter;
1717
import io.debezium.server.iceberg.converter.StructEventConverter;
18+
import io.debezium.server.iceberg.mapper.IcebergTableMapper;
1819
import io.debezium.server.iceberg.tableoperator.IcebergTableOperator;
1920
import io.debezium.util.Clock;
2021
import io.debezium.util.Strings;
@@ -30,7 +31,6 @@
3031
import org.apache.iceberg.Schema;
3132
import org.apache.iceberg.Table;
3233
import org.apache.iceberg.catalog.Catalog;
33-
import org.apache.iceberg.catalog.Namespace;
3434
import org.apache.iceberg.catalog.TableIdentifier;
3535
import org.slf4j.Logger;
3636
import org.slf4j.LoggerFactory;
@@ -69,6 +69,11 @@ public class IcebergChangeConsumer implements DebeziumEngine.ChangeConsumer<Embe
6969
@Inject
7070
GlobalConfig config;
7171

72+
@Inject
73+
@Any
74+
Instance<IcebergTableMapper> mappers;
75+
IcebergTableMapper mapper;
76+
7277
@PostConstruct
7378
void connect() {
7479
JsonEventConverter.initializeJsonSerde();
@@ -79,6 +84,7 @@ void connect() {
7984
icebergCatalog = CatalogUtil.buildIcebergCatalog(config.iceberg().catalogName(), config.iceberg().icebergConfigs(), hadoopConf);
8085
batchSizeWait = IcebergUtil.selectInstance(batchSizeWaitInstances, config.batch().batchSizeWaitName());
8186
batchSizeWait.initizalize();
87+
mapper = IcebergUtil.selectInstance(mappers, config.iceberg().mapper());
8288
}
8389

8490
@Override
@@ -162,16 +168,6 @@ protected void logConsumerProgress(long numUploadedEvents) {
162168
}
163169

164170
public TableIdentifier mapDestination(String destination) {
165-
final String tableName = destination
166-
.replaceAll(config.iceberg().destinationRegexp().orElse(""), config.iceberg().destinationRegexpReplace().orElse(""))
167-
.replace(".", "_");
168-
169-
if (config.iceberg().destinationUppercaseTableNames()) {
170-
return TableIdentifier.of(Namespace.of(config.iceberg().namespace()), (config.iceberg().tablePrefix().orElse("") + tableName).toUpperCase());
171-
} else if (config.iceberg().destinationLowercaseTableNames()) {
172-
return TableIdentifier.of(Namespace.of(config.iceberg().namespace()), (config.iceberg().tablePrefix().orElse("") + tableName).toLowerCase());
173-
} else {
174-
return TableIdentifier.of(Namespace.of(config.iceberg().namespace()), config.iceberg().tablePrefix().orElse("") + tableName);
175-
}
171+
return mapper.mapDestination(destination);
176172
}
177173
}

‎debezium-server-iceberg-sink/src/main/java/io/debezium/server/iceberg/IcebergConfig.java‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -40,6 +40,10 @@ public interface IcebergConfig {
4040
@WithName("debezium.sink.iceberg." + CatalogProperties.WAREHOUSE_LOCATION)
4141
String warehouseLocation();
4242

43+
@WithName("debezium.sink.iceberg.mapper")
44+
@WithDefault("default-mapper")
45+
String mapper();
46+
4347
@WithName("debezium.sink.iceberg.destination-regexp")
4448
// @WithDefault("")
4549
Optional<String> destinationRegexp();

‎debezium-server-iceberg-sink/src/main/java/io/debezium/server/iceberg/IcebergUtil.java‎

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -69,15 +69,15 @@ public static Map<String, String> getConfigSubset(Config config, String prefix)
6969
}
7070

7171
public static <T> T selectInstance(Instance<T> instances, String name) {
72-
7372
Instance<T> instance = instances.select(NamedLiteral.of(name));
73+
String className = instance.getClass().getName();
7474
if (instance.isAmbiguous()) {
75-
throw new DebeziumException("Multiple batch size wait class named '" + name + "' were found");
75+
throw new DebeziumException("Multiple '" + className + "' class instances named '" + name + "' were found");
7676
} else if (instance.isUnsatisfied()) {
77-
throw new DebeziumException("No batch size wait class named '" + name + "' is available");
77+
throw new DebeziumException("No '" + className + "' class instance named '" + name + "' is available");
7878
}
7979

80-
LOGGER.info("Using {}", instance.getClass().getName());
80+
LOGGER.info("Using {}", className);
8181
return instance.get();
8282
}
8383

‎debezium-server-iceberg-sink/src/main/java/io/debezium/server/iceberg/mapper/DefaultIcebergTableMapper.java‎

Lines changed: 30 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,30 @@
1+
package io.debezium.server.iceberg.mapper;
2+
3+
import io.debezium.server.iceberg.GlobalConfig;
4+
import jakarta.enterprise.context.Dependent;
5+
import jakarta.inject.Inject;
6+
import jakarta.inject.Named;
7+
import org.apache.iceberg.catalog.Namespace;
8+
import org.apache.iceberg.catalog.TableIdentifier;
9+
10+
@Named("default-mapper")
11+
@Dependent
12+
public class DefaultIcebergTableMapper implements IcebergTableMapper {
13+
@Inject
14+
GlobalConfig config;
15+
16+
@Override
17+
public TableIdentifier mapDestination(String destination) {
18+
final String tableName = destination
19+
.replaceAll(config.iceberg().destinationRegexp().orElse(""), config.iceberg().destinationRegexpReplace().orElse(""))
20+
.replace(".", "_");
21+
22+
if (config.iceberg().destinationUppercaseTableNames()) {
23+
return TableIdentifier.of(Namespace.of(config.iceberg().namespace()), (config.iceberg().tablePrefix().orElse("") + tableName).toUpperCase());
24+
} else if (config.iceberg().destinationLowercaseTableNames()) {
25+
return TableIdentifier.of(Namespace.of(config.iceberg().namespace()), (config.iceberg().tablePrefix().orElse("") + tableName).toLowerCase());
26+
} else {
27+
return TableIdentifier.of(Namespace.of(config.iceberg().namespace()), config.iceberg().tablePrefix().orElse("") + tableName);
28+
}
29+
}
30+
}

‎debezium-server-iceberg-sink/src/main/java/io/debezium/server/iceberg/mapper/IcebergTableMapper.java‎

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,7 @@
1+
package io.debezium.server.iceberg.mapper;
2+
3+
import org.apache.iceberg.catalog.TableIdentifier;
4+
5+
public interface IcebergTableMapper {
6+
TableIdentifier mapDestination(String destination);
7+
}

‎debezium-server-iceberg-sink/src/test/java/io/debezium/server/iceberg/TestConfigSource.java‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -56,6 +56,7 @@ public TestConfigSource() {
5656
config.put("debezium.sink.iceberg.table-prefix", "debeziumcdc_");
5757
config.put("debezium.sink.iceberg.table-namespace", ICEBERG_CATALOG_TABLE_NAMESPACE);
5858
config.put("debezium.sink.iceberg.catalog-name", ICEBERG_CATALOG_NAME);
59+
5960
// drop tombstones for delete events
6061
config.put("debezium.source.tombstones.on.delete", "false");
6162

‎debezium-server-iceberg-sink/src/test/java/io/debezium/server/iceberg/mapper/CustomMapper.java‎

Lines changed: 22 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,22 @@
1+
package io.debezium.server.iceberg.mapper;
2+
3+
import io.debezium.server.iceberg.GlobalConfig;
4+
import jakarta.enterprise.context.Dependent;
5+
import jakarta.inject.Inject;
6+
import jakarta.inject.Named;
7+
import org.apache.iceberg.catalog.Namespace;
8+
import org.apache.iceberg.catalog.TableIdentifier;
9+
10+
@Named("custom-mapper")
11+
@Dependent
12+
public class CustomMapper implements IcebergTableMapper {
13+
@Inject
14+
GlobalConfig config;
15+
16+
@Override
17+
public TableIdentifier mapDestination(String destination) {
18+
String[] parts = destination.split("\\.");
19+
String tableName = parts[2];
20+
return TableIdentifier.of(Namespace.of(config.iceberg().namespace()), "CUSTOM_MAPPER_" + tableName);
21+
}
22+
}

‎debezium-server-iceberg-sink/src/test/java/io/debezium/server/iceberg/mapper/CustomMapperTest.java‎

Lines changed: 59 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,59 @@
1+
package io.debezium.server.iceberg.mapper;
2+
3+
import io.debezium.server.iceberg.BaseSparkTest;
4+
import io.debezium.server.iceberg.testresources.CatalogJdbc;
5+
import io.debezium.server.iceberg.testresources.S3Minio;
6+
import io.debezium.server.iceberg.testresources.SourcePostgresqlDB;
7+
import io.quarkus.test.common.QuarkusTestResource;
8+
import io.quarkus.test.junit.QuarkusTest;
9+
import io.quarkus.test.junit.QuarkusTestProfile;
10+
import io.quarkus.test.junit.TestProfile;
11+
import org.awaitility.Awaitility;
12+
import org.junit.jupiter.api.Assertions;
13+
import org.junit.jupiter.api.Test;
14+
15+
import java.time.Duration;
16+
import java.util.HashMap;
17+
import java.util.Map;
18+
19+
import static io.debezium.server.iceberg.TestConfigSource.ICEBERG_CATALOG_TABLE_NAMESPACE;
20+
import static org.junit.jupiter.api.Assertions.assertEquals;
21+
22+
@QuarkusTest
23+
@QuarkusTestResource(value = S3Minio.class, restrictToAnnotatedClass = true)
24+
@QuarkusTestResource(value = SourcePostgresqlDB.class, restrictToAnnotatedClass = true)
25+
@QuarkusTestResource(value = CatalogJdbc.class, restrictToAnnotatedClass = true)
26+
@TestProfile(CustomMapperTest.TestProfile.class)
27+
public class CustomMapperTest extends BaseSparkTest {
28+
29+
@Test
30+
public void testCustomMapper() throws Exception {
31+
assertEquals(sinkType, "iceberg");
32+
String sql = """
33+
DROP TABLE IF EXISTS inventory.sample;
34+
CREATE TABLE IF NOT EXISTS inventory.sample (id INTEGER, val INTEGER);
35+
""";
36+
SourcePostgresqlDB.runSQL(sql);
37+
SourcePostgresqlDB.runSQL("INSERT INTO inventory.sample (id, val) VALUES (1, 123)");
38+
Awaitility.await().atMost(Duration.ofSeconds(320)).until(() -> {
39+
try {
40+
var df = spark.newSession().table(ICEBERG_CATALOG_TABLE_NAMESPACE + ".CUSTOM_MAPPER_sample");
41+
Assertions.assertEquals(1, df.count());
42+
43+
return true;
44+
} catch (Exception e) {
45+
e.printStackTrace();
46+
return false;
47+
}
48+
});
49+
}
50+
51+
public static class TestProfile implements QuarkusTestProfile {
52+
@Override
53+
public Map<String, String> getConfigOverrides() {
54+
Map<String, String> config = new HashMap<>();
55+
config.put("debezium.sink.iceberg.mapper", "custom-mapper");
56+
return config;
57+
}
58+
}
59+
}

0 commit comments

Comments
 (0)