ruanwenjun commented on code in PR #5485:
URL: https://github.com/apache/seatunnel/pull/5485#discussion_r1326714662
##########
seatunnel-api/src/main/java/org/apache/seatunnel/api/table/catalog/CatalogTableUtil.java:
##########
@@ -132,6 +135,54 @@ public static List<CatalogTable> getCatalogTables(Config
config, ClassLoader cla
.orElse(Collections.emptyList());
}
+ /**
+ * Get catalog table from config, if schema is specified, return a catalog
table with specified
+ * schema, otherwise, return a catalog table with schema from catalog.
+ *
+ * @deprecated DO NOT invoke it in any new
TableSourceFactory/TableSinkFactory, please directly
+ * use TableSourceFactory/TableSinkFactory instance to get
CatalogTable. We just use it to
+ * transition the old CatalogTable creation logic. Details please <a
+ *
href="https://cwiki.apache.org/confluence/display/SEATUNNEL/STIP5-Refactor+Catalog+and+CatalogTable";>check
+ * </a>
+ */
+ @Deprecated
+ public static List<CatalogTable> getCatalogTablesFromConfig(
+ Config config, ClassLoader classLoader) {
+ ReadonlyConfig readonlyConfig = ReadonlyConfig.fromConfig(config);
+
+ // We use plugin_name as factoryId, so MySQL-CDC should be MySQL
+ String factoryId =
readonlyConfig.get(CommonOptions.PLUGIN_NAME).replace("-CDC", "");
+ // Highest priority: specified schema
+ Map<String, String> schemaMap =
readonlyConfig.get(CatalogTableUtil.SCHEMA);
+ if (schemaMap != null) {
+ if (schemaMap.isEmpty()) {
+ throw new SeaTunnelException("Schema config can not be empty");
+ }
+ CatalogTable catalogTable =
CatalogTableUtil.buildWithConfig(config).getCatalogTable();
+ return Collections.singletonList(catalogTable);
+ }
+
+ Optional<Catalog> optionalCatalog =
+ FactoryUtil.createOptionalCatalog(
+ factoryId, readonlyConfig, classLoader, factoryId);
+ return optionalCatalog
+ .map(
+ c -> {
+ long startTime = System.currentTimeMillis();
+ try (Catalog catalog = c) {
+ catalog.open();
+ List<CatalogTable> catalogTables =
+ catalog.getTables(readonlyConfig);
+ log.info(
+ String.format(
+ "Get catalog tables, cost
time: %d",
+ System.currentTimeMillis() -
startTime));
+ return catalogTables;
+ }
+ })
+ .orElse(Collections.emptyList());
Review Comment:
Do we need to throw exception if the Catalog is not exist or catalogTable is
empty.
##########
seatunnel-api/src/test/java/org/apache/seatunnel/api/table/catalog/CatalogTableUtilTest.java:
##########
@@ -37,17 +36,13 @@
import java.net.URISyntaxException;
import java.net.URL;
import java.nio.file.Paths;
+import java.util.List;
public class CatalogTableUtilTest {
@Test
public void testSimpleSchemaParse() throws FileNotFoundException,
URISyntaxException {
String path = getTestConfigFile("/conf/simple.schema.conf");
- Config config =
Review Comment:
After remove this ,can we replace the system properties in config file?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
