HAWQ-1025. Refactor hawq register to implement HAWQ-1025: using actual sizes, check files number multiple of bucket number if hash table.
Project: http://git-wip-us.apache.org/repos/asf/incubator-hawq/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-hawq/commit/af12c6b3 Tree: http://git-wip-us.apache.org/repos/asf/incubator-hawq/tree/af12c6b3 Diff: http://git-wip-us.apache.org/repos/asf/incubator-hawq/diff/af12c6b3 Branch: refs/heads/master Commit: af12c6b3bfb4e676c0d960bd0282b58c1a8221b6 Parents: 2c4f25c Author: xunzhang <xunzhang...@gmail.com> Authored: Thu Sep 1 18:12:41 2016 +0800 Committer: rlei <r...@pivotal.io> Committed: Fri Sep 2 10:53:27 2016 +0800 ---------------------------------------------------------------------- src/test/feature/ManagementTool/incorrect1.yml | 1 + src/test/feature/ManagementTool/incorrect2.yml | 1 + src/test/feature/ManagementTool/incorrect3.yml | 1 + src/test/feature/ManagementTool/incorrect4.yml | 1 + src/test/feature/ManagementTool/incorrect5.yml | 1 + src/test/feature/ManagementTool/incorrect6.yml | 1 + src/test/feature/ManagementTool/incorrect7.yml | 29 +++++ src/test/feature/ManagementTool/incorrect8.yml | 28 +++++ .../ManagementTool/test_hawq_register.cpp | 12 +-- tools/bin/hawqregister | 107 +++++++++++-------- 10 files changed, 128 insertions(+), 54 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/incubator-hawq/blob/af12c6b3/src/test/feature/ManagementTool/incorrect1.yml ---------------------------------------------------------------------- diff --git a/src/test/feature/ManagementTool/incorrect1.yml b/src/test/feature/ManagementTool/incorrect1.yml index 8258a90..cfb2983 100755 --- a/src/test/feature/ManagementTool/incorrect1.yml +++ b/src/test/feature/ManagementTool/incorrect1.yml @@ -19,3 +19,4 @@ Parquet_Schema: type: int4 TableName: public.hawqregister Version: 1.0.0 +Bucketnum: 6 http://git-wip-us.apache.org/repos/asf/incubator-hawq/blob/af12c6b3/src/test/feature/ManagementTool/incorrect2.yml ---------------------------------------------------------------------- diff --git a/src/test/feature/ManagementTool/incorrect2.yml b/src/test/feature/ManagementTool/incorrect2.yml index 8258a90..cfb2983 100755 --- a/src/test/feature/ManagementTool/incorrect2.yml +++ b/src/test/feature/ManagementTool/incorrect2.yml @@ -19,3 +19,4 @@ Parquet_Schema: type: int4 TableName: public.hawqregister Version: 1.0.0 +Bucketnum: 6 http://git-wip-us.apache.org/repos/asf/incubator-hawq/blob/af12c6b3/src/test/feature/ManagementTool/incorrect3.yml ---------------------------------------------------------------------- diff --git a/src/test/feature/ManagementTool/incorrect3.yml b/src/test/feature/ManagementTool/incorrect3.yml index f45b5e5..ff56f4e 100755 --- a/src/test/feature/ManagementTool/incorrect3.yml +++ b/src/test/feature/ManagementTool/incorrect3.yml @@ -19,3 +19,4 @@ Parquet_Schema: type: int4 TableName: public.hawqregister Version: 1.0.0 +Bucketnum: 6 http://git-wip-us.apache.org/repos/asf/incubator-hawq/blob/af12c6b3/src/test/feature/ManagementTool/incorrect4.yml ---------------------------------------------------------------------- diff --git a/src/test/feature/ManagementTool/incorrect4.yml b/src/test/feature/ManagementTool/incorrect4.yml index 3b8b921..aec3a7f 100755 --- a/src/test/feature/ManagementTool/incorrect4.yml +++ b/src/test/feature/ManagementTool/incorrect4.yml @@ -19,3 +19,4 @@ Parquet_Schema: - name: i TableName: public.hawqregister Version: 1.0.0 +Bucketnum: 6 http://git-wip-us.apache.org/repos/asf/incubator-hawq/blob/af12c6b3/src/test/feature/ManagementTool/incorrect5.yml ---------------------------------------------------------------------- diff --git a/src/test/feature/ManagementTool/incorrect5.yml b/src/test/feature/ManagementTool/incorrect5.yml index 42a37ae..540744b 100755 --- a/src/test/feature/ManagementTool/incorrect5.yml +++ b/src/test/feature/ManagementTool/incorrect5.yml @@ -19,3 +19,4 @@ Encoding: UTF8 FileFormat: AO TableName: public.t1 Version: 1.0.0 +Bucketnum: 6 http://git-wip-us.apache.org/repos/asf/incubator-hawq/blob/af12c6b3/src/test/feature/ManagementTool/incorrect6.yml ---------------------------------------------------------------------- diff --git a/src/test/feature/ManagementTool/incorrect6.yml b/src/test/feature/ManagementTool/incorrect6.yml index 0d572dc..f42d0f7 100755 --- a/src/test/feature/ManagementTool/incorrect6.yml +++ b/src/test/feature/ManagementTool/incorrect6.yml @@ -20,3 +20,4 @@ Encoding: UTF8 FileFormat: AO TableName: public.t1 Version: 1.0.0 +Bucketnum: 6 http://git-wip-us.apache.org/repos/asf/incubator-hawq/blob/af12c6b3/src/test/feature/ManagementTool/incorrect7.yml ---------------------------------------------------------------------- diff --git a/src/test/feature/ManagementTool/incorrect7.yml b/src/test/feature/ManagementTool/incorrect7.yml new file mode 100644 index 0000000..d9a9185 --- /dev/null +++ b/src/test/feature/ManagementTool/incorrect7.yml @@ -0,0 +1,29 @@ +AO_FileLocations: + Blocksize: 32768 + Checksum: false + CompressionLevel: 0 + CompressionType: null + Files: + - path: /hawq_default/16385/16387/17015/1 + size: 16 + - path: /hawq_default/16385/16387/17015/2 + size: 0 + - path: /hawq_default/16385/16387/17015/3 + size: 0 + - path: /hawq_default/16385/16387/17015/4 + size: 16 + - path: /hawq_default/16385/16387/17015/5 + size: 16 +AO_Schema: +- name: i + type: int4 +DBVersion: PostgreSQL 8.2.15 (Greenplum Database 4.2.0 build 1) (HAWQ 2.0.1.0 build + dev) on x86_64-apple-darwin15.5.0, compiled by GCC Apple LLVM version 7.3.0 (clang-703.0.31) + compiled on Aug 31 2016 11:10:07 +DFS_URL: hdfs://localhost:8020 +Distribution_Policy: DISTRIBUTED BY (i) +Encoding: UTF8 +FileFormat: AO +TableName: public.t10 +Version: 1.0.0 +Bucketnum: 6 http://git-wip-us.apache.org/repos/asf/incubator-hawq/blob/af12c6b3/src/test/feature/ManagementTool/incorrect8.yml ---------------------------------------------------------------------- diff --git a/src/test/feature/ManagementTool/incorrect8.yml b/src/test/feature/ManagementTool/incorrect8.yml new file mode 100644 index 0000000..32d3b0d --- /dev/null +++ b/src/test/feature/ManagementTool/incorrect8.yml @@ -0,0 +1,28 @@ +AO_FileLocations: + Blocksize: 32768 + Checksum: false + CompressionLevel: 0 + CompressionType: null + Files: + - path: /hawq_default/16385/16387/17015/1 + size: 16 + - path: /hawq_default/16385/16387/17015/2 + size: 0 + - path: /hawq_default/16385/16387/17015/3 + size: 0 + - path: /hawq_default/16385/16387/17015/4 + size: 16 + - path: /hawq_default/16385/16387/17015/5 + size: 16 +AO_Schema: +- name: i + type: int4 +DBVersion: PostgreSQL 8.2.15 (Greenplum Database 4.2.0 build 1) (HAWQ 2.0.1.0 build + dev) on x86_64-apple-darwin15.5.0, compiled by GCC Apple LLVM version 7.3.0 (clang-703.0.31) + compiled on Aug 31 2016 11:10:07 +DFS_URL: hdfs://localhost:8020 +Distribution_Policy: DISTRIBUTED BY (i) +Encoding: UTF8 +FileFormat: AO +TableName: public.t10 +Version: 1.0.0 http://git-wip-us.apache.org/repos/asf/incubator-hawq/blob/af12c6b3/src/test/feature/ManagementTool/test_hawq_register.cpp ---------------------------------------------------------------------- diff --git a/src/test/feature/ManagementTool/test_hawq_register.cpp b/src/test/feature/ManagementTool/test_hawq_register.cpp index f7f67c0..97489b8 100644 --- a/src/test/feature/ManagementTool/test_hawq_register.cpp +++ b/src/test/feature/ManagementTool/test_hawq_register.cpp @@ -341,15 +341,11 @@ TEST_F(TestHawqRegister, TestIncorrectYaml) { EXPECT_EQ(1, Command::getCommandStatus("hawq register -d " + (string) HAWQ_DB + " -c " + filePath + "incorrect4.yml xx")); EXPECT_EQ(1, Command::getCommandStatus("hawq register -d " + (string) HAWQ_DB + " -c " + filePath + "incorrect5.yml xx")); EXPECT_EQ(1, Command::getCommandStatus("hawq register -d " + (string) HAWQ_DB + " -c " + filePath + "incorrect6.yml xx")); + EXPECT_EQ(1, Command::getCommandStatus("hawq register -d " + (string) HAWQ_DB + " -c " + filePath + "incorrect8.yml xx")); } -TEST_F(TestHawqRegister, TestCreateExistedTable) { +TEST_F(TestHawqRegister, TestDismatchFileNumber) { SQLUtility util; - util.execute("drop table if exists t10;"); - util.execute("create table t10(i int) with (appendonly=true, orientation=row) distributed by (i);"); - util.execute("insert into t10 values(1), (2), (3);"); - EXPECT_EQ(0, Command::getCommandStatus("hawq extract -d " + (string) HAWQ_DB + " -o t10.yml testhawqregister_testcreateexistedtable.t10")); - auto tmp = Command::getCommandOutput("hawq register -d " + (string) HAWQ_DB + " -c t10.yml testhawqregister_testcreateexistedtable.t10"); - auto out = hawq::test::trim(hawq::test::trimNewLine(tmp)); - EXPECT_EQ(1, hawq::test::endsWith(out, "has already existed.")); + string filePath = util.getTestRootPath() + "/ManagementTool/"; + EXPECT_EQ(1, Command::getCommandStatus("hawq register -d " + (string) HAWQ_DB + " -c " + filePath + "incorrect7.yml xx")); } http://git-wip-us.apache.org/repos/asf/incubator-hawq/blob/af12c6b3/tools/bin/hawqregister ---------------------------------------------------------------------- diff --git a/tools/bin/hawqregister b/tools/bin/hawqregister index 6b2c918..91f087a 100755 --- a/tools/bin/hawqregister +++ b/tools/bin/hawqregister @@ -17,8 +17,9 @@ # specific language governing permissions and limitations # under the License. -# Usage1: hawq register [-h hostname] [-p port] [-U username] [-d database] [-f filepath] tablename -# Usage2: hawq register [-h hostname] [-p port] [-U username] [-d database] [-c config] tablename +# Usage1: hawq register [-h hostname] [-p port] [-U username] [-d database] [-f filepath] [-e eof] <tablename> +# Usage2: hawq register [-h hostname] [-p port] [-U username] [-d database] [-c config] [--force] [--repair] <tablename> + import os, sys, optparse, getpass, re, urlparse try: from gppylib.commands.unix import getLocalHostname, getUserName @@ -44,14 +45,16 @@ def option_parser(): usage='usage: %prog [options] table_name', version='%prog version $Revision: #1 $') parser.remove_option('-h') - parser.add_option('-?', '--help', action='help') - parser.add_option('-h', '--host', help='host of the target DB') - parser.add_option('-p', '--port', help='port of the target DB', type='int', default=0) - parser.add_option('-U', '--user', help='username of the target DB') + parser.add_option('-?', '--help', action = 'help') + parser.add_option('-h', '--host', help = 'host of the target DB') + parser.add_option('-p', '--port', help = 'port of the target DB', type = 'int', default = 0) + parser.add_option('-U', '--user', help = 'username of the target DB') parser.add_option('-d', '--database', default = 'postgres', dest = 'database', help='database name') - parser.add_option('-f', '--filepath', dest = 'filepath', help='file name in HDFS') - parser.add_option('-e', '--eof', dest = 'filesizes', type='list', default = [], help = 'eof list of files to be registered') - parser.add_option('-c', '--config', dest = 'yml_config', default = '', help='configuration file in YAML format') + parser.add_option('-f', '--filepath', dest = 'filepath', help = 'file name in HDFS') + parser.add_option('-e', '--eof', dest = 'filesize', type = 'int', default = 0, help = 'eof of the file to be registered') + parser.add_option('-c', '--config', dest = 'yml_config', default = '', help = 'configuration file in YAML format') + parser.add_option('--force', action = 'store_true', default = False) + parser.add_option('--repair', action = 'store_true', default = False) return parser @@ -114,22 +117,16 @@ def option_parser_yml(yml_file): register_yaml_dict_check(params) if params['FileFormat'] == 'Parquet': if not len(params['Parquet_FileLocations']['Files']): - return 'Parquet', '', params['Parquet_Schema'], params['Distribution_Policy'], params['Parquet_FileLocations'], params['Bucketnum'] - offset = params['Parquet_FileLocations']['Files'][0]['path'].rfind('/') - filepath = (params['DFS_URL'] + params['Parquet_FileLocations']['Files'][0]['path'][:offset] - if len(params['Parquet_FileLocations']['Files']) != 1 - else params['DFS_URL'] + params['Parquet_FileLocations']['Files'][0]['path']) - return 'Parquet', filepath, params['Parquet_Schema'], params['Distribution_Policy'], params['Parquet_FileLocations'], params['Bucketnum'] + return 'Parquet', [], [], params['Parquet_Schema'], params['Distribution_Policy'], params['Parquet_FileLocations'], params['Bucketnum'] + files, sizes = [params['DFS_URL'] + d['path'] for d in params['Parquet_FileLocations']['Files']], [d['size'] for d in params['Parquet_FileLocations']['Files']] + return 'Parquet', files, sizes, params['Parquet_Schema'], params['Distribution_Policy'], params['Parquet_FileLocations'], params['Bucketnum'] if not len(params['AO_FileLocations']['Files']): - return 'AO', '', params['AO_Schema'], params['Distribution_Policy'], params['AO_FileLocations'], params['Bucketnum'] - offset = params['AO_FileLocations']['Files'][0]['path'].rfind('/') - filepath = (params['DFS_URL'] + params['AO_FileLocations']['Files'][0]['path'][:offset] - if len(params['AO_FileLocations']['Files']) != 1 - else params['DFS_URL'] + params['AO_FileLocations']['Files'][0]['path']) - return 'AO', filepath, params['AO_Schema'], params['Distribution_Policy'], params['AO_FileLocations'], params['Bucketnum'] + return 'AO', [], [], params['AO_Schema'], params['Distribution_Policy'], params['AO_FileLocations'], params['Bucketnum'] + files, sizes = [params['DFS_URL'] + d['path'] for d in params['AO_FileLocations']['Files']], [d['size'] for d in params['AO_FileLocations']['Files']] + return 'AO', files, sizes, params['AO_Schema'], params['Distribution_Policy'], params['AO_FileLocations'], params['Bucketnum'] -def create_table(dburl, tablename, schema_info, fmt, distrbution_policy, file_locations): +def create_table(dburl, tablename, schema_info, fmt, distrbution_policy, file_locations, bucket_number): try: query = "select count(*) from pg_class where relname = '%s'" % tablename.split('.')[-1].lower() conn = dbconn.connect(dburl, False) @@ -137,24 +134,26 @@ def create_table(dburl, tablename, schema_info, fmt, distrbution_policy, file_lo conn.commit() for row in rows: if row[0] != 0: - logger.error("Register failed: table %s has already existed." % tablename) - sys.exit(1) + # TODO + pass except DatabaseError, ex: logger.error('Failed to execute query "%s"' % query) sys.exit(1) + try: schema = ','.join([k['name'] + ' ' + k['type'] for k in schema_info]) fmt = 'ROW' if fmt == 'AO' else fmt if fmt == 'ROW': - query = ('create table %s(%s) with (appendonly=true, orientation=%s, compresstype=%s, compresslevel=%s, checksum=%s) %s;' - % (tablename, schema, fmt, file_locations['CompressionType'], file_locations['CompressionLevel'], file_locations['Checksum'], distrbution_policy)) + query = ('create table %s(%s) with (appendonly=true, orientation=%s, compresstype=%s, compresslevel=%s, checksum=%s, bucketnum=%s) %s;' + % (tablename, schema, fmt, file_locations['CompressionType'], file_locations['CompressionLevel'], file_locations['Checksum'], bucket_number, distrbution_policy)) else: # Parquet - query = ('create table %s(%s) with (appendonly=true, orientation=%s, compresstype=%s, compresslevel=%s, pagesize=%s, rowgroupsize=%s) %s;' - % (tablename, schema, fmt, file_locations['CompressionType'], file_locations['CompressionLevel'], file_locations['PageSize'], file_locations['RowGroupSize'], distrbution_policy)) + query = ('create table %s(%s) with (appendonly=true, orientation=%s, compresstype=%s, compresslevel=%s, pagesize=%s, rowgroupsize=%s, bucketnum=%s) %s;' + % (tablename, schema, fmt, file_locations['CompressionType'], file_locations['CompressionLevel'], file_locations['PageSize'], file_locations['RowGroupSize'], bucket_number, distrbution_policy)) conn = dbconn.connect(dburl, False) rows = dbconn.execSQL(conn, query) conn.commit() except DatabaseError, ex: + print DatabaseError, ex logger.error('Failed to execute query "%s"' % query) sys.exit(1) @@ -222,9 +221,9 @@ def get_metadata_from_database(dburl, tablename, seg_name): try: # get the full path of correspoding file for target table query = ("select location, gp_persistent_tablespace_node.tablespace_oid, database_oid, relfilenode from pg_class, gp_persistent_relation_node, " - "gp_persistent_tablespace_node, gp_persistent_filespace_node where relname = '%s' and pg_class.relfilenode = " - "gp_persistent_relation_node.relfilenode_oid and gp_persistent_relation_node.tablespace_oid = gp_persistent_tablespace_node.tablespace_oid " - "and gp_persistent_filespace_node.filespace_oid = gp_persistent_filespace_node.filespace_oid;") % tablename.split('.')[-1] + "gp_persistent_tablespace_node, gp_persistent_filespace_node where relname = '%s' and pg_class.relfilenode = " + "gp_persistent_relation_node.relfilenode_oid and gp_persistent_relation_node.tablespace_oid = gp_persistent_tablespace_node.tablespace_oid " + "and gp_persistent_filespace_node.filespace_oid = gp_persistent_filespace_node.filespace_oid;") % tablename.split('.')[-1] conn = dbconn.connect(dburl, False) rows = dbconn.execSQL(conn, query) conn.commit() @@ -328,56 +327,72 @@ def insert_metadata_into_database(dburl, databasename, tablename, seg_name, firs '''Insert the metadata into database''' try: query = "SET allow_system_table_mods='dml';" - segno = firstsegno - for eof in eofs: - query += "insert into pg_aoseg.%s values(%d, %d, %d, %d);" % (seg_name, segno, eof, -1, -1) - segno += 1 + query += 'insert into pg_aoseg.%s values(%d, %d, %d, %d)' % (seg_name, firstsegno, eofs[0], -1, -1) + for k, eof in enumerate(eofs[1:]): + query += ',(%d, %d, %d, %d)' % (firstsegno + k + 1, eof, -1, -1) + query += ';' conn = dbconn.connect(dburl, True) rows = dbconn.execSQL(conn, query) conn.commit() conn.close() except DatabaseError, ex: logger.error('Failed to connect to database, this script can only be run when the database is up') - move_files_in_hdfs(options.database, options.tablename, files, firstsegno, tabledir, False) + move_files_in_hdfs(database, tablename, files, firstsegno, tabledir, False) sys.exit(1) if __name__ == '__main__': + parser = option_parser() options, args = parser.parse_args() - if len(args) != 1 or (options.yml_config and options.filepath): + + if len(args) != 1 or ((options.yml_config or options.force or options.repair) and options.filepath) or (options.force and options.repair): parser.print_help(sys.stderr) sys.exit(1) if local_ssh('hadoop', logger): logger.error('command "hadoop" is not available.') sys.exit(1) - dburl = dbconn.DbURL(hostname=options.host, port=options.port, username=options.user, dbname=options.database) + dburl = dbconn.DbURL(hostname = options.host, port = options.port, username = options.user, dbname = options.database) filepath, database, tablename = options.filepath, options.database, args[0] if options.yml_config: # Usage2 - fileformat, filepath, schema, distribution_policy, file_locations, _ = option_parser_yml(options.yml_config) - create_table(dburl, tablename, schema, fileformat, distribution_policy, file_locations) + fileformat, files, sizes, schema, distribution_policy, file_locations, bucket_number = option_parser_yml(options.yml_config) + filepath = files[0][:files[0].rfind('/')] if files else '' + if distribution_policy.startswith('DISTRIBUTED BY'): + if len(files) % bucket_number != 0: + logger.error('Files to be registered must match the bucket number of hash table.') + sys.exit(1) + create_table(dburl, tablename, schema, fileformat, distribution_policy, file_locations, bucket_number) else: fileformat = 'Parquet' check_hash_type(dburl, tablename) # Usage1 only support randomly distributed table + + # check filepath if not filepath: sys.exit(0) + seg_name = get_seg_name(dburl, tablename, database, fileformat) firstsegno, tabledir = get_metadata_from_database(dburl, tablename, seg_name) - sizes = 0 + check_files_and_table_in_same_hdfs_cluster(filepath, tabledir) - files, sizes = get_files_in_hdfs(filepath) + + if not options.yml_config: + files, sizes = get_files_in_hdfs(filepath) print 'File(s) to be registered:', files + # set specified eofs - if options.filesizes: - if len(options.filesizes) != len(files): - logger.error('Specified eof list is incomplete.') + if options.filesize: + if options.filesize != len(files): + logger.error('-e option is only supported with single file case.') sys.exit(1) - sizes = options.sizes + sizes = [options.filesize] if fileformat == 'Parquet': check_parquet_format(files) move_files_in_hdfs(database, tablename, files, firstsegno, tabledir, True) + + # update catalog table insert_metadata_into_database(dburl, database, tablename, seg_name, firstsegno, tabledir, sizes) + logger.info('Hawq Register Succeed.')