Add a serializer for FileScanTask

[stevenzwu]

For batch/bounded mode, Java Serializable works well as there is no concern of schema evolution. If we are going to support the streaming read with long-running jobs, we need to consider schema evolution for checkpoint state. Otherwise, change in the code might break the Java serialization and ability to restore from checkpoint.

Here are some high-level thoughts.

• Move most of the schema defined in DataFile to parent interface ContentFile. Extend the schema with the additional fields in DataFile and DeleteFile.
• Add a schema to FileScanTask where ResidualEvaluator and PartitionSpec fields will be defined as string type.
• CombinedScanTask schema is straightforward. it should be just a collection of FileScanTask
• Add ScanTasks util class in iceberg-core that handles the serialization and deserialization of FileScanTask and CombinedScanTask

One challenge is how to plugin custom field serializers for ResidualEvaluator and PartitionSpec.

Overall, this seems like a large change. not sure if there is a simpler way.

[stevenzwu]

@openinx @JingsongLi @rdblue any comment?

[openinx]

Thanks @stevenzwu for bringing this up. It's indeed an problem for flink streaming reader because it depends on the java serialization in StreamingReaderOperator now, it's easy to crash when we upgrade iceberg lib version (which changes the CombinedScanTask classes) and restart the flink job.

I'd prefer to define the avro schema in BaseCombinedScanTask which is similar to BaseDataFile, then we could maintain the binary bytes which is serialized by avro approach to flink state backend. Let me evaluate the work.

[coolderli]

@stevenzwu @openinx Is there a patch to fix this problem ？

[stevenzwu]

@rdblue @pvary I updated the description with some high-level thoughts on how this can potentially achieved. Can you please share your thoughts?

[rdblue]

@stevenzwu, I think that we should introduce a JSON format and parser for these tasks. The information in a FileScanTask has been stable for a really long time so it wouldn't be a problem to maintain it. And we've had other projects ask for this before as well, since it is more common to use JSON to serialize in some settings. Trino uses JSON for RPC and we've also recently discussed adding job planning to the REST catalog interface.

[stevenzwu]

@rdblue JSON would work, although it is less efficient in terms of space and serialization. But I see the benefit that it can be useful in some other scenarios. I can look into that direction.

[stevenzwu]

@aokolnychyi would also like to get your input. With the recent changelog scan, we may also need to document the JSON format for those changelog scan tasks in the future. not needed right now especially as we are still iterating on those interfaces.

[aokolnychyi]

I am also +1 on trying to come up with a reasonable JSON representation. Handling job planning via the REST catalog is something I'd be interested to see.

[stevenzwu]

Anton, thanks a lot for the input. Looks like we have a direction moving forward.