DataFile Serialization for REST Scan Planning

[geruh]

Related to #2775

In order to support, scan planning for the REST catalog. The API returns file scan tasks as JSON, and I need to deserialize them into DataFile and DeleteFile objects. The API returns JSON like this:

  {
      "plan-status": "completed",
      "delete-files": [
          {
              "spec-id": 0,
              "content": "position-deletes",
              "file-path": "s3://bucket/deletes.parquet",
              "file-format": "parquet",
              "partition": ["test"],
              "file-size-in-bytes": 1529,
              "record-count": 1,
              "column-sizes": {"keys": [2147483546], "values": [134]},
              "lower-bounds": {"keys": [2147483546], "values": ["73333A2F..."]},
              ...
          }
      ],
      "file-scan-tasks": [
          {
              "data-file": {
                  "spec-id": 0,
                  "content": "data",
                  "file-path": "s3://bucket/data.parquet",
                  ...
              },
              "delete-file-references": [0],
              "residual-filter": true
          }
      ]
  }

The format is defined in the https://github.com/apache/iceberg/blob/main/open-api/rest-catalog-open-api.yaml#L4337-L4389, and Java parses it via ContentFileParser.java.

Issue

The REST API representation differs from our internal representation:

• Partition is unbound ["test"] instead of a Record
• Maps are {"keys": [...], "values": [...]} instead of {key: value}
• Bounds are primitives (bytes, hex)
• content is position-deletes string instead of enum int

The current state of our python DataFile:

• Extends Record (for Avro compatibility)
• Uses positional array access (_data[pos])
• Constructed via DataFile.from_args() factory
• Tightly coupled to Avro reader/writer via StructProtocol

Our DataFile isn't Pydantic, so we can't just do DataFile.model_validate(json) with validators to handle these conversions. Also, DataFile handles both data files and delete files via the content field. So it's really a content file.

Options

1. Translation layer (RestContentFile)

Create a separate Pydantic model that parses JSON with validators, then converts to DataFile.

Pros:

• Clean separation of concerns
• No risk to Avro code path
• Easy to test independently

Cons:

• Significant code duplication (all fields defined twice)
• Maintenance burden (keep two classes in sync)
• Conversion overhead
• I've prototyped this here and it's quite verbose

Example:

class RestContentFile(IcebergBaseModel):
    # All fields with validators...
    content: str  # Validates and converts to our content enum
    partition: list[Any]  # Unbound values
    
    def to_datafile(self) -> DataFile:
        # Manual conversion logic...

2. Make DataFile Pydantic
Then it could parse JSON directly with pydantic.

The Challenge with this is that DataFile is coupled to Avro for fields, and extends Record. the Avro reader constructs objects with positional args like DataFile(None, None, ...) then fills by index. We'd need to converge here.

3. Manual parsing

Transform raw JSON dict manually and construct DataFile without Pydantic.

Pros:

• No duplication
• Full control over conversion
• Simple and don't need to mess with existing avro functionality

Cons:

• Lose Pydantic's validation benefits

Reccomendation

I'm Leaning towards B, as it would reduce a lot of duplication. However, it seems can't directly extend both Record and BaseModel due to a metaclass conflict.

I'm Leaning towards option 2 since it would reduce a lot of duplication. However, we can't directly extend both Record and BaseModel due to a metaclass conflict, but we can implement the same StructProtocol interface:

  class DataFile(IcebergBaseModel):
      content: DataFileContent = Field(default=DataFileContent.DATA)
      file_path: str = Field(alias="file-path")
      file_format: FileFormat = Field(alias="file-format")
      # fields with validators for pydantic conversion

      # Field order must match DATA_FILE_TYPE for Avro StructProtocol compatibility.
      # The Avro reader/writer accesses fields by position, not name.
      _FIELD_ORDER: ClassVar[tuple[str, ...]] = ("content", "file_path", ...)

      def __new__(cls, *args, **kwargs):
          if args and not kwargs:
              # Positional args from Avro reader and bypass validation
              return cls.model_construct(**dict(zip(cls._FIELD_ORDER, args)))
          return super().__new__(cls)

      # StructProtocol interface
      def __getitem__(self, pos: int):
          return getattr(self, self._FIELD_ORDER[pos])

      def __setitem__(self, pos: int, value):
          setattr(self, self._FIELD_ORDER[pos], value)

      def __len__(self):
          return len(self._FIELD_ORDER)

But ultimately, I wanted to get input before making changes since this touches a core model. Open to suggestions on the approach.

cc: @Fokko @kevinjqliu @HonahX

[Fokko]

Oof, this is a good point @geruh. I would also be leaning towards B. If we want to fit this perfectly into PyIceberg today, then we need another decoder. Right now, we have the BinaryDecorder for Avro, but we could also add a JsonDecoder that takes the REST payload and reads it into the Record. You would need a schema that matches the response, and that would fix it (except the maps :/)

Partition is unbound ["test"] instead of a Record

This should be pretty straightforward and we should be able to convert a string into a record pretty easily.

Maps are {"keys": [...], "values": [...]} instead of {key: value}

I think this one can be a bit more tricky, and a Pydantic pre-validator might be able to solve that (we do similar stuff for the Iceberg Metadata JSON).

Bounds are primitives (bytes, hex)

Do we need these in the first place? This should be evaluated by the catalog. I would also be fine dropping these first, since they are probably pretty big in memory as well.

content is position-deletes string instead of enum int

Same as the maps, a pre-validator should do the trick here.

[kevinjqliu]

Thanks for bringing this up! This looks like a good opportunity for some refactoring to match the scan planning api.

I like option 2. Pydantic is great at serializing/deserializing (and validation!). We can use pydantic as the json decoder. I think it makes sense to create a DataFile class that both avro and json can reuse.

One question about keeping the original Record behavior. Is it needed for DataFile? The DataFile class is created using DataFile.from_args() and accessed via its properties.

[kevinjqliu]

Answering my own question here about Record. Seems like this is from the Avro reader implementation

iceberg-python/pyiceberg/manifest.py

Lines 894 to 900 in abae20f

	with AvroFile[ManifestFile](
	input_file,
	MANIFEST_LIST_FILE_SCHEMAS[DEFAULT_READ_VERSION],
	read_types={-1: ManifestFile, 508: PartitionFieldSummary},
	read_enums={517: ManifestContent},
	) as reader:
	yield from reader

These classes are essentially avro.ManifestFile / avro.DataFile and is directly serialized from the avro reader.
What we want is something more generic, that can be serialized from both avro and json...

Option 2 makes DataFile pydantic (json serializable) while keeping its Record behavior for avro serialization.