Skip to content

Add time travel syntax support for iceberg tables (VERSION and TIMESTAMP). #20991

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
tdcmeehan merged 1 commit into from
Nov 17, 2023
Merged

Add time travel syntax support for iceberg tables (VERSION and TIMESTAMP). #20991

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
7 changes: 7 additions & 0 deletions presto-main/src/main/java/com/facebook/presto/metadata/DelegatingMetadataManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@
import com.facebook.presto.Session;
import com.facebook.presto.common.CatalogSchemaName;
import com.facebook.presto.common.QualifiedObjectName;
import com.facebook.presto.common.block.Block;
import com.facebook.presto.common.block.BlockEncodingSerde;
import com.facebook.presto.common.predicate.TupleDomain;
import com.facebook.presto.common.type.Type;
Expand Down Expand Up @@ -98,6 +99,12 @@ public Optional<SystemTable> getSystemTable(Session session, QualifiedObjectName
return delegate.getSystemTable(session, tableName);
}

@Override
public Optional<TableHandle> getHandleVersion(Session session, QualifiedObjectName tableName, Optional<Block> tableVersionBlock)
{
return delegate.getHandleVersion(session, tableName, tableVersionBlock);
}

@Override
public Optional<TableHandle> getTableHandleForStatisticsCollection(Session session, QualifiedObjectName tableName, Map<String, Object> analyzeProperties)
{
Expand Down
6 changes: 6 additions & 0 deletions presto-main/src/main/java/com/facebook/presto/metadata/Metadata.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@
import com.facebook.presto.Session;
import com.facebook.presto.common.CatalogSchemaName;
import com.facebook.presto.common.QualifiedObjectName;
import com.facebook.presto.common.block.Block;
import com.facebook.presto.common.block.BlockEncodingSerde;
import com.facebook.presto.common.predicate.TupleDomain;
import com.facebook.presto.common.type.Type;
Expand Down Expand Up @@ -73,6 +74,11 @@ public interface Metadata

Optional<SystemTable> getSystemTable(Session session, QualifiedObjectName tableName);

/**
* Returns a table handle for time travel expression
*/
Optional<TableHandle> getHandleVersion(Session session, QualifiedObjectName tableName, Optional<Block> tableVersionBlock);

Optional<TableHandle> getTableHandleForStatisticsCollection(Session session, QualifiedObjectName tableName, Map<String, Object> analyzeProperties);

/**
Expand Down
13 changes: 10 additions & 3 deletions presto-main/src/main/java/com/facebook/presto/metadata/MetadataManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@
import com.facebook.presto.Session;
import com.facebook.presto.common.CatalogSchemaName;
import com.facebook.presto.common.QualifiedObjectName;
import com.facebook.presto.common.block.Block;
import com.facebook.presto.common.block.BlockEncodingManager;
import com.facebook.presto.common.block.BlockEncodingSerde;
import com.facebook.presto.common.function.OperatorType;
Expand Down Expand Up @@ -326,6 +327,12 @@ public Optional<TableHandle> getTableHandleForStatisticsCollection(Session sessi
return Optional.empty();
}

@Override
public Optional<TableHandle> getHandleVersion(Session session, QualifiedObjectName tableName, Optional<Block> tableVersionBlock)
{
return getOptionalTableHandle(session, transactionManager, tableName, tableVersionBlock);
}

@Override
public Optional<SystemTable> getSystemTable(Session session, QualifiedObjectName tableName)
{
Expand Down Expand Up @@ -1010,7 +1017,7 @@ public void dropMaterializedView(Session session, QualifiedObjectName viewName)

private MaterializedViewStatus getMaterializedViewStatus(Session session, QualifiedObjectName materializedViewName, TupleDomain<String> baseQueryDomain)
{
Optional<TableHandle> materializedViewHandle = getOptionalTableHandle(session, transactionManager, materializedViewName);
Optional<TableHandle> materializedViewHandle = getOptionalTableHandle(session, transactionManager, materializedViewName, Optional.empty());

ConnectorId connectorId = materializedViewHandle.get().getConnectorId();
ConnectorMetadata metadata = getMetadata(session, connectorId);
Expand Down Expand Up @@ -1322,13 +1329,13 @@ public boolean schemaExists(CatalogSchemaName schema)
@Override
public boolean tableExists(QualifiedObjectName tableName)
{
return getOptionalTableHandle(session, transactionManager, tableName).isPresent();
return getOptionalTableHandle(session, transactionManager, tableName, Optional.empty()).isPresent();
}

@Override
public Optional<TableHandle> getTableHandle(QualifiedObjectName tableName)
{
return getOptionalTableHandle(session, transactionManager, tableName);
return getOptionalTableHandle(session, transactionManager, tableName, Optional.empty());
}

@Override
Expand Down
8 changes: 6 additions & 2 deletions presto-main/src/main/java/com/facebook/presto/metadata/MetadataUtil.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@
import com.facebook.presto.Session;
import com.facebook.presto.common.CatalogSchemaName;
import com.facebook.presto.common.QualifiedObjectName;
import com.facebook.presto.common.block.Block;
import com.facebook.presto.common.type.Type;
import com.facebook.presto.spi.ColumnMetadata;
import com.facebook.presto.spi.ConnectorId;
Expand Down Expand Up @@ -167,7 +168,7 @@ public static Optional<CatalogMetadata> getOptionalCatalogMetadata(Session sessi
return transactionManager.getOptionalCatalogMetadata(session.getRequiredTransactionId(), catalogName);
}

public static Optional<TableHandle> getOptionalTableHandle(Session session, TransactionManager transactionManager, QualifiedObjectName table)
public static Optional<TableHandle> getOptionalTableHandle(Session session, TransactionManager transactionManager, QualifiedObjectName table, Optional<Block> tableVersionBlock)
{
requireNonNull(table, "table is null");

Expand All @@ -177,7 +178,10 @@ public static Optional<TableHandle> getOptionalTableHandle(Session session, Tran
ConnectorId connectorId = catalogMetadata.getConnectorId(session, table);
ConnectorMetadata metadata = catalogMetadata.getMetadataFor(connectorId);

ConnectorTableHandle tableHandle = metadata.getTableHandle(session.toConnectorSession(connectorId), toSchemaTableName(table));
ConnectorTableHandle tableHandle;
tableHandle = tableVersionBlock
.map(expression -> metadata.getTableHandle(session.toConnectorSession(connectorId), toSchemaTableName(table), Optional.of(expression)))
.orElseGet(() -> metadata.getTableHandle(session.toConnectorSession(connectorId), toSchemaTableName(table)));
if (tableHandle != null) {
return Optional.of(new TableHandle(
connectorId,
Expand Down
56 changes: 55 additions & 1 deletion presto-main/src/main/java/com/facebook/presto/sql/analyzer/StatementAnalyzer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,14 +18,18 @@
import com.facebook.presto.SystemSessionProperties;
import com.facebook.presto.common.QualifiedObjectName;
import com.facebook.presto.common.Subfield;
import com.facebook.presto.common.block.Block;
import com.facebook.presto.common.block.BlockBuilder;
import com.facebook.presto.common.function.OperatorType;
import com.facebook.presto.common.predicate.Domain;
import com.facebook.presto.common.predicate.TupleDomain;
import com.facebook.presto.common.type.ArrayType;
import com.facebook.presto.common.type.BigintType;
import com.facebook.presto.common.type.DoubleType;
import com.facebook.presto.common.type.MapType;
import com.facebook.presto.common.type.RealType;
import com.facebook.presto.common.type.RowType;
import com.facebook.presto.common.type.TimestampWithTimeZoneType;
import com.facebook.presto.common.type.Type;
import com.facebook.presto.metadata.Metadata;
import com.facebook.presto.metadata.OperatorNotFoundException;
Expand Down Expand Up @@ -147,6 +151,7 @@
import com.facebook.presto.sql.tree.Statement;
import com.facebook.presto.sql.tree.Table;
import com.facebook.presto.sql.tree.TableSubquery;
import com.facebook.presto.sql.tree.TableVersionExpression;
import com.facebook.presto.sql.tree.TruncateTable;
import com.facebook.presto.sql.tree.Union;
import com.facebook.presto.sql.tree.Unnest;
Expand Down Expand Up @@ -188,10 +193,12 @@
import static com.facebook.presto.common.type.BooleanType.BOOLEAN;
import static com.facebook.presto.common.type.DoubleType.DOUBLE;
import static com.facebook.presto.common.type.TypeSignature.parseTypeSignature;
import static com.facebook.presto.common.type.TypeUtils.writeNativeValue;
import static com.facebook.presto.common.type.UnknownType.UNKNOWN;
import static com.facebook.presto.common.type.VarcharType.VARCHAR;
import static com.facebook.presto.metadata.MetadataUtil.createQualifiedObjectName;
import static com.facebook.presto.metadata.MetadataUtil.toSchemaTableName;
import static com.facebook.presto.spi.StandardErrorCode.INVALID_ARGUMENTS;
import static com.facebook.presto.spi.StandardErrorCode.INVALID_FUNCTION_ARGUMENT;
import static com.facebook.presto.spi.StandardErrorCode.NOT_FOUND;
import static com.facebook.presto.spi.StandardWarningCode.PERFORMANCE_WARNING;
Expand Down Expand Up @@ -268,6 +275,8 @@
import static com.facebook.presto.sql.tree.FrameBound.Type.PRECEDING;
import static com.facebook.presto.sql.tree.FrameBound.Type.UNBOUNDED_FOLLOWING;
import static com.facebook.presto.sql.tree.FrameBound.Type.UNBOUNDED_PRECEDING;
import static com.facebook.presto.sql.tree.TableVersionExpression.TableVersionType.TIMESTAMP;
import static com.facebook.presto.sql.tree.TableVersionExpression.TableVersionType.VERSION;
import static com.facebook.presto.util.AnalyzerUtil.createParsingOptions;
import static com.facebook.presto.util.MetadataUtils.getMaterializedViewDefinition;
import static com.facebook.presto.util.MetadataUtils.getTableColumnsMetadata;
Expand Down Expand Up @@ -1280,7 +1289,7 @@ protected Scope visitTable(Table table, Optional<Scope> scope)
}

TableColumnMetadata tableColumnsMetadata = getTableColumnsMetadata(session, metadataResolver, analysis.getMetadataHandle(), name);
Optional<TableHandle> tableHandle = tableColumnsMetadata.getTableHandle();
Optional<TableHandle> tableHandle = getTableHandle(tableColumnsMetadata, table, name, scope);

Map<String, ColumnHandle> columnHandles = tableColumnsMetadata.getColumnHandles();

Expand Down Expand Up @@ -1321,6 +1330,51 @@ protected Scope visitTable(Table table, Optional<Scope> scope)
return createAndAssignScope(table, scope, fields.build());
}

private Optional<TableHandle> getTableHandle(TableColumnMetadata tableColumnsMetadata, Table table, QualifiedObjectName name, Optional<Scope> scope)
{
// Process table version AS OF expression
if (table.getTableVersionExpression().isPresent()) {
return processTableVersion(table, name, scope);
}
else {
return tableColumnsMetadata.getTableHandle();
}
}
private Optional<TableHandle> processTableVersion(Table table, QualifiedObjectName name, Optional<Scope> scope)
{
Expression asOfExpr = table.getTableVersionExpression().get().getAsOfExpression();
TableVersionExpression.TableVersionType tableVersionType = table.getTableVersionExpression().get().getTableVersionType();
ExpressionAnalysis expressionAnalysis = analyzeExpression(asOfExpr, scope.get());
analysis.recordSubqueries(table, expressionAnalysis);
Type asOfExprType = expressionAnalysis.getType(asOfExpr);
if (asOfExprType == UNKNOWN) {
throw new PrestoException(INVALID_ARGUMENTS, format("Table version AS OF expression cannot be NULL for %s", name.toString()));
}
Object evalAsOfExpr = evaluateConstantExpression(asOfExpr, asOfExprType, metadata, session, analysis.getParameters());
if (tableVersionType == TIMESTAMP) {
if (!(asOfExprType instanceof TimestampWithTimeZoneType)) {
throw new SemanticException(TYPE_MISMATCH, asOfExpr,
"Type %s is invalid. Supported table version AS OF expression type is Timestamp with Time Zone.",
asOfExprType.getDisplayName());
}
}
Comment thread
hantangwangd marked this conversation as resolved.
Outdated
if (tableVersionType == VERSION) {
if (!(asOfExprType instanceof BigintType)) {
throw new SemanticException(TYPE_MISMATCH, asOfExpr,
"Type %s is invalid. Supported table version AS OF expression type is BIGINT",
asOfExprType.getDisplayName());
}
}

// Two block entries for table version type and expression.
BlockBuilder blockBuilder = asOfExprType.createBlockBuilder(null, 2);
writeNativeValue(asOfExprType, blockBuilder, tableVersionType.ordinal());
writeNativeValue(asOfExprType, blockBuilder, evalAsOfExpr);
Block block = blockBuilder.build();

return metadata.getHandleVersion(session, name, Optional.of(block));
}

private Scope getScopeFromTable(Table table, Optional<Scope> scope)
{
QualifiedObjectName tableName = createQualifiedObjectName(session, table, table.getName());
Expand Down
7 changes: 7 additions & 0 deletions presto-main/src/test/java/com/facebook/presto/metadata/AbstractMockMetadata.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@
import com.facebook.presto.Session;
import com.facebook.presto.common.CatalogSchemaName;
import com.facebook.presto.common.QualifiedObjectName;
import com.facebook.presto.common.block.Block;
import com.facebook.presto.common.block.BlockEncodingSerde;
import com.facebook.presto.common.predicate.TupleDomain;
import com.facebook.presto.common.type.Type;
Expand Down Expand Up @@ -143,6 +144,12 @@ public Optional<TableHandle> getTableHandleForStatisticsCollection(Session sessi
throw new UnsupportedOperationException();
}

@Override
public Optional<TableHandle> getHandleVersion(Session session, QualifiedObjectName tableName, Optional<Block> tableVersionBlock)
{
throw new UnsupportedOperationException();
}

@Override
public Optional<SystemTable> getSystemTable(Session session, QualifiedObjectName tableName)
{
Expand Down
16 changes: 12 additions & 4 deletions presto-parser/src/main/antlr4/com/facebook/presto/sql/parser/SqlBase.g4
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -324,7 +324,7 @@ columnAliases
;

relationPrimary
: qualifiedName #tableName
: qualifiedName tableVersionExpression? #tableName
| '(' query ')' #subqueryRelation
| UNNEST '(' expression (',' expression)* ')' (WITH ORDINALITY)? #unnest
| LATERAL '(' query ')' #lateral
Expand Down Expand Up @@ -531,6 +531,10 @@ qualifiedName
: identifier ('.' identifier)*
;

tableVersionExpression
: FOR tableVersionType=(SYSTEM_TIME | SYSTEM_VERSION | TIMESTAMP | VERSION) AS OF valueExpression #tableVersion
;

grantor
: CURRENT_USER #currentUserGrantor
| CURRENT_ROLE #currentRoleGrantor
Expand Down Expand Up @@ -576,14 +580,14 @@ nonReserved
| LANGUAGE | LAST | LATERAL | LEVEL | LIMIT | LOGICAL
| MAP | MATERIALIZED | MINUTE | MONTH
| NAME | NFC | NFD | NFKC | NFKD | NO | NONE | NULLIF | NULLS
| OFFSET | ONLY | OPTION | ORDINALITY | OUTPUT | OVER
| OF | OFFSET | ONLY | OPTION | ORDINALITY | OUTPUT | OVER
| PARTITION | PARTITIONS | POSITION | PRECEDING | PRIVILEGES | PROPERTIES
| RANGE | READ | REFRESH | RENAME | REPEATABLE | REPLACE | RESET | RESPECT | RESTRICT | RETURN | RETURNS | REVOKE | ROLE | ROLES | ROLLBACK | ROW | ROWS
| SCHEMA | SCHEMAS | SECOND | SECURITY | SERIALIZABLE | SESSION | SET | SETS | SQL
| SHOW | SOME | START | STATS | SUBSTRING | SYSTEM
| SHOW | SOME | START | STATS | SUBSTRING | SYSTEM | SYSTEM_TIME | SYSTEM_VERSION
| TABLES | TABLESAMPLE | TEMPORARY | TEXT | TIME | TIMESTAMP | TO | TRANSACTION | TRUNCATE | TRY_CAST | TYPE
| UNBOUNDED | UNCOMMITTED | USE | USER
| VALIDATE | VERBOSE | VIEW
| VALIDATE | VERBOSE | VERSION | VIEW
| WORK | WRITE
| YEAR
| ZONE
Expand Down Expand Up @@ -709,6 +713,7 @@ NOT: 'NOT';
NULL: 'NULL';
NULLIF: 'NULLIF';
NULLS: 'NULLS';
OF: 'OF';
OFFSET: 'OFFSET';
ON: 'ON';
ONLY: 'ONLY';
Expand Down Expand Up @@ -762,6 +767,8 @@ START: 'START';
STATS: 'STATS';
SUBSTRING: 'SUBSTRING';
SYSTEM: 'SYSTEM';
SYSTEM_TIME: 'SYSTEM_TIME';
SYSTEM_VERSION: 'SYSTEM_VERSION';
TABLE: 'TABLE';
TABLES: 'TABLES';
TABLESAMPLE: 'TABLESAMPLE';
Expand All @@ -787,6 +794,7 @@ USING: 'USING';
VALIDATE: 'VALIDATE';
VALUES: 'VALUES';
VERBOSE: 'VERBOSE';
VERSION: 'VERSION';
VIEW: 'VIEW';
WHEN: 'WHEN';
WHERE: 'WHERE';
Expand Down
6 changes: 6 additions & 0 deletions presto-parser/src/main/java/com/facebook/presto/sql/ExpressionFormatter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -73,6 +73,7 @@
import com.facebook.presto.sql.tree.SubqueryExpression;
import com.facebook.presto.sql.tree.SubscriptExpression;
import com.facebook.presto.sql.tree.SymbolReference;
import com.facebook.presto.sql.tree.TableVersionExpression;
import com.facebook.presto.sql.tree.TimeLiteral;
import com.facebook.presto.sql.tree.TimestampLiteral;
import com.facebook.presto.sql.tree.TryExpression;
Expand Down Expand Up @@ -689,6 +690,11 @@ private String joinExpressions(List<Expression> expressions)
.map((e) -> process(e, null))
.iterator());
}

protected String visitTableVersion(TableVersionExpression node, Void context)
{
return "FOR " + node.getTableVersionType().name() + " AS OF " + process(node.getAsOfExpression(), context) + " ";
}
}

static String formatStringLiteral(String s)
Expand Down
4 changes: 4 additions & 0 deletions presto-parser/src/main/java/com/facebook/presto/sql/SqlFormatter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -401,6 +401,10 @@ protected Void visitAllColumns(AllColumns node, Integer context)
protected Void visitTable(Table node, Integer indent)
{
builder.append(formatName(node.getName()));
Comment thread
gupteaj marked this conversation as resolved.
Outdated
if (node.getTableVersionExpression().isPresent()) {
builder.append(' ');
process(node.getTableVersionExpression().get(), indent);
}

return null;
}
Expand Down
24 changes: 24 additions & 0 deletions presto-parser/src/main/java/com/facebook/presto/sql/parser/AstBuilder.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -164,6 +164,7 @@
import com.facebook.presto.sql.tree.Table;
import com.facebook.presto.sql.tree.TableElement;
import com.facebook.presto.sql.tree.TableSubquery;
import com.facebook.presto.sql.tree.TableVersionExpression;
import com.facebook.presto.sql.tree.TimeLiteral;
import com.facebook.presto.sql.tree.TimestampLiteral;
import com.facebook.presto.sql.tree.TransactionAccessMode;
Expand Down Expand Up @@ -199,6 +200,8 @@
import static com.facebook.presto.sql.tree.RoutineCharacteristics.NullCallClause;
import static com.facebook.presto.sql.tree.RoutineCharacteristics.NullCallClause.CALLED_ON_NULL_INPUT;
import static com.facebook.presto.sql.tree.RoutineCharacteristics.NullCallClause.RETURNS_NULL_ON_NULL_INPUT;
import static com.facebook.presto.sql.tree.TableVersionExpression.timestampExpression;
import static com.facebook.presto.sql.tree.TableVersionExpression.versionExpression;
import static com.google.common.collect.ImmutableList.toImmutableList;
import static com.google.common.collect.Iterables.getOnlyElement;
import static java.lang.String.format;
Expand Down Expand Up @@ -1296,6 +1299,10 @@ public Node visitAliasedRelation(SqlBaseParser.AliasedRelationContext context)
@Override
public Node visitTableName(SqlBaseParser.TableNameContext context)
{
if (context.tableVersionExpression() != null) {
return new Table(getLocation(context), getQualifiedName(context.qualifiedName()), (TableVersionExpression) visit(context.tableVersionExpression()));
}

return new Table(getLocation(context), getQualifiedName(context.qualifiedName()));
}

Expand Down Expand Up @@ -1454,6 +1461,23 @@ public Node visitQuantifiedComparison(SqlBaseParser.QuantifiedComparisonContext

// ************** value expressions **************

@Override
public Node visitTableVersion(SqlBaseParser.TableVersionContext context)
{
Expression child = (Expression) visit(context.valueExpression());

switch (context.tableVersionType.getType()) {
case SqlBaseLexer.SYSTEM_TIME:
case SqlBaseLexer.TIMESTAMP:
return timestampExpression(getLocation(context), child);
case SqlBaseLexer.SYSTEM_VERSION:
case SqlBaseLexer.VERSION:
return versionExpression(getLocation(context), child);
default:
throw new UnsupportedOperationException("Unsupported Type: " + context.tableVersionType.getText());
}
}

@Override
public Node visitArithmeticUnary(SqlBaseParser.ArithmeticUnaryContext context)
{
Expand Down
Loading