<commit_before>#
# Copyright 2015 LinkedIn Corp. All rights reserved.
#
# Licensed 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.
#

import sys
from org.slf4j import LoggerFactory
from com.ziclix.python.sql import zxJDBC
from wherehows.common import Constant

from org.python.core import codecs
codecs.setDefaultEncoding('utf-8')

class HBaseLoad:
  def __init__(self, exec_id):
    self.logger = LoggerFactory.getLogger(self.__class__.__name__ + ':' + str(exec_id))

  def load_metadata(self):
    """
    Load dataset metadata into final table
    :return: nothing
    """
    cursor = self.conn_mysql.cursor()
    load_cmd = '''
        DELETE FROM stg_dict_dataset WHERE db_id = {db_id};
        LOAD DATA LOCAL INFILE '{source_file}'
        INTO TABLE stg_dict_dataset
        FIELDS TERMINATED BY '\Z' ESCAPED BY '\0'
        (`name`, `schema`, properties, fields, urn, source,  @dataset_type, @storage_type,
        sample_partition_full_path, source_created_time, source_modified_time)
        SET db_id = {db_id},wh_etl_exec_id = {wh_etl_exec_id};
        
        DELETE FROM stg_dict_dataset
        where db_id = {db_id}
            AND (length(`name`) = 0
             OR name like 'tmp\_%'
             OR urn like '%/\_tmp'
             OR urn like '%/\_distcp\_%');
        
        update stg_dict_dataset
        set location_prefix =
           case
            when source = 'Hbase'
            then '/hbase'
           end
        WHERE db_id = {db_id} and location_prefix is null;

        update stg_dict_dataset
        set name = substring_index(urn, '/', -1)
        where db_id = {db_id};

        update stg_dict_dataset
        set parent_name = 'hbase'
        where db_id = {db_id};
        
        DELETE FROM stg_dict_dataset_instance WHERE db_id = {db_id};
        INSERT INTO stg_dict_dataset_instance
        ( dataset_urn,
          db_id,
          deployment_tier,
          data_center,
          server_cluster,
          slice,
          status_id,
          native_name,
          logical_name,
          `version`,
          instance_created_time,
          created_time,
          wh_etl_exec_id,
          abstract_dataset_urn,
          schema_text
        )
        select s.urn,{db_id}, d.deployment_tier,d.data_center, d.cluster, '*', 0,concat('hbase.',s.name), s.name,
          0, s.source_created_time, s.created_time,{wh_etl_exec_id}, s.urn, s.schema
        from stg_dict_dataset s JOIN cfg_database d on s.db_id = d.db_id
        where s.db_id = {db_id}
        on duplicate key update
          deployment_tier=d.deployment_tier, data_center=d.data_center,
          server_cluster=d.cluster, native_name=s.name, logical_name=s.name,
          instance_created_time=s.source_created_time, created_time=s.created_time,
          wh_etl_exec_id={wh_etl_exec_id}, abstract_dataset_urn=s.urn, schema_text=s.schema;
        
        
        INSERT IGNORE INTO dict_dataset
        ( `name`,
          `schema`,
          schema_type,
          `fields`,
          properties,
          urn,
          `source`,
          location_prefix,
          parent_name,
          storage_type,
          ref_dataset_id,
          status_id,
          dataset_type,
          hive_serdes_class,
          is_partitioned,
          partition_layout_pattern_id,
          sample_partition_full_path,
          source_created_time,
          source_modified_time,
          created_time,
          wh_etl_exec_id
        )
        select s.name, s.schema, s.schema_type, s.fields,
          s.properties, s.urn,
          s.source, s.location_prefix, s.parent_name,
          s.storage_type, s.ref_dataset_id, s.status_id,
          s.dataset_type, s.hive_serdes_class, s.is_partitioned,
          s.partition_layout_pattern_id, s.sample_partition_full_path,
          s.source_created_time, s.source_modified_time, UNIX_TIMESTAMP(now()),
          s.wh_etl_exec_id
        from stg_dict_dataset s
        where s.db_id = {db_id}
        on duplicate key update
          `name`=s.name, `schema`=s.schema, schema_type=s.schema_type, `fields`=s.fields,
          properties=s.properties, `source`=s.source, location_prefix=s.location_prefix, parent_name=s.parent_name,
          storage_type=s.storage_type, ref_dataset_id=s.ref_dataset_id, status_id=s.status_id,
          dataset_type=s.dataset_type, hive_serdes_class=s.hive_serdes_class, is_partitioned=s.is_partitioned,
          partition_layout_pattern_id=s.partition_layout_pattern_id, sample_partition_full_path=s.sample_partition_full_path,
          source_created_time=s.source_created_time, source_modified_time=s.source_modified_time,
          modified_time=UNIX_TIMESTAMP(now()), wh_etl_exec_id=s.wh_etl_exec_id
        ;
        analyze table dict_dataset;
        
        
        update stg_dict_dataset_instance sdi, dict_dataset d
        set sdi.dataset_id = d.id where sdi.abstract_dataset_urn = d.urn
        and sdi.db_id = {db_id};
        
        INSERT IGNORE INTO dict_dataset_instance
        ( dataset_id,
          db_id,
          deployment_tier,
          data_center,
          server_cluster,
          slice,
          status_id,
          native_name,
          logical_name,
          version,
          version_sort_id,
          schema_text,
          ddl_text,
          instance_created_time,
          created_time,
          wh_etl_exec_id
        )
        select s.dataset_id, s.db_id, s.deployment_tier, s.data_center,
          s.server_cluster, s.slice, s.status_id,  s.native_name, s.logical_name, s.version,
          case when s.version regexp '[0-9]+\.[0-9]+\.[0-9]+'
            then cast(substring_index(s.version, '.', 1) as unsigned) * 100000000 +
                 cast(substring_index(substring_index(s.version, '.', 2), '.', -1) as unsigned) * 10000 +
                 cast(substring_index(s.version, '.', -1) as unsigned)
          else 0
          end version_sort_id, s.schema_text, s.ddl_text,
          s.instance_created_time, s.created_time, s.wh_etl_exec_id
        from stg_dict_dataset_instance s
        where s.db_id = {db_id}
        on duplicate key update
          deployment_tier=s.deployment_tier, data_center=s.data_center, server_cluster=s.server_cluster, slice=s.slice,
          status_id=s.status_id, native_name=s.native_name, logical_name=s.logical_name, version=s.version,
          schema_text=s.schema_text, ddl_text=s.ddl_text,
          instance_created_time=s.instance_created_time, created_time=s.created_time, wh_etl_exec_id=s.wh_etl_exec_id
          ;

        '''.format(source_file=self.input_file, db_id=self.db_id, wh_etl_exec_id=self.wh_etl_exec_id)
    for state in load_cmd.split(";"):
        self.logger.debug(state)
        cursor.execute(state)
        self.conn_mysql.commit()
    cursor.close()
    self.logger.info("finish loading hbase metadata db_id={db_id} to dict_dataset".format(db_id=self.db_id))


  def load_field(self):
    cursor = self.conn_mysql.cursor()
    load_field_cmd = '''
        DELETE FROM stg_dict_field_detail where db_id = {db_id};
        
        LOAD DATA LOCAL INFILE '{source_file}'
        INTO TABLE stg_dict_field_detail
        FIELDS TERMINATED BY '\Z'
        (urn, sort_id, parent_sort_id, parent_path, field_name, data_type,
         is_nullable, default_value, data_size, namespace, description)
        SET db_id = {db_id};
        
        analyze table stg_dict_field_detail;
        
        update stg_dict_field_detail
        set description = null
        where db_id = {db_id}
           and (char_length(trim(description)) = 0
           or description in ('null', 'N/A', 'nothing', 'empty', 'none'));

        insert into field_comments (
          user_id, comment, created, modified, comment_crc32_checksum
        )
        select 0 user_id, description, now() created, now() modified, crc32(description) from
        (
          select sf.description
          from stg_dict_field_detail sf left join field_comments fc
            on sf.description = fc.comment
          where sf.description is not null
            and fc.id is null
            and sf.db_id = {db_id}
          group by 1 order by 1
        ) d;

        analyze table field_comments;

        -- delete old record if it does not exist in this load batch anymore (but have the dataset id)
        create temporary table if not exists t_deleted_fields (primary key (field_id))
          select x.field_id
            from stg_dict_field_detail s
              join dict_dataset i
                on s.urn = i.urn
                and s.db_id = {db_id}
              right join dict_field_detail x
                on i.id = x.dataset_id
                and s.field_name = x.field_name
                and s.parent_path = x.parent_path
          where s.field_name is null
            and x.dataset_id in (
                       select d.id dataset_id
                       from stg_dict_field_detail k join dict_dataset d
                         on k.urn = d.urn
                        and k.db_id = {db_id}
            )
        ; -- run time : ~2min

        delete from dict_field_detail where field_id in (select field_id from t_deleted_fields);

        -- update the old record if some thing changed
        update dict_field_detail t join
        (
          select x.field_id, s.*
          from (select * from stg_dict_field_detail where db_id = {db_id}) s
            join dict_dataset d
              on s.urn = d.urn
            join dict_field_detail x
              on s.field_name = x.field_name
              and coalesce(s.parent_path, '*') = coalesce(x.parent_path, '*')
              and d.id = x.dataset_id
          where (x.sort_id <> s.sort_id
                or x.parent_sort_id <> s.parent_sort_id
                or x.data_type <> s.data_type
                or x.data_size <> s.data_size or (x.data_size is null XOR s.data_size is null)
                or x.data_precision <> s.data_precision or (x.data_precision is null XOR s.data_precision is null)
                or x.is_nullable <> s.is_nullable or (x.is_nullable is null XOR s.is_nullable is null)
                or x.is_partitioned <> s.is_partitioned or (x.is_partitioned is null XOR s.is_partitioned is null)
                or x.is_distributed <> s.is_distributed or (x.is_distributed is null XOR s.is_distributed is null)
                or x.default_value <> s.default_value or (x.default_value is null XOR s.default_value is null)
                or x.namespace <> s.namespace or (x.namespace is null XOR s.namespace is null)
            )
        ) p
          on t.field_id = p.field_id
        set t.sort_id = p.sort_id,
            t.parent_sort_id = p.parent_sort_id,
            t.data_type = p.data_type,
            t.data_size = p.data_size,
            t.data_precision = p.data_precision,
            t.is_nullable = p.is_nullable,
            t.is_partitioned = p.is_partitioned,
            t.is_distributed = p.is_distributed,
            t.default_value = p.default_value,
            t.namespace = p.namespace,
            t.modified = now()
        ;

       insert ignore into dict_field_detail (
          dataset_id, fields_layout_id, sort_id, parent_sort_id, parent_path,
          field_name, namespace, data_type, data_size, is_nullable, default_value,
           modified
        )
        select
          d.id, 0, sf.sort_id, sf.parent_sort_id, sf.parent_path,
          sf.field_name, sf.namespace, sf.data_type, sf.data_size, sf.is_nullable, sf.default_value, now()
        from stg_dict_field_detail sf join dict_dataset d
          on sf.urn = d.urn
             left join dict_field_detail t
          on d.id = t.dataset_id
         and sf.field_name = t.field_name
         and sf.parent_path = t.parent_path
        where db_id = {db_id} and t.field_id is null
        ;

        analyze table dict_field_detail;

        -- delete old record in stagging
        delete from stg_dict_dataset_field_comment where db_id = {db_id};

        -- insert
        insert into stg_dict_dataset_field_comment
        select t.field_id field_id, fc.id comment_id,  d.id dataset_id, {db_id}
                from stg_dict_field_detail sf join dict_dataset d
                  on sf.urn = d.urn
                      join field_comments fc
                  on sf.description = fc.comment
                      join dict_field_detail t
                  on d.id = t.dataset_id
                 and sf.field_name = t.field_name
                 and sf.parent_path = t.parent_path
        where sf.db_id = {db_id};

        -- have default comment, insert it set default to 0
        insert ignore into dict_dataset_field_comment
        select field_id, comment_id, dataset_id, 0 is_default from stg_dict_dataset_field_comment where field_id in (
          select field_id from dict_dataset_field_comment
          where field_id in (select field_id from stg_dict_dataset_field_comment)
        and is_default = 1 ) and db_id = {db_id};


        -- doesn't have this comment before, insert into it and set as default
        insert ignore into dict_dataset_field_comment
        select sd.field_id, sd.comment_id, sd.dataset_id, 1 from stg_dict_dataset_field_comment sd
        left join dict_dataset_field_comment d
        on d.field_id = sd.field_id
         and d.comment_id = sd.comment_id
        where d.comment_id is null
        and sd.db_id = {db_id};


        DELETE FROM stg_dict_field_detail where db_id = {db_id};

        '''.format(source_file=self.input_field_file, db_id=self.db_id)
    for state in load_field_cmd.split(";"):
        self.logger.debug(state)
        cursor.execute(state)
        self.conn_mysql.commit()
    cursor.close()
    self.logger.info("finish loading hbase metadata db_id={db_id} to dict_field_detail".format(db_id=self.db_id))

  def load_sample(self):
    cursor = self.conn_mysql.cursor()
    load_sample_cmd = '''
    DELETE FROM stg_dict_dataset_sample where db_id = {db_id};

    LOAD DATA LOCAL INFILE '{source_file}'
    INTO TABLE stg_dict_dataset_sample
    FIELDS TERMINATED BY '\Z' ESCAPED BY '\0'
    (urn,ref_urn,data)
    SET db_id = {db_id};

    -- update reference id in stagging table
    UPDATE  stg_dict_dataset_sample s
    JOIN dict_dataset d ON s.ref_urn = d.urn
    SET s.ref_id = d.id
    WHERE s.db_id = {db_id} AND s.ref_urn > '' AND s.ref_urn <> 'null';

    -- first insert ref_id as 0
    INSERT INTO dict_dataset_sample
    ( `dataset_id`,
      `urn`,
      `ref_id`,
      `data`,
      created
    )
    select d.id as dataset_id, s.urn, s.ref_id, s.data, now()
    from stg_dict_dataset_sample s join dict_dataset d on d.urn = s.urn
          where s.db_id = {db_id}
    on duplicate key update
      `data`=s.data, modified=now();

      -- update reference id in final table
    UPDATE dict_dataset_sample d
    RIGHT JOIN stg_dict_dataset_sample s ON d.urn = s.urn
    SET d.ref_id = s.ref_id
    WHERE s.db_id = {db_id} AND d.ref_id = 0;
    '''.format(source_file=self.input_sample_file, db_id=self.db_id)
    for state in load_sample_cmd.split(";"):
        self.logger.debug(state)
        cursor.execute(state)
        self.conn_mysql.commit()
    cursor.close()
    self.logger.info("finish loading hbase sample data db_id={db_id} to dict_dataset_sample".format(db_id=self.db_id))

if __name__ == "__main__":
  args = sys.argv[1]

  l = HBaseLoad(args[Constant.WH_EXEC_ID_KEY])

  # set up connection
  username = args[Constant.WH_DB_USERNAME_KEY]
  password = args[Constant.WH_DB_PASSWORD_KEY]
  JDBC_DRIVER = args[Constant.WH_DB_DRIVER_KEY]
  JDBC_URL = args[Constant.WH_DB_URL_KEY]

  l.input_file = "/var/tmp/wherehows/hbase_meta"
  l.input_field_file = "/var/tmp/wherehows/hbase_field_meta"
  l.input_sample_file = "/var/tmp/wherehows/hbase_sample"

  l.db_id = args[Constant.DB_ID_KEY]
  l.wh_etl_exec_id = args[Constant.WH_EXEC_ID_KEY]
  l.conn_mysql = zxJDBC.connect(JDBC_URL, username, password, JDBC_DRIVER)

  if Constant.INNODB_LOCK_WAIT_TIMEOUT in args:
    lock_wait_time = args[Constant.INNODB_LOCK_WAIT_TIMEOUT]
    l.conn_mysql.cursor().execute("SET innodb_lock_wait_timeout = %s;" % lock_wait_time)

  try:
    l.load_metadata()
    l.load_field()
    l.load_sample()
  finally:
    l.conn_mysql.close()




<commit_msg>hbase web display info bug repair<commit_after>#
# Copyright 2015 LinkedIn Corp. All rights reserved.
#
# Licensed 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.
#

import sys
from org.slf4j import LoggerFactory
from com.ziclix.python.sql import zxJDBC
from wherehows.common import Constant

from org.python.core import codecs
codecs.setDefaultEncoding('utf-8')

class HBaseLoad:
  def __init__(self, exec_id):
    self.logger = LoggerFactory.getLogger(self.__class__.__name__ + ':' + str(exec_id))

  def load_metadata(self):
    """
    Load dataset metadata into final table
    :return: nothing
    """
    cursor = self.conn_mysql.cursor()
    load_cmd = '''
        DELETE FROM stg_dict_dataset WHERE db_id = {db_id};
        LOAD DATA LOCAL INFILE '{source_file}'
        INTO TABLE stg_dict_dataset
        FIELDS TERMINATED BY '\Z' ESCAPED BY '\0'
        (`name`, `schema`, properties, fields, urn, source,  @dataset_type, @storage_type,
        sample_partition_full_path, source_created_time, source_modified_time)
        SET db_id = {db_id},wh_etl_exec_id = {wh_etl_exec_id};

        UPDATE stg_dict_dataset set dataset_type = 'Hbase'
        ,storage_type = 'Table' where source = 'Hbase';
        
        DELETE FROM stg_dict_dataset
        where db_id = {db_id}
            AND (length(`name`) = 0
             OR name like 'tmp\_%'
             OR urn like '%/\_tmp'
             OR urn like '%/\_distcp\_%');
        
        update stg_dict_dataset
        set location_prefix =
           case
            when source = 'Hbase'
            then '/hbase'
           end
        WHERE db_id = {db_id} and location_prefix is null;

        update stg_dict_dataset
        set name = substring_index(urn, '/', -1)
        where db_id = {db_id};

        update stg_dict_dataset
        set parent_name = 'hbase'
        where db_id = {db_id};
        
        DELETE FROM stg_dict_dataset_instance WHERE db_id = {db_id};
        INSERT INTO stg_dict_dataset_instance
        ( dataset_urn,
          db_id,
          deployment_tier,
          data_center,
          server_cluster,
          slice,
          status_id,
          native_name,
          logical_name,
          `version`,
          instance_created_time,
          created_time,
          wh_etl_exec_id,
          abstract_dataset_urn,
          schema_text
        )
        select s.urn,{db_id}, d.deployment_tier,d.data_center, d.cluster, '*', 0,concat('hbase.',s.name), s.name,
          0, s.source_created_time, s.created_time,{wh_etl_exec_id}, s.urn, s.schema
        from stg_dict_dataset s JOIN cfg_database d on s.db_id = d.db_id
        where s.db_id = {db_id}
        on duplicate key update
          deployment_tier=d.deployment_tier, data_center=d.data_center,
          server_cluster=d.cluster, native_name=s.name, logical_name=s.name,
          instance_created_time=s.source_created_time, created_time=s.created_time,
          wh_etl_exec_id={wh_etl_exec_id}, abstract_dataset_urn=s.urn, schema_text=s.schema;
        
        
        INSERT IGNORE INTO dict_dataset
        ( `name`,
          `schema`,
          schema_type,
          `fields`,
          properties,
          urn,
          `source`,
          location_prefix,
          parent_name,
          storage_type,
          ref_dataset_id,
          status_id,
          dataset_type,
          hive_serdes_class,
          is_partitioned,
          partition_layout_pattern_id,
          sample_partition_full_path,
          source_created_time,
          source_modified_time,
          created_time,
          wh_etl_exec_id
        )
        select s.name, s.schema, s.schema_type, s.fields,
          s.properties, s.urn,
          s.source, s.location_prefix, s.parent_name,
          s.storage_type, s.ref_dataset_id, s.status_id,
          s.dataset_type, s.hive_serdes_class, s.is_partitioned,
          s.partition_layout_pattern_id, s.sample_partition_full_path,
          s.source_created_time, s.source_modified_time, UNIX_TIMESTAMP(now()),
          s.wh_etl_exec_id
        from stg_dict_dataset s
        where s.db_id = {db_id}
        on duplicate key update
          `name`=s.name, `schema`=s.schema, schema_type=s.schema_type, `fields`=s.fields,
          properties=s.properties, `source`=s.source, location_prefix=s.location_prefix, parent_name=s.parent_name,
          storage_type=s.storage_type, ref_dataset_id=s.ref_dataset_id, status_id=s.status_id,
          dataset_type=s.dataset_type, hive_serdes_class=s.hive_serdes_class, is_partitioned=s.is_partitioned,
          partition_layout_pattern_id=s.partition_layout_pattern_id, sample_partition_full_path=s.sample_partition_full_path,
          source_created_time=s.source_created_time, source_modified_time=s.source_modified_time,
          modified_time=UNIX_TIMESTAMP(now()), wh_etl_exec_id=s.wh_etl_exec_id
        ;
        analyze table dict_dataset;
        
        
        update stg_dict_dataset_instance sdi, dict_dataset d
        set sdi.dataset_id = d.id where sdi.abstract_dataset_urn = d.urn
        and sdi.db_id = {db_id};
        
        INSERT IGNORE INTO dict_dataset_instance
        ( dataset_id,
          db_id,
          deployment_tier,
          data_center,
          server_cluster,
          slice,
          status_id,
          native_name,
          logical_name,
          version,
          version_sort_id,
          schema_text,
          ddl_text,
          instance_created_time,
          created_time,
          wh_etl_exec_id
        )
        select s.dataset_id, s.db_id, s.deployment_tier, s.data_center,
          s.server_cluster, s.slice, s.status_id,  s.native_name, s.logical_name, s.version,
          case when s.version regexp '[0-9]+\.[0-9]+\.[0-9]+'
            then cast(substring_index(s.version, '.', 1) as unsigned) * 100000000 +
                 cast(substring_index(substring_index(s.version, '.', 2), '.', -1) as unsigned) * 10000 +
                 cast(substring_index(s.version, '.', -1) as unsigned)
          else 0
          end version_sort_id, s.schema_text, s.ddl_text,
          s.instance_created_time, s.created_time, s.wh_etl_exec_id
        from stg_dict_dataset_instance s
        where s.db_id = {db_id}
        on duplicate key update
          deployment_tier=s.deployment_tier, data_center=s.data_center, server_cluster=s.server_cluster, slice=s.slice,
          status_id=s.status_id, native_name=s.native_name, logical_name=s.logical_name, version=s.version,
          schema_text=s.schema_text, ddl_text=s.ddl_text,
          instance_created_time=s.instance_created_time, created_time=s.created_time, wh_etl_exec_id=s.wh_etl_exec_id
          ;

        '''.format(source_file=self.input_file, db_id=self.db_id, wh_etl_exec_id=self.wh_etl_exec_id)
    for state in load_cmd.split(";"):
        self.logger.debug(state)
        cursor.execute(state)
        self.conn_mysql.commit()
    cursor.close()
    self.logger.info("finish loading hbase metadata db_id={db_id} to dict_dataset".format(db_id=self.db_id))


  def load_field(self):
    cursor = self.conn_mysql.cursor()
    load_field_cmd = '''
        DELETE FROM stg_dict_field_detail where db_id = {db_id};
        
        LOAD DATA LOCAL INFILE '{source_file}'
        INTO TABLE stg_dict_field_detail
        FIELDS TERMINATED BY '\Z'
        (urn, sort_id, parent_sort_id, parent_path, field_name, data_type,
         is_nullable, default_value, data_size, namespace, description)
        SET db_id = {db_id};
        
        analyze table stg_dict_field_detail;
        
        update stg_dict_field_detail
        set description = null
        where db_id = {db_id}
           and (char_length(trim(description)) = 0
           or description in ('null', 'N/A', 'nothing', 'empty', 'none'));

        insert into field_comments (
          user_id, comment, created, modified, comment_crc32_checksum
        )
        select 0 user_id, description, now() created, now() modified, crc32(description) from
        (
          select sf.description
          from stg_dict_field_detail sf left join field_comments fc
            on sf.description = fc.comment
          where sf.description is not null
            and fc.id is null
            and sf.db_id = {db_id}
          group by 1 order by 1
        ) d;

        analyze table field_comments;

        -- delete old record if it does not exist in this load batch anymore (but have the dataset id)
        create temporary table if not exists t_deleted_fields (primary key (field_id))
          select x.field_id
            from stg_dict_field_detail s
              join dict_dataset i
                on s.urn = i.urn
                and s.db_id = {db_id}
              right join dict_field_detail x
                on i.id = x.dataset_id
                and s.field_name = x.field_name
                and s.parent_path = x.parent_path
          where s.field_name is null
            and x.dataset_id in (
                       select d.id dataset_id
                       from stg_dict_field_detail k join dict_dataset d
                         on k.urn = d.urn
                        and k.db_id = {db_id}
            )
        ; -- run time : ~2min

        delete from dict_field_detail where field_id in (select field_id from t_deleted_fields);

        -- update the old record if some thing changed
        update dict_field_detail t join
        (
          select x.field_id, s.*
          from (select * from stg_dict_field_detail where db_id = {db_id}) s
            join dict_dataset d
              on s.urn = d.urn
            join dict_field_detail x
              on s.field_name = x.field_name
              and coalesce(s.parent_path, '*') = coalesce(x.parent_path, '*')
              and d.id = x.dataset_id
          where (x.sort_id <> s.sort_id
                or x.parent_sort_id <> s.parent_sort_id
                or x.data_type <> s.data_type
                or x.data_size <> s.data_size or (x.data_size is null XOR s.data_size is null)
                or x.data_precision <> s.data_precision or (x.data_precision is null XOR s.data_precision is null)
                or x.is_nullable <> s.is_nullable or (x.is_nullable is null XOR s.is_nullable is null)
                or x.is_partitioned <> s.is_partitioned or (x.is_partitioned is null XOR s.is_partitioned is null)
                or x.is_distributed <> s.is_distributed or (x.is_distributed is null XOR s.is_distributed is null)
                or x.default_value <> s.default_value or (x.default_value is null XOR s.default_value is null)
                or x.namespace <> s.namespace or (x.namespace is null XOR s.namespace is null)
            )
        ) p
          on t.field_id = p.field_id
        set t.sort_id = p.sort_id,
            t.parent_sort_id = p.parent_sort_id,
            t.data_type = p.data_type,
            t.data_size = p.data_size,
            t.data_precision = p.data_precision,
            t.is_nullable = p.is_nullable,
            t.is_partitioned = p.is_partitioned,
            t.is_distributed = p.is_distributed,
            t.default_value = p.default_value,
            t.namespace = p.namespace,
            t.modified = now()
        ;

       insert ignore into dict_field_detail (
          dataset_id, fields_layout_id, sort_id, parent_sort_id, parent_path,
          field_name, namespace, data_type, data_size, is_nullable, default_value,
           modified
        )
        select
          d.id, 0, sf.sort_id, sf.parent_sort_id, sf.parent_path,
          sf.field_name, sf.namespace, sf.data_type, sf.data_size, sf.is_nullable, sf.default_value, now()
        from stg_dict_field_detail sf join dict_dataset d
          on sf.urn = d.urn
             left join dict_field_detail t
          on d.id = t.dataset_id
         and sf.field_name = t.field_name
         and sf.parent_path = t.parent_path
        where db_id = {db_id} and t.field_id is null
        ;

        analyze table dict_field_detail;

        -- delete old record in stagging
        delete from stg_dict_dataset_field_comment where db_id = {db_id};

        -- insert
        insert into stg_dict_dataset_field_comment
        select t.field_id field_id, fc.id comment_id,  d.id dataset_id, {db_id}
                from stg_dict_field_detail sf join dict_dataset d
                  on sf.urn = d.urn
                      join field_comments fc
                  on sf.description = fc.comment
                      join dict_field_detail t
                  on d.id = t.dataset_id
                 and sf.field_name = t.field_name
                 and sf.parent_path = t.parent_path
        where sf.db_id = {db_id};

        -- have default comment, insert it set default to 0
        insert ignore into dict_dataset_field_comment
        select field_id, comment_id, dataset_id, 0 is_default from stg_dict_dataset_field_comment where field_id in (
          select field_id from dict_dataset_field_comment
          where field_id in (select field_id from stg_dict_dataset_field_comment)
        and is_default = 1 ) and db_id = {db_id};


        -- doesn't have this comment before, insert into it and set as default
        insert ignore into dict_dataset_field_comment
        select sd.field_id, sd.comment_id, sd.dataset_id, 1 from stg_dict_dataset_field_comment sd
        left join dict_dataset_field_comment d
        on d.field_id = sd.field_id
         and d.comment_id = sd.comment_id
        where d.comment_id is null
        and sd.db_id = {db_id};


        DELETE FROM stg_dict_field_detail where db_id = {db_id};

        '''.format(source_file=self.input_field_file, db_id=self.db_id)
    for state in load_field_cmd.split(";"):
        self.logger.debug(state)
        cursor.execute(state)
        self.conn_mysql.commit()
    cursor.close()
    self.logger.info("finish loading hbase metadata db_id={db_id} to dict_field_detail".format(db_id=self.db_id))

  def load_sample(self):
    cursor = self.conn_mysql.cursor()
    load_sample_cmd = '''
    DELETE FROM stg_dict_dataset_sample where db_id = {db_id};

    LOAD DATA LOCAL INFILE '{source_file}'
    INTO TABLE stg_dict_dataset_sample
    FIELDS TERMINATED BY '\Z' ESCAPED BY '\0'
    (urn,ref_urn,data)
    SET db_id = {db_id};

    -- update reference id in stagging table
    UPDATE  stg_dict_dataset_sample s
    JOIN dict_dataset d ON s.ref_urn = d.urn
    SET s.ref_id = d.id
    WHERE s.db_id = {db_id} AND s.ref_urn > '' AND s.ref_urn <> 'null';

    -- first insert ref_id as 0
    INSERT INTO dict_dataset_sample
    ( `dataset_id`,
      `urn`,
      `ref_id`,
      `data`,
      created
    )
    select d.id as dataset_id, s.urn, s.ref_id, s.data, now()
    from stg_dict_dataset_sample s join dict_dataset d on d.urn = s.urn
          where s.db_id = {db_id}
    on duplicate key update
      `data`=s.data, modified=now();

      -- update reference id in final table
    UPDATE dict_dataset_sample d
    RIGHT JOIN stg_dict_dataset_sample s ON d.urn = s.urn
    SET d.ref_id = s.ref_id
    WHERE s.db_id = {db_id} AND d.ref_id = 0;
    '''.format(source_file=self.input_sample_file, db_id=self.db_id)
    for state in load_sample_cmd.split(";"):
        self.logger.debug(state)
        cursor.execute(state)
        self.conn_mysql.commit()
    cursor.close()
    self.logger.info("finish loading hbase sample data db_id={db_id} to dict_dataset_sample".format(db_id=self.db_id))

if __name__ == "__main__":
  args = sys.argv[1]

  l = HBaseLoad(args[Constant.WH_EXEC_ID_KEY])

  # set up connection
  username = args[Constant.WH_DB_USERNAME_KEY]
  password = args[Constant.WH_DB_PASSWORD_KEY]
  JDBC_DRIVER = args[Constant.WH_DB_DRIVER_KEY]
  JDBC_URL = args[Constant.WH_DB_URL_KEY]

  l.input_file = "/var/tmp/wherehows/hbase_meta"
  l.input_field_file = "/var/tmp/wherehows/hbase_field_meta"
  l.input_sample_file = "/var/tmp/wherehows/hbase_sample"

  l.db_id = args[Constant.DB_ID_KEY]
  l.wh_etl_exec_id = args[Constant.WH_EXEC_ID_KEY]
  l.conn_mysql = zxJDBC.connect(JDBC_URL, username, password, JDBC_DRIVER)

  if Constant.INNODB_LOCK_WAIT_TIMEOUT in args:
    lock_wait_time = args[Constant.INNODB_LOCK_WAIT_TIMEOUT]
    l.conn_mysql.cursor().execute("SET innodb_lock_wait_timeout = %s;" % lock_wait_time)

  try:
    l.load_metadata()
    l.load_field()
    l.load_sample()
  finally:
    l.conn_mysql.close()





{'fraction_non_alphanumeric': 0.08378200526441733, 'fraction_numerical': 0.0034697296003828666, 'mean_word_length': 2.281605810757754, 'pattern_counts': {'":': 2, '<': 25, '<?xml version=': 0, '>': 27, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 2, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1557303', 'n_tokens_mistral': 10993, 'n_tokens_neox': 10676, 'n_words': 2459}