@Override
  public Option<IndexedRecord> combineAndGetUpdateValue(IndexedRecord currentValue, Schema schema, Properties properties) throws IOException {
    // Specific to Postgres: If the updated record has TOASTED columns,
    // we will need to keep the previous value for those columns
    // see https://debezium.io/documentation/reference/connectors/postgresql.html#postgresql-toasted-values
    Option<IndexedRecord> insertOrDeleteRecord = super.combineAndGetUpdateValue(currentValue, schema, properties);

    if (insertOrDeleteRecord.isPresent()) {
      mergeToastedValuesIfPresent(insertOrDeleteRecord.get(), currentValue);
    }
    return insertOrDeleteRecord;
  }
@Test
  public void testMergeWithToastedValues() throws IOException {
    Schema avroSchema = SchemaBuilder.builder()
        .record("test_schema")
        .namespace("test_namespace")
        .fields()
        .name(DebeziumConstants.FLATTENED_LSN_COL_NAME).type().longType().noDefault()
        .name("string_col").type().stringType().noDefault()
        .name("byte_col").type().bytesType().noDefault()
        .name("string_null_col_1").type().nullable().stringType().noDefault()
        .name("byte_null_col_1").type().nullable().bytesType().noDefault()
        .name("string_null_col_2").type().nullable().stringType().noDefault()
        .name("byte_null_col_2").type().nullable().bytesType().noDefault()
        .endRecord();

    GenericRecord oldVal = new GenericData.Record(avroSchema);
    oldVal.put(DebeziumConstants.FLATTENED_LSN_COL_NAME, 100L);
    oldVal.put("string_col", "valid string value");
    oldVal.put("byte_col", ByteBuffer.wrap(getUTF8Bytes("valid byte value")));
    oldVal.put("string_null_col_1", "valid string value");
    oldVal.put("byte_null_col_1", ByteBuffer.wrap(getUTF8Bytes("valid byte value")));
    oldVal.put("string_null_col_2", null);
    oldVal.put("byte_null_col_2", null);

    GenericRecord newVal = new GenericData.Record(avroSchema);
    newVal.put(DebeziumConstants.FLATTENED_LSN_COL_NAME, 105L);
    newVal.put("string_col", PostgresDebeziumAvroPayload.DEBEZIUM_TOASTED_VALUE);
    newVal.put("byte_col", ByteBuffer.wrap(getUTF8Bytes(PostgresDebeziumAvroPayload.DEBEZIUM_TOASTED_VALUE)));
    newVal.put("string_null_col_1", null);
    newVal.put("byte_null_col_1", null);
    newVal.put("string_null_col_2", "valid string value");
    newVal.put("byte_null_col_2", ByteBuffer.wrap(getUTF8Bytes("valid byte value")));

    PostgresDebeziumAvroPayload payload = new PostgresDebeziumAvroPayload(Option.of(newVal));

    GenericRecord outputRecord = (GenericRecord) payload
        .combineAndGetUpdateValue(oldVal, avroSchema).get();

    assertEquals("valid string value", outputRecord.get("string_col"));
    assertEquals("valid byte value", fromUTF8Bytes(((ByteBuffer) outputRecord.get("byte_col")).array()));
    assertNull(outputRecord.get("string_null_col_1"));
    assertNull(outputRecord.get("byte_null_col_1"));
    assertEquals("valid string value", ((Utf8) outputRecord.get("string_null_col_2")).toString());
    assertEquals("valid byte value", fromUTF8Bytes(((ByteBuffer) outputRecord.get("byte_null_col_2")).array()));
  }