Add persistent IDs to partition fields

[rdblue]

Partition fields are assigned IDs for when they are stored in manifest files. ID assignment is done in PartitionSpec#partitionType(). That assigns IDs for each field starting at 1000.

This assignment scheme reuses IDs across partition specs. Because a manifest file is written for a single partition spec, this doesn't cause problems when multiple specs exist. But this causes problems in the entries and files metadata tables because the data file partition may have a different schema across manifest files, but reuse IDs.

For example, if part of a table is partitioned by days(ts) and another part is partitioned by hours(ts), both of these will show up in the entries table's partition struct with ID 1000.

A simple solution is to assign partition field IDs starting at 1000 across all table specs and keep the last assigned ID in table metadata. This would ensure that partition tuples will be read correctly in metadata tables when a table has multiple partition specs. In the example above, days(ts) would be assigned ID 1000, and when the second partition spec is added, hours(ts) is assigned ID 1001.

[timmylicheng]

@rdblue Now every manifest only has one partitionSpecId. Do this mean if a table had multiple partition specs, there would be multiple manifest for it?

[rdblue]

Yes. If a table has multiple partition specs, it will probably have multiple manifests, each written with one of those specs.

[timmylicheng]

@rdblue So that for every current() TableMetadata, the spec() should return the one with the last assigned ID instead of the defaultSpecId?

[rdblue]

@timmylicheng, sorry for the confusion. The partition spec ID and the IDs I'm talking about here aren't the same thing. Partition specs are assigned IDs so that we can write manifest files that reference those specs, I think you're talking about those IDs.

The IDs I'm talking about here are schema field IDs that are used to write the record of partition data in the manifest file. Right now, those IDs are assigned each time a manifest file is created, starting at 1000. Instead, we should use persistent IDs for each unique partition field and keep track of the last assigned ID for partition evolution.

[timmylicheng]

@rdblue Ah no worries. I'm using this issue to get familiar with the code case :P.
Thanks for patience to guide thru. I'm looking around for the schema field ID you refer to. At the same time, it would be awesome if you could point me at the class or the member.

[manishmalhotrawork]

@rdblue thanks for the explanation !

Please If I understood it correctly:

We are talking about this constant
which is used here?

So, we can keep PARTITION_DATA_ID_START = 1000 at the table level, as PartitionSpec is instantiated for each new partition of the table.

[rdblue]

@manishmalhotrawork, we need to keep track of IDs that have been assigned to partition fields in a table and reuse them when partition specs change. They should probably continue to start at 1,000.

@timmylicheng, Schema field IDs are the integers passed in when creating struct fields, maps, and lists. See http://iceberg.apache.org/api/#nested-types

[manishmalhotrawork]

@rdblue thanks for the answer.
Curious on the flow and as how fieldId are maintained in PartitionSpec vs Table Schema.

Here PartitionSpec is initialized
And fields are copied to partition spec, which also included original IDs assigned to fields.

and in partitionType method
where new fieldID is assigned ( 1000+) for the partitionFields and returns StructType with those fields.

Not sure if Im missing something.

So, why original fieldId and ids assigned in partitionSpec are different ?
How the partitionSpec returned StructType with IDs as 1000+ is used?

[rdblue]

@manishmalhotrawork, those IDs have different contexts. The source ID in a partition field is the ID of the source data column in the table schema. The ID added by partitionType is the ID in the manifest file schema.

[manishmalhotrawork]

@rdblue thanks.

Let me take a stab on this, may be as you mentioned in description.
Passing initial value for the partitionId to PartitionSpec from TableMetadata. And also as TableMetadata knows how many fields are in partition, so can maintain the nextIDValue as well.
So, that when PartitionSpec.partitionType will be called, it will use the passed value as first ID and then increment.

Also the TableMetadata#updatePartitionSpec should also use nextIDValue to pass to PartitionSpec.

Does modifying/dropping columns also needs to be taken care, I believe not?

yeah, for my understanding, a sample partitioned table manifest file's schema has this entry for partition. Where field-id is 1000+.

        "type" : {
          "type" : "record",
          "name" : "r102",
          "fields" : [ {
            "name" : "order_status",
            "type" : [ "null", "string" ],
            "default" : null,
            "field-id" : 1000
          }, {
            "name" : "ship_priority",
            "type" : [ "null", "int" ],
            "default" : null,
            "field-id" : 1001
          }, {
            "name" : "order_key_bucket",
            "type" : [ "null", "int" ],
            "default" : null,
            "field-id" : 1002
          } ]```

[rdblue]

And also as TableMetadata knows how many fields are in partition, so can maintain the nextIDValue as well.

The next partition field ID is the highest field ID in all of the table's partition specs +1. Once a partition spec is removed, we can reuse the ID. Alternatively, we can keep track of the last assigned ID, like we do for the table schema.

Also the TableMetadata#updatePartitionSpec should also use nextIDValue to pass to PartitionSpec.

I think the spec's IDs will be assigned by the time that method is called because the partition spec passed in is already created.

Does modifying/dropping columns also needs to be taken care, I believe not?

No. updateSchema already checks compatibility: https://github.com/apache/incubator-iceberg/blob/master/core/src/main/java/org/apache/iceberg/TableMetadata.java#L284

[manishmalhotrawork]

@rdblue thanks.

Update:

I did changes in local where PartitionSpec has builder.property for initPartitionFieldId which will be 1000 for new partitionSpec. And partitionSpec will also maintain the lastPartitionFieldId.
When new PartitionSpec is added the first field's partitionFieldId will be last added partitionSpec.lastPartitionFieldId + 1. Irrespective of the field name.

wondering how the cases when the new partitionSpec will have the old column as well, like TestMergeAppend#testChangedPartitionSpec
It has 1st partitonSpec with data and then adding data, id. This test-case uses Snapshot/pending updates to verify.

So, as per our logic after adding 1st PartitonSpec, Manifest will have this schema:

        "name" : "partition",
        "type" : {
          "type" : "record",
          "name" : "r102",
          "fields" : [ {
            "name" : "data_bucket",
            "type" : [ "null", "int" ],
            "default" : null,
            "field-id" : 1000
          }

and after adding 2nd PartitionSpec, new manifest will have this schema

        "name" : "partition",
        "type" : {
          "type" : "record",
          "name" : "r102",
          "fields" : [ {
            "name" : "data_bucket",
            "type" : [ "null", "int" ],
            "default" : null,
            "field-id" : 1001
          }, {
            "name" : "id_bucket",
            "type" : [ "null", "int" ],
            "default" : null,
            "field-id" : 1002
          }

But problem is it fails when it reads 2nd manifest file in the test-case to validate the manifest files.

org.apache.avro.AvroRuntimeException: Duplicate field id_bucket in record org.apache.iceberg.PartitionData: id_bucket type:UNION pos:1 and id_bucket type:UNION pos:0.

Interestingly it fails with id_bucket as dupes, which is added only in the 2nd spec.
Will debug further why this is a problem.

[rdblue]

@manishmalhotrawork, one strange thing about your test is that data_bucket has a different ID. It should continue to use id 1000 because it hasn't changed. Either the assignment logic or the evolution logic (like how SchemaUpdate works) should detect that the column has not changed and not assign a different ID.

Without seeing more of what's happening, I'm not really able to tell why you're getting that error. Can you open a WIP pull request?

[manishmalhotrawork]

@rdblue yeah, will raise WIP.

one strange thing about your test is that data_bucket has a different ID. It should continue to use id 1000 because it hasn't changed. Either the assignment logic or the evolution logic (like how SchemaUpdate works) should detect that the column has not changed and not assign a different ID.

yeah thats what I was wondering, the assignment logic (which I added) doesn't check if the column has repeated. And can see updatePartitionSpec also doesn't check it. It uses the same specId only if all the fields are same PartitionSpec#compatibleWith otherwise just add a new partitionSpec into specs.