Core: Fix Partitions table for evolved partition specs #4560

szehon-ho · 2022-04-14T07:49:16Z

#4516 fixes the schema of the partitions table in the case of changed partition specs to use Partitioning.partitionType (a union of all previous partition specs), but the data is still wrong in some cases:

The PartitionsMap is constructed with the table's current spec, which is used to generate hashcode of partition values. This leads to colliding of some values which get lost.
- Ref: https://github.com/apache/iceberg/blob/apache-iceberg-0.13.1/core/src/main/java/org/apache/iceberg/PartitionsTable.java#L99.
- Example: p1=foo, p2=bar collides with p1=foo if PartitionMap is instantiated with current spec of {p1}, and other instance is lost.
The partition values are just listed in order without any transformation, meaning they may in the wrong field in the unified schema.
- Ref:
  
  iceberg/core/src/main/java/org/apache/iceberg/PartitionsTable.java
  
  Line 93 in a78aa2d
  
  return StaticDataTask.Row.of(partition.key, partition.recordCount, partition.fileCount);
  
  .
- Example, spec evolution of p1,p2 => p2,p1 leads to wrong ordering of earlier partitions.

This fixes the problem by :

Instantiating the PartitionMap by Partitioning.partitionType, so all types are used in the hashcode generation.
Transforming the PartitionDatas to fit the final schema Partioning.partitionType()

szehon-ho · 2022-04-14T16:30:57Z

FYI @rdblue @aokolnychyi @RussellSpitzer @szlta if can you help review, thanks

RussellSpitzer · 2022-04-18T17:07:50Z

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

+      originalPartitionIndex++;
+    }
+
+    PartitionData result = new PartitionData(newSchema);


result -> normalizedPartition?

Good point, renamed relevant fields in this method.

RussellSpitzer · 2022-04-18T17:08:44Z

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

    return partitions.all();
  }

+  private static PartitionData normalizePartition(PartitionData partition, Types.StructType newSchema) {


newSchema -> normalizedPartitionSchema?

Maybe that's too long?

RussellSpitzer · 2022-04-18T17:12:49Z

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

+
+    PartitionData result = new PartitionData(newSchema);
+
+    int finalPartitionIndex = 0;


normalizedPartitionIndex (or normalizedIndex)

RussellSpitzer · 2022-04-18T17:38:15Z

.../src/test/java/org/apache/iceberg/spark/source/TestMetadataTablesWithPartitionEvolution.java

  }

+  @Test
+  public void testPartitionMetadataTable() throws ParseException {


As one more thing to test, can you check reordering partition transforms?
Going from data, category -> category,data

I think you have this covered but I want to make sure we have a test in there since I think this is a pretty common usecase

Added the test. Unfortunately it's a bit weird and does not work in V2 until #4292 is completely fixed (I believe @szlta is taking a look at the remaining point there which would fix this issue).

RussellSpitzer

One suggested additional test, and I'm not quite sure about the Precondition check but other than that looks good

szlta · 2022-04-19T16:13:27Z

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

    for (FileScanTask task : tasks) {
-      partitions.get(task.file().partition()).update(task.file());
+      PartitionData original = (PartitionData) task.file().partition();
+      PartitionData normalized = normalizePartition(original, Partitioning.partitionType(table));


Should we reuse the result of the Partitioning.partitionType(table) invocation at line 99 rather than calling it in every iteration?

Yea good point

szlta · 2022-04-19T16:19:01Z

Thanks for catching this @szehon-ho, this change looks good to me, just added a nit comment.

RussellSpitzer · 2022-04-19T22:03:33Z

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

  }

+  private static PartitionData normalizePartition(PartitionData partition, Types.StructType normalizedPartitionSchema) {
+    Map<Integer, Object> fieldIdToValues = Maps.newHashMap();


One thing we may want to consider here is caching these normalization by spec rather than recomputing the mapping for every partition value. Thinking about this code being run on millions of files

Still wasn't sure how to cache something per spec, I added a cache per partition for the normalized key, let me know if you have some thought?

By the way, I tried to use the un-normalized partition in the map itself as map key, but doesn't work, as it duplicates if the partition field has name change. So we still need to use normalized partition as map key.

Discussed offline, added cache of positional mappings to the final partition type, for each spec-id

RussellSpitzer · 2022-04-19T22:06:25Z

.../src/test/java/org/apache/iceberg/spark/source/TestMetadataTablesWithPartitionEvolution.java

+    // Re-added partition fields currently not re-associated: https://github.com/apache/iceberg/issues/4292
+    // In V1, dropped partition fields show separately when field is re-added
+    // In V2, re-added field currently conflicts with its deleted form
+    if (formatVersion == 1) {


Putting this in a new test with the "Assume(formatVersion ==1)" would be a bit cleaner.

Done, split this into 3 tests

RussellSpitzer

Two suggestions but I think this is good as is,

Spitting up the test so that we have individual tests for the different alterations
Caching the field mapping by partition spec

szehon-ho · 2022-04-21T00:00:50Z

@RussellSpitzer added additional cache of field positional mapping per spec-id as suggested , it's changed a bit, let me know if you can take another look

RussellSpitzer · 2022-04-21T19:23:51Z

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

+
+    LoadingCache<Integer, Integer[]> originalPartitionFieldPositionsBySpec = Caffeine.newBuilder().build(specId ->
+        originalPositions(table, specId, normalizedPartitionType));
+    LoadingCache<Pair<PartitionData, Integer>, PartitionData> normalizedPartitions = Caffeine.newBuilder().build(


This I don't think is super important to cache since the getters and setters are pretty fast and you have the integer mapping already. So I probably would drop this.

It was saving the construction of the new partitionData object, at the cost of memory. but yea it might not be worth it, dropped it.

RussellSpitzer · 2022-04-21T19:27:28Z

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

+    Types.StructType normalizedPartitionType = Partitioning.partitionType(table);
+    PartitionMap partitions = new PartitionMap();
+
+    LoadingCache<Integer, Integer[]> originalPartitionFieldPositionsBySpec = Caffeine.newBuilder().build(specId ->


I think the number of partition specs should be very small so I would probably just use a
Maps.newHashMap()

Then .computeIfAbsent

Not that I think there is something wrong with Caffeine, just seems a bit heavy weight to me in this use case

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

RussellSpitzer · 2022-04-21T20:00:43Z

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

    for (FileScanTask task : tasks) {
-      partitions.get(task.file().partition()).update(task.file());
+      PartitionData original = (PartitionData) task.file().partition();
+      int specId = task.spec().specId();


I was thinking this would look something more like

positionMapping = originalPositions.computeIfAbsent(specId, specId -> originalPositions()) make normalizedPartitionData for (originalIndex = 0; originalIndex < original.length; originalIndex ++) { normalizedPartitionData.put(positionMapping[originalIndex], original.get(originalIndex)) }

This could be a separate function like we have now but I don't know if it's that bad if we just inline it here

Done, yea your suggestion is cleaner.

It's a bit easier to read that logic in a separate function, so kept it outside.

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

RussellSpitzer · 2022-04-22T21:10:23Z

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

+      Partition partition = partitions.get(key);
      if (partition == null) {
        partition = new Partition(key);
-        partitions.put(StructLikeWrapper.forType(type).set(key), partition);


I'm a bit confused by this change, I believe the issue here is that StructType does not have a well defined hashFunction (since implementations can do whatever they like) which is why we use the Wrapper to make sure we have a valid hash. (and equals)

Changed to map of PartitionData (I feel, it should have been that way in the beginning)

RussellSpitzer · 2022-04-22T21:11:37Z

.../src/test/java/org/apache/iceberg/spark/source/TestMetadataTablesWithPartitionEvolution.java

    sql("CREATE TABLE %s (id bigint NOT NULL, category string, data string) USING iceberg " +
        "TBLPROPERTIES ('commit.manifest-merge.enabled' 'false')", tableName);
    initTable();
-


Unnecessary white space change here

I think if we want to make this change we need to do it in all the tests and currently this is missing format changes for testFilesMetadataTable and testWithUnknownTransfer. Probably fine to just keep it as is and match in the new tests

Removed whitespace

RussellSpitzer · 2022-04-22T21:22:56Z

.../src/test/java/org/apache/iceberg/spark/source/TestMetadataTablesWithPartitionEvolution.java

+
+    Table table = validationCatalog.loadTable(tableIdent);
+
+    table.updateSpec()


I think you could just do this in SparkSQL if you like and skip the refresh, but this is fine too.

Will just keep it for now then, matches existing tests

RussellSpitzer

LGTM! Thanks for baring with my many comments. Feel free to merge whenever you feel ready.

szehon-ho · 2022-04-25T21:48:32Z

No problem, thanks for all the suggestions !

(cherry picked from commit 4c3aac2)

github-actions bot added core spark labels Apr 14, 2022

szehon-ho mentioned this pull request Apr 14, 2022

Adding specId for partitions metadata table #4516

Merged

RussellSpitzer reviewed Apr 18, 2022

View reviewed changes

szlta reviewed Apr 19, 2022

View reviewed changes

RussellSpitzer reviewed Apr 19, 2022

View reviewed changes

RussellSpitzer approved these changes Apr 19, 2022

View reviewed changes

szehon-ho force-pushed the partition_key_evolving_spec branch from 8253071 to 3d2860f Compare April 20, 2022 07:10

RussellSpitzer reviewed Apr 21, 2022

View reviewed changes

core/src/main/java/org/apache/iceberg/PartitionsTable.java Outdated Show resolved Hide resolved

RussellSpitzer reviewed Apr 21, 2022

View reviewed changes

szehon-ho added 6 commits April 21, 2022 23:02

Core: Fix Partitions table for evolved partition specs

aac0ba8

Review comments

1f2eb63

Address review comments

da0dc4d

Split tests and some optimizations

983ec0e

Add cache for specId -> partition field position mappings

34bcae8

Review comments, use map instead of caffeine and cleanup code

5129068

szehon-ho force-pushed the partition_key_evolving_spec branch from 6d11188 to 5129068 Compare April 22, 2022 06:09

szehon-ho added 2 commits April 22, 2022 10:22

Cleanup - remove redundant code

b974f0e

Cleanup variable names for consistency (schema -> type)

fb53467

RussellSpitzer reviewed Apr 22, 2022

View reviewed changes

core/src/main/java/org/apache/iceberg/PartitionsTable.java Outdated Show resolved Hide resolved

RussellSpitzer reviewed Apr 22, 2022

View reviewed changes

More review comments

3cfb65c

szehon-ho force-pushed the partition_key_evolving_spec branch from 2c7d303 to 3cfb65c Compare April 22, 2022 23:29

RussellSpitzer approved these changes Apr 25, 2022

View reviewed changes

szehon-ho merged commit 4c3aac2 into apache:master Apr 25, 2022

szehon-ho mentioned this pull request May 4, 2022

Core: Fix Partitions table filtering for evolved partition specs #4637

Merged

szehon-ho mentioned this pull request Jun 21, 2022

StructLikeWrapper equals method is broken #5064

Closed

szehon-ho mentioned this pull request Jun 29, 2022

Support catalog method to set table metadata #5163

Closed

southernriver mentioned this pull request Oct 26, 2022

[Refactor] Update iceberg version to 0.14.1 StarRocks/starrocks#12306

Closed

9 tasks

szehon-ho mentioned this pull request May 2, 2023

Core: Simplify Partitions table partition-coercion code #7503

Merged

sunchao pushed a commit to sunchao/iceberg that referenced this pull request May 9, 2023

Core: Fix Partitions table for evolved partition specs (apache#4560)

69bf69c

(cherry picked from commit 4c3aac2)


		PartitionData result = new PartitionData(newSchema);

		int finalPartitionIndex = 0;


		Table table = validationCatalog.loadTable(tableIdent);

		table.updateSpec()

Core: Fix Partitions table for evolved partition specs #4560

Core: Fix Partitions table for evolved partition specs #4560

Uh oh!

Conversation

szehon-ho commented Apr 14, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

szehon-ho commented Apr 14, 2022

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!

szehon-ho Apr 18, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

RussellSpitzer left a comment

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!

szlta commented Apr 19, 2022

Uh oh!

Choose a reason for hiding this comment

Uh oh!

szehon-ho Apr 20, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

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!

RussellSpitzer left a comment

Choose a reason for hiding this comment

Uh oh!

szehon-ho commented Apr 21, 2022

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

RussellSpitzer Apr 21, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

szehon-ho Apr 21, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

RussellSpitzer Apr 22, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

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

szehon-ho commented Apr 14, 2022 •

edited

Loading

szehon-ho Apr 18, 2022 •

edited

Loading

szehon-ho Apr 20, 2022 •

edited

Loading

RussellSpitzer Apr 21, 2022 •

edited

Loading

szehon-ho Apr 21, 2022 •

edited

Loading

RussellSpitzer Apr 22, 2022 •

edited

Loading