From 0a216401d007a77829730714e9f66590fb25c808 Mon Sep 17 00:00:00 2001
From: Eduard Tudenhoefner <etudenhoefner@gmail.com>
Date: Tue, 23 Apr 2024 16:46:24 +0200
Subject: [PATCH] Core: Assign fresh IDs to view schema

---
 .../iceberg/rest/RESTSessionCatalog.java      | 10 +++
 .../view/BaseMetastoreViewCatalog.java        | 10 +++
 .../iceberg/view/ViewVersionReplace.java      |  8 +++
 .../apache/iceberg/view/ViewCatalogTests.java | 61 +++++++++++++------
 4 files changed, 70 insertions(+), 19 deletions(-)

diff --git a/core/src/main/java/org/apache/iceberg/rest/RESTSessionCatalog.java b/core/src/main/java/org/apache/iceberg/rest/RESTSessionCatalog.java
index 94f3057f9fe7..223c0f859ae6 100644
--- a/core/src/main/java/org/apache/iceberg/rest/RESTSessionCatalog.java
+++ b/core/src/main/java/org/apache/iceberg/rest/RESTSessionCatalog.java
@@ -32,6 +32,7 @@
 import java.util.concurrent.Future;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
 import java.util.function.BiFunction;
 import java.util.function.Function;
 import java.util.function.Supplier;
@@ -92,6 +93,7 @@
 import org.apache.iceberg.rest.responses.LoadViewResponse;
 import org.apache.iceberg.rest.responses.OAuthTokenResponse;
 import org.apache.iceberg.rest.responses.UpdateNamespacePropertiesResponse;
+import org.apache.iceberg.types.TypeUtil;
 import org.apache.iceberg.util.EnvironmentUtil;
 import org.apache.iceberg.util.Pair;
 import org.apache.iceberg.util.PropertyUtil;
@@ -1185,6 +1187,8 @@ public View create() {
       Preconditions.checkState(
           null != defaultNamespace, "Cannot create view without specifying a default namespace");
 
+      schema = TypeUtil.assignFreshIds(schema, new AtomicInteger(0)::incrementAndGet);
+
       ViewVersion viewVersion =
           ImmutableViewVersion.builder()
               .versionId(1)
@@ -1262,6 +1266,12 @@ private View replace(LoadViewResponse response) {
               .max(Integer::compareTo)
               .orElseGet(metadata::currentVersionId);
 
+      schema =
+          TypeUtil.assignFreshIds(
+              schema,
+              metadata.schema(),
+              new AtomicInteger(metadata.schema().highestFieldId())::incrementAndGet);
+
       ViewVersion viewVersion =
           ImmutableViewVersion.builder()
               .versionId(maxVersionId + 1)
diff --git a/core/src/main/java/org/apache/iceberg/view/BaseMetastoreViewCatalog.java b/core/src/main/java/org/apache/iceberg/view/BaseMetastoreViewCatalog.java
index 6e2d6ff5e864..89af4138c07d 100644
--- a/core/src/main/java/org/apache/iceberg/view/BaseMetastoreViewCatalog.java
+++ b/core/src/main/java/org/apache/iceberg/view/BaseMetastoreViewCatalog.java
@@ -20,6 +20,7 @@
 
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.atomic.AtomicInteger;
 import org.apache.iceberg.BaseMetastoreCatalog;
 import org.apache.iceberg.EnvironmentContext;
 import org.apache.iceberg.Schema;
@@ -33,6 +34,7 @@
 import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
 import org.apache.iceberg.relocated.com.google.common.collect.Lists;
 import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.apache.iceberg.types.TypeUtil;
 
 public abstract class BaseMetastoreViewCatalog extends BaseMetastoreCatalog implements ViewCatalog {
   protected abstract ViewOperations newViewOps(TableIdentifier identifier);
@@ -155,6 +157,8 @@ private View create(ViewOperations ops) {
       Preconditions.checkState(
           null != defaultNamespace, "Cannot create view without specifying a default namespace");
 
+      schema = TypeUtil.assignFreshIds(schema, new AtomicInteger(0)::incrementAndGet);
+
       ViewVersion viewVersion =
           ImmutableViewVersion.builder()
               .versionId(1)
@@ -204,6 +208,12 @@ private View replace(ViewOperations ops) {
               .max(Integer::compareTo)
               .orElseGet(metadata::currentVersionId);
 
+      schema =
+          TypeUtil.assignFreshIds(
+              schema,
+              metadata.schema(),
+              new AtomicInteger(metadata.schema().highestFieldId())::incrementAndGet);
+
       ViewVersion viewVersion =
           ImmutableViewVersion.builder()
               .versionId(maxVersionId + 1)
diff --git a/core/src/main/java/org/apache/iceberg/view/ViewVersionReplace.java b/core/src/main/java/org/apache/iceberg/view/ViewVersionReplace.java
index 8b3d087940a5..0ddf42c6e981 100644
--- a/core/src/main/java/org/apache/iceberg/view/ViewVersionReplace.java
+++ b/core/src/main/java/org/apache/iceberg/view/ViewVersionReplace.java
@@ -28,12 +28,14 @@
 import static org.apache.iceberg.TableProperties.COMMIT_TOTAL_RETRY_TIME_MS_DEFAULT;
 
 import java.util.List;
+import java.util.concurrent.atomic.AtomicInteger;
 import org.apache.iceberg.EnvironmentContext;
 import org.apache.iceberg.Schema;
 import org.apache.iceberg.catalog.Namespace;
 import org.apache.iceberg.exceptions.CommitFailedException;
 import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
 import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.types.TypeUtil;
 import org.apache.iceberg.util.PropertyUtil;
 import org.apache.iceberg.util.Tasks;
 
@@ -71,6 +73,12 @@ ViewMetadata internalApply() {
             .max(Integer::compareTo)
             .orElseGet(viewVersion::versionId);
 
+    schema =
+        TypeUtil.assignFreshIds(
+            schema,
+            base.schema(),
+            new AtomicInteger(base.schema().highestFieldId())::incrementAndGet);
+
     ViewVersion newVersion =
         ImmutableViewVersion.builder()
             .versionId(maxVersionId + 1)
diff --git a/core/src/test/java/org/apache/iceberg/view/ViewCatalogTests.java b/core/src/test/java/org/apache/iceberg/view/ViewCatalogTests.java
index b3765bb1eae7..3c128d70c29d 100644
--- a/core/src/test/java/org/apache/iceberg/view/ViewCatalogTests.java
+++ b/core/src/test/java/org/apache/iceberg/view/ViewCatalogTests.java
@@ -52,9 +52,20 @@ public abstract class ViewCatalogTests<C extends ViewCatalog & SupportsNamespace
           required(3, "id", Types.IntegerType.get(), "unique ID"),
           required(4, "data", Types.StringType.get()));
 
+  // actual schema for the view, with column IDs reassigned
+  protected static final Schema VIEW_SCHEMA =
+      new Schema(
+          0,
+          required(1, "id", Types.IntegerType.get(), "unique ID"),
+          required(2, "data", Types.StringType.get()));
+
   private static final Schema OTHER_SCHEMA =
       new Schema(7, required(1, "some_id", Types.IntegerType.get()));
 
+  // actual replaced schema for the view, with column IDs reassigned
+  private static final Schema OTHER_VIEW_SCHEMA =
+      new Schema(1, required(3, "some_id", Types.IntegerType.get()));
+
   protected abstract C catalog();
 
   protected abstract Catalog tableCatalog();
@@ -102,10 +113,10 @@ public void basicCreateView() {
         .first()
         .extracting(ViewHistoryEntry::versionId)
         .isEqualTo(1);
-    assertThat(view.schema().schemaId()).isEqualTo(0);
-    assertThat(view.schema().asStruct()).isEqualTo(SCHEMA.asStruct());
+    assertThat(view.schema().schemaId()).isEqualTo(VIEW_SCHEMA.schemaId());
+    assertThat(view.schema().asStruct()).isEqualTo(VIEW_SCHEMA.asStruct());
     assertThat(view.currentVersion().operation()).isEqualTo("create");
-    assertThat(view.schemas()).hasSize(1).containsKey(0);
+    assertThat(view.schemas()).hasSize(1).containsKey(VIEW_SCHEMA.schemaId());
     assertThat(view.versions()).hasSize(1).containsExactly(view.currentVersion());
 
     assertThat(view.currentVersion())
@@ -113,7 +124,7 @@ public void basicCreateView() {
             ImmutableViewVersion.builder()
                 .timestampMillis(view.currentVersion().timestampMillis())
                 .versionId(1)
-                .schemaId(0)
+                .schemaId(VIEW_SCHEMA.schemaId())
                 .summary(view.currentVersion().summary())
                 .defaultNamespace(identifier.namespace())
                 .addRepresentations(
@@ -173,9 +184,9 @@ public void completeCreateView() {
         .extracting(ViewHistoryEntry::versionId)
         .isEqualTo(1);
     assertThat(view.currentVersion().operation()).isEqualTo("create");
-    assertThat(view.schema().schemaId()).isEqualTo(0);
-    assertThat(view.schema().asStruct()).isEqualTo(SCHEMA.asStruct());
-    assertThat(view.schemas()).hasSize(1).containsKey(0);
+    assertThat(view.schema().schemaId()).isEqualTo(VIEW_SCHEMA.schemaId());
+    assertThat(view.schema().asStruct()).isEqualTo(VIEW_SCHEMA.asStruct());
+    assertThat(view.schemas()).hasSize(1).containsKey(VIEW_SCHEMA.schemaId());
     assertThat(view.versions()).hasSize(1).containsExactly(view.currentVersion());
 
     assertThat(view.currentVersion())
@@ -183,7 +194,7 @@ public void completeCreateView() {
             ImmutableViewVersion.builder()
                 .timestampMillis(view.currentVersion().timestampMillis())
                 .versionId(1)
-                .schemaId(0)
+                .schemaId(VIEW_SCHEMA.schemaId())
                 .summary(view.currentVersion().summary())
                 .defaultNamespace(identifier.namespace())
                 .defaultCatalog(catalog().name())
@@ -885,9 +896,12 @@ public void createOrReplaceView(boolean useCreateOrReplace) {
         .extracting(ViewHistoryEntry::versionId)
         .isEqualTo(2);
 
-    assertThat(replacedView.schema().schemaId()).isEqualTo(1);
-    assertThat(replacedView.schema().asStruct()).isEqualTo(OTHER_SCHEMA.asStruct());
-    assertThat(replacedView.schemas()).hasSize(2).containsKey(0).containsKey(1);
+    assertThat(replacedView.schema().schemaId()).isEqualTo(OTHER_VIEW_SCHEMA.schemaId());
+    assertThat(replacedView.schema().asStruct()).isEqualTo(OTHER_VIEW_SCHEMA.asStruct());
+    assertThat(replacedView.schemas())
+        .hasSize(2)
+        .containsKey(VIEW_SCHEMA.schemaId())
+        .containsKey(OTHER_VIEW_SCHEMA.schemaId());
 
     ViewVersion replacedViewVersion = replacedView.currentVersion();
     assertThat(replacedView.versions())
@@ -895,7 +909,7 @@ public void createOrReplaceView(boolean useCreateOrReplace) {
         .containsExactly(viewVersion, replacedViewVersion);
     assertThat(replacedViewVersion).isNotNull();
     assertThat(replacedViewVersion.versionId()).isEqualTo(2);
-    assertThat(replacedViewVersion.schemaId()).isEqualTo(1);
+    assertThat(replacedViewVersion.schemaId()).isEqualTo(OTHER_VIEW_SCHEMA.schemaId());
     assertThat(replacedViewVersion.operation()).isEqualTo("replace");
     assertThat(replacedViewVersion.representations())
         .containsExactly(
@@ -1120,7 +1134,12 @@ public void replaceViewVersion() {
         .element(1)
         .extracting(ViewHistoryEntry::versionId)
         .isEqualTo(updatedView.currentVersion().versionId());
-    assertThat(updatedView.schemas()).hasSize(2).containsKey(0).containsKey(1);
+    assertThat(updatedView.schemas())
+        .hasSize(2)
+        .containsKey(VIEW_SCHEMA.schemaId())
+        .containsKey(OTHER_VIEW_SCHEMA.schemaId());
+    assertThat(updatedView.schema().schemaId()).isEqualTo(OTHER_VIEW_SCHEMA.schemaId());
+    assertThat(updatedView.schema().asStruct()).isEqualTo(OTHER_VIEW_SCHEMA.asStruct());
     assertThat(updatedView.versions())
         .hasSize(2)
         .containsExactly(viewVersion, updatedView.currentVersion());
@@ -1130,7 +1149,7 @@ public void replaceViewVersion() {
     assertThat(updatedViewVersion.versionId()).isEqualTo(viewVersion.versionId() + 1);
     assertThat(updatedViewVersion.operation()).isEqualTo("replace");
     assertThat(updatedViewVersion.representations()).hasSize(1).containsExactly(trino);
-    assertThat(updatedViewVersion.schemaId()).isEqualTo(1);
+    assertThat(updatedViewVersion.schemaId()).isEqualTo(OTHER_VIEW_SCHEMA.schemaId());
     assertThat(updatedViewVersion.defaultCatalog()).isEqualTo("default");
     assertThat(updatedViewVersion.defaultNamespace()).isEqualTo(identifier.namespace());
 
@@ -1585,6 +1604,8 @@ public void concurrentReplaceViewVersion() {
       viewOps.commit(current, sparkUpdate);
 
       View updatedView = catalog().loadView(identifier);
+      assertThat(updatedView.schema().schemaId()).isEqualTo(VIEW_SCHEMA.schemaId());
+      assertThat(updatedView.schema().asStruct()).isEqualTo(VIEW_SCHEMA.asStruct());
       ViewVersion viewVersion = updatedView.currentVersion();
       assertThat(viewVersion.versionId()).isEqualTo(3);
       assertThat(updatedView.versions()).hasSize(3);
@@ -1593,7 +1614,7 @@ public void concurrentReplaceViewVersion() {
               ImmutableViewVersion.builder()
                   .timestampMillis(updatedView.version(1).timestampMillis())
                   .versionId(1)
-                  .schemaId(0)
+                  .schemaId(VIEW_SCHEMA.schemaId())
                   .summary(updatedView.version(1).summary())
                   .defaultNamespace(identifier.namespace())
                   .addRepresentations(
@@ -1608,7 +1629,7 @@ public void concurrentReplaceViewVersion() {
               ImmutableViewVersion.builder()
                   .timestampMillis(updatedView.version(2).timestampMillis())
                   .versionId(2)
-                  .schemaId(1)
+                  .schemaId(OTHER_VIEW_SCHEMA.schemaId())
                   .summary(updatedView.version(2).summary())
                   .defaultNamespace(identifier.namespace())
                   .addRepresentations(
@@ -1623,7 +1644,7 @@ public void concurrentReplaceViewVersion() {
               ImmutableViewVersion.builder()
                   .timestampMillis(updatedView.version(3).timestampMillis())
                   .versionId(3)
-                  .schemaId(0)
+                  .schemaId(VIEW_SCHEMA.schemaId())
                   .summary(updatedView.version(3).summary())
                   .defaultNamespace(identifier.namespace())
                   .addRepresentations(
@@ -1638,6 +1659,8 @@ public void concurrentReplaceViewVersion() {
           .hasMessageContaining("Cannot commit");
 
       View updatedView = catalog().loadView(identifier);
+      assertThat(updatedView.schema().schemaId()).isEqualTo(OTHER_VIEW_SCHEMA.schemaId());
+      assertThat(updatedView.schema().asStruct()).isEqualTo(OTHER_VIEW_SCHEMA.asStruct());
       ViewVersion viewVersion = updatedView.currentVersion();
       assertThat(viewVersion.versionId()).isEqualTo(2);
       assertThat(updatedView.versions()).hasSize(2);
@@ -1646,7 +1669,7 @@ public void concurrentReplaceViewVersion() {
               ImmutableViewVersion.builder()
                   .timestampMillis(updatedView.version(1).timestampMillis())
                   .versionId(1)
-                  .schemaId(0)
+                  .schemaId(VIEW_SCHEMA.schemaId())
                   .summary(updatedView.version(1).summary())
                   .defaultNamespace(identifier.namespace())
                   .addRepresentations(
@@ -1661,7 +1684,7 @@ public void concurrentReplaceViewVersion() {
               ImmutableViewVersion.builder()
                   .timestampMillis(updatedView.version(2).timestampMillis())
                   .versionId(2)
-                  .schemaId(1)
+                  .schemaId(OTHER_VIEW_SCHEMA.schemaId())
                   .summary(updatedView.version(2).summary())
                   .defaultNamespace(identifier.namespace())
                   .addRepresentations(