Spec: Add file and position delete files

[rdblue]

One option for encoding row-level deletes is to track rows by file and row position in the file. This format should be documented in the spec.

[chenjunjiedada]

What format do we want for the delete files? Avro?

We need to read through the filtered parquet to construct the delete files, right?

[chenjunjiedada]

Just for information and answer questions myself. I did some POC to generate delete files in Parquet format with additional meta columns. The approach is to generate dataframe according to filters and then extend the internal row with metadata. It works fine but looks a bit strange that we spawn spark tasks in a table format, which means it has some write amplification when deleting by filters. But it should be ok if the deletion records are provided by Spark like Hudi. This reminds me that we can add an API like deleteByRowKeys(Iteratable<InternalRows>)

[chenjunjiedada]

Just FYI, I did an early PoC based on file and position deletion files here. Since the file name should be most same so I store it as parquet file for dictionary encoding. I will submit a PR for this soon.

[rdblue]

I think the delete file format (Avro, Parquet, ORC) should be up to the implementation. I would use the table's default format. That is normally Parquet, but ORC users would probably want both data and delete files to be ORC. Similarly, Avro doesn't require additional modules, so if a table sets its format to Avro then it makes sense to also use Avro for deletes.

Looking forward to a read/writer PR for this. Thank you, @chenjunjiedada!

[chenjunjiedada]

Make sense. Thanks!

[chenjunjiedada]

Hi @rdblue , I 'd like to confirm a couple of things to continue:

1. For the write path, before we have spark interfaces and SortOrder, my idea is to build a filtered and sorted data frame and save it to staging table, append the added files and then drop staging table metadata. What do you think?

    dataFrame.select( s"_file", s"_pos").sort(s"_file")
      .write
      .format("iceberg")
      .option(DEFAULT_FILE_FORMAT, format)
      .mode("append")
      .save(tablePath)

2. For the read path, my understanding is we don't want the deletion file filtering logic right now because that depends on the sequence number. So the read path here contains only the logic that extracts the deletion files from scan task and read them as Iterators for mergeInterator. Does this make sense to you?

[rdblue]

@chenjunjiedada, I'm not sure I understand.

This issue tracks updates to the spec that detail how to write file/position delete files. That boils down to things like this:

• The files must be a supported Iceberg format and are valid Iceberg data files, just like all Iceberg metadata files
• The schema of the file/position files; probably string file path and long position in a file. Maybe that should be int instead of long?
• Rows in the delete file must be sorted by position. Should we also sort by path?

We also need a reader and writer implementation for these.

[chenjunjiedada]

I put an early PR for the spec.

It uses integer for the position, that should be enough for most cases unless we have files that larger than 8G (roughly 2^31 rows of integer). But there may exist very special cases like files with most same rows and that could be dictionary and RLE encoded.

For the sorting requirement, I prefer to sort with both path and position(I assume the records already cluster by path, so sort with path should be easy) so that the merge sort could take advantage of it.

For the reader and writer, I will create a separate PR later.

[rdblue]

I don't understand what #885 has to do with the spec. That updates DataFile. The spec is this document that details the Iceberg at-rest format. Maybe you used the wrong branch?

For position, I think we should use a long after all. That's what DataFile uses to encode the number of rows, and I've seen Parquet files with more than 2 billion rows before.

I agree that we should sort by both path and position, but this will require support from the processing engine because the file contents could be large.

[chenjunjiedada]

Sorry for the confusion, I just open #887.

No matter what column it sort with, we all need a processing engine to handle it, don't we? I 'm thinking of the case of the Spark SupportsDelete interface, with the given filter we have to call the spark to build the dataframe for deletion file. But Iceberg may have APIs to support deleting rows in file scope or partition scope in the future. That may not need the processing engine.

[rdblue]

SupportsDelete won't be used to create delete files because Iceberg should not call into Spark's API as you describe. Instead, Spark should attempt the metadata-level delete and fall back to a merge-based delete.

[chenjunjiedada]

Hi @rdblue @jerryshao, I'm working on the reader and writer for delete files. As for the reader, the input and output are clear as the delete files would be extracted from FileScanTask and the output would be iterable of InternalRow or GenericRecord. While for the writer, I 'm not very sure since how metadata-level delete will be implemented. For the spark engine, I could imagine two possible ways:

• spark build the table scan with the filter expression and projection of the partition columns and metadata columns then save the data frame in append mode with an option like data=diff. In this case, Iceberg can generate partition level delete files from the input partition. That means we may not need a separate writer, maybe some update in the current table writer.

• spark has a logical plan RowDeleteScan like the table scan, each task read the input file with filter(residual) and then write to delete file. In this case, we need a separate writer that takes the input file and filter expression as input and writes the filtered records. This would need residual evaluation for every supported format, but we can use IcebergGenerics to handle this.

Both of them depend on spark side implementation, any thought?

[rdblue]

The object model used by these files can be swapped out just like any Avro file. If we want to pass this information to Spark, then we would use the Spark reader classes. For most internals, we build classes that implement IndexedRecord.

[chenjunjiedada]

@rdblue , I don't quite follow the last comment. Do you mean the object model for delete files could be Avro in-memory data model so that it could be used by other engines as well?