Repository: NetEase/arctic (java)

Issue:
[Bug]: Global Catalog configuration not valid for table build check
### What happened?

![image](https://github.com/NetEase/arctic/assets/109209550/bd86789d-e7b4-4a1c-8329-9efcbd7dbe46)
![image](https://github.com/NetEase/arctic/assets/109209550/ad30e9e3-8b92-4748-a988-ba1b56216807)
As you can see, after I configured table.log-store.topic = xxxxx in the global configuration of catalog, I still get an error saying that this configuration cannot be null.

### Affects Versions

0.4.x

### What engines are you seeing the problem on?

Core

### How to reproduce

1.Create a catalog and set properties log-store.topic = xxxxx
2.Create a table under this catalog and set properties log-store.enabled = true

### Relevant log output

```shell
2023/06/12 20:12:10 java.lang.IllegalArgumentException: log-store.topic must not be null when log-store.enabled is true.
	at org.apache.iceberg.relocated.com.google.common.base.Preconditions.checkArgument(Preconditions.java:142)
	at com.netease.arctic.catalog.BasicArcticCatalog$ArcticTableBuilder.checkProperties(BasicArcticCatalog.java:549)
	at com.netease.arctic.catalog.BasicArcticCatalog$ArcticTableBuilder.doCreateCheck(BasicArcticCatalog.java:539)
	at com.netease.arctic.hive.catalog.ArcticHiveCatalog$ArcticHiveTableBuilder.doCreateCheck(ArcticHiveCatalog.java:249)
	at com.netease.arctic.catalog.BasicArcticCatalog$ArcticTableBuilder.create(BasicArcticCatalog.java:488)
	at com.netease.arctic.spark.ArcticSparkCatalog.createTable(ArcticSparkCatalog.java:198)
	at org.apache.spark.sql.execution.datasources.v2.CreateTableExec.run(CreateTableExec.scala:41)
	at org.apache.spark.sql.execution.datasources.v2.V2CommandExec.result$lzycompute(V2CommandExec.scala:40)
	at org.apache.spark.sql.execution.datasources.v2.V2CommandExec.result(V2CommandExec.scala:40)
	at org.apache.spark.sql.execution.datasources.v2.V2CommandExec.executeCollect(V2CommandExec.scala:46)
	at org.apache.spark.sql.Dataset.$anonfun$logicalPlan$1(Dataset.scala:228)
	at org.apache.spark.sql.Dataset.$anonfun$withAction$1(Dataset.scala:3687)
	at org.apache.spark.sql.execution.SQLExecution$.$anonfun$withNewExecutionId$5(SQLExecution.scala:103)
	at org.apache.spark.sql.execution.SQLExecution$.withSQLConfPropagated(SQLExecution.scala:163)
	at org.apache.spark.sql.execution.SQLExecution$.$anonfun$withNewExecutionId$1(SQLExecution.scala:90)
	at org.apache.spark.sql.SparkSession.withActive(SparkSession.scala:772)
	at org.apache.spark.sql.execution.SQLExecution$.withNewExecutionId(SQLExecution.scala:64)
	at org.apache.spark.sql.Dataset.withAction(Dataset.scala:3685)
	at org.apache.spark.sql.Dataset.(Dataset.scala:228)
	at org.apache.spark.sql.Dataset$.$anonfun$ofRows$2(Dataset.scala:99)
	at org.apache.spark.sql.SparkSession.withActive(SparkSession.scala:772)
	at org.apache.spark.sql.Dataset$.ofRows(Dataset.scala:96)
	at org.apache.spark.sql.SparkSession.$anonfun$sql$1(SparkSession.scala:615)
	at org.apache.spark.sql.SparkSession.withActive(SparkSession.scala:772)
	at org.apache.spark.sql.SparkSession.sql(SparkSession.scala:610)
	at com.netease.arctic.ams.server.terminal.local.LocalTerminalSession.executeStatement(LocalTerminalSession.java:65)
	at com.netease.arctic.ams.server.terminal.TerminalSessionContext$ExecutionTask.executeStatement(TerminalSessionContext.java:275)
	at com.netease.arctic.ams.server.terminal.TerminalSessionContext$ExecutionTask.execute(TerminalSessionContext.java:238)
	at com.netease.arctic.ams.server.terminal.TerminalSessionContext$ExecutionTask.lambda$get$0(TerminalSessionContext.java:199)
	at com.netease.arctic.table.TableMetaStore.doAsUgi(TableMetaStore.java:365)
	at com.netease.arctic.table.TableMetaStore.lambda$doAs$0(TableMetaStore.java:345)
	at java.security.AccessController.doPrivileged(Native Method)
	at javax.security.auth.Subject.doAs(Subject.java:360)
	at org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1873)
	at com.netease.arctic.table.TableMetaStore.doAs(TableMetaStore.java:345)
	at com.netease.arctic.ams.server.terminal.TerminalSessionContext$ExecutionTask.get(TerminalSessionContext.java:191)
	at com.netease.arctic.ams.server.terminal.TerminalSessionContext$ExecutionTask.get(TerminalSessionContext.java:162)
	at java.util.concurrent.CompletableFuture$AsyncSupply.run(CompletableFuture.java:1604)
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
	at java.lang.Thread.run(Thread.java:748)
```


### Anything else

_No response_

### Code of Conduct

- [X] I agree to follow this project's Code of Conduct

Pull request description:
[ARCTIC-1551]: Global Catalog configuration not valid for table build check
<!--
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?
resolve #1551 

## Brief change log

Merge catalog properties before validation

## 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



Patch:
diff --git a/core/src/main/java/com/netease/arctic/catalog/BasicArcticCatalog.java b/core/src/main/java/com/netease/arctic/catalog/BasicArcticCatalog.java
index d1514b011..0521d2094 100644
--- a/core/src/main/java/com/netease/arctic/catalog/BasicArcticCatalog.java
+++ b/core/src/main/java/com/netease/arctic/catalog/BasicArcticCatalog.java
@@ -347,14 +347,16 @@ public class BasicArcticCatalog implements ArcticCatalog {
     }
 
     protected void checkProperties() {
-      boolean enableStream = CompatiblePropertyUtil.propertyAsBoolean(properties,
+      Map<String, String> mergedProperties = CatalogUtil.mergeCatalogPropertiesToTable(properties,
+          catalogMeta.getCatalogProperties());
+      boolean enableStream = CompatiblePropertyUtil.propertyAsBoolean(mergedProperties,
           TableProperties.ENABLE_LOG_STORE, TableProperties.ENABLE_LOG_STORE_DEFAULT);
       if (enableStream) {
-        Preconditions.checkArgument(properties.containsKey(TableProperties.LOG_STORE_MESSAGE_TOPIC),
+        Preconditions.checkArgument(mergedProperties.containsKey(TableProperties.LOG_STORE_MESSAGE_TOPIC),
             "log-store.topic must not be null when log-store.enabled is true.");
-        Preconditions.checkArgument(properties.containsKey(TableProperties.LOG_STORE_ADDRESS),
+        Preconditions.checkArgument(mergedProperties.containsKey(TableProperties.LOG_STORE_ADDRESS),
             "log-store.address must not be null when log-store.enabled is true.");
-        String logStoreType = properties.get(LOG_STORE_TYPE);
+        String logStoreType = mergedProperties.get(LOG_STORE_TYPE);
         Preconditions.checkArgument(logStoreType == null ||
                 logStoreType.equals(LOG_STORE_STORAGE_TYPE_KAFKA) ||
                 logStoreType.equals(LOG_STORE_STORAGE_TYPE_PULSAR),
diff --git a/core/src/main/java/com/netease/arctic/scan/BaseCombinedScanTask.java b/core/src/main/java/com/netease/arctic/scan/BaseCombinedScanTask.java
index 0e93474e3..88f94d1f6 100644
--- a/core/src/main/java/com/netease/arctic/scan/BaseCombinedScanTask.java
+++ b/core/src/main/java/com/netease/arctic/scan/BaseCombinedScanTask.java
@@ -50,7 +50,7 @@ public class BaseCombinedScanTask implements CombinedScanTask {
   @Override
   public String toString() {
     return MoreObjects.toStringHelper(this)
-        .add("tasks", Joiner.on(", ").join(tasks))
+        .add("\ntasks", Joiner.on(",\n").join(tasks).replaceAll("\n", "\n\t"))
         .toString();
   }
 }
diff --git a/core/src/main/java/com/netease/arctic/scan/NodeFileScanTask.java b/core/src/main/java/com/netease/arctic/scan/NodeFileScanTask.java
index 9e6e8c27e..7dc338fdc 100644
--- a/core/src/main/java/com/netease/arctic/scan/NodeFileScanTask.java
+++ b/core/src/main/java/com/netease/arctic/scan/NodeFileScanTask.java
@@ -21,6 +21,8 @@ package com.netease.arctic.scan;
 import com.netease.arctic.data.DataFileType;
 import com.netease.arctic.data.DataTreeNode;
 import com.netease.arctic.table.TableProperties;
+import com.netease.arctic.utils.FileScanTaskUtil;
+import org.apache.iceberg.relocated.com.google.common.base.MoreObjects;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -142,4 +144,13 @@ public class NodeFileScanTask implements KeyedTableScanTask {
   public DataTreeNode treeNode() {
     return treeNode;
   }
+
+  @Override
+  public String toString() {
+    return MoreObjects.toStringHelper(this)
+        .add("\nbaseTasks", FileScanTaskUtil.toString(baseTasks))
+        .add("\ninsertTasks", FileScanTaskUtil.toString(insertTasks))
+        .add("\ndeleteFiles", FileScanTaskUtil.toString(deleteFiles))
+        .toString();
+  }
 }
diff --git a/core/src/main/java/com/netease/arctic/utils/FileScanTaskUtil.java b/core/src/main/java/com/netease/arctic/utils/FileScanTaskUtil.java
new file mode 100644
index 000000000..847eb71d5
--- /dev/null
+++ b/core/src/main/java/com/netease/arctic/utils/FileScanTaskUtil.java
@@ -0,0 +1,58 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.netease.arctic.utils;
+
+import com.netease.arctic.scan.ArcticFileScanTask;
+import org.apache.iceberg.relocated.com.google.common.base.MoreObjects;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
+
+import java.util.Collection;
+import java.util.stream.Collectors;
+
+/**
+ * Utility class for working with file scan tasks.
+ */
+public class FileScanTaskUtil {
+
+  /**
+   * Converts a collection of ArcticFileScanTask objects to a string representation.
+   * The string representation includes details about each file, such as its path, type, mask, index,
+   * transaction ID, file size in bytes, and record count.
+   *
+   * @param tasks the collection of ArcticFileScanTask objects to convert to a string
+   * @return a string representation of the ArcticFileScanTask objects
+   */
+  public static String toString(Collection<ArcticFileScanTask> tasks) {
+    if (tasks == null) {
+      return "[]";
+    }
+    return Iterables.toString(tasks.stream()
+        .map(ArcticFileScanTask::file)
+        .map(primaryKeyedFile ->
+            MoreObjects.toStringHelper(primaryKeyedFile)
+                .add("\n\tfile", primaryKeyedFile.path().toString())
+                .add("\n\ttype", primaryKeyedFile.type().shortName())
+                .add("\n\tmask", primaryKeyedFile.node().mask())
+                .add("\n\tindex", primaryKeyedFile.node().index())
+                .add("\n\ttransactionId", primaryKeyedFile.transactionId())
+                .add("\n\tfileSizeInBytes", primaryKeyedFile.fileSizeInBytes())
+                .add("\n\trecordCount", primaryKeyedFile.recordCount())
+                .toString()).collect(Collectors.toList()));
+  }
+}
diff --git a/flink/v1.12/flink/src/main/java/com/netease/arctic/flink/read/hybrid/split/ArcticSplit.java b/flink/v1.12/flink/src/main/java/com/netease/arctic/flink/read/hybrid/split/ArcticSplit.java
index 99cc75a5a..3acd4dec1 100644
--- a/flink/v1.12/flink/src/main/java/com/netease/arctic/flink/read/hybrid/split/ArcticSplit.java
+++ b/flink/v1.12/flink/src/main/java/com/netease/arctic/flink/read/hybrid/split/ArcticSplit.java
@@ -19,14 +19,9 @@
 package com.netease.arctic.flink.read.hybrid.split;
 
 import com.netease.arctic.data.DataTreeNode;
-import com.netease.arctic.scan.ArcticFileScanTask;
 import org.apache.flink.api.connector.source.SourceSplit;
-import org.apache.iceberg.relocated.com.google.common.base.MoreObjects;
-import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
 
 import java.io.Serializable;
-import java.util.Collection;
-import java.util.stream.Collectors;
 
 /**
  * An abstract arctic source split.
@@ -95,24 +90,6 @@ public abstract class ArcticSplit implements SourceSplit, Serializable, Comparab
     return this.taskIndex().compareTo(that.taskIndex());
   }
 
-  protected String toString(Collection<ArcticFileScanTask> fileScanTasks) {
-    if (fileScanTasks == null) {
-      return "[]";
-    }
-    return Iterables.toString(fileScanTasks.stream()
-        .map(ArcticFileScanTask::file)
-        .map(primaryKeyedFile ->
-            MoreObjects.toStringHelper(primaryKeyedFile)
-                .add("\n\tfile", primaryKeyedFile.path().toString())
-                .add("\n\ttype", primaryKeyedFile.type().shortName())
-                .add("\n\tmask", primaryKeyedFile.node().mask())
-                .add("\n\tindex", primaryKeyedFile.node().index())
-                .add("\n\ttransactionId", primaryKeyedFile.transactionId())
-                .add("\n\tfileSizeInBytes", primaryKeyedFile.fileSizeInBytes())
-                .add("\n\trecordCount", primaryKeyedFile.recordCount())
-                .toString()).collect(Collectors.toList()));
-  }
-
   public abstract ArcticSplit copy();
 
 }
diff --git a/flink/v1.12/flink/src/main/java/com/netease/arctic/flink/read/hybrid/split/ChangelogSplit.java b/flink/v1.12/flink/src/main/java/com/netease/arctic/flink/read/hybrid/split/ChangelogSplit.java
index 4b886bc41..62a58e47e 100644
--- a/flink/v1.12/flink/src/main/java/com/netease/arctic/flink/read/hybrid/split/ChangelogSplit.java
+++ b/flink/v1.12/flink/src/main/java/com/netease/arctic/flink/read/hybrid/split/ChangelogSplit.java
@@ -21,6 +21,7 @@ package com.netease.arctic.flink.read.hybrid.split;
 import com.netease.arctic.data.DataTreeNode;
 import com.netease.arctic.data.PrimaryKeyedFile;
 import com.netease.arctic.scan.ArcticFileScanTask;
+import com.netease.arctic.utils.FileScanTaskUtil;
 import org.apache.iceberg.relocated.com.google.common.base.MoreObjects;
 import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
 
@@ -87,16 +88,16 @@ public class ChangelogSplit extends ArcticSplit {
   @Override
   public String splitId() {
     return MoreObjects.toStringHelper(this)
-        .add("insertTasks", toString(insertScanTasks))
-        .add("arcticEquityDeletes", toString(deleteScanTasks))
+        .add("insertTasks", FileScanTaskUtil.toString(insertScanTasks))
+        .add("arcticEquityDeletes", FileScanTaskUtil.toString(deleteScanTasks))
         .toString();
   }
 
   @Override
   public String toString() {
     return MoreObjects.toStringHelper(this)
-        .add("insertTasks", toString(insertScanTasks))
-        .add("arcticEquityDeletes", toString(deleteScanTasks))
+        .add("insertTasks", FileScanTaskUtil.toString(insertScanTasks))
+        .add("arcticEquityDeletes", FileScanTaskUtil.toString(deleteScanTasks))
         .add("dataTreeNode", dataTreeNode.toString())
         .toString();
   }
diff --git a/flink/v1.12/flink/src/main/java/com/netease/arctic/flink/read/hybrid/split/MergeOnReadSplit.java b/flink/v1.12/flink/src/main/java/com/netease/arctic/flink/read/hybrid/split/MergeOnReadSplit.java
index 6f881c64c..7f32973c8 100644
--- a/flink/v1.12/flink/src/main/java/com/netease/arctic/flink/read/hybrid/split/MergeOnReadSplit.java
+++ b/flink/v1.12/flink/src/main/java/com/netease/arctic/flink/read/hybrid/split/MergeOnReadSplit.java
@@ -19,6 +19,7 @@
 package com.netease.arctic.flink.read.hybrid.split;
 
 import com.netease.arctic.scan.KeyedTableScanTask;
+import com.netease.arctic.utils.FileScanTaskUtil;
 import org.apache.flink.util.FlinkRuntimeException;
 import org.apache.iceberg.relocated.com.google.common.base.MoreObjects;
 
@@ -54,18 +55,18 @@ public class MergeOnReadSplit extends ArcticSplit {
   @Override
   public String splitId() {
     return MoreObjects.toStringHelper(this)
-        .add("insertTasks", toString(keyedTableScanTask.insertTasks()))
-        .add("baseTasks", toString(keyedTableScanTask.baseTasks()))
-        .add("arcticEquityDeletes", toString(keyedTableScanTask.arcticEquityDeletes()))
+        .add("insertTasks", FileScanTaskUtil.toString(keyedTableScanTask.insertTasks()))
+        .add("baseTasks", FileScanTaskUtil.toString(keyedTableScanTask.baseTasks()))
+        .add("arcticEquityDeletes", FileScanTaskUtil.toString(keyedTableScanTask.arcticEquityDeletes()))
         .toString();
   }
 
   @Override
   public String toString() {
     return MoreObjects.toStringHelper(this)
-        .add("\ninsertTasks", toString(keyedTableScanTask.insertTasks()))
-        .add("\nbaseTasks", toString(keyedTableScanTask.baseTasks()))
-        .add("\narcticEquityDeletes", toString(keyedTableScanTask.arcticEquityDeletes()))
+        .add("\ninsertTasks", FileScanTaskUtil.toString(keyedTableScanTask.insertTasks()))
+        .add("\nbaseTasks", FileScanTaskUtil.toString(keyedTableScanTask.baseTasks()))
+        .add("\narcticEquityDeletes", FileScanTaskUtil.toString(keyedTableScanTask.arcticEquityDeletes()))
         .add("\ncost", keyedTableScanTask.cost() / 1024 + " KB")
         .add("\nrecordCount", keyedTableScanTask.recordCount())
         .toString();
diff --git a/flink/v1.12/flink/src/main/java/com/netease/arctic/flink/read/hybrid/split/SnapshotSplit.java b/flink/v1.12/flink/src/main/java/com/netease/arctic/flink/read/hybrid/split/SnapshotSplit.java
index 6d8c8180d..7840dc66f 100644
--- a/flink/v1.12/flink/src/main/java/com/netease/arctic/flink/read/hybrid/split/SnapshotSplit.java
+++ b/flink/v1.12/flink/src/main/java/com/netease/arctic/flink/read/hybrid/split/SnapshotSplit.java
@@ -21,6 +21,7 @@ package com.netease.arctic.flink.read.hybrid.split;
 import com.netease.arctic.data.DataTreeNode;
 import com.netease.arctic.data.PrimaryKeyedFile;
 import com.netease.arctic.scan.ArcticFileScanTask;
+import com.netease.arctic.utils.FileScanTaskUtil;
 import org.apache.iceberg.relocated.com.google.common.base.MoreObjects;
 import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
 
@@ -50,7 +51,7 @@ public class SnapshotSplit extends ArcticSplit {
   @Override
   public String splitId() {
     return MoreObjects.toStringHelper(this)
-        .add("insertTasks", toString(insertScanTasks))
+        .add("insertTasks", FileScanTaskUtil.toString(insertScanTasks))
         .toString();
   }
 
@@ -97,7 +98,7 @@ public class SnapshotSplit extends ArcticSplit {
   @Override
   public String toString() {
     return MoreObjects.toStringHelper(this)
-        .add("insertTasks", toString(insertScanTasks))
+        .add("insertTasks", FileScanTaskUtil.toString(insertScanTasks))
         .add("dataTreeNode", dataTreeNode.toString())
         .toString();
   }
diff --git a/flink/v1.14/flink/src/main/java/com/netease/arctic/flink/read/hybrid/split/ArcticSplit.java b/flink/v1.14/flink/src/main/java/com/netease/arctic/flink/read/hybrid/split/ArcticSplit.java
index 99cc75a5a..3acd4dec1 100644
--- a/flink/v1.14/flink/src/main/java/com/netease/arctic/flink/read/hybrid/split/ArcticSplit.java
+++ b/flink/v1.14/flink/src/main/java/com/netease/arctic/flink/read/hybrid/split/ArcticSplit.java
@@ -19,14 +19,9 @@
 package com.netease.arctic.flink.read.hybrid.split;
 
 import com.netease.arctic.data.DataTreeNode;
-import com.netease.arctic.scan.ArcticFileScanTask;
 import org.apache.flink.api.connector.source.SourceSplit;
-import org.apache.iceberg.relocated.com.google.common.base.MoreObjects;
-import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
 
 import java.io.Serializable;
-import java.util.Collection;
-import java.util.stream.Collectors;
 
 /**
  * An abstract arctic source split.
@@ -95,24 +90,6 @@ public abstract class ArcticSplit implements SourceSplit, Serializable, Comparab
     return this.taskIndex().compareTo(that.taskIndex());
   }
 
-  protected String toString(Collection<ArcticFileScanTask> fileScanTasks) {
-    if (fileScanTasks == null) {
-      return "[]";
-    }
-    return Iterables.toString(fileScanTasks.stream()
-        .map(ArcticFileScanTask::file)
-        .map(primaryKeyedFile ->
-            MoreObjects.toStringHelper(primaryKeyedFile)
-                .add("\n\tfile", primaryKeyedFile.path().toString())
-                .add("\n\ttype", primaryKeyedFile.type().shortName())
-                .add("\n\tmask", primaryKeyedFile.node().mask())
-                .add("\n\tindex", primaryKeyedFile.node().index())
-                .add("\n\ttransactionId", primaryKeyedFile.transactionId())
-                .add("\n\tfileSizeInBytes", primaryKeyedFile.fileSizeInBytes())
-                .add("\n\trecordCount", primaryKeyedFile.recordCount())
-                .toString()).collect(Collectors.toList()));
-  }
-
   public abstract ArcticSplit copy();
 
 }
diff --git a/flink/v1.14/flink/src/main/java/com/netease/arctic/flink/read/hybrid/split/ChangelogSplit.java b/flink/v1.14/flink/src/main/java/com/netease/arctic/flink/read/hybrid/split/ChangelogSplit.java
index 275b4ff98..0c2e32ad9 100644
--- a/flink/v1.14/flink/src/main/java/com/netease/arctic/flink/read/hybrid/split/ChangelogSplit.java
+++ b/flink/v1.14/flink/src/main/java/com/netease/arctic/flink/read/hybrid/split/ChangelogSplit.java
@@ -21,6 +21,7 @@ package com.netease.arctic.flink.read.hybrid.split;
 import com.netease.arctic.data.DataTreeNode;
 import com.netease.arctic.data.PrimaryKeyedFile;
 import com.netease.arctic.scan.ArcticFileScanTask;
+import com.netease.arctic.utils.FileScanTaskUtil;
 import org.apache.iceberg.relocated.com.google.common.base.MoreObjects;
 import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
 
@@ -87,16 +88,16 @@ public class ChangelogSplit extends ArcticSplit {
   @Override
   public String splitId() {
     return MoreObjects.toStringHelper(this)
-        .add("insertTasks", toString(insertScanTasks))
-        .add("arcticEquityDeletes", toString(deleteScanTasks))
+        .add("insertTasks", FileScanTaskUtil.toString(insertScanTasks))
+        .add("arcticEquityDeletes", FileScanTaskUtil.toString(deleteScanTasks))
         .toString();
   }
 
   @Override
   public String toString() {
     return MoreObjects.toStringHelper(this)
-        .add("insertTasks", toString(insertScanTasks))
-        .add("arcticEquityDeletes", toString(deleteScanTasks))
+        .add("insertTasks", FileScanTaskUtil.toString(insertScanTasks))
+        .add("arcticEquityDeletes", FileScanTaskUtil.toString(deleteScanTasks))
         .add("dataTreeNode", dataTreeNode.toString())
         .toString();
   }
diff --git a/flink/v1.14/flink/src/main/java/com/netease/arctic/flink/read/hybrid/split/MergeOnReadSplit.java b/flink/v1.14/flink/src/main/java/com/netease/arctic/flink/read/hybrid/split/MergeOnReadSplit.java
index 6f881c64c..7f32973c8 100644
--- a/flink/v1.14/flink/src/main/java/com/netease/arctic/flink/read/hybrid/split/MergeOnReadSplit.java
+++ b/flink/v1.14/flink/src/main/java/com/netease/arctic/flink/read/hybrid/split/MergeOnReadSplit.java
@@ -19,6 +19,7 @@
 package com.netease.arctic.flink.read.hybrid.split;
 
 import com.netease.arctic.scan.KeyedTableScanTask;
+import com.netease.arctic.utils.FileScanTaskUtil;
 import org.apache.flink.util.FlinkRuntimeException;
 import org.apache.iceberg.relocated.com.google.common.base.MoreObjects;
 
@@ -54,18 +55,18 @@ public class MergeOnReadSplit extends ArcticSplit {
   @Override
   public String splitId() {
     return MoreObjects.toStringHelper(this)
-        .add("insertTasks", toString(keyedTableScanTask.insertTasks()))
-        .add("baseTasks", toString(keyedTableScanTask.baseTasks()))
-        .add("arcticEquityDeletes", toString(keyedTableScanTask.arcticEquityDeletes()))
+        .add("insertTasks", FileScanTaskUtil.toString(keyedTableScanTask.insertTasks()))
+        .add("baseTasks", FileScanTaskUtil.toString(keyedTableScanTask.baseTasks()))
+        .add("arcticEquityDeletes", FileScanTaskUtil.toString(keyedTableScanTask.arcticEquityDeletes()))
         .toString();
   }
 
   @Override
   public String toString() {
     return MoreObjects.toStringHelper(this)
-        .add("\ninsertTasks", toString(keyedTableScanTask.insertTasks()))
-        .add("\nbaseTasks", toString(keyedTableScanTask.baseTasks()))
-        .add("\narcticEquityDeletes", toString(keyedTableScanTask.arcticEquityDeletes()))
+        .add("\ninsertTasks", FileScanTaskUtil.toString(keyedTableScanTask.insertTasks()))
+        .add("\nbaseTasks", FileScanTaskUtil.toString(keyedTableScanTask.baseTasks()))
+        .add("\narcticEquityDeletes", FileScanTaskUtil.toString(keyedTableScanTask.arcticEquityDeletes()))
         .add("\ncost", keyedTableScanTask.cost() / 1024 + " KB")
         .add("\nrecordCount", keyedTableScanTask.recordCount())
         .toString();
diff --git a/flink/v1.14/flink/src/main/java/com/netease/arctic/flink/read/hybrid/split/SnapshotSplit.java b/flink/v1.14/flink/src/main/java/com/netease/arctic/flink/read/hybrid/split/SnapshotSplit.java
index 263a8a847..b39f28c65 100644
--- a/flink/v1.14/flink/src/main/java/com/netease/arctic/flink/read/hybrid/split/SnapshotSplit.java
+++ b/flink/v1.14/flink/src/main/java/com/netease/arctic/flink/read/hybrid/split/SnapshotSplit.java
@@ -21,6 +21,7 @@ package com.netease.arctic.flink.read.hybrid.split;
 import com.netease.arctic.data.DataTreeNode;
 import com.netease.arctic.data.PrimaryKeyedFile;
 import com.netease.arctic.scan.ArcticFileScanTask;
+import com.netease.arctic.utils.FileScanTaskUtil;
 import org.apache.iceberg.relocated.com.google.common.base.MoreObjects;
 import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
 
@@ -50,7 +51,7 @@ public class SnapshotSplit extends ArcticSplit {
   @Override
   public String splitId() {
     return MoreObjects.toStringHelper(this)
-        .add("insertTasks", toString(insertScanTasks))
+        .add("insertTasks", FileScanTaskUtil.toString(insertScanTasks))
         .toString();
   }
 
@@ -97,7 +98,7 @@ public class SnapshotSplit extends ArcticSplit {
   @Override
   public String toString() {
     return MoreObjects.toStringHelper(this)
-        .add("insertTasks", toString(insertScanTasks))
+        .add("insertTasks", FileScanTaskUtil.toString(insertScanTasks))
         .add("dataTreeNode", dataTreeNode.toString())
         .toString();
   }
diff --git a/flink/v1.15/flink/src/main/java/com/netease/arctic/flink/read/hybrid/split/ArcticSplit.java b/flink/v1.15/flink/src/main/java/com/netease/arctic/flink/read/hybrid/split/ArcticSplit.java
index 99cc75a5a..3acd4dec1 100644
--- a/flink/v1.15/flink/src/main/java/com/netease/arctic/flink/read/hybrid/split/ArcticSplit.java
+++ b/flink/v1.15/flink/src/main/java/com/netease/arctic/flink/read/hybrid/split/ArcticSplit.java
@@ -19,14 +19,9 @@
 package com.netease.arctic.flink.read.hybrid.split;
 
 import com.netease.arctic.data.DataTreeNode;
-import com.netease.arctic.scan.ArcticFileScanTask;
 import org.apache.flink.api.connector.source.SourceSplit;
-import org.apache.iceberg.relocated.com.google.common.base.MoreObjects;
-import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
 
 import java.io.Serializable;
-import java.util.Collection;
-import java.util.stream.Collectors;
 
 /**
  * An abstract arctic source split.
@@ -95,24 +90,6 @@ public abstract class ArcticSplit implements SourceSplit, Serializable, Comparab
     return this.taskIndex().compareTo(that.taskIndex());
   }
 
-  protected String toString(Collection<ArcticFileScanTask> fileScanTasks) {
-    if (fileScanTasks == null) {
-      return "[]";
-    }
-    return Iterables.toString(fileScanTasks.stream()
-        .map(ArcticFileScanTask::file)
-        .map(primaryKeyedFile ->
-            MoreObjects.toStringHelper(primaryKeyedFile)
-                .add("\n\tfile", primaryKeyedFile.path().toString())
-                .add("\n\ttype", primaryKeyedFile.type().shortName())
-                .add("\n\tmask", primaryKeyedFile.node().mask())
-                .add("\n\tindex", primaryKeyedFile.node().index())
-                .add("\n\ttransactionId", primaryKeyedFile.transactionId())
-                .add("\n\tfileSizeInBytes", primaryKeyedFile.fileSizeInBytes())
-                .add("\n\trecordCount", primaryKeyedFile.recordCount())
-                .toString()).collect(Collectors.toList()));
-  }
-
   public abstract ArcticSplit copy();
 
 }
diff --git a/flink/v1.15/flink/src/main/java/com/netease/arctic/flink/read/hybrid/split/ChangelogSplit.java b/flink/v1.15/flink/src/main/java/com/netease/arctic/flink/read/hybrid/split/ChangelogSplit.java
index 275b4ff98..0c2e32ad9 100644
--- a/flink/v1.15/flink/src/main/java/com/netease/arctic/flink/read/hybrid/split/ChangelogSplit.java
+++ b/flink/v1.15/flink/src/main/java/com/netease/arctic/flink/read/hybrid/split/ChangelogSplit.java
@@ -21,6 +21,7 @@ package com.netease.arctic.flink.read.hybrid.split;
 import com.netease.arctic.data.DataTreeNode;
 import com.netease.arctic.data.PrimaryKeyedFile;
 import com.netease.arctic.scan.ArcticFileScanTask;
+import com.netease.arctic.utils.FileScanTaskUtil;
 import org.apache.iceberg.relocated.com.google.common.base.MoreObjects;
 import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
 
@@ -87,16 +88,16 @@ public class ChangelogSplit extends ArcticSplit {
   @Override
   public String splitId() {
     return MoreObjects.toStringHelper(this)
-        .add("insertTasks", toString(insertScanTasks))
-        .add("arcticEquityDeletes", toString(deleteScanTasks))
+        .add("insertTasks", FileScanTaskUtil.toString(insertScanTasks))
+        .add("arcticEquityDeletes", FileScanTaskUtil.toString(deleteScanTasks))
         .toString();
   }
 
   @Override
   public String toString() {
     return MoreObjects.toStringHelper(this)
-        .add("insertTasks", toString(insertScanTasks))
-        .add("arcticEquityDeletes", toString(deleteScanTasks))
+        .add("insertTasks", FileScanTaskUtil.toString(insertScanTasks))
+        .add("arcticEquityDeletes", FileScanTaskUtil.toString(deleteScanTasks))
         .add("dataTreeNode", dataTreeNode.toString())
         .toString();
   }
diff --git a/flink/v1.15/flink/src/main/java/com/netease/arctic/flink/read/hybrid/split/MergeOnReadSplit.java b/flink/v1.15/flink/src/main/java/com/netease/arctic/flink/read/hybrid/split/MergeOnReadSplit.java
index 6f881c64c..7f32973c8 100644
--- a/flink/v1.15/flink/src/main/java/com/netease/arctic/flink/read/hybrid/split/MergeOnReadSplit.java
+++ b/flink/v1.15/flink/src/main/java/com/netease/arctic/flink/read/hybrid/split/MergeOnReadSplit.java
@@ -19,6 +19,7 @@
 package com.netease.arctic.flink.read.hybrid.split;
 
 import com.netease.arctic.scan.KeyedTableScanTask;
+import com.netease.arctic.utils.FileScanTaskUtil;
 import org.apache.flink.util.FlinkRuntimeException;
 import org.apache.iceberg.relocated.com.google.common.base.MoreObjects;
 
@@ -54,18 +55,18 @@ public class MergeOnReadSplit extends ArcticSplit {
   @Override
   public String splitId() {
     return MoreObjects.toStringHelper(this)
-        .add("insertTasks", toString(keyedTableScanTask.insertTasks()))
-        .add("baseTasks", toString(keyedTableScanTask.baseTasks()))
-        .add("arcticEquityDeletes", toString(keyedTableScanTask.arcticEquityDeletes()))
+        .add("insertTasks", FileScanTaskUtil.toString(keyedTableScanTask.insertTasks()))
+        .add("baseTasks", FileScanTaskUtil.toString(keyedTableScanTask.baseTasks()))
+        .add("arcticEquityDeletes", FileScanTaskUtil.toString(keyedTableScanTask.arcticEquityDeletes()))
         .toString();
   }
 
   @Override
   public String toString() {
     return MoreObjects.toStringHelper(this)
-        .add("\ninsertTasks", toString(keyedTableScanTask.insertTasks()))
-        .add("\nbaseTasks", toString(keyedTableScanTask.baseTasks()))
-        .add("\narcticEquityDeletes", toString(keyedTableScanTask.arcticEquityDeletes()))
+        .add("\ninsertTasks", FileScanTaskUtil.toString(keyedTableScanTask.insertTasks()))
+        .add("\nbaseTasks", FileScanTaskUtil.toString(keyedTableScanTask.baseTasks()))
+        .add("\narcticEquityDeletes", FileScanTaskUtil.toString(keyedTableScanTask.arcticEquityDeletes()))
         .add("\ncost", keyedTableScanTask.cost() / 1024 + " KB")
         .add("\nrecordCount", keyedTableScanTask.recordCount())
         .toString();
diff --git a/flink/v1.15/flink/src/main/java/com/netease/arctic/flink/read/hybrid/split/SnapshotSplit.java b/flink/v1.15/flink/src/main/java/com/netease/arctic/flink/read/hybrid/split/SnapshotSplit.java
index 263a8a847..b39f28c65 100644
--- a/flink/v1.15/flink/src/main/java/com/netease/arctic/flink/read/hybrid/split/SnapshotSplit.java
+++ b/flink/v1.15/flink/src/main/java/com/netease/arctic/flink/read/hybrid/split/SnapshotSplit.java
@@ -21,6 +21,7 @@ package com.netease.arctic.flink.read.hybrid.split;
 import com.netease.arctic.data.DataTreeNode;
 import com.netease.arctic.data.PrimaryKeyedFile;
 import com.netease.arctic.scan.ArcticFileScanTask;
+import com.netease.arctic.utils.FileScanTaskUtil;
 import org.apache.iceberg.relocated.com.google.common.base.MoreObjects;
 import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
 
@@ -50,7 +51,7 @@ public class SnapshotSplit extends ArcticSplit {
   @Override
   public String splitId() {
     return MoreObjects.toStringHelper(this)
-        .add("insertTasks", toString(insertScanTasks))
+        .add("insertTasks", FileScanTaskUtil.toString(insertScanTasks))
         .toString();
   }
 
@@ -97,7 +98,7 @@ public class SnapshotSplit extends ArcticSplit {
   @Override
   public String toString() {
     return MoreObjects.toStringHelper(this)
-        .add("insertTasks", toString(insertScanTasks))
+        .add("insertTasks", FileScanTaskUtil.toString(insertScanTasks))
         .add("dataTreeNode", dataTreeNode.toString())
         .toString();
   }


Test patch:
diff --git a/core/src/test/java/com/netease/arctic/catalog/TestMixedCatalog.java b/core/src/test/java/com/netease/arctic/catalog/TestMixedCatalog.java
index 660d9daea..48cc7fcd1 100644
--- a/core/src/test/java/com/netease/arctic/catalog/TestMixedCatalog.java
+++ b/core/src/test/java/com/netease/arctic/catalog/TestMixedCatalog.java
@@ -142,6 +142,41 @@ public class TestMixedCatalog extends CatalogTestBase {
         TableProperties.ENABLE_SELF_OPTIMIZING, TableProperties.ENABLE_SELF_OPTIMIZING_DEFAULT));
   }
 
+  @Test
+  public void testCreateTableWithNewCatalogLogProperties() throws TException {
+    UnkeyedTable createTable = getCatalog()
+        .newTableBuilder(TableTestHelper.TEST_TABLE_ID, getCreateTableSchema())
+        .withPartitionSpec(getCreateTableSpec())
+        .create()
+        .asUnkeyedTable();
+    Assert.assertTrue(PropertyUtil.propertyAsBoolean(createTable.properties(),
+        TableProperties.ENABLE_SELF_OPTIMIZING, TableProperties.ENABLE_SELF_OPTIMIZING_DEFAULT));
+    getCatalog().dropTable(TableTestHelper.TEST_TABLE_ID, true);
+
+    CatalogMeta testCatalogMeta = TEST_AMS.getAmsHandler().getCatalog(CatalogTestHelper.TEST_CATALOG_NAME);
+    TEST_AMS.getAmsHandler().updateMeta(
+        testCatalogMeta,
+        CatalogMetaProperties.TABLE_PROPERTIES_PREFIX + TableProperties.LOG_STORE_ADDRESS,
+        "1.1.1.1");
+    TEST_AMS.getAmsHandler().updateMeta(
+        testCatalogMeta,
+        CatalogMetaProperties.TABLE_PROPERTIES_PREFIX + TableProperties.LOG_STORE_MESSAGE_TOPIC,
+        "test-topic");
+    TEST_AMS.getAmsHandler().updateMeta(
+        testCatalogMeta,
+        CatalogMetaProperties.TABLE_PROPERTIES_PREFIX + TableProperties.ENABLE_SELF_OPTIMIZING,
+        "false");
+    getCatalog().refresh();
+    createTable = getCatalog()
+        .newTableBuilder(TableTestHelper.TEST_TABLE_ID, getCreateTableSchema())
+        .withPartitionSpec(getCreateTableSpec())
+        .withProperty(TableProperties.ENABLE_LOG_STORE, "true")
+        .create()
+        .asUnkeyedTable();
+    Assert.assertFalse(PropertyUtil.propertyAsBoolean(createTable.properties(),
+        TableProperties.ENABLE_SELF_OPTIMIZING, TableProperties.ENABLE_SELF_OPTIMIZING_DEFAULT));
+  }
+
   @Test
   public void testUnkeyedRecoverableFileIO() throws TException {
     UnkeyedTable createTable = getCatalog()
diff --git a/core/src/test/java/com/netease/arctic/scan/TestBaseCombinedScanTask.java b/core/src/test/java/com/netease/arctic/scan/TestBaseCombinedScanTask.java
new file mode 100644
index 000000000..ff590c2fe
--- /dev/null
+++ b/core/src/test/java/com/netease/arctic/scan/TestBaseCombinedScanTask.java
@@ -0,0 +1,81 @@
+/*
+ * 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 com.netease.arctic.scan;
+
+import com.netease.arctic.IcebergFileEntry;
+import com.netease.arctic.data.DefaultKeyedFile;
+import com.netease.arctic.utils.ManifestEntryFields;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.DataFiles;
+import org.apache.iceberg.PartitionSpec;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.util.Collections;
+
+public class TestBaseCombinedScanTask {
+
+  @Test
+  public void testToString() {
+    IcebergFileEntry entry =
+        new IcebergFileEntry(
+            1L,
+            2L,
+            ManifestEntryFields.Status.ADDED,
+            DataFiles.builder(PartitionSpec.unpartitioned())
+                .withPath("/tmp/test.parquet")
+                .withFileSizeInBytes(10L)
+                .withRecordCount(100L)
+                .build()
+        );
+    DefaultKeyedFile keyedFile =
+        DefaultKeyedFile
+            .parseChange(
+                (DataFile) entry.getFile(),
+                entry.getSequenceNumber());
+    BasicArcticFileScanTask task =
+        new BasicArcticFileScanTask(
+            keyedFile,
+            null,
+            PartitionSpec.unpartitioned(),
+            null);
+
+
+    BaseCombinedScanTask baseCombinedScanTask =
+        new BaseCombinedScanTask(
+            new NodeFileScanTask(
+                Collections.singletonList(
+                    task
+                )));
+    String expected = "BaseCombinedScanTask{\n" +
+        "tasks=NodeFileScanTask{\n" +
+        "\tbaseTasks=[], \n" +
+        "\tinsertTasks=[DefaultKeyedFile{\n" +
+        "\t\tfile=/tmp/test.parquet, \n" +
+        "\t\ttype=I, \n" +
+        "\t\tmask=0, \n" +
+        "\t\tindex=0, \n" +
+        "\t\ttransactionId=2, \n" +
+        "\t\tfileSizeInBytes=10, \n" +
+        "\t\trecordCount=100}], \n" +
+        "\tdeleteFiles=[]}}";
+    Assert.assertEquals(expected, baseCombinedScanTask.toString());
+  }
+
+}
\ No newline at end of file