API, Core: Move schemaID from ViewRepresentation to ViewVersion and make it required

.palantir/revapi.yml

@@ -410,9 +410,30 @@ acceptedBreaks:
       old: "field org.apache.iceberg.actions.RewriteDataFiles.MAX_CONCURRENT_FILE_GROUP_REWRITES_DEFAULT"
       new: "field org.apache.iceberg.actions.RewriteDataFiles.MAX_CONCURRENT_FILE_GROUP_REWRITES_DEFAULT"
       justification: "Update default to align with recommendation and allow more parallelism"
+    - code: "java.method.addedToInterface"
+      new: "method int org.apache.iceberg.view.ViewVersion::schemaId()"
+      justification: "Moving SchemaID to ViewVersion. View Spec implementation has\
+        \ not been voted on yet so this break is acceptable"
     - code: "java.method.addedToInterface"
       new: "method java.lang.String org.apache.iceberg.view.ViewVersion::operation()"
       justification: "Add operation API to view version"
+    - code: "java.method.removed"
+      old: "method java.lang.Integer org.apache.iceberg.view.ImmutableSQLViewRepresentation::schemaId()"
+      justification: "Moving SchemaID to ViewVersion. View Spec implementation has\
+        \ not been voted on yet so this break is acceptable"
+    - code: "java.method.removed"
+      old: "method java.lang.Integer org.apache.iceberg.view.SQLViewRepresentation::schemaId()"
+      justification: "Moving SchemaID to ViewVersion. View Spec implementation has\
+        \ not been voted on yet so this break is acceptable"
+    - code: "java.method.removed"
+      old: "method org.apache.iceberg.view.ImmutableSQLViewRepresentation org.apache.iceberg.view.ImmutableSQLViewRepresentation::withSchemaId(java.lang.Integer)"
+      justification: "Moving SchemaID to ViewVersion. View Spec implementation has\
+        \ not been voted on yet so this break is acceptable"
+    - code: "java.method.removed"
+      old: "method org.apache.iceberg.view.ImmutableSQLViewRepresentation.Builder\
+        \ org.apache.iceberg.view.ImmutableSQLViewRepresentation.Builder::schemaId(java.lang.Integer)"
+      justification: "Moving SchemaID to ViewVersion. View Spec implementation has\
+        \ not been voted on yet so this break is acceptable"
     org.apache.iceberg:iceberg-core:
     - code: "java.class.removed"
       old: "class org.apache.iceberg.actions.BaseExpireSnapshotsActionResult"

api/src/main/java/org/apache/iceberg/view/SQLViewRepresentation.java

@@ -45,13 +45,6 @@ default String type() {
   @Nullable
   Namespace defaultNamespace();
 
-  /**
-   * The query output schema ID at version create time, without aliases or null if no schema is
-   * defined
-   */
-  @Nullable
-  Integer schemaId();
-
   /** The view field comments. */
   List<String> fieldComments();
 

api/src/main/java/org/apache/iceberg/view/ViewVersion.java

@@ -69,4 +69,7 @@ public interface ViewVersion {
   default String operation() {
     return summary().get("operation");
   }
+
+  /** The query output schema at version create time, without aliases */
+  int schemaId();
 }

core/src/main/java/org/apache/iceberg/view/SQLViewRepresentationParser.java

@@ -50,10 +50,6 @@ static void toJson(SQLViewRepresentation view, JsonGenerator generator) throws I
     generator.writeStringField(SQL, view.sql());
     generator.writeStringField(DIALECT, view.dialect());
 
-    if (view.schemaId() != null) {
-      generator.writeNumberField(SCHEMA_ID, view.schemaId());
-    }
-
     if (view.defaultCatalog() != null) {
       generator.writeStringField(DEFAULT_CATALOG, view.defaultCatalog());
     }
@@ -93,9 +89,6 @@ static SQLViewRepresentation fromJson(JsonNode node) {
     }
 
     Integer schemaId = JsonUtil.getIntOrNull(SCHEMA_ID, node);
-    if (schemaId != null) {
-      builder.schemaId(schemaId);
-    }
 
     List<String> namespace = JsonUtil.getStringListOrNull(DEFAULT_NAMESPACE, node);
     if (namespace != null && !namespace.isEmpty()) {

core/src/main/java/org/apache/iceberg/view/ViewVersionParser.java

@@ -33,6 +33,7 @@ class ViewVersionParser {
   private static final String SUMMARY = "summary";
   private static final String OPERATION = "operation";
   private static final String REPRESENTATIONS = "representations";
+  private static final String SCHEMA_ID = "schema-id";
 
   private ViewVersionParser() {}
 
@@ -42,6 +43,7 @@ static void toJson(ViewVersion version, JsonGenerator generator) throws IOExcept
 
     generator.writeNumberField(VERSION_ID, version.versionId());
     generator.writeNumberField(TIMESTAMP_MS, version.timestampMillis());
+    generator.writeNumberField(SCHEMA_ID, version.schemaId());
 
     generator.writeObjectFieldStart(SUMMARY);
     generator.writeStringField(OPERATION, version.operation());
@@ -77,6 +79,7 @@ static ViewVersion fromJson(JsonNode node) {
         node.isObject(), "Cannot parse view version from a non-object: %s", node);
 
     int versionId = JsonUtil.getInt(VERSION_ID, node);
+    int schemaId = JsonUtil.getInt(SCHEMA_ID, node);
     long timestamp = JsonUtil.getLong(TIMESTAMP_MS, node);
     Map<String, String> summary = JsonUtil.getStringMap(SUMMARY, node);
     Preconditions.checkArgument(
@@ -93,6 +96,7 @@ static ViewVersion fromJson(JsonNode node) {
     return ImmutableViewVersion.builder()
         .versionId(versionId)
         .timestampMillis(timestamp)
+        .schemaId(schemaId)
         .summary(summary)
         .representations(representations.build())
         .build();

core/src/test/java/org/apache/iceberg/view/TestSQLViewRepresentationParser.java

@@ -41,7 +41,7 @@ public void testParseSqlViewRepresentation() {
         SQLViewRepresentationParser.fromJson(requiredFields));
 
     String requiredAndOptionalFields =
-        "{\"type\":\"sql\", \"sql\": \"select * from foo\", \"schema-id\": 1, \"dialect\": \"spark-sql\", "
+        "{\"type\":\"sql\", \"sql\": \"select * from foo\", \"dialect\": \"spark-sql\", "
             + "\"default-catalog\":\"cat\", "
             + "\"default-namespace\":[\"part1\",\"part2\"], "
             + "\"field-aliases\":[\"col1\", \"col2\"], "
@@ -50,7 +50,6 @@ public void testParseSqlViewRepresentation() {
     SQLViewRepresentation viewWithOptionalFields =
         ImmutableSQLViewRepresentation.builder()
             .sql("select * from foo")
-            .schemaId(1)
             .dialect("spark-sql")
             .defaultCatalog("cat")
             .fieldAliases(ImmutableList.of("col1", "col2"))
@@ -91,7 +90,7 @@ public void testViewRepresentationSerialization() {
         ViewRepresentationParser.toJson(viewRepresentation));
 
     String requiredAndOptionalFields =
-        "{\"type\":\"sql\",\"sql\":\"select * from foo\",\"dialect\":\"spark-sql\",\"schema-id\":1,"
+        "{\"type\":\"sql\",\"sql\":\"select * from foo\",\"dialect\":\"spark-sql\","
             + "\"default-catalog\":\"cat\","
             + "\"default-namespace\":[\"part1\",\"part2\"],"
             + "\"field-aliases\":[\"col1\",\"col2\"],"
@@ -100,7 +99,6 @@ public void testViewRepresentationSerialization() {
     SQLViewRepresentation viewWithOptionalFields =
         ImmutableSQLViewRepresentation.builder()
             .sql("select * from foo")
-            .schemaId(1)
             .dialect("spark-sql")
             .defaultCatalog("cat")
             .fieldAliases(ImmutableList.of("col1", "col2"))

core/src/test/java/org/apache/iceberg/view/TestViewVersionParser.java

@@ -45,6 +45,7 @@ public void testParseViewVersion() {
             .timestampMillis(12345)
             .addRepresentations(firstRepresentation, secondRepresentation)
             .summary(ImmutableMap.of("operation", "create", "user", "some-user"))
+            .schemaId(1)
             .build();
 
     String serializedRepresentations =
@@ -53,7 +54,7 @@ public void testParseViewVersion() {
 
     String serializedViewVersion =
         String.format(
-            "{\"version-id\":1, \"timestamp-ms\":12345, \"summary\":{\"operation\":\"create\", \"user\":\"some-user\"}, \"representations\":%s}",
+            "{\"version-id\":1, \"timestamp-ms\":12345, \"schema-id\":1, \"summary\":{\"operation\":\"create\", \"user\":\"some-user\"}, \"representations\":%s}",
             serializedRepresentations);
 
     Assert.assertEquals(
@@ -81,6 +82,7 @@ public void testSerializeViewVersion() {
             .timestampMillis(12345)
             .addRepresentations(firstRepresentation, secondRepresentation)
             .summary(ImmutableMap.of("operation", "create", "user", "some-user"))
+            .schemaId(1)
             .build();
 
     String expectedRepresentations =
@@ -89,7 +91,7 @@ public void testSerializeViewVersion() {
 
     String expectedViewVersion =
         String.format(
-            "{\"version-id\":1,\"timestamp-ms\":12345,\"summary\":{\"operation\":\"create\",\"user\":\"some-user\"},\"representations\":%s}",
+            "{\"version-id\":1,\"timestamp-ms\":12345,\"schema-id\":1,\"summary\":{\"operation\":\"create\",\"user\":\"some-user\"},\"representations\":%s}",
             expectedRepresentations);
 
     Assert.assertEquals(
@@ -106,7 +108,7 @@ public void testFailParsingMissingOperation() {
 
     String viewVersionMissingOperation =
         String.format(
-            "{\"version-id\":1,\"timestamp-ms\":12345,\"summary\":{\"some-other-field\":\"some-other-value\"},\"representations\":%s}",
+            "{\"version-id\":1,\"timestamp-ms\":12345,\"summary\":{\"some-other-field\":\"some-other-value\"},\"representations\":%s,\"schema-id\":1}",
             serializedRepresentations);
 
     Assertions.assertThatThrownBy(() -> ViewVersionParser.fromJson(viewVersionMissingOperation))