@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 testMergeWithDeleteUsingEmptyRecord() throws IOException {
    // empty record being merged with current record.
    HoodieRecord<PostgresDebeziumAvroPayload> emptyRecord = new HoodieAvroRecord(new HoodieKey(), new PostgresDebeziumAvroPayload(Option.empty()));
    GenericRecord existingRecord = createRecord(2, Operation.UPDATE, 99L);
    Option<IndexedRecord> mergedRecord = emptyRecord.getData().combineAndGetUpdateValue(existingRecord, avroSchema, new TypedProperties());
    // expect nothing to be committed to table
    assertFalse(mergedRecord.isPresent());
    // Insert record being merged with empty record.
    GenericRecord insertedRecord = createRecord(1, Operation.INSERT, 100L);
    PostgresDebeziumAvroPayload insertPayload = new PostgresDebeziumAvroPayload(insertedRecord, 100L);
    PostgresDebeziumAvroPayload combinedPayload = (PostgresDebeziumAvroPayload) insertPayload.preCombine(emptyRecord.getData(), avroSchema, new TypedProperties());
    assertEquals(insertPayload, combinedPayload);
  }