[cdc] source postgres: add optional to schema fields

airbytehq · May 19, 2022 · 3fbe5d4 · 3fbe5d4
1 parent 1edc0de
commit 3fbe5d4
Showing 1 changed file with 3 additions and 3 deletions.
diff --git a/.../debezium/src/main/java/io/airbyte/integrations/debezium/internals/PostgresConverter.java b/.../debezium/src/main/java/io/airbyte/integrations/debezium/internals/PostgresConverter.java
@@ -42,7 +42,7 @@ public void converterFor(final RelationalColumn field, final ConverterRegistrati
   }
 
   private void registerText(final RelationalColumn field, final ConverterRegistration<SchemaBuilder> registration) {
-    registration.register(SchemaBuilder.string(), x -> {
+    registration.register(SchemaBuilder.string().optional(), x -> {
       if (x == null) {
         return DebeziumConverterUtils.convertDefaultValue(field);
       }
@@ -56,7 +56,7 @@ private void registerText(final RelationalColumn field, final ConverterRegistrat
   }
 
   private void registerDate(final RelationalColumn field, final ConverterRegistration<SchemaBuilder> registration) {
-    registration.register(SchemaBuilder.string(), x -> {
+    registration.register(SchemaBuilder.string().optional(), x -> {
       if (x == null) {
         return DebeziumConverterUtils.convertDefaultValue(field);
       } else if (x instanceof PGInterval) {
@@ -78,7 +78,7 @@ private String convertInterval(final PGInterval pgInterval) {
   }
 
   private void registerMoney(final RelationalColumn field, final ConverterRegistration<SchemaBuilder> registration) {
-    registration.register(SchemaBuilder.string(), x -> {
+    registration.register(SchemaBuilder.string().optional(), x -> {
       if (x == null) {
         return DebeziumConverterUtils.convertDefaultValue(field);
       } else if (x instanceof Double) {