/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 * <p/>
 * http://www.apache.org/licenses/LICENSE-2.0
 * <p/>
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.atlas.hive.hook.events;

import org.apache.atlas.hive.hook.AtlasHiveHookContext;
import org.apache.atlas.hive.hook.HiveHook.PreprocessAction;
import org.apache.atlas.utils.PathExtractorContext;
import org.apache.atlas.model.instance.AtlasEntity;
import org.apache.atlas.model.instance.AtlasEntity.AtlasEntitiesWithExtInfo;
import org.apache.atlas.model.instance.AtlasEntity.AtlasEntityWithExtInfo;
import org.apache.atlas.model.instance.AtlasEntity.AtlasEntityExtInfo;
import org.apache.atlas.model.instance.AtlasObjectId;
import org.apache.atlas.model.instance.AtlasRelatedObjectId;
import org.apache.atlas.model.instance.AtlasStruct;
import org.apache.atlas.model.notification.HookNotification;
import org.apache.atlas.type.AtlasTypeUtil;
import org.apache.atlas.utils.AtlasPathExtractorUtil;
import org.apache.atlas.utils.HdfsNameServiceResolver;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.collections.MapUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hive.metastore.api.Database;
import org.apache.hadoop.hive.metastore.api.FieldSchema;
import org.apache.hadoop.hive.metastore.api.Order;
import org.apache.hadoop.hive.metastore.api.SerDeInfo;
import org.apache.hadoop.hive.metastore.api.StorageDescriptor;
import org.apache.hadoop.hive.metastore.utils.SecurityUtils;
import org.apache.hadoop.hive.ql.hooks.*;
import org.apache.hadoop.hive.ql.hooks.LineageInfo.BaseColumnInfo;
import org.apache.hadoop.hive.ql.hooks.LineageInfo.DependencyKey;
import org.apache.hadoop.hive.ql.metadata.Hive;
import org.apache.hadoop.hive.ql.metadata.Table;
import org.apache.hadoop.hive.ql.plan.HiveOperation;
import org.apache.hadoop.security.UserGroupInformation;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.net.URI;
import java.util.*;

import static org.apache.atlas.hive.bridge.HiveMetaStoreBridge.getDatabaseName;
import static org.apache.atlas.hive.hook.AtlasHiveHookContext.QNAME_SEP_METADATA_NAMESPACE;
import static org.apache.atlas.hive.hook.AtlasHiveHookContext.QNAME_SEP_ENTITY_NAME;
import static org.apache.atlas.hive.hook.AtlasHiveHookContext.QNAME_SEP_PROCESS;

public abstract class BaseHiveEvent {
    private static final Logger LOG = LoggerFactory.getLogger(BaseHiveEvent.class);

    public static final String HIVE_TYPE_DB                        = "hive_db";
    public static final String HIVE_TYPE_TABLE                     = "hive_table";
    public static final String HIVE_TYPE_STORAGEDESC               = "hive_storagedesc";
    public static final String HIVE_TYPE_COLUMN                    = "hive_column";
    public static final String HIVE_TYPE_PROCESS                   = "hive_process";
    public static final String HIVE_TYPE_COLUMN_LINEAGE            = "hive_column_lineage";
    public static final String HIVE_TYPE_SERDE                     = "hive_serde";
    public static final String HIVE_TYPE_ORDER                     = "hive_order";
    public static final String HIVE_TYPE_PROCESS_EXECUTION         = "hive_process_execution";
    public static final String HIVE_DB_DDL                         = "hive_db_ddl";
    public static final String HIVE_TABLE_DDL                      = "hive_table_ddl";
    public static final String HBASE_TYPE_TABLE                    = "hbase_table";
    public static final String HBASE_TYPE_NAMESPACE                = "hbase_namespace";
    public static final String ATTRIBUTE_QUALIFIED_NAME            = "qualifiedName";
    public static final String ATTRIBUTE_NAME                      = "name";
    public static final String ATTRIBUTE_DESCRIPTION               = "description";
    public static final String ATTRIBUTE_OWNER                     = "owner";
    public static final String ATTRIBUTE_CLUSTER_NAME              = "clusterName";
    public static final String ATTRIBUTE_LOCATION                  = "location";
    public static final String ATTRIBUTE_LOCATION_PATH             = "locationPath";
    public static final String ATTRIBUTE_PARAMETERS                = "parameters";
    public static final String ATTRIBUTE_OWNER_TYPE                = "ownerType";
    public static final String ATTRIBUTE_COMMENT                   = "comment";
    public static final String ATTRIBUTE_CREATE_TIME               = "createTime";
    public static final String ATTRIBUTE_LAST_ACCESS_TIME          = "lastAccessTime";
    public static final String ATTRIBUTE_VIEW_ORIGINAL_TEXT        = "viewOriginalText";
    public static final String ATTRIBUTE_VIEW_EXPANDED_TEXT        = "viewExpandedText";
    public static final String ATTRIBUTE_TABLE_TYPE                = "tableType";
    public static final String ATTRIBUTE_TEMPORARY                 = "temporary";
    public static final String ATTRIBUTE_RETENTION                 = "retention";
    public static final String ATTRIBUTE_DB                        = "db";
    public static final String ATTRIBUTE_HIVE_DB                   = "hiveDb";
    public static final String ATTRIBUTE_STORAGEDESC               = "sd";
    public static final String ATTRIBUTE_PARTITION_KEYS            = "partitionKeys";
    public static final String ATTRIBUTE_COLUMNS                   = "columns";
    public static final String ATTRIBUTE_INPUT_FORMAT              = "inputFormat";
    public static final String ATTRIBUTE_OUTPUT_FORMAT             = "outputFormat";
    public static final String ATTRIBUTE_COMPRESSED                = "compressed";
    public static final String ATTRIBUTE_BUCKET_COLS               = "bucketCols";
    public static final String ATTRIBUTE_NUM_BUCKETS               = "numBuckets";
    public static final String ATTRIBUTE_STORED_AS_SUB_DIRECTORIES = "storedAsSubDirectories";
    public static final String ATTRIBUTE_TABLE                     = "table";
    public static final String ATTRIBUTE_SERDE_INFO                = "serdeInfo";
    public static final String ATTRIBUTE_SERIALIZATION_LIB         = "serializationLib";
    public static final String ATTRIBUTE_SORT_COLS                 = "sortCols";
    public static final String ATTRIBUTE_COL_TYPE                  = "type";
    public static final String ATTRIBUTE_COL_POSITION              = "position";
    public static final String ATTRIBUTE_PATH                      = "path";
    public static final String ATTRIBUTE_NAMESERVICE_ID            = "nameServiceId";
    public static final String ATTRIBUTE_INPUTS                    = "inputs";
    public static final String ATTRIBUTE_OUTPUTS                   = "outputs";
    public static final String ATTRIBUTE_OPERATION_TYPE            = "operationType";
    public static final String ATTRIBUTE_START_TIME                = "startTime";
    public static final String ATTRIBUTE_USER_NAME                 = "userName";
    public static final String ATTRIBUTE_QUERY_TEXT                = "queryText";
    public static final String ATTRIBUTE_PROCESS                   = "process";
    public static final String ATTRIBUTE_PROCESS_EXECUTIONS        = "processExecutions";
    public static final String ATTRIBUTE_QUERY_ID                  = "queryId";
    public static final String ATTRIBUTE_QUERY_PLAN                = "queryPlan";
    public static final String ATTRIBUTE_END_TIME                  = "endTime";
    public static final String ATTRIBUTE_RECENT_QUERIES            = "recentQueries";
    public static final String ATTRIBUTE_QUERY                     = "query";
    public static final String ATTRIBUTE_DEPENDENCY_TYPE           = "depenendencyType";
    public static final String ATTRIBUTE_EXPRESSION                = "expression";
    public static final String ATTRIBUTE_ALIASES                   = "aliases";
    public static final String ATTRIBUTE_URI                       = "uri";
    public static final String ATTRIBUTE_STORAGE_HANDLER           = "storage_handler";
    public static final String ATTRIBUTE_NAMESPACE                 = "namespace";
    public static final String ATTRIBUTE_HOSTNAME                  = "hostName";
    public static final String ATTRIBUTE_EXEC_TIME                 = "execTime";
    public static final String ATTRIBUTE_DDL_QUERIES               = "ddlQueries";
    public static final String ATTRIBUTE_SERVICE_TYPE              = "serviceType";
    public static final String HBASE_STORAGE_HANDLER_CLASS         = "org.apache.hadoop.hive.hbase.HBaseStorageHandler";
    public static final String HBASE_DEFAULT_NAMESPACE             = "default";
    public static final String HBASE_NAMESPACE_TABLE_DELIMITER     = ":";
    public static final String HBASE_PARAM_TABLE_NAME              = "hbase.table.name";
    public static final long   MILLIS_CONVERT_FACTOR               = 1000;
    public static final String HDFS_PATH_PREFIX                    = "hdfs://";
    public static final String EMPTY_ATTRIBUTE_VALUE = "";

    public static final String RELATIONSHIP_DATASET_PROCESS_INPUTS        = "dataset_process_inputs";
    public static final String RELATIONSHIP_PROCESS_DATASET_OUTPUTS       = "process_dataset_outputs";
    public static final String RELATIONSHIP_HIVE_PROCESS_COLUMN_LINEAGE   = "hive_process_column_lineage";
    public static final String RELATIONSHIP_HIVE_TABLE_DB                 = "hive_table_db";
    public static final String RELATIONSHIP_HIVE_TABLE_PART_KEYS          = "hive_table_partitionkeys";
    public static final String RELATIONSHIP_HIVE_TABLE_COLUMNS            = "hive_table_columns";
    public static final String RELATIONSHIP_HIVE_TABLE_STORAGE_DESC       = "hive_table_storagedesc";
    public static final String RELATIONSHIP_HIVE_PROCESS_PROCESS_EXE      = "hive_process_process_executions";
    public static final String RELATIONSHIP_HIVE_DB_DDL_QUERIES           = "hive_db_ddl_queries";
    public static final String RELATIONSHIP_HIVE_DB_LOCATION              = "hive_db_location";
    public static final String RELATIONSHIP_HIVE_TABLE_DDL_QUERIES        = "hive_table_ddl_queries";
    public static final String RELATIONSHIP_HBASE_TABLE_NAMESPACE         = "hbase_table_namespace";


    public static final Map<Integer, String> OWNER_TYPE_TO_ENUM_VALUE = new HashMap<>();


    static {
        OWNER_TYPE_TO_ENUM_VALUE.put(1, "USER");
        OWNER_TYPE_TO_ENUM_VALUE.put(2, "ROLE");
        OWNER_TYPE_TO_ENUM_VALUE.put(3, "GROUP");
    }

    protected final AtlasHiveHookContext context;


    protected BaseHiveEvent(AtlasHiveHookContext context) {
        this.context = context;
    }

    public AtlasHiveHookContext getContext() {
        return context;
    }

    public List<HookNotification> getNotificationMessages() throws Exception {
        return null;
    }

    public static long getTableCreateTime(Table table) {
        return table.getTTable() != null ? (table.getTTable().getCreateTime() * MILLIS_CONVERT_FACTOR) : System.currentTimeMillis();
    }

    public static String getTableOwner(Table table) {
        return table.getTTable() != null ? (table.getOwner()): "";
    }


    public static List<AtlasObjectId> getObjectIds(List<AtlasEntity> entities) {
        final List<AtlasObjectId> ret;

        if (CollectionUtils.isNotEmpty(entities)) {
            ret = new ArrayList<>(entities.size());

            for (AtlasEntity entity : entities) {
                ret.add(AtlasTypeUtil.getObjectId(entity));
            }
        } else {
            ret = Collections.emptyList();
        }

        return ret;
    }


    protected void addProcessedEntities(AtlasEntitiesWithExtInfo entitiesWithExtInfo) {
        for (AtlasEntity entity : context.getEntities()) {
            entitiesWithExtInfo.addReferredEntity(entity);
        }

        entitiesWithExtInfo.compact();

        context.addToKnownEntities(entitiesWithExtInfo.getEntities());

        if (entitiesWithExtInfo.getReferredEntities() != null) {
            context.addToKnownEntities(entitiesWithExtInfo.getReferredEntities().values());
        }
    }

    protected AtlasEntity getInputOutputEntity(Entity entity, AtlasEntityExtInfo entityExtInfo, boolean skipTempTables) throws Exception {
        AtlasEntity ret = null;

        switch(entity.getType()) {
            case TABLE:
            case PARTITION:
            case DFS_DIR: {
                ret = toAtlasEntity(entity, entityExtInfo, skipTempTables);
            }
            break;
        }

        return ret;
    }

    protected AtlasEntity toAtlasEntity(Entity entity, AtlasEntityExtInfo entityExtInfo, boolean skipTempTables) throws Exception {
        AtlasEntity ret = null;

        switch (entity.getType()) {
            case DATABASE: {
                String dbName = getDatabaseName(entity.getDatabase());

                if (!context.getIgnoreDummyDatabaseName().contains(dbName)) {
                    Database db = getHive().getDatabase(dbName);

                    ret = toDbEntity(db);
                }
            }
            break;

            case TABLE:
            case PARTITION: {
                String  dbName    = entity.getTable().getDbName();
                String  tableName = entity.getTable().getTableName();
                boolean skipTable = StringUtils.isNotEmpty(context.getIgnoreValuesTmpTableNamePrefix()) && tableName.toLowerCase().startsWith(context.getIgnoreValuesTmpTableNamePrefix());

                if (!skipTable) {
                    skipTable = context.getIgnoreDummyTableName().contains(tableName) && context.getIgnoreDummyDatabaseName().contains(dbName);
                }

                if (!skipTable) {
                    skipTable = skipTempTables && entity.getTable().isTemporary();
                }

                if (!skipTable) {
                    Table table = getHive().getTable(dbName, tableName);

                    ret = toTableEntity(table, entityExtInfo);
                } else {
                    context.registerSkippedEntity(entity);
                }
            }
            break;

            case DFS_DIR: {
                URI location = entity.getLocation();

                if (location != null) {
                    ret = getPathEntity(new Path(entity.getLocation()), entityExtInfo);
                }
            }
            break;

            default:
            break;
        }

        return ret;
    }

    protected AtlasEntity toDbEntity(Database db) throws Exception {
        String      dbName          = getDatabaseName(db);
        String      dbQualifiedName = getQualifiedName(db);
        boolean     isKnownDatabase = context.isKnownDatabase(dbQualifiedName);
        AtlasEntity ret             = context.getEntity(dbQualifiedName);

        if (ret == null) {
            ret = new AtlasEntity(HIVE_TYPE_DB);

            // if this DB was sent in an earlier notification, set 'guid' to null - which will:
            //  - result in this entity to be not included in 'referredEntities'
            //  - cause Atlas server to resolve the entity by its qualifiedName
            if (isKnownDatabase) {
                ret.setGuid(null);
            }

            ret.setAttribute(ATTRIBUTE_QUALIFIED_NAME, dbQualifiedName);
            ret.setAttribute(ATTRIBUTE_NAME, dbName);

            if (StringUtils.isNotEmpty(db.getDescription())) {
                ret.setAttribute(ATTRIBUTE_DESCRIPTION, db.getDescription());
            }

            ret.setAttribute(ATTRIBUTE_OWNER, db.getOwnerName());

            ret.setAttribute(ATTRIBUTE_CLUSTER_NAME, getMetadataNamespace());
            ret.setAttribute(ATTRIBUTE_LOCATION, HdfsNameServiceResolver.getPathWithNameServiceID(db.getLocationUri()));
            ret.setAttribute(ATTRIBUTE_PARAMETERS, db.getParameters());

            if (db.getOwnerType() != null) {
                ret.setAttribute(ATTRIBUTE_OWNER_TYPE, OWNER_TYPE_TO_ENUM_VALUE.get(db.getOwnerType().getValue()));
            }

            context.putEntity(dbQualifiedName, ret);
        }

        return ret;
    }

    protected AtlasEntityWithExtInfo toTableEntity(Table table) throws Exception {
        AtlasEntityWithExtInfo ret = new AtlasEntityWithExtInfo();

        AtlasEntity entity = toTableEntity(table, ret);

        if (entity != null) {
            ret.setEntity(entity);
        } else {
            ret = null;
        }

        return ret;
    }

    protected AtlasEntity toTableEntity(Table table, AtlasEntitiesWithExtInfo entities) throws Exception {
        AtlasEntity ret = toTableEntity(table, (AtlasEntityExtInfo) entities);

        if (ret != null) {
            entities.addEntity(ret);
        }

        return ret;
    }

    protected AtlasEntity toTableEntity(Table table, AtlasEntityExtInfo entityExtInfo) throws Exception {
        Database    db       = getDatabases(table.getDbName());
        AtlasEntity dbEntity = toDbEntity(db);

        if (entityExtInfo != null) {
            if (dbEntity != null) {
                entityExtInfo.addReferredEntity(dbEntity);
            }
        }

        AtlasEntity ret = toTableEntity(AtlasTypeUtil.getObjectId(dbEntity), table, entityExtInfo);

        return ret;
    }

    protected AtlasEntity toTableEntity(AtlasObjectId dbId, Table table, AtlasEntityExtInfo entityExtInfo) throws Exception {
        String  tblQualifiedName = getQualifiedName(table);
        boolean isKnownTable     = context.isKnownTable(tblQualifiedName);

        AtlasEntity ret = context.getEntity(tblQualifiedName);

        if (ret == null) {
            PreprocessAction action = context.getPreprocessActionForHiveTable(tblQualifiedName);

            if (action == PreprocessAction.IGNORE) {
                LOG.info("ignoring table {}", tblQualifiedName);
            } else {
                ret = new AtlasEntity(HIVE_TYPE_TABLE);

                // if this table was sent in an earlier notification, set 'guid' to null - which will:
                //  - result in this entity to be not included in 'referredEntities'
                //  - cause Atlas server to resolve the entity by its qualifiedName
                if (isKnownTable && !isAlterTableOperation()) {
                    ret.setGuid(null);
                }

                long createTime     = getTableCreateTime(table);
                long lastAccessTime = table.getLastAccessTime() > 0 ? (table.getLastAccessTime() * MILLIS_CONVERT_FACTOR) : createTime;

                AtlasRelatedObjectId dbRelatedObject =     new AtlasRelatedObjectId(dbId, RELATIONSHIP_HIVE_TABLE_DB);

                ret.setRelationshipAttribute(ATTRIBUTE_DB, dbRelatedObject );
                ret.setAttribute(ATTRIBUTE_QUALIFIED_NAME, tblQualifiedName);
                ret.setAttribute(ATTRIBUTE_NAME, table.getTableName().toLowerCase());
                ret.setAttribute(ATTRIBUTE_OWNER, table.getOwner());
                ret.setAttribute(ATTRIBUTE_CREATE_TIME, createTime);
                ret.setAttribute(ATTRIBUTE_LAST_ACCESS_TIME, lastAccessTime);
                ret.setAttribute(ATTRIBUTE_RETENTION, table.getRetention());
                ret.setAttribute(ATTRIBUTE_PARAMETERS, table.getParameters());
                ret.setAttribute(ATTRIBUTE_COMMENT, table.getParameters().get(ATTRIBUTE_COMMENT));
                ret.setAttribute(ATTRIBUTE_TABLE_TYPE, table.getTableType().name());
                ret.setAttribute(ATTRIBUTE_TEMPORARY, table.isTemporary());

                if (table.getViewOriginalText() != null) {
                    ret.setAttribute(ATTRIBUTE_VIEW_ORIGINAL_TEXT, table.getViewOriginalText());
                }

                if (table.getViewExpandedText() != null) {
                    ret.setAttribute(ATTRIBUTE_VIEW_EXPANDED_TEXT, table.getViewExpandedText());
                }

                boolean pruneTable = table.isTemporary() || action == PreprocessAction.PRUNE;

                if (pruneTable) {
                    LOG.info("ignoring details of table {}", tblQualifiedName);
                } else {
                    AtlasObjectId     tableId       = AtlasTypeUtil.getObjectId(ret);
                    AtlasEntity       sd            = getStorageDescEntity(tableId, table);
                    List<AtlasEntity> partitionKeys = getColumnEntities(tableId, table, table.getPartitionKeys(), RELATIONSHIP_HIVE_TABLE_PART_KEYS);
                    List<AtlasEntity> columns       = getColumnEntities(tableId, table, table.getCols(), RELATIONSHIP_HIVE_TABLE_COLUMNS);


                    if (entityExtInfo != null) {
                        entityExtInfo.addReferredEntity(sd);

                        if (partitionKeys != null) {
                            for (AtlasEntity partitionKey : partitionKeys) {
                                entityExtInfo.addReferredEntity(partitionKey);
                            }
                        }

                        if (columns != null) {
                            for (AtlasEntity column : columns) {
                                entityExtInfo.addReferredEntity(column);
                            }
                        }
                    }


                    ret.setRelationshipAttribute(ATTRIBUTE_STORAGEDESC, AtlasTypeUtil.getAtlasRelatedObjectId(sd, RELATIONSHIP_HIVE_TABLE_STORAGE_DESC));
                    ret.setRelationshipAttribute(ATTRIBUTE_PARTITION_KEYS, AtlasTypeUtil.getAtlasRelatedObjectIds(partitionKeys, RELATIONSHIP_HIVE_TABLE_PART_KEYS));
                    ret.setRelationshipAttribute(ATTRIBUTE_COLUMNS, AtlasTypeUtil.getAtlasRelatedObjectIds(columns, RELATIONSHIP_HIVE_TABLE_COLUMNS));
                }

                context.putEntity(tblQualifiedName, ret);
            }
        }

        return ret;
    }

    protected AtlasEntity getStorageDescEntity(AtlasObjectId tableId, Table table) {
        String  sdQualifiedName = getQualifiedName(table, table.getSd());
        boolean isKnownTable    = tableId.getGuid() == null;

        AtlasEntity ret = context.getEntity(sdQualifiedName);

        if (ret == null) {
            ret = new AtlasEntity(HIVE_TYPE_STORAGEDESC);

            // if sd's table was sent in an earlier notification, set 'guid' to null - which will:
            //  - result in this entity to be not included in 'referredEntities'
            //  - cause Atlas server to resolve the entity by its qualifiedName
            if (isKnownTable) {
                ret.setGuid(null);
            }

            StorageDescriptor sd = table.getSd();

            AtlasRelatedObjectId tableRelatedObject =     new AtlasRelatedObjectId(tableId, RELATIONSHIP_HIVE_TABLE_STORAGE_DESC);

            ret.setRelationshipAttribute(ATTRIBUTE_TABLE, tableRelatedObject);
            ret.setAttribute(ATTRIBUTE_QUALIFIED_NAME, sdQualifiedName);
            ret.setAttribute(ATTRIBUTE_PARAMETERS, sd.getParameters());
            ret.setAttribute(ATTRIBUTE_LOCATION, HdfsNameServiceResolver.getPathWithNameServiceID(sd.getLocation()));
            ret.setAttribute(ATTRIBUTE_INPUT_FORMAT, sd.getInputFormat());
            ret.setAttribute(ATTRIBUTE_OUTPUT_FORMAT, sd.getOutputFormat());
            ret.setAttribute(ATTRIBUTE_COMPRESSED, sd.isCompressed());
            ret.setAttribute(ATTRIBUTE_NUM_BUCKETS, sd.getNumBuckets());
            ret.setAttribute(ATTRIBUTE_STORED_AS_SUB_DIRECTORIES, sd.isStoredAsSubDirectories());

            if (sd.getBucketCols() != null && sd.getBucketCols().size() > 0) {
                ret.setAttribute(ATTRIBUTE_BUCKET_COLS, sd.getBucketCols());
            }

            if (sd.getSerdeInfo() != null) {
                AtlasStruct serdeInfo   = new AtlasStruct(HIVE_TYPE_SERDE);
                SerDeInfo   sdSerDeInfo = sd.getSerdeInfo();

                serdeInfo.setAttribute(ATTRIBUTE_NAME, sdSerDeInfo.getName());
                serdeInfo.setAttribute(ATTRIBUTE_SERIALIZATION_LIB, sdSerDeInfo.getSerializationLib());
                serdeInfo.setAttribute(ATTRIBUTE_PARAMETERS, sdSerDeInfo.getParameters());

                ret.setAttribute(ATTRIBUTE_SERDE_INFO, serdeInfo);
            }

            if (CollectionUtils.isNotEmpty(sd.getSortCols())) {
                List<AtlasStruct> sortCols = new ArrayList<>(sd.getSortCols().size());

                for (Order sdSortCol : sd.getSortCols()) {
                    AtlasStruct sortcol = new AtlasStruct(HIVE_TYPE_ORDER);

                    sortcol.setAttribute("col", sdSortCol.getCol());
                    sortcol.setAttribute("order", sdSortCol.getOrder());

                    sortCols.add(sortcol);
                }

                ret.setAttribute(ATTRIBUTE_SORT_COLS, sortCols);
            }

            context.putEntity(sdQualifiedName, ret);
        }

        return ret;
    }

    protected List<AtlasEntity> getColumnEntities(AtlasObjectId tableId, Table table, List<FieldSchema> fieldSchemas, String relationshipType) {
        List<AtlasEntity> ret            = new ArrayList<>();
        boolean           isKnownTable   = tableId.getGuid() == null;
        int               columnPosition = 0;

        if (CollectionUtils.isNotEmpty(fieldSchemas)) {
            for (FieldSchema fieldSchema : fieldSchemas) {
                String      colQualifiedName = getQualifiedName(table, fieldSchema);
                AtlasEntity column           = context.getEntity(colQualifiedName);

                if (column == null) {
                    column = new AtlasEntity(HIVE_TYPE_COLUMN);

                    // if column's table was sent in an earlier notification, set 'guid' to null - which will:
                    //  - result in this entity to be not included in 'referredEntities'
                    //  - cause Atlas server to resolve the entity by its qualifiedName
                    if (isKnownTable) {
                        column.setGuid(null);
                    }
                    AtlasRelatedObjectId relatedObjectId = new AtlasRelatedObjectId(tableId, relationshipType);
                    column.setRelationshipAttribute(ATTRIBUTE_TABLE, (relatedObjectId));
                    column.setAttribute(ATTRIBUTE_QUALIFIED_NAME, colQualifiedName);
                    column.setAttribute(ATTRIBUTE_NAME, fieldSchema.getName());
                    column.setAttribute(ATTRIBUTE_OWNER, table.getOwner());
                    column.setAttribute(ATTRIBUTE_COL_TYPE, fieldSchema.getType());
                    column.setAttribute(ATTRIBUTE_COL_POSITION, columnPosition++);
                    column.setAttribute(ATTRIBUTE_COMMENT, fieldSchema.getComment());

                    context.putEntity(colQualifiedName, column);
                }

                ret.add(column);
            }
        }

        return ret;
    }

    protected AtlasEntity getPathEntity(Path path, AtlasEntityExtInfo extInfo) {
        String               strPath                  = path.toString();
        String               metadataNamespace        = getMetadataNamespace();
        boolean              isConvertPathToLowerCase = strPath.startsWith(HDFS_PATH_PREFIX) && context.isConvertHdfsPathToLowerCase();
        PathExtractorContext pathExtractorContext     = new PathExtractorContext(metadataNamespace, context.getQNameToEntityMap(),
                                                                                 isConvertPathToLowerCase, context.getAwsS3AtlasModelVersion());

        AtlasEntityWithExtInfo entityWithExtInfo = AtlasPathExtractorUtil.getPathEntity(path, pathExtractorContext);

        if (entityWithExtInfo.getReferredEntities() != null){
            for (AtlasEntity entity : entityWithExtInfo.getReferredEntities().values()) {
                extInfo.addReferredEntity(entity);
            }
        }

        return entityWithExtInfo.getEntity();
    }

    protected AtlasEntity getHiveProcessEntity(List<AtlasEntity> inputs, List<AtlasEntity> outputs) throws Exception {
        AtlasEntity ret           = new AtlasEntity(HIVE_TYPE_PROCESS);
        String      queryStr      = getQueryString();
        String      qualifiedName = getQualifiedName(inputs, outputs);

        if (queryStr != null) {
            queryStr = queryStr.toLowerCase().trim();
        }

        ret.setAttribute(ATTRIBUTE_OPERATION_TYPE, getOperationName());

        if (context.isMetastoreHook()) {
            HiveOperation operation = context.getHiveOperation();

            if (operation == HiveOperation.CREATETABLE || operation == HiveOperation.CREATETABLE_AS_SELECT) {
                AtlasEntity table      = outputs.get(0);
                long        createTime = Long.valueOf((Long)table.getAttribute(ATTRIBUTE_CREATE_TIME));
                qualifiedName          =  (String) table.getAttribute(ATTRIBUTE_QUALIFIED_NAME) + QNAME_SEP_PROCESS + createTime;

                ret.setAttribute(ATTRIBUTE_NAME, "dummyProcess:" + UUID.randomUUID());
                ret.setAttribute(ATTRIBUTE_OPERATION_TYPE, operation.getOperationName());
            }
        }

        ret.setAttribute(ATTRIBUTE_QUALIFIED_NAME, qualifiedName);
        ret.setAttribute(ATTRIBUTE_NAME, qualifiedName);
        ret.setRelationshipAttribute(ATTRIBUTE_INPUTS, AtlasTypeUtil.getAtlasRelatedObjectIds(inputs, RELATIONSHIP_DATASET_PROCESS_INPUTS));
        ret.setRelationshipAttribute(ATTRIBUTE_OUTPUTS, AtlasTypeUtil.getAtlasRelatedObjectIds(outputs, RELATIONSHIP_PROCESS_DATASET_OUTPUTS));

        // We are setting an empty value to these attributes, since now we have a new entity type called hive process
        // execution which captures these values. We have to set empty values here because these attributes are
        // mandatory attributes for hive process entity type.
        ret.setAttribute(ATTRIBUTE_START_TIME, System.currentTimeMillis());
        ret.setAttribute(ATTRIBUTE_END_TIME, System.currentTimeMillis());

        if (context.isHiveProcessPopulateDeprecatedAttributes()) {
            ret.setAttribute(ATTRIBUTE_USER_NAME, getUserName());
            ret.setAttribute(ATTRIBUTE_QUERY_TEXT, queryStr);
            ret.setAttribute(ATTRIBUTE_QUERY_ID, getQueryId());
        } else {
            ret.setAttribute(ATTRIBUTE_USER_NAME, EMPTY_ATTRIBUTE_VALUE);
            ret.setAttribute(ATTRIBUTE_QUERY_TEXT, EMPTY_ATTRIBUTE_VALUE);
            ret.setAttribute(ATTRIBUTE_QUERY_ID, EMPTY_ATTRIBUTE_VALUE);
        }

        ret.setAttribute(ATTRIBUTE_QUERY_PLAN, "Not Supported");
        ret.setAttribute(ATTRIBUTE_RECENT_QUERIES, Collections.singletonList(queryStr));
        ret.setAttribute(ATTRIBUTE_CLUSTER_NAME, getMetadataNamespace());

        return ret;
    }

    protected AtlasEntity getHiveProcessExecutionEntity(AtlasEntity hiveProcess) throws Exception {
        AtlasEntity ret         = new AtlasEntity(HIVE_TYPE_PROCESS_EXECUTION);
        String      queryStr    = getQueryString();

        if (queryStr != null) {
            queryStr = queryStr.toLowerCase().trim();
        }

        Long endTime = System.currentTimeMillis();
        ret.setAttribute(ATTRIBUTE_QUALIFIED_NAME, hiveProcess.getAttribute(ATTRIBUTE_QUALIFIED_NAME).toString() +
                QNAME_SEP_PROCESS + getQueryStartTime().toString() +
                QNAME_SEP_PROCESS + endTime.toString());
        ret.setAttribute(ATTRIBUTE_NAME, ret.getAttribute(ATTRIBUTE_QUALIFIED_NAME));
        ret.setAttribute(ATTRIBUTE_START_TIME, getQueryStartTime());
        ret.setAttribute(ATTRIBUTE_END_TIME, endTime);
        ret.setAttribute(ATTRIBUTE_USER_NAME, getUserName());
        ret.setAttribute(ATTRIBUTE_QUERY_TEXT, queryStr);
        ret.setAttribute(ATTRIBUTE_QUERY_ID, getQueryId());
        ret.setAttribute(ATTRIBUTE_QUERY_PLAN, "Not Supported");
        ret.setAttribute(ATTRIBUTE_HOSTNAME, getContext().getHostName()); //
        AtlasRelatedObjectId hiveProcessRelationObjectId = AtlasTypeUtil.toAtlasRelatedObjectId(hiveProcess, RELATIONSHIP_HIVE_PROCESS_PROCESS_EXE);
        ret.setRelationshipAttribute(ATTRIBUTE_PROCESS, hiveProcessRelationObjectId);
        return ret;
    }

    protected AtlasEntity createHiveDDLEntity(AtlasEntity dbOrTable) {
        return createHiveDDLEntity(dbOrTable, false);
    }

    protected AtlasEntity createHiveDDLEntity(AtlasEntity dbOrTable, boolean excludeEntityGuid) {
        AtlasObjectId objId   = AtlasTypeUtil.getObjectId(dbOrTable);
        AtlasEntity   hiveDDL = null;

        if (excludeEntityGuid) {
            objId.setGuid(null);
        }
        AtlasRelatedObjectId objIdRelatedObject =     new AtlasRelatedObjectId(objId);

        if (StringUtils.equals(objId.getTypeName(), HIVE_TYPE_DB)) {
            hiveDDL = new AtlasEntity(HIVE_DB_DDL);
            objIdRelatedObject.setRelationshipType(RELATIONSHIP_HIVE_DB_DDL_QUERIES);
            hiveDDL.setRelationshipAttribute(ATTRIBUTE_DB, objIdRelatedObject);
        } else if (StringUtils.equals(objId.getTypeName(), HIVE_TYPE_TABLE)) {
            hiveDDL = new AtlasEntity(HIVE_TABLE_DDL);
            objIdRelatedObject.setRelationshipType(RELATIONSHIP_HIVE_TABLE_DDL_QUERIES);
            hiveDDL.setRelationshipAttribute( ATTRIBUTE_TABLE, objIdRelatedObject);
        }

        if (hiveDDL != null) {
            hiveDDL.setAttribute(ATTRIBUTE_SERVICE_TYPE, "hive");
            hiveDDL.setAttribute(ATTRIBUTE_EXEC_TIME, getQueryStartTime());
            hiveDDL.setAttribute(ATTRIBUTE_QUERY_TEXT, getQueryString());
            hiveDDL.setAttribute(ATTRIBUTE_USER_NAME, getUserName());
            hiveDDL.setAttribute(ATTRIBUTE_NAME, getQueryString());
            hiveDDL.setAttribute(ATTRIBUTE_QUALIFIED_NAME, dbOrTable.getAttribute(ATTRIBUTE_QUALIFIED_NAME).toString()
                                                           + QNAME_SEP_PROCESS + getQueryStartTime().toString());
        }

        return hiveDDL;
    }

    protected AtlasEntity createHiveLocationEntity(AtlasEntity dbEntity, AtlasEntitiesWithExtInfo extInfoEntity) {
        AtlasEntity ret          = null;
        String      locationUri  = (String)dbEntity.getAttribute(ATTRIBUTE_LOCATION);

        if (StringUtils.isNotEmpty(locationUri)) {
            Path path = null;

            try {
                path = new Path(locationUri);
            } catch (IllegalArgumentException excp) {
                LOG.warn("failed to create Path from locationUri {}", locationUri, excp);
            }

            if (path != null) {
                ret = getPathEntity(path, extInfoEntity);

                if (ret != null) {
                    AtlasRelatedObjectId dbRelatedObjectId = AtlasTypeUtil.getAtlasRelatedObjectId(dbEntity, RELATIONSHIP_HIVE_DB_LOCATION);

                    ret.setRelationshipAttribute(ATTRIBUTE_HIVE_DB, dbRelatedObjectId);
                }
            }
        }

        return ret;
    }

    protected String getMetadataNamespace() {
        return context.getMetadataNamespace();
    }

    protected Database getDatabases(String dbName) throws Exception {
        return context.isMetastoreHook() ? context.getMetastoreHandler().get_database(dbName) :
                                           context.getHive().getDatabase(dbName);
    }

    protected Hive getHive() {
        return context.getHive();
    }

    protected Set<ReadEntity> getInputs() {
        return context != null ? context.getInputs() : Collections.emptySet();
    }

    protected Set<WriteEntity> getOutputs() {
        return context != null ? context.getOutputs() : Collections.emptySet();
    }

    protected LineageInfo getLineageInfo() {
        return context != null ? context.getLineageInfo() : null;
    }

    protected String getQueryString() {
        return isHiveContextValid() ? context.getHiveContext().getQueryPlan().getQueryStr() : null;
    }

    protected String getOperationName() {
        return isHiveContextValid() ? context.getHiveContext().getOperationName() : null;
    }

    protected String getHiveUserName() {
        return isHiveContextValid() ? context.getHiveContext().getUserName() : null;
    }

    protected UserGroupInformation getUgi() {
        return isHiveContextValid() ? context.getHiveContext().getUgi() : null;
    }

    protected Long getQueryStartTime() {
        return isHiveContextValid() ? context.getHiveContext().getQueryPlan().getQueryStartTime() : System.currentTimeMillis();
    }

    protected String getQueryId() {
        return isHiveContextValid() ? context.getHiveContext().getQueryPlan().getQueryId() : null;
    }

    private boolean isHiveContextValid() {
        return context != null && context.getHiveContext() != null;
    }

    protected String getUserName() {
        String               ret = null;
        UserGroupInformation ugi = null;

        if (context.isMetastoreHook()) {
            try {
                ugi = SecurityUtils.getUGI();
            } catch (Exception e) {
                //do nothing
            }
        } else {
            ret = getHiveUserName();

            if (StringUtils.isEmpty(ret)) {
                ugi = getUgi();
            }
        }

        if (ugi != null) {
            ret = ugi.getShortUserName();
        }

        if (StringUtils.isEmpty(ret)) {
            try {
                ret = UserGroupInformation.getCurrentUser().getShortUserName();
            } catch (IOException e) {
                LOG.warn("Failed for UserGroupInformation.getCurrentUser() ", e);

                ret = System.getProperty("user.name");
            }
        }

        return ret;
    }

    protected String getQualifiedName(Entity entity) throws Exception {
        switch (entity.getType()) {
            case DATABASE:
                return getQualifiedName(entity.getDatabase());

            case TABLE:
            case PARTITION:
                return getQualifiedName(entity.getTable());

            case DFS_DIR:
                return getQualifiedName(entity.getLocation());
        }

        return null;
    }

    protected String getQualifiedName(Database db) {
        return context.getQualifiedName(db);
    }

    protected String getQualifiedName(Table table) {
        return context.getQualifiedName(table);
    }

    protected String getQualifiedName(Table table, StorageDescriptor sd) {
        return getQualifiedName(table) + "_storage";
    }

    protected String getQualifiedName(Table table, FieldSchema column) {
        String tblQualifiedName = getQualifiedName(table);

        int sepPos = tblQualifiedName.lastIndexOf(QNAME_SEP_METADATA_NAMESPACE);

        if (sepPos == -1) {
            return tblQualifiedName + QNAME_SEP_ENTITY_NAME + column.getName().toLowerCase();
        } else {
            return tblQualifiedName.substring(0, sepPos) + QNAME_SEP_ENTITY_NAME + column.getName().toLowerCase() + tblQualifiedName.substring(sepPos);
        }
    }

    protected String getQualifiedName(DependencyKey column) {
        String dbName    = column.getDataContainer().getTable().getDbName();
        String tableName = column.getDataContainer().getTable().getTableName();
        String colName   = column.getFieldSchema().getName();

        return getQualifiedName(dbName, tableName, colName);
    }

    protected String getQualifiedName(BaseColumnInfo column) {
        String dbName            = column.getTabAlias().getTable().getDbName();
        String tableName         = column.getTabAlias().getTable().getTableName();
        String colName           = column.getColumn() != null ? column.getColumn().getName() : null;
        String metadataNamespace = getMetadataNamespace();

        if (colName == null) {
            return (dbName + QNAME_SEP_ENTITY_NAME + tableName + QNAME_SEP_METADATA_NAMESPACE).toLowerCase() + metadataNamespace;
        } else {
            return (dbName + QNAME_SEP_ENTITY_NAME + tableName + QNAME_SEP_ENTITY_NAME + colName + QNAME_SEP_METADATA_NAMESPACE).toLowerCase() + metadataNamespace;
        }
    }

    protected String getQualifiedName(String dbName, String tableName, String colName) {
        return (dbName + QNAME_SEP_ENTITY_NAME + tableName + QNAME_SEP_ENTITY_NAME + colName + QNAME_SEP_METADATA_NAMESPACE).toLowerCase() + getMetadataNamespace();
    }

    protected String getQualifiedName(URI location) {
        String strPath = new Path(location).toString();

        if (strPath.startsWith(HDFS_PATH_PREFIX) && context.isConvertHdfsPathToLowerCase()) {
            strPath = strPath.toLowerCase();
        }

        String nameServiceID = HdfsNameServiceResolver.getNameServiceIDForPath(strPath);
        String attrPath      = StringUtils.isEmpty(nameServiceID) ? strPath : HdfsNameServiceResolver.getPathWithNameServiceID(strPath);

        return getQualifiedName(attrPath);
    }

    protected String getQualifiedName(String path) {
        if (path.startsWith(HdfsNameServiceResolver.HDFS_SCHEME)) {
            return path + QNAME_SEP_METADATA_NAMESPACE + getMetadataNamespace();
        }

        return path.toLowerCase();
    }

    protected String getColumnQualifiedName(String tblQualifiedName, String columnName) {
        int sepPos = tblQualifiedName.lastIndexOf(QNAME_SEP_METADATA_NAMESPACE);

        if (sepPos == -1) {
            return tblQualifiedName + QNAME_SEP_ENTITY_NAME + columnName.toLowerCase();
        } else {
            return tblQualifiedName.substring(0, sepPos) + QNAME_SEP_ENTITY_NAME + columnName.toLowerCase() + tblQualifiedName.substring(sepPos);
        }

    }

    protected String getQualifiedName(List<AtlasEntity> inputs, List<AtlasEntity> outputs) throws Exception {
        HiveOperation operation = context.getHiveOperation();

        if (operation == HiveOperation.CREATETABLE ||
            operation == HiveOperation.CREATETABLE_AS_SELECT ||
            operation == HiveOperation.CREATEVIEW ||
            operation == HiveOperation.ALTERVIEW_AS ||
            operation == HiveOperation.ALTERTABLE_LOCATION) {
            List<? extends Entity> sortedEntities = new ArrayList<>(getOutputs());

            Collections.sort(sortedEntities, entityComparator);

            for (Entity entity : sortedEntities) {
                if (entity.getType() == Entity.Type.TABLE) {
                    Table table = entity.getTable();

                    table = getHive().getTable(table.getDbName(), table.getTableName());

                    long createTime = getTableCreateTime(table);

                    return getQualifiedName(table) + QNAME_SEP_PROCESS + createTime;
                }
            }
        }

        String qualifiedName = null;
        String operationName = getOperationName();

        if (operationName != null) {
            StringBuilder sb = new StringBuilder(operationName);

            boolean ignoreHDFSPaths = ignoreHDFSPathsinProcessQualifiedName();

            addToProcessQualifiedName(sb, getInputs(), ignoreHDFSPaths);
            sb.append("->");
            addToProcessQualifiedName(sb, getOutputs(), ignoreHDFSPaths);

            qualifiedName = sb.toString();
        }


        return qualifiedName;
    }

    protected AtlasEntity toReferencedHBaseTable(Table table, AtlasEntitiesWithExtInfo entities) {
        AtlasEntity    ret               = null;
        HBaseTableInfo hBaseTableInfo    = new HBaseTableInfo(table);
        String         hbaseNameSpace    = hBaseTableInfo.getHbaseNameSpace();
        String         hbaseTableName    = hBaseTableInfo.getHbaseTableName();
        String         metadataNamespace = getMetadataNamespace();

        if (hbaseTableName != null) {
            AtlasEntity nsEntity = new AtlasEntity(HBASE_TYPE_NAMESPACE);
            nsEntity.setAttribute(ATTRIBUTE_NAME, hbaseNameSpace);
            nsEntity.setAttribute(ATTRIBUTE_CLUSTER_NAME, metadataNamespace);
            nsEntity.setAttribute(ATTRIBUTE_QUALIFIED_NAME, getHBaseNameSpaceQualifiedName(metadataNamespace, hbaseNameSpace));

            ret = new AtlasEntity(HBASE_TYPE_TABLE);

            ret.setAttribute(ATTRIBUTE_NAME, hbaseTableName);
            ret.setAttribute(ATTRIBUTE_URI, hbaseTableName);

            AtlasRelatedObjectId objIdRelatedObject = new AtlasRelatedObjectId(AtlasTypeUtil.getObjectId(nsEntity), RELATIONSHIP_HBASE_TABLE_NAMESPACE);

            ret.setRelationshipAttribute(ATTRIBUTE_NAMESPACE, objIdRelatedObject);
            ret.setAttribute(ATTRIBUTE_QUALIFIED_NAME, getHBaseTableQualifiedName(metadataNamespace, hbaseNameSpace, hbaseTableName));

            entities.addReferredEntity(nsEntity);
            entities.addEntity(ret);
        }

        return ret;
    }

    protected boolean isHBaseStore(Table table) {
        boolean             ret        = false;
        Map<String, String> parameters = table.getParameters();

        if (MapUtils.isNotEmpty(parameters)) {
            String storageHandler = parameters.get(ATTRIBUTE_STORAGE_HANDLER);

            ret = (storageHandler != null && storageHandler.equals(HBASE_STORAGE_HANDLER_CLASS));
        }

        return ret;
    }

    private static String getHBaseTableQualifiedName(String metadataNamespace, String nameSpace, String tableName) {
        return String.format("%s:%s@%s", nameSpace.toLowerCase(), tableName.toLowerCase(), metadataNamespace);
    }

    private static String getHBaseNameSpaceQualifiedName(String metadataNamespace, String nameSpace) {
        return String.format("%s@%s", nameSpace.toLowerCase(), metadataNamespace);
    }

    private boolean ignoreHDFSPathsinProcessQualifiedName() {
        switch (context.getHiveOperation()) {
            case LOAD:
            case IMPORT:
                return hasPartitionEntity(getOutputs());
            case EXPORT:
                return hasPartitionEntity(getInputs());
            case QUERY:
                return true;
        }

        return false;
    }

    private boolean hasPartitionEntity(Collection<? extends Entity> entities) {
        if (entities != null) {
            for (Entity entity : entities) {
                if (entity.getType() == Entity.Type.PARTITION) {
                    return true;
                }
            }
        }

        return false;
    }

    private void addToProcessQualifiedName(StringBuilder processQualifiedName, Collection<? extends Entity> entities, boolean ignoreHDFSPaths) {
        if (entities == null) {
            return;
        }

        List<? extends Entity> sortedEntities = new ArrayList<>(entities);

        Collections.sort(sortedEntities, entityComparator);

        Set<String> dataSetsProcessed = new HashSet<>();
        Map<String, Table> tableMap   = new HashMap<>();

        for (Entity entity : sortedEntities) {
            if (ignoreHDFSPaths && (Entity.Type.DFS_DIR.equals(entity.getType()) || Entity.Type.LOCAL_DIR.equals(entity.getType()))) {
                continue;
            }

            String qualifiedName = null;
            long   createTime    = 0;

            try {
                if (entity.getType() == Entity.Type.PARTITION || entity.getType() == Entity.Type.TABLE) {
                    String tableKey = entity.getTable().getDbName() + "." + entity.getTable().getTableName();
                    Table  table    = tableMap.get(tableKey);

                    if (table == null) {
                        table = getHive().getTable(entity.getTable().getDbName(), entity.getTable().getTableName());

                        tableMap.put(tableKey, table); //since there could be several partitions in a table, store it to avoid hive calls.
                    }
                    if (table != null) {
                        createTime    = getTableCreateTime(table);
                        qualifiedName = getQualifiedName(table);
                    }
                } else {
                    qualifiedName = getQualifiedName(entity);
                }
            } catch (Exception excp) {
                LOG.error("error while computing qualifiedName for process", excp);
            }

            if (qualifiedName == null || !dataSetsProcessed.add(qualifiedName)) {
                continue;
            }

            if (entity instanceof WriteEntity) { // output entity
                WriteEntity writeEntity = (WriteEntity) entity;

                if (writeEntity.getWriteType() != null && HiveOperation.QUERY.equals(context.getHiveOperation())) {
                    boolean addWriteType = false;

                    switch (((WriteEntity) entity).getWriteType()) {
                        case INSERT:
                        case INSERT_OVERWRITE:
                        case UPDATE:
                        case DELETE:
                            addWriteType = true;
                        break;

                        case PATH_WRITE:
                            addWriteType = !Entity.Type.LOCAL_DIR.equals(entity.getType());
                        break;
                    }

                    if (addWriteType) {
                        processQualifiedName.append(QNAME_SEP_PROCESS).append(writeEntity.getWriteType().name());
                    }
                }
            }

            processQualifiedName.append(QNAME_SEP_PROCESS).append(qualifiedName.toLowerCase().replaceAll("/", ""));

            if (createTime != 0) {
                processQualifiedName.append(QNAME_SEP_PROCESS).append(createTime);
            }
        }
    }

    private boolean isAlterTableOperation() {
        switch (context.getHiveOperation()) {
            case ALTERTABLE_FILEFORMAT:
            case ALTERTABLE_CLUSTER_SORT:
            case ALTERTABLE_BUCKETNUM:
            case ALTERTABLE_PROPERTIES:
            case ALTERTABLE_SERDEPROPERTIES:
            case ALTERTABLE_SERIALIZER:
            case ALTERTABLE_ADDCOLS:
            case ALTERTABLE_REPLACECOLS:
            case ALTERTABLE_PARTCOLTYPE:
            case ALTERTABLE_LOCATION:
            case ALTERTABLE_RENAME:
            case ALTERTABLE_RENAMECOL:
            case ALTERVIEW_PROPERTIES:
            case ALTERVIEW_RENAME:
            case ALTERVIEW_AS:
                return true;
        }

        return false;
    }

    static final class EntityComparator implements Comparator<Entity> {
        @Override
        public int compare(Entity entity1, Entity entity2) {
            String name1 = entity1.getName();
            String name2 = entity2.getName();

            if (name1 == null || name2 == null) {
                name1 = entity1.getD().toString();
                name2 = entity2.getD().toString();
            }

            return name1.toLowerCase().compareTo(name2.toLowerCase());
        }
    }

    static final Comparator<Entity> entityComparator = new EntityComparator();

    static final class HBaseTableInfo {
        String hbaseNameSpace = null;
        String hbaseTableName = null;

         HBaseTableInfo(Table table) {
            Map<String, String> parameters = table.getParameters();

            if (MapUtils.isNotEmpty(parameters)) {
                hbaseNameSpace = HBASE_DEFAULT_NAMESPACE;
                hbaseTableName = parameters.get(HBASE_PARAM_TABLE_NAME);

                if (hbaseTableName != null) {
                    if (hbaseTableName.contains(HBASE_NAMESPACE_TABLE_DELIMITER)) {
                        String[] hbaseTableInfo = hbaseTableName.split(HBASE_NAMESPACE_TABLE_DELIMITER);

                        if (hbaseTableInfo.length > 1) {
                            hbaseNameSpace = hbaseTableInfo[0];
                            hbaseTableName = hbaseTableInfo[1];
                        }
                    }
                }
            }
        }

        public String getHbaseNameSpace() {
            return hbaseNameSpace;
        }

        public String getHbaseTableName() {
            return hbaseTableName;
        }
    }

    public static Table toTable(org.apache.hadoop.hive.metastore.api.Table table) {
        return new Table(table);
    }
}