Encapsulates functionality to run a query and provide a runtime report.
| 62 | |
| 63 | |
| 64 | class QueryRunner(object): |
| 65 | """Encapsulates functionality to run a query and provide a runtime report.""" |
| 66 | |
| 67 | SPILLED_PATTERNS = [re.compile("ExecOption:.*Spilled"), re.compile("SpilledRuns: [^0]")] |
| 68 | BATCH_SIZE = 1024 |
| 69 | |
| 70 | def __init__(self, impalad, results_dir, use_kerberos, common_query_options, |
| 71 | test_admission_control, check_if_mem_was_spilled=False): |
| 72 | """Creates a new instance, but does not start the process. """ |
| 73 | self.impalad = impalad |
| 74 | self.use_kerberos = use_kerberos |
| 75 | self.results_dir = results_dir |
| 76 | self.check_if_mem_was_spilled = check_if_mem_was_spilled |
| 77 | self.common_query_options = common_query_options |
| 78 | self.test_admission_control = test_admission_control |
| 79 | # proc is filled out by caller |
| 80 | self.proc = None |
| 81 | # impalad_conn is initialised in connect() |
| 82 | self.impalad_conn = None |
| 83 | |
| 84 | # All these values are shared values between processes. We want these to be accessible |
| 85 | # by the parent process that started this QueryRunner, for operational purposes. |
| 86 | self._metrics = { |
| 87 | NUM_QUERIES_DEQUEUED: Value("i", 0), |
| 88 | NUM_QUERIES_SUBMITTED: Value("i", 0), |
| 89 | NUM_QUERIES_STARTED_RUNNING_OR_CANCELLED: Value("i", 0), |
| 90 | NUM_QUERIES_FINISHED: Value("i", 0), |
| 91 | NUM_QUERIES_EXCEEDED_MEM_LIMIT: Value("i", 0), |
| 92 | NUM_QUERIES_AC_REJECTED: Value("i", 0), |
| 93 | NUM_QUERIES_AC_TIMEDOUT: Value("i", 0), |
| 94 | NUM_QUERIES_CANCELLED: Value("i", 0), |
| 95 | NUM_RESULT_MISMATCHES: Value("i", 0), |
| 96 | NUM_OTHER_ERRORS: Value("i", 0)} |
| 97 | |
| 98 | def connect(self): |
| 99 | """Connect to the server and start the query runner thread.""" |
| 100 | self.impalad_conn = self.impalad.impala.connect(impalad=self.impalad) |
| 101 | |
| 102 | def run_query(self, query, mem_limit_mb, run_set_up=False, |
| 103 | timeout_secs=maxsize, cancel_mech=None, retain_profile=False): |
| 104 | """Run a query and return an execution report. If 'run_set_up' is True, set up sql |
| 105 | will be executed before the main query. This should be the case during the binary |
| 106 | search phase of the stress test. 'cancel_mech' is optionally a CancelMechanism |
| 107 | value that should be used to cancel the query after timeout_secs. |
| 108 | If 'cancel_mech' is provided, don't get the query profile for timed out queries |
| 109 | because the query was purposely cancelled, rather than having some problem that needs |
| 110 | investigation. |
| 111 | """ |
| 112 | assert self.impalad_conn, "connect() must be called before run_query()" |
| 113 | assert cancel_mech is None or cancel_mech in CancelMechanism.ALL_MECHS |
| 114 | |
| 115 | timeout_unix_time = time() + timeout_secs |
| 116 | report = QueryReport(query) |
| 117 | try: |
| 118 | with self.impalad_conn.cursor() as cursor: |
| 119 | start_time = time() |
| 120 | self._set_db_and_options(cursor, query, run_set_up, mem_limit_mb, timeout_secs, |
| 121 | cancel_mech) |
no outgoing calls
no test coverage detected