Add persistent IDs to partition fields (WIP) #499

manishmalhotrawork · 2019-09-26T23:01:03Z

@rdblue can you please review.
Raising as WIP PR, as this might need some changes.

Summary: parsing manifest-schema json to find out the partition-field_ids and initializing PartitionSpec based on last ID (if available ) other 1000.

manishmalhotrawork · 2019-09-27T17:27:32Z

build is failed because of iceberg-spark:test Hive related.
Is there a way to find detail logs of the failed tests.
All tests are failed with this error.
Caused by: org.apache.thrift.transport.TTransportException at TestIcebergSourceHiveTables.java:152

So, possibly of intermittent error, because Metastore didnt start.
And we ran just re-run the PR build.

In local all tests ran fine.

manishmalhotrawork · 2019-09-27T21:24:18Z

finally got the good PR build.

manishmalhotrawork · 2019-10-03T22:38:07Z

@rdblue can you please check this one, thanks !

api/src/main/java/org/apache/iceberg/PartitionSpec.java

rdblue · 2019-10-09T19:13:19Z

core/src/main/java/org/apache/iceberg/PartitionSpecParser.java

+  private static final String FIELD_ID = "field-id";
+
+  private PartitionSpecParser() {
+  }


Please remove any non-functional changes, like this one that moves the private constructor.

rdblue · 2019-10-09T19:13:54Z

core/src/main/java/org/apache/iceberg/PartitionSpecParser.java

    return builder.build();
  }

+


Please avoid adding extra newlines. These can cause avoidable commit conflicts.

rdblue · 2019-10-09T19:31:04Z

It looks like this is trying to assign the same IDs for a spec each time it is created, but I think the approach should be to assign IDs to each field in a spec. The JSON serialization should be updated to parse an ID for each field. That's a good place to start, just adding the ability to track an ID for each partition field

manishmalhotrawork · 2019-10-10T22:13:23Z

@rdblue thanks.

It looks like this is trying to assign the same IDs for a spec each time it is created, but I think the approach should be to assign IDs to each field in a spec.

PartitionSpecParser.getLastPartitionField(JsonNode manifestSchemaJson) is parsing the old PartitionSpec manifestSchema to find the max id from the field list. and this Id will be used as the init Id if new PartitionSpec is added. My understanding was to use the persistent partition field_id, and then use +1 of it for next partitionSpec. So that 2 partitionSpec will not have same partition_field Id.

The JSON serialization should be updated to parse an ID for each field. That's a good place to start, just adding the ability to track an ID for each partition field

I believe you meant when Avro file is created using AvroFileAppender?

rdblue · 2019-10-10T22:23:16Z

this Id will be used as the init Id if new PartitionSpec is added

This is assigning an ID. Those IDs should be statically assigned when the partition spec is created the first time and stored with the field information when it is serialized. The highest assigned ID in any spec should be kept in table-level metadata, like the lastColumnId property.

I believe you meant when Avro file is created using AvroFileAppender?

No; JSON serialization of a partition spec should encode the IDs. That gets put into file metadata, but the main thing is to add field IDs to partition fields.

manishmalhotrawork · 2019-10-10T23:29:40Z

this Id will be used as the init Id if new PartitionSpec is added

This is assigning an ID. Those IDs should be statically assigned when the partition spec is created the first time and stored with the field information when it is serialized. The highest assigned ID in any spec should be kept in table-level metadata, like the lastColumnId property.

Ok, I started that way, but currently PartitionSpec doesn't store the partitionFieldId ( 1000+) but stores the schema fieldId ( 1+) which is real schema id, It stores the partitionFieldId in the avro-schema of the file. Also the schema field Ids starts from 0, when converted from say from spark schema, or assigned manually when created the NestedField as the columns of the schema.

So, in case of partitionFieldId also, when its firstTime newMetadata is created we can assign, but next time it should be part of the PartitionSpec Object.

just to verify the partitionFieldIds, this is the manifest file schema ( partial till has partition field-id )

  "type": "record",
  "name": "manifest_entry",
  "fields": [
    {
      "name": "status",
      "type": "int",
      "field-id": 0
    },
    {
      "name": "snapshot_id",
      "type": "long",
      "field-id": 1
    },
    {
      "name": "data_file",
      "type": {
        "type": "record",
        "name": "r2",
        "fields": [
          {
            "name": "file_path",
            "type": "string",
            "field-id": 100
          },
          {
            "name": "file_format",
            "type": "string",
            "field-id": 101
          },
          {
            "name": "partition",
            "type": {
              "type": "record",
              "name": "r102",
              "fields": [
                **{
                  "name": "data_bucket",
                  "type": [
                    "null",
                    "int"
                  ],
                  "default": null,
                  "field-id": 1000
                }
              ]
            },**
            "field-id": 102
          }```

rdblue · 2019-10-11T00:14:06Z

Yeah, the first step is to add a partition field ID in addition to the existing source field ID.

manishmalhotrawork · 2019-10-11T19:59:47Z

Yeah, the first step is to add a partition field ID in addition to the existing source field ID.

Cool. then we also have to also handle cases where table is created with old way (partition-field if in ) and not adding partitionSpec to that table?

rdblue · 2019-10-11T20:26:53Z

then we also have to also handle cases where table is created with old way (partition-field if in ) and not adding partitionSpec to that table?

Yes. In that case, we can get IDs by assigning the same way they would be in the method that returns the partition schema.

manishmalhotrawork · 2019-10-18T08:18:54Z

thanks @rdblue for giving more details !

Please see, with last few commits. I'm trying to add partitionFieldId to table-metadata.
So, please review if this approach make sense to you, then will add more test-cases as well.

add partitionFieldId to PartitionField, and to be part of the table-metadata.
providing that value statically, and keeping the lastPartitionFieldId at the TableMetadata level like lastColumnId.
also tried to handle the cases where pre-existing tables will not have partitionFieldId in the schema. so keeping same logic of giving 1000+ as the value.
able to run all test cases in local.

rdblue · 2019-10-18T17:40:29Z

api/src/main/java/org/apache/iceberg/PartitionSpec.java

      Type sourceType = schema.findType(field.sourceId());
      Type resultType = field.transform().getResultType(sourceType);
-      // assign ids for partition fields starting at PARTITION_DATA_ID_START to leave room for data file's other fields
+      // assign ids for partition fields starting at 1000 to leave room for data file's other fields


This is no longer assigning IDs, so the comment can be removed.

let me take care.

rdblue · 2019-10-18T17:43:36Z

api/src/main/java/org/apache/iceberg/PartitionField.java

  private final Transform<?, ?> transform;

-  PartitionField(int sourceId, String name, Transform<?, ?> transform) {
+  PartitionField(int sourceId, int partitionFieldId, String name, Transform<?, ?> transform) {


The name partitionFieldId is redundant because the class is PartitionField.

Let's use the same convention that is used in types. The PartitionField should have an id instance variable that is accessed by a fieldId method.

rdblue · 2019-10-18T17:46:48Z

core/src/main/java/org/apache/iceberg/ManifestReader.java

      .add("value_counts", "null_value_counts", "lower_bounds", "upper_bounds")
      .build();
+  private static final String PARTITION_SPEC = "partition-spec";
+  private static final String SCHEMA = "schema";


These changes aren't functional. Can you please remove them?

rdblue · 2019-10-18T17:47:17Z

core/src/main/java/org/apache/iceberg/PartitionSpecParser.java

  private static final String SPEC_ID = "spec-id";
  private static final String FIELDS = "fields";
  private static final String SOURCE_ID = "source-id";
+  private static final String PARTITION_FIELD_ID = "partition-field-id";


I think "field-id" is fine here. The "partition" part is clear from context.

rdblue · 2019-10-18T17:48:07Z

core/src/main/java/org/apache/iceberg/SchemaUpdate.java

  @Override
  public void commit() {
-    TableMetadata update = applyChangesToMapping(base.updateSchema(apply(), lastColumnId));
+    TableMetadata update = applyChangesToMapping(base.updateSchema(apply(), lastColumnId, lastPartitionFieldId));


Why is the last field ID passed here?

yeah, its not required.
As its a schema update, and ideally we dont need keep lastPartitionFieldId at the schemaUpdate level.

rdblue · 2019-10-18T17:49:40Z

core/src/main/java/org/apache/iceberg/TableMetadata.java

  }

+  public int lastPartitionFieldId() {
+    return lastPartitionFieldId;


These names are good.

manishmalhotrawork · 2019-10-18T18:27:30Z

thanks @rdblue.
realized one more thing, I believe we also need to reuse the partitionFieldId ?
by checking if the column already exist in the old partitionSpecs , so that it can reuse the same partitionFieldId ?
For that we have to maintain a Map<String, Integer> as partitionFieldIdByColumnName in the TableMetadata so that, when new partitionSpec is updated, it can find the existing one, and reuse . the fieldId.

rdblue · 2019-10-18T18:35:30Z

Yes, we do want to reuse the fields across specs. We might want to make equality ignore the field ID for this purpose.

manishmalhotrawork · 2019-10-19T07:30:51Z

@rdblue please see, updated PR to reusing field-id .
one downside I see is, have to iterate the PartitionSpec two times, otherwise have to initialize two maps from same method.
TableMetadata.indexSpecs and TableMetadata.indexPartitionFieldIdByColumnName are the ones. Though this will happen, only at the time of creating a new TableMetadata.
Please share your thoughts.

manishmalhotrawork · 2019-10-25T23:55:16Z

@rdblue it would be helpful, if you check this. thanks !

manishmalhotrawork · 2019-12-16T22:58:00Z

@rdblue it would be helpful, if you review this. thanks !

rdblue · 2019-12-23T21:31:16Z

api/src/main/java/org/apache/iceberg/PartitionSpec.java

 public class PartitionSpec implements Serializable {
-  // start assigning IDs for partition fields at 1000
-  private static final int PARTITION_DATA_ID_START = 1000;
+  public static final int PARTITION_DATA_ID_START = 1000;


Does this need to be public or can it be package-private?

Also, why remove the comment?

rdblue · 2019-12-23T21:34:57Z

api/src/main/java/org/apache/iceberg/PartitionField.java

      return false;
    }
-
+    // not considering field id, as field-id will be reused.


ID will be reused, but assignment is consistent because we assume that partition specs are not modified before the addition of partition field IDs. That means that tables start with only one spec that might not have IDs. Because we assign incrementally, IDs will always match when assigned using the default (1000, 1001, etc.).

Because we do have consistent IDs, I think this should check field ID here.

rdblue · 2019-12-23T21:35:23Z

api/src/main/java/org/apache/iceberg/PartitionField.java

      return false;
    }
-
+    // not considering field id, as field-id will be reused.


Nit: adding this comment removed a spacing line. Could you add it back?

rdblue · 2019-12-23T21:36:15Z

api/src/main/java/org/apache/iceberg/PartitionSpec.java

      this.schema = schema;
    }

+    private int incrementAndGetPartitionFieldId() {


How about nextFieldId? That's a much shorter name, but is still descriptive.

rdblue · 2019-12-23T21:44:23Z

core/src/main/java/org/apache/iceberg/PartitionSpecParser.java

+      } else {
+        partitionFieldId = partitionFieldId + 1;
+      }
+      builder.add(sourceId, partitionFieldId, name, transform);


It seems odd that partitionFieldId is incremented from the last value when missing. If a spec has one missing field ID, then it will be assigned based on the previous field's ID. I don't think this would cause problems because we expect either all fields to have assigned IDs, or no fields to have them.

I'd prefer to keep the logic for those cases separate to make this easier to follow. It isn't a good practice to rely on a hidden assumption that either all fields have ids or none do.

rdblue · 2019-12-23T21:44:54Z

core/src/main/java/org/apache/iceberg/TableMetadata.java

  }
+
+  private static Map<String, Integer> indexPartitionFieldIdByColumnName(List<PartitionSpec> specs) {
+    Map<String, Integer> result = new HashMap<>();


Please use Maps.newHashMap instead of instantiating one directly.

rdblue · 2019-12-23T21:45:52Z

hive/src/test/java/org/apache/iceberg/hive/TestHiveTableConcurrency.java


    executorService.shutdown();
-    Assert.assertTrue("Timeout", executorService.awaitTermination(2, TimeUnit.MINUTES));
+    Assert.assertTrue("Timeout", executorService.awaitTermination(5, TimeUnit.MINUTES));


I don't think these changes are still needed.

rdblue · 2019-12-23T21:46:52Z

core/src/main/java/org/apache/iceberg/TableMetadataParser.java

  static final String SNAPSHOT_ID = "snapshot-id";
  static final String TIMESTAMP_MS = "timestamp-ms";
  static final String SNAPSHOT_LOG = "snapshot-log";
+  static final String FIELDS = "fields";


Is this used?

rdblue · 2019-12-23T21:47:43Z

core/src/main/java/org/apache/iceberg/TableMetadataParser.java

+    List<PartitionField> fields = specs.get(specs.size() - 1).fields();
+    if (fields.size() > 0) {
+      // get the last lastPartitionFieldId
+      lastAssignedPartitionFieldId = fields.get(fields.size() - 1).fieldId();


Instead of getting the last ID, I think this should just keep track of the last assigned partition field ID, like the last-column-id. How about storing it as last-partition-id?

rdblue · 2019-12-23T21:50:21Z

core/src/main/java/org/apache/iceberg/TableMetadata.java

+          // increment and assign new id, if this column_transform has not used in partition yet.
+          (partitionFieldIdByColumnName == null) ? nextPartitionFieldId.incrementAndGet()
+          : ((partitionFieldIdByColumnName.containsKey(field.name())) ? partitionFieldIdByColumnName.get(field.name())
+              : nextPartitionFieldId.incrementAndGet()),


It is difficult to read nested ternary expressions. I recommend avoiding that pattern.

rdblue · 2019-12-23T21:51:18Z

core/src/main/java/org/apache/iceberg/TableMetadata.java

-
-    return specBuilder.build();
+    PartitionSpec freshSpec = specBuilder.build();
+    return freshSpec;


Looks like this change is unnecessary.

rdblue · 2019-12-23T22:02:47Z

core/src/main/java/org/apache/iceberg/TableMetadata.java

      // get a fresh spec to ensure the spec ID is set to the new default
-      builder.add(freshSpec(newDefaultSpecId, schema, newPartitionSpec));
+      PartitionSpec freshSpec = freshSpecWithAssignIds(newDefaultSpecId, schema, schema, newPartitionSpec,
+          nextPartitionFieldId, partitionFieldIdByColumnName);


I think this should work like updateSchema, where a different class is responsible for reassigning IDs. The TableMetadata class should validate consistency and help with tracking (like the snapshot log) but it shouldn't modify other objects that are passed in, like schemas, snapshots, and partition specs.

@rdblue
are we referring to this
Schema freshSchema = TypeUtil.assignFreshIds(schema, lastColumnId::incrementAndGet);

as TypeUtil.assignFreshIds assign id to the schema ?

rdblue · 2019-12-23T22:04:45Z

core/src/main/java/org/apache/iceberg/TableMetadata.java

-    PartitionSpec.Builder specBuilder = PartitionSpec.builderFor(schema)
+  private static PartitionSpec freshSpecWithAssignIds(int specId, Schema newSchema, Schema oldSchema,
+                                                      PartitionSpec partitionSpec, AtomicInteger nextPartitionFieldId,
+                                                      Map<String, Integer> partitionFieldIdByColumnName) {


I don't think that it is necessary to have this method here. The freshSpec method just ensures that all specs have the correct schema associated.

rdblue · 2019-12-23T22:11:59Z

core/src/main/java/org/apache/iceberg/TableMetadata.java

-    PartitionSpec freshSpec = specBuilder.build();
+    AtomicInteger lastPartitionFieldId = new AtomicInteger(PartitionSpec.PARTITION_DATA_ID_START - 1);
+    PartitionSpec freshSpec = freshSpecWithAssignIds(INITIAL_SPEC_ID, freshSchema, schema, spec, lastPartitionFieldId,
+        null);


I think that this should assign fresh IDs to the partition spec fields, but you can just add the ID to the existing code.

Also, if you are using an AtomicInteger, you can use getAndIncrement to avoid needing to subtract 1 from the ID starting point.

rdblue · 2019-12-23T22:12:42Z

@manishmalhotrawork, I've added review comments. Sorry I wasn't able to get back to this sooner!

manishmalhotrawork · 2020-01-05T09:48:52Z

@rdblue np, thanks for reviewing !

May be I'll raise a new PR with the required changes, it would be cleaner.

rdblue · 2020-01-07T18:51:02Z

Thanks for working on it, @manishmalhotrawork. If you do open a new PR, please remember to close this one. Up to you which one you want to do.

jun-he · 2020-03-03T07:35:50Z

@manishmalhotrawork
Can you let me know if you are still working on it? If you are busy, I will continue this work and finish it. Thanks.

rdblue · 2020-04-06T23:05:15Z

I'm closing this because it has been picked up as #845.

rdblue reviewed Oct 9, 2019

View reviewed changes

api/src/main/java/org/apache/iceberg/PartitionSpec.java Show resolved Hide resolved

rdblue reviewed Oct 9, 2019

View reviewed changes

rdblue reviewed Oct 18, 2019

View reviewed changes

manishmalhotrawork force-pushed the add_partition-field branch from c330531 to 9f15a3a Compare October 19, 2019 06:10

Add persistent IDs to partition fields

8e6d147

manishmalhotrawork force-pushed the add_partition-field branch from 0229751 to 8e6d147 Compare October 19, 2019 22:48

rdblue reviewed Dec 23, 2019

View reviewed changes

jun-he mentioned this pull request Mar 16, 2020

Add persistent IDs to partition fields #845

Merged

rdblue closed this Apr 6, 2020

Add persistent IDs to partition fields (WIP) #499

Add persistent IDs to partition fields (WIP) #499

Uh oh!

Conversation

manishmalhotrawork commented Sep 26, 2019

Uh oh!

manishmalhotrawork commented Sep 27, 2019

Uh oh!

manishmalhotrawork commented Sep 27, 2019

Uh oh!

manishmalhotrawork commented Oct 3, 2019

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

rdblue commented Oct 9, 2019

Uh oh!

manishmalhotrawork commented Oct 10, 2019 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

rdblue commented Oct 10, 2019

Uh oh!

manishmalhotrawork commented Oct 10, 2019

Uh oh!

rdblue commented Oct 11, 2019

Uh oh!

manishmalhotrawork commented Oct 11, 2019 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

rdblue commented Oct 11, 2019

Uh oh!

manishmalhotrawork commented Oct 18, 2019

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

manishmalhotrawork commented Oct 18, 2019

Uh oh!

rdblue commented Oct 18, 2019

Uh oh!

manishmalhotrawork commented Oct 19, 2019 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

manishmalhotrawork commented Oct 25, 2019

Uh oh!

manishmalhotrawork commented Dec 16, 2019

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

manishmalhotrawork commented Oct 10, 2019 •

edited

Loading

manishmalhotrawork commented Oct 11, 2019 •

edited

Loading

manishmalhotrawork commented Oct 19, 2019 •

edited

Loading