Skip to content

Add table redirection support in SPI #7606

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

Merged
phd3 merged 1 commit into from
Jun 1, 2021
Merged

Add table redirection support in SPI #7606

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
24 changes: 23 additions & 1 deletion ...no-main/src/main/java/io/trino/connector/informationschema/InformationSchemaMetadata.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,7 @@
import io.trino.metadata.Metadata;
import io.trino.metadata.QualifiedObjectName;
import io.trino.metadata.QualifiedTablePrefix;
import io.trino.spi.TrinoException;
import io.trino.spi.connector.ColumnHandle;
import io.trino.spi.connector.ColumnMetadata;
import io.trino.spi.connector.ConnectorMetadata;
Expand Down Expand Up @@ -65,6 +66,7 @@
import static io.trino.connector.informationschema.InformationSchemaTable.TABLE_PRIVILEGES;
import static io.trino.connector.informationschema.InformationSchemaTable.VIEWS;
import static io.trino.metadata.MetadataUtil.findColumnMetadata;
import static io.trino.spi.StandardErrorCode.TABLE_REDIRECTION_ERROR;
import static io.trino.spi.type.VarcharType.createUnboundedVarcharType;
import static java.util.Collections.emptyList;
import static java.util.Locale.ENGLISH;
Expand Down Expand Up @@ -353,7 +355,27 @@ private Set<QualifiedTablePrefix> calculatePrefixesWithTableName(
.flatMap(prefix -> tables.get().stream()
.filter(this::isLowerCase)
.map(table -> new QualifiedObjectName(catalogName, prefix.getSchemaName().get(), table)))
.filter(objectName -> !isColumnsEnumeratingTable(informationSchemaTable) || metadata.getTableHandle(session, objectName).isPresent() || metadata.getView(session, objectName).isPresent())
Comment thread
phd3 marked this conversation as resolved.
Outdated
.filter(objectName -> {
if (!isColumnsEnumeratingTable(informationSchemaTable) || metadata.getView(session, objectName).isPresent()) {
return true;
}

// This is a columns enumerating table and the object is not a view
try {
// Table redirection to enumerate columns from target table happens later in
// MetadataListing#listTableColumns, but also applying it here to avoid incorrect
// filtering in case the source table does not exist or there is a problem with redirection.
return metadata.getRedirectionAwareTableHandle(session, objectName).getTableHandle().isPresent();
}
catch (TrinoException e) {
if (e.getErrorCode().equals(TABLE_REDIRECTION_ERROR.toErrorCode())) {
// Ignore redirection errors for listing, treat as if the table does not exist
return false;
}

throw e;
}
})
.filter(objectName -> predicate.isEmpty() || predicate.get().test(asFixedValues(objectName)))
.map(QualifiedObjectName::asQualifiedTablePrefix)
.collect(toImmutableSet());
Expand Down
2 changes: 1 addition & 1 deletion core/trino-main/src/main/java/io/trino/connector/system/TableCommentSystemTable.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -108,7 +108,7 @@ public RecordCursor cursor(ConnectorTransactionHandle transactionHandle, Connect
QualifiedObjectName tableName = new QualifiedObjectName(prefix.getCatalogName(), name.getSchemaName(), name.getTableName());
Optional<String> comment = Optional.empty();
try {
comment = metadata.getTableHandle(session, tableName)
comment = metadata.getRedirectionAwareTableHandle(session, tableName).getTableHandle()
.map(handle -> metadata.getTableMetadata(session, handle))
.map(metadata -> metadata.getMetadata().getComment())
.get();
Expand Down
22 changes: 16 additions & 6 deletions core/trino-main/src/main/java/io/trino/execution/CreateTableTask.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,7 @@
import io.trino.execution.warnings.WarningCollector;
import io.trino.metadata.Metadata;
import io.trino.metadata.QualifiedObjectName;
import io.trino.metadata.RedirectionAwareTableHandle;
import io.trino.metadata.TableHandle;
import io.trino.metadata.TableMetadata;
import io.trino.security.AccessControl;
Expand Down Expand Up @@ -75,6 +76,7 @@
import static io.trino.sql.tree.LikeClause.PropertiesOption.EXCLUDING;
import static io.trino.sql.tree.LikeClause.PropertiesOption.INCLUDING;
import static io.trino.type.UnknownType.UNKNOWN;
import static java.lang.String.format;

public class CreateTableTask
implements DataDefinitionTask<CreateTable>
Expand Down Expand Up @@ -166,15 +168,23 @@ ListenableFuture<?> internalExecute(CreateTable statement, Metadata metadata, Ac
}
else if (element instanceof LikeClause) {
LikeClause likeClause = (LikeClause) element;
QualifiedObjectName likeTableName = createQualifiedObjectName(session, statement, likeClause.getTableName());
if (metadata.getCatalogHandle(session, likeTableName.getCatalogName()).isEmpty()) {
throw semanticException(CATALOG_NOT_FOUND, statement, "LIKE table catalog '%s' does not exist", likeTableName.getCatalogName());
QualifiedObjectName originalLikeTableName = createQualifiedObjectName(session, statement, likeClause.getTableName());
if (metadata.getCatalogHandle(session, originalLikeTableName.getCatalogName()).isEmpty()) {
throw semanticException(CATALOG_NOT_FOUND, statement, "LIKE table catalog '%s' does not exist", originalLikeTableName.getCatalogName());
}

RedirectionAwareTableHandle redirection = metadata.getRedirectionAwareTableHandle(session, originalLikeTableName);
TableHandle likeTable = redirection.getTableHandle()
.orElseThrow(() -> semanticException(TABLE_NOT_FOUND, statement, "LIKE table '%s' does not exist", originalLikeTableName));

QualifiedObjectName likeTableName = redirection.getRedirectedTableName().orElse(originalLikeTableName);
if (!tableName.getCatalogName().equals(likeTableName.getCatalogName())) {
throw semanticException(NOT_SUPPORTED, statement, "LIKE table across catalogs is not supported");
String message = "CREATE TABLE LIKE across catalogs is not supported";
if (!originalLikeTableName.equals(likeTableName)) {
message += format(". LIKE table '%s' redirected to '%s'.", originalLikeTableName, likeTableName);
}
throw semanticException(NOT_SUPPORTED, statement, message);
}
TableHandle likeTable = metadata.getTableHandle(session, likeTableName)
.orElseThrow(() -> semanticException(TABLE_NOT_FOUND, statement, "LIKE table '%s' does not exist", likeTableName));

TableMetadata likeTableMetadata = metadata.getTableMetadata(session, likeTable);

Expand Down
12 changes: 10 additions & 2 deletions core/trino-main/src/main/java/io/trino/metadata/Metadata.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -43,6 +43,7 @@
import io.trino.spi.connector.SampleType;
import io.trino.spi.connector.SortItem;
import io.trino.spi.connector.SystemTable;
import io.trino.spi.connector.TableColumnsMetadata;
import io.trino.spi.connector.TableScanRedirectApplicationResult;
import io.trino.spi.connector.TopNApplicationResult;
import io.trino.spi.expression.ConnectorExpression;
Expand Down Expand Up @@ -157,9 +158,10 @@ public interface Metadata
ColumnMetadata getColumnMetadata(Session session, TableHandle tableHandle, ColumnHandle columnHandle);

/**
* Gets the metadata for all columns that match the specified table prefix.
* Gets the columns metadata for all tables that match the specified prefix.
* TODO: consider returning a stream for more efficient processing
*/
Map<QualifiedObjectName, List<ColumnMetadata>> listTableColumns(Session session, QualifiedTablePrefix prefix);
Map<CatalogName, List<TableColumnsMetadata>> listTableColumns(Session session, QualifiedTablePrefix prefix);
Comment thread
phd3 marked this conversation as resolved.
Outdated
Comment thread
phd3 marked this conversation as resolved.
Outdated

/**
* Creates a schema.
Expand Down Expand Up @@ -640,4 +642,10 @@ default ResolvedFunction getCoercion(Type fromType, Type toType)
* This method is called after security checks against the original table.
*/
Optional<TableScanRedirectApplicationResult> applyTableScanRedirect(Session session, TableHandle tableHandle);

/**
* Get the target table handle after performing redirection.
*
*/
RedirectionAwareTableHandle getRedirectionAwareTableHandle(Session session, QualifiedObjectName tableName);
}
67 changes: 57 additions & 10 deletions core/trino-main/src/main/java/io/trino/metadata/MetadataListing.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,10 +20,11 @@
import io.trino.Session;
import io.trino.connector.CatalogName;
import io.trino.security.AccessControl;
import io.trino.spi.connector.CatalogSchemaTableName;
import io.trino.spi.TrinoException;
import io.trino.spi.connector.ColumnMetadata;
import io.trino.spi.connector.ConnectorViewDefinition;
import io.trino.spi.connector.SchemaTableName;
import io.trino.spi.connector.TableColumnsMetadata;
import io.trino.spi.security.GrantInfo;

import java.util.List;
Expand All @@ -37,6 +38,8 @@
import static com.google.common.collect.ImmutableList.toImmutableList;
import static com.google.common.collect.ImmutableMap.toImmutableMap;
import static com.google.common.collect.ImmutableSet.toImmutableSet;
import static com.google.common.collect.Iterables.getOnlyElement;
import static io.trino.spi.StandardErrorCode.TABLE_REDIRECTION_ERROR;

public final class MetadataListing
{
Expand Down Expand Up @@ -132,31 +135,75 @@ public static Set<GrantInfo> listTablePrivileges(Session session, Metadata metad

public static Map<SchemaTableName, List<ColumnMetadata>> listTableColumns(Session session, Metadata metadata, AccessControl accessControl, QualifiedTablePrefix prefix)
{
Map<SchemaTableName, List<ColumnMetadata>> tableColumns = metadata.listTableColumns(session, prefix).entrySet().stream()
.collect(toImmutableMap(entry -> entry.getKey().asSchemaTableName(), Entry::getValue));
List<TableColumnsMetadata> catalogColumns = getOnlyElement(metadata.listTableColumns(session, prefix).values(), List.of());

Map<SchemaTableName, Optional<List<ColumnMetadata>>> tableColumns = catalogColumns.stream()
.collect(toImmutableMap(TableColumnsMetadata::getTable, TableColumnsMetadata::getColumns));

Set<SchemaTableName> allowedTables = accessControl.filterTables(
session.toSecurityContext(),
prefix.getCatalogName(),
tableColumns.keySet());

ImmutableMap.Builder<SchemaTableName, List<ColumnMetadata>> result = ImmutableMap.builder();
for (Entry<SchemaTableName, List<ColumnMetadata>> entry : tableColumns.entrySet()) {
if (!allowedTables.contains(entry.getKey())) {
continue;

tableColumns.forEach((table, columnsOptional) -> {
if (!allowedTables.contains(table)) {
return;
}
List<ColumnMetadata> columns = entry.getValue();

QualifiedObjectName originalTableName = new QualifiedObjectName(prefix.getCatalogName(), table.getSchemaName(), table.getTableName());
List<ColumnMetadata> columns;
Optional<QualifiedObjectName> targetTableName = Optional.empty();

if (columnsOptional.isPresent()) {
Comment thread
phd3 marked this conversation as resolved.
Outdated
columns = columnsOptional.get();
}
else {
TableHandle targetTableHandle = null;
boolean redirectionSucceeded = false;

try {
// Handle redirection before filterColumns check
RedirectionAwareTableHandle redirection = metadata.getRedirectionAwareTableHandle(session, originalTableName);
targetTableName = redirection.getRedirectedTableName();

// The target table name should be non-empty. If it is empty, it means that there is an
// inconsistency in the connector's implementation of ConnectorMetadata#streamTableColumns and
// ConnectorMetadata#redirectTable.
if (targetTableName.isPresent()) {
redirectionSucceeded = true;
targetTableHandle = redirection.getTableHandle().orElseThrow();
}
}
catch (TrinoException e) {
// Ignore redirection errors
if (!e.getErrorCode().equals(TABLE_REDIRECTION_ERROR.toErrorCode())) {
throw e;
}
}

if (redirectionSucceeded == false) {
return;
}

columns = metadata.getTableMetadata(session, targetTableHandle).getColumns();
}

Set<String> allowedColumns = accessControl.filterColumns(
session.toSecurityContext(),
new CatalogSchemaTableName(prefix.getCatalogName(), entry.getKey()),
// Use redirected table name for applying column filters
targetTableName.orElse(originalTableName).asCatalogSchemaTableName(),
columns.stream()
.map(ColumnMetadata::getName)
.collect(toImmutableSet()));
result.put(
entry.getKey(),
table,
columns.stream()
.filter(column -> allowedColumns.contains(column.getName()))
.collect(toImmutableList()));
}
});

return result.build();
}
}
Loading