Delta Lake connector writes incorrect CDC entries when deletion vector is enabled

[ebyhr]

Spark returns the expected results:

CREATE TABLE default.test_delta
(col1 STRING, updated_column INT)
USING DELTA
LOCATION 's3://test-bucket/test_delta'
TBLPROPERTIES ('delta.enableChangeDataFeed'=true, 'delta.enableDeletionVectors'=true);

INSERT INTO default.test_delta VALUES ('testValue1', 1), ('testValue2', 2), ('testValue3', 3);
UPDATE default.test_delta SET updated_column = 30 WHERE col1 = 'testValue3';

SELECT col1, updated_column, _change_type, _commit_version FROM table_changes('default.test_delta', 0);
testValue3	3	update_preimage	2
testValue3	30	update_postimage	2
testValue1	1	insert	1
testValue2	2	insert	1
testValue3	3	insert	1

update_preimage is missing if we update on Trino:

CREATE TABLE default.test_trino
(col1 STRING, updated_column INT)
USING DELTA
LOCATION 's3://test-bucket/test_trino'
TBLPROPERTIES ('delta.enableChangeDataFeed'=true, 'delta.enableDeletionVectors'=true);

INSERT INTO default.test_trino VALUES ('testValue1', 1), ('testValue2', 2), ('testValue3', 3);
UPDATE default.test_trino SET updated_column = 30 WHERE col1 = 'testValue3';

SELECT col1, updated_column, _change_type, _commit_version FROM table_changes('default.test_trino', 0);
testValue3	30	update_postimage	2
testValue1	1	insert	1
testValue2	2	insert	1
testValue3	3	insert	1

test_delta.zip
test_trino.zip

We should disallow write operations when both CDC and DV is enabled, until we fix the issue.

[ebyhr]

@homar Can you take a look at this issue when you have time?