[core] RESTCatalog: fix NotImplementedException handle when create and alter table

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

@@ -75,6 +75,7 @@
 import org.apache.paimon.schema.TableSchema;
 import org.apache.paimon.table.Table;
 import org.apache.paimon.table.TableSnapshot;
+import org.apache.paimon.table.system.SystemTableLoader;
 import org.apache.paimon.utils.Pair;
 import org.apache.paimon.view.View;
 import org.apache.paimon.view.ViewImpl;
@@ -286,6 +287,9 @@ public void alterDatabase(String name, List<PropertyChange> changes, boolean ign
     @Override
     public List<String> listTables(String databaseName) throws DatabaseNotExistException {
         try {
+            if (isSystemDatabase(databaseName)) {
+                return SystemTableLoader.loadGlobalTableNames();
+            }
             return listDataFromPageApi(
                     queryParams ->
                             client.get(
@@ -465,6 +469,8 @@ public void createTable(Identifier identifier, Schema schema, boolean ignoreIfEx
             if (!ignoreIfExists) {
                 throw new TableAlreadyExistException(identifier);
             }
+        } catch (NotImplementedException e) {
+            throw new RuntimeException(new UnsupportedOperationException(e.getMessage()));
         } catch (NoSuchResourceException e) {
             throw new DatabaseNotExistException(identifier.getDatabaseName());
         } catch (BadRequestException e) {
@@ -522,6 +528,8 @@ public void alterTable(
             throw new TableNoPermissionException(identifier, e);
         } catch (ServiceFailureException e) {
             throw new IllegalStateException(e.getMessage());
+        } catch (NotImplementedException e) {
+            throw new UnsupportedOperationException(e.getMessage());
         } catch (BadRequestException e) {
             throw new RuntimeException(new IllegalArgumentException(e.getMessage()));
         }

paimon-core/src/main/java/org/apache/paimon/rest/auth/DLFAuthProvider.java

@@ -89,9 +89,11 @@ public DLFAuthProvider(
     public Map<String, String> header(
             Map<String, String> baseHeader, RESTAuthParameter restAuthParameter) {
         try {
-            ZonedDateTime now = ZonedDateTime.now(ZoneOffset.UTC);
-            String date = now.format(AUTH_DATE_FORMATTER);
-            String dateTime = now.format(AUTH_DATE_TIME_FORMATTER);
+            String dateTime =
+                    baseHeader.getOrDefault(
+                            DLF_DATE_HEADER_KEY.toLowerCase(),
+                            ZonedDateTime.now(ZoneOffset.UTC).format(AUTH_DATE_TIME_FORMATTER));
+            String date = dateTime.substring(0, 8);
             Map<String, String> signHeaders =
                     generateSignHeaders(
                             restAuthParameter.data(), dateTime, token.getSecurityToken());

paimon-core/src/main/java/org/apache/paimon/rest/responses/BaseResourceAuditResponse.java

@@ -0,0 +1,88 @@
+/*
+ * 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.responses;
+
+import org.apache.paimon.rest.RESTResponse;
+
+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;
+
+/** Base class for database, table, view, audit response. */
+public abstract class BaseResourceAuditResponse implements RESTResponse {
+    protected static final String FIELD_OWNER = "owner";
+    protected static final String FIELD_CREATED_AT = "createdAt";
+    protected static final String FIELD_CREATED_BY = "createdBy";
+    protected static final String FIELD_UPDATED_AT = "updatedAt";
+    protected static final String FIELD_UPDATED_BY = "updatedBy";
+
+    @JsonProperty(FIELD_OWNER)
+    private final String owner;
+
+    @JsonProperty(FIELD_CREATED_AT)
+    private final long createdAt;
+
+    @JsonProperty(FIELD_CREATED_BY)
+    private final String createdBy;
+
+    @JsonProperty(FIELD_UPDATED_AT)
+    private final long updatedAt;
+
+    @JsonProperty(FIELD_UPDATED_BY)
+    private final String updatedBy;
+
+    @JsonCreator
+    public BaseResourceAuditResponse(
+            @JsonProperty(FIELD_OWNER) String owner,
+            @JsonProperty(FIELD_CREATED_AT) long createdAt,
+            @JsonProperty(FIELD_CREATED_BY) String createdBy,
+            @JsonProperty(FIELD_UPDATED_AT) long updatedAt,
+            @JsonProperty(FIELD_UPDATED_BY) String updatedBy) {
+        this.owner = owner;
+        this.createdAt = createdAt;
+        this.createdBy = createdBy;
+        this.updatedAt = updatedAt;
+        this.updatedBy = updatedBy;
+    }
+
+    @JsonGetter(FIELD_OWNER)
+    public String getOwner() {
+        return owner;
+    }
+
+    @JsonGetter(FIELD_CREATED_AT)
+    public long getCreatedAt() {
+        return createdAt;
+    }
+
+    @JsonGetter(FIELD_CREATED_BY)
+    public String getCreatedBy() {
+        return createdBy;
+    }
+
+    @JsonGetter(FIELD_UPDATED_AT)
+    public long getUpdatedAt() {
+        return updatedAt;
+    }
+
+    @JsonGetter(FIELD_UPDATED_BY)
+    public String getUpdatedBy() {
+        return updatedBy;
+    }
+}

paimon-core/src/main/java/org/apache/paimon/rest/responses/GetDatabaseResponse.java

@@ -33,7 +33,8 @@
 
 /** Response for getting database. */
 @JsonIgnoreProperties(ignoreUnknown = true)
-public class GetDatabaseResponse implements RESTResponse, Database {
+public class GetDatabaseResponse extends BaseResourceAuditResponse
+        implements RESTResponse, Database {
 
     private static final String FIELD_ID = "id";
     private static final String FIELD_NAME = "name";
@@ -52,7 +53,13 @@ public class GetDatabaseResponse implements RESTResponse, Database {
     public GetDatabaseResponse(
             @JsonProperty(FIELD_ID) String id,
             @JsonProperty(FIELD_NAME) String name,
-            @JsonProperty(FIELD_OPTIONS) Map<String, String> options) {
+            @JsonProperty(FIELD_OPTIONS) Map<String, String> options,
+            @JsonProperty(FIELD_OWNER) String owner,
+            @JsonProperty(FIELD_CREATED_AT) long createdAt,
+            @JsonProperty(FIELD_CREATED_BY) String createdBy,
+            @JsonProperty(FIELD_UPDATED_AT) long updatedAt,
+            @JsonProperty(FIELD_UPDATED_BY) String updatedBy) {
+        super(owner, createdAt, createdBy, updatedAt, updatedBy);
         this.id = id;
         this.name = name;
         this.options = options;

paimon-core/src/main/java/org/apache/paimon/rest/responses/GetTableResponse.java

@@ -28,7 +28,7 @@
 
 /** Response for getting table. */
 @JsonIgnoreProperties(ignoreUnknown = true)
-public class GetTableResponse implements RESTResponse {
+public class GetTableResponse extends BaseResourceAuditResponse implements RESTResponse {
 
     private static final String FIELD_ID = "id";
     private static final String FIELD_NAME = "name";
@@ -57,7 +57,13 @@ public GetTableResponse(
             @JsonProperty(FIELD_NAME) String name,
             @JsonProperty(FIELD_IS_EXTERNAL) boolean isExternal,
             @JsonProperty(FIELD_SCHEMA_ID) long schemaId,
-            @JsonProperty(FIELD_SCHEMA) Schema schema) {
+            @JsonProperty(FIELD_SCHEMA) Schema schema,
+            @JsonProperty(FIELD_OWNER) String owner,
+            @JsonProperty(FIELD_CREATED_AT) long createdAt,
+            @JsonProperty(FIELD_CREATED_BY) String createdBy,
+            @JsonProperty(FIELD_UPDATED_AT) long updatedAt,
+            @JsonProperty(FIELD_UPDATED_BY) String updatedBy) {
+        super(owner, createdAt, createdBy, updatedAt, updatedBy);
         this.id = id;
         this.name = name;
         this.isExternal = isExternal;

paimon-core/src/main/java/org/apache/paimon/rest/responses/GetViewResponse.java

@@ -28,7 +28,7 @@
 
 /** Response for getting view. */
 @JsonIgnoreProperties(ignoreUnknown = true)
-public class GetViewResponse implements RESTResponse {
+public class GetViewResponse extends BaseResourceAuditResponse implements RESTResponse {
 
     private static final String FIELD_ID = "id";
     private static final String FIELD_NAME = "name";
@@ -47,7 +47,13 @@ public class GetViewResponse implements RESTResponse {
     public GetViewResponse(
             @JsonProperty(FIELD_ID) String id,
             @JsonProperty(FIELD_NAME) String name,
-            @JsonProperty(FIELD_SCHEMA) ViewSchema schema) {
+            @JsonProperty(FIELD_SCHEMA) ViewSchema schema,
+            @JsonProperty(FIELD_OWNER) String owner,
+            @JsonProperty(FIELD_CREATED_AT) long createdAt,
+            @JsonProperty(FIELD_CREATED_BY) String createdBy,
+            @JsonProperty(FIELD_UPDATED_AT) long updatedAt,
+            @JsonProperty(FIELD_UPDATED_BY) String updatedBy) {
+        super(owner, createdAt, createdBy, updatedAt, updatedBy);
         this.id = id;
         this.name = name;
         this.schema = schema;

paimon-core/src/test/java/org/apache/paimon/catalog/CatalogTestBase.java

@@ -36,12 +36,15 @@
 import org.apache.paimon.table.sink.BatchTableCommit;
 import org.apache.paimon.table.sink.BatchTableWrite;
 import org.apache.paimon.table.sink.BatchWriteBuilder;
+import org.apache.paimon.table.system.AllTableOptionsTable;
+import org.apache.paimon.table.system.CatalogOptionsTable;
 import org.apache.paimon.types.DataField;
 import org.apache.paimon.types.DataTypes;
 import org.apache.paimon.types.RowType;
 import org.apache.paimon.view.View;
 import org.apache.paimon.view.ViewImpl;
 
+import org.apache.paimon.shade.guava30.com.google.common.collect.ImmutableMap;
 import org.apache.paimon.shade.guava30.com.google.common.collect.Lists;
 import org.apache.paimon.shade.guava30.com.google.common.collect.Maps;
 
@@ -483,6 +486,20 @@ public void testCreateTable() throws Exception {
                 .hasRootCauseInstanceOf(IllegalArgumentException.class)
                 .hasRootCauseMessage(
                         "Unrecognized option for boolean: max. Expected either true or false(case insensitive)");
+
+        // conflict options
+        Schema conflictOptionsSchema =
+                Schema.newBuilder()
+                        .column("a", DataTypes.INT())
+                        .options(ImmutableMap.of("changelog-producer", "input"))
+                        .build();
+        assertThatThrownBy(
+                        () ->
+                                catalog.createTable(
+                                        Identifier.create("test_db", "conflict_options_table"),
+                                        conflictOptionsSchema,
+                                        false))
+                .isInstanceOf(RuntimeException.class);
     }
 
     @Test
@@ -538,6 +555,12 @@ public void testGetTable() throws Exception {
         assertThatExceptionOfType(Catalog.TableNotExistException.class)
                 .isThrownBy(
                         () -> catalog.getTable(Identifier.create(SYSTEM_DATABASE_NAME, "1111")));
+
+        List<String> sysTables = catalog.listTables(SYSTEM_DATABASE_NAME);
+        assertThat(sysTables)
+                .containsExactlyInAnyOrder(
+                        AllTableOptionsTable.ALL_TABLE_OPTIONS,
+                        CatalogOptionsTable.CATALOG_OPTIONS);
     }
 
     @Test
@@ -683,6 +706,19 @@ public void testAlterTable() throws Exception {
                         anyCauseMatches(
                                 Catalog.ColumnAlreadyExistException.class,
                                 "Column col1 already exists in the test_db.test_table table."));
+
+        // conflict options
+        assertThatThrownBy(
+                        () ->
+                                catalog.alterTable(
+                                        identifier,
+                                        Lists.newArrayList(
+                                                SchemaChange.setOption(
+                                                        "changelog-producer", "input")),
+                                        false))
+                .isInstanceOf(RuntimeException.class)
+                .hasMessageContaining(
+                        "Can not set changelog-producer on table without primary keys");
     }
 
     @Test

paimon-core/src/test/java/org/apache/paimon/rest/MockRESTMessage.java

@@ -82,7 +82,15 @@ public static GetDatabaseResponse getDatabaseResponse(String name) {
         Map<String, String> options = new HashMap<>();
         options.put("a", "b");
         options.put(COMMENT_PROP, "comment");
-        return new GetDatabaseResponse(UUID.randomUUID().toString(), name, options);
+        return new GetDatabaseResponse(
+                UUID.randomUUID().toString(),
+                name,
+                options,
+                "owner",
+                System.currentTimeMillis(),
+                "created",
+                System.currentTimeMillis(),
+                "updated");
     }
 
     public static ListDatabasesResponse listDatabasesResponse(String name) {
@@ -219,7 +227,17 @@ public static GetTableResponse getTableResponse() {
         Map<String, String> options = new HashMap<>();
         options.put("option-1", "value-1");
         options.put("option-2", "value-2");
-        return new GetTableResponse(UUID.randomUUID().toString(), "", false, 1, schema(options));
+        return new GetTableResponse(
+                UUID.randomUUID().toString(),
+                "",
+                false,
+                1,
+                schema(options),
+                "owner",
+                System.currentTimeMillis(),
+                "created",
+                System.currentTimeMillis(),
+                "updated");
     }
 
     public static CreateViewRequest createViewRequest(String name) {
@@ -228,7 +246,15 @@ public static CreateViewRequest createViewRequest(String name) {
     }
 
     public static GetViewResponse getViewResponse() {
-        return new GetViewResponse(UUID.randomUUID().toString(), "", viewSchema());
+        return new GetViewResponse(
+                UUID.randomUUID().toString(),
+                "",
+                viewSchema(),
+                "owner",
+                System.currentTimeMillis(),
+                "created",
+                System.currentTimeMillis(),
+                "updated");
     }
 
     public static ListViewsResponse listViewsResponse() {