|
| 1 | +# Licensed to the Apache Software Foundation (ASF) under one |
| 2 | +# or more contributor license agreements. See the NOTICE file |
| 3 | +# distributed with this work for additional information |
| 4 | +# regarding copyright ownership. The ASF licenses this file |
| 5 | +# to you under the Apache License, Version 2.0 (the |
| 6 | +# "License"); you may not use this file except in compliance |
| 7 | +# with the License. You may obtain a copy of the License at |
| 8 | +# |
| 9 | +# http://www.apache.org/licenses/LICENSE-2.0 |
| 10 | +# |
| 11 | +# Unless required by applicable law or agreed to in writing, |
| 12 | +# software distributed under the License is distributed on an |
| 13 | +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY |
| 14 | +# KIND, either express or implied. See the License for the |
| 15 | +# specific language governing permissions and limitations |
| 16 | +# under the License. |
| 17 | + |
| 18 | +from __future__ import absolute_import, division, print_function |
| 19 | +import pytest |
| 20 | + |
| 21 | +from time import sleep |
| 22 | + |
| 23 | +from tests.common.custom_cluster_test_suite import CustomClusterTestSuite |
| 24 | +from tests.common.impala_connection import RUNNING |
| 25 | +from tests.common.skip import SkipIfExploration |
| 26 | +from tests.util.retry import retry |
| 27 | + |
| 28 | + |
| 29 | +@SkipIfExploration.is_not_exhaustive() |
| 30 | +class TestConjuncts(CustomClusterTestSuite): |
| 31 | + |
| 32 | + @pytest.mark.stress |
| 33 | + @pytest.mark.execute_serially |
| 34 | + @CustomClusterTestSuite.with_args(cluster_size=1) |
| 35 | + def test_create_cache_many_tables(self, unique_database): |
| 36 | + """Replicates the situation behund IMPALA-14863 where processing a row batch can cause |
| 37 | + the memory usage of a fragment instance on an executor to consume up to the process |
| 38 | + memory limit. Must be a custom cluster test because the fragment instances from the |
| 39 | + bad query continues to run for a long time after the query is cancelled.""" |
| 40 | + self.client.execute("CREATE TABLE {}.test_left_fact (product_sk BIGINT," |
| 41 | + "drv_event_type STRING)".format(unique_database)) |
| 42 | + self.client.execute("CREATE TABLE {}.test_right_cte (product_sk BIGINT," |
| 43 | + "product_type STRING)".format(unique_database)) |
| 44 | + |
| 45 | + # Insert 1,024 rows into the Left Fact table (Exactly 1 Impala Batch). All rows share |
| 46 | + # the same join key (999). |
| 47 | + self.client.execute("""WITH ten AS ( |
| 48 | + SELECT 0 AS n UNION ALL SELECT 1 UNION ALL SELECT 2 UNION ALL SELECT 3 UNION ALL |
| 49 | + SELECT 4 UNION ALL SELECT 5 UNION ALL SELECT 6 UNION ALL SELECT 7 UNION ALL |
| 50 | + SELECT 8 UNION ALL SELECT 9 |
| 51 | + ) |
| 52 | + INSERT INTO {}.test_left_fact |
| 53 | + SELECT 999 AS product_sk, 'fail_event' AS drv_event_type FROM ten a |
| 54 | + CROSS JOIN ten b CROSS JOIN ten c LIMIT 1024""".format(unique_database)) |
| 55 | + |
| 56 | + # Insert 1,000,000 rows into the Right CTE table. All 1 Million rows share the exact |
| 57 | + # same join key (999). |
| 58 | + self.client.execute("""WITH ten AS ( |
| 59 | + SELECT 0 AS n UNION ALL SELECT 1 UNION ALL SELECT 2 UNION ALL SELECT 3 UNION ALL |
| 60 | + SELECT 4 UNION ALL SELECT 5 UNION ALL SELECT 6 UNION ALL SELECT 7 UNION ALL |
| 61 | + SELECT 8 UNION ALL SELECT 9 |
| 62 | + ) |
| 63 | + INSERT INTO {}.test_right_cte |
| 64 | + SELECT 999 AS product_sk, 'wrong_type' AS product_type FROM ten a |
| 65 | + CROSS JOIN ten b CROSS JOIN ten c CROSS JOIN ten d CROSS JOIN ten e |
| 66 | + CROSS JOIN ten f""".format(unique_database)) |
| 67 | + |
| 68 | + # Set a very low memory limit that will quickly be exceeded by the join conjuncts. |
| 69 | + self.client.set_configuration_option("mem_limit_executors", "185mb") |
| 70 | + |
| 71 | + # Run a query that will cause the join conjuncts to exceed the memory limit. |
| 72 | + handle = self.client.execute_async("""SELECT STRAIGHT_JOIN count(1) |
| 73 | + FROM {0}.test_left_fact f |
| 74 | + LEFT JOIN {0}.test_right_cte p2 |
| 75 | + ON f.product_sk = p2.product_sk |
| 76 | + WHERE ( |
| 77 | + UPPER(f.drv_event_type) LIKE '%_CONF' |
| 78 | + OR |
| 79 | + ( |
| 80 | + CASE |
| 81 | + WHEN UPPER(p2.product_type) = 'COMMODITY-SWAP' |
| 82 | + THEN UPPER(CONCAT(f.drv_event_type, '_SWAP_1')) |
| 83 | + WHEN UPPER(p2.product_type) = 'METAL-FUTURE' |
| 84 | + THEN UPPER(CONCAT(f.drv_event_type, '_METAL_2')) |
| 85 | + WHEN UPPER(p2.product_type) = 'ENERGY-FUTURE' |
| 86 | + THEN UPPER(CONCAT(f.drv_event_type, '_ENERGY_3')) |
| 87 | + WHEN UPPER(p2.product_type) = 'AGRICULTURAL-FUTURE' |
| 88 | + THEN UPPER(CONCAT(f.drv_event_type, '_AGRI_4')) |
| 89 | + WHEN UPPER(p2.product_type) = 'SOFT-FUTURE' |
| 90 | + THEN UPPER(CONCAT(f.drv_event_type, '_SOFT_5')) |
| 91 | + WHEN UPPER(p2.product_type) = 'OTHER-FUTURE' |
| 92 | + THEN UPPER(CONCAT(f.drv_event_type, '_OTHER_6')) |
| 93 | + WHEN UPPER(p2.product_type) = 'METAL-OPTION' |
| 94 | + THEN UPPER(CONCAT(f.drv_event_type, '_MOPT_7')) |
| 95 | + WHEN UPPER(p2.product_type) = 'ENERGY-OPTION' |
| 96 | + THEN UPPER(CONCAT(f.drv_event_type, '_EOPT_8')) |
| 97 | + WHEN UPPER(p2.product_type) = 'SPREAD-OPTION' |
| 98 | + THEN UPPER(CONCAT(f.drv_event_type, '_SOPT_9')) |
| 99 | + WHEN UPPER(p2.product_type) = 'COMMODITY-INDEX' |
| 100 | + THEN UPPER(CONCAT(f.drv_event_type, '_INDEX_10')) |
| 101 | + ELSE UPPER(CONCAT(UPPER(f.drv_event_type), '_UNKNOWN')) |
| 102 | + END = 'COMMODITY-FORWARD' |
| 103 | + ) |
| 104 | + )""".format(unique_database)) |
| 105 | + self.client.wait_for_impala_state(handle, RUNNING, 10) |
| 106 | + |
| 107 | + # Wait for the query to consume more than 270mb of memory indicating it has started |
| 108 | + # evaluating the join conjuncts. |
| 109 | + def __wait_for_270mb(): |
| 110 | + return self.get_metric("memory.total-used") > 270 * 1024 * 1024 |
| 111 | + retry(__wait_for_270mb, 10, 1, 1) |
| 112 | + mem_before = self.get_metric("memory.total-used") |
| 113 | + |
| 114 | + # Give the query some time to continue to run. |
| 115 | + sleep(10) |
| 116 | + |
| 117 | + # Assert the memory usage did not increase more than 25mb during the sleep. If the |
| 118 | + # expression results pool is not being cleared properly, the memory usage will |
| 119 | + # continue to grow as more rows are processed. |
| 120 | + mem_after = self.get_metric("memory.total-used") |
| 121 | + assert mem_after - mem_before < 25 * 1024 * 1024, "Memory usage increased during " \ |
| 122 | + "sleep. Before: {} After: {}".format(mem_before, mem_after) |
0 commit comments