change delete logic

[hameizi]

Previous delete logic is write all delete data in eq-delete file although there is same key in pos-delete file. This PR change this logic to just write the delete data what is not exist in pos-delete file in eq-delete file. And the old logic will send several delete but one insert of one primary key in one snapshot when we want reperform the data by streaming read, in some scene will cause error (like aggregete count).
the following is difference between old delete logic and the new.
table schema:

int key;  (primary key) 
string data;

old logic:
tx1:

insert (1,'aa');

tx2:

delete (1,'aa'); -->eq-delete file add (1,'aa')
insert(1,'bb');
delete (1,'bb'); -->eq-delete file add (1,'bb') pos-delete file add(1,filepath)
insert(1,'cc')

result:
eq-delete file has (1,'aa'),(1,'bb')
pos-delete file has (1,filepath)

new logic:
tx1:

insert (1,'aa');

tx2:

delete (1,'aa'); -->eq-delete file add (1,'aa')
insert(1,'bb');
delete (1,'bb'); --> pos-delete file add(1,filepath)

insert(1,'cc')

result:
eq-delete file has (1,'aa')
pos-delete file has (1,filepath)

Actually the data (1,'bb') is unnecessary in eq-delete file, because when we call function applyPosdelete that (1,'bb') will be delete from result so there is not data match (1,'bb') when we call applyEqdelete. In streaming read scene the old logic will send two delete (1,'aa'),(1,'bb') of key 1 but one insert (1,'cc') to down opeator, but new logic just send one delete and one insert.

[hameizi]

@openinx @rdblue can you help take a look?

[rdblue]

I'm assuming that the equality delete column in your example is the ID. Is that correct?

If so, then there is still some value in writing the delete because you get the previous columns written into the file. That allows reconstructing a CDC stream with the delete.

[hameizi]

I'm assuming that the equality delete column in your example is the ID. Is that correct?

If so, then there is still some value in writing the delete because you get the previous columns written into the file. That allows reconstructing a CDC stream with the delete.

@rdblue But when we reconstructing a CDC stream with the delete we can't guarantee the order of delete data and append data. So we just can guarantee eventual Consistency. This PR can also guarantee eventual Consistency.

[rdblue]

Style: there should be a space between ) and {.

[rdblue]

Why is this changing as well? I think that the logic here was intended to implement upsert just in case there are duplicate inserts without a delete. @openinx, can you take a look at this?

[rdblue]

@hameizi can you reply to this comment?

[hameizi]

@rdblue yes,as you said that the logic here was intended to implement upsert just in case there are duplicate inserts without a delete. This logic can be retain, just i think duplicate inserts should be avoid by upstream semantic but not writer. If need i will rollback this change.

[hameizi]

@rdblue After test i think this is necessary, and i think old logic is error. Because old logic only write pos-delete in write function what should happen in delete function. And below test case is also puzzle:

iceberg/data/src/test/java/org/apache/iceberg/io/TestTaskEqualityDeltaWriter.java

Lines 324 to 348 in 9b6b5e0

	public void testUpsertDataWithFullRowSchema() throws IOException {
	List<Integer> eqDeleteFieldIds = Lists.newArrayList(dataFieldId);
	Schema eqDeleteRowSchema = table.schema();
	GenericTaskDeltaWriter deltaWriter = createTaskWriter(eqDeleteFieldIds, eqDeleteRowSchema);
	deltaWriter.write(createRecord(1, "aaa"));
	deltaWriter.write(createRecord(2, "bbb"));
	deltaWriter.write(createRecord(3, "aaa"));
	deltaWriter.write(createRecord(3, "ccc"));
	deltaWriter.write(createRecord(4, "ccc"));
	// Commit the 1th transaction.
	WriteResult result = deltaWriter.complete();
	Assert.assertEquals("Should have a data file", 1, result.dataFiles().length);
	Assert.assertEquals("Should have a pos-delete file for deduplication purpose", 1, result.deleteFiles().length);
	Assert.assertEquals("Should be pos-delete file", FileContent.POSITION_DELETES, result.deleteFiles()[0].content());
	Assert.assertEquals(1, result.referencedDataFiles().length);
	commitTransaction(result);
	Assert.assertEquals("Should have expected records", expectedRowSet(ImmutableList.of(
	createRecord(2, "bbb"),
	createRecord(3, "aaa"),
	createRecord(4, "ccc")
	)), actualRowSet("*"));

this test write duplicate key return just one line but user should avoid write duplicate key depend on config write.upsert.enable #2863 to execute delete semantic but not write sementic(this is error sementic) when there is duplicate inserts.
So we just need record the postion of key in insertedRowMap when execute write function then write pos-delete file when execute delete function.

[rdblue]

Let's remove this change.

The logic here is to opportunistically catch duplicates when there are only inserts. This is not intended to replace the real upsert logic, which requires calling delete as you noted. Instead, it is here because we're updating the insertedRowMap and may get a previous insert location. When that happens, the right thing to do is to delete the duplicate row.

I also just realized that the changes below are incorrect. Instead of calling internalPosDelete(key), this checks the insertedRowMap itself using get. The logic in internalPosDeleteusedremove` so that the entry was removed before the insert occurred and we have a second opportunistic check.

To fix this, you should instead update internalPosDelete to return true if a row was deleted and false otherwise. Then you can update your previous != null check like this:

    public void delete(T row) throws IOException {
      if (!internalPosDelete(structProjection.wrap(asStructLike(row)))) {
        eqDeleteWriter.write(row);
      }
    }

[rdblue]

Style is incorrect here as well.

[hameizi]

@rdblue code style fix in new commit.

[hameizi]

@rdblue Is there any more question of this PR?

[rdblue]

@hameizi, I'm mainly just waiting for a reply to the comment above. I wanted more information about that change. Thanks!

[rdblue]

These lines don't need to change. Can you revert this?

[hameizi]

@rdblue fix all. Can you help take a look?

[rdblue]

Unnecessary whitespace change, and this is not the correct indentation.

[rdblue]

Thanks, @hameizi! I merged this since there was only a style problem left.

[Xiangakun]

The comment in TestDeltaTaskWriter.testCdcEvents should also be updated since it may confuse the user.