From 8024d3015523b4da04e721934ead72dfc4a1a5a5 Mon Sep 17 00:00:00 2001 From: Martin Traverso Date: Wed, 2 Nov 2022 16:07:04 -0700 Subject: [PATCH] Allow empty values within partitioned fragment In some cases, a subplan containing a remote exchange (which results in a partitioned fragment) can be replaced with an empty Values node. In the process, the remote exchange is removed, which leaves the Values node within the downstream partitioned fragment. The plan fragmenter cannot currently deal with that scenario. It fails with an error indicating the partitioning is already set. However, it is for an empty Values to appear within a partitioned fragment, since there are no issues with rows being "replicated" across multiple fragments. --- .../io/trino/sql/planner/PlanFragmenter.java | 11 ++- plugin/trino-hive/pom.xml | 40 +++++++++++ .../io/trino/plugin/hive/TestIssue14317.java | 70 +++++++++++++++++++ 3 files changed, 120 insertions(+), 1 deletion(-) create mode 100644 plugin/trino-hive/src/test/java/io/trino/plugin/hive/TestIssue14317.java diff --git a/core/trino-main/src/main/java/io/trino/sql/planner/PlanFragmenter.java b/core/trino-main/src/main/java/io/trino/sql/planner/PlanFragmenter.java index c83ec0f64c3e..d55b8cf7f614 100644 --- a/core/trino-main/src/main/java/io/trino/sql/planner/PlanFragmenter.java +++ b/core/trino-main/src/main/java/io/trino/sql/planner/PlanFragmenter.java @@ -347,7 +347,11 @@ public PlanNode visitMergeProcessor(MergeProcessorNode node, RewriteContext context) { - context.get().setSingleNodeDistribution(); + // An empty values node is compatible with any distribution, so + // don't attempt to overwrite one's already been chosen + if (node.getRowCount() != 0 || !context.get().hasDistribution()) { + context.get().setSingleNodeDistribution(); + } return context.defaultRewrite(node, context.get()); } @@ -435,6 +439,11 @@ public List getChildren() return children; } + public boolean hasDistribution() + { + return partitioningHandle.isPresent(); + } + public FragmentProperties setSingleNodeDistribution() { if (partitioningHandle.isPresent() && partitioningHandle.get().isSingleNode()) { diff --git a/plugin/trino-hive/pom.xml b/plugin/trino-hive/pom.xml index cb30495ce44f..8da81d97509e 100644 --- a/plugin/trino-hive/pom.xml +++ b/plugin/trino-hive/pom.xml @@ -430,6 +430,24 @@ test + + org.junit.jupiter + junit-jupiter-api + test + + + + org.junit.jupiter + junit-jupiter-engine + test + + + + org.junit.jupiter + junit-jupiter-params + test + + org.openjdk.jmh jmh-core @@ -455,6 +473,28 @@ + + + + org.apache.maven.plugins + maven-surefire-plugin + + + + org.apache.maven.surefire + surefire-junit-platform + ${dep.plugin.surefire.version} + + + org.apache.maven.surefire + surefire-testng + ${dep.plugin.surefire.version} + + + + + + default diff --git a/plugin/trino-hive/src/test/java/io/trino/plugin/hive/TestIssue14317.java b/plugin/trino-hive/src/test/java/io/trino/plugin/hive/TestIssue14317.java new file mode 100644 index 000000000000..e02584191cbe --- /dev/null +++ b/plugin/trino-hive/src/test/java/io/trino/plugin/hive/TestIssue14317.java @@ -0,0 +1,70 @@ +/* + * 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.plugin.hive; + +import io.trino.sql.query.QueryAssertions; +import io.trino.testing.DistributedQueryRunner; +import org.testng.annotations.Test; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * Test for https://github.com/trinodb/trino/issues/14317 + *

+ * The only known reproduction involves a connector that exposes partitioned tables + * and supports predicate pushdown. + */ +public class TestIssue14317 +{ + @Test + public void test() + throws Exception + { + HiveQueryRunner.Builder builder = HiveQueryRunner.builder() + .setCreateTpchSchemas(false); + + try (DistributedQueryRunner queryRunner = builder.build(); + QueryAssertions assertions = new QueryAssertions(queryRunner);) { + queryRunner.execute("CREATE SCHEMA s"); + + queryRunner.execute("CREATE TABLE s.t (a bigint, b bigint)"); + queryRunner.execute("CREATE TABLE s.u (c bigint, d bigint) WITH (partitioned_by = array['d'])"); + queryRunner.execute("CREATE TABLE s.v (e bigint, f bigint)"); + + queryRunner.execute("INSERT INTO s.t VALUES (5, 6)"); + queryRunner.execute("INSERT INTO s.u VALUES (5, 6)"); + queryRunner.execute("INSERT INTO s.v VALUES (5, 6)"); + + assertThat(assertions.query(""" + WITH t1 AS ( + SELECT a + FROM ( + SELECT a, ROW_NUMBER() OVER (PARTITION BY a) AS rn + FROM s.t) + WHERE rn = 1), + t2 AS (SELECT c FROM s.u WHERE d - 5 = 8) + SELECT v.e + FROM s.v + INNER JOIN t1 on v.e = t1.a + INNER JOIN t2 ON v.e = t2.c + """)) + .returnsEmptyResult(); + + queryRunner.execute("DROP TABLE s.t"); + queryRunner.execute("DROP TABLE s.u"); + queryRunner.execute("DROP TABLE s.v"); + queryRunner.execute("DROP SCHEMA s"); + } + } +}