Core: Add stage-create flag to REST create route

core/src/main/java/org/apache/iceberg/rest/RESTCatalog.java

@@ -336,6 +336,7 @@ private LoadTableResponse stageCreate() {
       Map<String, String> tableProperties = propertiesBuilder.build();
 
       CreateTableRequest request = CreateTableRequest.builder()
+          .stageCreate()
           .withName(ident.name())
           .withSchema(schema)
           .withPartitionSpec(spec)
@@ -344,10 +345,8 @@ private LoadTableResponse stageCreate() {
           .setProperties(tableProperties)
           .build();
 
-      // TODO: will this be a specific route or a modified create?
       return client.post(
-          paths.stageCreate(ident.namespace()),
-          request, LoadTableResponse.class, ErrorHandlers.tableErrorHandler());
+          paths.tables(ident.namespace()), request, LoadTableResponse.class, ErrorHandlers.tableErrorHandler());
     }
   }
 

core/src/main/java/org/apache/iceberg/rest/ResourcePaths.java

@@ -58,10 +58,6 @@ public String tables(Namespace ns) {
     return SLASH.join("v1", prefix, "namespaces", RESTUtil.encodeNamespace(ns), "tables");
   }
 
-  public String stageCreate(Namespace ns) {
-    return SLASH.join("v1", prefix, "namespaces", RESTUtil.encodeNamespace(ns), "stageCreate");
-  }
-
   public String table(TableIdentifier ident) {
     return SLASH.join(
         "v1", prefix, "namespaces", RESTUtil.encodeNamespace(ident.namespace()), "tables",

core/src/main/java/org/apache/iceberg/rest/requests/CreateTableRequest.java

@@ -41,26 +41,29 @@ public class CreateTableRequest implements RESTRequest {
   private PartitionSpec spec;
   private SortOrder order;
   private Map<String, String> properties;
+  private Boolean stageCreate;
 
   public CreateTableRequest() {
     // Needed for Jackson Deserialization.
   }
 
   private CreateTableRequest(String name, String location, Schema schema, PartitionSpec spec, SortOrder order,
-                             Map<String, String> properties) {
+                             Map<String, String> properties, boolean stageCreate) {
     this.name = name;
     this.location = location;
     this.schema = schema;
     this.spec = spec;
     this.order = order;
     this.properties = properties;
+    this.stageCreate = stageCreate;
     validate();
   }
 
   @Override
   public void validate() {
     Preconditions.checkArgument(name != null, "Invalid table name: null");
     Preconditions.checkArgument(schema != null, "Invalid schema: null");
+    Preconditions.checkArgument(stageCreate != null, "Invalid stageCreate flag: null");
   }
 
   public String name() {
@@ -87,6 +90,10 @@ public Map<String, String> properties() {
     return properties != null ? properties : ImmutableMap.of();
   }
 
+  public boolean stageCreate() {
+    return stageCreate;
+  }
+
   @Override
   public String toString() {
     return MoreObjects.toStringHelper(this)
@@ -110,6 +117,7 @@ public static class Builder {
     private PartitionSpec spec;
     private SortOrder order;
     private final ImmutableMap.Builder<String, String> properties = ImmutableMap.builder();
+    private boolean stageCreate = false;
 
     private Builder() {
     }
@@ -157,8 +165,13 @@ public Builder withWriteOrder(SortOrder writeOrder) {
       return this;
     }
 
+    public Builder stageCreate() {
+      this.stageCreate = true;
+      return this;
+    }
+
     public CreateTableRequest build() {
-      return new CreateTableRequest(name, location, schema, spec, order, properties.build());
+      return new CreateTableRequest(name, location, schema, spec, order, properties.build(), stageCreate);
     }
   }
 }

core/src/test/java/org/apache/iceberg/rest/RESTCatalogAdapter.java

@@ -96,7 +96,6 @@ private enum Route {
     UPDATE_NAMESPACE(HTTPMethod.POST, "v1/namespaces/{namespace}/properties"),
     LIST_TABLES(HTTPMethod.GET, "v1/namespaces/{namespace}/tables"),
     CREATE_TABLE(HTTPMethod.POST, "v1/namespaces/{namespace}/tables"),
-    STAGE_CREATE_TABLE(HTTPMethod.POST, "v1/namespaces/{namespace}/stageCreate"),
     LOAD_TABLE(HTTPMethod.GET, "v1/namespaces/{namespace}/tables/{table}"),
     UPDATE_TABLE(HTTPMethod.POST, "v1/namespaces/{namespace}/tables/{table}"),
     DROP_TABLE(HTTPMethod.DELETE, "v1/namespaces/{namespace}/tables/{table}");
@@ -203,13 +202,12 @@ public <T extends RESTResponse> T handleRequest(Route route, Map<String, String>
       case CREATE_TABLE: {
         Namespace namespace = namespaceFromPathVars(vars);
         CreateTableRequest request = castRequest(CreateTableRequest.class, body);
-        return castResponse(responseType, CatalogHandlers.createTable(catalog, namespace, request));
-      }
-
-      case STAGE_CREATE_TABLE: {
-        Namespace namespace = namespaceFromPathVars(vars);
-        CreateTableRequest request = castRequest(CreateTableRequest.class, body);
-        return castResponse(responseType, CatalogHandlers.stageTableCreate(catalog, namespace, request));
+        request.validate();
+        if (request.stageCreate()) {
+          return castResponse(responseType, CatalogHandlers.stageTableCreate(catalog, namespace, request));
+        } else {
+          return castResponse(responseType, CatalogHandlers.createTable(catalog, namespace, request));
+        }
       }
 
       case DROP_TABLE: {

core/src/test/java/org/apache/iceberg/rest/TestResourcePaths.java

@@ -105,27 +105,6 @@ public void testTablesWithMultipartNamespace() {
     Assert.assertEquals("v1/namespaces/n%00s/tables", withoutPrefix.tables(ns));
   }
 
-  @Test
-  public void testStageCreate() {
-    Namespace ns = Namespace.of("ns");
-    Assert.assertEquals("v1/ws/catalog/namespaces/ns/stageCreate", withPrefix.stageCreate(ns));
-    Assert.assertEquals("v1/namespaces/ns/stageCreate", withoutPrefix.stageCreate(ns));
-  }
-
-  @Test
-  public void testStageCreateWithSlash() {
-    Namespace ns = Namespace.of("n/s");
-    Assert.assertEquals("v1/ws/catalog/namespaces/n%2Fs/stageCreate", withPrefix.stageCreate(ns));
-    Assert.assertEquals("v1/namespaces/n%2Fs/stageCreate", withoutPrefix.stageCreate(ns));
-  }
-
-  @Test
-  public void testStageCreateWithMultipartNamespace() {
-    Namespace ns = Namespace.of("n", "s");
-    Assert.assertEquals("v1/ws/catalog/namespaces/n%00s/stageCreate", withPrefix.stageCreate(ns));
-    Assert.assertEquals("v1/namespaces/n%00s/stageCreate", withoutPrefix.stageCreate(ns));
-  }
-
   @Test
   public void testTable() {
     TableIdentifier ident = TableIdentifier.of("ns", "table");

open-api/rest-catalog-open-api.yaml

@@ -376,7 +376,19 @@ paths:
       tags:
         - Catalog API
       summary: Create a table in the given namespace
-      description: Create table
+      description:
+        Create a table or start a create transaction, like atomic CTAS.
+
+
+        If `stage-create` is false, the table is created immediately.
+
+
+        If `stage-create` is true, the table is not created, but table metadata is initialized and returned.
+        The service should prepare as needed for a commit to the table commit endpoint to complete the create
+        transaction. The client uses the returned metadata to begin a transaction. To commit the transaction,
+        the client sends all create and subsequent changes to the table commit route. Changes from the table
+        create operation include changes like AddSchemaUpdate and SetCurrentSchemaUpdate that set the initial
+        table state.
       operationId: createTable
       requestBody:
         content:
@@ -469,13 +481,21 @@ paths:
       description:
         Commit updates to a table.
 
+
         Commits have two parts, requirements and updates. Requirements are assertions that will be validated
         before attempting to make and commit changes. For example, `assert-ref-snapshot-id` will check that a
         named ref's snapshot ID has a certain value.
 
+
         Updates are changes to make to table metadata. For example, after asserting that the current main ref
         is at the expected snapshot, a commit may add a new child snapshot and set the ref to the new
         snapshot id.
+
+
+        Create table transactions that are started by createTable with `stage-create` set to true are
+        committed using this route. Transactions should include all changes to the table, including table
+        initialization, like AddSchemaUpdate and SetCurrentSchemaUpdate. The `assert-create` requirement is
+        used to ensure that the table was not created concurrently.
       requestBody:
         content:
           application/json:
@@ -1408,6 +1428,8 @@ components:
           $ref: '#/components/schemas/PartitionSpec'
         write-order:
           $ref: '#/components/schemas/SortOrder'
+        stage-create:
+          type: boolean
         properties:
           type: object
           additionalProperties: