support create table

paimon-core/src/main/java/org/apache/paimon/catalog/AbstractCatalog.java

@@ -300,22 +300,29 @@ public void createTable(Identifier identifier, Schema schema, boolean ignoreIfEx
         }
 
         copyTableDefaultOptions(schema.options());
-
-        switch (Options.fromMap(schema.options()).get(TYPE)) {
-            case TABLE:
-            case MATERIALIZED_TABLE:
-                createTableImpl(identifier, schema);
-                break;
-            case OBJECT_TABLE:
-                createObjectTable(identifier, schema);
-                break;
-            case FORMAT_TABLE:
-                createFormatTable(identifier, schema);
-                break;
+        try {
+            switch (Options.fromMap(schema.options()).get(TYPE)) {
+                case TABLE:
+                case MATERIALIZED_TABLE:
+                    createTableImpl(identifier, schema);
+                    break;
+                case OBJECT_TABLE:
+                    createObjectTable(identifier, schema);
+                    break;
+                case FORMAT_TABLE:
+                    createFormatTable(identifier, schema);
+                    break;
+            }
+        } catch (TableAlreadyExistException e) {
+            if (ignoreIfExists) {
+                return;
+            }
+            throw e;
         }
     }
 
-    private void createObjectTable(Identifier identifier, Schema schema) {
+    private void createObjectTable(Identifier identifier, Schema schema)
+            throws TableAlreadyExistException {
         RowType rowType = schema.rowType();
         checkArgument(
                 rowType.getFields().isEmpty()
@@ -330,7 +337,8 @@ private void createObjectTable(Identifier identifier, Schema schema) {
         createTableImpl(identifier, schema.copy(ObjectTable.SCHEMA));
     }
 
-    protected abstract void createTableImpl(Identifier identifier, Schema schema);
+    protected abstract void createTableImpl(Identifier identifier, Schema schema)
+            throws TableAlreadyExistException;
 
     @Override
     public void renameTable(Identifier fromTable, Identifier toTable, boolean ignoreIfNotExists)

paimon-core/src/main/java/org/apache/paimon/rest/RESTCatalog.java

@@ -30,9 +30,11 @@
 import org.apache.paimon.rest.auth.AuthSession;
 import org.apache.paimon.rest.auth.CredentialsProvider;
 import org.apache.paimon.rest.auth.CredentialsProviderFactory;
+import org.apache.paimon.rest.exceptions.AlreadyExistsException;
 import org.apache.paimon.rest.exceptions.NoSuchResourceException;
 import org.apache.paimon.rest.requests.AlterDatabaseRequest;
 import org.apache.paimon.rest.requests.CreateDatabaseRequest;
+import org.apache.paimon.rest.requests.CreateTableRequest;
 import org.apache.paimon.rest.responses.AlterDatabaseResponse;
 import org.apache.paimon.rest.responses.ConfigResponse;
 import org.apache.paimon.rest.responses.CreateDatabaseResponse;
@@ -206,7 +208,19 @@ protected TableSchema getDataTableSchema(Identifier identifier) throws TableNotE
     }
 
     @Override
-    protected void createTableImpl(Identifier identifier, Schema schema) {}
+    protected void createTableImpl(Identifier identifier, Schema schema)
+            throws TableAlreadyExistException {
+        try {
+            CreateTableRequest request = new CreateTableRequest(identifier, schema);
+            client.post(
+                    resourcePaths.tables(identifier.getDatabaseName()),
+                    request,
+                    GetTableResponse.class,
+                    headers());
+        } catch (AlreadyExistsException e) {
+            throw new TableAlreadyExistException(identifier);
+        }
+    }
 
     @Override
     protected void renameTableImpl(Identifier fromTable, Identifier toTable) {}

paimon-core/src/main/java/org/apache/paimon/rest/requests/CreateTableRequest.java

@@ -0,0 +1,58 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.rest.requests;
+
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.rest.RESTRequest;
+import org.apache.paimon.schema.Schema;
+
+import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
+import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonGetter;
+import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
+
+/** Request for creating table. */
+public class CreateTableRequest implements RESTRequest {
+
+    private static final String FIELD_IDENTIFIER = "identifier";
+    private static final String FIELD_SCHEMA = "schema";
+
+    @JsonProperty(FIELD_IDENTIFIER)
+    private Identifier identifier;
+
+    @JsonProperty(FIELD_SCHEMA)
+    private Schema schema;
+
+    @JsonCreator
+    public CreateTableRequest(
+            @JsonProperty(FIELD_IDENTIFIER) Identifier identifier,
+            @JsonProperty(FIELD_SCHEMA) Schema schema) {
+        this.identifier = identifier;
+        this.schema = schema;
+    }
+
+    @JsonGetter(FIELD_IDENTIFIER)
+    public Identifier getIdentifier() {
+        return identifier;
+    }
+
+    @JsonGetter(FIELD_SCHEMA)
+    public Schema getSchema() {
+        return schema;
+    }
+}