Spec: define meta columns for row-level deletes

[chenjunjiedada]

In order to support the row-level delete, we may need to add some meta columns to the table schema, such as file name, row index and key.

[rdblue]

I don't think we need to add columns to schemas, although we can make additional columns available. We've discussed this in the Spark community and the plan was to allow Spark to project additional metadata columns when pruning.

[chenjunjiedada]

Thanks for the explanation.

Do you mean that spark will pass schema with metadata columns and Iceberg doesn't need to add additional columns by itself? I 'm not sure what benefit if spark provides the meta columns, while if Iceberg doesn't have the meta columns, then other engines will need to contain redundant logic like what spark would do, that would bring the complexity for implementing the new readers such as Flink source and hive InputFormat.

BTW, do we have any jira or doc about the discussion? Many thanks if you could provide the link about the context.

[openinx]

@rdblue , from the perspective of flink, the iceberg update/delete feature should not depend on any spark-binded feature I think. Any detail about the spark addition meta columns projection ? I would see whether it work in flink, Thanks.

[rdblue]

@openinx, what I have working in Spark could be implemented in Flink as well. I've just updated Spark to be able to request additional columns, which I've currently named _file and _pos. The columns that can be requested are "metadata columns" so that select * doesn't show them.

The same code used to provide the metadata values to Spark could be used for Flink.

[rdblue]

Here's the Spark PR that adds metadata columns: apache/spark#28027

I just ported that over from our internal branch, where we have it working and can project _file from Iceberg tables. I still need to add unit tests for it.

[jerryshao]

@rdblue I have a question about metadata columns defined in Spark. IIUC, if we want to merge base file and delete file, the file needs to be sorted by metadata column, this should be done by compute framework, shall we require metadata columns to be sorted? If not, how do we merge the base file and delete files?

[rdblue]

Metadata columns are a way to project extra fields on read. That allows us to attach identifying information to each row: the data file and the row's position in that file. There is no need to sort this on read. But when writing out a set of rows to delete, we want Spark to cluster the data by file and order by position, ascending. This is where we would use Anton's proposal to request a sort order from a write exec node. That way, Spark would insert the necessary shuffles and sorts to cluster data by file and locally order by file and position.

Does that make sense?

[jerryshao]

But when writing out a set of rows to delete, we want Spark to cluster the data by file and order by position, ascending. This is where we would use Anton's proposal to request a sort order from a write exec node. That way, Spark would insert the necessary shuffles and sorts to cluster data by file and locally order by file and position.

Yes, that's actually what I asked. Adding metadata columns may not be enough for the readers to merge the base and diff files efficiently. I saw @aokolnychyi 's proposal about this, was thinking that proposal should be related to this thing.

[chenjunjiedada]

I was thinking to abstract the left join logic to handle both sorted case and unsorted case according to the column's sort property. In the case of unsorted one or small deletion files, we could do the hash join.