diff --git a/debezium-connector-mongodb/src/test/java/io/debezium/connector/mongodb/transforms/outbox/MongoEventRouterTest.java b/debezium-connector-mongodb/src/test/java/io/debezium/connector/mongodb/transforms/outbox/MongoEventRouterTest.java
index 0d4fe0b86..0b35583fc 100644
--- a/debezium-connector-mongodb/src/test/java/io/debezium/connector/mongodb/transforms/outbox/MongoEventRouterTest.java
+++ b/debezium-connector-mongodb/src/test/java/io/debezium/connector/mongodb/transforms/outbox/MongoEventRouterTest.java
@@ -222,7 +222,7 @@ public void canSkipUpdates() {
                 .build();
         final Struct before = new Struct(recordSchema);
         before.put("id", "772590bf-ef2d-4814-b4bf-ddc6f5f8b9c5");
-        final Struct payload = envelope.update(before, before, null, Instant.now());
+        final Struct payload = envelope.update(before, before, null, Instant.now(), new String[0]);
         final SourceRecord eventRecord = new SourceRecord(
                 new HashMap<>(),
                 new HashMap<>(),
@@ -253,7 +253,7 @@ public void canFailOnUpdates() {
                 .build();
         final Struct before = new Struct(recordSchema);
         before.put("id", "772590bf-ef2d-4814-b4bf-ddc6f5f8b9c5");
-        final Struct payload = envelope.update(before, before, null, Instant.now());
+        final Struct payload = envelope.update(before, before, null, Instant.now(), new String[0]);
         final SourceRecord eventRecord = new SourceRecord(
                 new HashMap<>(),
                 new HashMap<>(),
diff --git a/debezium-connector-mysql/src/main/java/io/debezium/connector/mysql/legacy/RecordMakers.java b/debezium-connector-mysql/src/main/java/io/debezium/connector/mysql/legacy/RecordMakers.java
index 8a9ea1159..c831c154a 100644
--- a/debezium-connector-mysql/src/main/java/io/debezium/connector/mysql/legacy/RecordMakers.java
+++ b/debezium-connector-mysql/src/main/java/io/debezium/connector/mysql/legacy/RecordMakers.java
@@ -311,7 +311,7 @@ public int update(SourceInfo source, Object[] before, Object[] after, int rowNum
                     else {
                         // The key has not changed, so a simple update is fine ...
                         SourceRecord record = new SourceRecord(partition, getSourceRecordOffset(offset), topicName, partitionNum,
-                                keySchema, newkey, envelope.schema(), envelope.update(valueBefore, valueAfter, origin, ts));
+                                keySchema, newkey, envelope.schema(), envelope.update(valueBefore, valueAfter, origin, ts, new String[0]));
                         consumer.accept(record);
                         ++count;
                     }
diff --git a/debezium-connector-postgres/src/main/java/io/debezium/connector/postgresql/PostgresChangeRecordEmitter.java b/debezium-connector-postgres/src/main/java/io/debezium/connector/postgresql/PostgresChangeRecordEmitter.java
index 0b0e9edab..d274ac120 100644
--- a/debezium-connector-postgres/src/main/java/io/debezium/connector/postgresql/PostgresChangeRecordEmitter.java
+++ b/debezium-connector-postgres/src/main/java/io/debezium/connector/postgresql/PostgresChangeRecordEmitter.java
@@ -156,6 +156,41 @@ private DataCollectionSchema synchronizeTableSchema(DataCollectionSchema tableSc
         return schema.schemaFor(tableId);
     }
 
+    @Override
+    public String[] getColumnsWithUnavailableValues() {
+        List<ReplicationMessage.Column> columns = message.getNewTupleList();
+
+        if (columns == null || columns.isEmpty()) {
+            return new String[0];
+        }
+
+        List<String> toastedCols = new ArrayList<>();
+        final Table table = schema.tableFor(tableId);
+        Objects.requireNonNull(table);
+
+        // based on the schema columns, create the values on the same position as the columns
+        List<Column> schemaColumns = table.columns();
+        // based on the replication message without toasted columns for now
+        List<ReplicationMessage.Column> columnsWithoutToasted = columns.stream().filter(Predicates.not(ReplicationMessage.Column::isToastedColumn))
+                .collect(Collectors.toList());
+        // JSON does not deliver a list of all columns for REPLICA IDENTITY DEFAULT
+        Object[] values = new Object[columnsWithoutToasted.size() < schemaColumns.size() ? schemaColumns.size() : columnsWithoutToasted.size()];
+
+        for (ReplicationMessage.Column column : columns) {
+            final String columnName = Strings.unquoteIdentifierPart(column.getName());
+
+            int position = getPosition(columnName, table, values);
+            if (position != -1) {
+                Object value = column.getValue(() -> (BaseConnection) connection.connection(), connectorConfig.includeUnknownDatatypes());
+                if (value == UnchangedToastedReplicationMessageColumn.UNCHANGED_TOAST_VALUE) {
+                    toastedCols.add(columnName);
+                }
+            }
+        }
+
+        return toastedCols.toArray(new String[0]);
+    }
+
     private Object[] columnValues(List<ReplicationMessage.Column> columns, TableId tableId, boolean refreshSchemaIfChanged, boolean metadataInMessage,
                                   boolean sourceOfToasted, boolean oldValues)
             throws SQLException {
diff --git a/debezium-connector-postgres/src/main/java/io/debezium/connector/postgresql/PostgresValueConverter.java b/debezium-connector-postgres/src/main/java/io/debezium/connector/postgresql/PostgresValueConverter.java
index 4c66c2a66..2ca3c3ff3 100644
--- a/debezium-connector-postgres/src/main/java/io/debezium/connector/postgresql/PostgresValueConverter.java
+++ b/debezium-connector-postgres/src/main/java/io/debezium/connector/postgresql/PostgresValueConverter.java
@@ -978,6 +978,9 @@ else if (data instanceof PgProto.Point) {
 
     protected Object convertArray(Column column, Field fieldDefn, PostgresType elementType, ValueConverter elementConverter, Object data) {
         return convertValue(column, fieldDefn, data, Collections.emptyList(), (r) -> {
+            if (data == UnchangedToastedReplicationMessageColumn.UNCHANGED_TOAST_VALUE) {
+                r.deliver(null);
+            }
             if (data instanceof List) {
                 r.deliver(((List<?>) data).stream()
                         .map(value -> resolveArrayValue(value, elementType))
diff --git a/debezium-core/src/main/java/io/debezium/data/Envelope.java b/debezium-core/src/main/java/io/debezium/data/Envelope.java
index 944813b99..13da8ab31 100644
--- a/debezium-core/src/main/java/io/debezium/data/Envelope.java
+++ b/debezium-core/src/main/java/io/debezium/data/Envelope.java
@@ -6,9 +6,7 @@
 package io.debezium.data;
 
 import java.time.Instant;
-import java.util.Collections;
-import java.util.HashSet;
-import java.util.Set;
+import java.util.*;
 
 import org.apache.kafka.connect.data.Field;
 import org.apache.kafka.connect.data.Schema;
@@ -109,6 +107,8 @@ public String code() {
          * variations.
          */
         public static final String TIMESTAMP = "ts_ms";
+
+        public static final String UNAVAILABLE_COLUMNS = "unavailable_columns";
     }
 
     /**
@@ -129,6 +129,7 @@ public String code() {
         fields.add(FieldName.AFTER);
         fields.add(FieldName.SOURCE);
         fields.add(FieldName.TRANSACTION);
+        fields.add(FieldName.UNAVAILABLE_COLUMNS);
         ALL_FIELD_NAMES = Collections.unmodifiableSet(fields);
     }
 
@@ -225,11 +226,13 @@ public Envelope build() {
                 builder.field(FieldName.OPERATION, OPERATION_REQUIRED ? Schema.STRING_SCHEMA : Schema.OPTIONAL_STRING_SCHEMA);
                 builder.field(FieldName.TIMESTAMP, Schema.OPTIONAL_INT64_SCHEMA);
                 builder.field(FieldName.TRANSACTION, TransactionMonitor.TRANSACTION_BLOCK_SCHEMA);
+                builder.field(FieldName.UNAVAILABLE_COLUMNS, SchemaBuilder.array(Schema.STRING_SCHEMA));
                 checkFieldIsDefined(FieldName.OPERATION);
                 checkFieldIsDefined(FieldName.BEFORE);
                 checkFieldIsDefined(FieldName.AFTER);
                 checkFieldIsDefined(FieldName.SOURCE);
                 checkFieldIsDefined(FieldName.TRANSACTION);
+                checkFieldIsDefined(FieldName.UNAVAILABLE_COLUMNS);
                 if (!missingFields.isEmpty()) {
                     throw new IllegalStateException("The envelope schema is missing field(s) " + String.join(", ", missingFields));
                 }
@@ -314,7 +317,7 @@ public Struct create(Object record, Struct source, Instant timestamp) {
      * @param timestamp the timestamp for this message; may be null
      * @return the update message; never null
      */
-    public Struct update(Object before, Struct after, Struct source, Instant timestamp) {
+    public Struct update(Object before, Struct after, Struct source, Instant timestamp, String[] unavailableCols) {
         Struct struct = new Struct(schema);
         struct.put(FieldName.OPERATION, Operation.UPDATE.code());
         if (before != null) {
@@ -327,6 +330,9 @@ public Struct update(Object before, Struct after, Struct source, Instant timesta
         if (timestamp != null) {
             struct.put(FieldName.TIMESTAMP, timestamp.toEpochMilli());
         }
+        if(unavailableCols != null) {
+            struct.put(FieldName.UNAVAILABLE_COLUMNS, Arrays.asList(unavailableCols));
+        }
         return struct;
     }
 
diff --git a/debezium-core/src/main/java/io/debezium/relational/RelationalChangeRecordEmitter.java b/debezium-core/src/main/java/io/debezium/relational/RelationalChangeRecordEmitter.java
index fc33234dd..f50779a88 100644
--- a/debezium-core/src/main/java/io/debezium/relational/RelationalChangeRecordEmitter.java
+++ b/debezium-core/src/main/java/io/debezium/relational/RelationalChangeRecordEmitter.java
@@ -5,6 +5,7 @@
  */
 package io.debezium.relational;
 
+import java.util.List;
 import java.util.Objects;
 
 import org.apache.kafka.connect.data.Struct;
@@ -109,7 +110,7 @@ protected void emitUpdateRecord(Receiver<P> receiver, TableSchema tableSchema)
         // some configurations does not provide old values in case of updates
         // in this case we handle all updates as regular ones
         if (oldKey == null || Objects.equals(oldKey, newKey)) {
-            Struct envelope = tableSchema.getEnvelopeSchema().update(oldValue, newValue, getOffset().getSourceInfo(), getClock().currentTimeAsInstant());
+            Struct envelope = tableSchema.getEnvelopeSchema().update(oldValue, newValue, getOffset().getSourceInfo(), getClock().currentTimeAsInstant(), getColumnsWithUnavailableValues());
             receiver.changeRecord(getPartition(), tableSchema, Operation.UPDATE, newKey, envelope, getOffset(), null);
         }
         // PK update -> emit as delete and re-insert with new key
@@ -157,6 +158,10 @@ protected void emitTruncateRecord(Receiver<P> receiver, TableSchema schema) thro
      */
     protected abstract Object[] getNewColumnValues();
 
+    protected String[] getColumnsWithUnavailableValues() {
+        return null;
+    }
+
     /**
      * Whether empty data messages should be ignored.
      *
diff --git a/debezium-core/src/test/java/io/debezium/transforms/ExtractNewRecordStateTest.java b/debezium-core/src/test/java/io/debezium/transforms/ExtractNewRecordStateTest.java
index ab7e162b8..a9018ddd7 100644
--- a/debezium-core/src/test/java/io/debezium/transforms/ExtractNewRecordStateTest.java
+++ b/debezium-core/src/test/java/io/debezium/transforms/ExtractNewRecordStateTest.java
@@ -163,7 +163,7 @@ private SourceRecord createUpdateRecord() {
         transaction.put("id", "571");
         transaction.put("total_order", 42L);
         transaction.put("data_collection_order", 42L);
-        final Struct payload = envelope.update(before, after, source, Instant.now());
+        final Struct payload = envelope.update(before, after, source, Instant.now(), new String[0]);
         payload.put("transaction", transaction);
         return new SourceRecord(new HashMap<>(), new HashMap<>(), "dummy", envelope.schema(), payload);
     }
diff --git a/debezium-core/src/test/java/io/debezium/transforms/outbox/EventRouterTest.java b/debezium-core/src/test/java/io/debezium/transforms/outbox/EventRouterTest.java
index 7c0bc808c..abb1b5d58 100644
--- a/debezium-core/src/test/java/io/debezium/transforms/outbox/EventRouterTest.java
+++ b/debezium-core/src/test/java/io/debezium/transforms/outbox/EventRouterTest.java
@@ -219,7 +219,7 @@ public void canSkipUpdates() {
                 .build();
         final Struct before = new Struct(recordSchema);
         before.put("id", "772590bf-ef2d-4814-b4bf-ddc6f5f8b9c5");
-        final Struct payload = envelope.update(before, before, null, Instant.now());
+        final Struct payload = envelope.update(before, before, null, Instant.now(), new String[0]);
         final SourceRecord eventRecord = new SourceRecord(
                 new HashMap<>(),
                 new HashMap<>(),
@@ -251,7 +251,7 @@ public void canFailOnUpdates() {
                 .build();
         final Struct before = new Struct(recordSchema);
         before.put("id", "772590bf-ef2d-4814-b4bf-ddc6f5f8b9c5");
-        final Struct payload = envelope.update(before, before, null, Instant.now());
+        final Struct payload = envelope.update(before, before, null, Instant.now(), new String[0]);
         final SourceRecord eventRecord = new SourceRecord(
                 new HashMap<>(),
                 new HashMap<>(),
