Skip to content

PoC: Add support for DML with properties in engine + Add support for branching in Iceberg #23513

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

Closed
wants to merge 2 commits into from
Closed

PoC: Add support for DML with properties in engine + Add support for branching in Iceberg #23513

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
dd419d1
Add support for DML with properties in engine
ebyhr Sep 18, 2024
8f470b9
Add support for branching in Iceberg
ebyhr Sep 18, 2024
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
10 changes: 6 additions & 4 deletions core/trino-grammar/src/main/antlr4/io/trino/grammar/sql/SqlBase.g4
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -71,8 +71,10 @@ statement
(COMMENT string)?
(WITH properties)? #createTable
| DROP TABLE (IF EXISTS)? qualifiedName #dropTable
| INSERT INTO qualifiedName columnAliases? rootQuery #insertInto
| DELETE FROM qualifiedName (WHERE booleanExpression)? #delete
| INSERT INTO qualifiedName (WITH properties)?
columnAliases? rootQuery #insertInto
| DELETE FROM qualifiedName (WITH properties)?
(WHERE booleanExpression)? #delete
| TRUNCATE TABLE qualifiedName #truncateTable
| COMMENT ON TABLE qualifiedName IS (string | NULL) #commentTable
| COMMENT ON VIEW qualifiedName IS (string | NULL) #commentView
Expand Down Expand Up @@ -191,10 +193,10 @@ statement
| DESCRIBE OUTPUT identifier #describeOutput
| SET PATH pathSpecification #setPath
| SET TIME ZONE (LOCAL | expression) #setTimeZone
| UPDATE qualifiedName
| UPDATE qualifiedName (WITH properties)?
SET updateAssignment (',' updateAssignment)*
(WHERE where=booleanExpression)? #update
| MERGE INTO qualifiedName (AS? identifier)?
| MERGE INTO qualifiedName (AS? identifier)? (WITH properties)?
USING relation ON expression mergeCase+ #merge
;

Expand Down
4 changes: 2 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 @@ -834,12 +834,12 @@ default boolean isMaterializedView(Session session, QualifiedObjectName viewName
/**
* Get the target table handle after performing redirection with a table version.
*/
RedirectionAwareTableHandle getRedirectionAwareTableHandle(Session session, QualifiedObjectName tableName, Optional<TableVersion> startVersion, Optional<TableVersion> endVersion);
RedirectionAwareTableHandle getRedirectionAwareTableHandle(Session session, QualifiedObjectName tableName, Optional<TableVersion> startVersion, Optional<TableVersion> endVersion, Map<String, Object> properties);

/**
* Returns a table handle for the specified table name with a specified version
*/
Optional<TableHandle> getTableHandle(Session session, QualifiedObjectName tableName, Optional<TableVersion> startVersion, Optional<TableVersion> endVersion);
Optional<TableHandle> getTableHandle(Session session, QualifiedObjectName tableName, Optional<TableVersion> startVersion, Optional<TableVersion> endVersion, Map<String, Object> properties);

/**
* Returns maximum number of tasks that can be created while writing data to specific connector.
Expand Down
15 changes: 8 additions & 7 deletions core/trino-main/src/main/java/io/trino/metadata/MetadataManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -270,11 +270,11 @@ public List<String> listSchemaNames(Session session, String catalogName)
@Override
public Optional<TableHandle> getTableHandle(Session session, QualifiedObjectName table)
{
return getTableHandle(session, table, Optional.empty(), Optional.empty());
return getTableHandle(session, table, Optional.empty(), Optional.empty(), ImmutableMap.of());
}

@Override
public Optional<TableHandle> getTableHandle(Session session, QualifiedObjectName table, Optional<TableVersion> startVersion, Optional<TableVersion> endVersion)
public Optional<TableHandle> getTableHandle(Session session, QualifiedObjectName table, Optional<TableVersion> startVersion, Optional<TableVersion> endVersion, Map<String, Object> properties)
{
requireNonNull(table, "table is null");
if (cannotExist(table)) {
Expand All @@ -294,7 +294,8 @@ public Optional<TableHandle> getTableHandle(Session session, QualifiedObjectName
connectorSession,
table.asSchemaTableName(),
startTableVersion,
endTableVersion);
endTableVersion,
properties);
return Optional.ofNullable(tableHandle)
.map(connectorTableHandle -> new TableHandle(
catalogHandle,
Expand Down Expand Up @@ -1969,18 +1970,18 @@ private QualifiedObjectName getRedirectedTableName(Session session, QualifiedObj
@Override
public RedirectionAwareTableHandle getRedirectionAwareTableHandle(Session session, QualifiedObjectName tableName)
{
return getRedirectionAwareTableHandle(session, tableName, Optional.empty(), Optional.empty());
return getRedirectionAwareTableHandle(session, tableName, Optional.empty(), Optional.empty(), ImmutableMap.of());
}

@Override
public RedirectionAwareTableHandle getRedirectionAwareTableHandle(Session session, QualifiedObjectName tableName, Optional<TableVersion> startVersion, Optional<TableVersion> endVersion)
public RedirectionAwareTableHandle getRedirectionAwareTableHandle(Session session, QualifiedObjectName tableName, Optional<TableVersion> startVersion, Optional<TableVersion> endVersion, Map<String, Object> properties)
{
QualifiedObjectName targetTableName = getRedirectedTableName(session, tableName, startVersion, endVersion);
if (targetTableName.equals(tableName)) {
return noRedirection(getTableHandle(session, tableName, startVersion, endVersion));
return noRedirection(getTableHandle(session, tableName, startVersion, endVersion, properties));
}

Optional<TableHandle> tableHandle = getTableHandle(session, targetTableName, startVersion, endVersion);
Optional<TableHandle> tableHandle = getTableHandle(session, targetTableName, startVersion, endVersion, properties);
if (tableHandle.isPresent()) {
return withRedirectionTo(targetTableName, tableHandle.get());
}
Expand Down
74 changes: 68 additions & 6 deletions core/trino-main/src/main/java/io/trino/sql/analyzer/StatementAnalyzer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -579,8 +579,21 @@ protected Scope visitInsert(Insert insert, Optional<Scope> scope)
// analyze the query that creates the data
Scope queryScope = analyze(insert.getQuery(), Optional.empty(), false);

String catalogName = targetTable.catalogName();
CatalogHandle catalogHandle = getRequiredCatalogHandle(plannerContext.getMetadata(), session, insert, catalogName);

Map<String, Object> properties = tablePropertyManager.getProperties(
catalogName,
catalogHandle,
insert.getTable().getProperties(),
session,
plannerContext,
accessControl,
analysis.getParameters(),
true);

// verify the insert destination columns match the query
RedirectionAwareTableHandle redirection = metadata.getRedirectionAwareTableHandle(session, targetTable);
RedirectionAwareTableHandle redirection = metadata.getRedirectionAwareTableHandle(session, targetTable, Optional.empty(), Optional.empty(), properties);
Optional<TableHandle> targetTableHandle = redirection.tableHandle();
targetTable = redirection.redirectedTableName().orElse(targetTable);
if (targetTableHandle.isEmpty()) {
Expand Down Expand Up @@ -821,7 +834,19 @@ protected Scope visitDelete(Delete node, Optional<Scope> scope)
throw semanticException(NOT_SUPPORTED, node, "Deleting from views is not supported");
}

RedirectionAwareTableHandle redirection = metadata.getRedirectionAwareTableHandle(session, originalName);
String catalogName = originalName.catalogName();
CatalogHandle catalogHandle = getRequiredCatalogHandle(plannerContext.getMetadata(), session, node, catalogName);
Map<String, Object> properties = tablePropertyManager.getProperties(
catalogName,
catalogHandle,
node.getTable().getProperties(),
session,
plannerContext,
accessControl,
analysis.getParameters(),
true);

RedirectionAwareTableHandle redirection = metadata.getRedirectionAwareTableHandle(session, originalName, Optional.empty(), Optional.empty(), properties);
QualifiedObjectName tableName = redirection.redirectedTableName().orElse(originalName);
TableHandle handle = redirection.tableHandle()
.orElseThrow(() -> semanticException(TABLE_NOT_FOUND, table, "Table '%s' does not exist", tableName));
Expand Down Expand Up @@ -3375,7 +3400,19 @@ protected Scope visitUpdate(Update update, Optional<Scope> scope)

analysis.setUpdateType("UPDATE");

RedirectionAwareTableHandle redirection = metadata.getRedirectionAwareTableHandle(session, originalName);
String catalogName = originalName.catalogName();
CatalogHandle catalogHandle = getRequiredCatalogHandle(plannerContext.getMetadata(), session, update, catalogName);
Map<String, Object> properties = tablePropertyManager.getProperties(
catalogName,
catalogHandle,
update.getTable().getProperties(),
session,
plannerContext,
accessControl,
analysis.getParameters(),
true);

RedirectionAwareTableHandle redirection = metadata.getRedirectionAwareTableHandle(session, originalName, Optional.empty(), Optional.empty(), properties);
QualifiedObjectName tableName = redirection.redirectedTableName().orElse(originalName);
TableHandle handle = redirection.tableHandle()
.orElseThrow(() -> semanticException(TABLE_NOT_FOUND, table, "Table '%s' does not exist", tableName));
Expand Down Expand Up @@ -3505,7 +3542,19 @@ protected Scope visitMerge(Merge merge, Optional<Scope> scope)

analysis.setUpdateType("MERGE");

RedirectionAwareTableHandle redirection = metadata.getRedirectionAwareTableHandle(session, originalTableName);
String catalogName = originalTableName.catalogName();
CatalogHandle catalogHandle = getRequiredCatalogHandle(plannerContext.getMetadata(), session, merge, catalogName);
Map<String, Object> properties = tablePropertyManager.getProperties(
catalogName,
catalogHandle,
merge.getTargetTable().getProperties(),
session,
plannerContext,
accessControl,
analysis.getParameters(),
true);

RedirectionAwareTableHandle redirection = metadata.getRedirectionAwareTableHandle(session, originalTableName, Optional.empty(), Optional.empty(), properties);
QualifiedObjectName tableName = redirection.redirectedTableName().orElse(originalTableName);
TableHandle targetTableHandle = redirection.tableHandle()
.orElseThrow(() -> semanticException(TABLE_NOT_FOUND, table, "Table '%s' does not exist", tableName));
Expand Down Expand Up @@ -5860,12 +5909,25 @@ private OutputColumn createOutputColumn(Field field)
*/
private RedirectionAwareTableHandle getTableHandle(Table table, QualifiedObjectName name, Optional<Scope> scope)
{
String catalogName = name.catalogName();
CatalogHandle catalogHandle = getRequiredCatalogHandle(plannerContext.getMetadata(), session, table, catalogName);

Map<String, Object> properties = tablePropertyManager.getProperties(
catalogName,
catalogHandle,
table.getProperties(),
session,
plannerContext,
accessControl,
analysis.getParameters(),
true);

if (table.getQueryPeriod().isPresent()) {
Optional<TableVersion> startVersion = extractTableVersion(table, table.getQueryPeriod().get().getStart(), scope);
Optional<TableVersion> endVersion = extractTableVersion(table, table.getQueryPeriod().get().getEnd(), scope);
return metadata.getRedirectionAwareTableHandle(session, name, startVersion, endVersion);
return metadata.getRedirectionAwareTableHandle(session, name, startVersion, endVersion, properties);
}
return metadata.getRedirectionAwareTableHandle(session, name);
return metadata.getRedirectionAwareTableHandle(session, name, Optional.empty(), Optional.empty(), properties);
}

/**
Expand Down
9 changes: 9 additions & 0 deletions core/trino-main/src/main/java/io/trino/tracing/TracingConnectorMetadata.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -143,6 +143,15 @@ public ConnectorTableHandle getTableHandle(ConnectorSession session, SchemaTable
}
}

@Override
public ConnectorTableHandle getTableHandle(ConnectorSession session, SchemaTableName tableName, Optional<ConnectorTableVersion> startVersion, Optional<ConnectorTableVersion> endVersion, Map<String, Object> properties)
{
Span span = startSpan("getTableHandle", tableName);
try (var _ = scopedSpan(span)) {
return delegate.getTableHandle(session, tableName, startVersion, endVersion, properties);
}
}

@Override
public Optional<ConnectorTableExecuteHandle> getTableHandleForExecute(ConnectorSession session, ConnectorTableHandle tableHandle, String procedureName, Map<String, Object> executeProperties, RetryMode retryMode)
{
Expand Down
8 changes: 4 additions & 4 deletions core/trino-main/src/main/java/io/trino/tracing/TracingMetadata.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1497,20 +1497,20 @@ public RedirectionAwareTableHandle getRedirectionAwareTableHandle(Session sessio
}

@Override
public RedirectionAwareTableHandle getRedirectionAwareTableHandle(Session session, QualifiedObjectName tableName, Optional<TableVersion> startVersion, Optional<TableVersion> endVersion)
public RedirectionAwareTableHandle getRedirectionAwareTableHandle(Session session, QualifiedObjectName tableName, Optional<TableVersion> startVersion, Optional<TableVersion> endVersion, Map<String, Object> properties)
{
Span span = startSpan("getRedirectionAwareTableHandle", tableName);
try (var _ = scopedSpan(span)) {
return delegate.getRedirectionAwareTableHandle(session, tableName, startVersion, endVersion);
return delegate.getRedirectionAwareTableHandle(session, tableName, startVersion, endVersion, properties);
}
}

@Override
public Optional<TableHandle> getTableHandle(Session session, QualifiedObjectName tableName, Optional<TableVersion> startVersion, Optional<TableVersion> endVersion)
public Optional<TableHandle> getTableHandle(Session session, QualifiedObjectName tableName, Optional<TableVersion> startVersion, Optional<TableVersion> endVersion, Map<String, Object> properties)
{
Span span = startSpan("getTableHandle", tableName);
try (var _ = scopedSpan(span)) {
return delegate.getTableHandle(session, tableName, startVersion, endVersion);
return delegate.getTableHandle(session, tableName, startVersion, endVersion, properties);
}
}

Expand Down
3 changes: 2 additions & 1 deletion core/trino-main/src/test/java/io/trino/execution/TestCreateViewTask.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -51,6 +51,7 @@
import static io.trino.sql.QueryUtil.simpleQuery;
import static io.trino.sql.QueryUtil.table;
import static io.trino.sql.analyzer.StatementAnalyzerFactory.createTestingStatementAnalyzerFactory;
import static io.trino.testing.TestingHandles.TEST_CATALOG_HANDLE;
import static io.trino.testing.assertions.TrinoExceptionAssert.assertTrinoExceptionThrownBy;
import static org.assertj.core.api.Assertions.assertThat;
import static org.junit.jupiter.api.TestInstance.Lifecycle.PER_METHOD;
Expand All @@ -73,7 +74,7 @@ public void setUp()
createTestingStatementAnalyzerFactory(
plannerContext,
new AllowAllAccessControl(),
new TablePropertyManager(CatalogServiceProvider.fail()),
new TablePropertyManager(CatalogServiceProvider.singleton(TEST_CATALOG_HANDLE, ImmutableMap.of())),
new AnalyzePropertyManager(CatalogServiceProvider.fail())),
new StatementRewrite(ImmutableSet.of()),
plannerContext.getTracer());
Expand Down
9 changes: 6 additions & 3 deletions core/trino-main/src/test/java/io/trino/metadata/AbstractMockMetadata.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1003,13 +1003,16 @@ public RedirectionAwareTableHandle getRedirectionAwareTableHandle(Session sessio
}

@Override
public RedirectionAwareTableHandle getRedirectionAwareTableHandle(Session session, QualifiedObjectName tableName, Optional<TableVersion> startVersion, Optional<TableVersion> endVersion)
public RedirectionAwareTableHandle getRedirectionAwareTableHandle(Session session, QualifiedObjectName tableName, Optional<TableVersion> startVersion, Optional<TableVersion> endVersion, Map<String, Object> properties)
{
throw new UnsupportedOperationException();
if (startVersion.isPresent() || endVersion.isPresent() || !properties.isEmpty()) {
throw new UnsupportedOperationException();
}
return noRedirection(getTableHandle(session, tableName));
}

@Override
public Optional<TableHandle> getTableHandle(Session session, QualifiedObjectName table, Optional<TableVersion> startVersion, Optional<TableVersion> endVersion)
public Optional<TableHandle> getTableHandle(Session session, QualifiedObjectName table, Optional<TableVersion> startVersion, Optional<TableVersion> endVersion, Map<String, Object> properties)
{
throw new UnsupportedOperationException();
}
Expand Down
14 changes: 12 additions & 2 deletions core/trino-parser/src/main/java/io/trino/sql/SqlFormatter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1067,6 +1067,8 @@ protected Void visitMerge(Merge node, Integer indent)
builder.append("MERGE INTO ")
.append(formatName(node.getTargetTable().getName()));

builder.append(formatPropertiesMultiLine(node.getTargetTable().getProperties()));

node.getTargetAlias().ifPresent(value -> builder
.append(' ')
.append(formatName(value)));
Expand Down Expand Up @@ -1451,6 +1453,8 @@ protected Void visitDelete(Delete node, Integer indent)
builder.append("DELETE FROM ")
.append(formatName(node.getTable().getName()));

builder.append(formatPropertiesMultiLine(node.getTable().getProperties()));

node.getWhere().ifPresent(where -> builder
.append(" WHERE ")
.append(formatExpression(where)));
Expand Down Expand Up @@ -1877,6 +1881,8 @@ protected Void visitInsert(Insert node, Integer indent)
builder.append("INSERT INTO ")
.append(formatName(node.getTarget()));

builder.append(formatPropertiesMultiLine(node.getTable().getProperties()));

node.getColumns().ifPresent(columns -> builder
.append(" (")
.append(Joiner.on(", ").join(columns))
Expand All @@ -1893,8 +1899,12 @@ protected Void visitInsert(Insert node, Integer indent)
protected Void visitUpdate(Update node, Integer indent)
{
builder.append("UPDATE ")
.append(formatName(node.getTable().getName()))
.append(" SET");
.append(formatName(node.getTable().getName()));

builder.append(formatPropertiesMultiLine(node.getTable().getProperties()));

builder.append(" SET");

int setCounter = node.getAssignments().size() - 1;
for (UpdateAssignment assignment : node.getAssignments()) {
builder.append("\n")
Expand Down
Loading
Loading