Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

[Feature] Add unsupported datatype check for all catalog #5890

Merged
merged 3 commits into from
Dec 10, 2023
Merged

[Feature] Add unsupported datatype check for all catalog #5890

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
1b25ca7
[Feature] Add unsupported datatype check for all catalog
Hisoka-X Nov 17, 2023
77eb6e1
update
Hisoka-X Nov 21, 2023
13f0be4
update
Hisoka-X Nov 21, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
76 changes: 65 additions & 11 deletions seatunnel-api/src/main/java/org/apache/seatunnel/api/table/catalog/Catalog.java
View file
Open in desktop
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
Open in desktop
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
Open in desktop
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
Open in desktop
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