From 28874246813d3e5a59b902cd97cdd31d71afb0d3 Mon Sep 17 00:00:00 2001
From: Jia Fan <fanjiaeminem@qq.com>
Date: Wed, 20 Sep 2023 15:48:48 +0800
Subject: [PATCH] [Improve] Remove CatalogTable field in CatalogTableUtil
 (#5521)

---
 .../api/table/catalog/CatalogTable.java       |  6 ++++
 .../api/table/catalog/CatalogTableUtil.java   | 32 ++++++-------------
 2 files changed, 16 insertions(+), 22 deletions(-)

diff --git a/seatunnel-api/src/main/java/org/apache/seatunnel/api/table/catalog/CatalogTable.java b/seatunnel-api/src/main/java/org/apache/seatunnel/api/table/catalog/CatalogTable.java
index 1be6de02841..5b1f1248631 100644
--- a/seatunnel-api/src/main/java/org/apache/seatunnel/api/table/catalog/CatalogTable.java
+++ b/seatunnel-api/src/main/java/org/apache/seatunnel/api/table/catalog/CatalogTable.java
@@ -17,6 +17,8 @@
 
 package org.apache.seatunnel.api.table.catalog;
 
+import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
+
 import java.io.Serializable;
 import java.util.List;
 import java.util.Map;
@@ -91,6 +93,10 @@ public TableSchema getTableSchema() {
         return tableSchema;
     }
 
+    public SeaTunnelRowType getSeaTunnelRowType() {
+        return tableSchema.toPhysicalRowDataType();
+    }
+
     public Map<String, String> getOptions() {
         return options;
     }
diff --git a/seatunnel-api/src/main/java/org/apache/seatunnel/api/table/catalog/CatalogTableUtil.java b/seatunnel-api/src/main/java/org/apache/seatunnel/api/table/catalog/CatalogTableUtil.java
index 07c3473bff9..8439222b536 100644
--- a/seatunnel-api/src/main/java/org/apache/seatunnel/api/table/catalog/CatalogTableUtil.java
+++ b/seatunnel-api/src/main/java/org/apache/seatunnel/api/table/catalog/CatalogTableUtil.java
@@ -42,7 +42,6 @@
 import org.apache.seatunnel.common.utils.JsonUtils;
 import org.apache.seatunnel.common.utils.SeaTunnelException;
 
-import lombok.Getter;
 import lombok.extern.slf4j.Slf4j;
 
 import java.io.Serializable;
@@ -70,12 +69,6 @@ public class CatalogTableUtil implements Serializable {
             new SeaTunnelRowType(
                     new String[] {"content"}, new SeaTunnelDataType<?>[] {BasicType.STRING_TYPE});
 
-    @Getter private final CatalogTable catalogTable;
-
-    private CatalogTableUtil(CatalogTable catalogTable) {
-        this.catalogTable = catalogTable;
-    }
-
     @Deprecated
     public static CatalogTable getCatalogTable(String tableName, SeaTunnelRowType rowType) {
         TableSchema.Builder schemaBuilder = TableSchema.builder();
@@ -108,7 +101,7 @@ public static List<CatalogTable> getCatalogTables(Config config, ClassLoader cla
         // Highest priority: specified schema
         Map<String, String> schemaMap = readonlyConfig.get(CatalogTableUtil.SCHEMA);
         if (schemaMap != null && schemaMap.size() > 0) {
-            CatalogTable catalogTable = CatalogTableUtil.buildWithConfig(config).getCatalogTable();
+            CatalogTable catalogTable = CatalogTableUtil.buildWithConfig(config);
             return Collections.singletonList(catalogTable);
         }
 
@@ -158,7 +151,7 @@ public static List<CatalogTable> getCatalogTablesFromConfig(
             if (schemaMap.isEmpty()) {
                 throw new SeaTunnelException("Schema config can not be empty");
             }
-            CatalogTable catalogTable = CatalogTableUtil.buildWithConfig(config).getCatalogTable();
+            CatalogTable catalogTable = CatalogTableUtil.buildWithConfig(config);
             return Collections.singletonList(catalogTable);
         }
 
@@ -194,31 +187,26 @@ public static List<CatalogTable> getCatalogTablesFromConfig(
                                                 factoryId)));
     }
 
-    public static CatalogTableUtil buildWithConfig(Config config) {
+    public static CatalogTable buildWithConfig(Config config) {
         CheckResult checkResult = CheckConfigUtil.checkAllExists(config, "schema");
         if (!checkResult.isSuccess()) {
             throw new RuntimeException(
                     "Schema config need option [schema], please correct your config first");
         }
         TableSchema tableSchema = parseTableSchema(config.getConfig("schema"));
-        return new CatalogTableUtil(
-                CatalogTable.of(
-                        // TODO: other table info
-                        TableIdentifier.of("", "", ""),
-                        tableSchema,
-                        new HashMap<>(),
-                        new ArrayList<>(),
-                        ""));
+        return CatalogTable.of(
+                // TODO: other table info
+                TableIdentifier.of("", "", ""),
+                tableSchema,
+                new HashMap<>(),
+                new ArrayList<>(),
+                "");
     }
 
     public static SeaTunnelRowType buildSimpleTextSchema() {
         return SIMPLE_SCHEMA;
     }
 
-    public SeaTunnelRowType getSeaTunnelRowType() {
-        return catalogTable.getTableSchema().toPhysicalRowDataType();
-    }
-
     public static SeaTunnelDataType<?> parseDataType(String columnStr) {
         SqlType sqlType = null;
         try {