diff --git a/core/trino-main/src/main/java/io/trino/connector/informationschema/InformationSchemaMetadata.java b/core/trino-main/src/main/java/io/trino/connector/informationschema/InformationSchemaMetadata.java index 5c5da9a224d3..eead8614f8c5 100644 --- a/core/trino-main/src/main/java/io/trino/connector/informationschema/InformationSchemaMetadata.java +++ b/core/trino-main/src/main/java/io/trino/connector/informationschema/InformationSchemaMetadata.java @@ -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; @@ -29,6 +30,7 @@ import io.trino.spi.connector.ConnectorTableHandle; import io.trino.spi.connector.ConnectorTableMetadata; import io.trino.spi.connector.ConnectorTableProperties; +import io.trino.spi.connector.ConnectorTableVersion; import io.trino.spi.connector.Constraint; import io.trino.spi.connector.ConstraintApplicationResult; import io.trino.spi.connector.LimitApplicationResult; @@ -62,6 +64,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.NOT_SUPPORTED; import static io.trino.spi.type.VarcharType.createUnboundedVarcharType; import static java.util.Collections.emptyList; import static java.util.Locale.ENGLISH; @@ -95,8 +98,12 @@ public List listSchemaNames(ConnectorSession session) } @Override - public ConnectorTableHandle getTableHandle(ConnectorSession connectorSession, SchemaTableName tableName) + public ConnectorTableHandle getTableHandle(ConnectorSession connectorSession, SchemaTableName tableName, Optional startVersion, Optional endVersion) { + if (startVersion.isPresent() || endVersion.isPresent()) { + throw new TrinoException(NOT_SUPPORTED, "This connector does not support versioned tables"); + } + return InformationSchemaTable.of(tableName) .map(table -> new InformationSchemaTableHandle(catalogName, table, defaultPrefixes(catalogName), OptionalLong.empty())) .orElse(null); diff --git a/core/trino-main/src/main/java/io/trino/connector/system/SystemTablesMetadata.java b/core/trino-main/src/main/java/io/trino/connector/system/SystemTablesMetadata.java index 2473f976c70f..e564ce1e6b33 100644 --- a/core/trino-main/src/main/java/io/trino/connector/system/SystemTablesMetadata.java +++ b/core/trino-main/src/main/java/io/trino/connector/system/SystemTablesMetadata.java @@ -23,6 +23,7 @@ import io.trino.spi.connector.ConnectorSession; import io.trino.spi.connector.ConnectorTableHandle; import io.trino.spi.connector.ConnectorTableMetadata; +import io.trino.spi.connector.ConnectorTableVersion; import io.trino.spi.connector.Constraint; import io.trino.spi.connector.ConstraintApplicationResult; import io.trino.spi.connector.SchemaTableName; @@ -39,6 +40,7 @@ import static io.trino.connector.system.SystemColumnHandle.toSystemColumnHandles; import static io.trino.metadata.MetadataUtil.findColumnMetadata; import static io.trino.spi.StandardErrorCode.NOT_FOUND; +import static io.trino.spi.StandardErrorCode.NOT_SUPPORTED; import static java.lang.String.format; import static java.util.Collections.singletonMap; import static java.util.Objects.requireNonNull; @@ -63,12 +65,17 @@ public List listSchemaNames(ConnectorSession session) } @Override - public ConnectorTableHandle getTableHandle(ConnectorSession session, SchemaTableName tableName) + public ConnectorTableHandle getTableHandle(ConnectorSession session, SchemaTableName tableName, Optional startVersion, Optional endVersion) { Optional table = tables.getSystemTable(session, tableName); if (table.isEmpty()) { return null; } + + if (startVersion.isPresent() || endVersion.isPresent()) { + throw new TrinoException(NOT_SUPPORTED, "This connector does not support versioned tables"); + } + return SystemTableHandle.fromSchemaTableName(tableName); } diff --git a/core/trino-main/src/main/java/io/trino/metadata/CatalogMetadata.java b/core/trino-main/src/main/java/io/trino/metadata/CatalogMetadata.java index 42d827961d4c..aba219759cef 100644 --- a/core/trino-main/src/main/java/io/trino/metadata/CatalogMetadata.java +++ b/core/trino-main/src/main/java/io/trino/metadata/CatalogMetadata.java @@ -22,9 +22,11 @@ import io.trino.spi.connector.CatalogSchemaName; import io.trino.spi.connector.ConnectorCapabilities; import io.trino.spi.connector.ConnectorMetadata; +import io.trino.spi.connector.ConnectorTableVersion; import io.trino.spi.connector.ConnectorTransactionHandle; import java.util.List; +import java.util.Optional; import java.util.Set; import static com.google.common.base.MoreObjects.toStringHelper; @@ -130,13 +132,14 @@ public CatalogHandle getConnectorHandleForSchema(CatalogSchemaName schema) return catalogTransaction.getCatalogHandle(); } - public CatalogHandle getCatalogHandle(Session session, QualifiedObjectName table) + public CatalogHandle getCatalogHandle(Session session, QualifiedObjectName table, Optional startVersion, Optional endVersion) { if (table.schemaName().equals(INFORMATION_SCHEMA_NAME)) { return informationSchemaTransaction.getCatalogHandle(); } - if (systemTransaction.getConnectorMetadata(session).getTableHandle(session.toConnectorSession(systemTransaction.getCatalogHandle()), table.asSchemaTableName()) != null) { + if (systemTransaction.getConnectorMetadata(session) + .getTableHandle(session.toConnectorSession(systemTransaction.getCatalogHandle()), table.asSchemaTableName(), startVersion, endVersion) != null) { return systemTransaction.getCatalogHandle(); } diff --git a/core/trino-main/src/main/java/io/trino/metadata/MetadataManager.java b/core/trino-main/src/main/java/io/trino/metadata/MetadataManager.java index cae0f47e71b3..1a4a3e04faef 100644 --- a/core/trino-main/src/main/java/io/trino/metadata/MetadataManager.java +++ b/core/trino-main/src/main/java/io/trino/metadata/MetadataManager.java @@ -275,7 +275,10 @@ public Optional getTableHandle(Session session, QualifiedObjectName } return getOptionalCatalogMetadata(session, table.catalogName()).flatMap(catalogMetadata -> { - CatalogHandle catalogHandle = catalogMetadata.getCatalogHandle(session, table); + Optional startTableVersion = toConnectorVersion(startVersion); + Optional endTableVersion = toConnectorVersion(endVersion); + + CatalogHandle catalogHandle = catalogMetadata.getCatalogHandle(session, table, startTableVersion, endTableVersion); ConnectorMetadata metadata = catalogMetadata.getMetadataFor(session, catalogHandle); ConnectorSession connectorSession = session.toConnectorSession(catalogHandle); @@ -283,8 +286,8 @@ public Optional getTableHandle(Session session, QualifiedObjectName ConnectorTableHandle tableHandle = metadata.getTableHandle( connectorSession, table.asSchemaTableName(), - toConnectorVersion(startVersion), - toConnectorVersion(endVersion)); + startTableVersion, + endTableVersion); return Optional.ofNullable(tableHandle) .map(connectorTableHandle -> new TableHandle( catalogHandle, @@ -1496,7 +1499,7 @@ public Map getViewProperties(Session session, QualifiedObjectNam Optional catalog = getOptionalCatalogMetadata(session, viewName.catalogName()); if (catalog.isPresent()) { CatalogMetadata catalogMetadata = catalog.get(); - CatalogHandle catalogHandle = catalogMetadata.getCatalogHandle(session, viewName); + CatalogHandle catalogHandle = catalogMetadata.getCatalogHandle(session, viewName, Optional.empty(), Optional.empty()); ConnectorMetadata metadata = catalogMetadata.getMetadataFor(session, catalogHandle); ConnectorSession connectorSession = session.toConnectorSession(catalogHandle); @@ -1537,7 +1540,7 @@ private Optional getViewInternal(Session session, Quali Optional catalog = getOptionalCatalogMetadata(session, viewName.catalogName()); if (catalog.isPresent()) { CatalogMetadata catalogMetadata = catalog.get(); - CatalogHandle catalogHandle = catalogMetadata.getCatalogHandle(session, viewName); + CatalogHandle catalogHandle = catalogMetadata.getCatalogHandle(session, viewName, Optional.empty(), Optional.empty()); ConnectorMetadata metadata = catalogMetadata.getMetadataFor(session, catalogHandle); ConnectorSession connectorSession = session.toConnectorSession(catalogHandle); @@ -1766,7 +1769,7 @@ private Optional getMaterializedViewInterna Optional catalog = getOptionalCatalogMetadata(session, viewName.catalogName()); if (catalog.isPresent()) { CatalogMetadata catalogMetadata = catalog.get(); - CatalogHandle catalogHandle = catalogMetadata.getCatalogHandle(session, viewName); + CatalogHandle catalogHandle = catalogMetadata.getCatalogHandle(session, viewName, Optional.empty(), Optional.empty()); ConnectorMetadata metadata = catalogMetadata.getMetadataFor(session, catalogHandle); ConnectorSession connectorSession = session.toConnectorSession(catalogHandle); @@ -1781,7 +1784,7 @@ public Map getMaterializedViewProperties(Session session, Qualif Optional catalog = getOptionalCatalogMetadata(session, viewName.catalogName()); if (catalog.isPresent()) { CatalogMetadata catalogMetadata = catalog.get(); - CatalogHandle catalogHandle = catalogMetadata.getCatalogHandle(session, viewName); + CatalogHandle catalogHandle = catalogMetadata.getCatalogHandle(session, viewName, Optional.empty(), Optional.empty()); ConnectorMetadata metadata = catalogMetadata.getMetadataFor(session, catalogHandle); ConnectorSession connectorSession = session.toConnectorSession(catalogHandle); @@ -1799,7 +1802,7 @@ public MaterializedViewFreshness getMaterializedViewFreshness(Session session, Q Optional catalog = getOptionalCatalogMetadata(session, viewName.catalogName()); if (catalog.isPresent()) { CatalogMetadata catalogMetadata = catalog.get(); - CatalogHandle catalogHandle = catalogMetadata.getCatalogHandle(session, viewName); + CatalogHandle catalogHandle = catalogMetadata.getCatalogHandle(session, viewName, Optional.empty(), Optional.empty()); ConnectorMetadata metadata = catalogMetadata.getMetadataFor(session, catalogHandle); ConnectorSession connectorSession = session.toConnectorSession(catalogHandle); @@ -1863,7 +1866,7 @@ public Optional applyTableScanRedirect(Sessi return metadata.applyTableScanRedirect(connectorSession, tableHandle.connectorHandle()); } - private QualifiedObjectName getRedirectedTableName(Session session, QualifiedObjectName originalTableName) + private QualifiedObjectName getRedirectedTableName(Session session, QualifiedObjectName originalTableName, Optional startVersion, Optional endVersion) { requireNonNull(session, "session is null"); requireNonNull(originalTableName, "originalTableName is null"); @@ -1884,7 +1887,7 @@ private QualifiedObjectName getRedirectedTableName(Session session, QualifiedObj } CatalogMetadata catalogMetadata = catalog.get(); - CatalogHandle catalogHandle = catalogMetadata.getCatalogHandle(session, tableName); + CatalogHandle catalogHandle = catalogMetadata.getCatalogHandle(session, tableName, toConnectorVersion(startVersion), toConnectorVersion(endVersion)); ConnectorMetadata metadata = catalogMetadata.getMetadataFor(session, catalogHandle); Optional redirectedTableName = metadata.redirectTable(session.toConnectorSession(catalogHandle), tableName.asSchemaTableName()) @@ -1917,7 +1920,7 @@ public RedirectionAwareTableHandle getRedirectionAwareTableHandle(Session sessio @Override public RedirectionAwareTableHandle getRedirectionAwareTableHandle(Session session, QualifiedObjectName tableName, Optional startVersion, Optional endVersion) { - QualifiedObjectName targetTableName = getRedirectedTableName(session, tableName); + QualifiedObjectName targetTableName = getRedirectedTableName(session, tableName, startVersion, endVersion); if (targetTableName.equals(tableName)) { return noRedirection(getTableHandle(session, tableName, startVersion, endVersion)); } @@ -2426,7 +2429,7 @@ public List listTablePrivileges(Session session, QualifiedTablePrefix ConnectorSession connectorSession = session.toConnectorSession(catalogMetadata.getCatalogHandle()); List catalogHandles = prefix.asQualifiedObjectName() - .map(qualifiedTableName -> singletonList(catalogMetadata.getCatalogHandle(session, qualifiedTableName))) + .map(qualifiedTableName -> singletonList(catalogMetadata.getCatalogHandle(session, qualifiedTableName, Optional.empty(), Optional.empty()))) .orElseGet(catalogMetadata::listCatalogHandles); for (CatalogHandle catalogHandle : catalogHandles) { ConnectorMetadata metadata = catalogMetadata.getMetadataFor(session, catalogHandle); @@ -2778,7 +2781,7 @@ public OptionalInt getMaxWriterTasks(Session session, String catalogName) public WriterScalingOptions getNewTableWriterScalingOptions(Session session, QualifiedObjectName tableName, Map tableProperties) { CatalogMetadata catalogMetadata = getCatalogMetadataForWrite(session, tableName.catalogName()); - CatalogHandle catalogHandle = catalogMetadata.getCatalogHandle(session, tableName); + CatalogHandle catalogHandle = catalogMetadata.getCatalogHandle(session, tableName, Optional.empty(), Optional.empty()); ConnectorMetadata metadata = catalogMetadata.getMetadataFor(session, catalogHandle); return metadata.getNewTableWriterScalingOptions(session.toConnectorSession(catalogHandle), tableName.asSchemaTableName(), tableProperties); } diff --git a/core/trino-main/src/main/java/io/trino/testing/TestingMetadata.java b/core/trino-main/src/main/java/io/trino/testing/TestingMetadata.java index 0c0e04517e80..1b20b7f5fd5e 100644 --- a/core/trino-main/src/main/java/io/trino/testing/TestingMetadata.java +++ b/core/trino-main/src/main/java/io/trino/testing/TestingMetadata.java @@ -32,6 +32,7 @@ import io.trino.spi.connector.ConnectorTableHandle; import io.trino.spi.connector.ConnectorTableLayout; import io.trino.spi.connector.ConnectorTableMetadata; +import io.trino.spi.connector.ConnectorTableVersion; import io.trino.spi.connector.ConnectorViewDefinition; import io.trino.spi.connector.MaterializedViewFreshness; import io.trino.spi.connector.MaterializedViewNotFoundException; @@ -62,6 +63,7 @@ import static com.google.common.base.MoreObjects.toStringHelper; import static com.google.common.base.Preconditions.checkArgument; import static io.trino.spi.StandardErrorCode.ALREADY_EXISTS; +import static io.trino.spi.StandardErrorCode.NOT_SUPPORTED; import static io.trino.spi.connector.MaterializedViewFreshness.Freshness.FRESH; import static io.trino.spi.connector.MaterializedViewFreshness.Freshness.STALE; import static java.util.Collections.synchronizedSet; @@ -90,8 +92,12 @@ public List listSchemaNames(ConnectorSession session) } @Override - public ConnectorTableHandle getTableHandle(ConnectorSession session, SchemaTableName tableName) + public ConnectorTableHandle getTableHandle(ConnectorSession session, SchemaTableName tableName, Optional startVersion, Optional endVersion) { + if (startVersion.isPresent() || endVersion.isPresent()) { + throw new TrinoException(NOT_SUPPORTED, "This connector does not support versioned tables"); + } + requireNonNull(tableName, "tableName is null"); if (!tables.containsKey(tableName)) { return null; diff --git a/core/trino-main/src/main/java/io/trino/tracing/TracingConnectorMetadata.java b/core/trino-main/src/main/java/io/trino/tracing/TracingConnectorMetadata.java index 9b39e578f820..09425d19da90 100644 --- a/core/trino-main/src/main/java/io/trino/tracing/TracingConnectorMetadata.java +++ b/core/trino-main/src/main/java/io/trino/tracing/TracingConnectorMetadata.java @@ -132,15 +132,6 @@ public List listSchemaNames(ConnectorSession session) } } - @Override - public ConnectorTableHandle getTableHandle(ConnectorSession session, SchemaTableName tableName) - { - Span span = startSpan("getTableHandle", tableName); - try (var _ = scopedSpan(span)) { - return delegate.getTableHandle(session, tableName); - } - } - @Override public ConnectorTableHandle getTableHandle(ConnectorSession session, SchemaTableName tableName, Optional startVersion, Optional endVersion) { diff --git a/core/trino-main/src/test/java/io/trino/connector/MockConnector.java b/core/trino-main/src/test/java/io/trino/connector/MockConnector.java index 643239e3eb49..cf03ea18f497 100644 --- a/core/trino-main/src/test/java/io/trino/connector/MockConnector.java +++ b/core/trino-main/src/test/java/io/trino/connector/MockConnector.java @@ -27,6 +27,7 @@ import io.trino.connector.MockConnectorFactory.ApplyTopN; import io.trino.connector.MockConnectorFactory.ListRoleGrants; import io.trino.spi.Page; +import io.trino.spi.TrinoException; import io.trino.spi.connector.AggregateFunction; import io.trino.spi.connector.AggregationApplicationResult; import io.trino.spi.connector.BeginTableExecuteResult; @@ -58,6 +59,7 @@ import io.trino.spi.connector.ConnectorTableLayout; import io.trino.spi.connector.ConnectorTableMetadata; import io.trino.spi.connector.ConnectorTableProperties; +import io.trino.spi.connector.ConnectorTableVersion; import io.trino.spi.connector.ConnectorTransactionHandle; import io.trino.spi.connector.ConnectorViewDefinition; import io.trino.spi.connector.Constraint; @@ -125,6 +127,7 @@ import static com.google.common.collect.ImmutableMap.toImmutableMap; import static com.google.common.collect.ImmutableSet.toImmutableSet; import static io.trino.connector.MockConnector.MockConnectorSplit.MOCK_CONNECTOR_SPLIT; +import static io.trino.spi.StandardErrorCode.NOT_SUPPORTED; import static io.trino.spi.connector.MaterializedViewFreshness.Freshness.FRESH; import static io.trino.spi.connector.MaterializedViewFreshness.Freshness.STALE; import static io.trino.spi.connector.RowChangeParadigm.DELETE_ROW_AND_INSERT_ROW; @@ -535,8 +538,12 @@ public void setSchemaAuthorization(ConnectorSession session, String schemaName, public void dropSchema(ConnectorSession session, String schemaName, boolean cascade) {} @Override - public ConnectorTableHandle getTableHandle(ConnectorSession session, SchemaTableName tableName) + public ConnectorTableHandle getTableHandle(ConnectorSession session, SchemaTableName tableName, Optional startVersion, Optional endVersion) { + if (startVersion.isPresent() || endVersion.isPresent()) { + throw new TrinoException(NOT_SUPPORTED, "This connector does not support versioned tables"); + } + return getTableHandle.apply(session, tableName); } diff --git a/core/trino-main/src/test/java/io/trino/metadata/TestInformationSchemaMetadata.java b/core/trino-main/src/test/java/io/trino/metadata/TestInformationSchemaMetadata.java index 731e52aa5f4a..7120cbadb9cc 100644 --- a/core/trino-main/src/test/java/io/trino/metadata/TestInformationSchemaMetadata.java +++ b/core/trino-main/src/test/java/io/trino/metadata/TestInformationSchemaMetadata.java @@ -128,7 +128,7 @@ public void testInformationSchemaPredicatePushdown() ConnectorSession session = createNewSession(transactionId); ConnectorMetadata metadata = new InformationSchemaMetadata("test_catalog", this.metadata, MAX_PREFIXES_COUNT); InformationSchemaTableHandle tableHandle = (InformationSchemaTableHandle) - metadata.getTableHandle(session, new SchemaTableName("information_schema", "views")); + metadata.getTableHandle(session, new SchemaTableName("information_schema", "views"), Optional.empty(), Optional.empty()); tableHandle = metadata.applyFilter(session, tableHandle, constraint) .map(ConstraintApplicationResult::getHandle) .map(InformationSchemaTableHandle.class::cast) @@ -145,7 +145,7 @@ public void testInformationSchemaPredicatePushdownWithConstraintPredicate() ConnectorSession session = createNewSession(transactionId); ConnectorMetadata metadata = new InformationSchemaMetadata("test_catalog", this.metadata, MAX_PREFIXES_COUNT); InformationSchemaTableHandle tableHandle = (InformationSchemaTableHandle) - metadata.getTableHandle(session, new SchemaTableName("information_schema", "columns")); + metadata.getTableHandle(session, new SchemaTableName("information_schema", "columns"), Optional.empty(), Optional.empty()); tableHandle = metadata.applyFilter(session, tableHandle, constraint) .map(ConstraintApplicationResult::getHandle) .map(InformationSchemaTableHandle.class::cast) @@ -167,7 +167,7 @@ public void testInformationSchemaPredicatePushdownWithoutSchemaPredicate() ConnectorSession session = createNewSession(transactionId); ConnectorMetadata metadata = new InformationSchemaMetadata("test_catalog", this.metadata, MAX_PREFIXES_COUNT); InformationSchemaTableHandle tableHandle = (InformationSchemaTableHandle) - metadata.getTableHandle(session, new SchemaTableName("information_schema", "views")); + metadata.getTableHandle(session, new SchemaTableName("information_schema", "views"), Optional.empty(), Optional.empty()); tableHandle = metadata.applyFilter(session, tableHandle, constraint) .map(ConstraintApplicationResult::getHandle) .map(InformationSchemaTableHandle.class::cast) @@ -191,7 +191,7 @@ public void testInformationSchemaPredicatePushdownWithoutTablePredicate() ConnectorSession session = createNewSession(transactionId); ConnectorMetadata metadata = new InformationSchemaMetadata("test_catalog", this.metadata, MAX_PREFIXES_COUNT); InformationSchemaTableHandle tableHandle = (InformationSchemaTableHandle) - metadata.getTableHandle(session, new SchemaTableName("information_schema", "views")); + metadata.getTableHandle(session, new SchemaTableName("information_schema", "views"), Optional.empty(), Optional.empty()); tableHandle = metadata.applyFilter(session, tableHandle, constraint) .map(ConstraintApplicationResult::getHandle) .map(InformationSchemaTableHandle.class::cast) @@ -209,7 +209,7 @@ public void testInformationSchemaPredicatePushdownWithConstraintPredicateOnViews ConnectorSession session = createNewSession(transactionId); ConnectorMetadata metadata = new InformationSchemaMetadata("test_catalog", this.metadata, MAX_PREFIXES_COUNT); InformationSchemaTableHandle tableHandle = (InformationSchemaTableHandle) - metadata.getTableHandle(session, new SchemaTableName("information_schema", "views")); + metadata.getTableHandle(session, new SchemaTableName("information_schema", "views"), Optional.empty(), Optional.empty()); tableHandle = metadata.applyFilter(session, tableHandle, constraint) .map(ConstraintApplicationResult::getHandle) .map(InformationSchemaTableHandle.class::cast) @@ -229,7 +229,7 @@ public void testInformationSchemaPredicatePushdownOnCatalogWiseTables() ConnectorSession session = createNewSession(transactionId); ConnectorMetadata metadata = new InformationSchemaMetadata("test_catalog", this.metadata, MAX_PREFIXES_COUNT); InformationSchemaTableHandle tableHandle = (InformationSchemaTableHandle) - metadata.getTableHandle(session, new SchemaTableName("information_schema", "schemata")); + metadata.getTableHandle(session, new SchemaTableName("information_schema", "schemata"), Optional.empty(), Optional.empty()); Optional> result = metadata.applyFilter(session, tableHandle, constraint); assertThat(result.isPresent()).isFalse(); } @@ -242,7 +242,7 @@ public void testInformationSchemaPredicatePushdownForEmptyNames() ConnectorMetadata metadata = new InformationSchemaMetadata("test_catalog", this.metadata, MAX_PREFIXES_COUNT); InformationSchemaColumnHandle tableSchemaColumn = new InformationSchemaColumnHandle("table_schema"); InformationSchemaColumnHandle tableNameColumn = new InformationSchemaColumnHandle("table_name"); - ConnectorTableHandle tableHandle = metadata.getTableHandle(session, new SchemaTableName("information_schema", "tables")); + ConnectorTableHandle tableHandle = metadata.getTableHandle(session, new SchemaTableName("information_schema", "tables"), Optional.empty(), Optional.empty()); // Empty schema name InformationSchemaTableHandle filtered = metadata.applyFilter(session, tableHandle, new Constraint(TupleDomain.withColumnDomains( diff --git a/core/trino-main/src/test/java/io/trino/sql/analyzer/TestAnalyzer.java b/core/trino-main/src/test/java/io/trino/sql/analyzer/TestAnalyzer.java index 1f448cefa5f0..b3ac2f71763a 100644 --- a/core/trino-main/src/test/java/io/trino/sql/analyzer/TestAnalyzer.java +++ b/core/trino-main/src/test/java/io/trino/sql/analyzer/TestAnalyzer.java @@ -944,11 +944,11 @@ public void testInvalidTable() .hasErrorCode(TABLE_NOT_FOUND) .hasMessage("line 1:15: Table 'tpch.s1.foo' does not exist"); assertFails("SELECT * FROM foo FOR TIMESTAMP AS OF TIMESTAMP '2021-03-01 00:00:01'") - .hasErrorCode(TABLE_NOT_FOUND) - .hasMessage("line 1:15: Table 'tpch.s1.foo' does not exist"); + .hasErrorCode(NOT_SUPPORTED) + .hasMessage("This connector does not support versioned tables"); assertFails("SELECT * FROM foo FOR VERSION AS OF 'version1'") - .hasErrorCode(TABLE_NOT_FOUND) - .hasMessage("line 1:15: Table 'tpch.s1.foo' does not exist"); + .hasErrorCode(NOT_SUPPORTED) + .hasMessage("This connector does not support versioned tables"); // table name containing dots assertFails("SELECT * FROM \"table.not.existing\"") .hasErrorCode(TABLE_NOT_FOUND) diff --git a/core/trino-spi/src/main/java/io/trino/spi/connector/ConnectorMetadata.java b/core/trino-spi/src/main/java/io/trino/spi/connector/ConnectorMetadata.java index 9d24fdb3dcf1..df1587f005d8 100644 --- a/core/trino-spi/src/main/java/io/trino/spi/connector/ConnectorMetadata.java +++ b/core/trino-spi/src/main/java/io/trino/spi/connector/ConnectorMetadata.java @@ -104,23 +104,6 @@ default List listSchemaNames(ConnectorSession session) return emptyList(); } - /** - * Returns a table handle for the specified table name, or {@code null} if {@code tableName} relation does not exist - * or is not a table (e.g. is a view, or a materialized view). - * - * @throws TrinoException implementation can throw this exception when {@code tableName} refers to a table that - * cannot be queried. - * @see #getView(ConnectorSession, SchemaTableName) - * @see #getMaterializedView(ConnectorSession, SchemaTableName) - * @deprecated Implement {@link #getTableHandle(ConnectorSession, SchemaTableName, Optional, Optional)}. - */ - @Nullable - @Deprecated - default ConnectorTableHandle getTableHandle(ConnectorSession session, SchemaTableName tableName) - { - return null; - } - /** * Returns a table handle for the specified table name and version, or {@code null} if {@code tableName} relation does not exist * or is not a table (e.g. is a view, or a materialized view). @@ -137,15 +120,7 @@ default ConnectorTableHandle getTableHandle( Optional startVersion, Optional endVersion) { - ConnectorTableHandle tableHandle = getTableHandle(session, tableName); - if (tableHandle == null) { - // Not found - return null; - } - if (startVersion.isEmpty() && endVersion.isEmpty()) { - return tableHandle; - } - throw new TrinoException(NOT_SUPPORTED, "This connector does not support versioned tables"); + throw new TrinoException(GENERIC_INTERNAL_ERROR, "ConnectorMetadata getTableHandle() is not implemented"); } /** @@ -1039,7 +1014,7 @@ default Map getViews(ConnectorSession * Gets the view data for the specified view name. Returns {@link Optional#empty()} if {@code viewName} * relation does not or is not a view (e.g. is a table, or a materialized view). * - * @see #getTableHandle(ConnectorSession, SchemaTableName) + * @see #getTableHandle(ConnectorSession, SchemaTableName, Optional, Optional) * @see #getMaterializedView(ConnectorSession, SchemaTableName) */ default Optional getView(ConnectorSession session, SchemaTableName viewName) @@ -1748,7 +1723,7 @@ default Map getMaterialize * Gets the materialized view data for the specified materialized view name. Returns {@link Optional#empty()} * if {@code viewName} relation does not or is not a materialized view (e.g. is a table, or a view). * - * @see #getTableHandle(ConnectorSession, SchemaTableName) + * @see #getTableHandle(ConnectorSession, SchemaTableName, Optional, Optional) * @see #getView(ConnectorSession, SchemaTableName) */ default Optional getMaterializedView(ConnectorSession session, SchemaTableName viewName) diff --git a/lib/trino-plugin-toolkit/src/main/java/io/trino/plugin/base/classloader/ClassLoaderSafeConnectorMetadata.java b/lib/trino-plugin-toolkit/src/main/java/io/trino/plugin/base/classloader/ClassLoaderSafeConnectorMetadata.java index 75cbc9c57ae8..0f9975b16953 100644 --- a/lib/trino-plugin-toolkit/src/main/java/io/trino/plugin/base/classloader/ClassLoaderSafeConnectorMetadata.java +++ b/lib/trino-plugin-toolkit/src/main/java/io/trino/plugin/base/classloader/ClassLoaderSafeConnectorMetadata.java @@ -199,14 +199,6 @@ public List listSchemaNames(ConnectorSession session) } } - @Override - public ConnectorTableHandle getTableHandle(ConnectorSession session, SchemaTableName tableName) - { - try (ThreadContextClassLoader _ = new ThreadContextClassLoader(classLoader)) { - return delegate.getTableHandle(session, tableName); - } - } - @Override public Optional getTableHandleForExecute(ConnectorSession session, ConnectorTableHandle tableHandle, String procedureName, Map executeProperties, RetryMode retryMode) { diff --git a/plugin/trino-accumulo/src/main/java/io/trino/plugin/accumulo/AccumuloMetadata.java b/plugin/trino-accumulo/src/main/java/io/trino/plugin/accumulo/AccumuloMetadata.java index e5fe9a8d5ec4..d23a5ad9c04f 100644 --- a/plugin/trino-accumulo/src/main/java/io/trino/plugin/accumulo/AccumuloMetadata.java +++ b/plugin/trino-accumulo/src/main/java/io/trino/plugin/accumulo/AccumuloMetadata.java @@ -35,6 +35,7 @@ import io.trino.spi.connector.ConnectorTableHandle; import io.trino.spi.connector.ConnectorTableLayout; import io.trino.spi.connector.ConnectorTableMetadata; +import io.trino.spi.connector.ConnectorTableVersion; import io.trino.spi.connector.ConnectorViewDefinition; import io.trino.spi.connector.Constraint; import io.trino.spi.connector.ConstraintApplicationResult; @@ -261,8 +262,12 @@ private static void rollbackInsert(ConnectorInsertTableHandle insertHandle) } @Override - public ConnectorTableHandle getTableHandle(ConnectorSession session, SchemaTableName tableName) + public ConnectorTableHandle getTableHandle(ConnectorSession session, SchemaTableName tableName, Optional startVersion, Optional endVersion) { + if (startVersion.isPresent() || endVersion.isPresent()) { + throw new TrinoException(NOT_SUPPORTED, "This connector does not support versioned tables"); + } + if (!listSchemaNames(session).contains(tableName.getSchemaName().toLowerCase(Locale.ENGLISH))) { return null; } @@ -446,7 +451,7 @@ private List listTables(ConnectorSession session, SchemaTablePr // Make sure requested table exists, returning the single table of it does SchemaTableName table = prefix.toSchemaTableName(); - if (getTableHandle(session, table) != null) { + if (getTableHandle(session, table, Optional.empty(), Optional.empty()) != null) { return ImmutableList.of(table); } diff --git a/plugin/trino-atop/src/main/java/io/trino/plugin/atop/AtopMetadata.java b/plugin/trino-atop/src/main/java/io/trino/plugin/atop/AtopMetadata.java index 368ae3bcd23c..6860a148755a 100644 --- a/plugin/trino-atop/src/main/java/io/trino/plugin/atop/AtopMetadata.java +++ b/plugin/trino-atop/src/main/java/io/trino/plugin/atop/AtopMetadata.java @@ -17,6 +17,7 @@ import com.google.common.collect.ImmutableMap; import com.google.inject.Inject; import io.trino.plugin.atop.AtopTable.AtopColumn; +import io.trino.spi.TrinoException; import io.trino.spi.connector.ColumnHandle; import io.trino.spi.connector.ColumnMetadata; import io.trino.spi.connector.ColumnNotFoundException; @@ -24,6 +25,7 @@ import io.trino.spi.connector.ConnectorSession; import io.trino.spi.connector.ConnectorTableHandle; import io.trino.spi.connector.ConnectorTableMetadata; +import io.trino.spi.connector.ConnectorTableVersion; import io.trino.spi.connector.Constraint; import io.trino.spi.connector.ConstraintApplicationResult; import io.trino.spi.connector.SchemaTableName; @@ -39,6 +41,7 @@ import static io.trino.plugin.atop.AtopTable.AtopColumn.END_TIME; import static io.trino.plugin.atop.AtopTable.AtopColumn.START_TIME; +import static io.trino.spi.StandardErrorCode.NOT_SUPPORTED; import static java.util.Locale.ENGLISH; import static java.util.Objects.requireNonNull; @@ -64,8 +67,12 @@ public List listSchemaNames(ConnectorSession session) } @Override - public ConnectorTableHandle getTableHandle(ConnectorSession session, SchemaTableName tableName) + public ConnectorTableHandle getTableHandle(ConnectorSession session, SchemaTableName tableName, Optional startVersion, Optional endVersion) { + if (startVersion.isPresent() || endVersion.isPresent()) { + throw new TrinoException(NOT_SUPPORTED, "This connector does not support versioned tables"); + } + requireNonNull(tableName, "tableName is null"); String schemaName = tableName.getSchemaName(); @@ -125,7 +132,7 @@ public Map> listTableColumns(ConnectorSess { ImmutableMap.Builder> columns = ImmutableMap.builder(); for (SchemaTableName tableName : listTables(session, prefix.getSchema())) { - ConnectorTableMetadata tableMetadata = getTableMetadata(session, getTableHandle(session, tableName)); + ConnectorTableMetadata tableMetadata = getTableMetadata(session, getTableHandle(session, tableName, Optional.empty(), Optional.empty())); columns.put(tableName, tableMetadata.getColumns()); } return columns.buildOrThrow(); diff --git a/plugin/trino-base-jdbc/src/main/java/io/trino/plugin/jdbc/DefaultJdbcMetadata.java b/plugin/trino-base-jdbc/src/main/java/io/trino/plugin/jdbc/DefaultJdbcMetadata.java index 173043f08e64..4495a89d58ff 100644 --- a/plugin/trino-base-jdbc/src/main/java/io/trino/plugin/jdbc/DefaultJdbcMetadata.java +++ b/plugin/trino-base-jdbc/src/main/java/io/trino/plugin/jdbc/DefaultJdbcMetadata.java @@ -39,6 +39,7 @@ import io.trino.spi.connector.ConnectorTableLayout; import io.trino.spi.connector.ConnectorTableMetadata; import io.trino.spi.connector.ConnectorTableSchema; +import io.trino.spi.connector.ConnectorTableVersion; import io.trino.spi.connector.Constraint; import io.trino.spi.connector.ConstraintApplicationResult; import io.trino.spi.connector.JoinApplicationResult; @@ -150,8 +151,12 @@ public List listSchemaNames(ConnectorSession session) } @Override - public JdbcTableHandle getTableHandle(ConnectorSession session, SchemaTableName tableName) + public JdbcTableHandle getTableHandle(ConnectorSession session, SchemaTableName tableName, Optional startVersion, Optional endVersion) { + if (startVersion.isPresent() || endVersion.isPresent()) { + throw new TrinoException(NOT_SUPPORTED, "This connector does not support versioned tables"); + } + return jdbcClient.getTableHandle(session, tableName) .orElse(null); } diff --git a/plugin/trino-base-jdbc/src/test/java/io/trino/plugin/jdbc/TestDefaultJdbcMetadata.java b/plugin/trino-base-jdbc/src/test/java/io/trino/plugin/jdbc/TestDefaultJdbcMetadata.java index fca57731653a..d6c4350fd10f 100644 --- a/plugin/trino-base-jdbc/src/test/java/io/trino/plugin/jdbc/TestDefaultJdbcMetadata.java +++ b/plugin/trino-base-jdbc/src/test/java/io/trino/plugin/jdbc/TestDefaultJdbcMetadata.java @@ -81,7 +81,7 @@ public void setUp() TimestampTimeZoneDomain.ANY, false, ImmutableSet.of()); - tableHandle = metadata.getTableHandle(SESSION, new SchemaTableName("example", "numbers")); + tableHandle = metadata.getTableHandle(SESSION, new SchemaTableName("example", "numbers"), Optional.empty(), Optional.empty()); } @Test @@ -140,11 +140,11 @@ public void testListSchemaNames() @Test public void testGetTableHandle() { - JdbcTableHandle tableHandle = metadata.getTableHandle(SESSION, new SchemaTableName("example", "numbers")); - assertThat(metadata.getTableHandle(SESSION, new SchemaTableName("example", "numbers"))).isEqualTo(tableHandle); - assertThat(metadata.getTableHandle(SESSION, new SchemaTableName("example", "unknown"))).isNull(); - assertThat(metadata.getTableHandle(SESSION, new SchemaTableName("unknown", "numbers"))).isNull(); - assertThat(metadata.getTableHandle(SESSION, new SchemaTableName("unknown", "unknown"))).isNull(); + JdbcTableHandle tableHandle = metadata.getTableHandle(SESSION, new SchemaTableName("example", "numbers"), Optional.empty(), Optional.empty()); + assertThat(metadata.getTableHandle(SESSION, new SchemaTableName("example", "numbers"), Optional.empty(), Optional.empty())).isEqualTo(tableHandle); + assertThat(metadata.getTableHandle(SESSION, new SchemaTableName("example", "unknown"), Optional.empty(), Optional.empty())).isNull(); + assertThat(metadata.getTableHandle(SESSION, new SchemaTableName("unknown", "numbers"), Optional.empty(), Optional.empty())).isNull(); + assertThat(metadata.getTableHandle(SESSION, new SchemaTableName("unknown", "unknown"), Optional.empty(), Optional.empty())).isNull(); } @Test @@ -180,7 +180,7 @@ public void getTableMetadata() new ColumnMetadata("value", BIGINT))); // escaping name patterns - JdbcTableHandle specialTableHandle = metadata.getTableHandle(SESSION, new SchemaTableName("exa_ple", "num_ers")); + JdbcTableHandle specialTableHandle = metadata.getTableHandle(SESSION, new SchemaTableName("exa_ple", "num_ers"), Optional.empty(), Optional.empty()); ConnectorTableMetadata specialTableMetadata = metadata.getTableMetadata(SESSION, specialTableHandle); assertThat(specialTableMetadata.getTable()).isEqualTo(new SchemaTableName("exa_ple", "num_ers")); assertThat(specialTableMetadata.getColumns()).isEqualTo(ImmutableList.of( @@ -248,7 +248,7 @@ public void testCreateAndAlterTable() SchemaTableName table = new SchemaTableName("example", "foo"); metadata.createTable(SESSION, new ConnectorTableMetadata(table, ImmutableList.of(new ColumnMetadata("text", VARCHAR))), false); - JdbcTableHandle handle = metadata.getTableHandle(SESSION, table); + JdbcTableHandle handle = metadata.getTableHandle(SESSION, table, Optional.empty(), Optional.empty()); ConnectorTableMetadata layout = metadata.getTableMetadata(SESSION, handle); assertThat(layout.getTable()).isEqualTo(table); @@ -270,7 +270,7 @@ public void testCreateAndAlterTable() SchemaTableName newTableName = new SchemaTableName("example", "bar"); metadata.renameTable(SESSION, handle, newTableName); - handle = metadata.getTableHandle(SESSION, newTableName); + handle = metadata.getTableHandle(SESSION, newTableName, Optional.empty(), Optional.empty()); layout = metadata.getTableMetadata(SESSION, handle); assertThat(layout.getTable()).isEqualTo(newTableName); assertThat(layout.getColumns()) @@ -300,13 +300,13 @@ public void testAggregationPushdownForTableHandle() ImmutableMap.of(), ImmutableList.of(ImmutableList.of(groupByColumn))); - ConnectorTableHandle baseTableHandle = metadata.getTableHandle(session, new SchemaTableName("example", "numbers")); + ConnectorTableHandle baseTableHandle = metadata.getTableHandle(session, new SchemaTableName("example", "numbers"), Optional.empty(), Optional.empty()); Optional> aggregationResult = applyAggregation.apply(baseTableHandle); assertThat(aggregationResult).isPresent(); SchemaTableName noAggregationPushdownTable = new SchemaTableName("example", "no_aggregation_pushdown"); metadata.createTable(SESSION, new ConnectorTableMetadata(noAggregationPushdownTable, ImmutableList.of(new ColumnMetadata("text", VARCHAR))), false); - ConnectorTableHandle noAggregationPushdownTableHandle = metadata.getTableHandle(session, noAggregationPushdownTable); + ConnectorTableHandle noAggregationPushdownTableHandle = metadata.getTableHandle(session, noAggregationPushdownTable, Optional.empty(), Optional.empty()); aggregationResult = applyAggregation.apply(noAggregationPushdownTableHandle); assertThat(aggregationResult).isEmpty(); } @@ -318,7 +318,7 @@ public void testApplyFilterAfterAggregationPushdown() .setPropertyMetadata(new JdbcMetadataSessionProperties(new JdbcMetadataConfig().setAggregationPushdownEnabled(true), Optional.empty()).getSessionProperties()) .build(); ColumnHandle groupByColumn = metadata.getColumnHandles(session, tableHandle).get("text"); - ConnectorTableHandle baseTableHandle = metadata.getTableHandle(session, new SchemaTableName("example", "numbers")); + ConnectorTableHandle baseTableHandle = metadata.getTableHandle(session, new SchemaTableName("example", "numbers"), Optional.empty(), Optional.empty()); ConnectorTableHandle aggregatedTable = applyCountAggregation(session, baseTableHandle, ImmutableList.of(ImmutableList.of(groupByColumn))); Domain domain = Domain.singleValue(VARCHAR, utf8Slice("one")); @@ -334,7 +334,7 @@ public void testCombineFiltersWithAggregationPushdown() .setPropertyMetadata(new JdbcMetadataSessionProperties(new JdbcMetadataConfig().setAggregationPushdownEnabled(true), Optional.empty()).getSessionProperties()) .build(); ColumnHandle groupByColumn = metadata.getColumnHandles(session, tableHandle).get("text"); - ConnectorTableHandle baseTableHandle = metadata.getTableHandle(session, new SchemaTableName("example", "numbers")); + ConnectorTableHandle baseTableHandle = metadata.getTableHandle(session, new SchemaTableName("example", "numbers"), Optional.empty(), Optional.empty()); Domain firstDomain = Domain.multipleValues(VARCHAR, ImmutableList.of(utf8Slice("one"), utf8Slice("two"))); JdbcTableHandle filterResult = applyFilter(session, baseTableHandle, new Constraint(TupleDomain.withColumnDomains(ImmutableMap.of(groupByColumn, firstDomain)))); @@ -365,7 +365,7 @@ public void testNonGroupKeyPredicatePushdown() ColumnHandle groupByColumn = columnHandles.get("text"); ColumnHandle nonGroupByColumn = columnHandles.get("value"); - ConnectorTableHandle baseTableHandle = metadata.getTableHandle(session, new SchemaTableName("example", "numbers")); + ConnectorTableHandle baseTableHandle = metadata.getTableHandle(session, new SchemaTableName("example", "numbers"), Optional.empty(), Optional.empty()); ConnectorTableHandle aggregatedTable = applyCountAggregation(session, baseTableHandle, ImmutableList.of(ImmutableList.of(groupByColumn))); Domain domain = Domain.singleValue(BIGINT, 123L); @@ -390,7 +390,7 @@ public void testMultiGroupKeyPredicatePushdown() ColumnHandle textColumn = columnHandles.get("text"); ColumnHandle valueColumn = columnHandles.get("value"); - ConnectorTableHandle baseTableHandle = metadata.getTableHandle(session, new SchemaTableName("example", "numbers")); + ConnectorTableHandle baseTableHandle = metadata.getTableHandle(session, new SchemaTableName("example", "numbers"), Optional.empty(), Optional.empty()); ConnectorTableHandle aggregatedTable = applyCountAggregation(session, baseTableHandle, ImmutableList.of(ImmutableList.of(textColumn, valueColumn), ImmutableList.of(textColumn))); @@ -424,7 +424,7 @@ public void testApplyTableScanRedirect() ConnectorSession session = TestingConnectorSession.builder() .setPropertyMetadata(new JdbcMetadataSessionProperties(new JdbcMetadataConfig().setAggregationPushdownEnabled(true), Optional.empty()).getSessionProperties()) .build(); - JdbcTableHandle baseTableHandle = metadata.getTableHandle(session, new SchemaTableName("example", "numbers")); + JdbcTableHandle baseTableHandle = metadata.getTableHandle(session, new SchemaTableName("example", "numbers"), Optional.empty(), Optional.empty()); // redirection is applied if constraintExpressions is empty assertThat(metadata.applyTableScanRedirect(session, baseTableHandle)).hasValueSatisfying(actualResult -> { diff --git a/plugin/trino-bigquery/src/main/java/io/trino/plugin/bigquery/BigQueryMetadata.java b/plugin/trino-bigquery/src/main/java/io/trino/plugin/bigquery/BigQueryMetadata.java index 380c493cd751..9eee648b90cf 100644 --- a/plugin/trino-bigquery/src/main/java/io/trino/plugin/bigquery/BigQueryMetadata.java +++ b/plugin/trino-bigquery/src/main/java/io/trino/plugin/bigquery/BigQueryMetadata.java @@ -54,6 +54,7 @@ import io.trino.spi.connector.ConnectorTableHandle; import io.trino.spi.connector.ConnectorTableLayout; import io.trino.spi.connector.ConnectorTableMetadata; +import io.trino.spi.connector.ConnectorTableVersion; import io.trino.spi.connector.ConnectorTransactionHandle; import io.trino.spi.connector.Constraint; import io.trino.spi.connector.ConstraintApplicationResult; @@ -112,6 +113,7 @@ import static io.trino.plugin.bigquery.BigQueryUtil.quote; import static io.trino.plugin.bigquery.BigQueryUtil.quoted; import static io.trino.spi.StandardErrorCode.GENERIC_INTERNAL_ERROR; +import static io.trino.spi.StandardErrorCode.NOT_SUPPORTED; import static io.trino.spi.type.BigintType.BIGINT; import static java.lang.String.format; import static java.util.Locale.ENGLISH; @@ -285,8 +287,12 @@ private static Stream listRelationCommentMetadata(Conne } @Override - public ConnectorTableHandle getTableHandle(ConnectorSession session, SchemaTableName schemaTableName) + public ConnectorTableHandle getTableHandle(ConnectorSession session, SchemaTableName schemaTableName, Optional startVersion, Optional endVersion) { + if (startVersion.isPresent() || endVersion.isPresent()) { + throw new TrinoException(NOT_SUPPORTED, "This connector does not support versioned tables"); + } + BigQueryClient client = bigQueryClientFactory.create(session); log.debug("getTableHandle(session=%s, schemaTableName=%s)", session, schemaTableName); DatasetId localDatasetId = client.toDatasetId(schemaTableName.getSchemaName()); diff --git a/plugin/trino-blackhole/src/main/java/io/trino/plugin/blackhole/BlackHoleMetadata.java b/plugin/trino-blackhole/src/main/java/io/trino/plugin/blackhole/BlackHoleMetadata.java index 9be92de0e061..aa6bd40808d8 100644 --- a/plugin/trino-blackhole/src/main/java/io/trino/plugin/blackhole/BlackHoleMetadata.java +++ b/plugin/trino-blackhole/src/main/java/io/trino/plugin/blackhole/BlackHoleMetadata.java @@ -32,6 +32,7 @@ import io.trino.spi.connector.ConnectorTableHandle; import io.trino.spi.connector.ConnectorTableLayout; import io.trino.spi.connector.ConnectorTableMetadata; +import io.trino.spi.connector.ConnectorTableVersion; import io.trino.spi.connector.ConnectorViewDefinition; import io.trino.spi.connector.RetryMode; import io.trino.spi.connector.RowChangeParadigm; @@ -71,6 +72,7 @@ import static io.trino.plugin.blackhole.BlackHolePageSourceProvider.isNumericType; import static io.trino.spi.StandardErrorCode.ALREADY_EXISTS; import static io.trino.spi.StandardErrorCode.INVALID_TABLE_PROPERTY; +import static io.trino.spi.StandardErrorCode.NOT_SUPPORTED; import static io.trino.spi.connector.RetryMode.NO_RETRIES; import static io.trino.spi.connector.RowChangeParadigm.DELETE_ROW_AND_INSERT_ROW; import static io.trino.spi.type.BigintType.BIGINT; @@ -108,8 +110,12 @@ public synchronized void createSchema(ConnectorSession session, String schemaNam } @Override - public ConnectorTableHandle getTableHandle(ConnectorSession session, SchemaTableName tableName) + public ConnectorTableHandle getTableHandle(ConnectorSession session, SchemaTableName tableName, Optional startVersion, Optional endVersion) { + if (startVersion.isPresent() || endVersion.isPresent()) { + throw new TrinoException(NOT_SUPPORTED, "This connector does not support versioned tables"); + } + return tables.get(tableName); } diff --git a/plugin/trino-cassandra/src/main/java/io/trino/plugin/cassandra/CassandraMetadata.java b/plugin/trino-cassandra/src/main/java/io/trino/plugin/cassandra/CassandraMetadata.java index 9360afc5a79d..0b5619a5813c 100644 --- a/plugin/trino-cassandra/src/main/java/io/trino/plugin/cassandra/CassandraMetadata.java +++ b/plugin/trino-cassandra/src/main/java/io/trino/plugin/cassandra/CassandraMetadata.java @@ -35,6 +35,7 @@ import io.trino.spi.connector.ConnectorTableHandle; import io.trino.spi.connector.ConnectorTableLayout; import io.trino.spi.connector.ConnectorTableMetadata; +import io.trino.spi.connector.ConnectorTableVersion; import io.trino.spi.connector.Constraint; import io.trino.spi.connector.ConstraintApplicationResult; import io.trino.spi.connector.NotFoundException; @@ -111,10 +112,13 @@ public List listSchemaNames(ConnectorSession session) .collect(toImmutableList()); } - @SuppressWarnings("deprecation") // TODO Implement getTableHandle(ConnectorSession, SchemaTableName, Optional, Optional) method @Override - public CassandraTableHandle getTableHandle(ConnectorSession session, SchemaTableName tableName) + public CassandraTableHandle getTableHandle(ConnectorSession session, SchemaTableName tableName, Optional startVersion, Optional endVersion) { + if (startVersion.isPresent() || endVersion.isPresent()) { + throw new TrinoException(NOT_SUPPORTED, "This connector does not support versioned tables"); + } + requireNonNull(tableName, "tableName is null"); try { return new CassandraTableHandle(cassandraSession.getTable(tableName).tableHandle()); diff --git a/plugin/trino-delta-lake/src/main/java/io/trino/plugin/deltalake/DeltaLakeMetadata.java b/plugin/trino-delta-lake/src/main/java/io/trino/plugin/deltalake/DeltaLakeMetadata.java index 3f67136f4220..63dcc05c4510 100644 --- a/plugin/trino-delta-lake/src/main/java/io/trino/plugin/deltalake/DeltaLakeMetadata.java +++ b/plugin/trino-delta-lake/src/main/java/io/trino/plugin/deltalake/DeltaLakeMetadata.java @@ -559,12 +559,6 @@ public Optional redirectTable(ConnectorSession session, return Optional.empty(); } - @Override - public LocatedTableHandle getTableHandle(ConnectorSession session, SchemaTableName tableName) - { - throw new UnsupportedOperationException("Calling this method is not supported. getTableHandle with versions is implemented"); - } - @Override public LocatedTableHandle getTableHandle( ConnectorSession session, diff --git a/plugin/trino-elasticsearch/src/main/java/io/trino/plugin/elasticsearch/ElasticsearchMetadata.java b/plugin/trino-elasticsearch/src/main/java/io/trino/plugin/elasticsearch/ElasticsearchMetadata.java index 33aa3b92216c..ece364295837 100644 --- a/plugin/trino-elasticsearch/src/main/java/io/trino/plugin/elasticsearch/ElasticsearchMetadata.java +++ b/plugin/trino-elasticsearch/src/main/java/io/trino/plugin/elasticsearch/ElasticsearchMetadata.java @@ -48,6 +48,7 @@ import io.trino.spi.connector.ConnectorTableHandle; import io.trino.spi.connector.ConnectorTableMetadata; import io.trino.spi.connector.ConnectorTableProperties; +import io.trino.spi.connector.ConnectorTableVersion; import io.trino.spi.connector.Constraint; import io.trino.spi.connector.ConstraintApplicationResult; import io.trino.spi.connector.LimitApplicationResult; @@ -92,6 +93,7 @@ import static io.trino.plugin.elasticsearch.ElasticsearchTableHandle.Type.QUERY; import static io.trino.plugin.elasticsearch.ElasticsearchTableHandle.Type.SCAN; import static io.trino.spi.StandardErrorCode.INVALID_FUNCTION_ARGUMENT; +import static io.trino.spi.StandardErrorCode.NOT_SUPPORTED; import static io.trino.spi.expression.StandardFunctions.LIKE_FUNCTION_NAME; import static io.trino.spi.type.BigintType.BIGINT; import static io.trino.spi.type.BooleanType.BOOLEAN; @@ -151,8 +153,12 @@ public List listSchemaNames(ConnectorSession session) } @Override - public ElasticsearchTableHandle getTableHandle(ConnectorSession session, SchemaTableName tableName) + public ElasticsearchTableHandle getTableHandle(ConnectorSession session, SchemaTableName tableName, Optional startVersion, Optional endVersion) { + if (startVersion.isPresent() || endVersion.isPresent()) { + throw new TrinoException(NOT_SUPPORTED, "This connector does not support versioned tables"); + } + requireNonNull(tableName, "tableName is null"); if (tableName.getSchemaName().equals(schemaName)) { diff --git a/plugin/trino-example-http/src/main/java/io/trino/plugin/example/ExampleMetadata.java b/plugin/trino-example-http/src/main/java/io/trino/plugin/example/ExampleMetadata.java index 2842528bbb2a..83aa1d2b406c 100644 --- a/plugin/trino-example-http/src/main/java/io/trino/plugin/example/ExampleMetadata.java +++ b/plugin/trino-example-http/src/main/java/io/trino/plugin/example/ExampleMetadata.java @@ -17,12 +17,14 @@ import com.google.common.collect.ImmutableMap; import com.google.common.collect.ImmutableSet; import com.google.inject.Inject; +import io.trino.spi.TrinoException; import io.trino.spi.connector.ColumnHandle; import io.trino.spi.connector.ColumnMetadata; import io.trino.spi.connector.ConnectorMetadata; import io.trino.spi.connector.ConnectorSession; import io.trino.spi.connector.ConnectorTableHandle; import io.trino.spi.connector.ConnectorTableMetadata; +import io.trino.spi.connector.ConnectorTableVersion; import io.trino.spi.connector.SchemaTableName; import io.trino.spi.connector.SchemaTablePrefix; import io.trino.spi.connector.TableNotFoundException; @@ -32,6 +34,7 @@ import java.util.Optional; import java.util.Set; +import static io.trino.spi.StandardErrorCode.NOT_SUPPORTED; import static java.util.Objects.requireNonNull; public class ExampleMetadata @@ -57,8 +60,12 @@ public List listSchemaNames() } @Override - public ExampleTableHandle getTableHandle(ConnectorSession session, SchemaTableName tableName) + public ExampleTableHandle getTableHandle(ConnectorSession session, SchemaTableName tableName, Optional startVersion, Optional endVersion) { + if (startVersion.isPresent() || endVersion.isPresent()) { + throw new TrinoException(NOT_SUPPORTED, "This connector does not support versioned tables"); + } + if (!listSchemaNames(session).contains(tableName.getSchemaName())) { return null; } diff --git a/plugin/trino-example-http/src/test/java/io/trino/plugin/example/TestExampleMetadata.java b/plugin/trino-example-http/src/test/java/io/trino/plugin/example/TestExampleMetadata.java index 704aba27723a..aaebbbaa60f2 100644 --- a/plugin/trino-example-http/src/test/java/io/trino/plugin/example/TestExampleMetadata.java +++ b/plugin/trino-example-http/src/test/java/io/trino/plugin/example/TestExampleMetadata.java @@ -64,10 +64,10 @@ public void testListSchemaNames() @Test public void testGetTableHandle() { - assertThat(metadata.getTableHandle(SESSION, new SchemaTableName("example", "numbers"))).isEqualTo(NUMBERS_TABLE_HANDLE); - assertThat(metadata.getTableHandle(SESSION, new SchemaTableName("example", "unknown"))).isNull(); - assertThat(metadata.getTableHandle(SESSION, new SchemaTableName("unknown", "numbers"))).isNull(); - assertThat(metadata.getTableHandle(SESSION, new SchemaTableName("unknown", "unknown"))).isNull(); + assertThat(metadata.getTableHandle(SESSION, new SchemaTableName("example", "numbers"), Optional.empty(), Optional.empty())).isEqualTo(NUMBERS_TABLE_HANDLE); + assertThat(metadata.getTableHandle(SESSION, new SchemaTableName("example", "unknown"), Optional.empty(), Optional.empty())).isNull(); + assertThat(metadata.getTableHandle(SESSION, new SchemaTableName("unknown", "numbers"), Optional.empty(), Optional.empty())).isNull(); + assertThat(metadata.getTableHandle(SESSION, new SchemaTableName("unknown", "unknown"), Optional.empty(), Optional.empty())).isNull(); } @Test diff --git a/plugin/trino-google-sheets/src/main/java/io/trino/plugin/google/sheets/SheetsMetadata.java b/plugin/trino-google-sheets/src/main/java/io/trino/plugin/google/sheets/SheetsMetadata.java index f51b022f75d4..c109186108bb 100644 --- a/plugin/trino-google-sheets/src/main/java/io/trino/plugin/google/sheets/SheetsMetadata.java +++ b/plugin/trino-google-sheets/src/main/java/io/trino/plugin/google/sheets/SheetsMetadata.java @@ -26,6 +26,7 @@ import io.trino.spi.connector.ConnectorSession; import io.trino.spi.connector.ConnectorTableHandle; import io.trino.spi.connector.ConnectorTableMetadata; +import io.trino.spi.connector.ConnectorTableVersion; import io.trino.spi.connector.RetryMode; import io.trino.spi.connector.SchemaTableName; import io.trino.spi.connector.SchemaTablePrefix; @@ -75,8 +76,12 @@ public List listSchemaNames() } @Override - public SheetsNamedTableHandle getTableHandle(ConnectorSession session, SchemaTableName tableName) + public SheetsNamedTableHandle getTableHandle(ConnectorSession session, SchemaTableName tableName, Optional startVersion, Optional endVersion) { + if (startVersion.isPresent() || endVersion.isPresent()) { + throw new TrinoException(NOT_SUPPORTED, "This connector does not support versioned tables"); + } + requireNonNull(tableName, "tableName is null"); if (!listSchemaNames(session).contains(tableName.getSchemaName())) { return null; diff --git a/plugin/trino-hive/src/main/java/io/trino/plugin/hive/HiveMetadata.java b/plugin/trino-hive/src/main/java/io/trino/plugin/hive/HiveMetadata.java index fd5b4c2f3951..c5abc539f989 100644 --- a/plugin/trino-hive/src/main/java/io/trino/plugin/hive/HiveMetadata.java +++ b/plugin/trino-hive/src/main/java/io/trino/plugin/hive/HiveMetadata.java @@ -80,6 +80,7 @@ import io.trino.spi.connector.ConnectorTableMetadata; import io.trino.spi.connector.ConnectorTablePartitioning; import io.trino.spi.connector.ConnectorTableProperties; +import io.trino.spi.connector.ConnectorTableVersion; import io.trino.spi.connector.ConnectorViewDefinition; import io.trino.spi.connector.Constraint; import io.trino.spi.connector.ConstraintApplicationResult; @@ -488,8 +489,12 @@ public List listSchemaNames(ConnectorSession session) } @Override - public HiveTableHandle getTableHandle(ConnectorSession session, SchemaTableName tableName) + public HiveTableHandle getTableHandle(ConnectorSession session, SchemaTableName tableName, Optional startVersion, Optional endVersion) { + if (startVersion.isPresent() || endVersion.isPresent()) { + throw new TrinoException(NOT_SUPPORTED, "This connector does not support versioned tables"); + } + requireNonNull(tableName, "tableName is null"); if (isHiveSystemSchema(tableName.getSchemaName())) { return null; @@ -982,7 +987,7 @@ public void dropSchema(ConnectorSession session, String schemaName, boolean casc } for (SchemaTableName tableName : tables) { - ConnectorTableHandle table = getTableHandle(session, tableName); + ConnectorTableHandle table = getTableHandle(session, tableName, Optional.empty(), Optional.empty()); if (table == null) { log.debug("Table disappeared during DROP SCHEMA CASCADE: %s", tableName); continue; diff --git a/plugin/trino-hive/src/main/java/io/trino/plugin/hive/procedure/CreateEmptyPartitionProcedure.java b/plugin/trino-hive/src/main/java/io/trino/plugin/hive/procedure/CreateEmptyPartitionProcedure.java index ccbdae22b227..785712786168 100644 --- a/plugin/trino-hive/src/main/java/io/trino/plugin/hive/procedure/CreateEmptyPartitionProcedure.java +++ b/plugin/trino-hive/src/main/java/io/trino/plugin/hive/procedure/CreateEmptyPartitionProcedure.java @@ -112,7 +112,7 @@ private void doCreateEmptyPartition(ConnectorSession session, ConnectorAccessCon TransactionalMetadata hiveMetadata = hiveMetadataFactory.create(session.getIdentity(), true); hiveMetadata.beginQuery(session); try (UncheckedCloseable ignore = () -> hiveMetadata.cleanupQuery(session)) { - HiveTableHandle tableHandle = (HiveTableHandle) hiveMetadata.getTableHandle(session, new SchemaTableName(schemaName, tableName)); + HiveTableHandle tableHandle = (HiveTableHandle) hiveMetadata.getTableHandle(session, new SchemaTableName(schemaName, tableName), Optional.empty(), Optional.empty()); if (tableHandle == null) { throw new TrinoException(INVALID_PROCEDURE_ARGUMENT, format("Table '%s' does not exist", new SchemaTableName(schemaName, tableName))); } diff --git a/plugin/trino-hive/src/main/java/io/trino/plugin/hive/procedure/DropStatsProcedure.java b/plugin/trino-hive/src/main/java/io/trino/plugin/hive/procedure/DropStatsProcedure.java index 1f255d7a8bee..1e529065b9cc 100644 --- a/plugin/trino-hive/src/main/java/io/trino/plugin/hive/procedure/DropStatsProcedure.java +++ b/plugin/trino-hive/src/main/java/io/trino/plugin/hive/procedure/DropStatsProcedure.java @@ -38,6 +38,7 @@ import java.lang.invoke.MethodHandle; import java.util.List; import java.util.Map; +import java.util.Optional; import static com.google.common.collect.ImmutableList.toImmutableList; import static io.trino.plugin.base.util.Procedures.checkProcedureArgument; @@ -105,7 +106,7 @@ private void doDropStats(ConnectorSession session, ConnectorAccessControl access TransactionalMetadata hiveMetadata = hiveMetadataFactory.create(session.getIdentity(), true); hiveMetadata.beginQuery(session); try (UncheckedCloseable ignore = () -> hiveMetadata.cleanupQuery(session)) { - HiveTableHandle handle = (HiveTableHandle) hiveMetadata.getTableHandle(session, new SchemaTableName(schema, table)); + HiveTableHandle handle = (HiveTableHandle) hiveMetadata.getTableHandle(session, new SchemaTableName(schema, table), Optional.empty(), Optional.empty()); if (handle == null) { throw new TrinoException(INVALID_PROCEDURE_ARGUMENT, format("Table '%s' does not exist", new SchemaTableName(schema, table))); } diff --git a/plugin/trino-hudi/src/main/java/io/trino/plugin/hudi/HudiMetadata.java b/plugin/trino-hudi/src/main/java/io/trino/plugin/hudi/HudiMetadata.java index 9f51e835d64e..3794275dee97 100644 --- a/plugin/trino-hudi/src/main/java/io/trino/plugin/hudi/HudiMetadata.java +++ b/plugin/trino-hudi/src/main/java/io/trino/plugin/hudi/HudiMetadata.java @@ -30,6 +30,7 @@ import io.trino.spi.connector.ConnectorSession; import io.trino.spi.connector.ConnectorTableHandle; import io.trino.spi.connector.ConnectorTableMetadata; +import io.trino.spi.connector.ConnectorTableVersion; import io.trino.spi.connector.Constraint; import io.trino.spi.connector.ConstraintApplicationResult; import io.trino.spi.connector.SchemaTableName; @@ -68,6 +69,7 @@ import static io.trino.plugin.hudi.HudiTableProperties.PARTITIONED_BY_PROPERTY; import static io.trino.plugin.hudi.HudiUtil.hudiMetadataExists; import static io.trino.plugin.hudi.model.HudiTableType.COPY_ON_WRITE; +import static io.trino.spi.StandardErrorCode.NOT_SUPPORTED; import static io.trino.spi.StandardErrorCode.QUERY_REJECTED; import static io.trino.spi.StandardErrorCode.UNSUPPORTED_TABLE_TYPE; import static io.trino.spi.connector.SchemaTableName.schemaTableName; @@ -99,8 +101,12 @@ public List listSchemaNames(ConnectorSession session) } @Override - public HudiTableHandle getTableHandle(ConnectorSession session, SchemaTableName tableName) + public HudiTableHandle getTableHandle(ConnectorSession session, SchemaTableName tableName, Optional startVersion, Optional endVersion) { + if (startVersion.isPresent() || endVersion.isPresent()) { + throw new TrinoException(NOT_SUPPORTED, "This connector does not support versioned tables"); + } + if (isHiveSystemSchema(tableName.getSchemaName())) { return null; } diff --git a/plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/IcebergMetadata.java b/plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/IcebergMetadata.java index 1ca4e15641f0..39cc36bb117b 100644 --- a/plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/IcebergMetadata.java +++ b/plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/IcebergMetadata.java @@ -374,12 +374,6 @@ public Optional getSchemaOwner(ConnectorSession session, String return catalog.getNamespacePrincipal(session, schemaName); } - @Override - public IcebergTableHandle getTableHandle(ConnectorSession session, SchemaTableName tableName) - { - throw new UnsupportedOperationException("This method is not supported because getTableHandle with versions is implemented instead"); - } - @Override public ConnectorTableHandle getTableHandle( ConnectorSession session, diff --git a/plugin/trino-ignite/src/main/java/io/trino/plugin/ignite/IgniteMetadata.java b/plugin/trino-ignite/src/main/java/io/trino/plugin/ignite/IgniteMetadata.java index 04c5b8eed56c..44b684a8036f 100644 --- a/plugin/trino-ignite/src/main/java/io/trino/plugin/ignite/IgniteMetadata.java +++ b/plugin/trino-ignite/src/main/java/io/trino/plugin/ignite/IgniteMetadata.java @@ -35,6 +35,7 @@ import io.trino.spi.connector.ConnectorTableLayout; import io.trino.spi.connector.ConnectorTableMetadata; import io.trino.spi.connector.ConnectorTableSchema; +import io.trino.spi.connector.ConnectorTableVersion; import io.trino.spi.connector.RetryMode; import io.trino.spi.connector.SchemaTableName; import io.trino.spi.statistics.ComputedStatistics; @@ -63,8 +64,12 @@ public IgniteMetadata(JdbcClient igniteClient, TimestampTimeZoneDomain timestamp } @Override - public JdbcTableHandle getTableHandle(ConnectorSession session, SchemaTableName schemaTableName) + public JdbcTableHandle getTableHandle(ConnectorSession session, SchemaTableName schemaTableName, Optional startVersion, Optional endVersion) { + if (startVersion.isPresent() || endVersion.isPresent()) { + throw new TrinoException(NOT_SUPPORTED, "This connector does not support versioned tables"); + } + return igniteClient.getTableHandle(session, schemaTableName) .map(JdbcTableHandle::asPlainTable) .map(JdbcNamedRelationHandle::getRemoteTableName) diff --git a/plugin/trino-jmx/src/main/java/io/trino/plugin/jmx/JmxMetadata.java b/plugin/trino-jmx/src/main/java/io/trino/plugin/jmx/JmxMetadata.java index 158471ca4fdd..26ce36603652 100644 --- a/plugin/trino-jmx/src/main/java/io/trino/plugin/jmx/JmxMetadata.java +++ b/plugin/trino-jmx/src/main/java/io/trino/plugin/jmx/JmxMetadata.java @@ -28,6 +28,7 @@ import io.trino.spi.connector.ConnectorSession; import io.trino.spi.connector.ConnectorTableHandle; import io.trino.spi.connector.ConnectorTableMetadata; +import io.trino.spi.connector.ConnectorTableVersion; import io.trino.spi.connector.Constraint; import io.trino.spi.connector.ConstraintApplicationResult; import io.trino.spi.connector.SchemaTableName; @@ -58,6 +59,7 @@ import static com.google.common.collect.ImmutableList.toImmutableList; import static io.trino.plugin.jmx.JmxErrorCode.JMX_INVALID_TABLE_NAME; +import static io.trino.spi.StandardErrorCode.NOT_SUPPORTED; import static io.trino.spi.type.BigintType.BIGINT; import static io.trino.spi.type.BooleanType.BOOLEAN; import static io.trino.spi.type.DoubleType.DOUBLE; @@ -97,8 +99,12 @@ public List listSchemaNames(ConnectorSession session) } @Override - public JmxTableHandle getTableHandle(ConnectorSession session, SchemaTableName tableName) + public JmxTableHandle getTableHandle(ConnectorSession session, SchemaTableName tableName, Optional startVersion, Optional endVersion) { + if (startVersion.isPresent() || endVersion.isPresent()) { + throw new TrinoException(NOT_SUPPORTED, "This connector does not support versioned tables"); + } + return getTableHandle(tableName); } @@ -255,7 +261,7 @@ public Iterator streamTableColumns(ConnectorSession sessio } return tableNames.stream() - .map(tableName -> TableColumnsMetadata.forTable(tableName, getTableHandle(session, tableName).getTableMetadata().getColumns())) + .map(tableName -> TableColumnsMetadata.forTable(tableName, getTableHandle(session, tableName, Optional.empty(), Optional.empty()).getTableMetadata().getColumns())) .iterator(); } diff --git a/plugin/trino-jmx/src/test/java/io/trino/plugin/jmx/TestJmxMetadata.java b/plugin/trino-jmx/src/test/java/io/trino/plugin/jmx/TestJmxMetadata.java index ffbbc61eb0c4..4452fd5b9c66 100644 --- a/plugin/trino-jmx/src/test/java/io/trino/plugin/jmx/TestJmxMetadata.java +++ b/plugin/trino-jmx/src/test/java/io/trino/plugin/jmx/TestJmxMetadata.java @@ -70,7 +70,7 @@ public void testListTables() @Test public void testGetTableHandle() { - JmxTableHandle handle = metadata.getTableHandle(SESSION, RUNTIME_TABLE); + JmxTableHandle handle = metadata.getTableHandle(SESSION, RUNTIME_TABLE, Optional.empty(), Optional.empty()); assertThat(handle.objectNames()).isEqualTo(ImmutableList.of(RUNTIME_OBJECT)); List columns = handle.columnHandles(); @@ -82,7 +82,7 @@ public void testGetTableHandle() @Test public void testGetTimeTableHandle() { - JmxTableHandle handle = metadata.getTableHandle(SESSION, RUNTIME_HISTORY_TABLE); + JmxTableHandle handle = metadata.getTableHandle(SESSION, RUNTIME_HISTORY_TABLE, Optional.empty(), Optional.empty()); assertThat(handle.objectNames()).isEqualTo(ImmutableList.of(RUNTIME_OBJECT)); List columns = handle.columnHandles(); @@ -95,7 +95,7 @@ public void testGetTimeTableHandle() @Test public void testGetCumulativeTableHandle() { - JmxTableHandle handle = metadata.getTableHandle(SESSION, new SchemaTableName(JMX_SCHEMA_NAME, "java.lang:*")); + JmxTableHandle handle = metadata.getTableHandle(SESSION, new SchemaTableName(JMX_SCHEMA_NAME, "java.lang:*"), Optional.empty(), Optional.empty()); assertThat(handle.objectNames()).contains(RUNTIME_OBJECT); assertThat(handle.objectNames()).hasSizeGreaterThan(1); @@ -105,16 +105,16 @@ public void testGetCumulativeTableHandle() assertThat(columns).contains(new JmxColumnHandle("Name", createUnboundedVarcharType())); assertThat(columns).contains(new JmxColumnHandle("StartTime", BIGINT)); - assertThat(metadata.getTableHandle(SESSION, new SchemaTableName(JMX_SCHEMA_NAME, "*java.lang:type=Runtime*")).objectNames()).contains(RUNTIME_OBJECT); - assertThat(metadata.getTableHandle(SESSION, new SchemaTableName(JMX_SCHEMA_NAME, "java.lang:*=Runtime")).objectNames()).contains(RUNTIME_OBJECT); - assertThat(metadata.getTableHandle(SESSION, new SchemaTableName(JMX_SCHEMA_NAME, "*")).objectNames()).contains(RUNTIME_OBJECT); - assertThat(metadata.getTableHandle(SESSION, new SchemaTableName(JMX_SCHEMA_NAME, "*:*")).objectNames()).contains(RUNTIME_OBJECT); + assertThat(metadata.getTableHandle(SESSION, new SchemaTableName(JMX_SCHEMA_NAME, "*java.lang:type=Runtime*"), Optional.empty(), Optional.empty()).objectNames()).contains(RUNTIME_OBJECT); + assertThat(metadata.getTableHandle(SESSION, new SchemaTableName(JMX_SCHEMA_NAME, "java.lang:*=Runtime"), Optional.empty(), Optional.empty()).objectNames()).contains(RUNTIME_OBJECT); + assertThat(metadata.getTableHandle(SESSION, new SchemaTableName(JMX_SCHEMA_NAME, "*"), Optional.empty(), Optional.empty()).objectNames()).contains(RUNTIME_OBJECT); + assertThat(metadata.getTableHandle(SESSION, new SchemaTableName(JMX_SCHEMA_NAME, "*:*"), Optional.empty(), Optional.empty()).objectNames()).contains(RUNTIME_OBJECT); } @Test public void testGetCumulativeTableHandleForHistorySchema() { - JmxTableHandle handle = metadata.getTableHandle(SESSION, new SchemaTableName(HISTORY_SCHEMA_NAME, PATTERN)); + JmxTableHandle handle = metadata.getTableHandle(SESSION, new SchemaTableName(HISTORY_SCHEMA_NAME, PATTERN), Optional.empty(), Optional.empty()); assertThat(handle.objectNames()).contains(RUNTIME_OBJECT); assertThat(handle.objectNames()).hasSizeGreaterThan(1); @@ -125,16 +125,16 @@ public void testGetCumulativeTableHandleForHistorySchema() assertThat(columns).contains(new JmxColumnHandle("Name", createUnboundedVarcharType())); assertThat(columns).contains(new JmxColumnHandle("StartTime", BIGINT)); - assertThat(metadata.getTableHandle(SESSION, new SchemaTableName(HISTORY_SCHEMA_NAME, "*java.lang:type=Runtime*")).objectNames()).contains(RUNTIME_OBJECT); - assertThat(metadata.getTableHandle(SESSION, new SchemaTableName(HISTORY_SCHEMA_NAME, "java.lang:*=Runtime")).objectNames()).contains(RUNTIME_OBJECT); - assertThat(metadata.getTableHandle(SESSION, new SchemaTableName(HISTORY_SCHEMA_NAME, "*")).objectNames()).contains(RUNTIME_OBJECT); - assertThat(metadata.getTableHandle(SESSION, new SchemaTableName(HISTORY_SCHEMA_NAME, "*:*")).objectNames()).contains(RUNTIME_OBJECT); + assertThat(metadata.getTableHandle(SESSION, new SchemaTableName(HISTORY_SCHEMA_NAME, "*java.lang:type=Runtime*"), Optional.empty(), Optional.empty()).objectNames()).contains(RUNTIME_OBJECT); + assertThat(metadata.getTableHandle(SESSION, new SchemaTableName(HISTORY_SCHEMA_NAME, "java.lang:*=Runtime"), Optional.empty(), Optional.empty()).objectNames()).contains(RUNTIME_OBJECT); + assertThat(metadata.getTableHandle(SESSION, new SchemaTableName(HISTORY_SCHEMA_NAME, "*"), Optional.empty(), Optional.empty()).objectNames()).contains(RUNTIME_OBJECT); + assertThat(metadata.getTableHandle(SESSION, new SchemaTableName(HISTORY_SCHEMA_NAME, "*:*"), Optional.empty(), Optional.empty()).objectNames()).contains(RUNTIME_OBJECT); } @Test public void testApplyFilterWithoutConstraint() { - JmxTableHandle handle = metadata.getTableHandle(SESSION, new SchemaTableName(JMX_SCHEMA_NAME, "java.lang:*")); + JmxTableHandle handle = metadata.getTableHandle(SESSION, new SchemaTableName(JMX_SCHEMA_NAME, "java.lang:*"), Optional.empty(), Optional.empty()); Optional> result = metadata.applyFilter(SESSION, handle, new Constraint(TupleDomain.all())); assertThat(result).isNotPresent(); @@ -143,7 +143,7 @@ public void testApplyFilterWithoutConstraint() @Test public void testApplyFilterWithConstraint() { - JmxTableHandle handle = metadata.getTableHandle(SESSION, new SchemaTableName(JMX_SCHEMA_NAME, "java.lang:*")); + JmxTableHandle handle = metadata.getTableHandle(SESSION, new SchemaTableName(JMX_SCHEMA_NAME, "java.lang:*"), Optional.empty(), Optional.empty()); JmxColumnHandle nodeColumnHandle = new JmxColumnHandle("node", createUnboundedVarcharType()); NullableValue nodeColumnValue = NullableValue.of(createUnboundedVarcharType(), utf8Slice(localNode.getNodeIdentifier())); @@ -165,7 +165,7 @@ public void testApplyFilterWithConstraint() @Test public void testApplyFilterWithSameConstraint() { - JmxTableHandle handle = metadata.getTableHandle(SESSION, new SchemaTableName(JMX_SCHEMA_NAME, "java.lang:*")); + JmxTableHandle handle = metadata.getTableHandle(SESSION, new SchemaTableName(JMX_SCHEMA_NAME, "java.lang:*"), Optional.empty(), Optional.empty()); JmxColumnHandle columnHandle = new JmxColumnHandle("node", createUnboundedVarcharType()); TupleDomain nodeTupleDomain = TupleDomain.fromFixedValues(ImmutableMap.of(columnHandle, NullableValue.of(createUnboundedVarcharType(), utf8Slice(localNode.getNodeIdentifier())))); diff --git a/plugin/trino-jmx/src/test/java/io/trino/plugin/jmx/TestJmxSplitManager.java b/plugin/trino-jmx/src/test/java/io/trino/plugin/jmx/TestJmxSplitManager.java index 2f8497152d29..9c1add71c322 100644 --- a/plugin/trino-jmx/src/test/java/io/trino/plugin/jmx/TestJmxSplitManager.java +++ b/plugin/trino-jmx/src/test/java/io/trino/plugin/jmx/TestJmxSplitManager.java @@ -199,7 +199,7 @@ private List readTimeStampsFrom(RecordSet recordSet) private RecordSet getRecordSet(SchemaTableName schemaTableName) throws Exception { - JmxTableHandle tableHandle = metadata.getTableHandle(SESSION, schemaTableName); + JmxTableHandle tableHandle = metadata.getTableHandle(SESSION, schemaTableName, Optional.empty(), Optional.empty()); List columnHandles = ImmutableList.copyOf(metadata.getColumnHandles(SESSION, tableHandle).values()); ConnectorSplitSource splitSource = splitManager.getSplits(JmxTransactionHandle.INSTANCE, SESSION, tableHandle, DynamicFilter.EMPTY, Constraint.alwaysTrue()); diff --git a/plugin/trino-kafka/src/main/java/io/trino/plugin/kafka/KafkaMetadata.java b/plugin/trino-kafka/src/main/java/io/trino/plugin/kafka/KafkaMetadata.java index c79895f0fccb..7747834c93d7 100644 --- a/plugin/trino-kafka/src/main/java/io/trino/plugin/kafka/KafkaMetadata.java +++ b/plugin/trino-kafka/src/main/java/io/trino/plugin/kafka/KafkaMetadata.java @@ -28,6 +28,7 @@ import io.trino.spi.connector.ConnectorSession; import io.trino.spi.connector.ConnectorTableHandle; import io.trino.spi.connector.ConnectorTableMetadata; +import io.trino.spi.connector.ConnectorTableVersion; import io.trino.spi.connector.Constraint; import io.trino.spi.connector.ConstraintApplicationResult; import io.trino.spi.connector.RetryMode; @@ -87,8 +88,12 @@ public List listSchemaNames(ConnectorSession session) } @Override - public KafkaTableHandle getTableHandle(ConnectorSession session, SchemaTableName schemaTableName) + public KafkaTableHandle getTableHandle(ConnectorSession session, SchemaTableName schemaTableName, Optional startVersion, Optional endVersion) { + if (startVersion.isPresent() || endVersion.isPresent()) { + throw new TrinoException(NOT_SUPPORTED, "This connector does not support versioned tables"); + } + return getTopicDescription(session, schemaTableName) .map(kafkaTopicDescription -> new KafkaTableHandle( schemaTableName.getSchemaName(), diff --git a/plugin/trino-kinesis/src/main/java/io/trino/plugin/kinesis/KinesisMetadata.java b/plugin/trino-kinesis/src/main/java/io/trino/plugin/kinesis/KinesisMetadata.java index 0685e1f17a2d..4a177c28c711 100644 --- a/plugin/trino-kinesis/src/main/java/io/trino/plugin/kinesis/KinesisMetadata.java +++ b/plugin/trino-kinesis/src/main/java/io/trino/plugin/kinesis/KinesisMetadata.java @@ -17,12 +17,14 @@ import com.google.common.collect.ImmutableMap; import com.google.inject.Inject; import io.trino.decoder.dummy.DummyRowDecoder; +import io.trino.spi.TrinoException; import io.trino.spi.connector.ColumnHandle; import io.trino.spi.connector.ColumnMetadata; import io.trino.spi.connector.ConnectorMetadata; import io.trino.spi.connector.ConnectorSession; import io.trino.spi.connector.ConnectorTableHandle; import io.trino.spi.connector.ConnectorTableMetadata; +import io.trino.spi.connector.ConnectorTableVersion; import io.trino.spi.connector.SchemaTableName; import io.trino.spi.connector.SchemaTablePrefix; import io.trino.spi.connector.TableNotFoundException; @@ -35,6 +37,7 @@ import static com.google.common.collect.ImmutableList.toImmutableList; import static io.trino.plugin.kinesis.KinesisCompressionCodec.UNCOMPRESSED; +import static io.trino.spi.StandardErrorCode.NOT_SUPPORTED; import static java.util.Objects.requireNonNull; public class KinesisMetadata @@ -64,8 +67,12 @@ public List listSchemaNames(ConnectorSession session) } @Override - public KinesisTableHandle getTableHandle(ConnectorSession session, SchemaTableName schemaTableName) + public KinesisTableHandle getTableHandle(ConnectorSession session, SchemaTableName schemaTableName, Optional startVersion, Optional endVersion) { + if (startVersion.isPresent() || endVersion.isPresent()) { + throw new TrinoException(NOT_SUPPORTED, "This connector does not support versioned tables"); + } + KinesisStreamDescription table = tableDescriptionSupplier.get().get(schemaTableName); if (table == null) { throw new TableNotFoundException(schemaTableName); diff --git a/plugin/trino-kinesis/src/test/java/io/trino/plugin/kinesis/TestKinesisTableDescriptionSupplier.java b/plugin/trino-kinesis/src/test/java/io/trino/plugin/kinesis/TestKinesisTableDescriptionSupplier.java index 36401fc1634b..a001f6fabdc7 100644 --- a/plugin/trino-kinesis/src/test/java/io/trino/plugin/kinesis/TestKinesisTableDescriptionSupplier.java +++ b/plugin/trino-kinesis/src/test/java/io/trino/plugin/kinesis/TestKinesisTableDescriptionSupplier.java @@ -30,6 +30,7 @@ import java.util.List; import java.util.Map; +import java.util.Optional; import static io.trino.testing.TestingConnectorSession.SESSION; import static org.assertj.core.api.Assertions.assertThat; @@ -68,7 +69,7 @@ public void testTableDefinition() { KinesisMetadata metadata = (KinesisMetadata) connector.getMetadata(SESSION, new ConnectorTransactionHandle() {}); SchemaTableName tblName = new SchemaTableName("prod", "test_table"); - KinesisTableHandle tableHandle = metadata.getTableHandle(SESSION, tblName); + KinesisTableHandle tableHandle = metadata.getTableHandle(SESSION, tblName, Optional.empty(), Optional.empty()); assertThat(metadata).isNotNull(); SchemaTableName tableSchemaName = tableHandle.schemaTableName(); assertThat(tableSchemaName.getSchemaName()).isEqualTo("prod"); @@ -91,7 +92,7 @@ public void testRelatedObjects() assertThat(schemas.size()).isEqualTo(1); assertThat(schemas.get(0)).isEqualTo("prod"); - KinesisTableHandle tblHandle = metadata.getTableHandle(null, tblName); + KinesisTableHandle tblHandle = metadata.getTableHandle(null, tblName, Optional.empty(), Optional.empty()); assertThat(tblHandle).isNotNull(); assertThat(tblHandle.schemaName()).isEqualTo("prod"); assertThat(tblHandle.tableName()).isEqualTo("test_table"); diff --git a/plugin/trino-kinesis/src/test/java/io/trino/plugin/kinesis/s3config/TestS3TableConfigClient.java b/plugin/trino-kinesis/src/test/java/io/trino/plugin/kinesis/s3config/TestS3TableConfigClient.java index 2d8aa97e2641..3e7381824435 100644 --- a/plugin/trino-kinesis/src/test/java/io/trino/plugin/kinesis/s3config/TestS3TableConfigClient.java +++ b/plugin/trino-kinesis/src/test/java/io/trino/plugin/kinesis/s3config/TestS3TableConfigClient.java @@ -29,6 +29,7 @@ import org.junit.jupiter.api.parallel.Execution; import java.util.Map; +import java.util.Optional; import static io.trino.testing.TestingConnectorSession.SESSION; import static org.assertj.core.api.Assertions.assertThat; @@ -119,7 +120,7 @@ public void testTableReading() KinesisMetadata metadata = (KinesisMetadata) kinesisConnector.getMetadata(SESSION, new ConnectorTransactionHandle() {}); SchemaTableName tblName = new SchemaTableName("default", "test123"); - KinesisTableHandle tableHandle = metadata.getTableHandle(SESSION, tblName); + KinesisTableHandle tableHandle = metadata.getTableHandle(SESSION, tblName, Optional.empty(), Optional.empty()); assertThat(metadata).isNotNull(); SchemaTableName tableSchemaName = tableHandle.schemaTableName(); assertThat(tableSchemaName.getSchemaName()).isEqualTo("default"); diff --git a/plugin/trino-kudu/src/main/java/io/trino/plugin/kudu/KuduMetadata.java b/plugin/trino-kudu/src/main/java/io/trino/plugin/kudu/KuduMetadata.java index ee1cfbb0fa67..cd24abeeebee 100755 --- a/plugin/trino-kudu/src/main/java/io/trino/plugin/kudu/KuduMetadata.java +++ b/plugin/trino-kudu/src/main/java/io/trino/plugin/kudu/KuduMetadata.java @@ -33,6 +33,7 @@ import io.trino.spi.connector.ConnectorTableLayout; import io.trino.spi.connector.ConnectorTableMetadata; import io.trino.spi.connector.ConnectorTableProperties; +import io.trino.spi.connector.ConnectorTableVersion; import io.trino.spi.connector.Constraint; import io.trino.spi.connector.ConstraintApplicationResult; import io.trino.spi.connector.LimitApplicationResult; @@ -111,7 +112,7 @@ public Map> listTableColumns(ConnectorSess ImmutableMap.Builder> columns = ImmutableMap.builder(); for (SchemaTableName tableName : tables) { - KuduTableHandle tableHandle = getTableHandle(session, tableName); + KuduTableHandle tableHandle = getTableHandle(session, tableName, Optional.empty(), Optional.empty()); if (tableHandle != null) { KuduTable table = tableHandle.getTable(clientSession); columns.put(tableName, getColumnsMetadata(table.getSchema())); @@ -213,8 +214,12 @@ public ColumnMetadata getColumnMetadata(ConnectorSession session, ConnectorTable } @Override - public KuduTableHandle getTableHandle(ConnectorSession session, SchemaTableName schemaTableName) + public KuduTableHandle getTableHandle(ConnectorSession session, SchemaTableName schemaTableName, Optional startVersion, Optional endVersion) { + if (startVersion.isPresent() || endVersion.isPresent()) { + throw new TrinoException(NOT_SUPPORTED, "This connector does not support versioned tables"); + } + try { KuduTable table = clientSession.openTable(schemaTableName); OptionalInt bucketCount = OptionalInt.empty(); diff --git a/plugin/trino-local-file/src/main/java/io/trino/plugin/localfile/LocalFileMetadata.java b/plugin/trino-local-file/src/main/java/io/trino/plugin/localfile/LocalFileMetadata.java index 9d0a03f9f8ef..239908baf5aa 100644 --- a/plugin/trino-local-file/src/main/java/io/trino/plugin/localfile/LocalFileMetadata.java +++ b/plugin/trino-local-file/src/main/java/io/trino/plugin/localfile/LocalFileMetadata.java @@ -16,12 +16,14 @@ import com.google.common.collect.ImmutableList; import com.google.common.collect.ImmutableMap; import com.google.inject.Inject; +import io.trino.spi.TrinoException; import io.trino.spi.connector.ColumnHandle; import io.trino.spi.connector.ColumnMetadata; import io.trino.spi.connector.ConnectorMetadata; import io.trino.spi.connector.ConnectorSession; import io.trino.spi.connector.ConnectorTableHandle; import io.trino.spi.connector.ConnectorTableMetadata; +import io.trino.spi.connector.ConnectorTableVersion; import io.trino.spi.connector.Constraint; import io.trino.spi.connector.ConstraintApplicationResult; import io.trino.spi.connector.SchemaTableName; @@ -34,6 +36,7 @@ import static io.trino.plugin.localfile.LocalFileColumnHandle.SERVER_ADDRESS_COLUMN_NAME; import static io.trino.plugin.localfile.LocalFileColumnHandle.SERVER_ADDRESS_ORDINAL_POSITION; +import static io.trino.spi.StandardErrorCode.NOT_SUPPORTED; import static io.trino.spi.type.VarcharType.createUnboundedVarcharType; import static java.util.Objects.requireNonNull; @@ -59,8 +62,12 @@ public List listSchemaNames(ConnectorSession session) } @Override - public ConnectorTableHandle getTableHandle(ConnectorSession session, SchemaTableName tableName) + public ConnectorTableHandle getTableHandle(ConnectorSession session, SchemaTableName tableName, Optional startVersion, Optional endVersion) { + if (startVersion.isPresent() || endVersion.isPresent()) { + throw new TrinoException(NOT_SUPPORTED, "This connector does not support versioned tables"); + } + requireNonNull(tableName, "tableName is null"); return localFileTables.getTable(tableName); } diff --git a/plugin/trino-memory/src/main/java/io/trino/plugin/memory/MemoryMetadata.java b/plugin/trino-memory/src/main/java/io/trino/plugin/memory/MemoryMetadata.java index ba943bf03d37..f05eceda1c89 100644 --- a/plugin/trino-memory/src/main/java/io/trino/plugin/memory/MemoryMetadata.java +++ b/plugin/trino-memory/src/main/java/io/trino/plugin/memory/MemoryMetadata.java @@ -160,24 +160,18 @@ private boolean isSchemaEmpty(String schemaName) views.keySet().stream().noneMatch(view -> view.getSchemaName().equals(schemaName)); } - @Override - public ConnectorTableHandle getTableHandle(ConnectorSession session, SchemaTableName schemaTableName) - { - throw new UnsupportedOperationException("This method is not supported because getTableHandle with versions is implemented instead"); - } - @Override public synchronized ConnectorTableHandle getTableHandle(ConnectorSession session, SchemaTableName schemaTableName, Optional startVersion, Optional endVersion) { + if (startVersion.isPresent() || endVersion.isPresent()) { + throw new TrinoException(NOT_SUPPORTED, "This connector does not support versioned tables"); + } + Long id = tableIds.get(schemaTableName); if (id == null) { return null; } - if (startVersion.isPresent() || endVersion.isPresent()) { - throw new TrinoException(NOT_SUPPORTED, "This connector does not support versioned tables"); - } - return new MemoryTableHandle(id, OptionalLong.empty(), OptionalDouble.empty()); } diff --git a/plugin/trino-mongodb/src/main/java/io/trino/plugin/mongodb/MongoMetadata.java b/plugin/trino-mongodb/src/main/java/io/trino/plugin/mongodb/MongoMetadata.java index 4caa77d70543..e023e61e7211 100644 --- a/plugin/trino-mongodb/src/main/java/io/trino/plugin/mongodb/MongoMetadata.java +++ b/plugin/trino-mongodb/src/main/java/io/trino/plugin/mongodb/MongoMetadata.java @@ -169,6 +169,7 @@ public MongoTableHandle getTableHandle( if (startVersion.isPresent() || endVersion.isPresent()) { throw new TrinoException(NOT_SUPPORTED, "This connector does not support versioned tables"); } + requireNonNull(tableName, "tableName is null"); try { return mongoSession.getTable(tableName).tableHandle(); diff --git a/plugin/trino-mongodb/src/test/java/io/trino/plugin/mongodb/TestMongoConnectorTest.java b/plugin/trino-mongodb/src/test/java/io/trino/plugin/mongodb/TestMongoConnectorTest.java index d8da7d5c4772..954b31077fa8 100644 --- a/plugin/trino-mongodb/src/test/java/io/trino/plugin/mongodb/TestMongoConnectorTest.java +++ b/plugin/trino-mongodb/src/test/java/io/trino/plugin/mongodb/TestMongoConnectorTest.java @@ -147,17 +147,6 @@ public void testSortItemsReflectedInExplain() "TopNPartial\\[count = 5, orderBy = \\[nationkey DESC"); } - @Test - @Override // Override because the failure message is different - public void testSelectVersionOfNonExistentTable() - { - String tableName = "foo_" + randomNameSuffix(); - assertThat(query("SELECT * FROM " + tableName + " FOR TIMESTAMP AS OF TIMESTAMP '2021-03-01 00:00:01'")) - .failure().hasMessage("This connector does not support versioned tables"); - assertThat(query("SELECT * FROM " + tableName + " FOR VERSION AS OF 'version1'")) - .failure().hasMessage("This connector does not support versioned tables"); - } - @Override protected Optional filterDataMappingSmokeTestData(DataMappingTestSetup dataMappingTestSetup) { diff --git a/plugin/trino-opensearch/src/main/java/io/trino/plugin/opensearch/OpenSearchMetadata.java b/plugin/trino-opensearch/src/main/java/io/trino/plugin/opensearch/OpenSearchMetadata.java index ab54663df4a8..b79091d3a7fe 100644 --- a/plugin/trino-opensearch/src/main/java/io/trino/plugin/opensearch/OpenSearchMetadata.java +++ b/plugin/trino-opensearch/src/main/java/io/trino/plugin/opensearch/OpenSearchMetadata.java @@ -48,6 +48,7 @@ import io.trino.spi.connector.ConnectorTableHandle; import io.trino.spi.connector.ConnectorTableMetadata; import io.trino.spi.connector.ConnectorTableProperties; +import io.trino.spi.connector.ConnectorTableVersion; import io.trino.spi.connector.Constraint; import io.trino.spi.connector.ConstraintApplicationResult; import io.trino.spi.connector.LimitApplicationResult; @@ -90,6 +91,7 @@ import static io.airlift.slice.SliceUtf8.getCodePointAt; import static io.airlift.slice.SliceUtf8.lengthOfCodePoint; import static io.trino.spi.StandardErrorCode.INVALID_FUNCTION_ARGUMENT; +import static io.trino.spi.StandardErrorCode.NOT_SUPPORTED; import static io.trino.spi.expression.StandardFunctions.LIKE_FUNCTION_NAME; import static io.trino.spi.type.BigintType.BIGINT; import static io.trino.spi.type.BooleanType.BOOLEAN; @@ -148,10 +150,13 @@ public List listSchemaNames(ConnectorSession session) return ImmutableList.of(schemaName); } - @SuppressWarnings("deprecation") // TODO Implement getTableHandle(ConnectorSession, SchemaTableName, Optional, Optional) method @Override - public OpenSearchTableHandle getTableHandle(ConnectorSession session, SchemaTableName tableName) + public OpenSearchTableHandle getTableHandle(ConnectorSession session, SchemaTableName tableName, Optional startVersion, Optional endVersion) { + if (startVersion.isPresent() || endVersion.isPresent()) { + throw new TrinoException(NOT_SUPPORTED, "This connector does not support versioned tables"); + } + requireNonNull(tableName, "tableName is null"); if (tableName.getSchemaName().equals(schemaName)) { diff --git a/plugin/trino-phoenix5/src/main/java/io/trino/plugin/phoenix5/PhoenixMetadata.java b/plugin/trino-phoenix5/src/main/java/io/trino/plugin/phoenix5/PhoenixMetadata.java index cb86b7fc8a6d..7baab164ddd2 100644 --- a/plugin/trino-phoenix5/src/main/java/io/trino/plugin/phoenix5/PhoenixMetadata.java +++ b/plugin/trino-phoenix5/src/main/java/io/trino/plugin/phoenix5/PhoenixMetadata.java @@ -40,6 +40,7 @@ import io.trino.spi.connector.ConnectorTableMetadata; import io.trino.spi.connector.ConnectorTableProperties; import io.trino.spi.connector.ConnectorTableSchema; +import io.trino.spi.connector.ConnectorTableVersion; import io.trino.spi.connector.LocalProperty; import io.trino.spi.connector.RetryMode; import io.trino.spi.connector.SchemaTableName; @@ -97,8 +98,12 @@ public PhoenixMetadata( } @Override - public JdbcTableHandle getTableHandle(ConnectorSession session, SchemaTableName schemaTableName) + public JdbcTableHandle getTableHandle(ConnectorSession session, SchemaTableName schemaTableName, Optional startVersion, Optional endVersion) { + if (startVersion.isPresent() || endVersion.isPresent()) { + throw new TrinoException(NOT_SUPPORTED, "This connector does not support versioned tables"); + } + return phoenixClient.getTableHandle(session, schemaTableName) .map(JdbcTableHandle::asPlainTable) .map(JdbcNamedRelationHandle::getRemoteTableName) diff --git a/plugin/trino-pinot/src/main/java/io/trino/plugin/pinot/PinotMetadata.java b/plugin/trino-pinot/src/main/java/io/trino/plugin/pinot/PinotMetadata.java index ceb8ad89e37b..a8100d1c7414 100755 --- a/plugin/trino-pinot/src/main/java/io/trino/plugin/pinot/PinotMetadata.java +++ b/plugin/trino-pinot/src/main/java/io/trino/plugin/pinot/PinotMetadata.java @@ -35,6 +35,7 @@ import io.trino.plugin.pinot.query.aggregation.ImplementCountDistinct; import io.trino.plugin.pinot.query.aggregation.ImplementMinMax; import io.trino.plugin.pinot.query.aggregation.ImplementSum; +import io.trino.spi.TrinoException; import io.trino.spi.connector.AggregateFunction; import io.trino.spi.connector.AggregationApplicationResult; import io.trino.spi.connector.Assignment; @@ -44,6 +45,7 @@ import io.trino.spi.connector.ConnectorSession; import io.trino.spi.connector.ConnectorTableHandle; import io.trino.spi.connector.ConnectorTableMetadata; +import io.trino.spi.connector.ConnectorTableVersion; import io.trino.spi.connector.Constraint; import io.trino.spi.connector.ConstraintApplicationResult; import io.trino.spi.connector.LimitApplicationResult; @@ -79,6 +81,7 @@ import static io.trino.plugin.pinot.PinotSessionProperties.isAggregationPushdownEnabled; import static io.trino.plugin.pinot.query.AggregateExpression.replaceIdentifier; import static io.trino.plugin.pinot.query.DynamicTablePqlExtractor.quoteIdentifier; +import static io.trino.spi.StandardErrorCode.NOT_SUPPORTED; import static io.trino.spi.type.BigintType.BIGINT; import static io.trino.spi.type.DoubleType.DOUBLE; import static io.trino.spi.type.IntegerType.INTEGER; @@ -149,8 +152,12 @@ public List listSchemaNames(ConnectorSession session) } @Override - public PinotTableHandle getTableHandle(ConnectorSession session, SchemaTableName tableName) + public PinotTableHandle getTableHandle(ConnectorSession session, SchemaTableName tableName, Optional startVersion, Optional endVersion) { + if (startVersion.isPresent() || endVersion.isPresent()) { + throw new TrinoException(NOT_SUPPORTED, "This connector does not support versioned tables"); + } + if (tableName.getTableName().trim().contains("select ")) { DynamicTable dynamicTable = DynamicTableBuilder.buildFromPql(this, tableName, pinotClient, typeConverter); return new PinotTableHandle(tableName.getSchemaName(), dynamicTable.tableName(), TupleDomain.all(), OptionalLong.empty(), Optional.of(dynamicTable)); diff --git a/plugin/trino-pinot/src/test/java/io/trino/plugin/pinot/TestPinotMetadata.java b/plugin/trino-pinot/src/test/java/io/trino/plugin/pinot/TestPinotMetadata.java index cba44c329ce0..bad57262a4b5 100755 --- a/plugin/trino-pinot/src/test/java/io/trino/plugin/pinot/TestPinotMetadata.java +++ b/plugin/trino-pinot/src/test/java/io/trino/plugin/pinot/TestPinotMetadata.java @@ -44,11 +44,23 @@ public void testTables() .build()); List schemas = metadata.listSchemaNames(session); assertThat(ImmutableList.copyOf(schemas)).isEqualTo(ImmutableList.of("default")); - PinotTableHandle withWeirdSchema = metadata.getTableHandle(session, new SchemaTableName("foo", TestPinotSplitManager.realtimeOnlyTable.getTableName())); + PinotTableHandle withWeirdSchema = metadata.getTableHandle( + session, + new SchemaTableName("foo", TestPinotSplitManager.realtimeOnlyTable.getTableName()), + Optional.empty(), + Optional.empty()); assertThat(withWeirdSchema.getTableName()).isEqualTo(TestPinotSplitManager.realtimeOnlyTable.getTableName()); - PinotTableHandle withAnotherSchema = metadata.getTableHandle(session, new SchemaTableName(TestPinotSplitManager.realtimeOnlyTable.getTableName(), TestPinotSplitManager.realtimeOnlyTable.getTableName())); + PinotTableHandle withAnotherSchema = metadata.getTableHandle( + session, + new SchemaTableName(TestPinotSplitManager.realtimeOnlyTable.getTableName(), TestPinotSplitManager.realtimeOnlyTable.getTableName()), + Optional.empty(), + Optional.empty()); assertThat(withAnotherSchema.getTableName()).isEqualTo(TestPinotSplitManager.realtimeOnlyTable.getTableName()); - PinotTableHandle withUppercaseTable = metadata.getTableHandle(session, new SchemaTableName("default", TEST_TABLE)); + PinotTableHandle withUppercaseTable = metadata.getTableHandle( + session, + new SchemaTableName("default", TEST_TABLE), + Optional.empty(), + Optional.empty()); assertThat(withUppercaseTable.getTableName()).isEqualTo("airlineStats"); } } diff --git a/plugin/trino-prometheus/src/main/java/io/trino/plugin/prometheus/PrometheusMetadata.java b/plugin/trino-prometheus/src/main/java/io/trino/plugin/prometheus/PrometheusMetadata.java index c4a93cf1ede4..d4ca19d72e66 100644 --- a/plugin/trino-prometheus/src/main/java/io/trino/plugin/prometheus/PrometheusMetadata.java +++ b/plugin/trino-prometheus/src/main/java/io/trino/plugin/prometheus/PrometheusMetadata.java @@ -17,12 +17,14 @@ import com.google.common.collect.ImmutableMap; import com.google.common.collect.ImmutableSet; import com.google.inject.Inject; +import io.trino.spi.TrinoException; import io.trino.spi.connector.ColumnHandle; import io.trino.spi.connector.ColumnMetadata; import io.trino.spi.connector.ConnectorMetadata; import io.trino.spi.connector.ConnectorSession; import io.trino.spi.connector.ConnectorTableHandle; import io.trino.spi.connector.ConnectorTableMetadata; +import io.trino.spi.connector.ConnectorTableVersion; import io.trino.spi.connector.Constraint; import io.trino.spi.connector.ConstraintApplicationResult; import io.trino.spi.connector.SchemaTableName; @@ -35,6 +37,7 @@ import java.util.Set; import static com.google.common.collect.ImmutableList.toImmutableList; +import static io.trino.spi.StandardErrorCode.NOT_SUPPORTED; import static java.util.Objects.requireNonNull; public class PrometheusMetadata @@ -60,8 +63,12 @@ private static List listSchemaNames() } @Override - public PrometheusTableHandle getTableHandle(ConnectorSession session, SchemaTableName tableName) + public PrometheusTableHandle getTableHandle(ConnectorSession session, SchemaTableName tableName, Optional startVersion, Optional endVersion) { + if (startVersion.isPresent() || endVersion.isPresent()) { + throw new TrinoException(NOT_SUPPORTED, "This connector does not support versioned tables"); + } + if (!listSchemaNames(session).contains(tableName.getSchemaName())) { return null; } diff --git a/plugin/trino-raptor-legacy/src/main/java/io/trino/plugin/raptor/legacy/RaptorMetadata.java b/plugin/trino-raptor-legacy/src/main/java/io/trino/plugin/raptor/legacy/RaptorMetadata.java index f745d4cc0e29..4d3c1843997a 100644 --- a/plugin/trino-raptor-legacy/src/main/java/io/trino/plugin/raptor/legacy/RaptorMetadata.java +++ b/plugin/trino-raptor-legacy/src/main/java/io/trino/plugin/raptor/legacy/RaptorMetadata.java @@ -49,6 +49,7 @@ import io.trino.spi.connector.ConnectorTableMetadata; import io.trino.spi.connector.ConnectorTablePartitioning; import io.trino.spi.connector.ConnectorTableProperties; +import io.trino.spi.connector.ConnectorTableVersion; import io.trino.spi.connector.ConnectorViewDefinition; import io.trino.spi.connector.Constraint; import io.trino.spi.connector.ConstraintApplicationResult; @@ -168,8 +169,12 @@ public List listSchemaNames(ConnectorSession session) } @Override - public ConnectorTableHandle getTableHandle(ConnectorSession session, SchemaTableName tableName) + public ConnectorTableHandle getTableHandle(ConnectorSession session, SchemaTableName tableName, Optional startVersion, Optional endVersion) { + if (startVersion.isPresent() || endVersion.isPresent()) { + throw new TrinoException(NOT_SUPPORTED, "This connector does not support versioned tables"); + } + return getTableHandle(tableName); } diff --git a/plugin/trino-raptor-legacy/src/test/java/io/trino/plugin/raptor/legacy/TestRaptorConnector.java b/plugin/trino-raptor-legacy/src/test/java/io/trino/plugin/raptor/legacy/TestRaptorConnector.java index 7d8e5eb5cd2c..ca14a782cbdf 100644 --- a/plugin/trino-raptor-legacy/src/test/java/io/trino/plugin/raptor/legacy/TestRaptorConnector.java +++ b/plugin/trino-raptor-legacy/src/test/java/io/trino/plugin/raptor/legacy/TestRaptorConnector.java @@ -284,6 +284,6 @@ private ConnectorTransactionHandle beginTransaction() private static ConnectorTableHandle getTableHandle(ConnectorMetadata metadata, String name) { - return metadata.getTableHandle(SESSION, new SchemaTableName("test", name)); + return metadata.getTableHandle(SESSION, new SchemaTableName("test", name), Optional.empty(), Optional.empty()); } } diff --git a/plugin/trino-raptor-legacy/src/test/java/io/trino/plugin/raptor/legacy/metadata/TestRaptorMetadata.java b/plugin/trino-raptor-legacy/src/test/java/io/trino/plugin/raptor/legacy/metadata/TestRaptorMetadata.java index cb29121e8916..227bb7a238ba 100644 --- a/plugin/trino-raptor-legacy/src/test/java/io/trino/plugin/raptor/legacy/metadata/TestRaptorMetadata.java +++ b/plugin/trino-raptor-legacy/src/test/java/io/trino/plugin/raptor/legacy/metadata/TestRaptorMetadata.java @@ -116,9 +116,9 @@ public void cleanupDatabase() @Test public void testRenameColumn() { - assertThat(metadata.getTableHandle(SESSION, DEFAULT_TEST_ORDERS)).isNull(); + assertThat(metadata.getTableHandle(SESSION, DEFAULT_TEST_ORDERS, Optional.empty(), Optional.empty())).isNull(); metadata.createTable(SESSION, getOrdersTable(), false); - ConnectorTableHandle tableHandle = metadata.getTableHandle(SESSION, DEFAULT_TEST_ORDERS); + ConnectorTableHandle tableHandle = metadata.getTableHandle(SESSION, DEFAULT_TEST_ORDERS, Optional.empty(), Optional.empty()); assertInstanceOf(tableHandle, RaptorTableHandle.class); RaptorTableHandle raptorTableHandle = (RaptorTableHandle) tableHandle; @@ -133,12 +133,12 @@ public void testRenameColumn() @Test public void testAddColumn() { - assertThat(metadata.getTableHandle(SESSION, DEFAULT_TEST_ORDERS)).isNull(); + assertThat(metadata.getTableHandle(SESSION, DEFAULT_TEST_ORDERS, Optional.empty(), Optional.empty())).isNull(); metadata.createTable(SESSION, buildTable(ImmutableMap.of(), tableMetadataBuilder(DEFAULT_TEST_ORDERS) .column("orderkey", BIGINT) .column("price", BIGINT)), false); - ConnectorTableHandle tableHandle = metadata.getTableHandle(SESSION, DEFAULT_TEST_ORDERS); + ConnectorTableHandle tableHandle = metadata.getTableHandle(SESSION, DEFAULT_TEST_ORDERS, Optional.empty(), Optional.empty()); assertInstanceOf(tableHandle, RaptorTableHandle.class); RaptorTableHandle raptorTableHandle = (RaptorTableHandle) tableHandle; @@ -150,12 +150,12 @@ public void testAddColumn() @Test public void testDropColumn() { - assertThat(metadata.getTableHandle(SESSION, DEFAULT_TEST_ORDERS)).isNull(); + assertThat(metadata.getTableHandle(SESSION, DEFAULT_TEST_ORDERS, Optional.empty(), Optional.empty())).isNull(); metadata.createTable(SESSION, buildTable(ImmutableMap.of(), tableMetadataBuilder(DEFAULT_TEST_ORDERS) .column("orderkey", BIGINT) .column("price", BIGINT)), false); - ConnectorTableHandle tableHandle = metadata.getTableHandle(SESSION, DEFAULT_TEST_ORDERS); + ConnectorTableHandle tableHandle = metadata.getTableHandle(SESSION, DEFAULT_TEST_ORDERS, Optional.empty(), Optional.empty()); assertInstanceOf(tableHandle, RaptorTableHandle.class); RaptorTableHandle raptorTableHandle = (RaptorTableHandle) tableHandle; @@ -168,12 +168,12 @@ public void testDropColumn() @Test public void testAddColumnAfterDropColumn() { - assertThat(metadata.getTableHandle(SESSION, DEFAULT_TEST_ORDERS)).isNull(); + assertThat(metadata.getTableHandle(SESSION, DEFAULT_TEST_ORDERS, Optional.empty(), Optional.empty())).isNull(); metadata.createTable(SESSION, buildTable(ImmutableMap.of(), tableMetadataBuilder(DEFAULT_TEST_ORDERS) .column("orderkey", BIGINT) .column("price", BIGINT)), false); - ConnectorTableHandle tableHandle = metadata.getTableHandle(SESSION, DEFAULT_TEST_ORDERS); + ConnectorTableHandle tableHandle = metadata.getTableHandle(SESSION, DEFAULT_TEST_ORDERS, Optional.empty(), Optional.empty()); assertInstanceOf(tableHandle, RaptorTableHandle.class); RaptorTableHandle raptorTableHandle = (RaptorTableHandle) tableHandle; @@ -188,7 +188,7 @@ public void testAddColumnAfterDropColumn() @Test public void testDropColumnDisallowed() { - assertThat(metadata.getTableHandle(SESSION, DEFAULT_TEST_ORDERS)).isNull(); + assertThat(metadata.getTableHandle(SESSION, DEFAULT_TEST_ORDERS, Optional.empty(), Optional.empty())).isNull(); Map properties = ImmutableMap.of( BUCKET_COUNT_PROPERTY, 16, BUCKETED_ON_PROPERTY, ImmutableList.of("orderkey"), @@ -201,7 +201,7 @@ public void testDropColumnDisallowed() .column("highestid", BIGINT)); metadata.createTable(SESSION, ordersTable, false); - ConnectorTableHandle ordersTableHandle = metadata.getTableHandle(SESSION, DEFAULT_TEST_ORDERS); + ConnectorTableHandle ordersTableHandle = metadata.getTableHandle(SESSION, DEFAULT_TEST_ORDERS, Optional.empty(), Optional.empty()); assertInstanceOf(ordersTableHandle, RaptorTableHandle.class); RaptorTableHandle ordersRaptorTableHandle = (RaptorTableHandle) ordersTableHandle; assertThat(ordersRaptorTableHandle.getTableId()).isEqualTo(1); @@ -233,17 +233,17 @@ public void testDropColumnDisallowed() @Test public void testRenameTable() { - assertThat(metadata.getTableHandle(SESSION, DEFAULT_TEST_ORDERS)).isNull(); + assertThat(metadata.getTableHandle(SESSION, DEFAULT_TEST_ORDERS, Optional.empty(), Optional.empty())).isNull(); metadata.createTable(SESSION, getOrdersTable(), false); - ConnectorTableHandle tableHandle = metadata.getTableHandle(SESSION, DEFAULT_TEST_ORDERS); + ConnectorTableHandle tableHandle = metadata.getTableHandle(SESSION, DEFAULT_TEST_ORDERS, Optional.empty(), Optional.empty()); assertInstanceOf(tableHandle, RaptorTableHandle.class); RaptorTableHandle raptorTableHandle = (RaptorTableHandle) tableHandle; SchemaTableName renamedTable = new SchemaTableName(raptorTableHandle.getSchemaName(), "orders_renamed"); metadata.renameTable(SESSION, raptorTableHandle, renamedTable); - assertThat(metadata.getTableHandle(SESSION, DEFAULT_TEST_ORDERS)).isNull(); - ConnectorTableHandle renamedTableHandle = metadata.getTableHandle(SESSION, renamedTable); + assertThat(metadata.getTableHandle(SESSION, DEFAULT_TEST_ORDERS, Optional.empty(), Optional.empty())).isNull(); + ConnectorTableHandle renamedTableHandle = metadata.getTableHandle(SESSION, renamedTable, Optional.empty(), Optional.empty()); assertThat(renamedTableHandle).isNotNull(); assertThat(((RaptorTableHandle) renamedTableHandle).getTableName()).isEqualTo(renamedTable.getTableName()); } @@ -251,11 +251,11 @@ public void testRenameTable() @Test public void testCreateTable() { - assertThat(metadata.getTableHandle(SESSION, DEFAULT_TEST_ORDERS)).isNull(); + assertThat(metadata.getTableHandle(SESSION, DEFAULT_TEST_ORDERS, Optional.empty(), Optional.empty())).isNull(); metadata.createTable(SESSION, getOrdersTable(), false); - ConnectorTableHandle tableHandle = metadata.getTableHandle(SESSION, DEFAULT_TEST_ORDERS); + ConnectorTableHandle tableHandle = metadata.getTableHandle(SESSION, DEFAULT_TEST_ORDERS, Optional.empty(), Optional.empty()); assertInstanceOf(tableHandle, RaptorTableHandle.class); assertThat(((RaptorTableHandle) tableHandle).getTableId()).isEqualTo(1); @@ -275,14 +275,14 @@ public void testCreateTable() @Test public void testTableProperties() { - assertThat(metadata.getTableHandle(SESSION, DEFAULT_TEST_ORDERS)).isNull(); + assertThat(metadata.getTableHandle(SESSION, DEFAULT_TEST_ORDERS, Optional.empty(), Optional.empty())).isNull(); ConnectorTableMetadata ordersTable = getOrdersTable(ImmutableMap.of( ORDERING_PROPERTY, ImmutableList.of("orderdate", "custkey"), TEMPORAL_COLUMN_PROPERTY, "orderdate")); metadata.createTable(SESSION, ordersTable, false); - ConnectorTableHandle tableHandle = metadata.getTableHandle(SESSION, DEFAULT_TEST_ORDERS); + ConnectorTableHandle tableHandle = metadata.getTableHandle(SESSION, DEFAULT_TEST_ORDERS, Optional.empty(), Optional.empty()); assertInstanceOf(tableHandle, RaptorTableHandle.class); RaptorTableHandle raptorTableHandle = (RaptorTableHandle) tableHandle; assertThat(raptorTableHandle.getTableId()).isEqualTo(1); @@ -308,14 +308,14 @@ public void testTableProperties() @Test public void testTablePropertiesWithOrganization() { - assertThat(metadata.getTableHandle(SESSION, DEFAULT_TEST_ORDERS)).isNull(); + assertThat(metadata.getTableHandle(SESSION, DEFAULT_TEST_ORDERS, Optional.empty(), Optional.empty())).isNull(); ConnectorTableMetadata ordersTable = getOrdersTable(ImmutableMap.of( ORDERING_PROPERTY, ImmutableList.of("orderdate", "custkey"), ORGANIZED_PROPERTY, true)); metadata.createTable(SESSION, ordersTable, false); - ConnectorTableHandle tableHandle = metadata.getTableHandle(SESSION, DEFAULT_TEST_ORDERS); + ConnectorTableHandle tableHandle = metadata.getTableHandle(SESSION, DEFAULT_TEST_ORDERS, Optional.empty(), Optional.empty()); assertInstanceOf(tableHandle, RaptorTableHandle.class); RaptorTableHandle raptorTableHandle = (RaptorTableHandle) tableHandle; assertThat(raptorTableHandle.getTableId()).isEqualTo(1); @@ -338,14 +338,14 @@ public void testTablePropertiesWithOrganization() @Test public void testCreateBucketedTable() { - assertThat(metadata.getTableHandle(SESSION, DEFAULT_TEST_ORDERS)).isNull(); + assertThat(metadata.getTableHandle(SESSION, DEFAULT_TEST_ORDERS, Optional.empty(), Optional.empty())).isNull(); ConnectorTableMetadata ordersTable = getOrdersTable(ImmutableMap.of( BUCKET_COUNT_PROPERTY, 16, BUCKETED_ON_PROPERTY, ImmutableList.of("custkey", "orderkey"))); metadata.createTable(SESSION, ordersTable, false); - ConnectorTableHandle tableHandle = metadata.getTableHandle(SESSION, DEFAULT_TEST_ORDERS); + ConnectorTableHandle tableHandle = metadata.getTableHandle(SESSION, DEFAULT_TEST_ORDERS, Optional.empty(), Optional.empty()); assertInstanceOf(tableHandle, RaptorTableHandle.class); RaptorTableHandle raptorTableHandle = (RaptorTableHandle) tableHandle; assertThat(raptorTableHandle.getTableId()).isEqualTo(1); @@ -365,7 +365,7 @@ public void testCreateBucketedTable() // create a new table and verify it has a different distribution metadata.createTable(SESSION, ordersTable, false); - tableId = ((RaptorTableHandle) metadata.getTableHandle(SESSION, DEFAULT_TEST_ORDERS)).getTableId(); + tableId = ((RaptorTableHandle) metadata.getTableHandle(SESSION, DEFAULT_TEST_ORDERS, Optional.empty(), Optional.empty())).getTableId(); assertThat(tableId).isEqualTo(2); assertThat(getTableDistributionId(tableId)).isEqualTo(Long.valueOf(2)); } @@ -373,7 +373,7 @@ public void testCreateBucketedTable() @Test public void testCreateBucketedTableAsSelect() { - assertThat(metadata.getTableHandle(SESSION, DEFAULT_TEST_ORDERS)).isNull(); + assertThat(metadata.getTableHandle(SESSION, DEFAULT_TEST_ORDERS, Optional.empty(), Optional.empty())).isNull(); ConnectorTableMetadata ordersTable = getOrdersTable(ImmutableMap.of( BUCKET_COUNT_PROPERTY, 32, @@ -389,7 +389,7 @@ public void testCreateBucketedTableAsSelect() ConnectorOutputTableHandle outputHandle = metadata.beginCreateTable(SESSION, ordersTable, Optional.of(layout), NO_RETRIES); metadata.finishCreateTable(SESSION, outputHandle, ImmutableList.of(), ImmutableList.of()); - ConnectorTableHandle tableHandle = metadata.getTableHandle(SESSION, DEFAULT_TEST_ORDERS); + ConnectorTableHandle tableHandle = metadata.getTableHandle(SESSION, DEFAULT_TEST_ORDERS, Optional.empty(), Optional.empty()); assertInstanceOf(tableHandle, RaptorTableHandle.class); RaptorTableHandle raptorTableHandle = (RaptorTableHandle) tableHandle; assertThat(raptorTableHandle.getTableId()).isEqualTo(1); @@ -414,7 +414,7 @@ public void testCreateBucketedTableExistingDistribution() MetadataDao metadataDao = dbi.onDemand(MetadataDao.class); // create orders table - assertThat(metadata.getTableHandle(SESSION, DEFAULT_TEST_ORDERS)).isNull(); + assertThat(metadata.getTableHandle(SESSION, DEFAULT_TEST_ORDERS, Optional.empty(), Optional.empty())).isNull(); ConnectorTableMetadata table = getOrdersTable(ImmutableMap.of( BUCKET_COUNT_PROPERTY, 16, @@ -422,7 +422,7 @@ public void testCreateBucketedTableExistingDistribution() DISTRIBUTION_NAME_PROPERTY, "orders")); metadata.createTable(SESSION, table, false); - ConnectorTableHandle tableHandle = metadata.getTableHandle(SESSION, DEFAULT_TEST_ORDERS); + ConnectorTableHandle tableHandle = metadata.getTableHandle(SESSION, DEFAULT_TEST_ORDERS, Optional.empty(), Optional.empty()); assertInstanceOf(tableHandle, RaptorTableHandle.class); RaptorTableHandle raptorTableHandle = (RaptorTableHandle) tableHandle; @@ -437,7 +437,7 @@ public void testCreateBucketedTableExistingDistribution() assertThat(getTableDistributionId(tableId)).isEqualTo(Long.valueOf(1)); // create lineitems table - assertThat(metadata.getTableHandle(SESSION, DEFAULT_TEST_LINEITEMS)).isNull(); + assertThat(metadata.getTableHandle(SESSION, DEFAULT_TEST_LINEITEMS, Optional.empty(), Optional.empty())).isNull(); table = getLineItemsTable(ImmutableMap.of( BUCKET_COUNT_PROPERTY, 16, @@ -445,7 +445,7 @@ public void testCreateBucketedTableExistingDistribution() DISTRIBUTION_NAME_PROPERTY, "orders")); metadata.createTable(SESSION, table, false); - tableHandle = metadata.getTableHandle(SESSION, DEFAULT_TEST_LINEITEMS); + tableHandle = metadata.getTableHandle(SESSION, DEFAULT_TEST_LINEITEMS, Optional.empty(), Optional.empty()); assertInstanceOf(tableHandle, RaptorTableHandle.class); raptorTableHandle = (RaptorTableHandle) tableHandle; @@ -463,7 +463,7 @@ public void testCreateBucketedTableExistingDistribution() @Test public void testInvalidOrderingColumns() { - assertThat(metadata.getTableHandle(SESSION, DEFAULT_TEST_ORDERS)).isNull(); + assertThat(metadata.getTableHandle(SESSION, DEFAULT_TEST_ORDERS, Optional.empty(), Optional.empty())).isNull(); assertThatThrownBy(() -> metadata.createTable(SESSION, getOrdersTable(ImmutableMap.of(ORDERING_PROPERTY, ImmutableList.of("orderdatefoo"))), false)) .isInstanceOf(TrinoException.class) @@ -473,7 +473,7 @@ public void testInvalidOrderingColumns() @Test public void testInvalidTemporalColumn() { - assertThat(metadata.getTableHandle(SESSION, DEFAULT_TEST_ORDERS)).isNull(); + assertThat(metadata.getTableHandle(SESSION, DEFAULT_TEST_ORDERS, Optional.empty(), Optional.empty())).isNull(); assertThatThrownBy(() -> metadata.createTable(SESSION, getOrdersTable(ImmutableMap.of(TEMPORAL_COLUMN_PROPERTY, "foo")), false)) .isInstanceOf(TrinoException.class) @@ -483,7 +483,7 @@ public void testInvalidTemporalColumn() @Test public void testInvalidTemporalColumnType() { - assertThat(metadata.getTableHandle(SESSION, DEFAULT_TEST_ORDERS)).isNull(); + assertThat(metadata.getTableHandle(SESSION, DEFAULT_TEST_ORDERS, Optional.empty(), Optional.empty())).isNull(); assertThatThrownBy(() -> metadata.createTable(SESSION, getOrdersTable(ImmutableMap.of(TEMPORAL_COLUMN_PROPERTY, "orderkey")), false)) .isInstanceOf(TrinoException.class) .hasMessage("Temporal column must be of type timestamp or date: orderkey"); @@ -492,7 +492,7 @@ public void testInvalidTemporalColumnType() @Test public void testInvalidTemporalOrganization() { - assertThat(metadata.getTableHandle(SESSION, DEFAULT_TEST_ORDERS)).isNull(); + assertThat(metadata.getTableHandle(SESSION, DEFAULT_TEST_ORDERS, Optional.empty(), Optional.empty())).isNull(); assertThatThrownBy(() -> metadata.createTable(SESSION, getOrdersTable(ImmutableMap.of( TEMPORAL_COLUMN_PROPERTY, "orderdate", ORGANIZED_PROPERTY, true)), @@ -504,7 +504,7 @@ public void testInvalidTemporalOrganization() @Test public void testInvalidOrderingOrganization() { - assertThat(metadata.getTableHandle(SESSION, DEFAULT_TEST_ORDERS)).isNull(); + assertThat(metadata.getTableHandle(SESSION, DEFAULT_TEST_ORDERS, Optional.empty(), Optional.empty())).isNull(); assertThatThrownBy(() -> metadata.createTable(SESSION, getOrdersTable(ImmutableMap.of(ORGANIZED_PROPERTY, true)), false)) .isInstanceOf(TrinoException.class) .hasMessage("Table organization requires an ordering"); @@ -513,12 +513,12 @@ public void testInvalidOrderingOrganization() @Test public void testSortOrderProperty() { - assertThat(metadata.getTableHandle(SESSION, DEFAULT_TEST_ORDERS)).isNull(); + assertThat(metadata.getTableHandle(SESSION, DEFAULT_TEST_ORDERS, Optional.empty(), Optional.empty())).isNull(); ConnectorTableMetadata ordersTable = getOrdersTable(ImmutableMap.of(ORDERING_PROPERTY, ImmutableList.of("orderdate", "custkey"))); metadata.createTable(SESSION, ordersTable, false); - ConnectorTableHandle tableHandle = metadata.getTableHandle(SESSION, DEFAULT_TEST_ORDERS); + ConnectorTableHandle tableHandle = metadata.getTableHandle(SESSION, DEFAULT_TEST_ORDERS, Optional.empty(), Optional.empty()); assertInstanceOf(tableHandle, RaptorTableHandle.class); RaptorTableHandle raptorTableHandle = (RaptorTableHandle) tableHandle; assertThat(raptorTableHandle.getTableId()).isEqualTo(1); @@ -540,12 +540,12 @@ public void testSortOrderProperty() @Test public void testTemporalColumn() { - assertThat(metadata.getTableHandle(SESSION, DEFAULT_TEST_ORDERS)).isNull(); + assertThat(metadata.getTableHandle(SESSION, DEFAULT_TEST_ORDERS, Optional.empty(), Optional.empty())).isNull(); ConnectorTableMetadata ordersTable = getOrdersTable(ImmutableMap.of(TEMPORAL_COLUMN_PROPERTY, "orderdate")); metadata.createTable(SESSION, ordersTable, false); - ConnectorTableHandle tableHandle = metadata.getTableHandle(SESSION, DEFAULT_TEST_ORDERS); + ConnectorTableHandle tableHandle = metadata.getTableHandle(SESSION, DEFAULT_TEST_ORDERS, Optional.empty(), Optional.empty()); assertInstanceOf(tableHandle, RaptorTableHandle.class); RaptorTableHandle raptorTableHandle = (RaptorTableHandle) tableHandle; assertThat(raptorTableHandle.getTableId()).isEqualTo(1); @@ -677,7 +677,7 @@ public void testTransactionInsert() // start insert transactionId++; - ConnectorTableHandle tableHandle = metadata.getTableHandle(SESSION, DEFAULT_TEST_ORDERS); + ConnectorTableHandle tableHandle = metadata.getTableHandle(SESSION, DEFAULT_TEST_ORDERS, Optional.empty(), Optional.empty()); ConnectorInsertTableHandle insertHandle = metadata.beginInsert(SESSION, tableHandle, ImmutableList.of(), NO_RETRIES); // transaction is in progress diff --git a/plugin/trino-raptor-legacy/src/test/java/io/trino/plugin/raptor/legacy/metadata/TestRaptorSplitManager.java b/plugin/trino-raptor-legacy/src/test/java/io/trino/plugin/raptor/legacy/metadata/TestRaptorSplitManager.java index 6e510897ef06..b3a357c72569 100644 --- a/plugin/trino-raptor-legacy/src/test/java/io/trino/plugin/raptor/legacy/metadata/TestRaptorSplitManager.java +++ b/plugin/trino-raptor-legacy/src/test/java/io/trino/plugin/raptor/legacy/metadata/TestRaptorSplitManager.java @@ -111,7 +111,7 @@ public void setup() metadata = new RaptorMetadata(dbi, shardManager); metadata.createTable(SESSION, TEST_TABLE, false); - tableHandle = metadata.getTableHandle(SESSION, TEST_TABLE.getTable()); + tableHandle = metadata.getTableHandle(SESSION, TEST_TABLE.getTable(), Optional.empty(), Optional.empty()); List shards = ImmutableList.builder() .add(shardInfo(UUID.randomUUID(), nodeName)) diff --git a/plugin/trino-redis/src/main/java/io/trino/plugin/redis/RedisMetadata.java b/plugin/trino-redis/src/main/java/io/trino/plugin/redis/RedisMetadata.java index 25f39c1ea545..5575870bf1aa 100644 --- a/plugin/trino-redis/src/main/java/io/trino/plugin/redis/RedisMetadata.java +++ b/plugin/trino-redis/src/main/java/io/trino/plugin/redis/RedisMetadata.java @@ -20,12 +20,14 @@ import com.google.inject.Inject; import io.airlift.log.Logger; import io.trino.decoder.dummy.DummyRowDecoder; +import io.trino.spi.TrinoException; import io.trino.spi.connector.ColumnHandle; import io.trino.spi.connector.ColumnMetadata; import io.trino.spi.connector.ConnectorMetadata; import io.trino.spi.connector.ConnectorSession; import io.trino.spi.connector.ConnectorTableHandle; import io.trino.spi.connector.ConnectorTableMetadata; +import io.trino.spi.connector.ConnectorTableVersion; import io.trino.spi.connector.Constraint; import io.trino.spi.connector.ConstraintApplicationResult; import io.trino.spi.connector.SchemaTableName; @@ -49,6 +51,7 @@ import java.util.stream.Collectors; import static io.trino.plugin.redis.RedisSplit.toRedisDataType; +import static io.trino.spi.StandardErrorCode.NOT_SUPPORTED; import static java.util.Objects.requireNonNull; import static java.util.concurrent.TimeUnit.MILLISECONDS; @@ -92,8 +95,12 @@ public List listSchemaNames(ConnectorSession session) } @Override - public RedisTableHandle getTableHandle(ConnectorSession session, SchemaTableName schemaTableName) + public RedisTableHandle getTableHandle(ConnectorSession session, SchemaTableName schemaTableName, Optional startVersion, Optional endVersion) { + if (startVersion.isPresent() || endVersion.isPresent()) { + throw new TrinoException(NOT_SUPPORTED, "This connector does not support versioned tables"); + } + RedisTableDescription table = getDefinedTables().get(schemaTableName); if (table == null) { return null; diff --git a/plugin/trino-thrift/src/main/java/io/trino/plugin/thrift/ThriftMetadata.java b/plugin/trino-thrift/src/main/java/io/trino/plugin/thrift/ThriftMetadata.java index 286d9354354e..a3ad9a94013f 100644 --- a/plugin/trino-thrift/src/main/java/io/trino/plugin/thrift/ThriftMetadata.java +++ b/plugin/trino-thrift/src/main/java/io/trino/plugin/thrift/ThriftMetadata.java @@ -37,6 +37,7 @@ import io.trino.spi.connector.ConnectorSession; import io.trino.spi.connector.ConnectorTableHandle; import io.trino.spi.connector.ConnectorTableMetadata; +import io.trino.spi.connector.ConnectorTableVersion; import io.trino.spi.connector.Constraint; import io.trino.spi.connector.ConstraintApplicationResult; import io.trino.spi.connector.ProjectionApplicationResult; @@ -60,6 +61,7 @@ import static io.trino.cache.SafeCaches.buildNonEvictableCache; import static io.trino.plugin.thrift.ThriftErrorCode.THRIFT_SERVICE_INVALID_RESPONSE; import static io.trino.plugin.thrift.util.ThriftExceptions.toTrinoException; +import static io.trino.spi.StandardErrorCode.NOT_SUPPORTED; import static java.util.Objects.requireNonNull; import static java.util.concurrent.TimeUnit.MILLISECONDS; import static java.util.concurrent.TimeUnit.MINUTES; @@ -105,8 +107,12 @@ public List listSchemaNames(ConnectorSession session) } @Override - public ConnectorTableHandle getTableHandle(ConnectorSession session, SchemaTableName tableName) + public ConnectorTableHandle getTableHandle(ConnectorSession session, SchemaTableName tableName, Optional startVersion, Optional endVersion) { + if (startVersion.isPresent() || endVersion.isPresent()) { + throw new TrinoException(NOT_SUPPORTED, "This connector does not support versioned tables"); + } + return tableCache.getUnchecked(tableName) .map(ThriftTableMetadata::getSchemaTableName) .map(ThriftTableHandle::new) diff --git a/plugin/trino-tpcds/src/main/java/io/trino/plugin/tpcds/TpcdsMetadata.java b/plugin/trino-tpcds/src/main/java/io/trino/plugin/tpcds/TpcdsMetadata.java index 79ec83c0b6a5..2479968cd4ce 100644 --- a/plugin/trino-tpcds/src/main/java/io/trino/plugin/tpcds/TpcdsMetadata.java +++ b/plugin/trino-tpcds/src/main/java/io/trino/plugin/tpcds/TpcdsMetadata.java @@ -17,12 +17,14 @@ import com.google.common.collect.ImmutableMap; import com.google.common.collect.ImmutableSet; import io.trino.plugin.tpcds.statistics.TpcdsTableStatisticsFactory; +import io.trino.spi.TrinoException; import io.trino.spi.connector.ColumnHandle; import io.trino.spi.connector.ColumnMetadata; import io.trino.spi.connector.ConnectorMetadata; import io.trino.spi.connector.ConnectorSession; import io.trino.spi.connector.ConnectorTableHandle; import io.trino.spi.connector.ConnectorTableMetadata; +import io.trino.spi.connector.ConnectorTableVersion; import io.trino.spi.connector.SchemaTableName; import io.trino.spi.connector.SchemaTablePrefix; import io.trino.spi.statistics.TableStatistics; @@ -40,6 +42,7 @@ import java.util.Optional; import java.util.Set; +import static io.trino.spi.StandardErrorCode.NOT_SUPPORTED; import static io.trino.spi.type.CharType.createCharType; import static io.trino.spi.type.DecimalType.createDecimalType; import static io.trino.spi.type.VarcharType.createVarcharType; @@ -81,8 +84,12 @@ public List listSchemaNames(ConnectorSession session) } @Override - public ConnectorTableHandle getTableHandle(ConnectorSession session, SchemaTableName tableName) + public ConnectorTableHandle getTableHandle(ConnectorSession session, SchemaTableName tableName, Optional startVersion, Optional endVersion) { + if (startVersion.isPresent() || endVersion.isPresent()) { + throw new TrinoException(NOT_SUPPORTED, "This connector does not support versioned tables"); + } + requireNonNull(tableName, "tableName is null"); if (!tableNames.contains(tableName.getTableName())) { return null; diff --git a/plugin/trino-tpcds/src/test/java/io/trino/plugin/tpcds/TestTpcdsMetadataStatistics.java b/plugin/trino-tpcds/src/test/java/io/trino/plugin/tpcds/TestTpcdsMetadataStatistics.java index 4b2496e556cb..379d3d87b0ae 100644 --- a/plugin/trino-tpcds/src/test/java/io/trino/plugin/tpcds/TestTpcdsMetadataStatistics.java +++ b/plugin/trino-tpcds/src/test/java/io/trino/plugin/tpcds/TestTpcdsMetadataStatistics.java @@ -28,6 +28,7 @@ import org.junit.jupiter.api.Test; import java.util.Map; +import java.util.Optional; import java.util.stream.Stream; import static org.assertj.core.api.Assertions.assertThat; @@ -45,7 +46,7 @@ public void testNoTableStatsForNotSupportedSchema() .forEach(schemaName -> Table.getBaseTables() .forEach(table -> { SchemaTableName schemaTableName = new SchemaTableName(schemaName, table.getName()); - ConnectorTableHandle tableHandle = metadata.getTableHandle(session, schemaTableName); + ConnectorTableHandle tableHandle = metadata.getTableHandle(session, schemaTableName, Optional.empty(), Optional.empty()); TableStatistics tableStatistics = metadata.getTableStatistics(session, tableHandle); assertThat(tableStatistics.getRowCount().isUnknown()).isTrue(); assertThat(tableStatistics.getColumnStatistics().isEmpty()).isTrue(); @@ -59,7 +60,7 @@ public void testTableStatsExistenceSupportedSchema() .forEach(schemaName -> Table.getBaseTables() .forEach(table -> { SchemaTableName schemaTableName = new SchemaTableName(schemaName, table.getName()); - ConnectorTableHandle tableHandle = metadata.getTableHandle(session, schemaTableName); + ConnectorTableHandle tableHandle = metadata.getTableHandle(session, schemaTableName, Optional.empty(), Optional.empty()); TableStatistics tableStatistics = metadata.getTableStatistics(session, tableHandle); assertThat(tableStatistics.getRowCount().isUnknown()).isFalse(); for (ColumnHandle column : metadata.getColumnHandles(session, tableHandle).values()) { @@ -73,7 +74,7 @@ public void testTableStatsExistenceSupportedSchema() public void testTableStatsDetails() { SchemaTableName schemaTableName = new SchemaTableName("sf1", Table.CALL_CENTER.getName()); - ConnectorTableHandle tableHandle = metadata.getTableHandle(session, schemaTableName); + ConnectorTableHandle tableHandle = metadata.getTableHandle(session, schemaTableName, Optional.empty(), Optional.empty()); TableStatistics tableStatistics = metadata.getTableStatistics(session, tableHandle); estimateAssertion.assertClose(tableStatistics.getRowCount(), Estimate.of(6), "Row count does not match"); @@ -143,7 +144,7 @@ public void testTableStatsDetails() public void testNullFraction() { SchemaTableName schemaTableName = new SchemaTableName("sf1", Table.WEB_SITE.getName()); - ConnectorTableHandle tableHandle = metadata.getTableHandle(session, schemaTableName); + ConnectorTableHandle tableHandle = metadata.getTableHandle(session, schemaTableName, Optional.empty(), Optional.empty()); TableStatistics tableStatistics = metadata.getTableStatistics(session, tableHandle); Map columnHandles = metadata.getColumnHandles(session, tableHandle); diff --git a/plugin/trino-tpch/src/main/java/io/trino/plugin/tpch/TpchMetadata.java b/plugin/trino-tpch/src/main/java/io/trino/plugin/tpch/TpchMetadata.java index 6d01794d483f..90cd6c52a498 100644 --- a/plugin/trino-tpch/src/main/java/io/trino/plugin/tpch/TpchMetadata.java +++ b/plugin/trino-tpch/src/main/java/io/trino/plugin/tpch/TpchMetadata.java @@ -26,6 +26,7 @@ import io.trino.plugin.tpch.statistics.StatisticsEstimator; import io.trino.plugin.tpch.statistics.TableStatisticsData; import io.trino.plugin.tpch.statistics.TableStatisticsDataRepository; +import io.trino.spi.TrinoException; import io.trino.spi.connector.CatalogSchemaTableName; import io.trino.spi.connector.ColumnHandle; import io.trino.spi.connector.ColumnMetadata; @@ -37,6 +38,7 @@ import io.trino.spi.connector.ConnectorTableMetadata; import io.trino.spi.connector.ConnectorTablePartitioning; import io.trino.spi.connector.ConnectorTableProperties; +import io.trino.spi.connector.ConnectorTableVersion; import io.trino.spi.connector.Constraint; import io.trino.spi.connector.ConstraintApplicationResult; import io.trino.spi.connector.LocalProperty; @@ -80,6 +82,7 @@ import static com.google.common.collect.Maps.asMap; import static io.trino.plugin.tpch.util.PredicateUtils.convertToPredicate; import static io.trino.plugin.tpch.util.PredicateUtils.filterOutColumnFromPredicate; +import static io.trino.spi.StandardErrorCode.NOT_SUPPORTED; import static io.trino.spi.statistics.TableStatisticType.ROW_COUNT; import static io.trino.spi.type.BigintType.BIGINT; import static io.trino.spi.type.DateType.DATE; @@ -187,8 +190,12 @@ public List listSchemaNames(ConnectorSession session) } @Override - public TpchTableHandle getTableHandle(ConnectorSession session, SchemaTableName tableName) + public TpchTableHandle getTableHandle(ConnectorSession session, SchemaTableName tableName, Optional startVersion, Optional endVersion) { + if (startVersion.isPresent() || endVersion.isPresent()) { + throw new TrinoException(NOT_SUPPORTED, "This connector does not support versioned tables"); + } + requireNonNull(tableName, "tableName is null"); if (!tableNames.contains(tableName.getTableName())) { return null; diff --git a/plugin/trino-tpch/src/test/java/io/trino/plugin/tpch/TestTpchMetadata.java b/plugin/trino-tpch/src/test/java/io/trino/plugin/tpch/TestTpchMetadata.java index 13ad4a3fc982..3f1fc0bcdfe3 100644 --- a/plugin/trino-tpch/src/test/java/io/trino/plugin/tpch/TestTpchMetadata.java +++ b/plugin/trino-tpch/src/test/java/io/trino/plugin/tpch/TestTpchMetadata.java @@ -146,7 +146,7 @@ public void testGetTableMetadata() private void testGetTableMetadata(String schema, TpchTable table) { - TpchTableHandle tableHandle = tpchMetadata.getTableHandle(session, new SchemaTableName(schema, table.getTableName())); + TpchTableHandle tableHandle = tpchMetadata.getTableHandle(session, new SchemaTableName(schema, table.getTableName()), Optional.empty(), Optional.empty()); ConnectorTableMetadata tableMetadata = tpchMetadata.getTableMetadata(session, tableHandle); assertThat(tableMetadata.getTableSchema().getTable().getTableName()).isEqualTo(table.getTableName()); assertThat(tableMetadata.getTableSchema().getTable().getSchemaName()).isEqualTo(schema); @@ -170,7 +170,7 @@ private void testTableStats(String schema, TpchTable table, double expectedRo private void testTableStats(String schema, TpchTable table, Constraint constraint, double expectedRowCount) { - TpchTableHandle tableHandle = tpchMetadata.getTableHandle(session, new SchemaTableName(schema, table.getTableName())); + TpchTableHandle tableHandle = tpchMetadata.getTableHandle(session, new SchemaTableName(schema, table.getTableName()), Optional.empty(), Optional.empty()); Optional> result = tpchMetadata.applyFilter(session, tableHandle, constraint); if (result.isPresent()) { tableHandle = (TpchTableHandle) result.get().getHandle(); @@ -185,7 +185,7 @@ private void testTableStats(String schema, TpchTable table, Constraint constr private void testNoTableStats(String schema, TpchTable table) { - TpchTableHandle tableHandle = tpchMetadata.getTableHandle(session, new SchemaTableName(schema, table.getTableName())); + TpchTableHandle tableHandle = tpchMetadata.getTableHandle(session, new SchemaTableName(schema, table.getTableName()), Optional.empty(), Optional.empty()); TableStatistics tableStatistics = tpchMetadata.getTableStatistics(session, tableHandle); assertThat(tableStatistics.getRowCount().isUnknown()).isTrue(); } @@ -276,7 +276,7 @@ private void testColumnStats(String schema, TpchTable table, TpchColumn co private void testColumnStats(String schema, TpchTable table, TpchColumn column, Constraint constraint, ColumnStatistics expected) { - TpchTableHandle tableHandle = tpchMetadata.getTableHandle(session, new SchemaTableName(schema, table.getTableName())); + TpchTableHandle tableHandle = tpchMetadata.getTableHandle(session, new SchemaTableName(schema, table.getTableName()), Optional.empty(), Optional.empty()); Optional> result = tpchMetadata.applyFilter(session, tableHandle, constraint); if (result.isPresent()) { tableHandle = (TpchTableHandle) result.get().getHandle(); @@ -297,7 +297,7 @@ private void testColumnStats(String schema, TpchTable table, TpchColumn co @Test public void testOrdersOrderStatusPredicatePushdown() { - TpchTableHandle tableHandle = tpchMetadata.getTableHandle(session, new SchemaTableName("sf1", ORDERS.getTableName())); + TpchTableHandle tableHandle = tpchMetadata.getTableHandle(session, new SchemaTableName("sf1", ORDERS.getTableName()), Optional.empty(), Optional.empty()); TupleDomain domain; ConstraintApplicationResult result; @@ -324,7 +324,7 @@ public void testOrdersOrderStatusPredicatePushdown() @Test public void testPartTypeAndPartContainerPredicatePushdown() { - TpchTableHandle tableHandle = tpchMetadata.getTableHandle(session, new SchemaTableName("sf1", PART.getTableName())); + TpchTableHandle tableHandle = tpchMetadata.getTableHandle(session, new SchemaTableName("sf1", PART.getTableName()), Optional.empty(), Optional.empty()); TupleDomain domain; ConstraintApplicationResult result; diff --git a/testing/trino-testing/src/main/java/io/trino/testing/BaseConnectorTest.java b/testing/trino-testing/src/main/java/io/trino/testing/BaseConnectorTest.java index 184100487357..bbe1ff4fc525 100644 --- a/testing/trino-testing/src/main/java/io/trino/testing/BaseConnectorTest.java +++ b/testing/trino-testing/src/main/java/io/trino/testing/BaseConnectorTest.java @@ -690,13 +690,13 @@ public void testSelectInTransaction() @Test public void testSelectVersionOfNonExistentTable() { + String tableName = "foo_" + randomNameSuffix(); String catalog = getSession().getCatalog().orElseThrow(); String schema = getSession().getSchema().orElseThrow(); - String tableName = "foo_" + randomNameSuffix(); assertThat(query("SELECT * FROM " + tableName + " FOR TIMESTAMP AS OF TIMESTAMP '2021-03-01 00:00:01'")) - .failure().hasMessage(format("line 1:15: Table '%s.%s.%s' does not exist", catalog, schema, tableName)); + .failure().hasMessageMatching("line 1:15: Table '%s.%s.%s' does not exist|This connector does not support versioned tables".formatted(catalog, schema, tableName)); assertThat(query("SELECT * FROM " + tableName + " FOR VERSION AS OF 'version1'")) - .failure().hasMessage(format("line 1:15: Table '%s.%s.%s' does not exist", catalog, schema, tableName)); + .failure().hasMessageMatching("line 1:15: Table '%s.%s.%s' does not exist|This connector does not support versioned tables".formatted(catalog, schema, tableName)); } /** diff --git a/testing/trino-testing/src/main/java/io/trino/testing/H2QueryRunner.java b/testing/trino-testing/src/main/java/io/trino/testing/H2QueryRunner.java index 52de82add49c..059e7784e887 100644 --- a/testing/trino-testing/src/main/java/io/trino/testing/H2QueryRunner.java +++ b/testing/trino-testing/src/main/java/io/trino/testing/H2QueryRunner.java @@ -177,7 +177,7 @@ public H2QueryRunner() private void insertRows(TpchMetadata tpchMetadata, TpchTable tpchTable) { - TpchTableHandle tableHandle = tpchMetadata.getTableHandle(null, new SchemaTableName(TINY_SCHEMA_NAME, tpchTable.getTableName())); + TpchTableHandle tableHandle = tpchMetadata.getTableHandle(null, new SchemaTableName(TINY_SCHEMA_NAME, tpchTable.getTableName()), Optional.empty(), Optional.empty()); insertRows(tpchMetadata.getTableMetadata(null, tableHandle), handle, createTpchRecordSet(tpchTable, tableHandle.scaleFactor())); } diff --git a/testing/trino-testing/src/main/java/io/trino/testing/tpch/TpchIndexedData.java b/testing/trino-testing/src/main/java/io/trino/testing/tpch/TpchIndexedData.java index 53f1c2a1087d..1961aefd674d 100644 --- a/testing/trino-testing/src/main/java/io/trino/testing/tpch/TpchIndexedData.java +++ b/testing/trino-testing/src/main/java/io/trino/testing/tpch/TpchIndexedData.java @@ -64,7 +64,7 @@ public TpchIndexedData(TpchIndexSpec tpchIndexSpec) Set tables = tpchIndexSpec.listIndexedTables(); for (TpchScaledTable table : tables) { SchemaTableName tableName = new SchemaTableName("sf" + table.getScaleFactor(), table.getTableName()); - TpchTableHandle tableHandle = tpchMetadata.getTableHandle(null, tableName); + TpchTableHandle tableHandle = tpchMetadata.getTableHandle(null, tableName, Optional.empty(), Optional.empty()); Map columnHandles = new LinkedHashMap<>(tpchMetadata.getColumnHandles(null, tableHandle)); for (Set columnNames : tpchIndexSpec.getColumnIndexes(table)) { List keyColumnNames = ImmutableList.copyOf(columnNames); // Finalize the key order