Skip to content

Commit

Permalink
[Feature] Add unsupported datatype check for all catalog (#5890)
Browse files Browse the repository at this point in the history 
* [Feature] Add unsupported datatype check for all catalog

* update

* update
  • Loading branch information
@Hisoka-X
Hisoka-X authored Dec 10, 2023
1 parent 5aabb14 commit b979128
Show file tree
Hide file tree
Showing 12 changed files with 287 additions and 90 deletions.
76 changes: 65 additions & 11 deletions seatunnel-api/src/main/java/org/apache/seatunnel/api/table/catalog/Catalog.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -24,13 +24,20 @@
import org.apache.seatunnel.api.table.catalog.exception.TableAlreadyExistException;
import org.apache.seatunnel.api.table.catalog.exception.TableNotExistException;
import org.apache.seatunnel.api.table.factory.Factory;
import org.apache.seatunnel.common.exception.CommonError;
import org.apache.seatunnel.common.exception.CommonErrorCode;
import org.apache.seatunnel.common.exception.SeaTunnelRuntimeException;

import org.apache.commons.lang3.StringUtils;

import java.util.ArrayList;
import java.util.Collections;
import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.function.Function;
import java.util.regex.Pattern;

/**
Expand Down Expand Up @@ -58,6 +65,9 @@ default Optional<Factory> getFactory() {
*/
void close() throws CatalogException;

/** Get the name of the catalog. */
String name();

// --------------------------------------------------------------------------------------------
// database
// --------------------------------------------------------------------------------------------
Expand Down Expand Up @@ -124,15 +134,10 @@ default Optional<Factory> getFactory() {
default List<CatalogTable> getTables(ReadonlyConfig config) throws CatalogException {
// Get the list of specified tables
List<String> tableNames = config.get(CatalogOptions.TABLE_NAMES);
List<CatalogTable> catalogTables = new ArrayList<>();
if (tableNames != null && !tableNames.isEmpty()) {
for (String tableName : tableNames) {
TablePath tablePath = TablePath.of(tableName);
if (this.tableExists(tablePath)) {
catalogTables.add(this.getTable(tablePath));
}
}
return catalogTables;
Iterator<TablePath> tablePaths =
tableNames.stream().map(TablePath::of).filter(this::tableExists).iterator();
return buildCatalogTablesWithErrorCheck(tablePaths);
}

// Get the list of table pattern
Expand All @@ -144,17 +149,66 @@ default List<CatalogTable> getTables(ReadonlyConfig config) throws CatalogExcept
Pattern tablePattern = Pattern.compile(config.get(CatalogOptions.TABLE_PATTERN));
List<String> allDatabase = this.listDatabases();
allDatabase.removeIf(s -> !databasePattern.matcher(s).matches());
List<TablePath> tablePaths = new ArrayList<>();
for (String databaseName : allDatabase) {
tableNames = this.listTables(databaseName);
for (String tableName : tableNames) {
if (tablePattern.matcher(databaseName + "." + tableName).matches()) {
catalogTables.add(this.getTable(TablePath.of(databaseName, tableName)));
tableNames.forEach(
tableName -> {
if (tablePattern.matcher(databaseName + "." + tableName).matches()) {
tablePaths.add(TablePath.of(databaseName, tableName));
}
});
}
return buildCatalogTablesWithErrorCheck(tablePaths.iterator());
}

default List<CatalogTable> buildCatalogTablesWithErrorCheck(Iterator<TablePath> tablePaths) {
Map<String, Map<String, String>> unsupportedTable = new LinkedHashMap<>();
List<CatalogTable> catalogTables = new ArrayList<>();
while (tablePaths.hasNext()) {
try {
catalogTables.add(getTable(tablePaths.next()));
} catch (SeaTunnelRuntimeException e) {
if (e.getSeaTunnelErrorCode()
.equals(CommonErrorCode.GET_CATALOG_TABLE_WITH_UNSUPPORTED_TYPE_ERROR)) {
unsupportedTable.put(
e.getParams().get("tableName"),
e.getParamsValueAsMap("fieldWithDataTypes"));
} else {
throw e;
}
}
}
if (!unsupportedTable.isEmpty()) {
throw CommonError.getCatalogTablesWithUnsupportedType(name(), unsupportedTable);
}
return catalogTables;
}

default <T> void buildColumnsWithErrorCheck(
TablePath tablePath,
TableSchema.Builder builder,
Iterator<T> keys,
Function<T, Column> getColumn) {
Map<String, String> unsupported = new LinkedHashMap<>();
while (keys.hasNext()) {
try {
builder.column(getColumn.apply(keys.next()));
} catch (SeaTunnelRuntimeException e) {
if (e.getSeaTunnelErrorCode()
.equals(CommonErrorCode.CONVERT_TO_SEATUNNEL_TYPE_ERROR_SIMPLE)) {
unsupported.put(e.getParams().get("field"), e.getParams().get("dataType"));
} else {
throw e;
}
}
}
if (!unsupported.isEmpty()) {
throw CommonError.getCatalogTableWithUnsupportedType(
name(), tablePath.getFullName(), unsupported);
}
}

/**
* Create a new table in this catalog.
*
Expand Down
54 changes: 54 additions & 0 deletions seatunnel-api/src/test/java/org/apache/seatunnel/api/table/catalog/CatalogTableTest.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -17,9 +17,17 @@

package org.apache.seatunnel.api.table.catalog;

import org.apache.seatunnel.api.configuration.ReadonlyConfig;
import org.apache.seatunnel.common.exception.SeaTunnelRuntimeException;

import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;

import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.Map;

public class CatalogTableTest {

Expand All @@ -35,4 +43,50 @@ public void testCatalogTableModifyOptionsOrPartitionKeys() {
catalogTable.getOptions().put("test", "value");
catalogTable.getPartitionKeys().add("test");
}

@Test
public void testReadCatalogTableWithUnsupportedType() {
Catalog catalog =
new InMemoryCatalogFactory()
.createCatalog("InMemory", ReadonlyConfig.fromMap(new HashMap<>()));
SeaTunnelRuntimeException exception =
Assertions.assertThrows(
SeaTunnelRuntimeException.class,
() ->
catalog.getTables(
ReadonlyConfig.fromMap(
new HashMap<String, Object>() {
{
put(
CatalogOptions.TABLE_NAMES.key(),
Arrays.asList(
"unsupported.public.table1",
"unsupported.public.table2"));
}
})));
Assertions.assertEquals(
"ErrorCode:[COMMON-21], ErrorDescription:['InMemory' tables unsupported get catalog table，"
+ "the corresponding field types in the following tables are not supported: '{\"unsupported.public.table1\""
+ ":{\"field1\":\"interval\",\"field2\":\"interval2\"},\"unsupported.public.table2\":{\"field1\":\"interval\","
+ "\"field2\":\"interval2\"}}']",
exception.getMessage());
Map<String, Map<String, String>> result = new LinkedHashMap<>();
result.put(
"unsupported.public.table1",
new HashMap<String, String>() {
{
put("field1", "interval");
put("field2", "interval2");
}
});
result.put(
"unsupported.public.table2",
new HashMap<String, String>() {
{
put("field1", "interval");
put("field2", "interval2");
}
});
Assertions.assertEquals(result, exception.getParamsValueAs("tableUnsupportedTypes"));
}
}
48 changes: 46 additions & 2 deletions seatunnel-api/src/test/java/org/apache/seatunnel/api/table/catalog/InMemoryCatalog.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -25,11 +25,15 @@
import org.apache.seatunnel.api.table.catalog.exception.TableNotExistException;
import org.apache.seatunnel.api.table.type.BasicType;
import org.apache.seatunnel.api.table.type.LocalTimeType;
import org.apache.seatunnel.common.exception.CommonError;

import org.apache.commons.lang3.tuple.Pair;

import com.google.common.collect.Lists;
import lombok.extern.slf4j.Slf4j;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
Expand All @@ -42,6 +46,7 @@ public class InMemoryCatalog implements Catalog {
// database -> tables
private final Map<String, List<CatalogTable>> catalogTables;
private static final String DEFAULT_DATABASE = "default";
private static final String UNSUPPORTED_DATABASE = "unsupported";

InMemoryCatalog(String catalogName, ReadonlyConfig options) {
this.name = catalogName;
Expand All @@ -53,6 +58,7 @@ public class InMemoryCatalog implements Catalog {
// Add some default table for testing
private void addDefaultTable() {
this.catalogTables.put(DEFAULT_DATABASE, new ArrayList<>());
this.catalogTables.put(UNSUPPORTED_DATABASE, new ArrayList<>());
List<CatalogTable> tables = new ArrayList<>();
this.catalogTables.put("st", tables);
TableSchema tableSchema =
Expand Down Expand Up @@ -92,20 +98,40 @@ private void addDefaultTable() {
CatalogTable catalogTable1 =
CatalogTable.of(
TableIdentifier.of(name, TablePath.of("st", "public", "table1")),
tableSchema,
TableSchema.builder().build(),
new HashMap<>(),
new ArrayList<>(),
"In Memory Table");
CatalogTable catalogTable2 =
CatalogTable.of(
TableIdentifier.of(name, TablePath.of("st", "public", "table2")),
tableSchema,
TableSchema.builder().build(),
new HashMap<>(),
new ArrayList<>(),
"In Memory Table",
name);
tables.add(catalogTable1);
tables.add(catalogTable2);

CatalogTable unsupportedTable1 =
CatalogTable.of(
TableIdentifier.of(
name, TablePath.of(UNSUPPORTED_DATABASE, "public", "table1")),
tableSchema,
new HashMap<>(),
new ArrayList<>(),
"In Memory Table");
CatalogTable unsupportedTable2 =
CatalogTable.of(
TableIdentifier.of(
name, TablePath.of(UNSUPPORTED_DATABASE, "public", "table2")),
tableSchema,
new HashMap<>(),
new ArrayList<>(),
"In Memory Table",
name);
this.catalogTables.get(UNSUPPORTED_DATABASE).add(unsupportedTable1);
this.catalogTables.get(UNSUPPORTED_DATABASE).add(unsupportedTable2);
}

@Override
Expand All @@ -125,6 +151,11 @@ public void close() throws CatalogException {
log.trace(String.format("InMemoryCatalog %s closing", name));
}

@Override
public String name() {
return "InMemory";
}

@Override
public String getDefaultDatabase() throws CatalogException {
return DEFAULT_DATABASE;
Expand Down Expand Up @@ -165,6 +196,19 @@ public boolean tableExists(TablePath tablePath) throws CatalogException {
public CatalogTable getTable(TablePath tablePath)
throws CatalogException, TableNotExistException {
if (catalogTables.containsKey(tablePath.getDatabaseName())) {
if (tablePath.getDatabaseName().equals(UNSUPPORTED_DATABASE)) {
List<Pair<String, String>> unsupportedFields =
Arrays.asList(
Pair.of("field1", "interval"), Pair.of("field2", "interval2"));
buildColumnsWithErrorCheck(
tablePath,
new TableSchema.Builder(),
unsupportedFields.iterator(),
field -> {
throw CommonError.convertToSeaTunnelTypeError(
name(), field.getValue(), field.getKey());
});
}
List<CatalogTable> tables = catalogTables.get(tablePath.getDatabaseName());
return tables.stream()
.filter(t -> t.getTableId().toTablePath().equals(tablePath))
Expand Down
30 changes: 18 additions & 12 deletions ...org/apache/seatunnel/connectors/seatunnel/elasticsearch/catalog/ElasticSearchCatalog.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -91,6 +91,11 @@ public void close() throws CatalogException {
esRestClient.close();
}

@Override
public String name() {
return catalogName;
}

@Override
public String getDefaultDatabase() throws CatalogException {
return defaultDatabase;
Expand Down Expand Up @@ -142,19 +147,20 @@ public CatalogTable getTable(TablePath tablePath)
TableSchema.Builder builder = TableSchema.builder();
Map<String, String> fieldTypeMapping =
esRestClient.getFieldTypeMapping(tablePath.getTableName(), Collections.emptyList());
fieldTypeMapping.forEach(
(fieldName, fieldType) -> {
buildColumnsWithErrorCheck(
tablePath,
builder,
fieldTypeMapping.entrySet().iterator(),
nameAndType -> {
// todo: we need to add a new type TEXT or add length in STRING type
PhysicalColumn physicalColumn =
PhysicalColumn.of(
fieldName,
elasticSearchDataTypeConvertor.toSeaTunnelType(
fieldName, fieldType),
null,
true,
null,
null);
builder.column(physicalColumn);
return PhysicalColumn.of(
nameAndType.getKey(),
elasticSearchDataTypeConvertor.toSeaTunnelType(
nameAndType.getKey(), nameAndType.getValue()),
null,
true,
null,
null);
});

return CatalogTable.of(
Expand Down
Loading

0 comments on commit b979128

Please sign in to comment.