Skip to content

Commit

Permalink
[FLINK-14490][table] Introduced temporary tables to CatalogManager
Browse files Browse the repository at this point in the history
  • Loading branch information
@dawidwys
dawidwys committed Oct 29, 2019
1 parent 96a62d9 commit c0f4b7f
Show file tree
Hide file tree
Showing 21 changed files with 862 additions and 259 deletions.
8 changes: 4 additions & 4 deletions ...client/src/test/java/org/apache/flink/table/client/gateway/local/LocalExecutorITCase.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -145,13 +145,13 @@ public void testValidateSession() throws Exception {

List<String> actualTables = executor.listTables(session);
List<String> expectedTables = Arrays.asList(
"AdditionalView1",
"AdditionalView2",
"TableNumber1",
"TableNumber2",
"TableSourceSink",
"TestView1",
"TestView2",
"AdditionalView1",
"AdditionalView2");
"TestView2");
assertEquals(expectedTables, actualTables);

session.removeView("AdditionalView1");
Expand Down Expand Up @@ -183,8 +183,8 @@ public void testListCatalogs() throws Exception {
final List<String> actualCatalogs = executor.listCatalogs(session);

final List<String> expectedCatalogs = Arrays.asList(
"default_catalog",
"catalog1",
"default_catalog",
"simple-catalog");
assertEquals(expectedCatalogs, actualCatalogs);
}
Expand Down
25 changes: 24 additions & 1 deletion ...table/flink-table-api-java/src/main/java/org/apache/flink/table/api/TableEnvironment.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -272,12 +272,35 @@ static TableEnvironment create(EnvironmentSettings settings) {
String[] listDatabases();

/**
* Gets the names of all tables registered in the current database of the current catalog.
* Gets the names of all tables available in the current namespace (the current database of the current catalog).
* It returns both temporary and permanent tables and views.
*
* @return A list of the names of all registered tables in the current database of the current catalog.
* @see #listTemporaryTables()
* @see #listTemporaryViews()
*/
String[] listTables();

/**
* Gets the names of all temporary tables and views available in the current namespace (the current
* database of the current catalog).
*
* @return A list of the names of all registered temporary tables and views in the current database
* of the current catalog.
* @see #listTables()
*/
String[] listTemporaryTables();

/**
* Gets the names of all temporary views available in the current namespace (the current
* database of the current catalog).
*
* @return A list of the names of all registered temporary views in the current database
* of the current catalog.
* @see #listTables()
*/
String[] listTemporaryViews();

/**
* Gets the names of all user defined functions registered in this environment.
*/
Expand Down
59 changes: 36 additions & 23 deletions ...able-api-java/src/main/java/org/apache/flink/table/api/internal/TableEnvironmentImpl.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -38,7 +38,6 @@
import org.apache.flink.table.catalog.ObjectIdentifier;
import org.apache.flink.table.catalog.QueryOperationCatalogView;
import org.apache.flink.table.catalog.UnresolvedIdentifier;
import org.apache.flink.table.catalog.exceptions.DatabaseNotExistException;
import org.apache.flink.table.delegation.Executor;
import org.apache.flink.table.delegation.ExecutorFactory;
import org.apache.flink.table.delegation.Parser;
Expand Down Expand Up @@ -199,7 +198,7 @@ public void registerTable(String name, Table table) {
}

CatalogBaseTable tableTable = new QueryOperationCatalogView(table.getQueryOperation());
catalogManager.createTable(tableTable, getTemporaryObjectIdentifier(name), false);
catalogManager.createTemporaryTable(tableTable, getTemporaryObjectIdentifier(name), false);
}

@Override
Expand Down Expand Up @@ -239,7 +238,7 @@ public Table scan(String... tablePath) {
private Optional<CatalogQueryOperation> scanInternal(String... tablePath) {
ObjectIdentifier objectIdentifier = catalogManager.qualifyIdentifier(UnresolvedIdentifier.of(tablePath));
return catalogManager.getTable(objectIdentifier)
.map(t -> new CatalogQueryOperation(objectIdentifier, t.getSchema()));
.map(t -> new CatalogQueryOperation(objectIdentifier, t.getTable().getSchema()));
}

@Override
Expand All @@ -249,7 +248,10 @@ public ConnectTableDescriptor connect(ConnectorDescriptor connectorDescriptor) {

@Override
public String[] listCatalogs() {
return catalogManager.getCatalogs().toArray(new String[0]);
return catalogManager.listCatalogs()
.stream()
.sorted()
.toArray(String[]::new);
}

@Override
Expand All @@ -267,17 +269,26 @@ public String[] listDatabases() {

@Override
public String[] listTables() {
String currentCatalogName = catalogManager.getCurrentCatalog();
Optional<Catalog> currentCatalog = catalogManager.getCatalog(currentCatalogName);

return currentCatalog.map(catalog -> {
try {
return catalog.listTables(catalogManager.getCurrentDatabase()).toArray(new String[0]);
} catch (DatabaseNotExistException e) {
throw new ValidationException("Current database does not exist", e);
}
}).orElseThrow(() ->
new TableException(String.format("The current catalog %s does not exist.", currentCatalogName)));
return catalogManager.listTables()
.stream()
.sorted()
.toArray(String[]::new);
}

@Override
public String[] listTemporaryTables() {
return catalogManager.listTemporaryTables()
.stream()
.sorted()
.toArray(String[]::new);
}

@Override
public String[] listTemporaryViews() {
return catalogManager.listTemporaryViews()
.stream()
.sorted()
.toArray(String[]::new);
}

@Override
Expand Down Expand Up @@ -465,7 +476,7 @@ private ObjectIdentifier getTemporaryObjectIdentifier(String name) {

private void registerTableSourceInternal(String name, TableSource<?> tableSource) {
validateTableSource(tableSource);
Optional<CatalogBaseTable> table = getCatalogTable(
Optional<CatalogBaseTable> table = getTemporaryTable(
catalogManager.getBuiltInCatalogName(),
catalogManager.getBuiltInDatabaseName(),
name);
Expand All @@ -482,20 +493,20 @@ private void registerTableSourceInternal(String name, TableSource<?> tableSource
tableSource,
sourceSinkTable.getTableSink().get(),
!IS_STREAM_TABLE);
catalogManager.alterTable(sourceAndSink, getTemporaryObjectIdentifier(name), false);
catalogManager.createTemporaryTable(sourceAndSink, getTemporaryObjectIdentifier(name), true);
}
} else {
throw new ValidationException(String.format(
"Table '%s' already exists. Please choose a different name.", name));
}
} else {
ConnectorCatalogTable source = ConnectorCatalogTable.source(tableSource, !IS_STREAM_TABLE);
catalogManager.createTable(source, getTemporaryObjectIdentifier(name), false);
catalogManager.createTemporaryTable(source, getTemporaryObjectIdentifier(name), false);
}
}

private void registerTableSinkInternal(String name, TableSink<?> tableSink) {
Optional<CatalogBaseTable> table = getCatalogTable(
Optional<CatalogBaseTable> table = getTemporaryTable(
catalogManager.getBuiltInCatalogName(),
catalogManager.getBuiltInDatabaseName(),
name);
Expand All @@ -510,20 +521,22 @@ private void registerTableSinkInternal(String name, TableSink<?> tableSink) {
// wrapper contains only sink (not source)
ConnectorCatalogTable sourceAndSink = ConnectorCatalogTable
.sourceAndSink(sourceSinkTable.getTableSource().get(), tableSink, !IS_STREAM_TABLE);
catalogManager.alterTable(sourceAndSink, getTemporaryObjectIdentifier(name), false);
catalogManager.createTemporaryTable(sourceAndSink, getTemporaryObjectIdentifier(name), true);
}
} else {
throw new ValidationException(String.format(
"Table '%s' already exists. Please choose a different name.", name));
}
} else {
ConnectorCatalogTable sink = ConnectorCatalogTable.sink(tableSink, !IS_STREAM_TABLE);
catalogManager.createTable(sink, getTemporaryObjectIdentifier(name), false);
catalogManager.createTemporaryTable(sink, getTemporaryObjectIdentifier(name), false);
}
}

private Optional<CatalogBaseTable> getCatalogTable(String... name) {
return catalogManager.getTable(catalogManager.qualifyIdentifier(UnresolvedIdentifier.of(name)));
private Optional<CatalogBaseTable> getTemporaryTable(String... name) {
return catalogManager.getTable(catalogManager.qualifyIdentifier(UnresolvedIdentifier.of(name)))
.filter(CatalogManager.TableLookupResult::isTemporary)
.map(CatalogManager.TableLookupResult::getTable);
}

protected TableImpl createTable(QueryOperation tableOperation) {
Expand Down
Loading

0 comments on commit c0f4b7f

Please sign in to comment.