You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@hawq.apache.org by wcl14 <gi...@git.apache.org> on 2016/10/01 14:09:40 UTC
[GitHub] incubator-hawq pull request #950: HAWQ-991. Refator HAWQ Register code for p...
GitHub user wcl14 opened a pull request:
https://github.com/apache/incubator-hawq/pull/950
HAWQ-991. Refator HAWQ Register code for partition table.
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/wcl14/incubator-hawq HAWQ-991
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/incubator-hawq/pull/950.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #950
----
commit 4f9222792ef4dc0e9c9d130752faaa9e178aa8a0
Author: Chunling Wang <wa...@126.com>
Date: 2016-10-01T14:06:51Z
HAWQ-991. Refator HAWQ Register code for partition table.
----
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-hawq pull request #950: HAWQ-991. Refator HAWQ Register code for p...
Posted by xunzhang <gi...@git.apache.org>.
Github user xunzhang commented on a diff in the pull request:
https://github.com/apache/incubator-hawq/pull/950#discussion_r82498881
--- Diff: tools/bin/hawqregister ---
@@ -325,250 +326,221 @@ class HawqRegister(object):
return 'usage2_table_not_exist'
else:
if not table_existed():
- logger.error('Table %s does not exist.\nYou should create table before registering the data.' % self.tablename)
+ logger.error('Table %s does not exist.\nYou should create table before registering the data.' % self.dst_table_name)
sys.exit(1)
else:
return 'usage1'
- def _init(self):
- def check_hash_type():
- self.accessor.check_hash_type(self.tablename)
-
- # check conflicting distributed policy
- def check_distribution_policy():
- if self.distribution_policy.startswith('DISTRIBUTED BY'):
- if len(self.files) % self.bucket_number != 0:
- logger.error('Files to be registered must be multiple times to the bucket number of hash table.')
- self.failure_handler.rollback()
- sys.exit(1)
+ def check_hash_type(self):
+ self.accessor.check_hash_type(self.dst_table_name)
- def check_database_encoding():
- encoding_indx = self.accessor.get_database_encoding_indx(self.database)
- encoding = self.accessor.get_database_encoding(encoding_indx)
- if self.encoding.strip() != encoding:
- logger.error('Database encoding from yaml configuration file(%s) is not consistent with encoding from input args(%s).' % (self.encoding, encoding))
- sys.exit(1)
-
- def create_table():
- try:
- (ret, query) = self.accessor.do_create_table(self.src_table_name, self.tablename, self.schema, self.file_format, self.distribution_policy, self.file_locations, self.bucket_number,
+ def create_table(self):
+ try:
+ (ret, query) = self.accessor.do_create_table(self.src_table_name, self.dst_table_name, self.schema, self.file_format, self.distribution_policy, self.file_locations, self.bucket_number,
self.partitionby, self.partitions_constraint, self.partitions_name)
- except pg.DatabaseError as e:
- print e
+ except pg.DatabaseError as e:
+ print e
+ sys.exit(1)
+ if ret:
+ self.failure_handler.commit(('SQL', query))
+ return ret
+
+ def check_database_encoding(self):
+ encoding_indx = self.accessor.get_database_encoding_indx(self.database)
+ encoding = self.accessor.get_database_encoding(encoding_indx)
+ if self.encoding.strip() != encoding:
+ logger.error('Database encoding from yaml configuration file(%s) is not consistent with encoding from input args(%s).' % (self.encoding, encoding))
+ sys.exit(1)
+
+ def check_policy_consistency(self):
+ policy = self.get_distribution_policy() # "" or "{1,3}"
+ if policy is None:
+ return
+ if self.distribution_policy == 'DISTRIBUTED RANDOMLY':
+ logger.error('Distribution policy of %s from yaml is not consistent with the policy of existing table.' % self.tablename)
+ self.failure_handler.rollback()
+ sys.exit(1)
+ tmp_dict = {}
+ for i, d in enumerate(self.schema):
+ tmp_dict[d['name']] = i + 1
+ # 'DISTRIBUETD BY (1,3)' -> {1,3}
+ cols = self.distribution_policy.strip().split()[-1].strip('(').strip(')').split(',')
+ original_policy = ','.join([str(tmp_dict[col]) for col in cols])
+ if policy.strip('{').strip('}') != original_policy:
+ logger.error('Distribution policy of %s from yaml file is not consistent with the policy of existing table.' % self.dst_table_name)
+ self.failure_handler.rollback()
+ sys.exit(1)
+
+ def set_yml_dataa(self, file_format, files, sizes, tablename, schema, distribution_policy, file_locations,\
+ bucket_number, partitionby, partitions_constraint, partitions_name, partitions_compression_level,\
+ partitions_compression_type, partitions_checksum, partitions_filepaths, partitions_filesizes, encoding):
+ self.file_format = file_format
+ self.files = files
+ self.sizes = sizes
+ self.src_table_name = tablename
+ self.schema = schema
+ self.distribution_policy = distribution_policy
+ self.file_locations = file_locations
+ self.bucket_number = bucket_number
+ self.partitionby = partitionby
+ self.partitions_constraint = partitions_constraint
+ self.partitions_name = partitions_name
+ self.partitions_compression_level = partitions_compression_level
+ self.partitions_compression_type = partitions_compression_type
+ self.partitions_checksum = partitions_checksum
+ self.partitions_filepaths = partitions_filepaths
+ self.partitions_filesizes = partitions_filesizes
+ self.encoding = encoding
+
+ def option_parser_yml(self, yml_file):
+ import yaml
+ with open(yml_file, 'r') as f:
+ params = yaml.load(f)
+ table_column_num = self.accessor.get_table_column_num(self.tablename)
+ register_yaml_dict_check(params, table_column_num, self.tablename)
+ partitions_filepaths = []
+ partitions_filesizes = []
+ partitions_constraint = []
+ partitions_name = []
+ partitions_checksum = []
+ partitions_compression_level = []
+ partitions_compression_type = []
+ files, sizes = [], []
+
+ if params['FileFormat'].lower() == 'parquet':
+ Format = 'Parquet'
+ else: #AO format
+ Format = 'AO'
+ Format_FileLocations = '%s_FileLocations' % Format
+ partitionby = params.get(Format_FileLocations).get('PartitionBy')
+ if partitionby:
+ logger.info('Partition table is not supported in current release of hawq register.')
+ sys.exit(0)
+ if params.get(Format_FileLocations).get('Partitions') and len(params[Format_FileLocations]['Partitions']):
+ partitions_checksum = [d['Checksum'] for d in params[Format_FileLocations]['Partitions']]
+ partitions_compression_level = [d['CompressionLevel'] for d in params[Format_FileLocations]['Partitions']]
+ partitions_compression_type = [d['CompressionType'] for d in params[Format_FileLocations]['Partitions']]
+ partitions_constraint = [d['Constraint'] for d in params[Format_FileLocations]['Partitions']]
+ partitions_files = [d['Files'] for d in params[Format_FileLocations]['Partitions']]
+ if len(partitions_files):
+ for pfile in partitions_files:
+ partitions_filepaths.append([params['DFS_URL'] + item['path'] for item in pfile])
+ partitions_filesizes.append([item['size'] for item in pfile])
+ partitions_name = [d['Name'] for d in params[Format_FileLocations]['Partitions']]
+ if len(params[Format_FileLocations]['Files']):
+ files, sizes = [params['DFS_URL'] + d['path'] for d in params[Format_FileLocations]['Files']], [d['size'] for d in params[Format_FileLocations]['Files']]
+ encoding = params['Encoding']
+ self.set_yml_dataa(Format, files, sizes, params['TableName'], params['%s_Schema' % Format], params['Distribution_Policy'], params[Format_FileLocations], params['Bucketnum'], partitionby,\
+ partitions_constraint, partitions_name, partitions_compression_level, partitions_compression_type, partitions_checksum, partitions_filepaths, partitions_filesizes, encoding)
+
+
+ # check conflicting distributed policy
+ def check_distribution_policy(self):
+ if self.distribution_policy.startswith('DISTRIBUTED BY'):
+ if len(self.files) % self.bucket_number != 0:
+ logger.error('Files to be registered must be multiple times to the bucket number of hash table.')
+ self.failure_handler.rollback()
sys.exit(1)
- if ret:
- self.failure_handler.commit(('SQL', query))
- return ret
- def get_seg_name():
- return self.utility_accessor.get_seg_name(self.tablename, self.database, self.file_format)
+ def get_seg_name(self):
+ return self.utility_accessor.get_seg_name(self.tablename, self.database, self.file_format)
- def get_metadata():
- return self.accessor.get_metadata_from_database(self.tablename, self.seg_name)
+ def get_metadata(self):
+ return self.accessor.get_metadata_from_database(self.tablename, self.seg_name)
- def get_metadata_from_table():
- return self.accessor.get_metadata_from_seg_name(self.seg_name)
+ def get_metadata_from_table(self):
+ return self.accessor.get_metadata_from_seg_name(self.seg_name)
- def get_distribution_policy():
- return self.accessor.get_distribution_policy_info(self.tablename)
+ def get_distribution_policy(self):
+ return self.accessor.get_distribution_policy_info(self.tablename)
- def check_policy_consistency():
- policy = get_distribution_policy() # "" or "{1,3}"
- if policy is None:
- return
- if self.distribution_policy == 'DISTRIBUTED RANDOMLY':
- logger.error('Distribution policy of %s from yaml is not consistent with the policy of existing table.' % self.tablename)
- self.failure_handler.rollback()
+ def check_bucket_number(self):
+ def get_bucket_number():
+ return self.accessor.get_bucket_number(self.tablename)
+
+ if self.bucket_number != get_bucket_number():
+ logger.error('Bucket number of %s is not consistent with previous bucket number.' % self.tablename)
+ self.failure_handler.rollback()
+ sys.exit(1)
+
+ def check_file_not_folder(self):
+ for fn in self.files:
+ hdfscmd = 'hadoop fs -test -f %s' % fn
+ if local_ssh(hdfscmd, logger):
+ logger.info('%s is not a file in hdfs, please check the yaml configuration file.' % fn)
sys.exit(1)
- tmp_dict = {}
- for i, d in enumerate(self.schema):
- tmp_dict[d['name']] = i + 1
- # 'DISTRIBUETD BY (1,3)' -> {1,3}
- cols = self.distribution_policy.strip().split()[-1].strip('(').strip(')').split(',')
- original_policy = ','.join([str(tmp_dict[col]) for col in cols])
- if policy.strip('{').strip('}') != original_policy:
- logger.error('Distribution policy of %s from yaml file is not consistent with the policy of existing table.' % self.tablename)
+
+ def check_sizes_valid(self):
+ for sz in self.sizes:
+ if type(sz) != type(1):
+ logger.error('File size(%s) in yaml configuration file should be int type.' % sz)
self.failure_handler.rollback()
sys.exit(1)
-
- def check_bucket_number():
- def get_bucket_number():
- return self.accessor.get_bucket_number(self.tablename)
-
- if self.bucket_number != get_bucket_number():
- logger.error('Bucket number of %s is not consistent with previous bucket number.' % self.tablename)
+ if sz < 0:
+ logger.error('File size(%s) in yaml configuration file should not be less than 0.' % sz)
self.failure_handler.rollback()
sys.exit(1)
-
- def set_yml_dataa(file_format, files, sizes, tablename, schema, distribution_policy, file_locations,\
- bucket_number, partitionby, partitions_constraint, partitions_name, partitions_compression_level,\
- partitions_compression_type, partitions_checksum, partitions_filepaths, partitions_filesizes, encoding):
- self.file_format = file_format
- self.files = files
- self.sizes = sizes
- self.src_table_name = tablename
- self.schema = schema
- self.distribution_policy = distribution_policy
- self.file_locations = file_locations
- self.bucket_number = bucket_number
- self.partitionby = partitionby
- self.partitions_constraint = partitions_constraint
- self.partitions_name = partitions_name
- self.partitions_compression_level = partitions_compression_level
- self.partitions_compression_type = partitions_compression_type
- self.partitions_checksum = partitions_checksum
- self.partitions_filepaths = partitions_filepaths
- self.partitions_filesizes = partitions_filesizes
- self.encoding = encoding
-
- def option_parser_yml(yml_file):
- import yaml
- try:
- with open(yml_file, 'r') as f:
- params = yaml.load(f)
- except yaml.scanner.ScannerError as e:
- print e
+ for k, fn in enumerate(self.files):
+ hdfscmd = 'hadoop fs -du %s' % fn
+ _, out, _ = local_ssh_output(hdfscmd)
+ if self.sizes[k] > int(out.strip().split()[0]):
+ logger.error('File size(%s) in yaml configuration file should not exceed actual length(%s) of file %s.' % (self.sizes[k], out.strip().split()[0], fn))
self.failure_handler.rollback()
sys.exit(1)
-
- table_column_num = self.accessor.get_table_column_num(self.tablename)
- register_yaml_dict_check(params, table_column_num, self.tablename)
- partitions_filepaths = []
- partitions_filesizes = []
- partitions_constraint = []
- partitions_name = []
- partitions_checksum = []
- partitions_compression_level = []
- partitions_compression_type = []
- files, sizes = [], []
-
- if params['FileFormat'].lower() == 'parquet':
- partitionby = params.get('Parquet_FileLocations').get('PartitionBy')
- if partitionby:
- logger.info('Partition table is not supported in current release of hawq register.')
- sys.exit(0)
- if params.get('Parquet_FileLocations').get('Partitions') and len(params['Parquet_FileLocations']['Partitions']):
- partitions_checksum = [d['Checksum'] for d in params['Parquet_FileLocations']['Partitions']]
- partitions_compression_level = [d['CompressionLevel'] for d in params['Parquet_FileLocations']['Partitions']]
- partitions_compression_type = [d['CompressionType'] for d in params['Parquet_FileLocations']['Partitions']]
- partitions_constraint = [d['Constraint'] for d in params['Parquet_FileLocations']['Partitions']]
- partitions_files = [d['Files'] for d in params['Parquet_FileLocations']['Partitions']]
- if len(partitions_files):
- for pfile in partitions_files:
- partitions_filepaths.append([params['DFS_URL'] + item['path'] for item in pfile])
- partitions_filesizes.append([item['size'] for item in pfile])
- partitions_name = [d['Name'] for d in params['Parquet_FileLocations']['Partitions']]
- if len(params['Parquet_FileLocations']['Files']):
- files, sizes = [params['DFS_URL'] + d['path'] for d in params['Parquet_FileLocations']['Files']], [d['size'] for d in params['Parquet_FileLocations']['Files']]
- encoding = params['Encoding']
- set_yml_dataa('Parquet', files, sizes, params['TableName'], params['Parquet_Schema'], params['Distribution_Policy'], params['Parquet_FileLocations'], params['Bucketnum'], partitionby,\
- partitions_constraint, partitions_name, partitions_compression_level, partitions_compression_type, partitions_checksum, partitions_filepaths, partitions_filesizes, encoding)
-
- else: #AO format
- partitionby = params.get('AO_FileLocations').get('PartitionBy')
- if partitionby:
- logger.info('Partition table is not supported in current release of hawq register.')
- sys.exit(0)
- if params.get('AO_FileLocations').get('Partitions') and len(params['AO_FileLocations']['Partitions']):
- partitions_checksum = [d['Checksum'] for d in params['AO_FileLocations']['Partitions']]
- partitions_compressionLevel = [d['CompressionLevel'] for d in params['AO_FileLocations']['Partitions']]
- partitions_compressionType = [d['CompressionType'] for d in params['AO_FileLocations']['Partitions']]
- partitions_constraint = [d['Constraint'] for d in params['AO_FileLocations']['Partitions']]
- partitions_files = [d['Files'] for d in params['AO_FileLocations']['Partitions']]
- if len(partitions_files):
- for pfile in partitions_files:
- partitions_filepaths.append([params['DFS_URL'] + item['path'] for item in pfile])
- partitions_filesizes.append([item['size'] for item in pfile])
- partitions_name = [d['Name'] for d in params['AO_FileLocations']['Partitions']]
- if len(params['AO_FileLocations']['Files']):
- files, sizes = [params['DFS_URL'] + d['path'] for d in params['AO_FileLocations']['Files']], [d['size'] for d in params['AO_FileLocations']['Files']]
- encoding = params['Encoding']
- set_yml_dataa('AO', files, sizes, params['TableName'], params['AO_Schema'], params['Distribution_Policy'], params['AO_FileLocations'], params['Bucketnum'], partitionby, partitions_constraint,\
- partitions_name, partitions_compression_level, partitions_compression_type, partitions_checksum, partitions_filepaths, partitions_filesizes, encoding)
-
- def check_file_not_folder():
- for fn in self.files:
- hdfscmd = 'hadoop fs -test -f %s' % fn
- if local_ssh(hdfscmd, logger):
- logger.info('%s is not a file in hdfs, please check the yaml configuration file.' % fn)
- sys.exit(1)
- def is_folder(filepath):
- hdfscmd = 'hadoop fs -test -d %s' % filepath
- if local_ssh(hdfscmd, logger):
- return False
- else:
- return True
-
- def check_sizes_valid():
- for sz in self.sizes:
- if type(sz) != type(1):
- logger.error('File size(%s) in yaml configuration file should be int type.' % sz)
- self.failure_handler.rollback()
- sys.exit(1)
- if sz < 0:
- logger.error('File size(%s) in yaml configuration file should not be less than 0.' % sz)
+ def check_no_regex_filepath(self, files):
+ for fn in files:
+ tmp_lst = fn.split('/')
+ for v in tmp_lst:
+ if v == '.':
+ logger.error('Hawq register does not support file path with regex: %s.' % fn)
self.failure_handler.rollback()
sys.exit(1)
- for k, fn in enumerate(self.files):
- hdfscmd = 'hadoop fs -du %s' % fn
- _, out, _ = local_ssh_output(hdfscmd)
- if self.sizes[k] > int(out.strip().split()[0]):
- logger.error('File size(%s) in yaml configuration file should not exceed actual length(%s) of file %s.' % (self.sizes[k], out.strip().split()[0], fn))
+ for ch in ['..', '*']:
+ if fn.find(ch) != -1:
+ logger.error('Hawq register does not support file path with regex: %s.' % fn)
self.failure_handler.rollback()
sys.exit(1)
- def check_no_regex_filepath(files):
- for fn in files:
- tmp_lst = fn.split('/')
- for v in tmp_lst:
- if v == '.':
- logger.error('Hawq register does not support file path with regex: %s.' % fn)
- self.failure_handler.rollback()
- sys.exit(1)
- for ch in ['..', '*']:
- if fn.find(ch) != -1:
- logger.error('Hawq register does not support file path with regex: %s.' % fn)
- self.failure_handler.rollback()
- sys.exit(1)
-
+ def _init(self):
if self.yml:
- option_parser_yml(options.yml_config)
+ self.option_parser_yml(options.yml_config)
self.filepath = self.files[0][:self.files[0].rfind('/')] if self.files else ''
- check_file_not_folder()
- check_database_encoding()
+ self.check_file_not_folder()
+ self.check_database_encoding()
if self.mode != 'repair':
- if not create_table() and self.mode != 'force':
+ if not self.create_table() and self.mode != 'force':
self.mode = 'usage2_table_exist'
- check_bucket_number()
- check_distribution_policy()
- check_policy_consistency()
- check_no_regex_filepath(self.files)
else:
- if is_folder(self.filepath) and self.filesize:
- logger.error('-e option is only supported with single file case.')
- sys.exit(1)
self.file_format = 'Parquet'
- check_hash_type() # Usage1 only support randomly distributed table
+ self.check_hash_type() # Usage1 only support randomly distributed table
--- End diff --
Here your original code is not up-to-date, please merge with latest code. (https://github.com/apache/incubator-hawq/blob/master/tools/bin/hawqregister#L548)
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-hawq pull request #950: HAWQ-991. Refator HAWQ Register code for p...
Posted by linwen <gi...@git.apache.org>.
Github user linwen commented on a diff in the pull request:
https://github.com/apache/incubator-hawq/pull/950#discussion_r82497532
--- Diff: tools/bin/hawqregister ---
@@ -325,250 +326,221 @@ class HawqRegister(object):
return 'usage2_table_not_exist'
else:
if not table_existed():
- logger.error('Table %s does not exist.\nYou should create table before registering the data.' % self.tablename)
+ logger.error('Table %s does not exist.\nYou should create table before registering the data.' % self.dst_table_name)
sys.exit(1)
else:
return 'usage1'
- def _init(self):
- def check_hash_type():
- self.accessor.check_hash_type(self.tablename)
-
- # check conflicting distributed policy
- def check_distribution_policy():
- if self.distribution_policy.startswith('DISTRIBUTED BY'):
- if len(self.files) % self.bucket_number != 0:
- logger.error('Files to be registered must be multiple times to the bucket number of hash table.')
- self.failure_handler.rollback()
- sys.exit(1)
+ def check_hash_type(self):
+ self.accessor.check_hash_type(self.dst_table_name)
- def check_database_encoding():
- encoding_indx = self.accessor.get_database_encoding_indx(self.database)
- encoding = self.accessor.get_database_encoding(encoding_indx)
- if self.encoding.strip() != encoding:
- logger.error('Database encoding from yaml configuration file(%s) is not consistent with encoding from input args(%s).' % (self.encoding, encoding))
- sys.exit(1)
-
- def create_table():
- try:
- (ret, query) = self.accessor.do_create_table(self.src_table_name, self.tablename, self.schema, self.file_format, self.distribution_policy, self.file_locations, self.bucket_number,
+ def create_table(self):
+ try:
+ (ret, query) = self.accessor.do_create_table(self.src_table_name, self.dst_table_name, self.schema, self.file_format, self.distribution_policy, self.file_locations, self.bucket_number,
self.partitionby, self.partitions_constraint, self.partitions_name)
- except pg.DatabaseError as e:
- print e
+ except pg.DatabaseError as e:
+ print e
+ sys.exit(1)
+ if ret:
+ self.failure_handler.commit(('SQL', query))
+ return ret
+
+ def check_database_encoding(self):
+ encoding_indx = self.accessor.get_database_encoding_indx(self.database)
+ encoding = self.accessor.get_database_encoding(encoding_indx)
+ if self.encoding.strip() != encoding:
+ logger.error('Database encoding from yaml configuration file(%s) is not consistent with encoding from input args(%s).' % (self.encoding, encoding))
+ sys.exit(1)
+
+ def check_policy_consistency(self):
+ policy = self.get_distribution_policy() # "" or "{1,3}"
+ if policy is None:
+ return
+ if self.distribution_policy == 'DISTRIBUTED RANDOMLY':
+ logger.error('Distribution policy of %s from yaml is not consistent with the policy of existing table.' % self.tablename)
+ self.failure_handler.rollback()
+ sys.exit(1)
+ tmp_dict = {}
+ for i, d in enumerate(self.schema):
+ tmp_dict[d['name']] = i + 1
+ # 'DISTRIBUETD BY (1,3)' -> {1,3}
+ cols = self.distribution_policy.strip().split()[-1].strip('(').strip(')').split(',')
+ original_policy = ','.join([str(tmp_dict[col]) for col in cols])
+ if policy.strip('{').strip('}') != original_policy:
+ logger.error('Distribution policy of %s from yaml file is not consistent with the policy of existing table.' % self.dst_table_name)
+ self.failure_handler.rollback()
+ sys.exit(1)
+
+ def set_yml_dataa(self, file_format, files, sizes, tablename, schema, distribution_policy, file_locations,\
--- End diff --
Why uses set_yml_dataa not set_yml_data? What does the additional "a" stand for?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-hawq pull request #950: HAWQ-991. Refator HAWQ Register code for p...
Posted by xunzhang <gi...@git.apache.org>.
Github user xunzhang commented on a diff in the pull request:
https://github.com/apache/incubator-hawq/pull/950#discussion_r82499108
--- Diff: tools/bin/hawqregister ---
@@ -325,250 +326,221 @@ class HawqRegister(object):
return 'usage2_table_not_exist'
else:
if not table_existed():
- logger.error('Table %s does not exist.\nYou should create table before registering the data.' % self.tablename)
+ logger.error('Table %s does not exist.\nYou should create table before registering the data.' % self.dst_table_name)
sys.exit(1)
else:
return 'usage1'
- def _init(self):
- def check_hash_type():
- self.accessor.check_hash_type(self.tablename)
-
- # check conflicting distributed policy
- def check_distribution_policy():
- if self.distribution_policy.startswith('DISTRIBUTED BY'):
- if len(self.files) % self.bucket_number != 0:
- logger.error('Files to be registered must be multiple times to the bucket number of hash table.')
- self.failure_handler.rollback()
- sys.exit(1)
+ def check_hash_type(self):
+ self.accessor.check_hash_type(self.dst_table_name)
- def check_database_encoding():
- encoding_indx = self.accessor.get_database_encoding_indx(self.database)
- encoding = self.accessor.get_database_encoding(encoding_indx)
- if self.encoding.strip() != encoding:
- logger.error('Database encoding from yaml configuration file(%s) is not consistent with encoding from input args(%s).' % (self.encoding, encoding))
- sys.exit(1)
-
- def create_table():
- try:
- (ret, query) = self.accessor.do_create_table(self.src_table_name, self.tablename, self.schema, self.file_format, self.distribution_policy, self.file_locations, self.bucket_number,
+ def create_table(self):
+ try:
+ (ret, query) = self.accessor.do_create_table(self.src_table_name, self.dst_table_name, self.schema, self.file_format, self.distribution_policy, self.file_locations, self.bucket_number,
self.partitionby, self.partitions_constraint, self.partitions_name)
- except pg.DatabaseError as e:
- print e
+ except pg.DatabaseError as e:
+ print e
+ sys.exit(1)
+ if ret:
+ self.failure_handler.commit(('SQL', query))
+ return ret
+
+ def check_database_encoding(self):
+ encoding_indx = self.accessor.get_database_encoding_indx(self.database)
+ encoding = self.accessor.get_database_encoding(encoding_indx)
+ if self.encoding.strip() != encoding:
+ logger.error('Database encoding from yaml configuration file(%s) is not consistent with encoding from input args(%s).' % (self.encoding, encoding))
+ sys.exit(1)
+
+ def check_policy_consistency(self):
+ policy = self.get_distribution_policy() # "" or "{1,3}"
+ if policy is None:
+ return
+ if self.distribution_policy == 'DISTRIBUTED RANDOMLY':
+ logger.error('Distribution policy of %s from yaml is not consistent with the policy of existing table.' % self.tablename)
+ self.failure_handler.rollback()
+ sys.exit(1)
+ tmp_dict = {}
+ for i, d in enumerate(self.schema):
+ tmp_dict[d['name']] = i + 1
+ # 'DISTRIBUETD BY (1,3)' -> {1,3}
+ cols = self.distribution_policy.strip().split()[-1].strip('(').strip(')').split(',')
+ original_policy = ','.join([str(tmp_dict[col]) for col in cols])
+ if policy.strip('{').strip('}') != original_policy:
+ logger.error('Distribution policy of %s from yaml file is not consistent with the policy of existing table.' % self.dst_table_name)
+ self.failure_handler.rollback()
+ sys.exit(1)
+
+ def set_yml_dataa(self, file_format, files, sizes, tablename, schema, distribution_policy, file_locations,\
+ bucket_number, partitionby, partitions_constraint, partitions_name, partitions_compression_level,\
+ partitions_compression_type, partitions_checksum, partitions_filepaths, partitions_filesizes, encoding):
+ self.file_format = file_format
+ self.files = files
+ self.sizes = sizes
+ self.src_table_name = tablename
+ self.schema = schema
+ self.distribution_policy = distribution_policy
+ self.file_locations = file_locations
+ self.bucket_number = bucket_number
+ self.partitionby = partitionby
+ self.partitions_constraint = partitions_constraint
+ self.partitions_name = partitions_name
+ self.partitions_compression_level = partitions_compression_level
+ self.partitions_compression_type = partitions_compression_type
+ self.partitions_checksum = partitions_checksum
+ self.partitions_filepaths = partitions_filepaths
+ self.partitions_filesizes = partitions_filesizes
+ self.encoding = encoding
+
+ def option_parser_yml(self, yml_file):
--- End diff --
-1 on this. Can you move the function outside? In my view, partition logic is not related to class `HawqRegister`. Besides, can you move `_mapping_tablename_from_yml` logic outside of HawqRegister class?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-hawq pull request #950: HAWQ-991. Refator HAWQ Register code for p...
Posted by linwen <gi...@git.apache.org>.
Github user linwen commented on a diff in the pull request:
https://github.com/apache/incubator-hawq/pull/950#discussion_r82497872
--- Diff: tools/bin/hawqregister ---
@@ -847,6 +819,33 @@ class HawqRegister(object):
self._modify_metadata('insert')
logger.info('Hawq Register Succeed.')
+class HawqRegister_Partition(HawqRegister):
--- End diff --
Class name in this file is like this format:AaaBbbCcc, so HawqRegisterPartition is better here.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-hawq pull request #950: HAWQ-991. Refator HAWQ Register code for p...
Posted by xunzhang <gi...@git.apache.org>.
Github user xunzhang commented on a diff in the pull request:
https://github.com/apache/incubator-hawq/pull/950#discussion_r82499122
--- Diff: tools/bin/hawqregister ---
@@ -325,250 +326,221 @@ class HawqRegister(object):
return 'usage2_table_not_exist'
else:
if not table_existed():
- logger.error('Table %s does not exist.\nYou should create table before registering the data.' % self.tablename)
+ logger.error('Table %s does not exist.\nYou should create table before registering the data.' % self.dst_table_name)
sys.exit(1)
else:
return 'usage1'
- def _init(self):
- def check_hash_type():
- self.accessor.check_hash_type(self.tablename)
-
- # check conflicting distributed policy
- def check_distribution_policy():
- if self.distribution_policy.startswith('DISTRIBUTED BY'):
- if len(self.files) % self.bucket_number != 0:
- logger.error('Files to be registered must be multiple times to the bucket number of hash table.')
- self.failure_handler.rollback()
- sys.exit(1)
+ def check_hash_type(self):
+ self.accessor.check_hash_type(self.dst_table_name)
- def check_database_encoding():
- encoding_indx = self.accessor.get_database_encoding_indx(self.database)
- encoding = self.accessor.get_database_encoding(encoding_indx)
- if self.encoding.strip() != encoding:
- logger.error('Database encoding from yaml configuration file(%s) is not consistent with encoding from input args(%s).' % (self.encoding, encoding))
- sys.exit(1)
-
- def create_table():
- try:
- (ret, query) = self.accessor.do_create_table(self.src_table_name, self.tablename, self.schema, self.file_format, self.distribution_policy, self.file_locations, self.bucket_number,
+ def create_table(self):
+ try:
+ (ret, query) = self.accessor.do_create_table(self.src_table_name, self.dst_table_name, self.schema, self.file_format, self.distribution_policy, self.file_locations, self.bucket_number,
self.partitionby, self.partitions_constraint, self.partitions_name)
- except pg.DatabaseError as e:
- print e
+ except pg.DatabaseError as e:
+ print e
+ sys.exit(1)
+ if ret:
+ self.failure_handler.commit(('SQL', query))
+ return ret
+
+ def check_database_encoding(self):
+ encoding_indx = self.accessor.get_database_encoding_indx(self.database)
+ encoding = self.accessor.get_database_encoding(encoding_indx)
+ if self.encoding.strip() != encoding:
+ logger.error('Database encoding from yaml configuration file(%s) is not consistent with encoding from input args(%s).' % (self.encoding, encoding))
+ sys.exit(1)
+
+ def check_policy_consistency(self):
+ policy = self.get_distribution_policy() # "" or "{1,3}"
+ if policy is None:
+ return
+ if self.distribution_policy == 'DISTRIBUTED RANDOMLY':
+ logger.error('Distribution policy of %s from yaml is not consistent with the policy of existing table.' % self.tablename)
+ self.failure_handler.rollback()
+ sys.exit(1)
+ tmp_dict = {}
+ for i, d in enumerate(self.schema):
+ tmp_dict[d['name']] = i + 1
+ # 'DISTRIBUETD BY (1,3)' -> {1,3}
+ cols = self.distribution_policy.strip().split()[-1].strip('(').strip(')').split(',')
+ original_policy = ','.join([str(tmp_dict[col]) for col in cols])
+ if policy.strip('{').strip('}') != original_policy:
+ logger.error('Distribution policy of %s from yaml file is not consistent with the policy of existing table.' % self.dst_table_name)
+ self.failure_handler.rollback()
+ sys.exit(1)
+
+ def set_yml_dataa(self, file_format, files, sizes, tablename, schema, distribution_policy, file_locations,\
+ bucket_number, partitionby, partitions_constraint, partitions_name, partitions_compression_level,\
+ partitions_compression_type, partitions_checksum, partitions_filepaths, partitions_filesizes, encoding):
+ self.file_format = file_format
+ self.files = files
+ self.sizes = sizes
+ self.src_table_name = tablename
+ self.schema = schema
+ self.distribution_policy = distribution_policy
+ self.file_locations = file_locations
+ self.bucket_number = bucket_number
+ self.partitionby = partitionby
--- End diff --
Ditto, -1 on this. Please make the logic of class `HawqRegister` clean. Please do not only refactor from coding perspective.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-hawq issue #950: HAWQ-991. Refator HAWQ Register code for partitio...
Posted by xunzhang <gi...@git.apache.org>.
Github user xunzhang commented on the issue:
https://github.com/apache/incubator-hawq/pull/950
@ictmalili @linwen , Since this pull is based on not that latest code of `hawqregister`, I do not recommend to merge this kind of pull requests before we finish the automative negative test cases for 2.0.1 `hawqregister` usage. Otherwise, some of our manually tests effort before is in vain.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-hawq pull request #950: HAWQ-991. Refator HAWQ Register code for p...
Posted by xunzhang <gi...@git.apache.org>.
Github user xunzhang commented on a diff in the pull request:
https://github.com/apache/incubator-hawq/pull/950#discussion_r82498424
--- Diff: tools/bin/hawqregister ---
@@ -325,250 +326,221 @@ class HawqRegister(object):
return 'usage2_table_not_exist'
else:
if not table_existed():
- logger.error('Table %s does not exist.\nYou should create table before registering the data.' % self.tablename)
+ logger.error('Table %s does not exist.\nYou should create table before registering the data.' % self.dst_table_name)
sys.exit(1)
else:
return 'usage1'
- def _init(self):
- def check_hash_type():
- self.accessor.check_hash_type(self.tablename)
-
- # check conflicting distributed policy
- def check_distribution_policy():
- if self.distribution_policy.startswith('DISTRIBUTED BY'):
- if len(self.files) % self.bucket_number != 0:
- logger.error('Files to be registered must be multiple times to the bucket number of hash table.')
- self.failure_handler.rollback()
- sys.exit(1)
+ def check_hash_type(self):
+ self.accessor.check_hash_type(self.dst_table_name)
- def check_database_encoding():
- encoding_indx = self.accessor.get_database_encoding_indx(self.database)
- encoding = self.accessor.get_database_encoding(encoding_indx)
- if self.encoding.strip() != encoding:
- logger.error('Database encoding from yaml configuration file(%s) is not consistent with encoding from input args(%s).' % (self.encoding, encoding))
- sys.exit(1)
-
- def create_table():
- try:
- (ret, query) = self.accessor.do_create_table(self.src_table_name, self.tablename, self.schema, self.file_format, self.distribution_policy, self.file_locations, self.bucket_number,
+ def create_table(self):
+ try:
+ (ret, query) = self.accessor.do_create_table(self.src_table_name, self.dst_table_name, self.schema, self.file_format, self.distribution_policy, self.file_locations, self.bucket_number,
self.partitionby, self.partitions_constraint, self.partitions_name)
- except pg.DatabaseError as e:
- print e
+ except pg.DatabaseError as e:
+ print e
+ sys.exit(1)
+ if ret:
+ self.failure_handler.commit(('SQL', query))
+ return ret
+
+ def check_database_encoding(self):
+ encoding_indx = self.accessor.get_database_encoding_indx(self.database)
+ encoding = self.accessor.get_database_encoding(encoding_indx)
+ if self.encoding.strip() != encoding:
+ logger.error('Database encoding from yaml configuration file(%s) is not consistent with encoding from input args(%s).' % (self.encoding, encoding))
+ sys.exit(1)
+
+ def check_policy_consistency(self):
+ policy = self.get_distribution_policy() # "" or "{1,3}"
+ if policy is None:
+ return
+ if self.distribution_policy == 'DISTRIBUTED RANDOMLY':
+ logger.error('Distribution policy of %s from yaml is not consistent with the policy of existing table.' % self.tablename)
+ self.failure_handler.rollback()
+ sys.exit(1)
+ tmp_dict = {}
+ for i, d in enumerate(self.schema):
+ tmp_dict[d['name']] = i + 1
+ # 'DISTRIBUETD BY (1,3)' -> {1,3}
+ cols = self.distribution_policy.strip().split()[-1].strip('(').strip(')').split(',')
+ original_policy = ','.join([str(tmp_dict[col]) for col in cols])
+ if policy.strip('{').strip('}') != original_policy:
+ logger.error('Distribution policy of %s from yaml file is not consistent with the policy of existing table.' % self.dst_table_name)
+ self.failure_handler.rollback()
+ sys.exit(1)
+
+ def set_yml_dataa(self, file_format, files, sizes, tablename, schema, distribution_policy, file_locations,\
--- End diff --
@linwen , @zhangh43 did this hah!
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-hawq pull request #950: HAWQ-991. Refator HAWQ Register code for p...
Posted by xunzhang <gi...@git.apache.org>.
Github user xunzhang commented on a diff in the pull request:
https://github.com/apache/incubator-hawq/pull/950#discussion_r82498969
--- Diff: tools/bin/hawqregister ---
@@ -325,250 +326,221 @@ class HawqRegister(object):
return 'usage2_table_not_exist'
else:
if not table_existed():
- logger.error('Table %s does not exist.\nYou should create table before registering the data.' % self.tablename)
+ logger.error('Table %s does not exist.\nYou should create table before registering the data.' % self.dst_table_name)
sys.exit(1)
else:
return 'usage1'
- def _init(self):
- def check_hash_type():
- self.accessor.check_hash_type(self.tablename)
-
- # check conflicting distributed policy
- def check_distribution_policy():
- if self.distribution_policy.startswith('DISTRIBUTED BY'):
- if len(self.files) % self.bucket_number != 0:
- logger.error('Files to be registered must be multiple times to the bucket number of hash table.')
- self.failure_handler.rollback()
- sys.exit(1)
+ def check_hash_type(self):
+ self.accessor.check_hash_type(self.dst_table_name)
- def check_database_encoding():
- encoding_indx = self.accessor.get_database_encoding_indx(self.database)
- encoding = self.accessor.get_database_encoding(encoding_indx)
- if self.encoding.strip() != encoding:
- logger.error('Database encoding from yaml configuration file(%s) is not consistent with encoding from input args(%s).' % (self.encoding, encoding))
- sys.exit(1)
-
- def create_table():
- try:
- (ret, query) = self.accessor.do_create_table(self.src_table_name, self.tablename, self.schema, self.file_format, self.distribution_policy, self.file_locations, self.bucket_number,
+ def create_table(self):
+ try:
+ (ret, query) = self.accessor.do_create_table(self.src_table_name, self.dst_table_name, self.schema, self.file_format, self.distribution_policy, self.file_locations, self.bucket_number,
self.partitionby, self.partitions_constraint, self.partitions_name)
- except pg.DatabaseError as e:
- print e
+ except pg.DatabaseError as e:
+ print e
+ sys.exit(1)
+ if ret:
+ self.failure_handler.commit(('SQL', query))
+ return ret
+
+ def check_database_encoding(self):
+ encoding_indx = self.accessor.get_database_encoding_indx(self.database)
+ encoding = self.accessor.get_database_encoding(encoding_indx)
+ if self.encoding.strip() != encoding:
+ logger.error('Database encoding from yaml configuration file(%s) is not consistent with encoding from input args(%s).' % (self.encoding, encoding))
+ sys.exit(1)
+
+ def check_policy_consistency(self):
+ policy = self.get_distribution_policy() # "" or "{1,3}"
+ if policy is None:
+ return
+ if self.distribution_policy == 'DISTRIBUTED RANDOMLY':
+ logger.error('Distribution policy of %s from yaml is not consistent with the policy of existing table.' % self.tablename)
+ self.failure_handler.rollback()
+ sys.exit(1)
+ tmp_dict = {}
+ for i, d in enumerate(self.schema):
+ tmp_dict[d['name']] = i + 1
+ # 'DISTRIBUETD BY (1,3)' -> {1,3}
+ cols = self.distribution_policy.strip().split()[-1].strip('(').strip(')').split(',')
+ original_policy = ','.join([str(tmp_dict[col]) for col in cols])
+ if policy.strip('{').strip('}') != original_policy:
+ logger.error('Distribution policy of %s from yaml file is not consistent with the policy of existing table.' % self.dst_table_name)
+ self.failure_handler.rollback()
+ sys.exit(1)
+
+ def set_yml_dataa(self, file_format, files, sizes, tablename, schema, distribution_policy, file_locations,\
+ bucket_number, partitionby, partitions_constraint, partitions_name, partitions_compression_level,\
+ partitions_compression_type, partitions_checksum, partitions_filepaths, partitions_filesizes, encoding):
+ self.file_format = file_format
+ self.files = files
+ self.sizes = sizes
+ self.src_table_name = tablename
+ self.schema = schema
+ self.distribution_policy = distribution_policy
+ self.file_locations = file_locations
+ self.bucket_number = bucket_number
+ self.partitionby = partitionby
+ self.partitions_constraint = partitions_constraint
+ self.partitions_name = partitions_name
+ self.partitions_compression_level = partitions_compression_level
+ self.partitions_compression_type = partitions_compression_type
+ self.partitions_checksum = partitions_checksum
+ self.partitions_filepaths = partitions_filepaths
+ self.partitions_filesizes = partitions_filesizes
+ self.encoding = encoding
+
+ def option_parser_yml(self, yml_file):
+ import yaml
+ with open(yml_file, 'r') as f:
+ params = yaml.load(f)
+ table_column_num = self.accessor.get_table_column_num(self.tablename)
+ register_yaml_dict_check(params, table_column_num, self.tablename)
+ partitions_filepaths = []
+ partitions_filesizes = []
+ partitions_constraint = []
+ partitions_name = []
+ partitions_checksum = []
+ partitions_compression_level = []
+ partitions_compression_type = []
+ files, sizes = [], []
+
+ if params['FileFormat'].lower() == 'parquet':
+ Format = 'Parquet'
+ else: #AO format
+ Format = 'AO'
+ Format_FileLocations = '%s_FileLocations' % Format
+ partitionby = params.get(Format_FileLocations).get('PartitionBy')
+ if partitionby:
+ logger.info('Partition table is not supported in current release of hawq register.')
+ sys.exit(0)
+ if params.get(Format_FileLocations).get('Partitions') and len(params[Format_FileLocations]['Partitions']):
+ partitions_checksum = [d['Checksum'] for d in params[Format_FileLocations]['Partitions']]
+ partitions_compression_level = [d['CompressionLevel'] for d in params[Format_FileLocations]['Partitions']]
+ partitions_compression_type = [d['CompressionType'] for d in params[Format_FileLocations]['Partitions']]
+ partitions_constraint = [d['Constraint'] for d in params[Format_FileLocations]['Partitions']]
+ partitions_files = [d['Files'] for d in params[Format_FileLocations]['Partitions']]
+ if len(partitions_files):
+ for pfile in partitions_files:
+ partitions_filepaths.append([params['DFS_URL'] + item['path'] for item in pfile])
+ partitions_filesizes.append([item['size'] for item in pfile])
+ partitions_name = [d['Name'] for d in params[Format_FileLocations]['Partitions']]
+ if len(params[Format_FileLocations]['Files']):
+ files, sizes = [params['DFS_URL'] + d['path'] for d in params[Format_FileLocations]['Files']], [d['size'] for d in params[Format_FileLocations]['Files']]
+ encoding = params['Encoding']
+ self.set_yml_dataa(Format, files, sizes, params['TableName'], params['%s_Schema' % Format], params['Distribution_Policy'], params[Format_FileLocations], params['Bucketnum'], partitionby,\
+ partitions_constraint, partitions_name, partitions_compression_level, partitions_compression_type, partitions_checksum, partitions_filepaths, partitions_filesizes, encoding)
+
+
+ # check conflicting distributed policy
+ def check_distribution_policy(self):
+ if self.distribution_policy.startswith('DISTRIBUTED BY'):
+ if len(self.files) % self.bucket_number != 0:
+ logger.error('Files to be registered must be multiple times to the bucket number of hash table.')
+ self.failure_handler.rollback()
sys.exit(1)
- if ret:
- self.failure_handler.commit(('SQL', query))
- return ret
- def get_seg_name():
- return self.utility_accessor.get_seg_name(self.tablename, self.database, self.file_format)
+ def get_seg_name(self):
+ return self.utility_accessor.get_seg_name(self.tablename, self.database, self.file_format)
- def get_metadata():
- return self.accessor.get_metadata_from_database(self.tablename, self.seg_name)
+ def get_metadata(self):
+ return self.accessor.get_metadata_from_database(self.tablename, self.seg_name)
- def get_metadata_from_table():
- return self.accessor.get_metadata_from_seg_name(self.seg_name)
+ def get_metadata_from_table(self):
+ return self.accessor.get_metadata_from_seg_name(self.seg_name)
- def get_distribution_policy():
- return self.accessor.get_distribution_policy_info(self.tablename)
+ def get_distribution_policy(self):
+ return self.accessor.get_distribution_policy_info(self.tablename)
- def check_policy_consistency():
- policy = get_distribution_policy() # "" or "{1,3}"
- if policy is None:
- return
- if self.distribution_policy == 'DISTRIBUTED RANDOMLY':
- logger.error('Distribution policy of %s from yaml is not consistent with the policy of existing table.' % self.tablename)
- self.failure_handler.rollback()
+ def check_bucket_number(self):
+ def get_bucket_number():
+ return self.accessor.get_bucket_number(self.tablename)
+
+ if self.bucket_number != get_bucket_number():
+ logger.error('Bucket number of %s is not consistent with previous bucket number.' % self.tablename)
+ self.failure_handler.rollback()
+ sys.exit(1)
+
+ def check_file_not_folder(self):
+ for fn in self.files:
+ hdfscmd = 'hadoop fs -test -f %s' % fn
+ if local_ssh(hdfscmd, logger):
+ logger.info('%s is not a file in hdfs, please check the yaml configuration file.' % fn)
sys.exit(1)
- tmp_dict = {}
- for i, d in enumerate(self.schema):
- tmp_dict[d['name']] = i + 1
- # 'DISTRIBUETD BY (1,3)' -> {1,3}
- cols = self.distribution_policy.strip().split()[-1].strip('(').strip(')').split(',')
- original_policy = ','.join([str(tmp_dict[col]) for col in cols])
- if policy.strip('{').strip('}') != original_policy:
- logger.error('Distribution policy of %s from yaml file is not consistent with the policy of existing table.' % self.tablename)
+
+ def check_sizes_valid(self):
+ for sz in self.sizes:
+ if type(sz) != type(1):
+ logger.error('File size(%s) in yaml configuration file should be int type.' % sz)
self.failure_handler.rollback()
sys.exit(1)
-
- def check_bucket_number():
- def get_bucket_number():
- return self.accessor.get_bucket_number(self.tablename)
-
- if self.bucket_number != get_bucket_number():
- logger.error('Bucket number of %s is not consistent with previous bucket number.' % self.tablename)
+ if sz < 0:
+ logger.error('File size(%s) in yaml configuration file should not be less than 0.' % sz)
self.failure_handler.rollback()
sys.exit(1)
-
- def set_yml_dataa(file_format, files, sizes, tablename, schema, distribution_policy, file_locations,\
- bucket_number, partitionby, partitions_constraint, partitions_name, partitions_compression_level,\
- partitions_compression_type, partitions_checksum, partitions_filepaths, partitions_filesizes, encoding):
- self.file_format = file_format
- self.files = files
- self.sizes = sizes
- self.src_table_name = tablename
- self.schema = schema
- self.distribution_policy = distribution_policy
- self.file_locations = file_locations
- self.bucket_number = bucket_number
- self.partitionby = partitionby
- self.partitions_constraint = partitions_constraint
- self.partitions_name = partitions_name
- self.partitions_compression_level = partitions_compression_level
- self.partitions_compression_type = partitions_compression_type
- self.partitions_checksum = partitions_checksum
- self.partitions_filepaths = partitions_filepaths
- self.partitions_filesizes = partitions_filesizes
- self.encoding = encoding
-
- def option_parser_yml(yml_file):
- import yaml
- try:
- with open(yml_file, 'r') as f:
- params = yaml.load(f)
- except yaml.scanner.ScannerError as e:
- print e
+ for k, fn in enumerate(self.files):
+ hdfscmd = 'hadoop fs -du %s' % fn
+ _, out, _ = local_ssh_output(hdfscmd)
+ if self.sizes[k] > int(out.strip().split()[0]):
+ logger.error('File size(%s) in yaml configuration file should not exceed actual length(%s) of file %s.' % (self.sizes[k], out.strip().split()[0], fn))
self.failure_handler.rollback()
sys.exit(1)
-
- table_column_num = self.accessor.get_table_column_num(self.tablename)
- register_yaml_dict_check(params, table_column_num, self.tablename)
- partitions_filepaths = []
- partitions_filesizes = []
- partitions_constraint = []
- partitions_name = []
- partitions_checksum = []
- partitions_compression_level = []
- partitions_compression_type = []
- files, sizes = [], []
-
- if params['FileFormat'].lower() == 'parquet':
- partitionby = params.get('Parquet_FileLocations').get('PartitionBy')
- if partitionby:
- logger.info('Partition table is not supported in current release of hawq register.')
- sys.exit(0)
- if params.get('Parquet_FileLocations').get('Partitions') and len(params['Parquet_FileLocations']['Partitions']):
- partitions_checksum = [d['Checksum'] for d in params['Parquet_FileLocations']['Partitions']]
- partitions_compression_level = [d['CompressionLevel'] for d in params['Parquet_FileLocations']['Partitions']]
- partitions_compression_type = [d['CompressionType'] for d in params['Parquet_FileLocations']['Partitions']]
- partitions_constraint = [d['Constraint'] for d in params['Parquet_FileLocations']['Partitions']]
- partitions_files = [d['Files'] for d in params['Parquet_FileLocations']['Partitions']]
- if len(partitions_files):
- for pfile in partitions_files:
- partitions_filepaths.append([params['DFS_URL'] + item['path'] for item in pfile])
- partitions_filesizes.append([item['size'] for item in pfile])
- partitions_name = [d['Name'] for d in params['Parquet_FileLocations']['Partitions']]
- if len(params['Parquet_FileLocations']['Files']):
- files, sizes = [params['DFS_URL'] + d['path'] for d in params['Parquet_FileLocations']['Files']], [d['size'] for d in params['Parquet_FileLocations']['Files']]
- encoding = params['Encoding']
- set_yml_dataa('Parquet', files, sizes, params['TableName'], params['Parquet_Schema'], params['Distribution_Policy'], params['Parquet_FileLocations'], params['Bucketnum'], partitionby,\
- partitions_constraint, partitions_name, partitions_compression_level, partitions_compression_type, partitions_checksum, partitions_filepaths, partitions_filesizes, encoding)
-
- else: #AO format
- partitionby = params.get('AO_FileLocations').get('PartitionBy')
- if partitionby:
- logger.info('Partition table is not supported in current release of hawq register.')
- sys.exit(0)
- if params.get('AO_FileLocations').get('Partitions') and len(params['AO_FileLocations']['Partitions']):
- partitions_checksum = [d['Checksum'] for d in params['AO_FileLocations']['Partitions']]
- partitions_compressionLevel = [d['CompressionLevel'] for d in params['AO_FileLocations']['Partitions']]
- partitions_compressionType = [d['CompressionType'] for d in params['AO_FileLocations']['Partitions']]
- partitions_constraint = [d['Constraint'] for d in params['AO_FileLocations']['Partitions']]
- partitions_files = [d['Files'] for d in params['AO_FileLocations']['Partitions']]
- if len(partitions_files):
- for pfile in partitions_files:
- partitions_filepaths.append([params['DFS_URL'] + item['path'] for item in pfile])
- partitions_filesizes.append([item['size'] for item in pfile])
- partitions_name = [d['Name'] for d in params['AO_FileLocations']['Partitions']]
- if len(params['AO_FileLocations']['Files']):
- files, sizes = [params['DFS_URL'] + d['path'] for d in params['AO_FileLocations']['Files']], [d['size'] for d in params['AO_FileLocations']['Files']]
- encoding = params['Encoding']
- set_yml_dataa('AO', files, sizes, params['TableName'], params['AO_Schema'], params['Distribution_Policy'], params['AO_FileLocations'], params['Bucketnum'], partitionby, partitions_constraint,\
- partitions_name, partitions_compression_level, partitions_compression_type, partitions_checksum, partitions_filepaths, partitions_filesizes, encoding)
-
- def check_file_not_folder():
- for fn in self.files:
- hdfscmd = 'hadoop fs -test -f %s' % fn
- if local_ssh(hdfscmd, logger):
- logger.info('%s is not a file in hdfs, please check the yaml configuration file.' % fn)
- sys.exit(1)
- def is_folder(filepath):
- hdfscmd = 'hadoop fs -test -d %s' % filepath
- if local_ssh(hdfscmd, logger):
- return False
- else:
- return True
-
- def check_sizes_valid():
- for sz in self.sizes:
- if type(sz) != type(1):
- logger.error('File size(%s) in yaml configuration file should be int type.' % sz)
- self.failure_handler.rollback()
- sys.exit(1)
- if sz < 0:
- logger.error('File size(%s) in yaml configuration file should not be less than 0.' % sz)
+ def check_no_regex_filepath(self, files):
+ for fn in files:
+ tmp_lst = fn.split('/')
+ for v in tmp_lst:
+ if v == '.':
+ logger.error('Hawq register does not support file path with regex: %s.' % fn)
self.failure_handler.rollback()
sys.exit(1)
- for k, fn in enumerate(self.files):
- hdfscmd = 'hadoop fs -du %s' % fn
- _, out, _ = local_ssh_output(hdfscmd)
- if self.sizes[k] > int(out.strip().split()[0]):
- logger.error('File size(%s) in yaml configuration file should not exceed actual length(%s) of file %s.' % (self.sizes[k], out.strip().split()[0], fn))
+ for ch in ['..', '*']:
+ if fn.find(ch) != -1:
+ logger.error('Hawq register does not support file path with regex: %s.' % fn)
self.failure_handler.rollback()
sys.exit(1)
- def check_no_regex_filepath(files):
- for fn in files:
- tmp_lst = fn.split('/')
- for v in tmp_lst:
- if v == '.':
- logger.error('Hawq register does not support file path with regex: %s.' % fn)
- self.failure_handler.rollback()
- sys.exit(1)
- for ch in ['..', '*']:
- if fn.find(ch) != -1:
- logger.error('Hawq register does not support file path with regex: %s.' % fn)
- self.failure_handler.rollback()
- sys.exit(1)
-
+ def _init(self):
if self.yml:
- option_parser_yml(options.yml_config)
+ self.option_parser_yml(options.yml_config)
self.filepath = self.files[0][:self.files[0].rfind('/')] if self.files else ''
- check_file_not_folder()
- check_database_encoding()
+ self.check_file_not_folder()
+ self.check_database_encoding()
if self.mode != 'repair':
- if not create_table() and self.mode != 'force':
+ if not self.create_table() and self.mode != 'force':
self.mode = 'usage2_table_exist'
- check_bucket_number()
- check_distribution_policy()
- check_policy_consistency()
- check_no_regex_filepath(self.files)
else:
- if is_folder(self.filepath) and self.filesize:
- logger.error('-e option is only supported with single file case.')
- sys.exit(1)
self.file_format = 'Parquet'
- check_hash_type() # Usage1 only support randomly distributed table
+ self.check_hash_type() # Usage1 only support randomly distributed table
+ self.do_check()
--- End diff --
The name here is a little confused I think.
Firstly, your L509 use `check_file_not_folder `, L510 `check_database_encoding `, L516 use `check_hash_type`, what's the difference between `check_file_not_folder`, `check_database_encoding`, `check_hash_type` and `do_check`. (I know the reason why you do this, but please refactor with semantic, not just code itself...)
Secondly, `do_check` is an internal interface used in `_init`, which means it can only be invoked in the right place of `_init` function. I strongly recommend define this kind of function inside `_init`, it can not be invoked independently. Anther reason defining a function variable inside a function is simplify the invoking process for readability.
Thirdly, a internal interface in Python should be named starting with `_` sign.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-hawq pull request #950: HAWQ-991. Refator HAWQ Register code to ma...
Posted by wcl14 <gi...@git.apache.org>.
Github user wcl14 closed the pull request at:
https://github.com/apache/incubator-hawq/pull/950
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-hawq issue #950: HAWQ-991. Refator HAWQ Register code for partitio...
Posted by ictmalili <gi...@git.apache.org>.
Github user ictmalili commented on the issue:
https://github.com/apache/incubator-hawq/pull/950
@wcl14 Could you change the description for this Pull Request, since it only covers code refactor part. The partition table support can be processed by another PR. Thanks
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-hawq issue #950: HAWQ-991. Refator HAWQ Register code to make sure...
Posted by linwen <gi...@git.apache.org>.
Github user linwen commented on the issue:
https://github.com/apache/incubator-hawq/pull/950
LGTM
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---