Repository: NetEase/arctic
Pull request #802 (2022-12-01 07:15:02)

[ARCTIC-797]Validate catalog properties when creating
<!--
Thanks for sending a pull request!

Here are some tips for you:
  1. If this is your first time, please read our contributor guidelines: https://arctic.netease.com/ch/contribute/
  2. If the PR is related to an issue in https://github.com/NetEase/arctic/issues, add '[ARCTIC-XXXX]' in your PR title, e.g., '[ARCTIC-XXXX] Your PR title ...'.
  3. If the PR is unfinished, add '[WIP]' in your PR title, e.g., '[WIP][ARCTIC #XXXX] Your PR title ...'.
-->

## Why are the changes needed?
<!--
Please clarify why the changes are needed. For instance,
  1. If you add a feature, you can talk about its use case.
  2. If you fix a bug, you can clarify why it is a bug.
-->

Resove #797

## Brief change log

  - *Add property validation for Arctic Metastore and Hadoop Catalog*
  - *Change catalog property name `warehouse.dir` to `warehouse`*

## How was this patch tested?
- [x] Add some test cases that check the changes thoroughly including negative and positive cases if possible

- [ ] Add screenshots for manual tests if appropriate

- [x] Run test locally before making a pull request

## Documentation

  - Does this pull request introduces a new feature? (yes / **no**)
  - If yes, how is the feature documented? (not applicable / docs / JavaDocs / **not documented**)



Patch:
diff --git a/ams/ams-api/src/main/java/com/netease/arctic/ams/api/properties/CatalogMetaProperties.java b/ams/ams-api/src/main/java/com/netease/arctic/ams/api/properties/CatalogMetaProperties.java
index 73a8a7917..873588d71 100644
--- a/ams/ams-api/src/main/java/com/netease/arctic/ams/api/properties/CatalogMetaProperties.java
+++ b/ams/ams-api/src/main/java/com/netease/arctic/ams/api/properties/CatalogMetaProperties.java
@@ -37,7 +37,10 @@ public class CatalogMetaProperties {
     public static final String AUTH_CONFIGS_VALUE_TYPE_SIMPLE = "simple";
     public static final String AUTH_CONFIGS_VALUE_TYPE_KERBEROS = "kerberos";
 
+    //Deprecated from version v0.4.0, use KEY_WAREHOUSE
+    @Deprecated
     public static final String KEY_WAREHOUSE_DIR = "warehouse.dir";
+    public static final String KEY_WAREHOUSE = "warehouse";
     public static final String KEY_DATABASE_FILTER_REGULAR_EXPRESSION = "database.filter-regular-expression";
 
     public static final String CATALOG_TYPE_HADOOP = "hadoop";
diff --git a/ams/ams-server/src/main/java/com/netease/arctic/ams/server/config/ServerTableProperties.java b/ams/ams-server/src/main/java/com/netease/arctic/ams/server/config/ServerTableProperties.java
index 42605a794..4f08d9816 100644
--- a/ams/ams-server/src/main/java/com/netease/arctic/ams/server/config/ServerTableProperties.java
+++ b/ams/ams-server/src/main/java/com/netease/arctic/ams/server/config/ServerTableProperties.java
@@ -46,7 +46,7 @@ public class ServerTableProperties {
     HIDDEN_EXPOSED.add(TableProperties.READ_DISTRIBUTION_HASH_AUTO);
     HIDDEN_EXPOSED.add(TableProperties.READ_DISTRIBUTION_MODE_NONE);
     HIDDEN_EXPOSED.add(TableProperties.READ_DISTRIBUTION_MODE_HASH);
-    HIDDEN_EXPOSED.add(TableProperties.PROTECTED_PROPERTIES.toString());
+    HIDDEN_EXPOSED.add(TableProperties.WRITE_PROTECTED_PROPERTIES.toString());
     HIDDEN_EXPOSED.add(TableProperties.TABLE_EVENT_TIME_FIELD);
 
     HIDDEN_INTERNAL.add("meta_store_site");
diff --git a/ams/ams-server/src/main/java/com/netease/arctic/ams/server/controller/CatalogController.java b/ams/ams-server/src/main/java/com/netease/arctic/ams/server/controller/CatalogController.java
index fc1b5722c..480da92f9 100644
--- a/ams/ams-server/src/main/java/com/netease/arctic/ams/server/controller/CatalogController.java
+++ b/ams/ams-server/src/main/java/com/netease/arctic/ams/server/controller/CatalogController.java
@@ -19,6 +19,8 @@
 package com.netease.arctic.ams.server.controller;
 
 import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
 import com.netease.arctic.ams.api.CatalogMeta;
 import com.netease.arctic.ams.api.properties.CatalogMetaProperties;
 import com.netease.arctic.ams.api.properties.TableFormat;
@@ -36,6 +38,7 @@ import com.netease.arctic.ams.server.utils.CatalogUtil;
 import io.javalin.http.Context;
 import io.javalin.http.HttpCode;
 import org.apache.commons.lang.StringUtils;
+import org.apache.iceberg.CatalogProperties;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -58,6 +61,7 @@ import static com.netease.arctic.ams.api.properties.CatalogMetaProperties.CATALO
 import static com.netease.arctic.ams.api.properties.CatalogMetaProperties.CATALOG_TYPE_CUSTOM;
 import static com.netease.arctic.ams.api.properties.CatalogMetaProperties.CATALOG_TYPE_HADOOP;
 import static com.netease.arctic.ams.api.properties.CatalogMetaProperties.CATALOG_TYPE_HIVE;
+import static com.netease.arctic.ams.api.properties.CatalogMetaProperties.KEY_WAREHOUSE;
 import static com.netease.arctic.ams.api.properties.CatalogMetaProperties.STORAGE_CONFIGS_KEY_CORE_SITE;
 import static com.netease.arctic.ams.api.properties.CatalogMetaProperties.STORAGE_CONFIGS_KEY_HDFS_SITE;
 import static com.netease.arctic.ams.api.properties.CatalogMetaProperties.STORAGE_CONFIGS_KEY_HIVE_SITE;
@@ -76,6 +80,15 @@ public class CatalogController extends RestBaseController {
   // <configuration></configuration>  encoded with base64
   private static final String EMPTY_XML_BASE64 = "PGNvbmZpZ3VyYXRpb24+PC9jb25maWd1cmF0aW9uPg==";
 
+  private static final Map<String, List<String>> CATALOG_REQUIRED_PROPERTIES;
+
+  static {
+    CATALOG_REQUIRED_PROPERTIES = Maps.newHashMap();
+    CATALOG_REQUIRED_PROPERTIES.put(CATALOG_TYPE_AMS, Lists.newArrayList(KEY_WAREHOUSE));
+    CATALOG_REQUIRED_PROPERTIES.put(CATALOG_TYPE_HADOOP, Lists.newArrayList(CatalogProperties.WAREHOUSE_LOCATION));
+    CATALOG_REQUIRED_PROPERTIES.put(CATALOG_TYPE_CUSTOM, Lists.newArrayList(CatalogProperties.CATALOG_IMPL));
+  }
+
   /**
    * Get catalog Type list
    */
@@ -207,10 +220,13 @@ public class CatalogController extends RestBaseController {
     } catch (Exception e) {
       throw new RuntimeException("Invalid table format list, " + String.join(",", info.getTableFormatList()));
     }
-    if (CATALOG_TYPE_CUSTOM.equals(info.getType())) {
-      // check properties contains key 'catalog-impl'
-      if (info.getProperties().containsKey("catalog-impl")) {
-        throw new RuntimeException("You must config catalog-impl in properties when catalog type is custom!");
+    List<String> requiredProperties = CATALOG_REQUIRED_PROPERTIES.get(info.getType());
+    if (requiredProperties != null && !requiredProperties.isEmpty()) {
+      for (String propertyName : requiredProperties) {
+        if (!info.getProperties().containsKey(propertyName)) {
+          throw new RuntimeException(
+              String.format("Catalog type:%s require property:%s.", info.getType(), propertyName));
+        }
       }
     }
     catalogMeta.getCatalogProperties().put(CatalogMetaProperties.TABLE_FORMATS, tableFormats.toString());
diff --git a/ams/ams-server/src/main/java/com/netease/arctic/ams/server/controller/TableController.java b/ams/ams-server/src/main/java/com/netease/arctic/ams/server/controller/TableController.java
index 7ad19784c..e5fbc52c0 100644
--- a/ams/ams-server/src/main/java/com/netease/arctic/ams/server/controller/TableController.java
+++ b/ams/ams-server/src/main/java/com/netease/arctic/ams/server/controller/TableController.java
@@ -66,11 +66,11 @@ import io.javalin.http.Context;
 import io.javalin.http.HttpCode;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.hadoop.hive.metastore.api.Table;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.nio.charset.StandardCharsets;
-import java.text.SimpleDateFormat;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.HashMap;
@@ -86,12 +86,12 @@ import java.util.stream.Collectors;
 public class TableController extends RestBaseController {
   private static final Logger LOG = LoggerFactory.getLogger(TableController.class);
 
-  private static ITableInfoService tableInfoService = ServiceContainer.getTableInfoService();
-  private static IOptimizeService optimizeService = ServiceContainer.getOptimizeService();
-  private static FileInfoCacheService fileInfoCacheService = ServiceContainer.getFileInfoCacheService();
-  private static CatalogMetadataService catalogMetadataService = ServiceContainer.getCatalogMetadataService();
-  private static AdaptHiveService adaptHiveService = ServiceContainer.getAdaptHiveService();
-  private static DDLTracerService ddlTracerService = ServiceContainer.getDdlTracerService();
+  private static final ITableInfoService tableInfoService = ServiceContainer.getTableInfoService();
+  private static final IOptimizeService optimizeService = ServiceContainer.getOptimizeService();
+  private static final FileInfoCacheService fileInfoCacheService = ServiceContainer.getFileInfoCacheService();
+  private static final CatalogMetadataService catalogMetadataService = ServiceContainer.getCatalogMetadataService();
+  private static final AdaptHiveService adaptHiveService = ServiceContainer.getAdaptHiveService();
+  private static final DDLTracerService ddlTracerService = ServiceContainer.getDdlTracerService();
 
   /**
    * get table detail.
@@ -111,7 +111,7 @@ public class TableController extends RestBaseController {
       return;
     }
 
-    TableBasicInfo tableBasicInfo = null;
+    TableBasicInfo tableBasicInfo;
 
     try {
       // set basic info
@@ -121,7 +121,6 @@ public class TableController extends RestBaseController {
       ctx.json(new ErrorResponse(HttpCode.BAD_REQUEST, "", ""));
       return;
     }
-    SimpleDateFormat sd = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
     ServerTableMeta serverTableMeta = MetaService.getServerTableMeta(ac, TableIdentifier.of(catalog, db, table));
     if (CatalogUtil.isIcebergCatalog(catalog)) {
       serverTableMeta.setTableType(TableMeta.TableType.ICEBERG.toString());
@@ -130,7 +129,7 @@ public class TableController extends RestBaseController {
     } else {
       serverTableMeta.setTableType(TableMeta.TableType.HIVE.toString());
     }
-    Map baseMetrics = new HashMap();
+    Map<String, Object> baseMetrics = Maps.newHashMap();
     FilesStatistics baseFilesStatistics = tableBasicInfo.getBaseStatistics().getTotalFilesStat();
     Map<String, String> baseSummary = tableBasicInfo.getBaseStatistics().getSummary();
     baseMetrics.put("lastCommitTime", AmsUtils.longOrNull(baseSummary.get("visibleTime")));
@@ -139,7 +138,7 @@ public class TableController extends RestBaseController {
     baseMetrics.put("averageFile", AmsUtils.byteToXB(baseFilesStatistics.getAverageSize()));
     serverTableMeta.setBaseMetrics(baseMetrics);
 
-    Map changeMetrics = new HashMap();
+    Map<String, Object> changeMetrics = Maps.newHashMap();
     if (tableBasicInfo.getChangeStatistics() != null) {
       FilesStatistics changeFilesStatistics = tableBasicInfo.getChangeStatistics().getTotalFilesStat();
       Map<String, String> changeSummary = tableBasicInfo.getChangeStatistics().getSummary();
@@ -172,7 +171,7 @@ public class TableController extends RestBaseController {
         = (ArcticHiveCatalog)CatalogUtil.getArcticCatalog(thriftHost, thriftPort, catalog);
 
     TableIdentifier tableIdentifier = TableIdentifier.of(catalog, db, table);
-    HiveTableInfo hiveTableInfo = null;
+    HiveTableInfo hiveTableInfo;
     try {
       Table hiveTable = HiveTableUtil.loadHmsTable(arcticHiveCatalog.getHMSClient(), tableIdentifier);
       List<AMSColumnInfo> schema =
@@ -254,7 +253,7 @@ public class TableController extends RestBaseController {
     checkOffsetAndLimit(offset, limit);
 
     TableIdentifier tableIdentifier = TableIdentifier.of(catalog, db, table);
-    List<BaseMajorCompactRecord> baseMajorCompactRecords = null;
+    List<BaseMajorCompactRecord> baseMajorCompactRecords;
     try {
       List<OptimizeHistory> tmpRecords = optimizeService.getOptimizeHistory(
           tableIdentifier);
@@ -278,7 +277,6 @@ public class TableController extends RestBaseController {
         .limit(limit)
         .collect(Collectors.toList());
     ctx.json(OkResponse.of(PageResult.of(result, total)));
-    return;
   }
 
   /**
@@ -295,15 +293,13 @@ public class TableController extends RestBaseController {
     try {
       List<TransactionsOfTable> transactionsOfTables = fileInfoCacheService.getTxExcludeOptimize(
           AmsUtils.toTableIdentifier(TableIdentifier.of(catalog, db, table)));
-      Integer offset = (page - 1) * pageSize;
+      int offset = (page - 1) * pageSize;
       PageResult<TransactionsOfTable, AMSTransactionsOfTable> pageResult = PageResult.of(transactionsOfTables,
           offset, pageSize, AmsUtils::toTransactionsOfTable);
       ctx.json(OkResponse.of(pageResult));
-      return;
     } catch (Exception e) {
       LOG.error("Failed to list transactions ", e);
       ctx.json(new ErrorResponse(HttpCode.BAD_REQUEST, "Failed to list transactions", ""));
-      return;
     }
   }
 
@@ -321,15 +317,13 @@ public class TableController extends RestBaseController {
     try {
       List<AMSDataFileInfo> dataFileInfo = fileInfoCacheService.getDatafilesInfo(
           AmsUtils.toTableIdentifier(TableIdentifier.of(catalog, db, table)), Long.valueOf(transactionId));
-      Integer offset = (page - 1) * pageSize;
+      int offset = (page - 1) * pageSize;
       PageResult<DataFileInfo, AMSDataFileInfo> amsPageResult = PageResult.of(dataFileInfo,
           offset, pageSize);
       ctx.json(OkResponse.of(amsPageResult));
-      return;
     } catch (Exception e) {
       LOG.error("Failed to get transactions detail", e);
       ctx.json(new ErrorResponse(HttpCode.BAD_REQUEST, "Failed to get transactions detail", ""));
-      return;
     }
   }
 
@@ -347,11 +341,10 @@ public class TableController extends RestBaseController {
       // First determine whether there is a partitioned table, and then get different information
       List<PartitionBaseInfo> partitionBaseInfos = fileInfoCacheService.getPartitionBaseInfoList(
           AmsUtils.toTableIdentifier(TableIdentifier.of(catalog, db, table)));
-      Integer offset = (page - 1) * pageSize;
+      int offset = (page - 1) * pageSize;
       PageResult<PartitionBaseInfo, PartitionBaseInfo> amsPageResult = PageResult.of(partitionBaseInfos,
           offset, pageSize);
       ctx.json(OkResponse.of(amsPageResult));
-      return;
     } catch (Exception e) {
       LOG.error("Failed to get transactions detail", e);
       ctx.json(new ErrorResponse(HttpCode.BAD_REQUEST, "Failed to get transactions detail", ""));
@@ -390,15 +383,13 @@ public class TableController extends RestBaseController {
       }
       List<PartitionFileBaseInfo> partitionFileBaseInfos = fileInfoCacheService.getPartitionFileList(
           AmsUtils.toTableIdentifier(TableIdentifier.of(catalog, db, table)), partition);
-      Integer offset = (page - 1) * pageSize;
+      int offset = (page - 1) * pageSize;
       PageResult<PartitionFileBaseInfo, PartitionFileBaseInfo> amsPageResult = PageResult.of(partitionFileBaseInfos,
           offset, pageSize);
       ctx.json(OkResponse.of(amsPageResult));
-      return;
     } catch (Exception e) {
       LOG.error("Failed to get partition file list", e);
       ctx.json(new ErrorResponse(HttpCode.BAD_REQUEST, "Failed to get partition file list", ""));
-      return;
     }
   }
 
@@ -410,7 +401,7 @@ public class TableController extends RestBaseController {
 
     Integer page = ctx.queryParamAsClass("page", Integer.class).getOrDefault(1);
     Integer pageSize = ctx.queryParamAsClass("pageSize", Integer.class).getOrDefault(20);
-    Integer offset = (page - 1) * pageSize;
+    int offset = (page - 1) * pageSize;
 
     List<DDLInfo> ddlInfos = ddlTracerService.getDDL(TableIdentifier.of(catalog, db, table).buildTableIdentifier());
     PageResult<DDLInfo, TableOperation> amsPageResult = PageResult.of(ddlInfos,
@@ -486,8 +477,6 @@ public class TableController extends RestBaseController {
 
   /**
    * get single page query token
-   *
-   * @param ctx
    */
   public static void getTableDetailTabToken(Context ctx) {
     String catalog = ctx.pathParam("catalog");
diff --git a/ams/ams-server/src/main/java/com/netease/arctic/ams/server/controller/response/ErrorResponse.java b/ams/ams-server/src/main/java/com/netease/arctic/ams/server/controller/response/ErrorResponse.java
index b6256106b..901dcf4fe 100644
--- a/ams/ams-server/src/main/java/com/netease/arctic/ams/server/controller/response/ErrorResponse.java
+++ b/ams/ams-server/src/main/java/com/netease/arctic/ams/server/controller/response/ErrorResponse.java
@@ -23,6 +23,11 @@ import io.javalin.http.HttpCode;
 public class ErrorResponse extends Response {
   private String requestId;
 
+  public ErrorResponse(int code, String message, String requestId) {
+    super(code, message);
+    this.requestId = requestId;
+  }
+
   public ErrorResponse(HttpCode httpStatus, String message, String requestId) {
     super(httpStatus.getStatus(), message);
     this.requestId = requestId;
@@ -33,7 +38,6 @@ public class ErrorResponse extends Response {
     this.requestId = null;
   }
 
-
   public String getRequestId() {
     return requestId;
   }
@@ -42,6 +46,7 @@ public class ErrorResponse extends Response {
     this.requestId = requestId;
   }
 
+
   @Override
   public String toString() {
     return "ErrorResponse{" +
diff --git a/ams/ams-server/src/main/java/com/netease/arctic/ams/server/service/MetaService.java b/ams/ams-server/src/main/java/com/netease/arctic/ams/server/service/MetaService.java
index ffd02877b..dd0102923 100644
--- a/ams/ams-server/src/main/java/com/netease/arctic/ams/server/service/MetaService.java
+++ b/ams/ams-server/src/main/java/com/netease/arctic/ams/server/service/MetaService.java
@@ -18,6 +18,7 @@
 
 package com.netease.arctic.ams.server.service;
 
+import com.google.common.collect.Maps;
 import com.netease.arctic.ams.server.model.AMSColumnInfo;
 import com.netease.arctic.ams.server.model.AMSPartitionField;
 import com.netease.arctic.ams.server.model.ServerTableMeta;
@@ -30,14 +31,18 @@ import org.apache.iceberg.util.PropertyUtil;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.text.DateFormat;
+import java.text.SimpleDateFormat;
 import java.util.ArrayList;
-import java.util.HashMap;
+import java.util.Date;
 import java.util.Map;
 import java.util.stream.Collectors;
 
 public class MetaService {
   private static final Logger LOG = LoggerFactory.getLogger(MetaService.class);
 
+  private static final DateFormat WATERMARK_FORMAT = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
+
   /**
    * getServerTableMeta.
    *
@@ -52,13 +57,13 @@ public class MetaService {
     ArcticTable at = ac.loadTable(ti);
     ServerTableMeta serverTableMeta = new ServerTableMeta();
     serverTableMeta.setTableIdentifier(ti);
-    Map<String, String> icebergProperties = at.properties();
-    Map<String, String> properties = new HashMap<>(icebergProperties);
+    Map<String, String> properties = Maps.newHashMap(at.properties());
+    formatTableProperties(properties);
     serverTableMeta.setProperties(properties);
     serverTableMeta.setCreateTime(PropertyUtil.propertyAsLong(properties, TableProperties.TABLE_CREATE_TIME,
             TableProperties.TABLE_CREATE_TIME_DEFAULT));
 
-    TableProperties.PROTECTED_PROPERTIES.forEach(serverTableMeta.getProperties()::remove);
+    TableProperties.READ_PROTECTED_PROPERTIES.forEach(serverTableMeta.getProperties()::remove);
     serverTableMeta.getProperties().remove(TableProperties.TABLE_CREATE_TIME);
 
     serverTableMeta.setBaseLocation(at.location());
@@ -93,4 +98,20 @@ public class MetaService {
     }
     return serverTableMeta;
   }
+
+  private static void formatTableProperties(Map<String, String> tableProperties) {
+    formatTableWatermark(tableProperties, TableProperties.WATERMARK_TABLE);
+    formatTableWatermark(tableProperties, TableProperties.WATERMARK_BASE_STORE);
+  }
+
+  private static void formatTableWatermark(Map<String, String> tableProperties, String watermarkPropertyName) {
+    if (tableProperties.containsKey(watermarkPropertyName)) {
+      Long watermarkTimestamp = Long.valueOf(tableProperties.get(watermarkPropertyName));
+      if (watermarkTimestamp > 0) {
+        tableProperties.put(watermarkPropertyName, WATERMARK_FORMAT.format(new Date(watermarkTimestamp)));
+      } else {
+        tableProperties.remove(watermarkPropertyName);
+      }
+    }
+  }
 }
diff --git a/ams/ams-server/src/main/java/com/netease/arctic/ams/server/service/impl/DDLTracerService.java b/ams/ams-server/src/main/java/com/netease/arctic/ams/server/service/impl/DDLTracerService.java
index 8943a45d2..eccc6ad22 100644
--- a/ams/ams-server/src/main/java/com/netease/arctic/ams/server/service/impl/DDLTracerService.java
+++ b/ams/ams-server/src/main/java/com/netease/arctic/ams/server/service/impl/DDLTracerService.java
@@ -176,7 +176,7 @@ public class DDLTracerService extends IJDBCService {
     StringBuilder unsetPro = new StringBuilder();
     int c = 0;
     for (String oldPro : before.keySet()) {
-      if (TableProperties.PROTECTED_PROPERTIES.contains(oldPro)) {
+      if (TableProperties.WRITE_PROTECTED_PROPERTIES.contains(oldPro)) {
         continue;
       }
       if (!after.containsKey(oldPro)) {
@@ -190,7 +190,7 @@ public class DDLTracerService extends IJDBCService {
     StringBuilder setPro = new StringBuilder();
     int c1 = 0;
     for (String newPro : after.keySet()) {
-      if (TableProperties.PROTECTED_PROPERTIES.contains(newPro)) {
+      if (TableProperties.WRITE_PROTECTED_PROPERTIES.contains(newPro)) {
         continue;
       }
       if (!after.get(newPro).equals(before.get(newPro))) {
@@ -321,7 +321,7 @@ public class DDLTracerService extends IJDBCService {
     StringBuilder unsetPro = new StringBuilder();
     int c = 0;
     for (String oldPro : before.keySet()) {
-      if (TableProperties.PROTECTED_PROPERTIES.contains(oldPro)) {
+      if (TableProperties.WRITE_PROTECTED_PROPERTIES.contains(oldPro)) {
         continue;
       }
       if (!after.containsKey(oldPro)) {
@@ -335,7 +335,7 @@ public class DDLTracerService extends IJDBCService {
     StringBuilder setPro = new StringBuilder();
     int c1 = 0;
     for (String newPro : after.keySet()) {
-      if (TableProperties.PROTECTED_PROPERTIES.contains(newPro)) {
+      if (TableProperties.WRITE_PROTECTED_PROPERTIES.contains(newPro)) {
         continue;
       }
       if (!after.get(newPro).equals(before.get(newPro))) {
diff --git a/ams/ams-server/src/main/resources/sql/derby/ams-init.sql b/ams/ams-server/src/main/resources/sql/derby/ams-init.sql
index 036c93785..fa3f1923b 100644
--- a/ams/ams-server/src/main/resources/sql/derby/ams-init.sql
+++ b/ams/ams-server/src/main/resources/sql/derby/ams-init.sql
@@ -266,4 +266,4 @@ CREATE TABLE platform_file_info (
   PRIMARY KEY (id)
 );
 
-INSERT INTO catalog_metadata(catalog_name,catalog_type,storage_configs,auth_configs, catalog_properties) VALUES ('local_catalog','ams','{"storage.type":"hdfs","hive.site":"PGNvbmZpZ3VyYXRpb24+PC9jb25maWd1cmF0aW9uPg==","hadoop.core.site":"PGNvbmZpZ3VyYXRpb24+PC9jb25maWd1cmF0aW9uPg==","hadoop.hdfs.site":"PGNvbmZpZ3VyYXRpb24+PC9jb25maWd1cmF0aW9uPg=="}','{"auth.type":"simple","auth.simple.hadoop_username":"root"}','{"warehouse.dir":"/tmp/arctic/warehouse","table-formats":"ICEBERG"}');
\ No newline at end of file
+INSERT INTO catalog_metadata(catalog_name,catalog_type,storage_configs,auth_configs, catalog_properties) VALUES ('local_catalog','ams','{"storage.type":"hdfs","hive.site":"PGNvbmZpZ3VyYXRpb24+PC9jb25maWd1cmF0aW9uPg==","hadoop.core.site":"PGNvbmZpZ3VyYXRpb24+PC9jb25maWd1cmF0aW9uPg==","hadoop.hdfs.site":"PGNvbmZpZ3VyYXRpb24+PC9jb25maWd1cmF0aW9uPg=="}','{"auth.type":"simple","auth.simple.hadoop_username":"root"}','{"warehouse":"/tmp/arctic/warehouse","table-formats":"MIXED_ICEBERG"}');
\ No newline at end of file
diff --git a/ams/ams-server/src/main/resources/sql/mysql/0.4.0-init.sql b/ams/ams-server/src/main/resources/sql/mysql/0.4.0-init.sql
index c75ab14f9..32e76ec7b 100644
--- a/ams/ams-server/src/main/resources/sql/mysql/0.4.0-init.sql
+++ b/ams/ams-server/src/main/resources/sql/mysql/0.4.0-init.sql
@@ -287,4 +287,4 @@ CREATE TABLE `platform_file_info` (
   PRIMARY KEY (`id`)
 ) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8mb4 COMMENT='store files info saved in the platform';
 
-INSERT INTO catalog_metadata(catalog_name,display_name,catalog_type,storage_configs,auth_configs, catalog_properties) VALUES ('local_catalog',NULL,'ams','{"storage.type":"hdfs","hive.site":"PGNvbmZpZ3VyYXRpb24+PC9jb25maWd1cmF0aW9uPg==","hadoop.core.site":"PGNvbmZpZ3VyYXRpb24+PC9jb25maWd1cmF0aW9uPg==","hadoop.hdfs.site":"PGNvbmZpZ3VyYXRpb24+PC9jb25maWd1cmF0aW9uPg=="}','{"auth.type":"simple","auth.simple.hadoop_username":"root"}','{"warehouse.dir":"/tmp/arctic/warehouse","table-formats":"ICEBERG"}');
\ No newline at end of file
+INSERT INTO catalog_metadata(catalog_name,display_name,catalog_type,storage_configs,auth_configs, catalog_properties) VALUES ('local_catalog',NULL,'ams','{"storage.type":"hdfs","hive.site":"PGNvbmZpZ3VyYXRpb24+PC9jb25maWd1cmF0aW9uPg==","hadoop.core.site":"PGNvbmZpZ3VyYXRpb24+PC9jb25maWd1cmF0aW9uPg==","hadoop.hdfs.site":"PGNvbmZpZ3VyYXRpb24+PC9jb25maWd1cmF0aW9uPg=="}','{"auth.type":"simple","auth.simple.hadoop_username":"root"}','{"warehouse":"/tmp/arctic/warehouse","table-formats":"MIXED_ICEBERG"}');
\ No newline at end of file
diff --git a/core/src/main/java/com/netease/arctic/catalog/BaseArcticCatalog.java b/core/src/main/java/com/netease/arctic/catalog/BaseArcticCatalog.java
index a84a3c6bf..c0e413d2d 100644
--- a/core/src/main/java/com/netease/arctic/catalog/BaseArcticCatalog.java
+++ b/core/src/main/java/com/netease/arctic/catalog/BaseArcticCatalog.java
@@ -653,7 +653,7 @@ public class BaseArcticCatalog implements ArcticCatalog {
       } else {
         throw new IllegalStateException(
             "either `location` in table properties or " +
-                "`warehouse.dir` in catalog properties is specified");
+                "`warehouse` in catalog properties is specified");
       }
     }
 
@@ -665,16 +665,21 @@ public class BaseArcticCatalog implements ArcticCatalog {
 
     protected String getDatabaseLocation() {
       if (catalogMeta.getCatalogProperties() != null) {
-        String catalogWarehouseDir = catalogMeta.getCatalogProperties().getOrDefault(
-            CatalogMetaProperties.KEY_WAREHOUSE_DIR,
-            null
-        );
-        if (!Objects.equals("/", catalogWarehouseDir) && catalogWarehouseDir.endsWith("/")) {
-          catalogWarehouseDir = catalogWarehouseDir.substring(
-              0, catalogWarehouseDir.length() - 1);
+        String catalogWarehouse = catalogMeta.getCatalogProperties().getOrDefault(
+            CatalogMetaProperties.KEY_WAREHOUSE,null);
+        if (catalogWarehouse == null) {
+          catalogWarehouse = catalogMeta.getCatalogProperties().getOrDefault(
+              CatalogMetaProperties.KEY_WAREHOUSE_DIR,null);
         }
-        if (StringUtils.isNotBlank(catalogWarehouseDir)) {
-          return catalogWarehouseDir + '/' + identifier.getDatabase();
+        if (catalogWarehouse == null) {
+          throw new NullPointerException("Catalog warehouse is null.");
+        }
+        if (!Objects.equals("/", catalogWarehouse) && catalogWarehouse.endsWith("/")) {
+          catalogWarehouse = catalogWarehouse.substring(
+              0, catalogWarehouse.length() - 1);
+        }
+        if (StringUtils.isNotBlank(catalogWarehouse)) {
+          return catalogWarehouse + '/' + identifier.getDatabase();
         }
       }
       return null;
diff --git a/core/src/main/java/com/netease/arctic/table/BaseKeyedTable.java b/core/src/main/java/com/netease/arctic/table/BaseKeyedTable.java
index a3eef14df..44b30db36 100644
--- a/core/src/main/java/com/netease/arctic/table/BaseKeyedTable.java
+++ b/core/src/main/java/com/netease/arctic/table/BaseKeyedTable.java
@@ -105,17 +105,19 @@ public class BaseKeyedTable implements KeyedTable {
     long changeWatermark = TablePropertyUtil.getTableWatermark(changeTable.properties());
     long baseWatermark = TablePropertyUtil.getTableWatermark(baseTable.properties());
 
+    ImmutableMap.Builder<String, String> builder = ImmutableMap.builder();
     if (changeWatermark > baseWatermark) {
-      ImmutableMap.Builder<String, String> builder = ImmutableMap.builder();
       baseTable.properties().forEach((k, v) -> {
         if (!TableProperties.WATERMARK_TABLE.equals(k)) {
           builder.put(k, v);
         }
       });
       builder.put(TableProperties.WATERMARK_TABLE, String.valueOf(changeWatermark));
-      return builder.build();
+    } else {
+      builder.putAll(baseTable.properties());
     }
-    return baseTable.properties();
+    builder.put(TableProperties.WATERMARK_BASE_STORE, String.valueOf(baseWatermark));
+    return builder.build();
   }
 
   @Override
diff --git a/core/src/main/java/com/netease/arctic/table/TableProperties.java b/core/src/main/java/com/netease/arctic/table/TableProperties.java
index e2e53ce08..cb4237be5 100644
--- a/core/src/main/java/com/netease/arctic/table/TableProperties.java
+++ b/core/src/main/java/com/netease/arctic/table/TableProperties.java
@@ -64,7 +64,7 @@ public class TableProperties {
 
   public static final String WATERMARK_TABLE = "watermark.table";
 
-  public static final String WATERMARK_BASE_STORE = "watermark.base-store";
+  public static final String WATERMARK_BASE_STORE = "watermark.base";
 
   /**
    * table optimize related properties
@@ -275,17 +275,32 @@ public class TableProperties {
   public static final String OWNER = "owner";
 
   /**
-   * Protected properties which should not be exposed to user.
+   * Protected properties which should not be read by user.
    */
-  public static final Set<String> PROTECTED_PROPERTIES = new HashSet<>();
+  public static final Set<String> READ_PROTECTED_PROPERTIES = new HashSet<>();
+
+  /**
+   * Protected properties which should not be written by user.
+   */
+  public static final Set<String> WRITE_PROTECTED_PROPERTIES = new HashSet<>();
 
   static {
-    PROTECTED_PROPERTIES.add(TableProperties.BASE_TABLE_MAX_TRANSACTION_ID);
-    PROTECTED_PROPERTIES.add(TableProperties.PARTITION_MAX_TRANSACTION_ID);
-    PROTECTED_PROPERTIES.add(TableProperties.LOCATION);
-    PROTECTED_PROPERTIES.add(TableProperties.TABLE_PARTITION_PROPERTIES);
-    PROTECTED_PROPERTIES.add(DEFAULT_NAME_MAPPING);
-    PROTECTED_PROPERTIES.add(FORMAT_VERSION);
-    PROTECTED_PROPERTIES.add(WATERMARK_TABLE);
+    READ_PROTECTED_PROPERTIES.add(TableProperties.BASE_TABLE_MAX_TRANSACTION_ID);
+    READ_PROTECTED_PROPERTIES.add(TableProperties.PARTITION_MAX_TRANSACTION_ID);
+    READ_PROTECTED_PROPERTIES.add(TableProperties.LOCATION);
+    READ_PROTECTED_PROPERTIES.add(TableProperties.TABLE_PARTITION_PROPERTIES);
+    READ_PROTECTED_PROPERTIES.add(DEFAULT_NAME_MAPPING);
+    READ_PROTECTED_PROPERTIES.add(FORMAT_VERSION);
+
+
+    WRITE_PROTECTED_PROPERTIES.add(TableProperties.BASE_TABLE_MAX_TRANSACTION_ID);
+    WRITE_PROTECTED_PROPERTIES.add(TableProperties.PARTITION_MAX_TRANSACTION_ID);
+    WRITE_PROTECTED_PROPERTIES.add(TableProperties.LOCATION);
+    WRITE_PROTECTED_PROPERTIES.add(TableProperties.TABLE_PARTITION_PROPERTIES);
+    WRITE_PROTECTED_PROPERTIES.add(DEFAULT_NAME_MAPPING);
+    WRITE_PROTECTED_PROPERTIES.add(FORMAT_VERSION);
+    WRITE_PROTECTED_PROPERTIES.add(WATERMARK_TABLE);
+    WRITE_PROTECTED_PROPERTIES.add(WATERMARK_BASE_STORE);
   }
+
 }


Test patch:
diff --git a/ams/ams-api/src/test/java/com/netease/arctic/ams/api/MockArcticMetastoreServer.java b/ams/ams-api/src/test/java/com/netease/arctic/ams/api/MockArcticMetastoreServer.java
index f387ffe03..2d1184149 100644
--- a/ams/ams-api/src/test/java/com/netease/arctic/ams/api/MockArcticMetastoreServer.java
+++ b/ams/ams-api/src/test/java/com/netease/arctic/ams/api/MockArcticMetastoreServer.java
@@ -76,7 +76,7 @@ public class MockArcticMetastoreServer implements Runnable {
           System.getProperty("user.name"));
 
       Map<String, String> catalogProperties = new HashMap<>();
-      catalogProperties.put(CatalogMetaProperties.KEY_WAREHOUSE_DIR, "/tmp");
+      catalogProperties.put(CatalogMetaProperties.KEY_WAREHOUSE, "/tmp");
 
       CatalogMeta catalogMeta = new CatalogMeta(TEST_CATALOG_NAME, CATALOG_TYPE_HADOOP,
           storageConfig, authConfig, catalogProperties);
diff --git a/ams/ams-server/src/test/java/com/netease/arctic/ams/server/AmsEnvironment.java b/ams/ams-server/src/test/java/com/netease/arctic/ams/server/AmsEnvironment.java
index 7e7d7e8b8..1445271bf 100644
--- a/ams/ams-server/src/test/java/com/netease/arctic/ams/server/AmsEnvironment.java
+++ b/ams/ams-server/src/test/java/com/netease/arctic/ams/server/AmsEnvironment.java
@@ -185,7 +185,7 @@ public class AmsEnvironment {
       LOG.error("failed to create iceberg warehouse dir {}", warehouseDir, e);
       throw new RuntimeException(e);
     }
-    properties.put("warehouse.dir", warehouseDir);
+    properties.put("warehouse", warehouseDir);
     properties.put("table-formats", "ICEBERG");
     localCatalog.setCatalogProperties(properties);
     ServiceContainer.getCatalogMetadataService().addCatalog(localCatalog);
diff --git a/ams/ams-server/src/test/java/com/netease/arctic/ams/server/AmsTestBase.java b/ams/ams-server/src/test/java/com/netease/arctic/ams/server/AmsTestBase.java
index 2c514f8df..894c632b5 100644
--- a/ams/ams-server/src/test/java/com/netease/arctic/ams/server/AmsTestBase.java
+++ b/ams/ams-server/src/test/java/com/netease/arctic/ams/server/AmsTestBase.java
@@ -240,7 +240,7 @@ public class AmsTestBase {
             System.getProperty("user.name"));
 
     Map<String, String> catalogProperties = new HashMap<>();
-    catalogProperties.put(CatalogMetaProperties.KEY_WAREHOUSE_DIR, "/tmp");
+    catalogProperties.put(CatalogMetaProperties.KEY_WAREHOUSE, "/tmp");
     CatalogMeta catalogMeta = new CatalogMeta(CATALOG_CONTROLLER_UNITTEST_NAME, CATALOG_TYPE_HADOOP,
             storageConfig, authConfig, catalogProperties);
     List<CatalogMeta> catalogMetas = Lists.newArrayList(catalogMeta);
@@ -265,7 +265,7 @@ public class AmsTestBase {
         System.getProperty("user.name"));
 
     Map<String, String> catalogProperties = new HashMap<>();
-    catalogProperties.put(CatalogMetaProperties.KEY_WAREHOUSE_DIR, tempFolder.newFolder().getPath());
+    catalogProperties.put(CatalogMetaProperties.KEY_WAREHOUSE, tempFolder.newFolder().getPath());
     CatalogMeta catalogMeta = new CatalogMeta(AMS_TEST_CATALOG_NAME, CATALOG_TYPE_HADOOP,
         storageConfig, authConfig, catalogProperties);
     List<CatalogMeta> catalogMetas = Lists.newArrayList(catalogMeta);
diff --git a/ams/ams-server/src/test/java/com/netease/arctic/ams/server/controller/CatalogControllerTest.java b/ams/ams-server/src/test/java/com/netease/arctic/ams/server/controller/CatalogControllerTest.java
index fdb0e9759..c6a2526be 100644
--- a/ams/ams-server/src/test/java/com/netease/arctic/ams/server/controller/CatalogControllerTest.java
+++ b/ams/ams-server/src/test/java/com/netease/arctic/ams/server/controller/CatalogControllerTest.java
@@ -19,11 +19,13 @@
 package com.netease.arctic.ams.server.controller;
 
 import com.alibaba.fastjson.JSONObject;
+import com.netease.arctic.ams.server.controller.response.ErrorResponse;
 import com.netease.arctic.ams.server.controller.response.OkResponse;
 import com.netease.arctic.ams.server.controller.response.Response;
 import com.netease.arctic.ams.server.model.CatalogSettingInfo;
 import io.javalin.testtools.HttpClient;
 import io.javalin.testtools.JavalinTest;
+import org.junit.Assert;
 import org.junit.Test;
 
 import java.util.List;
@@ -58,18 +60,20 @@ public class CatalogControllerTest {
     String name = "unittest";
     String newAuthUser = "UnitTest";
     JavalinTest.test((app, client) -> {
-      // add one catalog
+
       app.post("/", CatalogController::createCatalog);
       app.get("/{catalogName}", CatalogController::getCatalogDetail);
       app.put("/{catalogName}", CatalogController::updateCatalog);
       app.delete("/{catalogName}", CatalogController::deleteCatalog);
-      String paramString = "{\"name\":\"unittest\",\"type\":\"hadoop\","
+
+      // add one catalog
+      String paramString = "{\"name\":\"unittest\",\"type\":\"ams\","
               + "\"storageConfig\":{"
               + "\"hadoop.core.site\":\"1\","
               + "\"hadoop.hdfs.site\":\"2\","
               + "\"hive.site\":\"3\"},"
               + "\"authConfig\":{\"auth.type\":\"SIMPLE\",\"auth.simple.hadoop_username\":\"arctic\"},"
-              + "\"properties\":{},\"tableFormatList\":[\"MIXED_HIVE\"]}";
+              + "\"properties\":{\"warehouse\":\"/data/warehouse\"},\"tableFormatList\":[\"MIXED_HIVE\"]}";
       JSONObject param = JSONObject.parseObject(paramString);
       final okhttp3.Response resp = client.post("/", param);
       Response result = JSONObject.parseObject(resp.body().string(), Response.class);
@@ -108,6 +112,20 @@ public class CatalogControllerTest {
       assert result5.getCode() == 200;
       assert result5.getResult() == null;
 
+      // validate catalog property
+      paramString = "{\"name\":\"unittest2\",\"type\":\"ams\","
+          + "\"storageConfig\":{"
+          + "\"hadoop.core.site\":\"1\","
+          + "\"hadoop.hdfs.site\":\"2\","
+          + "\"hive.site\":\"3\"},"
+          + "\"authConfig\":{\"auth.type\":\"SIMPLE\",\"auth.simple.hadoop_username\":\"arctic\"},"
+          + "\"properties\":{},\"tableFormatList\":[\"MIXED_HIVE\"]}";
+      JSONObject param6 = JSONObject.parseObject(paramString);
+      final okhttp3.Response resp6 = client.post("/", param6);
+      String resp6Body = resp6.body().string();
+      final ErrorResponse result6 = JSONObject.parseObject(resp6Body, ErrorResponse.class);
+      assert result6.getCode() == 400;
+      assert result6.getMessage().equals("Catalog type:ams require property:warehouse.");
     });
   }
 }
diff --git a/ams/ams-server/src/test/java/com/netease/arctic/ams/server/controller/TableControllerTest.java b/ams/ams-server/src/test/java/com/netease/arctic/ams/server/controller/TableControllerTest.java
index c4abd58ed..e7dc76345 100644
--- a/ams/ams-server/src/test/java/com/netease/arctic/ams/server/controller/TableControllerTest.java
+++ b/ams/ams-server/src/test/java/com/netease/arctic/ams/server/controller/TableControllerTest.java
@@ -575,7 +575,7 @@ public class TableControllerTest {
         System.getProperty("user.name"));
 
     Map<String, String> catalogProperties = new HashMap<>();
-    catalogProperties.put(CatalogMetaProperties.KEY_WAREHOUSE_DIR, "/tmp");
+    catalogProperties.put(CatalogMetaProperties.KEY_WAREHOUSE, "/tmp");
 
     CatalogMeta catalogMeta = new CatalogMeta(TEST_CATALOG_NAME, CATALOG_TYPE_HADOOP,
         storageConfig, authConfig, catalogProperties);
diff --git a/ams/ams-server/src/test/java/com/netease/arctic/ams/server/optimize/TestSupportHiveBase.java b/ams/ams-server/src/test/java/com/netease/arctic/ams/server/optimize/TestSupportHiveBase.java
index 94983a88e..4a2e27375 100644
--- a/ams/ams-server/src/test/java/com/netease/arctic/ams/server/optimize/TestSupportHiveBase.java
+++ b/ams/ams-server/src/test/java/com/netease/arctic/ams/server/optimize/TestSupportHiveBase.java
@@ -185,7 +185,7 @@ public class TestSupportHiveBase implements TestOptimizeBase {
           System.getProperty("user.name"));
 
       Map<String, String> hiveCatalogProperties = new HashMap<>();
-      hiveCatalogProperties.put(CatalogMetaProperties.KEY_WAREHOUSE_DIR, "/hive_tmp");
+      hiveCatalogProperties.put(CatalogMetaProperties.KEY_WAREHOUSE, "/hive_tmp");
       CatalogMeta hiveCatalogMeta = new CatalogMeta(AMS_TEST_HIVE_CATALOG_NAME, CATALOG_TYPE_HIVE,
           storageConfig, authConfig, hiveCatalogProperties);
       List<CatalogMeta> catalogMetas = Lists.newArrayList(hiveCatalogMeta);
diff --git a/core/src/test/java/com/netease/arctic/CatalogMetaTestUtil.java b/core/src/test/java/com/netease/arctic/CatalogMetaTestUtil.java
index 4fbecd287..051cda1d1 100644
--- a/core/src/test/java/com/netease/arctic/CatalogMetaTestUtil.java
+++ b/core/src/test/java/com/netease/arctic/CatalogMetaTestUtil.java
@@ -56,7 +56,7 @@ public class CatalogMetaTestUtil {
         System.getProperty("user.name"));
 
     meta.putToCatalogProperties(
-        CatalogMetaProperties.KEY_WAREHOUSE_DIR,
+        CatalogMetaProperties.KEY_WAREHOUSE,
         arcticBaseDir.getAbsolutePath()
     );
     return meta;
diff --git a/core/src/test/java/com/netease/arctic/catalog/BaseCatalogTest.java b/core/src/test/java/com/netease/arctic/catalog/BaseCatalogTest.java
index de9733992..b6cad513e 100644
--- a/core/src/test/java/com/netease/arctic/catalog/BaseCatalogTest.java
+++ b/core/src/test/java/com/netease/arctic/catalog/BaseCatalogTest.java
@@ -68,12 +68,12 @@ public class BaseCatalogTest extends TableTestBase {
   @Test
   public void refreshCatalog() throws TException {
     CatalogMeta catalog = AMS.handler().getCatalog(TEST_CATALOG_NAME);
-    AMS.handler().updateMeta(catalog, CatalogMetaProperties.KEY_WAREHOUSE_DIR, "/test");
+    AMS.handler().updateMeta(catalog, CatalogMetaProperties.KEY_WAREHOUSE, "/test");
     testCatalog = CatalogLoader.load(AMS.getUrl());
     testCatalog.refresh();
     Assert.assertEquals("/test",
         AMS.handler().getCatalog(TEST_CATALOG_NAME).
-            getCatalogProperties().get(CatalogMetaProperties.KEY_WAREHOUSE_DIR));
+            getCatalogProperties().get(CatalogMetaProperties.KEY_WAREHOUSE));
   }
 
 }
diff --git a/spark/v2.3/spark/src/test/java/com/netease/arctic/spark/ArcticCatalogMetaTestUtil.java b/spark/v2.3/spark/src/test/java/com/netease/arctic/spark/ArcticCatalogMetaTestUtil.java
index 0578fbc1b..8f667484e 100644
--- a/spark/v2.3/spark/src/test/java/com/netease/arctic/spark/ArcticCatalogMetaTestUtil.java
+++ b/spark/v2.3/spark/src/test/java/com/netease/arctic/spark/ArcticCatalogMetaTestUtil.java
@@ -60,7 +60,7 @@ public class ArcticCatalogMetaTestUtil {
         System.getProperty("user.name"));
 
     meta.putToCatalogProperties(
-        CatalogMetaProperties.KEY_WAREHOUSE_DIR,
+        CatalogMetaProperties.KEY_WAREHOUSE,
         arcticBaseDir.getAbsolutePath()
     );
     return meta;
diff --git a/spark/v3.1/spark/src/test/java/com/netease/arctic/spark/TestKeyedTableDDL.java b/spark/v3.1/spark/src/test/java/com/netease/arctic/spark/TestKeyedTableDDL.java
index e6b96ebc1..856fdaa48 100644
--- a/spark/v3.1/spark/src/test/java/com/netease/arctic/spark/TestKeyedTableDDL.java
+++ b/spark/v3.1/spark/src/test/java/com/netease/arctic/spark/TestKeyedTableDDL.java
@@ -178,7 +178,7 @@ public class TestKeyedTableDDL extends SparkTestBase {
     File testArcticDir = new File(testBaseDir, "arctic2");
     sql("set `spark.sql.arctic.refresh-catalog-before-usage` = {0}", "true");
     CatalogMeta catalog = ams.handler().getCatalog(catalogNameArctic);
-    ams.handler().updateMeta(catalog, CatalogMetaProperties.KEY_WAREHOUSE_DIR, testArcticDir.getPath());
+    ams.handler().updateMeta(catalog, CatalogMetaProperties.KEY_WAREHOUSE, testArcticDir.getPath());
     TableIdentifier identifier = TableIdentifier.of(catalogNameArctic, database, table);
 
     sql("create table {0}.{1} ( \n" +
diff --git a/spark/v3.1/spark/src/test/java/com/netease/arctic/spark/hive/HiveCatalogMetaTestUtil.java b/spark/v3.1/spark/src/test/java/com/netease/arctic/spark/hive/HiveCatalogMetaTestUtil.java
index 4c2739534..4c73e6104 100644
--- a/spark/v3.1/spark/src/test/java/com/netease/arctic/spark/hive/HiveCatalogMetaTestUtil.java
+++ b/spark/v3.1/spark/src/test/java/com/netease/arctic/spark/hive/HiveCatalogMetaTestUtil.java
@@ -41,7 +41,7 @@ public class HiveCatalogMetaTestUtil {
         System.getProperty("user.name"));
 
     meta.putToCatalogProperties(
-        CatalogMetaProperties.KEY_WAREHOUSE_DIR,
+        CatalogMetaProperties.KEY_WAREHOUSE,
         arcticBaseDir.getAbsolutePath()
     );
     return meta;