From 237dbc06cb2fb4aacc57491c4f839722e3ec7701 Mon Sep 17 00:00:00 2001 From: David Phillips Date: Tue, 17 Sep 2024 17:48:14 -0700 Subject: [PATCH 1/8] Add TPCDS for Iceberg development query runners --- plugin/trino-iceberg/pom.xml | 6 +++ .../plugin/iceberg/IcebergQueryRunner.java | 47 ++++++++++++------- 2 files changed, 35 insertions(+), 18 deletions(-) diff --git a/plugin/trino-iceberg/pom.xml b/plugin/trino-iceberg/pom.xml index 48b035b4d0be..bae1cc87934d 100644 --- a/plugin/trino-iceberg/pom.xml +++ b/plugin/trino-iceberg/pom.xml @@ -558,6 +558,12 @@ test + + io.trino + trino-tpcds + test + + io.trino trino-tpch diff --git a/plugin/trino-iceberg/src/test/java/io/trino/plugin/iceberg/IcebergQueryRunner.java b/plugin/trino-iceberg/src/test/java/io/trino/plugin/iceberg/IcebergQueryRunner.java index 274646e6ac2c..99dc8bd8535d 100644 --- a/plugin/trino-iceberg/src/test/java/io/trino/plugin/iceberg/IcebergQueryRunner.java +++ b/plugin/trino-iceberg/src/test/java/io/trino/plugin/iceberg/IcebergQueryRunner.java @@ -25,6 +25,7 @@ import io.trino.plugin.hive.containers.HiveMinioDataLake; import io.trino.plugin.iceberg.catalog.jdbc.TestingIcebergJdbcServer; import io.trino.plugin.iceberg.catalog.rest.TestingPolarisCatalog; +import io.trino.plugin.tpcds.TpcdsPlugin; import io.trino.plugin.tpch.TpchPlugin; import io.trino.testing.DistributedQueryRunner; import io.trino.testing.QueryRunner; @@ -84,6 +85,7 @@ public static class Builder private Optional metastoreDirectory = Optional.empty(); private ImmutableMap.Builder icebergProperties = ImmutableMap.builder(); private Optional schemaInitializer = Optional.empty(); + private boolean tpcdsCatalogEnabled; protected Builder() { @@ -139,6 +141,12 @@ public Builder setSchemaInitializer(SchemaInitializer schemaInitializer) return self(); } + public Builder setTpcdsCatalogEnabled(boolean tpcdsCatalogEnabled) + { + this.tpcdsCatalogEnabled = tpcdsCatalogEnabled; + return self(); + } + @Override public DistributedQueryRunner build() throws Exception @@ -148,6 +156,11 @@ public DistributedQueryRunner build() queryRunner.installPlugin(new TpchPlugin()); queryRunner.createCatalog("tpch", "tpch"); + if (tpcdsCatalogEnabled) { + queryRunner.installPlugin(new TpcdsPlugin()); + queryRunner.createCatalog("tpcds", "tpcds"); + } + if (!icebergProperties.buildOrThrow().containsKey("fs.hadoop.enabled")) { icebergProperties.put("fs.hadoop.enabled", "true"); } @@ -166,6 +179,13 @@ public DistributedQueryRunner build() } } + private static Builder icebergQueryRunnerMainBuilder() + { + return IcebergQueryRunner.builder() + .addCoordinatorProperty("http-server.http.port", "8080") + .setTpcdsCatalogEnabled(true); + } + public static final class IcebergRestQueryRunnerMain { private IcebergRestQueryRunnerMain() {} @@ -186,8 +206,7 @@ public static void main(String[] args) testServer.start(); @SuppressWarnings("resource") - QueryRunner queryRunner = IcebergQueryRunner.builder() - .addCoordinatorProperty("http-server.http.port", "8080") + QueryRunner queryRunner = icebergQueryRunnerMainBuilder() .setBaseDataDir(Optional.of(warehouseLocation.toPath())) .setIcebergProperties(ImmutableMap.of( "iceberg.catalog.type", "rest", @@ -215,8 +234,7 @@ public static void main(String[] args) TestingPolarisCatalog polarisCatalog = new TestingPolarisCatalog(warehouseLocation.getPath()); @SuppressWarnings("resource") - QueryRunner queryRunner = IcebergQueryRunner.builder() - .addCoordinatorProperty("http-server.http.port", "8080") + QueryRunner queryRunner = icebergQueryRunnerMainBuilder() .setBaseDataDir(Optional.of(warehouseLocation.toPath())) .addIcebergProperty("iceberg.catalog.type", "rest") .addIcebergProperty("iceberg.rest-catalog.uri", polarisCatalog.restUri() + "/api/catalog") @@ -243,8 +261,7 @@ public static void main(String[] args) // Requires AWS credentials, which can be provided any way supported by the DefaultProviderChain // See https://docs.aws.amazon.com/sdk-for-java/v1/developer-guide/credentials.html#credentials-default @SuppressWarnings("resource") - QueryRunner queryRunner = IcebergQueryRunner.builder() - .addCoordinatorProperty("http-server.http.port", "8080") + QueryRunner queryRunner = icebergQueryRunnerMainBuilder() .setIcebergProperties(ImmutableMap.of("iceberg.catalog.type", "glue")) .build(); @@ -310,8 +327,7 @@ public static void main(String[] args) minio.createBucket(bucketName); @SuppressWarnings("resource") - QueryRunner queryRunner = IcebergQueryRunner.builder() - .addCoordinatorProperty("http-server.http.port", "8080") + QueryRunner queryRunner = icebergQueryRunnerMainBuilder() .setIcebergProperties(Map.of( "iceberg.catalog.type", "TESTING_FILE_METASTORE", "hive.metastore.catalog.dir", "s3://%s/".formatted(bucketName), @@ -364,8 +380,7 @@ public static void main(String[] args) hiveHadoop.start(); @SuppressWarnings("resource") - QueryRunner queryRunner = IcebergQueryRunner.builder() - .addCoordinatorProperty("http-server.http.port", "8080") + QueryRunner queryRunner = icebergQueryRunnerMainBuilder() .setIcebergProperties(Map.of( "iceberg.catalog.type", "HIVE_METASTORE", "hive.metastore.uri", hiveHadoop.getHiveMetastoreEndpoint().toString(), @@ -400,8 +415,7 @@ public static void main(String[] args) TestingIcebergJdbcServer server = new TestingIcebergJdbcServer(); @SuppressWarnings("resource") - QueryRunner queryRunner = IcebergQueryRunner.builder() - .addCoordinatorProperty("http-server.http.port", "8080") + QueryRunner queryRunner = icebergQueryRunnerMainBuilder() .setIcebergProperties(ImmutableMap.builder() .put("iceberg.catalog.type", "jdbc") .put("iceberg.jdbc-catalog.driver-class", "org.postgresql.Driver") @@ -428,8 +442,7 @@ public static void main(String[] args) throws Exception { @SuppressWarnings("resource") - QueryRunner queryRunner = IcebergQueryRunner.builder() - .addCoordinatorProperty("http-server.http.port", "8080") + QueryRunner queryRunner = icebergQueryRunnerMainBuilder() .setIcebergProperties(ImmutableMap.builder() .put("iceberg.catalog.type", "snowflake") .put("fs.native-s3.enabled", "true") @@ -466,8 +479,7 @@ public static void main(String[] args) metastoreDir.deleteOnExit(); @SuppressWarnings("resource") - QueryRunner queryRunner = IcebergQueryRunner.builder() - .addCoordinatorProperty("http-server.http.port", "8080") + QueryRunner queryRunner = icebergQueryRunnerMainBuilder() .addIcebergProperty("hive.metastore.catalog.dir", metastoreDir.toURI().toString()) .setInitialTables(TpchTable.getTables()) .build(); @@ -495,8 +507,7 @@ public static void main(String[] args) metastoreDir.deleteOnExit(); @SuppressWarnings("resource") - QueryRunner queryRunner = IcebergQueryRunner.builder() - .addCoordinatorProperty("http-server.http.port", "8080") + QueryRunner queryRunner = icebergQueryRunnerMainBuilder() .addIcebergProperty("hive.metastore.catalog.dir", metastoreDir.toURI().toString()) .setExtraProperties(ImmutableMap.builder() .put("retry-policy", "TASK") From b684002ca639207580d69f632c03d2721ef4a897 Mon Sep 17 00:00:00 2001 From: David Phillips Date: Tue, 17 Sep 2024 13:25:14 -0700 Subject: [PATCH 2/8] Rename compatibility product test config files --- .../env/environment/EnvSinglenodeCompatibility.java | 6 +++--- ...th-system-memory.properties => config-pre369.properties} | 0 .../{hive-hadoop2.properties => hive-pre359.properties} | 0 .../{iceberg_old.properties => iceberg-pre359.properties} | 0 4 files changed, 3 insertions(+), 3 deletions(-) rename testing/trino-product-tests-launcher/src/main/resources/docker/presto-product-tests/conf/environment/singlenode-compatibility/{config-with-system-memory.properties => config-pre369.properties} (100%) rename testing/trino-product-tests-launcher/src/main/resources/docker/presto-product-tests/conf/environment/singlenode-compatibility/{hive-hadoop2.properties => hive-pre359.properties} (100%) rename testing/trino-product-tests-launcher/src/main/resources/docker/presto-product-tests/conf/environment/singlenode-compatibility/{iceberg_old.properties => iceberg-pre359.properties} (100%) diff --git a/testing/trino-product-tests-launcher/src/main/java/io/trino/tests/product/launcher/env/environment/EnvSinglenodeCompatibility.java b/testing/trino-product-tests-launcher/src/main/java/io/trino/tests/product/launcher/env/environment/EnvSinglenodeCompatibility.java index 7ffb510e4b43..147967fc3c6f 100644 --- a/testing/trino-product-tests-launcher/src/main/java/io/trino/tests/product/launcher/env/environment/EnvSinglenodeCompatibility.java +++ b/testing/trino-product-tests-launcher/src/main/java/io/trino/tests/product/launcher/env/environment/EnvSinglenodeCompatibility.java @@ -107,7 +107,7 @@ protected String getConfigurationDirectory(String dockerImage) private String getConfigFileFor(String dockerImage) { if (getVersionFromDockerImageName(dockerImage) < 369) { - return "config-with-system-memory.properties"; + return "config-pre369.properties"; } return "config.properties"; } @@ -115,7 +115,7 @@ private String getConfigFileFor(String dockerImage) private String getHiveConfigFor(String dockerImage) { if (getVersionFromDockerImageName(dockerImage) < 359) { - return "hive-hadoop2.properties"; + return "hive-pre359.properties"; } return "hive.properties"; } @@ -123,7 +123,7 @@ private String getHiveConfigFor(String dockerImage) private String getIcebergConfigFor(String dockerImage) { if (getVersionFromDockerImageName(dockerImage) < 359) { - return "iceberg_old.properties"; + return "iceberg-pre359.properties"; } return "iceberg.properties"; } diff --git a/testing/trino-product-tests-launcher/src/main/resources/docker/presto-product-tests/conf/environment/singlenode-compatibility/config-with-system-memory.properties b/testing/trino-product-tests-launcher/src/main/resources/docker/presto-product-tests/conf/environment/singlenode-compatibility/config-pre369.properties similarity index 100% rename from testing/trino-product-tests-launcher/src/main/resources/docker/presto-product-tests/conf/environment/singlenode-compatibility/config-with-system-memory.properties rename to testing/trino-product-tests-launcher/src/main/resources/docker/presto-product-tests/conf/environment/singlenode-compatibility/config-pre369.properties diff --git a/testing/trino-product-tests-launcher/src/main/resources/docker/presto-product-tests/conf/environment/singlenode-compatibility/hive-hadoop2.properties b/testing/trino-product-tests-launcher/src/main/resources/docker/presto-product-tests/conf/environment/singlenode-compatibility/hive-pre359.properties similarity index 100% rename from testing/trino-product-tests-launcher/src/main/resources/docker/presto-product-tests/conf/environment/singlenode-compatibility/hive-hadoop2.properties rename to testing/trino-product-tests-launcher/src/main/resources/docker/presto-product-tests/conf/environment/singlenode-compatibility/hive-pre359.properties diff --git a/testing/trino-product-tests-launcher/src/main/resources/docker/presto-product-tests/conf/environment/singlenode-compatibility/iceberg_old.properties b/testing/trino-product-tests-launcher/src/main/resources/docker/presto-product-tests/conf/environment/singlenode-compatibility/iceberg-pre359.properties similarity index 100% rename from testing/trino-product-tests-launcher/src/main/resources/docker/presto-product-tests/conf/environment/singlenode-compatibility/iceberg_old.properties rename to testing/trino-product-tests-launcher/src/main/resources/docker/presto-product-tests/conf/environment/singlenode-compatibility/iceberg-pre359.properties From 7aecc320e00bcb873390d21a61161b99a94f400a Mon Sep 17 00:00:00 2001 From: David Phillips Date: Sun, 15 Sep 2024 16:55:00 -0700 Subject: [PATCH 3/8] Optimize isDynamicFilter planner check --- .../src/main/java/io/trino/sql/DynamicFilters.java | 10 +++------- 1 file changed, 3 insertions(+), 7 deletions(-) diff --git a/core/trino-main/src/main/java/io/trino/sql/DynamicFilters.java b/core/trino-main/src/main/java/io/trino/sql/DynamicFilters.java index 15b2395b9c29..02e9879121f8 100644 --- a/core/trino-main/src/main/java/io/trino/sql/DynamicFilters.java +++ b/core/trino-main/src/main/java/io/trino/sql/DynamicFilters.java @@ -150,20 +150,16 @@ public static Expression replaceDynamicFilterId(Call dynamicFilterFunctionCall, public static boolean isDynamicFilter(Expression expression) { - return getDescriptor(expression).isPresent(); + return (expression instanceof Call call) && isDynamicFilterFunction(call); } public static Optional getDescriptor(Expression expression) { - if (!(expression instanceof Call call)) { + if (!isDynamicFilter(expression)) { return Optional.empty(); } - if (!isDynamicFilterFunction(call)) { - return Optional.empty(); - } - - List arguments = call.arguments(); + List arguments = ((Call) expression).arguments(); checkArgument(arguments.size() == 4, "invalid arguments count: %s", arguments.size()); Expression probeSymbol = arguments.get(0); From b2fef758ce43f47a378e9fa78b5a1cc906fc9dd2 Mon Sep 17 00:00:00 2001 From: David Phillips Date: Mon, 16 Sep 2024 18:48:14 -0700 Subject: [PATCH 4/8] Use PlanTester session in tests --- ...tAddLocalExchangesForTaskScaleWriters.java | 38 +++++++++++-------- 1 file changed, 22 insertions(+), 16 deletions(-) diff --git a/core/trino-main/src/test/java/io/trino/sql/planner/optimizations/TestAddLocalExchangesForTaskScaleWriters.java b/core/trino-main/src/test/java/io/trino/sql/planner/optimizations/TestAddLocalExchangesForTaskScaleWriters.java index a277a2674ddb..0926ba9c65c3 100644 --- a/core/trino-main/src/test/java/io/trino/sql/planner/optimizations/TestAddLocalExchangesForTaskScaleWriters.java +++ b/core/trino-main/src/test/java/io/trino/sql/planner/optimizations/TestAddLocalExchangesForTaskScaleWriters.java @@ -18,6 +18,7 @@ import com.google.common.collect.ImmutableMap; import com.google.common.collect.ImmutableSet; import io.trino.Session; +import io.trino.Session.SessionBuilder; import io.trino.connector.MockConnector; import io.trino.connector.MockConnectorColumnHandle; import io.trino.connector.MockConnectorFactory; @@ -63,7 +64,7 @@ import static io.trino.sql.planner.plan.ExchangeNode.Scope.REMOTE; import static io.trino.sql.planner.plan.ExchangeNode.Type.GATHER; import static io.trino.sql.planner.plan.ExchangeNode.Type.REPARTITION; -import static io.trino.testing.TestingSession.testSessionBuilder; +import static io.trino.testing.TestingSession.testSession; public class TestAddLocalExchangesForTaskScaleWriters extends BasePlanTest @@ -73,7 +74,7 @@ public class TestAddLocalExchangesForTaskScaleWriters @Override protected PlanTester createPlanTester() { - PlanTester planTester = PlanTester.create(testSessionBuilder().build()); + PlanTester planTester = PlanTester.create(testSession()); planTester.createCatalog( "mock_with_scaled_writers", createConnectorFactory("mock_with_scaled_writers", true, true), @@ -150,7 +151,7 @@ public void testLocalScaledUnpartitionedWriterDistribution() { assertDistributedPlan( "INSERT INTO unpartitioned_table SELECT * FROM source_table", - testSessionBuilder() + testingSessionBuilder() .setCatalog("mock_without_multiple_writer_per_partition") .setSchema("mock") .setSystemProperty(TASK_SCALE_WRITERS_ENABLED, "true") @@ -166,7 +167,7 @@ public void testLocalScaledUnpartitionedWriterDistribution() assertDistributedPlan( "INSERT INTO unpartitioned_table SELECT * FROM source_table", - testSessionBuilder() + testingSessionBuilder() .setCatalog("mock_without_multiple_writer_per_partition") .setSchema("mock") .setSystemProperty(TASK_SCALE_WRITERS_ENABLED, "false") @@ -186,7 +187,7 @@ public void testLocalScaledUnpartitionedWriterWithPerTaskScalingDisabled() { assertDistributedPlan( "INSERT INTO unpartitioned_table SELECT * FROM source_table", - testSessionBuilder() + testingSessionBuilder() .setCatalog("mock_without_scaled_writers") .setSchema("mock") .setSystemProperty(TASK_SCALE_WRITERS_ENABLED, "true") @@ -202,7 +203,7 @@ public void testLocalScaledUnpartitionedWriterWithPerTaskScalingDisabled() assertDistributedPlan( "INSERT INTO unpartitioned_table SELECT * FROM source_table", - testSessionBuilder() + testingSessionBuilder() .setCatalog("mock_without_scaled_writers") .setSchema("mock") .setSystemProperty(TASK_SCALE_WRITERS_ENABLED, "false") @@ -229,7 +230,7 @@ public void testLocalScaledPartitionedWriterWithoutSupportForMultipleWritersPerP assertDistributedPlan( "INSERT INTO connector_partitioned_table SELECT * FROM source_table", - testSessionBuilder() + testingSessionBuilder() .setCatalog(catalogName) .setSchema("mock") .setSystemProperty(TASK_SCALE_WRITERS_ENABLED, String.valueOf(taskScaleWritersEnabled)) @@ -257,7 +258,7 @@ public void testLocalScaledPartitionedWriterWithPerTaskScalingDisabled() assertDistributedPlan( "INSERT INTO connector_partitioned_table SELECT * FROM source_table", - testSessionBuilder() + testingSessionBuilder() .setCatalog(catalogName) .setSchema("mock") .setSystemProperty(TASK_SCALE_WRITERS_ENABLED, String.valueOf(taskScaleWritersEnabled)) @@ -278,7 +279,7 @@ public void testLocalScaledPartitionedWriterForSystemPartitioningWithEnforcedPre { assertDistributedPlan( "INSERT INTO system_partitioned_table SELECT * FROM source_table", - testSessionBuilder() + testingSessionBuilder() .setCatalog("mock_with_scaled_writers") .setSchema("mock") // Enforce preferred partitioning @@ -296,7 +297,7 @@ public void testLocalScaledPartitionedWriterForSystemPartitioningWithEnforcedPre assertDistributedPlan( "INSERT INTO system_partitioned_table SELECT * FROM source_table", - testSessionBuilder() + testingSessionBuilder() .setCatalog("mock_with_scaled_writers") .setSchema("mock") // Enforce preferred partitioning @@ -329,7 +330,7 @@ public void testLocalScaledPartitionedWriterForConnectorPartitioning() assertDistributedPlan( "INSERT INTO connector_partitioned_table SELECT * FROM source_table", - testSessionBuilder() + testingSessionBuilder() .setCatalog(catalogName) .setSchema("mock") .setSystemProperty(TASK_SCALE_WRITERS_ENABLED, "true") @@ -345,7 +346,7 @@ public void testLocalScaledPartitionedWriterForConnectorPartitioning() assertDistributedPlan( "INSERT INTO connector_partitioned_table SELECT * FROM source_table", - testSessionBuilder() + testingSessionBuilder() .setCatalog(catalogName) .setSchema("mock") .setSystemProperty(TASK_SCALE_WRITERS_ENABLED, "false") @@ -365,7 +366,7 @@ public void testLocalScaledPartitionedWriterWithEnforcedLocalPreferredPartitioni { assertDistributedPlan( "INSERT INTO system_partitioned_table SELECT * FROM source_table", - testSessionBuilder() + testingSessionBuilder() .setCatalog("mock_with_scaled_writers") .setSchema("mock") .setSystemProperty(TASK_SCALE_WRITERS_ENABLED, "true") @@ -381,7 +382,7 @@ public void testLocalScaledPartitionedWriterWithEnforcedLocalPreferredPartitioni assertDistributedPlan( "INSERT INTO system_partitioned_table SELECT * FROM source_table", - testSessionBuilder() + testingSessionBuilder() .setCatalog("mock_with_scaled_writers") .setSchema("mock") .setSystemProperty(TASK_SCALE_WRITERS_ENABLED, "false") @@ -401,7 +402,7 @@ public void testTableExecuteLocalScalingDisabledForPartitionedTable() { @Language("SQL") String query = "ALTER TABLE system_partitioned_table EXECUTE optimize(file_size_threshold => '10MB')"; - Session session = Session.builder(getPlanTester().getDefaultSession()) + Session session = testingSessionBuilder() .setCatalog("mock_with_scaled_writers") .setSchema("mock") .setSystemProperty(TASK_SCALE_WRITERS_ENABLED, "true") @@ -422,7 +423,7 @@ public void testTableExecuteLocalScalingDisabledForUnpartitionedTable() { @Language("SQL") String query = "ALTER TABLE unpartitioned_table EXECUTE optimize(file_size_threshold => '10MB')"; - Session session = Session.builder(getPlanTester().getDefaultSession()) + Session session = testingSessionBuilder() .setCatalog("mock_with_scaled_writers") .setSchema("mock") .setSystemProperty(TASK_SCALE_WRITERS_ENABLED, "true") @@ -437,4 +438,9 @@ public void testTableExecuteLocalScalingDisabledForUnpartitionedTable() exchange(REMOTE, REPARTITION, SCALED_WRITER_ROUND_ROBIN_DISTRIBUTION, node(TableScanNode.class)))))); } + + private SessionBuilder testingSessionBuilder() + { + return Session.builder(getPlanTester().getDefaultSession()); + } } From c1f579a909bd675a41cb9ee01834764edeb430f9 Mon Sep 17 00:00:00 2001 From: David Phillips Date: Tue, 17 Sep 2024 00:39:01 -0700 Subject: [PATCH 5/8] Use QueryRunner session in TestQueryManager --- .../java/io/trino/tests/TestQueryManager.java | 15 +++------------ 1 file changed, 3 insertions(+), 12 deletions(-) diff --git a/testing/trino-tests/src/test/java/io/trino/tests/TestQueryManager.java b/testing/trino-tests/src/test/java/io/trino/tests/TestQueryManager.java index 214e854c6a54..b5178d0692df 100644 --- a/testing/trino-tests/src/test/java/io/trino/tests/TestQueryManager.java +++ b/testing/trino-tests/src/test/java/io/trino/tests/TestQueryManager.java @@ -15,7 +15,6 @@ import io.opentelemetry.api.trace.Span; import io.trino.Session; -import io.trino.client.ClientCapabilities; import io.trino.dispatcher.DispatchManager; import io.trino.execution.QueryInfo; import io.trino.execution.QueryManager; @@ -31,18 +30,14 @@ import org.junit.jupiter.api.Timeout; import org.junit.jupiter.api.parallel.Execution; -import static com.google.common.collect.ImmutableSet.toImmutableSet; import static io.trino.SessionTestUtils.TEST_SESSION; import static io.trino.execution.QueryRunnerUtil.createQuery; import static io.trino.execution.QueryRunnerUtil.waitForQueryState; import static io.trino.execution.QueryState.FAILED; import static io.trino.execution.QueryState.RUNNING; -import static io.trino.plugin.tpch.TpchMetadata.TINY_SCHEMA_NAME; import static io.trino.spi.StandardErrorCode.EXCEEDED_CPU_LIMIT; import static io.trino.spi.StandardErrorCode.EXCEEDED_SCAN_LIMIT; import static io.trino.spi.StandardErrorCode.GENERIC_INTERNAL_ERROR; -import static io.trino.testing.TestingSession.testSessionBuilder; -import static java.util.Arrays.stream; import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Fail.fail; import static org.junit.jupiter.api.parallel.ExecutionMode.SAME_THREAD; @@ -110,7 +105,8 @@ public void testQueryScanExceeded() throws Exception { try (QueryRunner queryRunner = TpchQueryRunner.builder().addExtraProperty("query.max-scan-physical-bytes", "0B").build()) { - QueryId queryId = createQuery(queryRunner, TEST_SESSION, "SELECT * FROM system.runtime.nodes"); + Session session = queryRunner.getDefaultSession(); + QueryId queryId = createQuery(queryRunner, session, "SELECT * FROM system.runtime.nodes"); waitForQueryState(queryRunner, queryId, FAILED); QueryManager queryManager = queryRunner.getCoordinator().getQueryManager(); BasicQueryInfo queryInfo = queryManager.getQueryInfo(queryId); @@ -125,12 +121,7 @@ public void testQueryScanExceededSession() throws Exception { try (QueryRunner queryRunner = TpchQueryRunner.builder().build()) { - Session session = testSessionBuilder() - .setCatalog("tpch") - .setSchema(TINY_SCHEMA_NAME) - .setClientCapabilities(stream(ClientCapabilities.values()) - .map(ClientCapabilities::toString) - .collect(toImmutableSet())) + Session session = Session.builder(queryRunner.getDefaultSession()) .setSystemProperty("query_max_scan_physical_bytes", "0B") .build(); QueryId queryId = createQuery(queryRunner, session, "SELECT * FROM system.runtime.nodes"); From 7ffc4eb5c4d05efe8f49778109d65a5a4aa92985 Mon Sep 17 00:00:00 2001 From: David Phillips Date: Sat, 14 Sep 2024 17:02:05 -0700 Subject: [PATCH 6/8] Add ValuesNode constructor for zero rows --- .../planner/iterative/rule/EvaluateEmptyIntersect.java | 3 +-- .../sql/planner/iterative/rule/EvaluateZeroSample.java | 3 +-- .../rule/PushFilterThroughBoolOrAggregation.java | 2 +- .../iterative/rule/PushFilterThroughCountAggregation.java | 2 +- .../iterative/rule/PushPredicateIntoTableScan.java | 5 ++--- .../rule/PushPredicateThroughProjectIntoRowNumber.java | 2 +- .../rule/PushPredicateThroughProjectIntoWindow.java | 2 +- .../iterative/rule/PushdownFilterIntoRowNumber.java | 3 +-- .../planner/iterative/rule/PushdownFilterIntoWindow.java | 3 +-- .../planner/iterative/rule/RemoveEmptyExceptBranches.java | 2 +- .../planner/iterative/rule/RemoveEmptyUnionBranches.java | 2 +- .../iterative/rule/RemoveRedundantDistinctLimit.java | 2 +- .../sql/planner/iterative/rule/RemoveRedundantJoin.java | 7 +------ .../sql/planner/iterative/rule/RemoveRedundantLimit.java | 3 +-- .../sql/planner/iterative/rule/RemoveRedundantOffset.java | 3 +-- .../rule/RemoveRedundantPredicateAboveTableScan.java | 4 ++-- .../sql/planner/iterative/rule/RemoveRedundantSort.java | 3 +-- .../iterative/rule/RemoveRedundantTableFunction.java | 3 +-- .../sql/planner/iterative/rule/RemoveRedundantTopN.java | 3 +-- .../sql/planner/iterative/rule/RemoveRedundantWindow.java | 3 +-- .../sql/planner/iterative/rule/RemoveTrivialFilters.java | 3 +-- .../io/trino/sql/planner/optimizations/LimitPushDown.java | 5 +---- .../sql/planner/optimizations/WindowFilterPushDown.java | 4 ++-- .../main/java/io/trino/sql/planner/plan/ValuesNode.java | 8 ++++++++ .../planner/iterative/rule/TestEliminateCrossJoins.java | 3 +-- .../java/io/trino/sql/planner/plan/TestWindowNode.java | 5 +---- .../sql/planner/sanity/TestVerifyOnlyOneOutputNode.java | 6 ++---- 27 files changed, 38 insertions(+), 56 deletions(-) diff --git a/core/trino-main/src/main/java/io/trino/sql/planner/iterative/rule/EvaluateEmptyIntersect.java b/core/trino-main/src/main/java/io/trino/sql/planner/iterative/rule/EvaluateEmptyIntersect.java index 025dfe7659a9..1fd7a8948f00 100644 --- a/core/trino-main/src/main/java/io/trino/sql/planner/iterative/rule/EvaluateEmptyIntersect.java +++ b/core/trino-main/src/main/java/io/trino/sql/planner/iterative/rule/EvaluateEmptyIntersect.java @@ -13,7 +13,6 @@ */ package io.trino.sql.planner.iterative.rule; -import com.google.common.collect.ImmutableList; import io.trino.matching.Captures; import io.trino.matching.Pattern; import io.trino.sql.planner.iterative.Rule; @@ -43,7 +42,7 @@ public Result apply(IntersectNode node, Captures captures, Context context) boolean hasEmptyBranches = node.getSources().stream().anyMatch(source -> isEmpty(source, context.getLookup())); if (hasEmptyBranches) { - return Result.ofPlanNode(new ValuesNode(node.getId(), node.getOutputSymbols(), ImmutableList.of())); + return Result.ofPlanNode(new ValuesNode(node.getId(), node.getOutputSymbols())); } return Result.empty(); diff --git a/core/trino-main/src/main/java/io/trino/sql/planner/iterative/rule/EvaluateZeroSample.java b/core/trino-main/src/main/java/io/trino/sql/planner/iterative/rule/EvaluateZeroSample.java index 7936be4da4ac..f4818f1317e3 100644 --- a/core/trino-main/src/main/java/io/trino/sql/planner/iterative/rule/EvaluateZeroSample.java +++ b/core/trino-main/src/main/java/io/trino/sql/planner/iterative/rule/EvaluateZeroSample.java @@ -13,7 +13,6 @@ */ package io.trino.sql.planner.iterative.rule; -import com.google.common.collect.ImmutableList; import io.trino.matching.Captures; import io.trino.matching.Pattern; import io.trino.sql.planner.iterative.Rule; @@ -41,6 +40,6 @@ public Pattern getPattern() @Override public Result apply(SampleNode sample, Captures captures, Context context) { - return Result.ofPlanNode(new ValuesNode(sample.getId(), sample.getOutputSymbols(), ImmutableList.of())); + return Result.ofPlanNode(new ValuesNode(sample.getId(), sample.getOutputSymbols())); } } diff --git a/core/trino-main/src/main/java/io/trino/sql/planner/iterative/rule/PushFilterThroughBoolOrAggregation.java b/core/trino-main/src/main/java/io/trino/sql/planner/iterative/rule/PushFilterThroughBoolOrAggregation.java index 80470f0feb40..a9eba75e4da3 100644 --- a/core/trino-main/src/main/java/io/trino/sql/planner/iterative/rule/PushFilterThroughBoolOrAggregation.java +++ b/core/trino-main/src/main/java/io/trino/sql/planner/iterative/rule/PushFilterThroughBoolOrAggregation.java @@ -199,7 +199,7 @@ private static Result pushFilter(FilterNode filterNode, AggregationNode aggregat if (tupleDomain.isNone()) { // Filter predicate is never satisfied. Replace filter with empty values. - return Result.ofPlanNode(new ValuesNode(filterNode.getId(), filterNode.getOutputSymbols(), ImmutableList.of())); + return Result.ofPlanNode(new ValuesNode(filterNode.getId(), filterNode.getOutputSymbols())); } // boolOrSymbol (in remaining expressions) should only be used in Coalesce(boolOrSymbol, FALSE) expression diff --git a/core/trino-main/src/main/java/io/trino/sql/planner/iterative/rule/PushFilterThroughCountAggregation.java b/core/trino-main/src/main/java/io/trino/sql/planner/iterative/rule/PushFilterThroughCountAggregation.java index fce621ccd975..246b96b330eb 100644 --- a/core/trino-main/src/main/java/io/trino/sql/planner/iterative/rule/PushFilterThroughCountAggregation.java +++ b/core/trino-main/src/main/java/io/trino/sql/planner/iterative/rule/PushFilterThroughCountAggregation.java @@ -189,7 +189,7 @@ private static Result pushFilter(FilterNode filterNode, AggregationNode aggregat if (tupleDomain.isNone()) { // Filter predicate is never satisfied. Replace filter with empty values. - return Result.ofPlanNode(new ValuesNode(filterNode.getId(), filterNode.getOutputSymbols(), ImmutableList.of())); + return Result.ofPlanNode(new ValuesNode(filterNode.getId(), filterNode.getOutputSymbols())); } Domain countDomain = tupleDomain.getDomains().get().get(countSymbol); if (countDomain == null) { diff --git a/core/trino-main/src/main/java/io/trino/sql/planner/iterative/rule/PushPredicateIntoTableScan.java b/core/trino-main/src/main/java/io/trino/sql/planner/iterative/rule/PushPredicateIntoTableScan.java index 15980aabb40f..6b2ade04d662 100644 --- a/core/trino-main/src/main/java/io/trino/sql/planner/iterative/rule/PushPredicateIntoTableScan.java +++ b/core/trino-main/src/main/java/io/trino/sql/planner/iterative/rule/PushPredicateIntoTableScan.java @@ -14,7 +14,6 @@ package io.trino.sql.planner.iterative.rule; import com.google.common.collect.ImmutableBiMap; -import com.google.common.collect.ImmutableList; import com.google.common.collect.ImmutableMap; import io.trino.Session; import io.trino.cost.StatsProvider; @@ -217,7 +216,7 @@ public static Optional pushFilterIntoTableScan( // TODO: DomainTranslator.fromPredicate can infer that the expression is "false" in some cases (TupleDomain.none()). // This should move to another rule that simplifies the filter using that logic and then rely on RemoveTrivialFilters // to turn the subtree into a Values node - return Optional.of(new ValuesNode(node.getId(), node.getOutputSymbols(), ImmutableList.of())); + return Optional.of(new ValuesNode(node.getId(), node.getOutputSymbols())); } Optional> result = plannerContext.getMetadata().applyFilter(session, node.getTable(), constraint); @@ -231,7 +230,7 @@ public static Optional pushFilterIntoTableScan( TableProperties newTableProperties = plannerContext.getMetadata().getTableProperties(session, newTable); Optional newTablePartitioning = newTableProperties.getTablePartitioning(); if (newTableProperties.getPredicate().isNone()) { - return Optional.of(new ValuesNode(node.getId(), node.getOutputSymbols(), ImmutableList.of())); + return Optional.of(new ValuesNode(node.getId(), node.getOutputSymbols())); } TupleDomain remainingFilter = result.get().getRemainingFilter(); diff --git a/core/trino-main/src/main/java/io/trino/sql/planner/iterative/rule/PushPredicateThroughProjectIntoRowNumber.java b/core/trino-main/src/main/java/io/trino/sql/planner/iterative/rule/PushPredicateThroughProjectIntoRowNumber.java index d10c61ded9ff..76b11475c061 100644 --- a/core/trino-main/src/main/java/io/trino/sql/planner/iterative/rule/PushPredicateThroughProjectIntoRowNumber.java +++ b/core/trino-main/src/main/java/io/trino/sql/planner/iterative/rule/PushPredicateThroughProjectIntoRowNumber.java @@ -116,7 +116,7 @@ public Result apply(FilterNode filter, Captures captures, Context context) return Result.empty(); } if (upperBound.getAsInt() <= 0) { - return Result.ofPlanNode(new ValuesNode(filter.getId(), filter.getOutputSymbols(), ImmutableList.of())); + return Result.ofPlanNode(new ValuesNode(filter.getId(), filter.getOutputSymbols())); } boolean updatedMaxRowCountPerPartition = false; if (rowNumber.getMaxRowCountPerPartition().isEmpty() || rowNumber.getMaxRowCountPerPartition().get() > upperBound.getAsInt()) { diff --git a/core/trino-main/src/main/java/io/trino/sql/planner/iterative/rule/PushPredicateThroughProjectIntoWindow.java b/core/trino-main/src/main/java/io/trino/sql/planner/iterative/rule/PushPredicateThroughProjectIntoWindow.java index 119e81b6fae7..d4070f9ddfd4 100644 --- a/core/trino-main/src/main/java/io/trino/sql/planner/iterative/rule/PushPredicateThroughProjectIntoWindow.java +++ b/core/trino-main/src/main/java/io/trino/sql/planner/iterative/rule/PushPredicateThroughProjectIntoWindow.java @@ -128,7 +128,7 @@ public Result apply(FilterNode filter, Captures captures, Context context) return Result.empty(); } if (upperBound.getAsInt() <= 0) { - return Result.ofPlanNode(new ValuesNode(filter.getId(), filter.getOutputSymbols(), ImmutableList.of())); + return Result.ofPlanNode(new ValuesNode(filter.getId(), filter.getOutputSymbols())); } RankingType rankingType = toTopNRankingType(window).orElseThrow(); project = (ProjectNode) project.replaceChildren(ImmutableList.of(new TopNRankingNode( diff --git a/core/trino-main/src/main/java/io/trino/sql/planner/iterative/rule/PushdownFilterIntoRowNumber.java b/core/trino-main/src/main/java/io/trino/sql/planner/iterative/rule/PushdownFilterIntoRowNumber.java index 32b55785ea1d..dca2d24a2439 100644 --- a/core/trino-main/src/main/java/io/trino/sql/planner/iterative/rule/PushdownFilterIntoRowNumber.java +++ b/core/trino-main/src/main/java/io/trino/sql/planner/iterative/rule/PushdownFilterIntoRowNumber.java @@ -13,7 +13,6 @@ */ package io.trino.sql.planner.iterative.rule; -import com.google.common.collect.ImmutableList; import io.trino.Session; import io.trino.matching.Capture; import io.trino.matching.Captures; @@ -84,7 +83,7 @@ public Result apply(FilterNode node, Captures captures, Context context) } if (upperBound.getAsInt() <= 0) { - return Result.ofPlanNode(new ValuesNode(node.getId(), node.getOutputSymbols(), ImmutableList.of())); + return Result.ofPlanNode(new ValuesNode(node.getId(), node.getOutputSymbols())); } RowNumberNode merged = mergeLimit(source, upperBound.getAsInt()); diff --git a/core/trino-main/src/main/java/io/trino/sql/planner/iterative/rule/PushdownFilterIntoWindow.java b/core/trino-main/src/main/java/io/trino/sql/planner/iterative/rule/PushdownFilterIntoWindow.java index dcf085749914..cece80c57aa2 100644 --- a/core/trino-main/src/main/java/io/trino/sql/planner/iterative/rule/PushdownFilterIntoWindow.java +++ b/core/trino-main/src/main/java/io/trino/sql/planner/iterative/rule/PushdownFilterIntoWindow.java @@ -13,7 +13,6 @@ */ package io.trino.sql.planner.iterative.rule; -import com.google.common.collect.ImmutableList; import io.trino.Session; import io.trino.matching.Capture; import io.trino.matching.Captures; @@ -103,7 +102,7 @@ public Result apply(FilterNode node, Captures captures, Context context) } if (upperBound.getAsInt() <= 0) { - return Result.ofPlanNode(new ValuesNode(node.getId(), node.getOutputSymbols(), ImmutableList.of())); + return Result.ofPlanNode(new ValuesNode(node.getId(), node.getOutputSymbols())); } TopNRankingNode newSource = new TopNRankingNode( windowNode.getId(), diff --git a/core/trino-main/src/main/java/io/trino/sql/planner/iterative/rule/RemoveEmptyExceptBranches.java b/core/trino-main/src/main/java/io/trino/sql/planner/iterative/rule/RemoveEmptyExceptBranches.java index 05992c88aed7..e782907f0526 100644 --- a/core/trino-main/src/main/java/io/trino/sql/planner/iterative/rule/RemoveEmptyExceptBranches.java +++ b/core/trino-main/src/main/java/io/trino/sql/planner/iterative/rule/RemoveEmptyExceptBranches.java @@ -60,7 +60,7 @@ public Pattern getPattern() public Result apply(ExceptNode node, Captures captures, Context context) { if (isEmpty(node.getSources().get(0), context.getLookup())) { - return Result.ofPlanNode(new ValuesNode(node.getId(), node.getOutputSymbols(), ImmutableList.of())); + return Result.ofPlanNode(new ValuesNode(node.getId(), node.getOutputSymbols())); } boolean hasEmptyBranches = false; diff --git a/core/trino-main/src/main/java/io/trino/sql/planner/iterative/rule/RemoveEmptyUnionBranches.java b/core/trino-main/src/main/java/io/trino/sql/planner/iterative/rule/RemoveEmptyUnionBranches.java index 8d581f6927d8..7fdb58f95502 100644 --- a/core/trino-main/src/main/java/io/trino/sql/planner/iterative/rule/RemoveEmptyUnionBranches.java +++ b/core/trino-main/src/main/java/io/trino/sql/planner/iterative/rule/RemoveEmptyUnionBranches.java @@ -75,7 +75,7 @@ public Result apply(UnionNode node, Captures captures, Context context) } if (emptyBranches == node.getSources().size()) { - return Result.ofPlanNode(new ValuesNode(node.getId(), node.getOutputSymbols(), ImmutableList.of())); + return Result.ofPlanNode(new ValuesNode(node.getId(), node.getOutputSymbols())); } List newSources = newSourcesBuilder.build(); diff --git a/core/trino-main/src/main/java/io/trino/sql/planner/iterative/rule/RemoveRedundantDistinctLimit.java b/core/trino-main/src/main/java/io/trino/sql/planner/iterative/rule/RemoveRedundantDistinctLimit.java index 30c144df4894..5f10ea8fc45d 100644 --- a/core/trino-main/src/main/java/io/trino/sql/planner/iterative/rule/RemoveRedundantDistinctLimit.java +++ b/core/trino-main/src/main/java/io/trino/sql/planner/iterative/rule/RemoveRedundantDistinctLimit.java @@ -53,7 +53,7 @@ public Result apply(DistinctLimitNode node, Captures captures, Context context) { checkArgument(node.getHashSymbol().isEmpty(), "HashSymbol should be empty"); if (node.getLimit() == 0) { - return Result.ofPlanNode(new ValuesNode(node.getId(), node.getOutputSymbols(), ImmutableList.of())); + return Result.ofPlanNode(new ValuesNode(node.getId(), node.getOutputSymbols())); } Cardinality sourceCardinality = extractCardinality(node.getSource(), context.getLookup()); if (sourceCardinality.isScalar()) { diff --git a/core/trino-main/src/main/java/io/trino/sql/planner/iterative/rule/RemoveRedundantJoin.java b/core/trino-main/src/main/java/io/trino/sql/planner/iterative/rule/RemoveRedundantJoin.java index e3bfce610c1f..75a2e25b8f15 100644 --- a/core/trino-main/src/main/java/io/trino/sql/planner/iterative/rule/RemoveRedundantJoin.java +++ b/core/trino-main/src/main/java/io/trino/sql/planner/iterative/rule/RemoveRedundantJoin.java @@ -13,7 +13,6 @@ */ package io.trino.sql.planner.iterative.rule; -import com.google.common.collect.ImmutableList; import io.trino.matching.Captures; import io.trino.matching.Pattern; import io.trino.sql.planner.iterative.Lookup; @@ -40,11 +39,7 @@ public Pattern getPattern() public Result apply(JoinNode node, Captures captures, Context context) { if (canRemoveJoin(node, context.getLookup())) { - return Result.ofPlanNode( - new ValuesNode( - context.getIdAllocator().getNextId(), - node.getOutputSymbols(), - ImmutableList.of())); + return Result.ofPlanNode(new ValuesNode(context.getIdAllocator().getNextId(), node.getOutputSymbols())); } return Result.empty(); diff --git a/core/trino-main/src/main/java/io/trino/sql/planner/iterative/rule/RemoveRedundantLimit.java b/core/trino-main/src/main/java/io/trino/sql/planner/iterative/rule/RemoveRedundantLimit.java index b159e541907d..0c41f91148c4 100644 --- a/core/trino-main/src/main/java/io/trino/sql/planner/iterative/rule/RemoveRedundantLimit.java +++ b/core/trino-main/src/main/java/io/trino/sql/planner/iterative/rule/RemoveRedundantLimit.java @@ -13,7 +13,6 @@ */ package io.trino.sql.planner.iterative.rule; -import com.google.common.collect.ImmutableList; import io.trino.matching.Captures; import io.trino.matching.Pattern; import io.trino.sql.planner.iterative.Rule; @@ -43,7 +42,7 @@ public Pattern getPattern() public Result apply(LimitNode limit, Captures captures, Context context) { if (limit.getCount() == 0) { - return Result.ofPlanNode(new ValuesNode(limit.getId(), limit.getOutputSymbols(), ImmutableList.of())); + return Result.ofPlanNode(new ValuesNode(limit.getId(), limit.getOutputSymbols())); } if (isAtMost(limit.getSource(), context.getLookup(), limit.getCount())) { return Result.ofPlanNode(limit.getSource()); diff --git a/core/trino-main/src/main/java/io/trino/sql/planner/iterative/rule/RemoveRedundantOffset.java b/core/trino-main/src/main/java/io/trino/sql/planner/iterative/rule/RemoveRedundantOffset.java index acb904e96edc..5018808d5f5f 100644 --- a/core/trino-main/src/main/java/io/trino/sql/planner/iterative/rule/RemoveRedundantOffset.java +++ b/core/trino-main/src/main/java/io/trino/sql/planner/iterative/rule/RemoveRedundantOffset.java @@ -13,7 +13,6 @@ */ package io.trino.sql.planner.iterative.rule; -import com.google.common.collect.ImmutableList; import io.trino.matching.Captures; import io.trino.matching.Pattern; import io.trino.sql.planner.iterative.Rule; @@ -43,7 +42,7 @@ public Pattern getPattern() public Result apply(OffsetNode offset, Captures captures, Context context) { if (isAtMost(offset.getSource(), context.getLookup(), offset.getCount())) { - return Result.ofPlanNode(new ValuesNode(offset.getId(), offset.getOutputSymbols(), ImmutableList.of())); + return Result.ofPlanNode(new ValuesNode(offset.getId(), offset.getOutputSymbols())); } if (offset.getCount() == 0) { return Result.ofPlanNode(offset.getSource()); diff --git a/core/trino-main/src/main/java/io/trino/sql/planner/iterative/rule/RemoveRedundantPredicateAboveTableScan.java b/core/trino-main/src/main/java/io/trino/sql/planner/iterative/rule/RemoveRedundantPredicateAboveTableScan.java index a97ac1f656ef..67ce9b3b4704 100644 --- a/core/trino-main/src/main/java/io/trino/sql/planner/iterative/rule/RemoveRedundantPredicateAboveTableScan.java +++ b/core/trino-main/src/main/java/io/trino/sql/planner/iterative/rule/RemoveRedundantPredicateAboveTableScan.java @@ -106,12 +106,12 @@ public Result apply(FilterNode filterNode, Captures captures, Context context) // TODO: DomainTranslator.fromPredicate can infer that the expression is "false" in some cases (TupleDomain.none()). // This should move to another rule that simplifies the filter using that logic and then rely on RemoveTrivialFilters // to turn the subtree into a Values node - return Result.ofPlanNode(new ValuesNode(node.getId(), node.getOutputSymbols(), ImmutableList.of())); + return Result.ofPlanNode(new ValuesNode(node.getId(), node.getOutputSymbols())); } if (node.getEnforcedConstraint().isNone()) { // table scans with none domain should be converted to ValuesNode - return Result.ofPlanNode(new ValuesNode(node.getId(), node.getOutputSymbols(), ImmutableList.of())); + return Result.ofPlanNode(new ValuesNode(node.getId(), node.getOutputSymbols())); } Map enforcedColumnDomains = node.getEnforcedConstraint().getDomains().orElseThrow(); // is not NONE diff --git a/core/trino-main/src/main/java/io/trino/sql/planner/iterative/rule/RemoveRedundantSort.java b/core/trino-main/src/main/java/io/trino/sql/planner/iterative/rule/RemoveRedundantSort.java index 5ebfd6f0d8f3..58a087695a2d 100644 --- a/core/trino-main/src/main/java/io/trino/sql/planner/iterative/rule/RemoveRedundantSort.java +++ b/core/trino-main/src/main/java/io/trino/sql/planner/iterative/rule/RemoveRedundantSort.java @@ -13,7 +13,6 @@ */ package io.trino.sql.planner.iterative.rule; -import com.google.common.collect.ImmutableList; import io.trino.matching.Captures; import io.trino.matching.Pattern; import io.trino.sql.planner.iterative.Rule; @@ -40,7 +39,7 @@ public Result apply(SortNode node, Captures captures, Context context) { Cardinality cardinality = extractCardinality(node.getSource(), context.getLookup()); if (cardinality.isEmpty()) { - return Result.ofPlanNode(new ValuesNode(node.getId(), node.getOutputSymbols(), ImmutableList.of())); + return Result.ofPlanNode(new ValuesNode(node.getId(), node.getOutputSymbols())); } if (cardinality.isScalar()) { return Result.ofPlanNode(node.getSource()); diff --git a/core/trino-main/src/main/java/io/trino/sql/planner/iterative/rule/RemoveRedundantTableFunction.java b/core/trino-main/src/main/java/io/trino/sql/planner/iterative/rule/RemoveRedundantTableFunction.java index 3139437a97b4..6496cba70984 100644 --- a/core/trino-main/src/main/java/io/trino/sql/planner/iterative/rule/RemoveRedundantTableFunction.java +++ b/core/trino-main/src/main/java/io/trino/sql/planner/iterative/rule/RemoveRedundantTableFunction.java @@ -13,7 +13,6 @@ */ package io.trino.sql.planner.iterative.rule; -import com.google.common.collect.ImmutableList; import io.trino.matching.Captures; import io.trino.matching.Pattern; import io.trino.sql.planner.iterative.Rule; @@ -49,7 +48,7 @@ public Result apply(TableFunctionProcessorNode node, Captures captures, Context { if (node.isPruneWhenEmpty() && node.getSource().isPresent()) { if (isEmpty(node.getSource().orElseThrow(), context.getLookup())) { - return Result.ofPlanNode(new ValuesNode(node.getId(), node.getOutputSymbols(), ImmutableList.of())); + return Result.ofPlanNode(new ValuesNode(node.getId(), node.getOutputSymbols())); } } diff --git a/core/trino-main/src/main/java/io/trino/sql/planner/iterative/rule/RemoveRedundantTopN.java b/core/trino-main/src/main/java/io/trino/sql/planner/iterative/rule/RemoveRedundantTopN.java index 040aab2aea8a..c51f0ae4493a 100644 --- a/core/trino-main/src/main/java/io/trino/sql/planner/iterative/rule/RemoveRedundantTopN.java +++ b/core/trino-main/src/main/java/io/trino/sql/planner/iterative/rule/RemoveRedundantTopN.java @@ -13,7 +13,6 @@ */ package io.trino.sql.planner.iterative.rule; -import com.google.common.collect.ImmutableList; import io.trino.matching.Captures; import io.trino.matching.Pattern; import io.trino.sql.planner.iterative.Rule; @@ -46,7 +45,7 @@ public Pattern getPattern() public Result apply(TopNNode node, Captures captures, Context context) { if (node.getCount() == 0) { - return Result.ofPlanNode(new ValuesNode(node.getId(), node.getOutputSymbols(), ImmutableList.of())); + return Result.ofPlanNode(new ValuesNode(node.getId(), node.getOutputSymbols())); } Cardinality sourceCardinality = extractCardinality(node.getSource(), context.getLookup()); if (sourceCardinality.isScalar()) { diff --git a/core/trino-main/src/main/java/io/trino/sql/planner/iterative/rule/RemoveRedundantWindow.java b/core/trino-main/src/main/java/io/trino/sql/planner/iterative/rule/RemoveRedundantWindow.java index 121e5cf1572a..88eb3abfa23a 100644 --- a/core/trino-main/src/main/java/io/trino/sql/planner/iterative/rule/RemoveRedundantWindow.java +++ b/core/trino-main/src/main/java/io/trino/sql/planner/iterative/rule/RemoveRedundantWindow.java @@ -13,7 +13,6 @@ */ package io.trino.sql.planner.iterative.rule; -import com.google.common.collect.ImmutableList; import io.trino.matching.Captures; import io.trino.matching.Pattern; import io.trino.sql.planner.iterative.Rule; @@ -38,7 +37,7 @@ public Pattern getPattern() public Result apply(WindowNode window, Captures captures, Context context) { if (isEmpty(window.getSource(), context.getLookup())) { - return Result.ofPlanNode(new ValuesNode(window.getId(), window.getOutputSymbols(), ImmutableList.of())); + return Result.ofPlanNode(new ValuesNode(window.getId(), window.getOutputSymbols())); } return Result.empty(); } diff --git a/core/trino-main/src/main/java/io/trino/sql/planner/iterative/rule/RemoveTrivialFilters.java b/core/trino-main/src/main/java/io/trino/sql/planner/iterative/rule/RemoveTrivialFilters.java index 6805f4c0ca79..7f9e658bc267 100644 --- a/core/trino-main/src/main/java/io/trino/sql/planner/iterative/rule/RemoveTrivialFilters.java +++ b/core/trino-main/src/main/java/io/trino/sql/planner/iterative/rule/RemoveTrivialFilters.java @@ -24,7 +24,6 @@ import static io.trino.sql.ir.Booleans.FALSE; import static io.trino.sql.ir.Booleans.TRUE; import static io.trino.sql.planner.plan.Patterns.filter; -import static java.util.Collections.emptyList; public class RemoveTrivialFilters implements Rule @@ -48,7 +47,7 @@ public Result apply(FilterNode filterNode, Captures captures, Context context) if (predicate.equals(FALSE) || predicate instanceof Constant literal && literal.value() == null) { - return Result.ofPlanNode(new ValuesNode(context.getIdAllocator().getNextId(), filterNode.getOutputSymbols(), emptyList())); + return Result.ofPlanNode(new ValuesNode(context.getIdAllocator().getNextId(), filterNode.getOutputSymbols())); } return Result.empty(); diff --git a/core/trino-main/src/main/java/io/trino/sql/planner/optimizations/LimitPushDown.java b/core/trino-main/src/main/java/io/trino/sql/planner/optimizations/LimitPushDown.java index c8ab4b56a06a..a12b8b699c38 100644 --- a/core/trino-main/src/main/java/io/trino/sql/planner/optimizations/LimitPushDown.java +++ b/core/trino-main/src/main/java/io/trino/sql/planner/optimizations/LimitPushDown.java @@ -13,7 +13,6 @@ */ package io.trino.sql.planner.optimizations; -import com.google.common.collect.ImmutableList; import io.trino.sql.planner.PlanNodeIdAllocator; import io.trino.sql.planner.plan.AggregationNode; import io.trino.sql.planner.plan.DistinctLimitNode; @@ -110,9 +109,7 @@ public PlanNode visitLimit(LimitNode node, RewriteContext context) // return empty ValuesNode in case of limit 0 if (count == 0) { - return new ValuesNode(idAllocator.getNextId(), - node.getOutputSymbols(), - ImmutableList.of()); + return new ValuesNode(idAllocator.getNextId(), node.getOutputSymbols()); } if (!node.requiresPreSortedInputs() && (!node.isWithTies() || (limit != null && node.getCount() >= limit.getCount()))) { diff --git a/core/trino-main/src/main/java/io/trino/sql/planner/optimizations/WindowFilterPushDown.java b/core/trino-main/src/main/java/io/trino/sql/planner/optimizations/WindowFilterPushDown.java index 2d2d8e32fe71..111201b4e9be 100644 --- a/core/trino-main/src/main/java/io/trino/sql/planner/optimizations/WindowFilterPushDown.java +++ b/core/trino-main/src/main/java/io/trino/sql/planner/optimizations/WindowFilterPushDown.java @@ -161,7 +161,7 @@ public PlanNode visitFilter(FilterNode node, RewriteContext context) if (upperBound.isPresent()) { if (upperBound.getAsInt() <= 0) { - return new ValuesNode(node.getId(), node.getOutputSymbols(), ImmutableList.of()); + return new ValuesNode(node.getId(), node.getOutputSymbols()); } source = mergeLimit((RowNumberNode) source, upperBound.getAsInt()); return rewriteFilterSource(node, source, rowNumberSymbol, ((RowNumberNode) source).getMaxRowCountPerPartition().get()); @@ -175,7 +175,7 @@ else if (source instanceof WindowNode windowNode && isOptimizeTopNRanking(sessio if (upperBound.isPresent()) { if (upperBound.getAsInt() <= 0) { - return new ValuesNode(node.getId(), node.getOutputSymbols(), ImmutableList.of()); + return new ValuesNode(node.getId(), node.getOutputSymbols()); } source = convertToTopNRanking(windowNode, rankingType.get(), upperBound.getAsInt()); return rewriteFilterSource(node, source, rankingSymbol, upperBound.getAsInt()); diff --git a/core/trino-main/src/main/java/io/trino/sql/planner/plan/ValuesNode.java b/core/trino-main/src/main/java/io/trino/sql/planner/plan/ValuesNode.java index 27330c6c2faa..73caec8f0a3a 100644 --- a/core/trino-main/src/main/java/io/trino/sql/planner/plan/ValuesNode.java +++ b/core/trino-main/src/main/java/io/trino/sql/planner/plan/ValuesNode.java @@ -50,6 +50,14 @@ public ValuesNode(PlanNodeId id, List outputSymbols, List ro this(id, outputSymbols, rows.size(), Optional.of(rows)); } + /** + * Constructor of ValuesNode with non-empty output symbols list and zero rows + */ + public ValuesNode(PlanNodeId id, List outputSymbols) + { + this(id, outputSymbols, 0, Optional.of(ImmutableList.of())); + } + /** * Constructor of ValuesNode with empty output symbols list */ diff --git a/core/trino-main/src/test/java/io/trino/sql/planner/iterative/rule/TestEliminateCrossJoins.java b/core/trino-main/src/test/java/io/trino/sql/planner/iterative/rule/TestEliminateCrossJoins.java index 5db70ebc540d..191f911b5700 100644 --- a/core/trino-main/src/test/java/io/trino/sql/planner/iterative/rule/TestEliminateCrossJoins.java +++ b/core/trino-main/src/test/java/io/trino/sql/planner/iterative/rule/TestEliminateCrossJoins.java @@ -351,7 +351,6 @@ private ValuesNode values(String... symbols) { return new ValuesNode( idAllocator.getNextId(), - Arrays.stream(symbols).map(name -> new Symbol(BIGINT, name)).collect(toImmutableList()), - ImmutableList.of()); + Arrays.stream(symbols).map(name -> new Symbol(BIGINT, name)).collect(toImmutableList())); } } diff --git a/core/trino-main/src/test/java/io/trino/sql/planner/plan/TestWindowNode.java b/core/trino-main/src/test/java/io/trino/sql/planner/plan/TestWindowNode.java index 8570763897c2..ae9209f87a80 100644 --- a/core/trino-main/src/test/java/io/trino/sql/planner/plan/TestWindowNode.java +++ b/core/trino-main/src/test/java/io/trino/sql/planner/plan/TestWindowNode.java @@ -74,10 +74,7 @@ public void testSerializationRoundtrip() Symbol columnB = symbolAllocator.newSymbol("b", BIGINT); Symbol columnC = symbolAllocator.newSymbol("c", BIGINT); - ValuesNode sourceNode = new ValuesNode( - newId(), - ImmutableList.of(columnA, columnB, columnC), - ImmutableList.of()); + ValuesNode sourceNode = new ValuesNode(newId(), ImmutableList.of(columnA, columnB, columnC)); Symbol windowSymbol = symbolAllocator.newSymbol("sum", BIGINT); ResolvedFunction resolvedFunction = functionResolution.resolveFunction("sum", fromTypes(BIGINT)); diff --git a/core/trino-main/src/test/java/io/trino/sql/planner/sanity/TestVerifyOnlyOneOutputNode.java b/core/trino-main/src/test/java/io/trino/sql/planner/sanity/TestVerifyOnlyOneOutputNode.java index f4dd8fa4962c..3ef8d6503729 100644 --- a/core/trino-main/src/test/java/io/trino/sql/planner/sanity/TestVerifyOnlyOneOutputNode.java +++ b/core/trino-main/src/test/java/io/trino/sql/planner/sanity/TestVerifyOnlyOneOutputNode.java @@ -40,8 +40,7 @@ public void testValidateSuccessful() PlanNode root = new OutputNode(idAllocator.getNextId(), new ProjectNode(idAllocator.getNextId(), - new ValuesNode( - idAllocator.getNextId(), ImmutableList.of(), ImmutableList.of()), + new ValuesNode(idAllocator.getNextId(), ImmutableList.of()), Assignments.of() ), ImmutableList.of(), ImmutableList.of()); new VerifyOnlyOneOutputNode().validate(root, null, PLANNER_CONTEXT, WarningCollector.NOOP); @@ -56,8 +55,7 @@ public void testValidateFailed() new ExplainAnalyzeNode(idAllocator.getNextId(), new OutputNode(idAllocator.getNextId(), new ProjectNode(idAllocator.getNextId(), - new ValuesNode( - idAllocator.getNextId(), ImmutableList.of(), ImmutableList.of()), + new ValuesNode(idAllocator.getNextId(), ImmutableList.of()), Assignments.of() ), ImmutableList.of(), ImmutableList.of() ), new Symbol(UNKNOWN, "a"), From 35f5d2d5740c095ae40d4235050bb5ac5538d649 Mon Sep 17 00:00:00 2001 From: David Phillips Date: Fri, 13 Sep 2024 11:49:16 -0700 Subject: [PATCH 7/8] Limit number of Values node rows in printed plan --- .../io/trino/sql/planner/planprinter/PlanPrinter.java | 10 ++++++++-- 1 file changed, 8 insertions(+), 2 deletions(-) diff --git a/core/trino-main/src/main/java/io/trino/sql/planner/planprinter/PlanPrinter.java b/core/trino-main/src/main/java/io/trino/sql/planner/planprinter/PlanPrinter.java index 7e905fb679b6..359ec00e9a32 100644 --- a/core/trino-main/src/main/java/io/trino/sql/planner/planprinter/PlanPrinter.java +++ b/core/trino-main/src/main/java/io/trino/sql/planner/planprinter/PlanPrinter.java @@ -181,6 +181,7 @@ import static java.util.concurrent.TimeUnit.MILLISECONDS; import static java.util.function.Function.identity; import static java.util.stream.Collectors.joining; +import static java.util.stream.Collectors.toCollection; import static java.util.stream.Collectors.toList; public class PlanPrinter @@ -1148,7 +1149,8 @@ public Void visitValues(ValuesNode node, Context context) } return null; } - List rows = node.getRows().get().stream() + List nodeRows = node.getRows().get(); + List rows = nodeRows.stream() .map(row -> { if (row instanceof Row) { return ((Row) row).items().stream() @@ -1157,7 +1159,11 @@ public Void visitValues(ValuesNode node, Context context) } return anonymizer.anonymize(row); }) - .collect(toImmutableList()); + .limit(11) + .collect(toCollection(ArrayList::new)); + if (nodeRows.size() > 11) { + rows.set(10, "(... %s more rows ...)".formatted(nodeRows.size() - 10)); + } for (String row : rows) { nodeOutput.appendDetails("%s", row); } From 27f3691eb3b0d8f9b1d994bd75efd768fe11f530 Mon Sep 17 00:00:00 2001 From: David Phillips Date: Fri, 13 Sep 2024 16:46:38 -0700 Subject: [PATCH 8/8] Materialize tables during planning --- .../trino/jdbc/TestTrinoDatabaseMetaData.java | 4 +- .../io/trino/SystemSessionProperties.java | 44 ++++ .../io/trino/server/CoordinatorModule.java | 4 + .../java/io/trino/sql/planner/ForPlanner.java | 29 +++ .../io/trino/sql/planner/OptimizerConfig.java | 61 ++++- .../io/trino/sql/planner/PlanOptimizers.java | 14 ++ .../AbstractMaterializeTableScan.java | 217 ++++++++++++++++++ .../MaterializeFilteredTableScan.java | 86 +++++++ .../optimizations/MaterializeTableScan.java | 60 +++++ .../java/io/trino/testing/PlanTester.java | 12 + .../trino/testing/StandaloneQueryRunner.java | 8 +- .../io/trino/cost/TestOptimizerConfig.java | 14 +- .../TestTableScanNodePartitioning.java | 3 + .../planprinter/TestJsonRepresentation.java | 2 +- .../EnvSinglenodeCompatibility.java | 3 + .../common/hadoop-kerberos/config.properties | 1 + .../multinode-master-config.properties | 1 + .../common/standard/config.properties | 1 + .../multinode-master-config.properties | 1 + .../coordinator-config.properties | 1 + .../config-master.properties | 1 + .../config-master.properties | 1 + .../multinode-tls/config-master.properties | 1 + .../config-pre458.properties | 7 + .../config.properties | 1 + .../config.properties | 1 + .../singlenode-ldap/config.properties | 1 + .../trino/config.properties | 1 + .../trino/config.properties | 1 + .../config.properties | 1 + .../config.properties | 1 + .../config.properties | 1 + .../singlenode-oauth2/config.properties | 1 + .../singlenode-oidc-refresh/config.properties | 1 + .../singlenode-oidc/config.properties | 1 + .../trino/testing/DistributedQueryRunner.java | 6 +- .../execution/TestEventListenerBasic.java | 8 +- 37 files changed, 590 insertions(+), 11 deletions(-) create mode 100644 core/trino-main/src/main/java/io/trino/sql/planner/ForPlanner.java create mode 100644 core/trino-main/src/main/java/io/trino/sql/planner/optimizations/AbstractMaterializeTableScan.java create mode 100644 core/trino-main/src/main/java/io/trino/sql/planner/optimizations/MaterializeFilteredTableScan.java create mode 100644 core/trino-main/src/main/java/io/trino/sql/planner/optimizations/MaterializeTableScan.java create mode 100644 testing/trino-product-tests-launcher/src/main/resources/docker/presto-product-tests/conf/environment/singlenode-compatibility/config-pre458.properties diff --git a/client/trino-jdbc/src/test/java/io/trino/jdbc/TestTrinoDatabaseMetaData.java b/client/trino-jdbc/src/test/java/io/trino/jdbc/TestTrinoDatabaseMetaData.java index 3dd9e36431b0..96c5c8d7ef42 100644 --- a/client/trino-jdbc/src/test/java/io/trino/jdbc/TestTrinoDatabaseMetaData.java +++ b/client/trino-jdbc/src/test/java/io/trino/jdbc/TestTrinoDatabaseMetaData.java @@ -107,7 +107,9 @@ public void setupServer() throws Exception { Logging.initialize(); - server = TestingTrinoServer.create(); + server = TestingTrinoServer.builder() + .addProperty("optimizer.materialize-table.enabled", "false") + .build(); server.installPlugin(new TpchPlugin()); server.createCatalog(TEST_CATALOG, "tpch"); diff --git a/core/trino-main/src/main/java/io/trino/SystemSessionProperties.java b/core/trino-main/src/main/java/io/trino/SystemSessionProperties.java index 4e4f94036880..63b902447df3 100644 --- a/core/trino-main/src/main/java/io/trino/SystemSessionProperties.java +++ b/core/trino-main/src/main/java/io/trino/SystemSessionProperties.java @@ -214,6 +214,10 @@ public final class SystemSessionProperties public static final String USE_EXACT_PARTITIONING = "use_exact_partitioning"; public static final String USE_COST_BASED_PARTITIONING = "use_cost_based_partitioning"; public static final String PUSH_FILTER_INTO_VALUES_MAX_ROW_COUNT = "push_filter_into_values_max_row_count"; + public static final String MATERIALIZE_TABLE_ENABLED = "materialize_table_enabled"; + public static final String MATERIALIZE_TABLE_MAX_ESTIMATED_ROW_COUNT = "materialize_table_max_estimated_row_count"; + public static final String MATERIALIZE_TABLE_MAX_ACTUAL_ROW_COUNT = "materialize_table_max_actual_row_count"; + public static final String MATERIALIZE_TABLE_TIMEOUT = "materialize_table_timeout"; public static final String FORCE_SPILLING_JOIN = "force_spilling_join"; public static final String PAGE_PARTITIONING_BUFFER_POOL_SIZE = "page_partitioning_buffer_pool_size"; public static final String IDLE_WRITER_MIN_DATA_SIZE_THRESHOLD = "idle_writer_min_data_size_threshold"; @@ -1102,6 +1106,26 @@ public SystemSessionProperties( "Maximum number of rows in values for which filter is pushed down into values", optimizerConfig.getPushFilterIntoValuesMaxRowCount(), false), + booleanProperty( + MATERIALIZE_TABLE_ENABLED, + "Materialize tables during planning", + optimizerConfig.isMaterializeTable(), + false), + integerProperty( + MATERIALIZE_TABLE_MAX_ESTIMATED_ROW_COUNT, + "Maximum estimated row count for a table to be materialized", + optimizerConfig.getMaterializeTableMaxEstimatedRowCount(), + false), + integerProperty( + MATERIALIZE_TABLE_MAX_ACTUAL_ROW_COUNT, + "Maximum actual row count for a materialized table", + optimizerConfig.getMaterializeTableMaxActualRowCount(), + false), + durationProperty( + MATERIALIZE_TABLE_TIMEOUT, + "Maximum time to wait for materializing a table", + optimizerConfig.getMaterializeTableTimeout(), + false), booleanProperty( FORCE_SPILLING_JOIN, "Force the usage of spliing join operator in favor of the non-spilling one, even if spill is not enabled", @@ -1993,6 +2017,26 @@ public static int getPushFilterIntoValuesMaxRowCount(Session session) return session.getSystemProperty(PUSH_FILTER_INTO_VALUES_MAX_ROW_COUNT, Integer.class); } + public static boolean isMaterializeTableEnabled(Session session) + { + return session.getSystemProperty(MATERIALIZE_TABLE_ENABLED, Boolean.class); + } + + public static int getMaterializeTableMaxEstimatedRowCount(Session session) + { + return session.getSystemProperty(MATERIALIZE_TABLE_MAX_ESTIMATED_ROW_COUNT, Integer.class); + } + + public static int getMaterializeTableMaxActualRowCount(Session session) + { + return session.getSystemProperty(MATERIALIZE_TABLE_MAX_ACTUAL_ROW_COUNT, Integer.class); + } + + public static Duration getMaterializeTableTimeout(Session session) + { + return session.getSystemProperty(MATERIALIZE_TABLE_TIMEOUT, Duration.class); + } + public static boolean isForceSpillingOperator(Session session) { return session.getSystemProperty(FORCE_SPILLING_JOIN, Boolean.class); diff --git a/core/trino-main/src/main/java/io/trino/server/CoordinatorModule.java b/core/trino-main/src/main/java/io/trino/server/CoordinatorModule.java index 1820818ace22..f1e9f6f51817 100644 --- a/core/trino-main/src/main/java/io/trino/server/CoordinatorModule.java +++ b/core/trino-main/src/main/java/io/trino/server/CoordinatorModule.java @@ -114,6 +114,7 @@ import io.trino.sql.PlannerContext; import io.trino.sql.analyzer.AnalyzerFactory; import io.trino.sql.analyzer.QueryExplainerFactory; +import io.trino.sql.planner.ForPlanner; import io.trino.sql.planner.OptimizerStatsMBeanExporter; import io.trino.sql.planner.PlanFragmenter; import io.trino.sql.planner.PlanOptimizers; @@ -307,6 +308,8 @@ List getCompositeOutputDataSizeEstimatorDelegateFac // planner binder.bind(PlanFragmenter.class).in(Scopes.SINGLETON); binder.bind(PlanOptimizersFactory.class).to(PlanOptimizers.class).in(Scopes.SINGLETON); + binder.bind(ExecutorService.class).annotatedWith(ForPlanner.class) + .toInstance(newCachedThreadPool(threadsNamed("planner-%s"))); // Optimizer/Rule Stats exporter binder.bind(RuleStatsRecorder.class).in(Scopes.SINGLETON); @@ -372,6 +375,7 @@ List getCompositeOutputDataSizeEstimatorDelegateFac closingBinder(binder).registerExecutor(Key.get(ScheduledExecutorService.class, ForStatementResource.class)); closingBinder(binder).registerExecutor(Key.get(ExecutorService.class, ForQueryExecution.class)); closingBinder(binder).registerExecutor(Key.get(ScheduledExecutorService.class, ForScheduler.class)); + closingBinder(binder).registerExecutor(Key.get(ExecutorService.class, ForPlanner.class)); } // working around circular dependency Metadata <-> PlannerContext diff --git a/core/trino-main/src/main/java/io/trino/sql/planner/ForPlanner.java b/core/trino-main/src/main/java/io/trino/sql/planner/ForPlanner.java new file mode 100644 index 000000000000..f124305d0cc3 --- /dev/null +++ b/core/trino-main/src/main/java/io/trino/sql/planner/ForPlanner.java @@ -0,0 +1,29 @@ +/* + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.trino.sql.planner; + +import com.google.inject.BindingAnnotation; + +import java.lang.annotation.Retention; +import java.lang.annotation.Target; + +import static java.lang.annotation.ElementType.FIELD; +import static java.lang.annotation.ElementType.METHOD; +import static java.lang.annotation.ElementType.PARAMETER; +import static java.lang.annotation.RetentionPolicy.RUNTIME; + +@Retention(RUNTIME) +@Target({FIELD, PARAMETER, METHOD}) +@BindingAnnotation +public @interface ForPlanner {} diff --git a/core/trino-main/src/main/java/io/trino/sql/planner/OptimizerConfig.java b/core/trino-main/src/main/java/io/trino/sql/planner/OptimizerConfig.java index 54f0bceedb3d..916fa3a9e979 100644 --- a/core/trino-main/src/main/java/io/trino/sql/planner/OptimizerConfig.java +++ b/core/trino-main/src/main/java/io/trino/sql/planner/OptimizerConfig.java @@ -29,6 +29,7 @@ import static io.airlift.units.DataSize.Unit.MEGABYTE; import static java.util.Objects.requireNonNull; import static java.util.concurrent.TimeUnit.MINUTES; +import static java.util.concurrent.TimeUnit.SECONDS; @DefunctConfig({ "adaptive-partial-aggregation.min-rows", @@ -92,7 +93,11 @@ public class OptimizerConfig private boolean forceSingleNodeOutput; private boolean useExactPartitioning; private boolean useCostBasedPartitioning = true; - private int pushFilterIntoValuesMaxRowCount = 100; + private int pushFilterIntoValuesMaxRowCount = 100_000; + private boolean materializeTable = true; + private int materializeTableMaxEstimatedRowCount = 50_000; + private int materializeTableMaxActualRowCount = 100_000; + private Duration materializeTableTimeout = new Duration(5, SECONDS); // adaptive partial aggregation private boolean adaptivePartialAggregationEnabled = true; private double adaptivePartialAggregationUniqueRowsRatioThreshold = 0.8; @@ -823,6 +828,60 @@ public OptimizerConfig setPushFilterIntoValuesMaxRowCount(int pushFilterIntoValu return this; } + public boolean isMaterializeTable() + { + return materializeTable; + } + + @Config("optimizer.materialize-table.enabled") + @ConfigDescription("Materialize tables during planning") + public OptimizerConfig setMaterializeTable(boolean materializeTable) + { + this.materializeTable = materializeTable; + return this; + } + + @Min(0) + public int getMaterializeTableMaxEstimatedRowCount() + { + return materializeTableMaxEstimatedRowCount; + } + + @Config("optimizer.materialize-table.max-estimated-row-count") + @ConfigDescription("Maximum estimated row count for a table to be materialized") + public OptimizerConfig setMaterializeTableMaxEstimatedRowCount(int materializeTableMaxEstimatedRowCount) + { + this.materializeTableMaxEstimatedRowCount = materializeTableMaxEstimatedRowCount; + return this; + } + + @Min(0) + public int getMaterializeTableMaxActualRowCount() + { + return materializeTableMaxActualRowCount; + } + + @Config("optimizer.materialize-table.max-actual-row-count") + @ConfigDescription("Maximum actual row count for a materialized table") + public OptimizerConfig setMaterializeTableMaxActualRowCount(int materializeTableMaxActualRowCount) + { + this.materializeTableMaxActualRowCount = materializeTableMaxActualRowCount; + return this; + } + + public Duration getMaterializeTableTimeout() + { + return materializeTableTimeout; + } + + @Config("optimizer.materialize-table.timeout") + @ConfigDescription("Maximum time to wait for materializing a table") + public OptimizerConfig setMaterializeTableTimeout(Duration materializeTableTimeout) + { + this.materializeTableTimeout = materializeTableTimeout; + return this; + } + public boolean isUnsafePushdownAllowed() { return allowUnsafePushdown; diff --git a/core/trino-main/src/main/java/io/trino/sql/planner/PlanOptimizers.java b/core/trino-main/src/main/java/io/trino/sql/planner/PlanOptimizers.java index d5528f541fb2..747a3a333d71 100644 --- a/core/trino-main/src/main/java/io/trino/sql/planner/PlanOptimizers.java +++ b/core/trino-main/src/main/java/io/trino/sql/planner/PlanOptimizers.java @@ -25,6 +25,7 @@ import io.trino.cost.StatsCalculator; import io.trino.cost.TaskCountEstimator; import io.trino.execution.TaskManagerConfig; +import io.trino.metadata.InternalNodeManager; import io.trino.metadata.Metadata; import io.trino.split.PageSourceManager; import io.trino.split.SplitManager; @@ -255,6 +256,8 @@ import io.trino.sql.planner.optimizations.HashGenerationOptimizer; import io.trino.sql.planner.optimizations.IndexJoinOptimizer; import io.trino.sql.planner.optimizations.LimitPushDown; +import io.trino.sql.planner.optimizations.MaterializeFilteredTableScan; +import io.trino.sql.planner.optimizations.MaterializeTableScan; import io.trino.sql.planner.optimizations.MetadataQueryOptimizer; import io.trino.sql.planner.optimizations.OptimizerStats; import io.trino.sql.planner.optimizations.PlanOptimizer; @@ -267,6 +270,7 @@ import java.util.List; import java.util.Map; import java.util.Set; +import java.util.concurrent.ExecutorService; import static java.util.Objects.requireNonNull; @@ -281,9 +285,11 @@ public class PlanOptimizers @Inject public PlanOptimizers( PlannerContext plannerContext, + @ForPlanner ExecutorService executor, TaskManagerConfig taskManagerConfig, SplitManager splitManager, PageSourceManager pageSourceManager, + InternalNodeManager nodeManager, StatsCalculator statsCalculator, ScalarStatsCalculator scalarStatsCalculator, CostCalculator costCalculatorWithoutEstimatedExchanges, @@ -294,10 +300,12 @@ public PlanOptimizers( RuleStatsRecorder ruleStats) { this(plannerContext, + executor, taskManagerConfig, false, splitManager, pageSourceManager, + nodeManager, statsCalculator, scalarStatsCalculator, costCalculatorWithoutEstimatedExchanges, @@ -310,10 +318,12 @@ public PlanOptimizers( public PlanOptimizers( PlannerContext plannerContext, + ExecutorService executor, TaskManagerConfig taskManagerConfig, boolean forceSingleNode, SplitManager splitManager, PageSourceManager pageSourceManager, + InternalNodeManager nodeManager, StatsCalculator statsCalculator, ScalarStatsCalculator scalarStatsCalculator, CostCalculator costCalculatorWithoutEstimatedExchanges, @@ -644,6 +654,10 @@ public PlanOptimizers( .add(new PushDistinctLimitIntoTableScan(plannerContext)) .add(new PushTopNIntoTableScan(metadata)) .add(new RewriteTableFunctionToTableScan(plannerContext)) // must run after ImplementTableFunctionSource + .add(new MaterializeFilteredTableScan(plannerContext, splitManager, pageSourceManager, nodeManager, executor)) + .add(new MaterializeTableScan(plannerContext, splitManager, pageSourceManager, nodeManager, executor)) + .add(new PushFilterIntoValues(plannerContext)) + .add(new ReplaceJoinOverConstantWithProject()) .build(); IterativeOptimizer pushIntoTableScanOptimizer = new IterativeOptimizer( plannerContext, diff --git a/core/trino-main/src/main/java/io/trino/sql/planner/optimizations/AbstractMaterializeTableScan.java b/core/trino-main/src/main/java/io/trino/sql/planner/optimizations/AbstractMaterializeTableScan.java new file mode 100644 index 000000000000..02f95723577b --- /dev/null +++ b/core/trino-main/src/main/java/io/trino/sql/planner/optimizations/AbstractMaterializeTableScan.java @@ -0,0 +1,217 @@ +/* + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.trino.sql.planner.optimizations; + +import com.google.common.collect.ImmutableList; +import io.airlift.concurrent.SetThreadName; +import io.airlift.log.Logger; +import io.airlift.units.Duration; +import io.trino.Session; +import io.trino.cost.PlanNodeStatsEstimate; +import io.trino.matching.Captures; +import io.trino.metadata.InternalNodeManager; +import io.trino.metadata.Split; +import io.trino.metadata.TableHandle; +import io.trino.spi.HostAddress; +import io.trino.spi.Page; +import io.trino.spi.block.Block; +import io.trino.spi.connector.ColumnHandle; +import io.trino.spi.connector.ConnectorPageSource; +import io.trino.spi.connector.Constraint; +import io.trino.spi.connector.DynamicFilter; +import io.trino.spi.type.Type; +import io.trino.split.PageSourceManager; +import io.trino.split.PageSourceProvider; +import io.trino.split.SplitManager; +import io.trino.split.SplitSource; +import io.trino.sql.PlannerContext; +import io.trino.sql.ir.Constant; +import io.trino.sql.ir.Expression; +import io.trino.sql.ir.Row; +import io.trino.sql.planner.Symbol; +import io.trino.sql.planner.iterative.Rule; +import io.trino.sql.planner.plan.PlanNode; +import io.trino.sql.planner.plan.TableScanNode; + +import java.io.IOException; +import java.io.InterruptedIOException; +import java.util.ArrayList; +import java.util.List; +import java.util.Optional; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Future; +import java.util.concurrent.TimeoutException; + +import static io.airlift.concurrent.MoreFutures.getFutureValue; +import static io.trino.SystemSessionProperties.getMaterializeTableMaxActualRowCount; +import static io.trino.SystemSessionProperties.getMaterializeTableMaxEstimatedRowCount; +import static io.trino.SystemSessionProperties.getMaterializeTableTimeout; +import static io.trino.SystemSessionProperties.isMaterializeTableEnabled; +import static io.trino.spi.type.TypeUtils.readNativeValue; +import static java.util.Objects.requireNonNull; +import static java.util.concurrent.TimeUnit.MILLISECONDS; + +public abstract sealed class AbstractMaterializeTableScan + implements Rule + permits MaterializeFilteredTableScan, MaterializeTableScan +{ + private static final int MAX_SPLITS = 10_000; + + protected final Logger log = Logger.get(getClass()); + protected final PlannerContext plannerContext; + private final SplitManager splitManager; + private final PageSourceManager pageSourceManager; + private final ExecutorService executor; + private final HostAddress currentNode; + + protected AbstractMaterializeTableScan( + PlannerContext plannerContext, + SplitManager splitManager, + PageSourceManager pageSourceManager, + InternalNodeManager nodeManager, + ExecutorService executor) + { + this.plannerContext = requireNonNull(plannerContext, "plannerContext is null"); + this.splitManager = requireNonNull(splitManager, "splitManager is null"); + this.pageSourceManager = requireNonNull(pageSourceManager, "pageSourceManager is null"); + this.executor = requireNonNull(executor, "executor is null"); + this.currentNode = nodeManager.getCurrentNode().getHostAndPort(); + } + + @Override + public final boolean isEnabled(Session session) + { + return isMaterializeTableEnabled(session) && + getMaterializeTableMaxEstimatedRowCount(session) > 0 && + getMaterializeTableMaxActualRowCount(session) > 0 && + getMaterializeTableTimeout(session).toMillis() > 0; + } + + @Override + public final Result apply(T node, Captures captures, Context context) + { + int maxRows = getMaterializeTableMaxEstimatedRowCount(context.getSession()); + PlanNodeStatsEstimate stats = context.getStatsProvider().getStats(node); + if (stats.getOutputRowCount() > maxRows) { + return Result.empty(); + } + + return apply(node, captures, context.getSession()); + } + + protected abstract Result apply(T node, Captures captures, Session session); + + protected Optional> materializeTable(Session session, TableScanNode tableScan, Constraint constraint) + { + TableHandle table = tableScan.getTable(); + List columns = tableScan.getOutputSymbols().stream() + .map(symbol -> tableScan.getAssignments().get(symbol)) + .toList(); + List types = tableScan.getOutputSymbols().stream() + .map(Symbol::type) + .toList(); + + Future>> future = executor.submit(() -> { + try (var _ = new SetThreadName("MaterializeTableScan-%s", table.catalogHandle())) { + return doMaterializeTable(session, table, columns, types, constraint); + } + }); + + Duration timeout = getMaterializeTableTimeout(session); + try { + return future.get(timeout.toMillis(), MILLISECONDS); + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + future.cancel(true); + throw new RuntimeException("interrupted", e); + } + catch (ExecutionException e) { + if (e.getCause() instanceof TableTooLargeException) { + log.warn("Table too large to materialize: %s", table); + } + else { + log.warn(e.getCause(), "Failed to materialize table: %s", table); + } + } + catch (TimeoutException e) { + log.warn("Timed out while materializing table: %s", table); + future.cancel(true); + } + return Optional.empty(); + } + + private Optional> doMaterializeTable(Session session, TableHandle table, List columns, List types, Constraint constraint) + throws IOException + { + SplitSource splitSource = splitManager.getSplits(session, session.getQuerySpan(), table, DynamicFilter.EMPTY, constraint); + List splits = new ArrayList<>(); + while (!splitSource.isFinished()) { + if (Thread.currentThread().isInterrupted()) { + throw new InterruptedIOException(); + } + List batch = getFutureValue(splitSource.getNextBatch(1000)).getSplits(); + for (Split split : batch) { + if (!split.isRemotelyAccessible() && !split.getAddresses().contains(currentNode)) { + return Optional.empty(); + } + } + splits.addAll(batch); + if (splits.size() > MAX_SPLITS) { + return Optional.empty(); + } + } + + PageSourceProvider pageSourceProvider = pageSourceManager.createPageSourceProvider(table.catalogHandle()); + int maxRows = getMaterializeTableMaxActualRowCount(session); + int rowCount = 0; + ImmutableList.Builder rows = ImmutableList.builder(); + for (Split split : splits) { + try (ConnectorPageSource pageSource = pageSourceProvider.createPageSource(session, split, table, columns, DynamicFilter.EMPTY)) { + while (!pageSource.isFinished()) { + if (Thread.currentThread().isInterrupted()) { + throw new InterruptedIOException(); + } + Page page = pageSource.getNextPage(); + if (page != null && page.getPositionCount() > 0) { + rowCount += page.getPositionCount(); + if (rowCount > maxRows) { + throw new TableTooLargeException(); + } + for (int i = 0; i < page.getPositionCount(); i++) { + rows.add(toRow(page, i, types)); + } + } + } + } + } + return Optional.of(rows.build()); + } + + private static Row toRow(Page page, int position, List types) + { + ImmutableList.Builder row = ImmutableList.builder(); + for (int i = 0; i < page.getChannelCount(); i++) { + Type type = types.get(i); + Block block = page.getBlock(i); + Object value = readNativeValue(type, block, position); + row.add(new Constant(type, value)); + } + return new Row(row.build()); + } + + private static final class TableTooLargeException + extends IOException {} +} diff --git a/core/trino-main/src/main/java/io/trino/sql/planner/optimizations/MaterializeFilteredTableScan.java b/core/trino-main/src/main/java/io/trino/sql/planner/optimizations/MaterializeFilteredTableScan.java new file mode 100644 index 000000000000..a7e9500d5873 --- /dev/null +++ b/core/trino-main/src/main/java/io/trino/sql/planner/optimizations/MaterializeFilteredTableScan.java @@ -0,0 +1,86 @@ +/* + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.trino.sql.planner.optimizations; + +import com.google.common.collect.ImmutableList; +import io.trino.Session; +import io.trino.matching.Capture; +import io.trino.matching.Captures; +import io.trino.matching.Pattern; +import io.trino.metadata.InternalNodeManager; +import io.trino.spi.connector.Constraint; +import io.trino.spi.predicate.TupleDomain; +import io.trino.split.PageSourceManager; +import io.trino.split.SplitManager; +import io.trino.sql.DynamicFilters; +import io.trino.sql.PlannerContext; +import io.trino.sql.planner.LayoutConstraintEvaluator; +import io.trino.sql.planner.plan.FilterNode; +import io.trino.sql.planner.plan.TableScanNode; +import io.trino.sql.planner.plan.ValuesNode; + +import java.util.Optional; +import java.util.concurrent.ExecutorService; + +import static io.trino.matching.Capture.newCapture; +import static io.trino.spi.connector.Constraint.alwaysTrue; +import static io.trino.sql.ir.IrUtils.filterConjuncts; +import static io.trino.sql.planner.plan.Patterns.filter; +import static io.trino.sql.planner.plan.Patterns.source; +import static io.trino.sql.planner.plan.Patterns.tableScan; +import static java.util.function.Predicate.not; + +public final class MaterializeFilteredTableScan + extends AbstractMaterializeTableScan +{ + private static final Capture TABLE_SCAN = newCapture(); + private static final Pattern PATTERN = filter() + .with(source().matching( + tableScan().capturedAs(TABLE_SCAN) + .matching(not(TableScanNode::isUpdateTarget)))); + + public MaterializeFilteredTableScan( + PlannerContext plannerContext, + SplitManager splitManager, + PageSourceManager pageSourceManager, + InternalNodeManager nodeManager, + ExecutorService executor) + { + super(plannerContext, splitManager, pageSourceManager, nodeManager, executor); + } + + @Override + public Pattern getPattern() + { + return PATTERN; + } + + @Override + protected Result apply(FilterNode filter, Captures captures, Session session) + { + TableScanNode tableScan = captures.get(TABLE_SCAN); + + Constraint constraint = Optional.of(filter.getPredicate()) + .map(predicate -> filterConjuncts(predicate, expression -> !DynamicFilters.isDynamicFilter(expression))) + .map(predicate -> new LayoutConstraintEvaluator(plannerContext, session, tableScan.getAssignments(), predicate)) + .map(evaluator -> new Constraint(TupleDomain.all(), evaluator::isCandidate, evaluator.getArguments())) + .orElse(alwaysTrue()); + + return materializeTable(session, tableScan, constraint) + .map(rows -> new ValuesNode(tableScan.getId(), tableScan.getOutputSymbols(), rows)) + .map(values -> filter.replaceChildren(ImmutableList.of(values))) + .map(Result::ofPlanNode) + .orElseGet(Result::empty); + } +} diff --git a/core/trino-main/src/main/java/io/trino/sql/planner/optimizations/MaterializeTableScan.java b/core/trino-main/src/main/java/io/trino/sql/planner/optimizations/MaterializeTableScan.java new file mode 100644 index 000000000000..a4b8a199b6c3 --- /dev/null +++ b/core/trino-main/src/main/java/io/trino/sql/planner/optimizations/MaterializeTableScan.java @@ -0,0 +1,60 @@ +/* + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.trino.sql.planner.optimizations; + +import io.trino.Session; +import io.trino.matching.Captures; +import io.trino.matching.Pattern; +import io.trino.metadata.InternalNodeManager; +import io.trino.split.PageSourceManager; +import io.trino.split.SplitManager; +import io.trino.sql.PlannerContext; +import io.trino.sql.planner.plan.TableScanNode; +import io.trino.sql.planner.plan.ValuesNode; + +import java.util.concurrent.ExecutorService; + +import static io.trino.spi.connector.Constraint.alwaysTrue; +import static io.trino.sql.planner.plan.Patterns.tableScan; +import static java.util.function.Predicate.not; + +public final class MaterializeTableScan + extends AbstractMaterializeTableScan +{ + public MaterializeTableScan( + PlannerContext plannerContext, + SplitManager splitManager, + PageSourceManager pageSourceManager, + InternalNodeManager nodeManager, + ExecutorService executor) + { + super(plannerContext, splitManager, pageSourceManager, nodeManager, executor); + } + + @Override + public Pattern getPattern() + { + return tableScan() + .matching(not(TableScanNode::isUpdateTarget)); + } + + @Override + protected Result apply(TableScanNode tableScan, Captures captures, Session session) + { + return materializeTable(session, tableScan, alwaysTrue()) + .map(rows -> new ValuesNode(tableScan.getId(), tableScan.getOutputSymbols(), rows)) + .map(Result::ofPlanNode) + .orElseGet(Result::empty); + } +} diff --git a/core/trino-main/src/main/java/io/trino/testing/PlanTester.java b/core/trino-main/src/main/java/io/trino/testing/PlanTester.java index 7fcee3d4f7c3..942a2887fe59 100644 --- a/core/trino-main/src/main/java/io/trino/testing/PlanTester.java +++ b/core/trino-main/src/main/java/io/trino/testing/PlanTester.java @@ -223,6 +223,7 @@ import static io.airlift.concurrent.Threads.daemonThreadsNamed; import static io.airlift.tracing.Tracing.noopTracer; import static io.opentelemetry.api.OpenTelemetry.noop; +import static io.trino.SystemSessionProperties.MATERIALIZE_TABLE_ENABLED; import static io.trino.connector.CatalogServiceProviderModule.createAccessControlProvider; import static io.trino.connector.CatalogServiceProviderModule.createAnalyzePropertyManager; import static io.trino.connector.CatalogServiceProviderModule.createColumnPropertyManager; @@ -262,6 +263,7 @@ public class PlanTester private final Session defaultSession; private final ExecutorService notificationExecutor; private final ScheduledExecutorService yieldExecutor; + private final ExecutorService plannerExecutor; private final FinalizerService finalizerService; private final SqlParser sqlParser; @@ -305,6 +307,7 @@ public class PlanTester private final TaskManagerConfig taskManagerConfig; private final OptimizerConfig optimizerConfig; private final StatementAnalyzerFactory statementAnalyzerFactory; + private boolean printPlan; public static PlanTester create(Session defaultSession) @@ -325,6 +328,7 @@ private PlanTester(Session defaultSession, int nodeCountForStats) this.taskManagerConfig = new TaskManagerConfig().setTaskConcurrency(4); this.notificationExecutor = newCachedThreadPool(daemonThreadsNamed("local-query-runner-executor-%s")); this.yieldExecutor = newScheduledThreadPool(2, daemonThreadsNamed("local-query-runner-scheduler-%s")); + this.plannerExecutor = newCachedThreadPool(daemonThreadsNamed("local-query-runner-planner-%s")); this.finalizerService = new FinalizerService(); finalizerService.start(); @@ -476,6 +480,11 @@ private PlanTester(Session defaultSession, int nodeCountForStats) catalogManager.registerGlobalSystemConnector(globalSystemConnector); languageFunctionManager.setPlannerContext(plannerContext); + defaultSession = Session.builder(defaultSession) + // disable this optimization globally as it invalidates every test + .setSystemProperty(MATERIALIZE_TABLE_ENABLED, "false") + .build(); + // rewrite session to use managed SessionPropertyMetadata this.defaultSession = new Session( defaultSession.getQueryId(), @@ -537,6 +546,7 @@ public void close() { notificationExecutor.shutdownNow(); yieldExecutor.shutdownNow(); + plannerExecutor.shutdownNow(); catalogManager.stop(); finalizerService.destroy(); } @@ -844,10 +854,12 @@ public PlanOptimizersFactory getPlanOptimizersFactory(boolean forceSingleNode) { return new PlanOptimizers( plannerContext, + plannerExecutor, taskManagerConfig, forceSingleNode, splitManager, pageSourceManager, + nodeManager, statsCalculator, scalarStatsCalculator, costCalculator, diff --git a/core/trino-main/src/main/java/io/trino/testing/StandaloneQueryRunner.java b/core/trino-main/src/main/java/io/trino/testing/StandaloneQueryRunner.java index d62041920b7f..4c2b5f0f0131 100644 --- a/core/trino-main/src/main/java/io/trino/testing/StandaloneQueryRunner.java +++ b/core/trino-main/src/main/java/io/trino/testing/StandaloneQueryRunner.java @@ -55,8 +55,8 @@ import java.util.function.Consumer; import static com.google.common.base.Preconditions.checkState; +import static io.trino.SystemSessionProperties.MATERIALIZE_TABLE_ENABLED; import static io.trino.execution.querystats.PlanOptimizersStatsCollector.createPlanOptimizersStatsCollector; -import static java.util.Objects.requireNonNull; public final class StandaloneQueryRunner implements QueryRunner @@ -77,7 +77,11 @@ public StandaloneQueryRunner(Session defaultSession) public StandaloneQueryRunner(Session defaultSession, Consumer serverProcessor) { - this.defaultSession = requireNonNull(defaultSession, "defaultSession is null"); + this.defaultSession = Session.builder(defaultSession) + // disable this optimization globally as it invalidates every test + .setSystemProperty(MATERIALIZE_TABLE_ENABLED, "false") + .build(); + TestingTrinoServer.Builder builder = TestingTrinoServer.builder() .setSpanProcessor(SimpleSpanProcessor.create(spanExporter)) .setProperties(ImmutableMap.builder() diff --git a/core/trino-main/src/test/java/io/trino/cost/TestOptimizerConfig.java b/core/trino-main/src/test/java/io/trino/cost/TestOptimizerConfig.java index a461a420f454..9f8964bdd826 100644 --- a/core/trino-main/src/test/java/io/trino/cost/TestOptimizerConfig.java +++ b/core/trino-main/src/test/java/io/trino/cost/TestOptimizerConfig.java @@ -91,7 +91,11 @@ public void testDefaults() .setMinInputRowsPerTask(10_000_000L) .setUseExactPartitioning(false) .setUseCostBasedPartitioning(true) - .setPushFilterIntoValuesMaxRowCount(100) + .setPushFilterIntoValuesMaxRowCount(100_000) + .setMaterializeTable(true) + .setMaterializeTableMaxEstimatedRowCount(50_000) + .setMaterializeTableMaxActualRowCount(100_000) + .setMaterializeTableTimeout(new Duration(5, SECONDS)) .setUnsafePushdownAllowed(false)); } @@ -151,6 +155,10 @@ public void testExplicitPropertyMappings() .put("optimizer.use-exact-partitioning", "true") .put("optimizer.use-cost-based-partitioning", "false") .put("optimizer.push-filter-into-values-max-row-count", "5") + .put("optimizer.materialize-table.enabled", "false") + .put("optimizer.materialize-table.max-estimated-row-count", "12345") + .put("optimizer.materialize-table.max-actual-row-count", "54321") + .put("optimizer.materialize-table.timeout", "7s") .put("optimizer.allow-unsafe-pushdown", "true") .buildOrThrow(); @@ -207,6 +215,10 @@ public void testExplicitPropertyMappings() .setUseExactPartitioning(true) .setUseCostBasedPartitioning(false) .setPushFilterIntoValuesMaxRowCount(5) + .setMaterializeTable(false) + .setMaterializeTableMaxEstimatedRowCount(12345) + .setMaterializeTableMaxActualRowCount(54321) + .setMaterializeTableTimeout(new Duration(7, SECONDS)) .setUnsafePushdownAllowed(true); assertFullMapping(properties, expected); } diff --git a/core/trino-main/src/test/java/io/trino/sql/planner/TestTableScanNodePartitioning.java b/core/trino-main/src/test/java/io/trino/sql/planner/TestTableScanNodePartitioning.java index 7d4026a7e603..6d45e1afd893 100644 --- a/core/trino-main/src/test/java/io/trino/sql/planner/TestTableScanNodePartitioning.java +++ b/core/trino-main/src/test/java/io/trino/sql/planner/TestTableScanNodePartitioning.java @@ -42,6 +42,7 @@ import java.util.Optional; import java.util.function.ToIntFunction; +import static io.trino.SystemSessionProperties.MATERIALIZE_TABLE_ENABLED; import static io.trino.SystemSessionProperties.TASK_CONCURRENCY; import static io.trino.SystemSessionProperties.USE_TABLE_SCAN_NODE_PARTITIONING; import static io.trino.spi.connector.ConnectorBucketNodeMap.createBucketNodeMap; @@ -72,12 +73,14 @@ public class TestTableScanNodePartitioning public static final Session ENABLE_PLAN_WITH_TABLE_NODE_PARTITIONING = testSessionBuilder() .setCatalog(TEST_CATALOG_NAME) .setSchema(TEST_SCHEMA) + .setSystemProperty(MATERIALIZE_TABLE_ENABLED, "false") .setSystemProperty(USE_TABLE_SCAN_NODE_PARTITIONING, "true") .setSystemProperty(TASK_CONCURRENCY, "2") // force parallel plan even on test nodes with single CPU .build(); public static final Session DISABLE_PLAN_WITH_TABLE_NODE_PARTITIONING = testSessionBuilder() .setCatalog(TEST_CATALOG_NAME) .setSchema(TEST_SCHEMA) + .setSystemProperty(MATERIALIZE_TABLE_ENABLED, "false") .setSystemProperty(USE_TABLE_SCAN_NODE_PARTITIONING, "false") .setSystemProperty(TASK_CONCURRENCY, "2") // force parallel plan even on test nodes with single CPU .build(); diff --git a/core/trino-main/src/test/java/io/trino/sql/planner/planprinter/TestJsonRepresentation.java b/core/trino-main/src/test/java/io/trino/sql/planner/planprinter/TestJsonRepresentation.java index b0b518c1399d..ab3e74cfc73b 100644 --- a/core/trino-main/src/test/java/io/trino/sql/planner/planprinter/TestJsonRepresentation.java +++ b/core/trino-main/src/test/java/io/trino/sql/planner/planprinter/TestJsonRepresentation.java @@ -103,7 +103,7 @@ public void testJsonPlan() ImmutableList.of("_col0 := field"), ImmutableList.of(new PlanNodeStatsAndCostSummary(1, 5, 0, 0, 0)), ImmutableList.of(new JsonRenderedNode( - "90", + "92", "Limit", ImmutableMap.of("count", "1", "withTies", "", "inputPreSortedBy", "[]"), ImmutableList.of(new Symbol(INTEGER, "field")), diff --git a/testing/trino-product-tests-launcher/src/main/java/io/trino/tests/product/launcher/env/environment/EnvSinglenodeCompatibility.java b/testing/trino-product-tests-launcher/src/main/java/io/trino/tests/product/launcher/env/environment/EnvSinglenodeCompatibility.java index 147967fc3c6f..06eb4d56629d 100644 --- a/testing/trino-product-tests-launcher/src/main/java/io/trino/tests/product/launcher/env/environment/EnvSinglenodeCompatibility.java +++ b/testing/trino-product-tests-launcher/src/main/java/io/trino/tests/product/launcher/env/environment/EnvSinglenodeCompatibility.java @@ -109,6 +109,9 @@ private String getConfigFileFor(String dockerImage) if (getVersionFromDockerImageName(dockerImage) < 369) { return "config-pre369.properties"; } + if (getVersionFromDockerImageName(dockerImage) < 458) { + return "config-pre458.properties"; + } return "config.properties"; } diff --git a/testing/trino-product-tests-launcher/src/main/resources/docker/presto-product-tests/common/hadoop-kerberos/config.properties b/testing/trino-product-tests-launcher/src/main/resources/docker/presto-product-tests/common/hadoop-kerberos/config.properties index facbc5966c27..28ef07d9cd36 100644 --- a/testing/trino-product-tests-launcher/src/main/resources/docker/presto-product-tests/common/hadoop-kerberos/config.properties +++ b/testing/trino-product-tests-launcher/src/main/resources/docker/presto-product-tests/common/hadoop-kerberos/config.properties @@ -4,6 +4,7 @@ node.internal-address-source=FQDN coordinator=true node-scheduler.include-coordinator=true +optimizer.materialize-table.enabled=false query.max-memory=1GB query.max-memory-per-node=1GB discovery.uri=https://presto-master.docker.cluster:7778 diff --git a/testing/trino-product-tests-launcher/src/main/resources/docker/presto-product-tests/common/standard-multinode/multinode-master-config.properties b/testing/trino-product-tests-launcher/src/main/resources/docker/presto-product-tests/common/standard-multinode/multinode-master-config.properties index e3dc3e74ade7..180e84f70a24 100644 --- a/testing/trino-product-tests-launcher/src/main/resources/docker/presto-product-tests/common/standard-multinode/multinode-master-config.properties +++ b/testing/trino-product-tests-launcher/src/main/resources/docker/presto-product-tests/common/standard-multinode/multinode-master-config.properties @@ -5,6 +5,7 @@ coordinator=true experimental.concurrent-startup=true node-scheduler.include-coordinator=false http-server.http.port=8080 +optimizer.materialize-table.enabled=false query.max-memory=1GB discovery.uri=http://presto-master:8080 diff --git a/testing/trino-product-tests-launcher/src/main/resources/docker/presto-product-tests/common/standard/config.properties b/testing/trino-product-tests-launcher/src/main/resources/docker/presto-product-tests/common/standard/config.properties index 6ad61ff2be58..b539f71fb714 100644 --- a/testing/trino-product-tests-launcher/src/main/resources/docker/presto-product-tests/common/standard/config.properties +++ b/testing/trino-product-tests-launcher/src/main/resources/docker/presto-product-tests/common/standard/config.properties @@ -5,6 +5,7 @@ coordinator=true experimental.concurrent-startup=true node-scheduler.include-coordinator=true http-server.http.port=8080 +optimizer.materialize-table.enabled=false query.max-memory=2GB query.max-memory-per-node=1.25GB discovery.uri=http://presto-master:8080 diff --git a/testing/trino-product-tests-launcher/src/main/resources/docker/presto-product-tests/common/task-retries-multinode/multinode-master-config.properties b/testing/trino-product-tests-launcher/src/main/resources/docker/presto-product-tests/common/task-retries-multinode/multinode-master-config.properties index 41056d70d698..7bdcaf4559c3 100644 --- a/testing/trino-product-tests-launcher/src/main/resources/docker/presto-product-tests/common/task-retries-multinode/multinode-master-config.properties +++ b/testing/trino-product-tests-launcher/src/main/resources/docker/presto-product-tests/common/task-retries-multinode/multinode-master-config.properties @@ -5,6 +5,7 @@ coordinator=true experimental.concurrent-startup=true node-scheduler.include-coordinator=false http-server.http.port=8080 +optimizer.materialize-table.enabled=false query.max-memory=1GB discovery.uri=http://presto-master:8080 diff --git a/testing/trino-product-tests-launcher/src/main/resources/docker/presto-product-tests/conf/environment/multinode-postgresql-spooling/coordinator-config.properties b/testing/trino-product-tests-launcher/src/main/resources/docker/presto-product-tests/conf/environment/multinode-postgresql-spooling/coordinator-config.properties index 3cb3e7ae064e..7d1424342746 100644 --- a/testing/trino-product-tests-launcher/src/main/resources/docker/presto-product-tests/conf/environment/multinode-postgresql-spooling/coordinator-config.properties +++ b/testing/trino-product-tests-launcher/src/main/resources/docker/presto-product-tests/conf/environment/multinode-postgresql-spooling/coordinator-config.properties @@ -5,6 +5,7 @@ coordinator=true experimental.concurrent-startup=true node-scheduler.include-coordinator=false http-server.http.port=8080 +optimizer.materialize-table.enabled=false query.max-memory=1GB discovery.uri=http://presto-master:8080 diff --git a/testing/trino-product-tests-launcher/src/main/resources/docker/presto-product-tests/conf/environment/multinode-secrets-provider/config-master.properties b/testing/trino-product-tests-launcher/src/main/resources/docker/presto-product-tests/conf/environment/multinode-secrets-provider/config-master.properties index c929813a2ce1..9667843f2109 100644 --- a/testing/trino-product-tests-launcher/src/main/resources/docker/presto-product-tests/conf/environment/multinode-secrets-provider/config-master.properties +++ b/testing/trino-product-tests-launcher/src/main/resources/docker/presto-product-tests/conf/environment/multinode-secrets-provider/config-master.properties @@ -4,6 +4,7 @@ node.internal-address-source=FQDN coordinator=true node-scheduler.include-coordinator=false +optimizer.materialize-table.enabled=false discovery.uri=https://presto-master.docker.cluster:7778 query.max-memory=1GB diff --git a/testing/trino-product-tests-launcher/src/main/resources/docker/presto-product-tests/conf/environment/multinode-tls-kerberos/config-master.properties b/testing/trino-product-tests-launcher/src/main/resources/docker/presto-product-tests/conf/environment/multinode-tls-kerberos/config-master.properties index 5bb94e071eae..3e05500cb42d 100644 --- a/testing/trino-product-tests-launcher/src/main/resources/docker/presto-product-tests/conf/environment/multinode-tls-kerberos/config-master.properties +++ b/testing/trino-product-tests-launcher/src/main/resources/docker/presto-product-tests/conf/environment/multinode-tls-kerberos/config-master.properties @@ -4,6 +4,7 @@ node.internal-address-source=FQDN coordinator=true node-scheduler.include-coordinator=false +optimizer.materialize-table.enabled=false discovery.uri=https://presto-master.docker.cluster:7778 query.max-memory=1GB diff --git a/testing/trino-product-tests-launcher/src/main/resources/docker/presto-product-tests/conf/environment/multinode-tls/config-master.properties b/testing/trino-product-tests-launcher/src/main/resources/docker/presto-product-tests/conf/environment/multinode-tls/config-master.properties index 4964430b0af9..160c39d6e89a 100644 --- a/testing/trino-product-tests-launcher/src/main/resources/docker/presto-product-tests/conf/environment/multinode-tls/config-master.properties +++ b/testing/trino-product-tests-launcher/src/main/resources/docker/presto-product-tests/conf/environment/multinode-tls/config-master.properties @@ -4,6 +4,7 @@ node.internal-address-source=FQDN coordinator=true node-scheduler.include-coordinator=false +optimizer.materialize-table.enabled=false discovery.uri=https://presto-master.docker.cluster:7778 query.max-memory=1GB diff --git a/testing/trino-product-tests-launcher/src/main/resources/docker/presto-product-tests/conf/environment/singlenode-compatibility/config-pre458.properties b/testing/trino-product-tests-launcher/src/main/resources/docker/presto-product-tests/conf/environment/singlenode-compatibility/config-pre458.properties new file mode 100644 index 000000000000..9822f0100f35 --- /dev/null +++ b/testing/trino-product-tests-launcher/src/main/resources/docker/presto-product-tests/conf/environment/singlenode-compatibility/config-pre458.properties @@ -0,0 +1,7 @@ +coordinator=true +node-scheduler.include-coordinator=true +http-server.http.port=8081 +query.max-memory=2GB +query.max-memory-per-node=1.25GB +discovery-server.enabled=true +discovery.uri=http://compatibility-test-coordinator:8081 diff --git a/testing/trino-product-tests-launcher/src/main/resources/docker/presto-product-tests/conf/environment/singlenode-compatibility/config.properties b/testing/trino-product-tests-launcher/src/main/resources/docker/presto-product-tests/conf/environment/singlenode-compatibility/config.properties index 9822f0100f35..3669e831341a 100644 --- a/testing/trino-product-tests-launcher/src/main/resources/docker/presto-product-tests/conf/environment/singlenode-compatibility/config.properties +++ b/testing/trino-product-tests-launcher/src/main/resources/docker/presto-product-tests/conf/environment/singlenode-compatibility/config.properties @@ -1,6 +1,7 @@ coordinator=true node-scheduler.include-coordinator=true http-server.http.port=8081 +optimizer.materialize-table.enabled=false query.max-memory=2GB query.max-memory-per-node=1.25GB discovery-server.enabled=true diff --git a/testing/trino-product-tests-launcher/src/main/resources/docker/presto-product-tests/conf/environment/singlenode-ldap-and-file/config.properties b/testing/trino-product-tests-launcher/src/main/resources/docker/presto-product-tests/conf/environment/singlenode-ldap-and-file/config.properties index 9574860d25d2..01dbc61e7514 100644 --- a/testing/trino-product-tests-launcher/src/main/resources/docker/presto-product-tests/conf/environment/singlenode-ldap-and-file/config.properties +++ b/testing/trino-product-tests-launcher/src/main/resources/docker/presto-product-tests/conf/environment/singlenode-ldap-and-file/config.properties @@ -11,6 +11,7 @@ node.internal-address-source=FQDN coordinator=true node-scheduler.include-coordinator=true +optimizer.materialize-table.enabled=false query.max-memory=1GB query.max-memory-per-node=1GB discovery.uri=https://presto-master:8443 diff --git a/testing/trino-product-tests-launcher/src/main/resources/docker/presto-product-tests/conf/environment/singlenode-ldap/config.properties b/testing/trino-product-tests-launcher/src/main/resources/docker/presto-product-tests/conf/environment/singlenode-ldap/config.properties index c4b994bae711..036098dc8929 100644 --- a/testing/trino-product-tests-launcher/src/main/resources/docker/presto-product-tests/conf/environment/singlenode-ldap/config.properties +++ b/testing/trino-product-tests-launcher/src/main/resources/docker/presto-product-tests/conf/environment/singlenode-ldap/config.properties @@ -11,6 +11,7 @@ node.internal-address-source=FQDN coordinator=true node-scheduler.include-coordinator=true +optimizer.materialize-table.enabled=false query.max-memory=1GB query.max-memory-per-node=1GB discovery.uri=https://presto-master:8443 diff --git a/testing/trino-product-tests-launcher/src/main/resources/docker/presto-product-tests/conf/environment/singlenode-oauth2-authenticated-http-proxy/trino/config.properties b/testing/trino-product-tests-launcher/src/main/resources/docker/presto-product-tests/conf/environment/singlenode-oauth2-authenticated-http-proxy/trino/config.properties index bc46205fe0ba..3b6e2da7b0cc 100644 --- a/testing/trino-product-tests-launcher/src/main/resources/docker/presto-product-tests/conf/environment/singlenode-oauth2-authenticated-http-proxy/trino/config.properties +++ b/testing/trino-product-tests-launcher/src/main/resources/docker/presto-product-tests/conf/environment/singlenode-oauth2-authenticated-http-proxy/trino/config.properties @@ -4,6 +4,7 @@ node.environment=test coordinator=true node-scheduler.include-coordinator=true http-server.http.port=8080 +optimizer.materialize-table.enabled=false query.max-memory=2GB query.max-memory-per-node=1.25GB discovery.uri=http://presto-master:8080 diff --git a/testing/trino-product-tests-launcher/src/main/resources/docker/presto-product-tests/conf/environment/singlenode-oauth2-authenticated-https-proxy/trino/config.properties b/testing/trino-product-tests-launcher/src/main/resources/docker/presto-product-tests/conf/environment/singlenode-oauth2-authenticated-https-proxy/trino/config.properties index 918d0d247eba..7e840b505791 100644 --- a/testing/trino-product-tests-launcher/src/main/resources/docker/presto-product-tests/conf/environment/singlenode-oauth2-authenticated-https-proxy/trino/config.properties +++ b/testing/trino-product-tests-launcher/src/main/resources/docker/presto-product-tests/conf/environment/singlenode-oauth2-authenticated-https-proxy/trino/config.properties @@ -4,6 +4,7 @@ node.environment=test coordinator=true node-scheduler.include-coordinator=true http-server.http.port=8080 +optimizer.materialize-table.enabled=false query.max-memory=2GB query.max-memory-per-node=1.25GB discovery.uri=http://presto-master:8080 diff --git a/testing/trino-product-tests-launcher/src/main/resources/docker/presto-product-tests/conf/environment/singlenode-oauth2-http-proxy/config.properties b/testing/trino-product-tests-launcher/src/main/resources/docker/presto-product-tests/conf/environment/singlenode-oauth2-http-proxy/config.properties index 64e7a3f53d4d..fff012b70ec0 100644 --- a/testing/trino-product-tests-launcher/src/main/resources/docker/presto-product-tests/conf/environment/singlenode-oauth2-http-proxy/config.properties +++ b/testing/trino-product-tests-launcher/src/main/resources/docker/presto-product-tests/conf/environment/singlenode-oauth2-http-proxy/config.properties @@ -4,6 +4,7 @@ node.environment=test coordinator=true node-scheduler.include-coordinator=true http-server.http.port=8080 +optimizer.materialize-table.enabled=false query.max-memory=2GB query.max-memory-per-node=1.25GB discovery.uri=http://presto-master:8080 diff --git a/testing/trino-product-tests-launcher/src/main/resources/docker/presto-product-tests/conf/environment/singlenode-oauth2-https-proxy/config.properties b/testing/trino-product-tests-launcher/src/main/resources/docker/presto-product-tests/conf/environment/singlenode-oauth2-https-proxy/config.properties index d3560c2305bc..26146324fabb 100644 --- a/testing/trino-product-tests-launcher/src/main/resources/docker/presto-product-tests/conf/environment/singlenode-oauth2-https-proxy/config.properties +++ b/testing/trino-product-tests-launcher/src/main/resources/docker/presto-product-tests/conf/environment/singlenode-oauth2-https-proxy/config.properties @@ -4,6 +4,7 @@ node.environment=test coordinator=true node-scheduler.include-coordinator=true http-server.http.port=8080 +optimizer.materialize-table.enabled=false query.max-memory=2GB query.max-memory-per-node=1.25GB discovery.uri=http://presto-master:8080 diff --git a/testing/trino-product-tests-launcher/src/main/resources/docker/presto-product-tests/conf/environment/singlenode-oauth2-refresh/config.properties b/testing/trino-product-tests-launcher/src/main/resources/docker/presto-product-tests/conf/environment/singlenode-oauth2-refresh/config.properties index 6a62a169a8e9..12d15c0fa9c8 100644 --- a/testing/trino-product-tests-launcher/src/main/resources/docker/presto-product-tests/conf/environment/singlenode-oauth2-refresh/config.properties +++ b/testing/trino-product-tests-launcher/src/main/resources/docker/presto-product-tests/conf/environment/singlenode-oauth2-refresh/config.properties @@ -4,6 +4,7 @@ node.environment=test coordinator=true node-scheduler.include-coordinator=true http-server.http.port=8080 +optimizer.materialize-table.enabled=false query.max-memory=2GB query.max-memory-per-node=1.25GB discovery.uri=http://presto-master:8080 diff --git a/testing/trino-product-tests-launcher/src/main/resources/docker/presto-product-tests/conf/environment/singlenode-oauth2/config.properties b/testing/trino-product-tests-launcher/src/main/resources/docker/presto-product-tests/conf/environment/singlenode-oauth2/config.properties index 31c18d830d20..2b3c39c9721a 100644 --- a/testing/trino-product-tests-launcher/src/main/resources/docker/presto-product-tests/conf/environment/singlenode-oauth2/config.properties +++ b/testing/trino-product-tests-launcher/src/main/resources/docker/presto-product-tests/conf/environment/singlenode-oauth2/config.properties @@ -4,6 +4,7 @@ node.environment=test coordinator=true node-scheduler.include-coordinator=true http-server.http.port=8080 +optimizer.materialize-table.enabled=false query.max-memory=2GB query.max-memory-per-node=1.25GB discovery.uri=http://presto-master:8080 diff --git a/testing/trino-product-tests-launcher/src/main/resources/docker/presto-product-tests/conf/environment/singlenode-oidc-refresh/config.properties b/testing/trino-product-tests-launcher/src/main/resources/docker/presto-product-tests/conf/environment/singlenode-oidc-refresh/config.properties index 6a452392cce9..70bbdf1b4d19 100644 --- a/testing/trino-product-tests-launcher/src/main/resources/docker/presto-product-tests/conf/environment/singlenode-oidc-refresh/config.properties +++ b/testing/trino-product-tests-launcher/src/main/resources/docker/presto-product-tests/conf/environment/singlenode-oidc-refresh/config.properties @@ -4,6 +4,7 @@ node.environment=test coordinator=true node-scheduler.include-coordinator=true http-server.http.port=8080 +optimizer.materialize-table.enabled=false query.max-memory=2GB query.max-memory-per-node=1.25GB discovery.uri=http://presto-master:8080 diff --git a/testing/trino-product-tests-launcher/src/main/resources/docker/presto-product-tests/conf/environment/singlenode-oidc/config.properties b/testing/trino-product-tests-launcher/src/main/resources/docker/presto-product-tests/conf/environment/singlenode-oidc/config.properties index 0d3de0c078c2..4966919b3f2f 100644 --- a/testing/trino-product-tests-launcher/src/main/resources/docker/presto-product-tests/conf/environment/singlenode-oidc/config.properties +++ b/testing/trino-product-tests-launcher/src/main/resources/docker/presto-product-tests/conf/environment/singlenode-oidc/config.properties @@ -4,6 +4,7 @@ node.environment=test coordinator=true node-scheduler.include-coordinator=true http-server.http.port=8080 +optimizer.materialize-table.enabled=false query.max-memory=2GB query.max-memory-per-node=1.25GB discovery.uri=http://presto-master:8080 diff --git a/testing/trino-testing/src/main/java/io/trino/testing/DistributedQueryRunner.java b/testing/trino-testing/src/main/java/io/trino/testing/DistributedQueryRunner.java index 6235b56eb0b5..372fc0bc6cb1 100644 --- a/testing/trino-testing/src/main/java/io/trino/testing/DistributedQueryRunner.java +++ b/testing/trino-testing/src/main/java/io/trino/testing/DistributedQueryRunner.java @@ -97,6 +97,7 @@ import static io.airlift.log.Level.WARN; import static io.airlift.testing.Closeables.closeAllSuppress; import static io.airlift.units.Duration.nanosSince; +import static io.trino.SystemSessionProperties.MATERIALIZE_TABLE_ENABLED; import static io.trino.client.StatementClientFactory.newStatementClient; import static io.trino.execution.querystats.PlanOptimizersStatsCollector.createPlanOptimizersStatsCollector; import static java.lang.Boolean.parseBoolean; @@ -720,7 +721,10 @@ public static class Builder> protected Builder(Session defaultSession) { - this.defaultSession = requireNonNull(defaultSession, "defaultSession is null"); + this.defaultSession = Session.builder(defaultSession) + // disable this optimization globally as it invalidates every test + .setSystemProperty(MATERIALIZE_TABLE_ENABLED, "false") + .build(); String tracingEnabled = firstNonNull(getenv("TESTS_TRACING_ENABLED"), "false"); this.withTracing = parseBoolean(tracingEnabled) || tracingEnabled.equals("1"); } diff --git a/testing/trino-tests/src/test/java/io/trino/execution/TestEventListenerBasic.java b/testing/trino-tests/src/test/java/io/trino/execution/TestEventListenerBasic.java index 8825ed975b9f..ae5c154dd555 100644 --- a/testing/trino-tests/src/test/java/io/trino/execution/TestEventListenerBasic.java +++ b/testing/trino-tests/src/test/java/io/trino/execution/TestEventListenerBasic.java @@ -1453,14 +1453,14 @@ public void testAnonymizedJsonPlan() ImmutableList.of(), ImmutableList.of(new PlanNodeStatsAndCostSummary(10., 90., 0., 0., 0.)), ImmutableList.of(new JsonRenderedNode( - "100", + "102", "Limit", ImmutableMap.of("count", "10", "withTies", "", "inputPreSortedBy", "[]"), ImmutableList.of(new Symbol(DOUBLE, "symbol_1")), ImmutableList.of(), ImmutableList.of(new PlanNodeStatsAndCostSummary(10., 90., 90., 0., 0.)), ImmutableList.of(new JsonRenderedNode( - "173", + "175", "LocalExchange", ImmutableMap.of( "partitioning", "[connectorHandleType = SystemPartitioningHandle, partitioning = SINGLE, function = SINGLE]", @@ -1471,7 +1471,7 @@ public void testAnonymizedJsonPlan() ImmutableList.of(), ImmutableList.of(new PlanNodeStatsAndCostSummary(10., 90., 0., 0., 0.)), ImmutableList.of(new JsonRenderedNode( - "140", + "142", "RemoteSource", ImmutableMap.of("sourceFragmentIds", "[1]"), ImmutableList.of(new Symbol(DOUBLE, "symbol_1")), @@ -1479,7 +1479,7 @@ public void testAnonymizedJsonPlan() ImmutableList.of(), ImmutableList.of()))))))), "1", new JsonRenderedNode( - "139", + "141", "LimitPartial", ImmutableMap.of( "count", "10",