| #!/usr/bin/env python |
| # -*- coding: utf-8 -*- |
| # Licensed to the Apache Software Foundation (ASF) under one |
| # or more contributor license agreements. See the NOTICE file |
| # distributed with this work for additional information |
| # regarding copyright ownership. The ASF licenses this file |
| # to you 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. |
| |
| |
| # import mainUtils FIRST to get python version check |
| from gppylib.mainUtils import * |
| from optparse import OptionParser |
| from Queue import Queue,Empty |
| import os |
| import shutil |
| import fnmatch |
| import tempfile |
| import time |
| from threading import Thread |
| import threading |
| from datetime import datetime |
| |
| try: |
| from gppylib import gplog, pgconf, userinput |
| from gppylib.commands.base import Command, WorkerPool, Worker |
| from gppylib.operations import Operation |
| from gppylib.gpversion import GpVersion |
| from gppylib.db import dbconn |
| from gppylib.operations.unix import CheckDir, CheckFile, MakeDir |
| from gppylib.operations.dump_analyzedb import get_partition_state, \ |
| get_pgstatlastoperations_dict, compare_metadata, compare_dict, \ |
| write_lines_to_file, verify_lines_in_file, ValidateIncludeTargets, \ |
| ValidateSchemaExists |
| from gppylib.operations.backup_utils import execute_sql, get_lines_from_file |
| |
| except ImportError, e: |
| sys.exit('Cannot import modules. Please check that you have sourced greenplum_path.sh. Detail: ' + str(e)) |
| |
| EXECNAME = 'analyzedb' |
| ANALYZE_SQL = """psql %s -p %s -c 'analyze %s'""" |
| ANALYZE_ROOT_SQL = """psql %s -p %s -c 'analyze rootpartition %s'""" |
| STATEFILE_DIR = 'db_analyze' |
| logger = gplog.get_default_logger() |
| CURR_TIME = None |
| |
| PG_PARTITIONS_SURROGATE = """ |
| SELECT n.nspname AS schemaname, cl.relname AS tablename, n2.nspname AS partitionschemaname, cl2.relname AS partitiontablename, cl3.relname AS parentpartitiontablename |
| FROM pg_namespace n, pg_namespace n2, pg_class cl, pg_class cl2, pg_partition pp, |
| pg_partition_rule pr1 |
| LEFT JOIN pg_partition_rule pr2 ON pr1.parparentrule = pr2.oid |
| LEFT JOIN pg_class cl3 ON pr2.parchildrelid = cl3.oid |
| WHERE pp.paristemplate = false AND pp.parrelid = cl.oid AND pr1.paroid = pp.oid AND cl2.oid = pr1.parchildrelid AND cl.relnamespace = n.oid AND cl2.relnamespace = n2.oid |
| """ |
| |
| GET_ALL_DATA_TABLES_SQL = """ select n.nspname as schemaname, c.relname as tablename from pg_class c, pg_namespace n where |
| c.relnamespace = n.oid and c.relkind='r'::char and (c.relnamespace > 16384 or n.nspname = 'public' or n.nspname = 'pg_catalog') and c.oid not in (select reloid from pg_exttable) |
| EXCEPT |
| select distinct schemaname, tablename from (%s) AS pps1 |
| EXCEPT |
| select distinct partitionschemaname, parentpartitiontablename from (%s) AS pps2 where parentpartitiontablename is not NULL |
| """ % (PG_PARTITIONS_SURROGATE, PG_PARTITIONS_SURROGATE) |
| |
| GET_VALID_DATA_TABLES_SQL = """ |
| select n.nspname as schemaname, c.relname as tablename from pg_class c, pg_namespace n where |
| c.relnamespace = n.oid and c.oid in (%s) and c.relkind='r'::char and (c.relnamespace > 16384 or n.nspname = 'public' or n.nspname = 'pg_catalog') and c.oid not in (select reloid from pg_exttable) |
| """ |
| |
| GET_REQUESTED_AO_DATA_TABLE_INFO_SQL = """ |
| SELECT ALL_DATA_TABLES.oid, ALL_DATA_TABLES.schemaname, ALL_DATA_TABLES.tablename, OUTER_PG_CLASS.relname as tupletable FROM |
| ( |
| select c.oid as oid, n.nspname as schemaname, c.relname as tablename from pg_class c, pg_namespace n where |
| c.relnamespace = n.oid and c.oid in (%s) |
| ) as ALL_DATA_TABLES, pg_appendonly, pg_class OUTER_PG_CLASS |
| WHERE ALL_DATA_TABLES.oid = pg_appendonly.relid |
| AND OUTER_PG_CLASS.oid = pg_appendonly.segrelid |
| """ |
| |
| GET_REQUESTED_LAST_OP_INFO_SQL = """ |
| SELECT PGN.nspname, PGC.relname, objid, staactionname, stasubtype, statime FROM pg_stat_last_operation, pg_class PGC, pg_namespace PGN |
| WHERE objid = PGC.oid |
| AND objid in (%s) |
| AND PGC.relnamespace = PGN.oid |
| AND staactionname IN ('CREATE', 'ALTER', 'TRUNCATE') |
| ORDER BY objid, staactionname |
| """ |
| |
| GET_ALL_DATA_TABLES_IN_SCHEMA_SQL = """ |
| select n.nspname as schemaname, c.relname as tablename from pg_class c, pg_namespace n where |
| c.relnamespace = n.oid and c.relkind='r'::char and (c.relnamespace > 16384 or n.nspname = 'public' or n.nspname = 'pg_catalog') and c.oid not in (select reloid from pg_exttable) |
| and n.nspname = '%s' |
| EXCEPT |
| select distinct schemaname, tablename from (%s) AS pps1 |
| EXCEPT |
| select distinct partitionschemaname, parentpartitiontablename from (%s) AS pps2 where parentpartitiontablename is not NULL |
| """ % ('%s', PG_PARTITIONS_SURROGATE, PG_PARTITIONS_SURROGATE) |
| |
| GET_LEAF_PARTITIONS_SQL = """ |
| select partitionschemaname || '.' || partitiontablename from (%s) AS pps1 where schemaname = '%s' and tablename = '%s' |
| EXCEPT |
| select distinct partitionschemaname || '.' || parentpartitiontablename from (%s) AS pps2 where parentpartitiontablename is not NULL |
| and schemaname = '%s' and tablename = '%s' |
| """ % (PG_PARTITIONS_SURROGATE, '%s', '%s', PG_PARTITIONS_SURROGATE, '%s', '%s') |
| |
| GET_MID_LEVEL_PARTITIONS_SQL = """ |
| select distinct partitionschemaname, parentpartitiontablename from (%s) AS pps1 where parentpartitiontablename is not NULL |
| """ % PG_PARTITIONS_SURROGATE |
| |
| GET_REQUESTED_NON_AO_TABLES_SQL = """ |
| select n.nspname as schemaname, c.relname as tablename from pg_class c, pg_namespace n where |
| c.relnamespace = n.oid and c.relkind='r'::char and (c.relnamespace > 16384 or n.nspname = 'public' or n.nspname = 'pg_catalog') |
| and c.oid not in (select relid from pg_appendonly) and c.oid in (%s) and c.oid not in (select reloid from pg_exttable) |
| EXCEPT |
| select distinct schemaname, tablename from (%s) AS pps1 |
| EXCEPT |
| select distinct partitionschemaname, parentpartitiontablename from (%s) AS pps2 where parentpartitiontablename is not NULL |
| """ % ('%s', PG_PARTITIONS_SURROGATE, PG_PARTITIONS_SURROGATE) |
| |
| GET_COLUMN_NAMES_SQL = """ |
| SELECT attname FROM pg_attribute WHERE attrelid = '%s'::regclass AND attnum > 0 AND NOT attisdropped |
| """ |
| |
| GET_INCLUDED_COLUMNS_FROM_EXCLUDE_SQL = """ |
| SELECT attname FROM pg_attribute WHERE attrelid = '%s'::regclass AND attname NOT IN (%s) AND attnum > 0 AND NOT attisdropped |
| """ |
| |
| VALIDATE_COLUMN_NAMES_SQL = """ |
| SELECT count(*) FROM pg_attribute WHERE attrelid = '%s'::regclass AND attname IN (%s) AND attnum > 0 AND NOT attisdropped |
| """ |
| |
| VALIDATE_TABLE_NAMES_SQL = """ |
| SELECT %s |
| """ |
| |
| GET_LEAF_ROOT_MAPPING_SQL = """ |
| SELECT n.nspname || '.' || c2.relname, n.nspname || '.' || c.relname from pg_class c, pg_class c2, pg_namespace n, pg_partition pp, pg_partition_rule ppr |
| WHERE ppr.parchildrelid in (%s) AND ppr.paroid = pp.oid AND pp.parrelid = c.oid AND c.relnamespace = n.oid AND ppr.parchildrelid = c2.oid; |
| """ |
| |
| GET_ALL_ROOT_PARTITION_TABLES_SQL = """ |
| SELECT distinct n.nspname || '.' || c.relname from pg_class c, pg_namespace n, pg_partition pp |
| WHERE pp.parrelid = c.oid AND c.relnamespace = n.oid AND pp.paristemplate = false |
| """ |
| |
| ORDER_CANDIDATES_BY_OID_SQL = """ |
| SELECT schemaname, tablename FROM |
| (SELECT c.oid as tableoid, n.nspname as schemaname, c.relname as tablename FROM pg_class c, pg_namespace n where c.relnamespace=n.oid and c.oid in (%s)) AS foo |
| ORDER BY tableoid DESC; |
| """ |
| |
| |
| class AnalyzeDb(Operation): |
| def __init__(self, options, args): |
| |
| if args: |
| logger.warn("Please note that some of the arguments (%s) aren't valid and will be ignored.", args) |
| if options.masterDataDirectory is None: |
| options.masterDataDirectory = gp.get_masterdatadir() |
| self.master_datadir = options.masterDataDirectory |
| self.analyze_dir = STATEFILE_DIR |
| self.pg_port = gp.get_master_port() |
| self.dbname = options.dbname |
| self.schema = options.schema |
| self.single_table = options.single_table |
| self.config_file = options.config_file |
| self.entire_db = False |
| self.include_cols = options.include_cols |
| self.exclude_cols = options.exclude_cols |
| self.full_analyze = options.full_analyze |
| self.dry_run = options.dry_run |
| self.parallel_level = options.parallel_level |
| self.rootstats = True |
| self.silent = options.silent |
| self.verbose = options.verbose |
| self.clean_last = options.clean_last |
| self.clean_all = options.clean_all |
| |
| self.success_list = [] |
| |
| self._validate_options() |
| self._preprocess_options() |
| |
| self.conn = dbconn.connect(dbconn.DbURL(port=self.pg_port, dbname=self.dbname), utility=False) |
| |
| def _validate_options(self): |
| """ |
| Validates the options passed in to the application. |
| """ |
| if not self.dbname: |
| raise ProgramArgumentValidationException("option -d required. Please see 'analyzedb -?' for usage.") |
| |
| if self.clean_all + self.clean_last > 1: |
| raise ProgramArgumentValidationException('options --clean_last and --clean_all are mutually exclusive') |
| |
| if (self.schema is not None) + (self.single_table is not None) \ |
| + (self.config_file is not None) > 1: |
| raise ProgramArgumentValidationException('options -s, -t and -f are mutually exclusive') |
| |
| if (self.include_cols is not None) + (self.exclude_cols is not None) > 1: |
| raise ProgramArgumentValidationException('options -i and -x are mutually exclusive') |
| |
| if (self.include_cols is not None or self.exclude_cols is not None) and self.single_table is None: |
| raise ProgramArgumentValidationException('option -i or -x can only be used together with -t') |
| |
| if self.parallel_level < 1 or self.parallel_level > 10: |
| raise ProgramArgumentValidationException('option -p requires a value between 1 and 10') |
| |
| |
| def _preprocess_options(self): |
| |
| if self.clean_all: |
| analyze_folder = os.path.join(self.master_datadir, self.analyze_dir, self.dbname) |
| if os.path.exists(analyze_folder): |
| if not self.silent and not userinput.ask_yesno(None, "\nDeleting all files and folders in %s ?" % analyze_folder, 'N'): |
| raise UserAbortedException() |
| for f in os.listdir(analyze_folder): |
| f_path = os.path.join(analyze_folder, f) |
| if os.path.isdir(f_path): |
| shutil.rmtree(f_path) |
| else: |
| os.remove(f_path) |
| else: |
| logger.warning("Folder %s does not exist. Exiting...") |
| |
| if self.clean_last: |
| last_analyze_timestamp = get_lastest_analyze_timestamp(self.master_datadir, self.analyze_dir, self.dbname) |
| if last_analyze_timestamp is not None: |
| analyze_folder = os.path.join(self.master_datadir, self.analyze_dir, self.dbname, last_analyze_timestamp) |
| if os.path.exists(analyze_folder): |
| if not self.silent and not userinput.ask_yesno(None, "\nDeleting folder %s and all files inside?" % analyze_folder, 'N'): |
| raise UserAbortedException() |
| shutil.rmtree(analyze_folder) |
| else: |
| logger.warning("Folder %s does not exist. Exiting...") |
| else: |
| logger.warning("No valid state files directories exist. Exiting...") |
| |
| if self.include_cols is not None: |
| self.include_cols = self.include_cols.strip().split(',') |
| |
| if self.exclude_cols is not None: |
| self.exclude_cols = self.exclude_cols.strip().split(',') |
| |
| if self.single_table is None and self.schema is None and self.config_file is None: |
| self.entire_db = True |
| |
| if self.verbose: |
| logger.setLevel(10) |
| |
| def execute(self): |
| try: |
| if self.clean_all or self.clean_last: |
| return 0 |
| |
| global CURR_TIME |
| CURR_TIME = generate_timestamp() |
| |
| # The input_col_dict keeps track of the requested columns to analyze. |
| # key: table name (e.g. 'public.foo') |
| # value: a set of requested column names (e.g. set(['col1','col2']), or '-1' indicating all columns |
| input_col_dict = {} |
| |
| # parse input and update input column dictionary |
| input_tables = self._get_input_tables(input_col_dict) # ['public.foo', 'public.bar', ...] |
| input_tables_set = set(input_tables) |
| |
| if len(input_tables_set) == 0: |
| logger.warning("There are no tables or partitions to be analyzed. Exiting...") |
| return 0 |
| |
| logger.info("Checking for tables with stale stats...") |
| # get all heap tables in the requested tables. all heap tables are regarded as dirty |
| heap_partitions = get_heap_tables_set(self.conn, input_tables_set) # set(['schema.table1', ...]) |
| |
| # get the current state of the requested tables |
| # curr_ao_state contains the number of DML commands that have been executed on an AO table |
| # curr_last_op contains the timestamp of the last DDL command (CREATE, ALTER, TRUNCATE, DROP) of an AO table |
| curr_ao_state = self._get_ao_state(input_tables_set) # ['schema, table, modcount', ...] |
| curr_last_op = self._get_lastop_state(input_tables_set) # e.g. ['public,ao_tab,67468,ALTER,ADD COLUMN,2014-10-15 14:49:27.658777-07', ...] |
| last_analyze_timestamp = get_lastest_analyze_timestamp(self.master_datadir, self.analyze_dir, self.dbname) |
| |
| # get the previous state of the database |
| prev_ao_state = get_prev_ao_state(last_analyze_timestamp, self.master_datadir, self.analyze_dir, self.dbname) |
| prev_last_op = get_prev_last_op(last_analyze_timestamp, self.master_datadir, self.analyze_dir, self.dbname) |
| |
| # compare two states to get dirty tables |
| dirty_partitions = self._get_dirty_data_tables(heap_partitions, curr_ao_state, curr_last_op, prev_ao_state, prev_last_op) |
| |
| # get the previous column states |
| # read from the file on disk which contains info about what columns had up-to-date stats after the last time the table was analyzed |
| prev_col_dict = get_prev_col_state(last_analyze_timestamp, self.master_datadir, self.analyze_dir, self.dbname) |
| |
| candidates = set() # set(['public.foo', 'public.bar', ...]) |
| |
| if self.full_analyze: # full analyze does not support column-level increments and will invalidate previous column-level state |
| candidates = input_tables_set |
| else: # incremental |
| for table in input_tables_set: |
| if table in dirty_partitions: # for dirty partitions, we invalidate all previous column-level state |
| candidates.add(table) |
| else: |
| # figure out which columns need to be analyzed |
| self._update_input_col_dict_with_column_increments(table, input_col_dict, prev_col_dict) |
| if len(input_col_dict[table]) > 0: |
| candidates.add(table) |
| |
| candidates = self._get_valid_candidates(candidates) |
| if len(candidates) < 1: |
| logger.warning("There are no tables or partitions to be analyzed. Exiting...") |
| return 0 |
| |
| root_partition_col_dict = {} |
| # root_partition_col_dict contains the mapping between the root partitions |
| # and its corresponding columns to be analyzed |
| # key: name of the root partition whose stats needs to be refreshed |
| # value: a set of column names to be analyzed, or '-1' meaning all columns of that table |
| if self.rootstats: |
| root_partition_col_dict = self._get_root_partition_col_dict(candidates, input_col_dict) |
| |
| ordered_candidates = self._get_ordered_candidates(candidates, root_partition_col_dict) |
| target_list = [] |
| logger.info("---------------------------------------------------") |
| logger.info("Tables or partitions to be analyzed") |
| logger.info("---------------------------------------------------") |
| for can in ordered_candidates: |
| if can in candidates: |
| target = self._get_tablename_with_cols(can, input_col_dict) |
| else: # can in root_partition_col_dict |
| target = self._get_tablename_with_cols(can, root_partition_col_dict) |
| logger.info(target) |
| target_list.append(target) |
| logger.info("---------------------------------------------------") |
| |
| if self.dry_run: |
| return 0 |
| |
| if not self.silent and not userinput.ask_yesno(None, "\nContinue with Analyze", 'N'): |
| raise UserAbortedException() |
| |
| num_workers = min(self.parallel_level, len(ordered_candidates)) |
| logger.info("Starting analyze with %d workers..." % num_workers) |
| pool = AnalyzeWorkerPool(numWorkers=num_workers) |
| for can in ordered_candidates: |
| if can in candidates: |
| target = self._get_tablename_with_cols(can, input_col_dict) |
| cmd = Command('analyze %s' % target, ANALYZE_SQL % (self.dbname, self.pg_port, quote_tbl(target))) |
| else: # can in root_partition_col_dict |
| target = self._get_tablename_with_cols(can, root_partition_col_dict) |
| cmd = Command('analyze rootpartition %s' % target, ANALYZE_ROOT_SQL % (self.dbname, self.pg_port, quote_tbl(target))) |
| pool.addCommand(cmd) |
| |
| wait_count = len(ordered_candidates) |
| try: |
| start_time = time.time() |
| while wait_count > 0: |
| done_cmd = pool.completed_queue.get() |
| if done_cmd.was_successful(): |
| subject = self._get_tablename_from_cmd_name(done_cmd.name) |
| self.success_list.append(subject) |
| if wait_count % 10 == 0: |
| logger.info("progress status: completed %d out of %d tables or partitions" % |
| (len(self.success_list), len(ordered_candidates))) |
| wait_count -= 1 |
| |
| for worker in pool.workers: |
| worker.join(1) |
| except: |
| pool.haltWork() |
| for worker in pool.workers: |
| worker.join(1) |
| raise |
| |
| finally: |
| self._write_back(curr_ao_state, curr_last_op, prev_ao_state, prev_last_op, heap_partitions, |
| input_col_dict, prev_col_dict, root_partition_col_dict, self.full_analyze, dirty_partitions, |
| target_list) |
| end_time = time.time() |
| logger.info("Total elapsed time: %d seconds. Analyzed %d out of %d table(s) or partition(s) successfully." |
| % (int(end_time-start_time), len(self.success_list), len(ordered_candidates))) |
| logger.info("Done.") |
| except: |
| raise |
| |
| finally: |
| if self.conn: |
| self.conn.close() |
| |
| return 0 |
| |
| def _get_input_tables(self, col_dict): |
| """ |
| Depending on the way this program was invoked, gather all the requested tables to be analyzed. |
| At the same time, parse the requested columns and populate the col_dict. |
| If a requested table is partitioned, expand all the leaf partitions. |
| """ |
| logger.info("Getting and verifying input tables...") |
| if self.single_table: |
| validate_include_tables(self.conn, [self.single_table], None) |
| # for single table, we always try to expand it to avoid getting all root partitions in the database |
| self._parse_column(col_dict, self.single_table, self.include_cols, self.exclude_cols, True) |
| |
| elif self.schema: # all tables in a schema |
| ValidateSchemaExists(self.dbname, self.schema, self.pg_port).run() |
| logger.debug("getting all tables in the schema...") |
| all_schema_tables = run_sql(self.conn, GET_ALL_DATA_TABLES_IN_SCHEMA_SQL % self.schema) |
| # convert table name from ['public','foo'] to 'public.foo' and populate col_dict as all columns requested |
| for table in all_schema_tables: |
| col_dict['.'.join(table)] = set(['-1']) |
| |
| elif self.config_file: |
| validate_include_tables(self.conn, None, self.config_file) |
| all_root_partitions = run_sql(self.conn, GET_ALL_ROOT_PARTITION_TABLES_SQL) |
| cfg_file = open(self.config_file, 'rU') |
| for line in cfg_file: |
| toks = line.strip().split() |
| table = toks[0] |
| included_cols = self._get_include_or_exclude_cols(toks, '-i') |
| excluded_cols = self._get_include_or_exclude_cols(toks, '-x') |
| self._parse_column(col_dict, table, included_cols, excluded_cols, [table] in all_root_partitions) |
| |
| else: # all user tables in database |
| alltables = run_sql(self.conn, GET_ALL_DATA_TABLES_SQL) |
| for table in alltables: |
| col_dict['.'.join(table)] = set(['-1']) |
| |
| return col_dict.keys() |
| |
| def _get_include_or_exclude_cols(self, line_tokens, option_str): |
| """ |
| Get included or excluded columns from a line in the config file |
| :param line_tokens: tokenized lien in the config file, e.g., ['<schema>.<table>', '-i', '<col1>,<col2>,...'] |
| :param option_str: 'include' or 'exclude' |
| :return: a list of included or excluded columns |
| """ |
| pos = line_tokens.index(option_str) if option_str in line_tokens else -1 |
| if pos < 0: |
| cols = None |
| else: |
| if pos+1 >= len(line_tokens): |
| raise Exception("No %s columns specified." % option_str) |
| cols = line_tokens[pos+1].split(',') |
| |
| return cols |
| |
| def _get_valid_candidates(self, candidates): |
| """ |
| Validate candidates for the following: |
| 1. check whether candidates set is empty |
| 2. check invalid characters in table names |
| 3. skip views and external tables |
| """ |
| ret = [] |
| mid_level_partitions = run_sql(self.conn, GET_MID_LEVEL_PARTITIONS_SQL) |
| for can in candidates: |
| if '\n' in can or ',' in can or ':' in can: |
| raise Exception('Table name has an invalid character "\\n", ":", "," : "%s"' % can) |
| if can.split('.') in mid_level_partitions: |
| logger.warning("Skipping mid-level partition %s" % can) |
| else: |
| ret.append(can) |
| |
| if self.config_file is not None or self.single_table is not None: |
| valid_tables = [] |
| if len(ret) > 0: |
| oid_str = get_oid_str(ret) |
| qresult = run_sql(self.conn, GET_VALID_DATA_TABLES_SQL % oid_str) |
| for tbl in qresult: |
| valid_tables.append('.'.join(tbl)) |
| return valid_tables |
| |
| return ret |
| |
| def _get_dirty_data_tables(self, heap_tables_set, curr_ao_state, curr_last_op, prev_ao_state, prev_last_op): |
| """ |
| dirty data tables include: |
| - heap tables |
| - ao tables that have gone through DML or DDL |
| """ |
| logger.debug("getting dirty data tables...") |
| dirty_heap_tables = heap_tables_set |
| dirty_ao_tables = self._get_dirty_ao_state_tables(curr_ao_state, prev_ao_state) |
| dirty_metadata_set = self._get_dirty_lastop_tables(curr_last_op, prev_last_op) |
| return dirty_heap_tables | dirty_ao_tables | dirty_metadata_set |
| |
| def _get_ao_state(self, input_tables_set): |
| logger.debug("getting ao state...") |
| oid_str = get_oid_str(input_tables_set) |
| ao_partition_info = run_sql(self.conn, GET_REQUESTED_AO_DATA_TABLE_INFO_SQL % oid_str) |
| return get_partition_state(self.pg_port, self.dbname, 'pg_aoseg', ao_partition_info) |
| |
| def _get_lastop_state(self, input_tables_set): |
| # oid, action, subtype, timestamp |
| logger.debug("getting last operation states...") |
| oid_str = get_oid_str(input_tables_set) |
| rows = run_sql(self.conn, GET_REQUESTED_LAST_OP_INFO_SQL % oid_str) |
| data = [] |
| for row in rows: |
| line = "%s,%s,%d,%s,%s,%s" % (row[0], row[1], row[2], row[3], row[4], row[5]) |
| data.append(line) |
| return data |
| |
| def _write_back(self, curr_ao_state, curr_last_op, prev_ao_state, prev_last_op, heap_partitions, |
| input_col_dict, prev_col_dict, root_partition_col_dict, is_full, dirty_partitions, target_list): |
| validate_dir("%s/%s/%s/%s" % (self.master_datadir, self.analyze_dir, self.dbname, CURR_TIME)) |
| |
| curr_ao_state_dict = create_ao_state_dict(curr_ao_state) |
| curr_last_op_dict = create_last_op_dict(curr_last_op) |
| prev_ao_state_dict = create_ao_state_dict(prev_ao_state) |
| prev_last_op_dict = create_last_op_dict(prev_last_op) |
| |
| for table in (x for x in self.success_list if x not in heap_partitions and x not in root_partition_col_dict): |
| # update modcount for tables that are successfully analyzed |
| new_modcount = curr_ao_state_dict[table] |
| prev_ao_state_dict[table] = new_modcount |
| |
| # update last op for tables that are successfully analyzed |
| last_op_info = curr_last_op_dict[table] # {'CREATE':'<entry>', 'ALTER':'<entry>', ...} |
| prev_last_op_dict[table] = last_op_info |
| |
| # update column dict |
| if is_full or table in dirty_partitions or table not in prev_col_dict or '-1' in input_col_dict[table]: |
| prev_col_dict[table] = input_col_dict[table] |
| else: |
| prev_col_dict[table] = prev_col_dict[table] | input_col_dict[table] |
| |
| ao_state_output = construct_entries_from_dict_aostate(prev_ao_state_dict) |
| last_op_output = construct_entries_from_dict_lastop(prev_last_op_dict) |
| col_state_output = construct_entries_from_dict_colstate(prev_col_dict) |
| |
| if len(ao_state_output) > 0: |
| ao_state_filename = generate_statefile_name('ao', self.master_datadir, self.analyze_dir, self.dbname, CURR_TIME) |
| logger.info("Writing ao state file %s" % ao_state_filename) |
| write_lines_to_file(ao_state_filename, ao_state_output) |
| logger.debug("Verifying ao state file ...") |
| verify_lines_in_file(ao_state_filename, ao_state_output) |
| |
| if len(last_op_output) > 0: |
| last_operation_filename = generate_statefile_name('lastop', self.master_datadir, self.analyze_dir, self.dbname, CURR_TIME) |
| logger.info("Writing last operation file %s" % last_operation_filename) |
| write_lines_to_file(last_operation_filename, last_op_output) |
| logger.debug("Verifying last operation file ...") |
| verify_lines_in_file(last_operation_filename, last_op_output) |
| |
| if len(prev_col_dict) > 0: |
| col_state_filename = generate_statefile_name('col', self.master_datadir, self.analyze_dir, self.dbname, CURR_TIME) |
| logger.info("Writing column state file %s" % col_state_filename) |
| write_lines_to_file(col_state_filename, col_state_output) |
| logger.debug("Verifying column state ...") |
| verify_lines_in_file(col_state_filename, col_state_output) |
| |
| report_filename = generate_statefile_name('report', self.master_datadir, self.analyze_dir, self.dbname, CURR_TIME) |
| logger.info("Writing report file %s" % report_filename) |
| with open(report_filename, 'w') as fp: |
| fp.write("%s:%s:%s:%s %s:%s:analyzedb %s\n\n" % (CURR_TIME[:8], CURR_TIME[8:10], CURR_TIME[10:12], CURR_TIME[12:14], |
| unix.getLocalHostname(), unix.getUserName(), ' '.join(sys.argv[1:]))) |
| fp.write("Tables or partitions to analyze:\n---------------------------------------\n") |
| for target in target_list: |
| fp.write("%s\n" % target.strip()) |
| fp.write("\n\nTables or partitions successfully analyzed:\n--------------------------------------------\n") |
| for tbl in self.success_list: |
| fp.write("%s\n" % tbl.strip()) |
| fp.write("\n%d out of %d tables are analyzed.\n" % (len(self.success_list), len(target_list))) |
| if len(target_list) == len(self.success_list): |
| fp.write("\nanalyzedb finished successfully.\n") |
| |
| def _get_dirty_lastop_tables(self, curr_last_op, prev_last_op): |
| old_pgstatoperations_dict = get_pgstatlastoperations_dict(prev_last_op) |
| dirty_tables = compare_metadata(old_pgstatoperations_dict, curr_last_op) |
| return dirty_tables |
| |
| def _get_dirty_ao_state_tables(self, curr_ao_state, prev_ao_state): |
| last_state_dict = create_ao_state_dict(prev_ao_state) |
| curr_state_dict = create_ao_state_dict(curr_ao_state) |
| return compare_dict(last_state_dict, curr_state_dict) |
| |
| def _parse_column(self, col_dict, table, include_cols, exclude_cols, is_root_partition): |
| """ |
| Given a list of included or excluded columns of a table, populate the column dictionary. |
| If the table is partitioned, expand it into all leaf partitions. |
| If both include_cols and exclude_cols are empty, use '-1' as the value indicating 'all columns'. |
| """ |
| if include_cols is not None: |
| validate_columns(self.conn, table, include_cols) |
| elif exclude_cols is not None: |
| validate_columns(self.conn, table, exclude_cols) |
| included_column_set = get_include_cols_from_exclude(self.conn, table, exclude_cols) |
| if len(included_column_set) == 0: |
| raise Exception("All columns have been excluded from table %s" % table) |
| if is_root_partition: |
| logger.debug("expanding partition tables...") |
| tbl_parts = self._expand_partition_tables([table]) |
| else: |
| tbl_parts = [table] |
| |
| for tbl in tbl_parts: |
| if include_cols is not None: |
| col_dict[tbl] = set(include_cols) |
| elif exclude_cols is not None: |
| col_dict[tbl] = included_column_set |
| else: # all columns |
| col_dict[tbl] = set(['-1']) |
| |
| def _update_input_col_dict_with_column_increments(self, table, input_col_dict, prev_col_dict): |
| if table in prev_col_dict: |
| # since expanding the default '-1' to all column name is expensive, we avoid this as much as possible |
| if '-1' not in input_col_dict[table] or '-1' not in prev_col_dict[table]: |
| input_col_set = self._expand_columns(input_col_dict, table) |
| prev_col_set = self._expand_columns(prev_col_dict, table) |
| pending_cols = input_col_set - prev_col_set # set difference |
| input_col_dict[table] = pending_cols |
| else: # both previous and current runs are without column specification |
| input_col_dict[table] = set() |
| |
| def _get_tablename_with_cols(self, table, col_dict): |
| if '-1' in col_dict[table]: |
| return table |
| else: |
| return table + '(' + ','.join(sorted(col_dict[table])) + ')' |
| |
| def _get_tablename_from_cmd_name(self, cmdName): |
| if '(' in cmdName: |
| subject = cmdName.split()[-1].split('(')[0] |
| else: |
| subject = cmdName.split()[-1] |
| |
| return subject |
| |
| def _expand_partition_tables(self, table_list): |
| ret = [] |
| for table in table_list: |
| (schema, parent) = table.split('.') |
| qresult = run_sql(self.conn, GET_LEAF_PARTITIONS_SQL % (schema, parent, schema, parent)) |
| child_parts = [x[0] for x in qresult] |
| if len(child_parts) == 0: |
| ret.append(table) |
| else: |
| ret += child_parts |
| |
| return ret |
| |
| def _get_root_partition_col_dict(self, candidates, input_col_dict): |
| """ |
| Examine the candidates and figure out the root partitions whose stats need refreshing and |
| what columns need to be analyzed on those root partitions. |
| If the program is invoked on whole schema or whole database, then we know all columns have |
| been requested. Thus we can use one query to obtain the root partitions associated with the |
| candidates. |
| If the program is invoked by '-t' or '-f', we need to either look up the partition_dict or |
| issue a query to get the leaf-root relationship. Then the columns to be analyzed on the root |
| level are the set union of the columns to be analyzed for all leaf partitions. |
| """ |
| logger.debug("getting mapping between leaf and root partition tables...") |
| ret = {} |
| # The leaf_root_dict keeps track of the mapping between a leaf partition and its root partition |
| # for the use of refreshing root stats. |
| leaf_root_dict = {} |
| oid_str = get_oid_str(candidates) |
| qresult = run_sql(self.conn, GET_LEAF_ROOT_MAPPING_SQL % oid_str) |
| for mapping in qresult: |
| leaf_root_dict[mapping[0]] = mapping[1] |
| |
| for can in candidates: |
| if can in leaf_root_dict: # this is a leaf partition |
| if leaf_root_dict[can] not in ret: |
| ret[leaf_root_dict[can]] = input_col_dict[can].copy() |
| else: |
| ret[leaf_root_dict[can]] |= input_col_dict[can].copy() |
| ## TODO: do we need column expansion here? |
| return ret |
| |
| def _get_ordered_candidates(self, candidates, root_partition_col_dict): |
| """ |
| Take all tables in candidates and root_partition_col_dict and order them |
| by descending order of their OIDs. This gives us two important benefits: |
| 1. The root partition will be analyzed right after the leaves |
| 2. The leaf partitions (if range partitioned, especially by date) will be ordered in descending |
| order of the partition key, so that newer partitions can be analyzed first. |
| """ |
| candidate_regclass_str = get_oid_str(candidates+root_partition_col_dict.keys()) |
| qresult = run_sql(self.conn, ORDER_CANDIDATES_BY_OID_SQL % candidate_regclass_str) |
| return ['.'.join(x) for x in qresult] |
| |
| def _expand_columns(self, col_dict, table): |
| if '-1' in col_dict[table]: |
| cols = run_sql(self.conn, GET_COLUMN_NAMES_SQL % quote_tbl(table)) |
| return set([x[0] for x in cols]) |
| else: |
| return col_dict[table] |
| |
| |
| def run_sql(conn, query): |
| try: |
| cursor = dbconn.execSQL(conn, query) |
| res = cursor.fetchall() |
| except Exception, db_err: |
| raise ExceptionNoStackTraceNeeded("%s" % db_err.__str__()) #.split('\n')[0]) |
| cursor.close() |
| return res |
| |
| def generate_timestamp(): |
| timestamp = datetime.now() |
| return timestamp.strftime("%Y%m%d%H%M%S") |
| |
| def quote_tbl(target): |
| """ |
| Quote schema names and table names before pass them to query strings. |
| If target contains column name, e.g. public.foo(a,b), quote column name separately. |
| E.g. public.foo(a,b) will become "public"."foo"("a","b") |
| """ |
| sch, tbl_raw = target.split('.') |
| cols_str = '' |
| if '(' in tbl_raw: |
| tbl = tbl_raw.split('(')[0] |
| cols = tbl_raw.split('(')[1][:-1].split(',') |
| cols_str = '(' + ','.join(['"%s"' % x for x in cols]) + ')' |
| else: |
| tbl = tbl_raw |
| |
| return '"%s"."%s"' % (sch, tbl) + cols_str |
| |
| |
| def get_oid_str(table_list): |
| return ','.join(["""'\"%s\".\"%s\"'::regclass""" % (x.split('.')[0], x.split('.')[1]) for x in table_list]) |
| |
| |
| def get_heap_tables_set(conn, input_tables_set): |
| logger.debug("getting heap tables...") |
| oid_str = get_oid_str(input_tables_set) |
| dirty_tables = set() |
| qresult = run_sql(conn, GET_REQUESTED_NON_AO_TABLES_SQL % oid_str) |
| for row in qresult: |
| dirty_tables.add('.'.join(row)) |
| return dirty_tables |
| |
| def get_lastest_analyze_timestamp(master_datadir, statefile_dir, dbname): |
| analyze_dirs = get_analyze_dirs(master_datadir, statefile_dir, dbname) |
| |
| for analyze_dir in analyze_dirs: |
| files = sorted(os.listdir(analyze_dir)) |
| |
| if len(files) == 0: |
| logger.warn('Analyze state file directory %s is empty. Ignoring this directory...' % analyze_dir) |
| continue |
| |
| analyze_report_files = fnmatch.filter(files, 'analyze_[0-9][0-9][0-9][0-9][0-9][0-9][0-9][0-9][0-9][0-9][0-9][0-9][0-9][0-9]_report') |
| |
| if len(analyze_report_files) == 0: |
| logger.warn('No analyze report files found in analyze directory %s. Ignoring this directory...' % analyze_dir) |
| continue |
| for report_file in analyze_report_files: |
| return report_file.split('_')[1] |
| |
| return None |
| |
| def get_prev_ao_state(timestamp, master_datadir, analyze_dir, dbname): |
| logger.debug("getting previous ao state...") |
| prev_state_filename = generate_statefile_name('ao', master_datadir, analyze_dir, dbname, timestamp) |
| if not os.path.isfile(prev_state_filename): |
| return [] |
| return get_lines_from_file(prev_state_filename) |
| |
| def get_prev_last_op(timestamp, master_datadir, analyze_dir, dbname): |
| logger.debug("getting previous last operation...") |
| old_pgstatoperations_file = generate_statefile_name('lastop', master_datadir, analyze_dir, dbname, timestamp) |
| if not os.path.isfile(old_pgstatoperations_file): |
| old_pgstatoperations = [] |
| else: |
| old_pgstatoperations = get_lines_from_file(old_pgstatoperations_file) |
| return old_pgstatoperations |
| |
| def get_prev_col_state(timestamp, master_datadir, analyze_dir, dbname): |
| logger.debug("getting previous column states...") |
| prev_col_state_file = generate_statefile_name('col', master_datadir, analyze_dir, dbname, timestamp) |
| if not os.path.isfile(prev_col_state_file): |
| return {} |
| lines = get_lines_from_file(prev_col_state_file) |
| prev_col_dict = {} |
| for line in lines: |
| toks = line.strip().split(',') |
| toks = map(str.strip, toks) |
| prev_col_dict['.'.join(toks[:2])] = set(toks[2:]) |
| |
| return prev_col_dict |
| |
| def create_ao_state_dict(ao_state_entries): |
| ao_state_dict = dict() |
| for entry in ao_state_entries: |
| fields = entry.split(',') |
| if len(fields) != 3: |
| raise Exception('Invalid ao state file format %s' % entry) |
| key = '%s.%s' % (fields[0].strip(), fields[1].strip()) |
| ao_state_dict[key] = fields[2].strip() |
| |
| return ao_state_dict |
| |
| def create_last_op_dict(last_op_entries): |
| last_op_dict = {} |
| for entry in last_op_entries: |
| toks = entry.split(',') |
| if len(toks) != 6: |
| raise Exception('Wrong number of tokens in last_operation data for last backup: "%s"' % entry) |
| key = '%s.%s' % (toks[0],toks[1]) |
| op = toks[3] |
| if key not in last_op_dict: |
| last_op_dict[key] = {op:entry} |
| else: |
| last_op_dict[key][op] = entry |
| |
| return last_op_dict |
| |
| def construct_entries_from_dict_aostate(ao_state_dict): |
| ret = [] |
| for key, item in ao_state_dict.iteritems(): |
| token = ','.join(key.split('.')) |
| ret.append(','.join([token, item])) |
| return ret |
| |
| def construct_entries_from_dict_lastop(last_op_dict): |
| ret = [] |
| for value in last_op_dict.itervalues(): |
| for entry in value.itervalues(): |
| ret.append(entry) |
| return ret |
| |
| def construct_entries_from_dict_colstate(prev_col_dict): |
| ret = [] |
| for table, col_set in prev_col_dict.iteritems(): |
| cols = ','.join(col_set) |
| ret.append(','.join([','.join(table.split('.')), cols])) # public,foo,a,b,c |
| return ret |
| |
| def generate_statefile_name(type_str, master_data_dir, analyze_dir, dbname, timestamp): |
| use_dir = "%s/%s/%s/%s" % (master_data_dir, analyze_dir, dbname, timestamp) |
| if type_str == 'lastop': |
| ret_str = "%s/analyze_%s_last_operation" |
| elif type_str == 'ao': |
| ret_str = "%s/analyze_%s_ao_state_file" |
| elif type_str == 'col': |
| ret_str = "%s/analyze_%s_col_state_file" |
| elif type_str == 'report': |
| ret_str = "%s/analyze_%s_report" |
| else: |
| raise Exception("Invalid type string for generating state file name") |
| return ret_str % (use_dir, timestamp) |
| |
| def get_analyze_dirs(master_datadir, statefile_dir, dbname): |
| analyze_path = os.path.join(master_datadir, statefile_dir, dbname) |
| |
| if not os.path.isdir(analyze_path): |
| return [] |
| |
| initial_list = os.listdir(analyze_path) |
| initial_list = fnmatch.filter(initial_list, '[0-9][0-9][0-9][0-9][0-9][0-9][0-9][0-9][0-9][0-9][0-9][0-9][0-9][0-9]') |
| |
| dirnames = [] |
| for d in initial_list: |
| pth = os.path.join(analyze_path, d) |
| if os.path.isdir(pth): |
| dirnames.append(pth) |
| |
| if len(dirnames) == 0: |
| return [] |
| dirnames = sorted(dirnames, key=lambda x: int(os.path.basename(x)), reverse=True) |
| return dirnames |
| |
| def validate_dir(path): |
| exists = CheckDir(path).run() |
| if exists: |
| logger.info("Directory %s exists" % path) |
| else: |
| try: |
| MakeDir(path).run() |
| except OSError, e: |
| logger.exception("Could not create directory %s" % path) |
| raise AnalyzeDirCreateFailed() |
| else: |
| logger.info("Created %s" % path) |
| try: |
| with tempfile.TemporaryFile(dir=path) as f: |
| pass |
| except Exception, e: |
| logger.exception("Cannot write to %s" % path) |
| raise AnalyzeDirNotWritable() |
| |
| def validate_include_tables(conn, table_list, include_file): |
| """ |
| table_list needs to be a list of 'schema.table's |
| include_file could contain columns specified by -i |
| """ |
| tables = {} |
| if table_list is not None: |
| for tbl in table_list: |
| if '.' not in tbl: |
| raise ExceptionNoStackTraceNeeded("No schema name supplied for table %s" % tbl) |
| tables[tbl] = 0 |
| |
| if include_file is not None: |
| in_file = open(include_file, 'rU') |
| line_no = 1 |
| for line in in_file: |
| toks = line.strip().split() |
| if len(toks) == 0: # emtpy line |
| continue |
| if len(toks) > 3: # we are expecting <schema>.<table> --in(ex)clude_column <col1>,<col2>,... |
| raise ExceptionNoStackTraceNeeded("Wrong input arguments in line %d of config file. Please check usage." % line_no) |
| if '.' not in toks[0]: |
| raise ExceptionNoStackTraceNeeded("No schema name supplied for table %s" % toks[0]) |
| if toks[0] in tables: |
| raise ExceptionNoStackTraceNeeded("Duplicate table name in line %d of config file." % line_no) |
| tables[toks[0]] = 0 |
| line_no += 1 |
| |
| # since the number of target entries cannot be greater than 1664 in GPDB/HAWQ, |
| # we validate the tables in batches of 1500 |
| tablenames = tables.keys() |
| batch_size = 1500 |
| nbatches = (len(tables)-1)/batch_size + 1 |
| curr_batch = 0 |
| |
| while curr_batch < nbatches: |
| oid_str = get_oid_str(tablenames[curr_batch*batch_size:(curr_batch+1)*batch_size]) |
| run_sql(conn, VALIDATE_TABLE_NAMES_SQL % oid_str) |
| curr_batch += 1 |
| |
| def get_include_cols_from_exclude(conn, table, exclude_cols): |
| """ |
| Given a list of excluded columns of a table, get the list of included columns |
| """ |
| quoted_exclude_cols = ','.join(["'%s'" % x for x in exclude_cols]) |
| cols = run_sql(conn, GET_INCLUDED_COLUMNS_FROM_EXCLUDE_SQL % (quote_tbl(table), quoted_exclude_cols)) |
| |
| return set([x[0] for x in cols]) |
| |
| def validate_columns(conn, table, column_list): |
| """ |
| Check whether all column names in a list are valid for a table |
| """ |
| if len(column_list) == 0: |
| return |
| valid_col_count = dbconn.execSQLForSingleton(conn, VALIDATE_COLUMN_NAMES_SQL % (quote_tbl(table), ','.join(["'%s'" % x for x in column_list]))) |
| |
| if int(valid_col_count) != len(column_list): |
| raise Exception("Invalid input columns for table %s." % table) |
| |
| def create_parser(): |
| parser = OptionParser(version='%prog version 1.0', |
| description= |
| "Analyze a database incrementally. 'Incremental' means if a table or partition has not been modified by" |
| "DML or DDL commands since the last analyzedb run, it will be automatically skipped since its statistics" |
| "must be up to date. Some restrictions apply: " |
| "1. The incremental semantics only applies to append-only tables or partitions. All heap tables are regarded" |
| "as having stale stats every time analyzedb is run. This is because we use AO metadata to check for DML or" |
| "DDL events, which is not available to heap tables. " |
| "2. Views, indices and external tables are automatically skipped. " |
| "3. Table names or schema names containing comma or period is not supported yet." |
| ) |
| parser.set_usage('%prog [options] ') |
| parser.remove_option('-h') |
| |
| parser.add_option('-d', dest='dbname', metavar="<database name>", |
| help="Database name. Required.") |
| parser.add_option('-s', dest='schema', metavar="<schema name>", |
| help="Specify a schema to analyze. All tables in the schema will be analyzed.") |
| parser.add_option('-t', dest='single_table', metavar="<schema name>.<table name>", |
| help="Analyze a single table. Table name needs to be qualified with schema name.") |
| parser.add_option('-i', type='string', dest='include_cols', metavar="<column1>,<column2>,...", |
| help="Columns to include to be analyzed, separated by comma. All columns will be analyzed if not specified.") |
| parser.add_option('-x', type='string', dest='exclude_cols', metavar="<column1>,<column2>,...", |
| help="Columns to exclude to be analyzed, separated by comma. All columns will be analyzed if not specified.") |
| parser.add_option('-f', '--file', dest='config_file', metavar="<config_file>", |
| help="Config file that includes a list of tables to be analyzed. " |
| "Table names must be qualified with schema name. Optionally a list of columns (separated by " |
| "comma) can be specified using -i or -x.") |
| parser.add_option('-l', '--list', action='store_true', dest='dry_run', default=False, |
| help="List the tables to be analyzed without actually running analyze (dry run).") |
| parser.add_option('-p', type='int', dest='parallel_level', default=5, metavar="<parallel level>", |
| help="Parallel level, i.e. the number of tables to be analyzed in parallel. Valid numbers are between 1 and 10. Default value is 5.") |
| parser.add_option('--full', action='store_true', dest='full_analyze', default=False, |
| help="Analyze without using incremental. All tables requested by the user will be analyzed.") |
| parser.add_option('--clean_last', action='store_true', dest='clean_last', default=False, |
| help="Clean the state files generated by last analyzedb run. All other options except -d will be ignored.") |
| parser.add_option('--clean_all', action='store_true', dest='clean_all', default=False, |
| help="Clean all the state files generated by analyzedb. All other options except -d will be ignored.") |
| parser.add_option('-h', '-?', '--help', action='help', |
| help='Show this help message and exit.') |
| parser.add_option('-v', '--verbose', action='store_true', dest='verbose', help='Print debug messages.') |
| parser.add_option('-a', action='store_true', dest='silent', default=False, |
| help="Quiet mode. Do not prompt for user confirmation.") |
| |
| return parser |
| |
| class AnalyzeDirCreateFailed(Exception): pass |
| class AnalyzeDirNotWritable(Exception): pass |
| |
| class AnalyzeWorkerPool(WorkerPool): |
| """ |
| a custom worker pool for analyze workers |
| """ |
| def __init__(self, numWorkers=5, items=None): |
| self.workers=[] |
| self.work_queue=Queue() |
| self.completed_queue=Queue() |
| self.num_assigned=0 |
| if items is not None: |
| for item in items: |
| self.work_queue.put(item) |
| self.num_assigned += 1 |
| |
| for i in range(0,numWorkers): |
| # use AnalyzeWorker instead of Worker |
| w = AnalyzeWorker("worker%d" % i,self) |
| self.workers.append(w) |
| w.start() |
| self.numWorkers = numWorkers |
| self.logger = logger |
| |
| |
| class AnalyzeWorker(Thread): |
| """ |
| a custom worker thread for Analyze |
| """ |
| pool=None |
| cmd=None |
| name=None |
| logger=None |
| |
| def __init__(self,name,pool,timeout=0.05): |
| self.name=name |
| self.pool=pool |
| self.timeout=timeout |
| self.logger=logger |
| self.stoprequest = threading.Event() |
| Thread.__init__(self) |
| |
| def run(self): |
| while not self.stoprequest.isSet(): |
| try: |
| self.cmd = self.pool.getNextWorkItem(timeout=self.timeout) |
| |
| if self.cmd is not None: |
| # tablename = self.cmd.cmdStr.split()[-1][:-1] |
| self.logger.info("[%s] started %s" % (self.name, self.cmd.name)) |
| start_time = time.time() |
| self.cmd.run() |
| end_time = time.time() |
| stderr = self.cmd.get_stderr_lines() |
| if len(stderr) > 0: # emit stderr if there is any |
| self.logger.warning('\n'.join(stderr)) |
| if self.cmd.was_successful(): |
| self.logger.info("[%s] finished %s. Elapsed time: %d seconds." % (self.name, self.cmd.name, |
| int(end_time-start_time))) |
| self.pool.addFinishedWorkItem(self.cmd) |
| self.cmd=None |
| |
| except Empty: |
| return |
| |
| except: |
| if self.cmd: |
| self.logger.debug("[%s] finished cmd with exception: %s" % (self.name, self.cmd)) |
| self.pool.addFinishedWorkItem(self.cmd) |
| self.cmd=None |
| raise |
| |
| def haltWork(self): |
| self.stoprequest.set() |
| c = self.cmd |
| if c is not None: |
| c.interrupt() |
| c.cancel() |
| |
| |
| if __name__ == '__main__': |
| sys.argv[0] = EXECNAME |
| simple_main(create_parser, AnalyzeDb) |
