This is an automated email from the ASF dual-hosted git repository.
zhoujinsong pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/amoro.git
The following commit(s) were added to refs/heads/master by this push:
new da51a2d58 [AMORO-4240][AMS] Support namespace properties in REST
catalog (#4283)
da51a2d58 is described below
commit da51a2d582deedab65b36c9c831e3a3f566caab0
Author: Power John <[email protected]>
AuthorDate: Thu Jul 23 14:15:19 2026 +0800
[AMORO-4240][AMS] Support namespace properties in REST catalog (#4283)
---
.../apache/amoro/server/RestCatalogService.java | 53 ++++++-
.../amoro/server/catalog/DatabaseMetadata.java | 52 +++++++
.../amoro/server/catalog/InternalCatalog.java | 58 ++++++-
.../amoro/server/catalog/InternalCatalogImpl.java | 7 +-
.../server/persistence/mapper/TableMetaMapper.java | 48 +++++-
.../table/internal/InternalIcebergCreator.java | 27 +++-
.../internal/InternalMixedIcebergCreator.java | 9 ++
.../src/main/resources/derby/ams-derby-init.sql | 3 +-
amoro-ams/src/main/resources/derby/upgrade.sql | 16 ++
.../src/main/resources/mysql/ams-mysql-init.sql | 1 +
amoro-ams/src/main/resources/mysql/upgrade.sql | 3 +
.../main/resources/postgres/ams-postgres-init.sql | 4 +-
amoro-ams/src/main/resources/postgres/upgrade.sql | 6 +-
.../server/TestInternalIcebergCatalogService.java | 126 ++++++++++++++++
.../amoro/server/TestRestCatalogService.java | 43 ++++++
.../amoro/server/persistence/TestDerbyUpgrade.java | 80 ++++++++++
.../persistence/mapper/TestTableMetaMapper.java | 166 +++++++++++++++++++++
17 files changed, 682 insertions(+), 20 deletions(-)
diff --git
a/amoro-ams/src/main/java/org/apache/amoro/server/RestCatalogService.java
b/amoro-ams/src/main/java/org/apache/amoro/server/RestCatalogService.java
index aa342beb9..d197021a8 100644
--- a/amoro-ams/src/main/java/org/apache/amoro/server/RestCatalogService.java
+++ b/amoro-ams/src/main/java/org/apache/amoro/server/RestCatalogService.java
@@ -60,15 +60,18 @@ import org.apache.iceberg.TableOperations;
import org.apache.iceberg.catalog.Namespace;
import org.apache.iceberg.catalog.TableIdentifier;
import org.apache.iceberg.exceptions.AlreadyExistsException;
+import org.apache.iceberg.exceptions.BadRequestException;
import org.apache.iceberg.exceptions.CommitFailedException;
import org.apache.iceberg.exceptions.NoSuchNamespaceException;
import org.apache.iceberg.exceptions.NoSuchTableException;
+import org.apache.iceberg.exceptions.UnprocessableEntityException;
import org.apache.iceberg.rest.RESTResponse;
import org.apache.iceberg.rest.RESTSerializers;
import org.apache.iceberg.rest.requests.CreateNamespaceRequest;
import org.apache.iceberg.rest.requests.CreateTableRequest;
import org.apache.iceberg.rest.requests.ReportMetricsRequest;
import org.apache.iceberg.rest.requests.ReportMetricsRequestParser;
+import org.apache.iceberg.rest.requests.UpdateNamespacePropertiesRequest;
import org.apache.iceberg.rest.requests.UpdateTableRequest;
import org.apache.iceberg.rest.responses.ConfigResponse;
import org.apache.iceberg.rest.responses.CreateNamespaceResponse;
@@ -77,6 +80,7 @@ import org.apache.iceberg.rest.responses.GetNamespaceResponse;
import org.apache.iceberg.rest.responses.ListNamespacesResponse;
import org.apache.iceberg.rest.responses.ListTablesResponse;
import org.apache.iceberg.rest.responses.LoadTableResponse;
+import org.apache.iceberg.rest.responses.UpdateNamespacePropertiesResponse;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -131,7 +135,9 @@ public class RestCatalogService extends PersistentBase
implements RestExtension
post("/v1/catalogs/{catalog}/namespaces", this::createNamespace);
get("/v1/catalogs/{catalog}/namespaces/{namespace}",
this::getNamespace);
delete("/v1/catalogs/{catalog}/namespaces/{namespace}",
this::dropNamespace);
- post("/v1/catalogs/{catalog}/namespaces/{namespace}",
this::setNamespaceProperties);
+ post(
+ "/v1/catalogs/{catalog}/namespaces/{namespace}/properties",
+ this::setNamespaceProperties);
get(
"/v1/catalogs/{catalog}/namespaces/{namespace}/tables",
this::listTablesInNamespace);
@@ -229,12 +235,16 @@ public class RestCatalogService extends PersistentBase
implements RestExtension
ctx,
catalog -> {
CreateNamespaceRequest request = bodyAsClass(ctx,
CreateNamespaceRequest.class);
+ request.validate();
Namespace ns = request.namespace();
checkUnsupported(ns.length() == 1, "multi-level namespace is not
supported now");
String database = ns.level(0);
checkAlreadyExists(!catalog.databaseExists(database), "Database",
database);
- catalog.createDatabase(database);
- return
CreateNamespaceResponse.builder().withNamespace(Namespace.of(database)).build();
+ catalog.createDatabase(database, request.properties());
+ return CreateNamespaceResponse.builder()
+ .withNamespace(Namespace.of(database))
+ .setProperties(catalog.getDatabaseProperties(database))
+ .build();
});
}
@@ -243,7 +253,10 @@ public class RestCatalogService extends PersistentBase
implements RestExtension
handleNamespace(
ctx,
(catalog, database) ->
-
GetNamespaceResponse.builder().withNamespace(Namespace.of(database)).build());
+ GetNamespaceResponse.builder()
+ .withNamespace(Namespace.of(database))
+ .setProperties(catalog.getDatabaseProperties(database))
+ .build());
}
/** DELETE PREFIX/v1/catalogs/{catalog}/namespaces/{namespace} */
@@ -258,7 +271,30 @@ public class RestCatalogService extends PersistentBase
implements RestExtension
/** POST PREFIX/v1/catalogs/{catalog}/namespaces/{namespace}/properties */
public void setNamespaceProperties(Context ctx) {
- throw new UnsupportedOperationException("namespace properties is not
supported");
+ handleNamespace(
+ ctx,
+ (catalog, database) -> {
+ UpdateNamespacePropertiesRequest request =
+ bodyAsClass(ctx, UpdateNamespacePropertiesRequest.class);
+ request.validate();
+
+ Map<String, String> properties =
+ catalog.applyDatabasePropertiesUpdate(
+ database, request.updates(), request.removals());
+ List<String> removed =
+ request.removals().stream()
+ .filter(properties::containsKey)
+ .collect(Collectors.toList());
+ List<String> missing =
+ request.removals().stream()
+ .filter(key -> !properties.containsKey(key))
+ .collect(Collectors.toList());
+ return UpdateNamespacePropertiesResponse.builder()
+ .addUpdated(request.updates().keySet())
+ .addRemoved(removed)
+ .addMissing(missing)
+ .build();
+ });
}
/** GET PREFIX/v1/catalogs/{catalog}/namespaces/{namespace}/tables */
@@ -488,6 +524,7 @@ public class RestCatalogService extends PersistentBase
implements RestExtension
AuthenticationTimeout(419),
NotFound(404),
Conflict(409),
+ UnprocessableEntity(422),
InternalServerError(500),
ServiceUnavailable(503);
public final int code;
@@ -497,7 +534,11 @@ public class RestCatalogService extends PersistentBase
implements RestExtension
}
public static IcebergRestErrorCode exceptionToCode(Exception e) {
- if (e instanceof UnsupportedOperationException) {
+ if (e instanceof BadRequestException || e instanceof
IllegalArgumentException) {
+ return BadRequest;
+ } else if (e instanceof UnprocessableEntityException) {
+ return UnprocessableEntity;
+ } else if (e instanceof UnsupportedOperationException) {
return UnsupportedOperation;
} else if (e instanceof ObjectNotExistsException) {
return NotFound;
diff --git
a/amoro-ams/src/main/java/org/apache/amoro/server/catalog/DatabaseMetadata.java
b/amoro-ams/src/main/java/org/apache/amoro/server/catalog/DatabaseMetadata.java
new file mode 100644
index 000000000..a35410f18
--- /dev/null
+++
b/amoro-ams/src/main/java/org/apache/amoro/server/catalog/DatabaseMetadata.java
@@ -0,0 +1,52 @@
+/*
+ * 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.amoro.server.catalog;
+
+import java.util.Map;
+
+public class DatabaseMetadata {
+
+ private String catalogName;
+ private String databaseName;
+ private Map<String, String> properties;
+
+ public String getCatalogName() {
+ return catalogName;
+ }
+
+ public void setCatalogName(String catalogName) {
+ this.catalogName = catalogName;
+ }
+
+ public String getDatabaseName() {
+ return databaseName;
+ }
+
+ public void setDatabaseName(String databaseName) {
+ this.databaseName = databaseName;
+ }
+
+ public Map<String, String> getProperties() {
+ return properties;
+ }
+
+ public void setProperties(Map<String, String> properties) {
+ this.properties = properties;
+ }
+}
diff --git
a/amoro-ams/src/main/java/org/apache/amoro/server/catalog/InternalCatalog.java
b/amoro-ams/src/main/java/org/apache/amoro/server/catalog/InternalCatalog.java
index d6a6bd989..a85ef80b5 100644
---
a/amoro-ams/src/main/java/org/apache/amoro/server/catalog/InternalCatalog.java
+++
b/amoro-ams/src/main/java/org/apache/amoro/server/catalog/InternalCatalog.java
@@ -34,7 +34,10 @@ import
org.apache.amoro.server.table.internal.InternalTableHandler;
import org.apache.amoro.table.TableIdentifier;
import org.apache.iceberg.rest.requests.CreateTableRequest;
+import java.util.Collections;
+import java.util.HashMap;
import java.util.List;
+import java.util.Map;
import java.util.Optional;
import java.util.stream.Collectors;
@@ -56,6 +59,12 @@ public abstract class InternalCatalog extends ServerCatalog {
}
public void createDatabase(String databaseName) {
+ createDatabase(databaseName, Collections.emptyMap());
+ }
+
+ public void createDatabase(String databaseName, Map<String, String>
properties) {
+ Map<String, String> databaseProperties =
+ properties == null ? Collections.emptyMap() : new
HashMap<>(properties);
if (!databaseExists(databaseName)) {
doAsTransaction(
// make sure catalog existed in database
@@ -67,13 +76,60 @@ public abstract class InternalCatalog extends ServerCatalog
{
() ->
doAs(
TableMetaMapper.class,
- mapper ->
mapper.insertDatabase(getMetadata().getCatalogName(), databaseName)),
+ mapper ->
+ mapper.insertDatabase(
+ getMetadata().getCatalogName(), databaseName,
databaseProperties)),
() -> createDatabaseInternal(databaseName));
} else {
throw new AlreadyExistsException("Database " + databaseName);
}
}
+ public Map<String, String> getDatabaseProperties(String databaseName) {
+ DatabaseMetadata databaseMetadata =
+ getAs(TableMetaMapper.class, mapper ->
mapper.selectDatabaseMetadata(name(), databaseName));
+ if (databaseMetadata == null) {
+ throw new ObjectNotExistsException(getDatabaseDesc(databaseName));
+ }
+
+ Map<String, String> properties = databaseMetadata.getProperties();
+ return properties == null ? new HashMap<>() : new HashMap<>(properties);
+ }
+
+ public Map<String, String> applyDatabasePropertiesUpdate(
+ String databaseName, Map<String, String> updates, List<String> removals)
{
+ Map<String, String> requestedUpdates =
+ updates == null ? Collections.emptyMap() : new HashMap<>(updates);
+ List<String> requestedRemovals =
+ removals == null ? Collections.emptyList() : List.copyOf(removals);
+ Map<String, String> previousProperties = new HashMap<>();
+
+ doAsTransaction(
+ () -> {
+ DatabaseMetadata metadata =
+ getAs(
+ TableMetaMapper.class,
+ mapper -> mapper.selectDatabaseMetadataForUpdate(name(),
databaseName));
+ if (metadata == null) {
+ throw new ObjectNotExistsException(getDatabaseDesc(databaseName));
+ }
+
+ Map<String, String> properties =
+ metadata.getProperties() == null
+ ? new HashMap<>()
+ : new HashMap<>(metadata.getProperties());
+ previousProperties.putAll(properties);
+ requestedRemovals.forEach(properties::remove);
+ properties.putAll(requestedUpdates);
+ doAsExisted(
+ TableMetaMapper.class,
+ mapper -> mapper.updateDatabaseProperties(name(), databaseName,
properties),
+ () -> new
ObjectNotExistsException(getDatabaseDesc(databaseName)));
+ });
+
+ return previousProperties;
+ }
+
public void dropDatabase(String databaseName) {
if (databaseExists(databaseName)) {
doAsTransaction(
diff --git
a/amoro-ams/src/main/java/org/apache/amoro/server/catalog/InternalCatalogImpl.java
b/amoro-ams/src/main/java/org/apache/amoro/server/catalog/InternalCatalogImpl.java
index 70e52a60b..5310fb432 100644
---
a/amoro-ams/src/main/java/org/apache/amoro/server/catalog/InternalCatalogImpl.java
+++
b/amoro-ams/src/main/java/org/apache/amoro/server/catalog/InternalCatalogImpl.java
@@ -193,10 +193,13 @@ public class InternalCatalogImpl extends InternalCatalog {
throw new AlreadyExistsException(
"Table " + name() + "." + database + "." + tableName + " already
exists.");
}
+ String namespaceLocation = getDatabaseProperties(database).get("location");
if (TableFormat.ICEBERG.equals(format)) {
- return new InternalIcebergCreator(getMetadata(), database, tableName,
creatorArguments);
+ return new InternalIcebergCreator(
+ getMetadata(), database, tableName, creatorArguments,
namespaceLocation);
} else if (TableFormat.MIXED_ICEBERG.equals(format)) {
- return new InternalMixedIcebergCreator(getMetadata(), database,
tableName, creatorArguments);
+ return new InternalMixedIcebergCreator(
+ getMetadata(), database, tableName, creatorArguments,
namespaceLocation);
} else {
throw new IllegalArgumentException("Unsupported table format:" + format);
}
diff --git
a/amoro-ams/src/main/java/org/apache/amoro/server/persistence/mapper/TableMetaMapper.java
b/amoro-ams/src/main/java/org/apache/amoro/server/persistence/mapper/TableMetaMapper.java
index 0dae33222..2ad5a1ad3 100644
---
a/amoro-ams/src/main/java/org/apache/amoro/server/persistence/mapper/TableMetaMapper.java
+++
b/amoro-ams/src/main/java/org/apache/amoro/server/persistence/mapper/TableMetaMapper.java
@@ -19,6 +19,7 @@
package org.apache.amoro.server.persistence.mapper;
import org.apache.amoro.ServerTableIdentifier;
+import org.apache.amoro.server.catalog.DatabaseMetadata;
import org.apache.amoro.server.persistence.converter.Map2StringConverter;
import org.apache.amoro.server.persistence.converter.TableFormatConverter;
import
org.apache.amoro.server.persistence.extension.InListExtendedLanguageDriver;
@@ -29,16 +30,24 @@ import org.apache.ibatis.annotations.Lang;
import org.apache.ibatis.annotations.Options;
import org.apache.ibatis.annotations.Param;
import org.apache.ibatis.annotations.Result;
+import org.apache.ibatis.annotations.ResultMap;
import org.apache.ibatis.annotations.Results;
import org.apache.ibatis.annotations.Select;
import org.apache.ibatis.annotations.Update;
import java.util.List;
+import java.util.Map;
public interface TableMetaMapper {
- @Insert("INSERT INTO database_metadata(catalog_name, db_name) VALUES(
#{catalogName}, #{dbName})")
- void insertDatabase(@Param("catalogName") String catalogName,
@Param("dbName") String dbName);
+ @Insert(
+ "INSERT INTO database_metadata(catalog_name, db_name, properties)"
+ + " VALUES(#{catalogName}, #{dbName},"
+ + " #{properties,
typeHandler=org.apache.amoro.server.persistence.converter.Map2StringConverter})")
+ void insertDatabase(
+ @Param("catalogName") String catalogName,
+ @Param("dbName") String dbName,
+ @Param("properties") Map<String, String> properties);
@Select("SELECT db_name FROM database_metadata WHERE catalog_name =
#{catalogName}")
List<String> selectDatabases(@Param("catalogName") String catalogName);
@@ -47,6 +56,41 @@ public interface TableMetaMapper {
"SELECT db_name FROM database_metadata WHERE catalog_name =
#{catalogName} AND db_name=#{dbName}")
String selectDatabase(@Param("catalogName") String catalogName,
@Param("dbName") String dbName);
+ @Select(
+ "SELECT catalog_name, db_name, properties FROM database_metadata"
+ + " WHERE catalog_name = #{catalogName} AND db_name = #{dbName}")
+ @Results(
+ id = "databaseMetadata",
+ value = {
+ @Result(property = "catalogName", column = "catalog_name"),
+ @Result(property = "databaseName", column = "db_name"),
+ @Result(
+ property = "properties",
+ column = "properties",
+ typeHandler = Map2StringConverter.class)
+ })
+ DatabaseMetadata selectDatabaseMetadata(
+ @Param("catalogName") String catalogName, @Param("dbName") String
dbName);
+
+ @Select(
+ "<script>"
+ + "SELECT catalog_name, db_name, properties FROM database_metadata"
+ + " WHERE catalog_name = #{catalogName} AND db_name = #{dbName} FOR
UPDATE"
+ + "<if test=\"_databaseId == 'derby'\"> WITH RS</if>"
+ + "</script>")
+ @ResultMap("databaseMetadata")
+ DatabaseMetadata selectDatabaseMetadataForUpdate(
+ @Param("catalogName") String catalogName, @Param("dbName") String
dbName);
+
+ @Update(
+ "UPDATE database_metadata SET properties ="
+ + " #{properties,
typeHandler=org.apache.amoro.server.persistence.converter.Map2StringConverter}"
+ + " WHERE catalog_name = #{catalogName} AND db_name = #{dbName}")
+ Integer updateDatabaseProperties(
+ @Param("catalogName") String catalogName,
+ @Param("dbName") String dbName,
+ @Param("properties") Map<String, String> properties);
+
@Delete(
"DELETE FROM database_metadata WHERE catalog_name = #{catalogName} AND
db_name = #{dbName}"
+ " AND table_count = 0")
diff --git
a/amoro-ams/src/main/java/org/apache/amoro/server/table/internal/InternalIcebergCreator.java
b/amoro-ams/src/main/java/org/apache/amoro/server/table/internal/InternalIcebergCreator.java
index 90255a30f..31f3c2389 100644
---
a/amoro-ams/src/main/java/org/apache/amoro/server/table/internal/InternalIcebergCreator.java
+++
b/amoro-ams/src/main/java/org/apache/amoro/server/table/internal/InternalIcebergCreator.java
@@ -42,6 +42,7 @@ public class InternalIcebergCreator implements
InternalTableCreator {
protected final AuthenticatedFileIO io;
protected final CreateTableRequest request;
private final CatalogMeta catalogMeta;
+ private final String namespaceLocation;
protected final String database;
protected final String tableName;
private boolean closed = false;
@@ -51,8 +52,18 @@ public class InternalIcebergCreator implements
InternalTableCreator {
public InternalIcebergCreator(
CatalogMeta catalog, String database, String tableName,
CreateTableRequest request) {
+ this(catalog, database, tableName, request, null);
+ }
+
+ public InternalIcebergCreator(
+ CatalogMeta catalog,
+ String database,
+ String tableName,
+ CreateTableRequest request,
+ String namespaceLocation) {
this.io = InternalTableUtil.newIcebergFileIo(catalog);
this.catalogMeta = catalog;
+ this.namespaceLocation = namespaceLocation;
this.database = database;
this.tableName = tableName;
this.request = request;
@@ -127,12 +138,16 @@ public class InternalIcebergCreator implements
InternalTableCreator {
private String tableLocation() {
String location = this.request.location();
if (StringUtils.isBlank(location)) {
- String warehouse =
-
catalogMeta.getCatalogProperties().get(CatalogMetaProperties.KEY_WAREHOUSE);
- Preconditions.checkState(
- StringUtils.isNotBlank(warehouse), "catalog warehouse is not
configured");
- warehouse = LocationUtil.stripTrailingSlash(warehouse);
- location = warehouse + "/" + database + "/" + tableName;
+ if (StringUtils.isNotBlank(namespaceLocation)) {
+ location = LocationUtil.stripTrailingSlash(namespaceLocation) + "/" +
tableName;
+ } else {
+ String warehouse =
+
catalogMeta.getCatalogProperties().get(CatalogMetaProperties.KEY_WAREHOUSE);
+ Preconditions.checkState(
+ StringUtils.isNotBlank(warehouse), "catalog warehouse is not
configured");
+ warehouse = LocationUtil.stripTrailingSlash(warehouse);
+ location = warehouse + "/" + database + "/" + tableName;
+ }
} else {
location = LocationUtil.stripTrailingSlash(location);
}
diff --git
a/amoro-ams/src/main/java/org/apache/amoro/server/table/internal/InternalMixedIcebergCreator.java
b/amoro-ams/src/main/java/org/apache/amoro/server/table/internal/InternalMixedIcebergCreator.java
index 4953f27ca..34a58ad16 100644
---
a/amoro-ams/src/main/java/org/apache/amoro/server/table/internal/InternalMixedIcebergCreator.java
+++
b/amoro-ams/src/main/java/org/apache/amoro/server/table/internal/InternalMixedIcebergCreator.java
@@ -44,6 +44,15 @@ public class InternalMixedIcebergCreator extends
InternalIcebergCreator {
super(catalog, database, tableName, request);
}
+ public InternalMixedIcebergCreator(
+ CatalogMeta catalog,
+ String database,
+ String tableName,
+ CreateTableRequest request,
+ String namespaceLocation) {
+ super(catalog, database, tableName, request, namespaceLocation);
+ }
+
@Override
protected TableFormat format() {
return TableFormat.MIXED_ICEBERG;
diff --git a/amoro-ams/src/main/resources/derby/ams-derby-init.sql
b/amoro-ams/src/main/resources/derby/ams-derby-init.sql
index 32d663288..72c43f4d8 100644
--- a/amoro-ams/src/main/resources/derby/ams-derby-init.sql
+++ b/amoro-ams/src/main/resources/derby/ams-derby-init.sql
@@ -29,6 +29,7 @@ CREATE TABLE catalog_metadata (
CREATE TABLE database_metadata (
catalog_name VARCHAR(64) NOT NULL,
db_name VARCHAR(128) NOT NULL,
+ properties CLOB(64m),
table_count INT NOT NULL DEFAULT 0,
PRIMARY KEY (catalog_name, db_name)
);
@@ -274,4 +275,4 @@ CREATE TABLE bucket_assignments (
last_update_time BIGINT NOT NULL DEFAULT 0,
node_heartbeat_ts BIGINT NOT NULL DEFAULT 0,
PRIMARY KEY (cluster_name, node_key)
-);
\ No newline at end of file
+);
diff --git a/amoro-ams/src/main/resources/derby/upgrade.sql
b/amoro-ams/src/main/resources/derby/upgrade.sql
new file mode 100644
index 000000000..0282c32ec
--- /dev/null
+++ b/amoro-ams/src/main/resources/derby/upgrade.sql
@@ -0,0 +1,16 @@
+-- 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.
+
+ALTER TABLE database_metadata ADD COLUMN properties CLOB(64m);
diff --git a/amoro-ams/src/main/resources/mysql/ams-mysql-init.sql
b/amoro-ams/src/main/resources/mysql/ams-mysql-init.sql
index f202695c8..b7c423776 100644
--- a/amoro-ams/src/main/resources/mysql/ams-mysql-init.sql
+++ b/amoro-ams/src/main/resources/mysql/ams-mysql-init.sql
@@ -31,6 +31,7 @@ CREATE TABLE `database_metadata`
(
`catalog_name` varchar(64) NOT NULL COMMENT 'catalog name',
`db_name` varchar(128) NOT NULL COMMENT 'database name',
+ `properties` mediumtext COMMENT 'Database properties',
`table_count` int(11) NOT NULL default 0,
PRIMARY KEY (`catalog_name`, `db_name`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT 'database metadata';
diff --git a/amoro-ams/src/main/resources/mysql/upgrade.sql
b/amoro-ams/src/main/resources/mysql/upgrade.sql
index c67807751..e8752b1ca 100644
--- a/amoro-ams/src/main/resources/mysql/upgrade.sql
+++ b/amoro-ams/src/main/resources/mysql/upgrade.sql
@@ -175,3 +175,6 @@ CREATE TABLE IF NOT EXISTS bucket_assignments (
-- ADD node_heartbeat_ts to table bucket_assignments
ALTER TABLE `bucket_assignments` ADD COLUMN `node_heartbeat_ts` BIGINT NOT
NULL DEFAULT 0 COMMENT 'Per-node heartbeat timestamp updated only by the owning
node (ms since epoch)';
+-- ADD properties to table database_metadata
+ALTER TABLE `database_metadata` ADD COLUMN `properties` MEDIUMTEXT COMMENT
'Database properties';
+
diff --git a/amoro-ams/src/main/resources/postgres/ams-postgres-init.sql
b/amoro-ams/src/main/resources/postgres/ams-postgres-init.sql
index 245e37744..5f1494bc0 100644
--- a/amoro-ams/src/main/resources/postgres/ams-postgres-init.sql
+++ b/amoro-ams/src/main/resources/postgres/ams-postgres-init.sql
@@ -38,12 +38,14 @@ CREATE TABLE database_metadata
(
catalog_name varchar(64) NOT NULL,
db_name varchar(128) NOT NULL,
+ properties text,
table_count integer NOT NULL DEFAULT 0,
PRIMARY KEY (catalog_name, db_name)
);
COMMENT ON TABLE database_metadata IS 'Database metadata';
COMMENT ON COLUMN database_metadata.catalog_name IS 'Catalog name';
COMMENT ON COLUMN database_metadata.db_name IS 'Database name';
+COMMENT ON COLUMN database_metadata.properties IS 'Database properties';
COMMENT ON COLUMN database_metadata.table_count IS 'Number of tables in the
database';
CREATE TABLE optimizer
@@ -474,4 +476,4 @@ COMMENT ON COLUMN node_ip IS 'Node IP address';
COMMENT ON COLUMN server_info_json IS 'JSON encoded server info
(AmsServerInfo)';
COMMENT ON COLUMN lease_expire_ts IS 'Lease expiration timestamp (ms since
epoch)';
COMMENT ON COLUMN version IS 'Optimistic lock version of the lease row';
-COMMENT ON COLUMN updated_at IS 'Last update timestamp (ms since epoch)';
\ No newline at end of file
+COMMENT ON COLUMN updated_at IS 'Last update timestamp (ms since epoch)';
diff --git a/amoro-ams/src/main/resources/postgres/upgrade.sql
b/amoro-ams/src/main/resources/postgres/upgrade.sql
index 8f26023f0..8c290769e 100644
--- a/amoro-ams/src/main/resources/postgres/upgrade.sql
+++ b/amoro-ams/src/main/resources/postgres/upgrade.sql
@@ -235,4 +235,8 @@ CREATE TABLE IF NOT EXISTS bucket_assignments (
CREATE INDEX IF NOT EXISTS idx_bucket_assignments_cluster ON
bucket_assignments (cluster_name);
-- ADD node_heartbeat_ts to table bucket_assignments
-ALTER TABLE bucket_assignments ADD COLUMN node_heartbeat_ts BIGINT NOT NULL
DEFAULT 0;
\ No newline at end of file
+ALTER TABLE bucket_assignments ADD COLUMN node_heartbeat_ts BIGINT NOT NULL
DEFAULT 0;
+
+-- ADD properties to table database_metadata
+ALTER TABLE database_metadata ADD COLUMN properties text;
+COMMENT ON COLUMN database_metadata.properties IS 'Database properties';
diff --git
a/amoro-ams/src/test/java/org/apache/amoro/server/TestInternalIcebergCatalogService.java
b/amoro-ams/src/test/java/org/apache/amoro/server/TestInternalIcebergCatalogService.java
index 19d4ace8a..5fa41d72b 100644
---
a/amoro-ams/src/test/java/org/apache/amoro/server/TestInternalIcebergCatalogService.java
+++
b/amoro-ams/src/test/java/org/apache/amoro/server/TestInternalIcebergCatalogService.java
@@ -24,10 +24,15 @@ import org.apache.amoro.io.IcebergDataTestHelpers;
import org.apache.amoro.io.MixedDataTestHelpers;
import org.apache.amoro.io.reader.GenericUnkeyedDataReader;
import org.apache.amoro.properties.CatalogMetaProperties;
+import org.apache.amoro.server.persistence.SqlSessionFactoryProvider;
import org.apache.amoro.shade.guava32.com.google.common.collect.Lists;
import org.apache.amoro.shade.guava32.com.google.common.collect.Maps;
+import org.apache.amoro.shade.guava32.com.google.common.collect.Sets;
import org.apache.amoro.shade.guava32.com.google.common.collect.Streams;
+import org.apache.amoro.shade.jackson2.com.fasterxml.jackson.databind.JsonNode;
+import
org.apache.amoro.shade.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.amoro.table.MixedTable;
+import org.apache.ibatis.session.SqlSession;
import org.apache.iceberg.AppendFiles;
import org.apache.iceberg.DataFile;
import org.apache.iceberg.FileScanTask;
@@ -50,9 +55,16 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.IOException;
+import java.net.URI;
+import java.net.http.HttpClient;
+import java.net.http.HttpRequest;
+import java.net.http.HttpResponse;
+import java.sql.PreparedStatement;
+import java.sql.SQLException;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
+import java.util.Set;
import java.util.stream.Collectors;
public class TestInternalIcebergCatalogService extends
RestCatalogServiceTestBase {
@@ -102,15 +114,112 @@ public class TestInternalIcebergCatalogService extends
RestCatalogServiceTestBas
@Nested
public class NamespaceTests {
+ @AfterEach
+ public void clean() {
+ if (serverCatalog.databaseExists(database)) {
+ serverCatalog.dropDatabase(database);
+ }
+ }
+
@Test
public void testNamespaceOperations() throws IOException {
Assertions.assertTrue(nsCatalog.listNamespaces().isEmpty());
nsCatalog.createNamespace(Namespace.of(database));
Assertions.assertEquals(1, nsCatalog.listNamespaces().size());
Assertions.assertEquals(0,
nsCatalog.listNamespaces(Namespace.of(database)).size());
+ Assertions.assertTrue(nsCatalog.loadNamespaceMetadata(ns).isEmpty());
nsCatalog.dropNamespace(Namespace.of(database));
Assertions.assertTrue(nsCatalog.listNamespaces().isEmpty());
}
+
+ @Test
+ public void testNamespaceProperties() {
+ Map<String, String> properties = Maps.newHashMap();
+ properties.put("location", "s3://warehouse/test_ns");
+ properties.put("owner", "analytics");
+
+ nsCatalog.createNamespace(ns, properties);
+
+ Assertions.assertEquals(properties, nsCatalog.loadNamespaceMetadata(ns));
+ Assertions.assertEquals(properties,
serverCatalog.getDatabaseProperties(database));
+ }
+
+ @Test
+ public void testUpdateNamespaceProperties() {
+ Map<String, String> properties = Maps.newHashMap();
+ properties.put("location", "s3://warehouse/test_ns");
+ properties.put("owner", "analytics");
+ nsCatalog.createNamespace(ns, properties);
+
+ Map<String, String> updates = Maps.newHashMap();
+ updates.put("owner", "platform");
+ updates.put("retention-days", "30");
+ nsCatalog.setProperties(ns, updates);
+ nsCatalog.removeProperties(ns, Sets.newHashSet("owner", "missing"));
+
+ Map<String, String> expected = Maps.newHashMap();
+ expected.put("location", "s3://warehouse/test_ns");
+ expected.put("retention-days", "30");
+ Assertions.assertEquals(expected, nsCatalog.loadNamespaceMetadata(ns));
+ }
+
+ @Test
+ public void testUpdateNamespacePropertiesResponse() throws IOException,
InterruptedException {
+ Map<String, String> properties = Maps.newHashMap();
+ properties.put("location", "s3://warehouse/test_ns");
+ properties.put("owner", "analytics");
+ nsCatalog.createNamespace(ns, properties);
+
+ String requestBody =
+ "{\"updates\":{\"retention-days\":\"30\"}," +
"\"removals\":[\"owner\",\"missing\"]}";
+ HttpRequest request =
+ HttpRequest.newBuilder(
+ URI.create(
+ ams.getHttpUrl()
+ + restCatalogUri
+ + "/v1/catalogs/"
+ + catalogName()
+ + "/namespaces/"
+ + database
+ + "/properties"))
+ .header("Content-Type", "application/json")
+ .POST(HttpRequest.BodyPublishers.ofString(requestBody))
+ .build();
+ HttpResponse<String> response =
+ HttpClient.newHttpClient().send(request,
HttpResponse.BodyHandlers.ofString());
+
+ Assertions.assertEquals(200, response.statusCode());
+ JsonNode responseBody = new ObjectMapper().readTree(response.body());
+ Assertions.assertEquals(
+ Sets.newHashSet("retention-days"),
stringSet(responseBody.get("updated")));
+ Assertions.assertEquals(Sets.newHashSet("owner"),
stringSet(responseBody.get("removed")));
+ Assertions.assertEquals(Sets.newHashSet("missing"),
stringSet(responseBody.get("missing")));
+ }
+
+ @Test
+ public void testNullNamespaceProperties() throws SQLException {
+ nsCatalog.createNamespace(ns);
+ try (SqlSession session =
SqlSessionFactoryProvider.getInstance().get().openSession(true);
+ PreparedStatement statement =
+ session
+ .getConnection()
+ .prepareStatement(
+ "UPDATE database_metadata SET properties = NULL"
+ + " WHERE catalog_name = ? AND db_name = ?")) {
+ statement.setString(1, catalogName());
+ statement.setString(2, database);
+ Assertions.assertEquals(1, statement.executeUpdate());
+ }
+
+
Assertions.assertTrue(serverCatalog.getDatabaseProperties(database).isEmpty());
+ Assertions.assertTrue(nsCatalog.loadNamespaceMetadata(ns).isEmpty());
+ }
+
+ private Set<String> stringSet(JsonNode values) {
+ Set<String> result = Sets.newHashSet();
+ values.forEach(value -> result.add(value.asText()));
+ return result;
+ }
}
@Nested
@@ -158,6 +267,23 @@ public class TestInternalIcebergCatalogService extends
RestCatalogServiceTestBas
Assertions.assertFalse(nsCatalog.tableExists(identifier));
}
+ @Test
+ public void testCreateTableInNamespaceLocation() {
+ String namespaceLocation =
+ serverCatalog
+ .getMetadata()
+ .getCatalogProperties()
+ .get(CatalogMetaProperties.KEY_WAREHOUSE)
+ + "/namespace-location";
+ Map<String, String> properties = Maps.newHashMap();
+ properties.put("location", namespaceLocation);
+ nsCatalog.setProperties(ns, properties);
+
+ Table created = nsCatalog.createTable(identifier, schema);
+
+ Assertions.assertEquals(namespaceLocation + "/" + table,
created.location());
+ }
+
@Test
public void testTableWriteAndCommit() throws IOException {
Table tbl = nsCatalog.createTable(identifier, schema);
diff --git
a/amoro-ams/src/test/java/org/apache/amoro/server/TestRestCatalogService.java
b/amoro-ams/src/test/java/org/apache/amoro/server/TestRestCatalogService.java
new file mode 100644
index 000000000..26c39dc41
--- /dev/null
+++
b/amoro-ams/src/test/java/org/apache/amoro/server/TestRestCatalogService.java
@@ -0,0 +1,43 @@
+/*
+ * 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.amoro.server;
+
+import org.apache.iceberg.exceptions.BadRequestException;
+import org.apache.iceberg.exceptions.UnprocessableEntityException;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class TestRestCatalogService {
+
+ @Test
+ public void testRequestErrorCodes() {
+ Assertions.assertEquals(
+ RestCatalogService.IcebergRestErrorCode.BadRequest,
+ RestCatalogService.IcebergRestErrorCode.exceptionToCode(
+ new BadRequestException("invalid request")));
+ Assertions.assertEquals(
+ RestCatalogService.IcebergRestErrorCode.BadRequest,
+ RestCatalogService.IcebergRestErrorCode.exceptionToCode(
+ new IllegalArgumentException("invalid argument")));
+ Assertions.assertEquals(
+ RestCatalogService.IcebergRestErrorCode.UnprocessableEntity,
+ RestCatalogService.IcebergRestErrorCode.exceptionToCode(
+ new UnprocessableEntityException("conflicting property changes")));
+ }
+}
diff --git
a/amoro-ams/src/test/java/org/apache/amoro/server/persistence/TestDerbyUpgrade.java
b/amoro-ams/src/test/java/org/apache/amoro/server/persistence/TestDerbyUpgrade.java
new file mode 100644
index 000000000..ea3d23044
--- /dev/null
+++
b/amoro-ams/src/test/java/org/apache/amoro/server/persistence/TestDerbyUpgrade.java
@@ -0,0 +1,80 @@
+/*
+ * 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.amoro.server.persistence;
+
+import org.apache.ibatis.jdbc.ScriptRunner;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.io.InputStream;
+import java.io.InputStreamReader;
+import java.nio.charset.StandardCharsets;
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.UUID;
+
+public class TestDerbyUpgrade {
+
+ @Test
+ public void testAddDatabasePropertiesColumn() throws Exception {
+ String databaseName = "upgrade_" +
UUID.randomUUID().toString().replace("-", "");
+ String connectionUrl = "jdbc:derby:memory:" + databaseName +
";create=true";
+
+ try (Connection connection = DriverManager.getConnection(connectionUrl);
+ Statement statement = connection.createStatement()) {
+ statement.execute(
+ "CREATE TABLE database_metadata ("
+ + "catalog_name VARCHAR(64) NOT NULL, "
+ + "db_name VARCHAR(128) NOT NULL, "
+ + "table_count INT NOT NULL DEFAULT 0, "
+ + "PRIMARY KEY (catalog_name, db_name))");
+ statement.execute(
+ "INSERT INTO database_metadata(catalog_name, db_name) VALUES
('catalog', 'database')");
+
+ try (InputStream script =
+
TestDerbyUpgrade.class.getClassLoader().getResourceAsStream("derby/upgrade.sql"))
{
+ Assertions.assertNotNull(script);
+ new ScriptRunner(connection)
+ .runScript(new InputStreamReader(script, StandardCharsets.UTF_8));
+ }
+
+ statement.execute(
+ "UPDATE database_metadata SET properties =
'{\"location\":\"file:/warehouse\"}'"
+ + " WHERE catalog_name = 'catalog' AND db_name = 'database'");
+ try (ResultSet resultSet =
+ statement.executeQuery(
+ "SELECT properties FROM database_metadata"
+ + " WHERE catalog_name = 'catalog' AND db_name =
'database'")) {
+ Assertions.assertTrue(resultSet.next());
+ Assertions.assertEquals("{\"location\":\"file:/warehouse\"}",
resultSet.getString(1));
+ }
+ connection.commit();
+ } finally {
+ try {
+ DriverManager.getConnection("jdbc:derby:memory:" + databaseName +
";drop=true");
+ Assertions.fail("Derby did not drop the database");
+ } catch (SQLException e) {
+ Assertions.assertEquals("08006", e.getSQLState());
+ }
+ }
+ }
+}
diff --git
a/amoro-ams/src/test/java/org/apache/amoro/server/persistence/mapper/TestTableMetaMapper.java
b/amoro-ams/src/test/java/org/apache/amoro/server/persistence/mapper/TestTableMetaMapper.java
new file mode 100644
index 000000000..3333654ee
--- /dev/null
+++
b/amoro-ams/src/test/java/org/apache/amoro/server/persistence/mapper/TestTableMetaMapper.java
@@ -0,0 +1,166 @@
+/*
+ * 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.amoro.server.persistence.mapper;
+
+import org.apache.amoro.config.Configurations;
+import org.apache.amoro.server.AmoroManagementConf;
+import org.apache.amoro.server.catalog.DatabaseMetadata;
+import org.apache.amoro.server.persistence.DataSourceFactory;
+import org.apache.commons.dbcp2.BasicDataSource;
+import org.apache.ibatis.mapping.Environment;
+import org.apache.ibatis.session.SqlSession;
+import org.apache.ibatis.session.SqlSessionFactory;
+import org.apache.ibatis.session.SqlSessionFactoryBuilder;
+import org.apache.ibatis.transaction.TransactionFactory;
+import org.apache.ibatis.transaction.jdbc.JdbcTransactionFactory;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+
+import java.sql.SQLException;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+public class TestTableMetaMapper {
+
+ private static BasicDataSource dataSource;
+ private static SqlSessionFactory sqlSessionFactory;
+
+ @BeforeAll
+ public static void initialize() {
+ Configurations configurations = new Configurations();
+ configurations.set(
+ AmoroManagementConf.DB_CONNECTION_URL,
+ String.format("jdbc:derby:memory:%s;create=true", UUID.randomUUID()));
+ configurations.set(AmoroManagementConf.DB_TYPE,
AmoroManagementConf.DB_TYPE_DERBY);
+ configurations.set(
+ AmoroManagementConf.DB_DRIVER_CLASS_NAME,
"org.apache.derby.jdbc.EmbeddedDriver");
+ dataSource = (BasicDataSource)
DataSourceFactory.createDataSource(configurations);
+
+ TransactionFactory transactionFactory = new JdbcTransactionFactory();
+ Environment environment = new Environment("test", transactionFactory,
dataSource);
+ org.apache.ibatis.session.Configuration configuration =
+ new org.apache.ibatis.session.Configuration(environment);
+ configuration.setDatabaseId("derby");
+ configuration.addMapper(TableMetaMapper.class);
+ sqlSessionFactory = new SqlSessionFactoryBuilder().build(configuration);
+ }
+
+ @AfterAll
+ public static void clean() throws SQLException {
+ dataSource.close();
+ }
+
+ @Test
+ public void testDatabaseProperties() {
+ Map<String, String> properties = new HashMap<>();
+ properties.put("location", "s3://warehouse/test_db");
+ properties.put("owner", "analytics");
+
+ try (SqlSession session = sqlSessionFactory.openSession(true)) {
+ TableMetaMapper mapper = session.getMapper(TableMetaMapper.class);
+ mapper.insertDatabase("test_catalog", "test_db", properties);
+
+ DatabaseMetadata metadata =
mapper.selectDatabaseMetadata("test_catalog", "test_db");
+ Assertions.assertEquals(properties, metadata.getProperties());
+
+ Map<String, String> updatedProperties = new HashMap<>();
+ updatedProperties.put("location", "s3://warehouse/test_db");
+ updatedProperties.put("retention-days", "30");
+ Assertions.assertEquals(
+ Integer.valueOf(1),
+ mapper.updateDatabaseProperties("test_catalog", "test_db",
updatedProperties));
+ Assertions.assertEquals(
+ updatedProperties,
+ mapper.selectDatabaseMetadata("test_catalog",
"test_db").getProperties());
+ }
+ }
+
+ @Test
+ public void testDatabasePropertiesUpdateIsSerialized() throws Exception {
+ Map<String, String> properties = new HashMap<>();
+ properties.put("base", "value");
+ try (SqlSession session = sqlSessionFactory.openSession(true)) {
+ session
+ .getMapper(TableMetaMapper.class)
+ .insertDatabase("lock_catalog", "lock_db", properties);
+ }
+
+ CountDownLatch secondUpdateStarted = new CountDownLatch(1);
+ ExecutorService executor = Executors.newSingleThreadExecutor();
+ try (SqlSession firstSession = sqlSessionFactory.openSession(false)) {
+ TableMetaMapper firstMapper =
firstSession.getMapper(TableMetaMapper.class);
+ DatabaseMetadata firstMetadata =
+ firstMapper.selectDatabaseMetadataForUpdate("lock_catalog",
"lock_db");
+
+ Future<Map<String, String>> secondUpdate =
+ executor.submit(
+ () -> {
+ try (SqlSession secondSession =
sqlSessionFactory.openSession(false)) {
+ TableMetaMapper secondMapper =
secondSession.getMapper(TableMetaMapper.class);
+ secondUpdateStarted.countDown();
+ DatabaseMetadata secondMetadata =
+
secondMapper.selectDatabaseMetadataForUpdate("lock_catalog", "lock_db");
+ Map<String, String> secondProperties =
+ new HashMap<>(secondMetadata.getProperties());
+ secondProperties.put("second", "2");
+ secondMapper.updateDatabaseProperties(
+ "lock_catalog", "lock_db", secondProperties);
+ secondSession.commit();
+ return secondProperties;
+ }
+ });
+
+ Assertions.assertTrue(secondUpdateStarted.await(5, TimeUnit.SECONDS));
+ Assertions.assertThrows(
+ TimeoutException.class, () -> secondUpdate.get(500,
TimeUnit.MILLISECONDS));
+
+ Map<String, String> firstProperties = new
HashMap<>(firstMetadata.getProperties());
+ firstProperties.put("first", "1");
+ firstMapper.updateDatabaseProperties("lock_catalog", "lock_db",
firstProperties);
+ firstSession.commit();
+
+ Map<String, String> mergedProperties = secondUpdate.get(5,
TimeUnit.SECONDS);
+ Assertions.assertEquals("1", mergedProperties.get("first"));
+ Assertions.assertEquals("2", mergedProperties.get("second"));
+ } finally {
+ executor.shutdownNow();
+ Assertions.assertTrue(executor.awaitTermination(5, TimeUnit.SECONDS));
+ }
+
+ try (SqlSession session = sqlSessionFactory.openSession(true)) {
+ Map<String, String> storedProperties =
+ session
+ .getMapper(TableMetaMapper.class)
+ .selectDatabaseMetadata("lock_catalog", "lock_db")
+ .getProperties();
+ Assertions.assertEquals("value", storedProperties.get("base"));
+ Assertions.assertEquals("1", storedProperties.get("first"));
+ Assertions.assertEquals("2", storedProperties.get("second"));
+ }
+ }
+}
