content: /**
   * 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;
  
  import com.google.common.annotations.VisibleForTesting;
  import org.apache.atlas.ApplicationProperties;
  import org.apache.atlas.AtlasClient;
  import org.apache.atlas.AtlasClientV2;
  import org.apache.atlas.AtlasServiceException;
  import org.apache.atlas.hive.bridge.ColumnLineageUtils;
  import org.apache.atlas.hive.bridge.HiveMetaStoreBridge;
  import org.apache.atlas.hive.hook.HiveHookIT;
  import org.apache.atlas.hive.model.HiveDataTypes;
  import org.apache.atlas.model.instance.AtlasEntity;
  import org.apache.atlas.model.instance.AtlasObjectId;
  import org.apache.atlas.model.instance.AtlasStruct;
  import org.apache.atlas.model.notification.HookNotification;
  import org.apache.atlas.utils.AuthenticationUtil;
  import org.apache.atlas.utils.ParamChecker;
  import org.apache.commons.configuration.Configuration;
  import org.apache.commons.lang.RandomStringUtils;
  import org.apache.commons.lang.StringUtils;
  import org.apache.hadoop.fs.Path;
  import org.apache.hadoop.hive.conf.HiveConf;
  import org.apache.hadoop.hive.ql.Driver;
  import org.apache.hadoop.hive.ql.hooks.Entity;
  import org.apache.hadoop.hive.ql.hooks.HookContext;
  import org.apache.hadoop.hive.ql.hooks.LineageInfo;
  import org.apache.hadoop.hive.ql.hooks.ReadEntity;
  import org.apache.hadoop.hive.ql.hooks.WriteEntity;
  import org.apache.hadoop.hive.ql.metadata.HiveException;
  import org.apache.hadoop.hive.ql.metadata.Table;
  import org.apache.hadoop.hive.ql.plan.HiveOperation;
  import org.apache.hadoop.hive.ql.processors.CommandProcessorResponse;
  import org.apache.hadoop.hive.ql.session.SessionState;
  import org.apache.hadoop.security.UserGroupInformation;
  import org.json.JSONObject;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  import org.testng.annotations.BeforeClass;
  
  import java.io.File;
  import java.util.ArrayList;
  import java.util.Collection;
  import java.util.Collections;
  import java.util.Comparator;
  import java.util.HashSet;
  import java.util.LinkedHashSet;
  import java.util.List;
  import java.util.Map;
  import java.util.Set;
  import java.util.SortedMap;
  import java.util.SortedSet;
  
  import static com.sun.jersey.api.client.ClientResponse.Status.NOT_FOUND;
  import static org.apache.atlas.hive.bridge.HiveMetaStoreBridge.HDFS_PATH;
  import static org.apache.atlas.hive.hook.events.BaseHiveEvent.ATTRIBUTE_QUALIFIED_NAME;
  import static org.apache.atlas.hive.model.HiveDataTypes.HIVE_DB;
  import static org.testng.Assert.assertEquals;
  import static org.testng.Assert.assertNotNull;
  import static org.testng.Assert.assertTrue;
  import static org.testng.Assert.fail;
  
  public class HiveITBase {
      private static final Logger LOG = LoggerFactory.getLogger(HiveITBase.class);
  
      public    static final String DEFAULT_DB   = "default";
      public    static final String SEP          = ":".intern();
      public    static final String IO_SEP       = "->".intern();
      protected static final String DGI_URL      = "http://localhost:21000/";
      protected static final String CLUSTER_NAME = "primary";
      protected static final String PART_FILE    = "2015-01-01";
      protected static final String INPUTS       = "inputs";
      protected static final String OUTPUTS      = "outputs";
  
  
      protected Driver              driver;
      protected AtlasClient         atlasClient;
      protected AtlasClientV2       atlasClientV2;
      protected HiveMetaStoreBridge hiveMetaStoreBridge;
      protected SessionState        ss;
      protected HiveConf            conf;
      protected Driver              driverWithoutContext;
  
      private static final String REFERENCEABLE_ATTRIBUTE_NAME = "qualifiedName";
      private static final String ATTR_NAME                    = "name";
  
  
      @BeforeClass
      public void setUp() throws Exception {
          //Set-up hive session
          conf = new HiveConf();
          conf.setClassLoader(Thread.currentThread().getContextClassLoader());
          conf.set("hive.metastore.event.listeners", "");
  
          // 'driver' using this configuration will be used for tests in HiveHookIT
          //  HiveHookIT will use this driver to test post-execution hooks in HiveServer2.
          //  initialize 'driver' with HMS hook disabled.
          driver = new Driver(conf);
          ss     = new SessionState(conf);
          ss     = SessionState.start(ss);
  
          SessionState.setCurrentSessionState(ss);
  
          Configuration configuration = ApplicationProperties.get();
  
          String[] atlasEndPoint = configuration.getStringArray(HiveMetaStoreBridge.ATLAS_ENDPOINT);
  
          if (atlasEndPoint == null || atlasEndPoint.length == 0) {
              atlasEndPoint = new String[] { DGI_URL };
          }
  
          if (!AuthenticationUtil.isKerberosAuthenticationEnabled()) {
              atlasClientV2 = new AtlasClientV2(atlasEndPoint, new String[]{"admin", "admin"});
              atlasClient   = new AtlasClient(atlasEndPoint, new String[]{"admin", "admin"});
          } else {
              atlasClientV2 = new AtlasClientV2(atlasEndPoint);
              atlasClient   = new AtlasClient(atlasEndPoint);
          }
  
          hiveMetaStoreBridge = new HiveMetaStoreBridge(configuration, conf, atlasClientV2);
  
          HiveConf conf = new HiveConf();
  
          conf.set("hive.exec.post.hooks", "");
  
          SessionState ss = new SessionState(conf);
          ss = SessionState.start(ss);
          SessionState.setCurrentSessionState(ss);
  
          // 'driverWithoutContext' using this configuration will be used for tests in HiveMetastoreHookIT
          //  HiveMetastoreHookIT will use this driver to test event listeners in HiveMetastore.
          //  initialize 'driverWithoutContext' with HiveServer2 post execution hook disabled.
          driverWithoutContext = new Driver(conf);
      }
  
      protected void runCommand(String cmd) throws Exception {
          runCommandWithDelay(cmd, 0);
      }
  
      protected void runCommand(Driver driver, String cmd) throws Exception {
          runCommandWithDelay(driver, cmd, 0);
      }
  
      protected void runCommandWithDelay(String cmd, int sleepMs) throws Exception {
          runCommandWithDelay(driver, cmd, sleepMs);
      }
  
      protected void runCommandWithDelay(Driver driver, String cmd, int sleepMs) throws Exception {
          LOG.debug("Running command '{}'", cmd);
  
          CommandProcessorResponse response = driver.run(cmd);
  
          assertEquals(response.getResponseCode(), 0);
  
          if (sleepMs != 0) {
              Thread.sleep(sleepMs);
          }
      }
  
      protected String createTestDFSPath(String path) throws Exception {
          return "file://" + mkdir(path);
      }
  
      protected String file(String tag) throws Exception {
          String filename = System.getProperty("user.dir") + "/target/" + tag + "-data-" + random();
          File file = new File(filename);
          file.createNewFile();
          return file.getAbsolutePath();
      }
  
      protected String mkdir(String tag) throws Exception {
          String filename = "./target/" + tag + "-data-" + random();
          File file = new File(filename);
          file.mkdirs();
          return file.getAbsolutePath();
      }
  
      public static String lower(String str) {
          if (StringUtils.isEmpty(str)) {
              return null;
          }
          return str.toLowerCase().trim();
      }
  
      protected String random() {
          return RandomStringUtils.randomAlphanumeric(10).toLowerCase();
      }
  
      protected String tableName() {
          return "table_" + random();
      }
  
      protected String dbName() {
          return "db_" + random();
      }
  
      protected String assertTableIsRegistered(String dbName, String tableName) throws Exception {
          return assertTableIsRegistered(dbName, tableName, null, false);
      }
  
      protected String assertTableIsRegistered(String dbName, String tableName, HiveHookIT.AssertPredicate assertPredicate, boolean isTemporary) throws Exception {
          LOG.debug("Searching for table {}.{}", dbName, tableName);
          String tableQualifiedName = HiveMetaStoreBridge.getTableQualifiedName(CLUSTER_NAME, dbName, tableName, isTemporary);
          return assertEntityIsRegistered(HiveDataTypes.HIVE_TABLE.getName(), REFERENCEABLE_ATTRIBUTE_NAME, tableQualifiedName,
                  assertPredicate);
      }
  
      protected String assertEntityIsRegistered(final String typeName, final String property, final String value,
                                                final HiveHookIT.AssertPredicate assertPredicate) throws Exception {
          waitFor(100000, new HiveHookIT.Predicate() {
              @Override
              public void evaluate() throws Exception {
                  AtlasEntity.AtlasEntityWithExtInfo atlasEntityWithExtInfo = atlasClientV2.getEntityByAttribute(typeName, Collections.singletonMap(property,value));
                  AtlasEntity entity = atlasEntityWithExtInfo.getEntity();
                  assertNotNull(entity);
                  if (assertPredicate != null) {
                      assertPredicate.assertOnEntity(entity);
                  }
              }
          });
          AtlasEntity.AtlasEntityWithExtInfo atlasEntityWithExtInfo = atlasClientV2.getEntityByAttribute(typeName, Collections.singletonMap(property,value));
          AtlasEntity entity = atlasEntityWithExtInfo.getEntity();
          return (String) entity.getGuid();
      }
  
      protected String assertEntityIsRegisteredViaGuid(String guid,
                                                final HiveHookIT.AssertPredicate assertPredicate) throws Exception {
          waitFor(100000, new HiveHookIT.Predicate() {
              @Override
              public void evaluate() throws Exception {
                      AtlasEntity.AtlasEntityWithExtInfo atlasEntityWithExtInfo = atlasClientV2.getEntityByGuid(guid);
                      AtlasEntity entity = atlasEntityWithExtInfo.getEntity();
                      assertNotNull(entity);
                      if (assertPredicate != null) {
                          assertPredicate.assertOnEntity(entity);
                      }
  
              }
          });
          AtlasEntity.AtlasEntityWithExtInfo atlasEntityWithExtInfo = atlasClientV2.getEntityByGuid(guid);
          AtlasEntity entity = atlasEntityWithExtInfo.getEntity();
          return (String) entity.getGuid();
      }
  
      protected AtlasEntity assertEntityIsRegistedViaEntity(final String typeName, final String property, final String value,
                                                final HiveHookIT.AssertPredicate assertPredicate) throws Exception {
          waitFor(80000, new HiveHookIT.Predicate() {
              @Override
              public void evaluate() throws Exception {
                  AtlasEntity.AtlasEntityWithExtInfo atlasEntityWithExtInfo = atlasClientV2.getEntityByAttribute(typeName, Collections.singletonMap(property,value));
                  AtlasEntity entity = atlasEntityWithExtInfo.getEntity();
                  assertNotNull(entity);
                  if (assertPredicate != null) {
                      assertPredicate.assertOnEntity(entity);
                  }
              }
          });
          AtlasEntity.AtlasEntityWithExtInfo atlasEntityWithExtInfo = atlasClientV2.getEntityByAttribute(typeName, Collections.singletonMap(property,value));
          AtlasEntity entity = atlasEntityWithExtInfo.getEntity();
          return entity;
      }
  
      public interface AssertPredicate {
          void assertOnEntity(AtlasEntity entity) throws Exception;
      }
  
      public interface Predicate {
          /**
           * Perform a predicate evaluation.
           *
           * @return the boolean result of the evaluation.
           * @throws Exception thrown if the predicate evaluation could not evaluate.
           */
          void evaluate() throws Exception;
      }
  
      /**
       * Wait for a condition, expressed via a {@link Predicate} to become true.
       *
       * @param timeout maximum time in milliseconds to wait for the predicate to become true.
       * @param predicate predicate waiting on.
       */
      protected void waitFor(int timeout, Predicate predicate) throws Exception {
          ParamChecker.notNull(predicate, "predicate");
          long mustEnd = System.currentTimeMillis() + timeout;
  
          while (true) {
              try {
                  predicate.evaluate();
                  return;
              } catch(Error | Exception e) {
                  if (System.currentTimeMillis() >= mustEnd) {
                      fail("Assertions failed. Failing after waiting for timeout " + timeout + " msecs", e);
                  }
                  LOG.debug("Waiting up to {} msec as assertion failed", mustEnd - System.currentTimeMillis(), e);
                  Thread.sleep(5000);
              }
          }
      }
  
      protected String getTableProcessQualifiedName(String dbName, String tableName) throws Exception {
          return HiveMetaStoreBridge.getTableProcessQualifiedName(CLUSTER_NAME,
                  hiveMetaStoreBridge.getHiveClient().getTable(dbName, tableName));
      }
  
      protected void validateHDFSPaths(AtlasEntity processEntity, String attributeName, String... testPaths) throws Exception {
          List<AtlasObjectId> hdfsPathIds = toAtlasObjectIdList(processEntity.getAttribute(attributeName));
  
          for (String testPath : testPaths) {
              Path   path           = new Path(testPath);
              String testPathNormed = lower(path.toString());
              String hdfsPathId     = assertHDFSPathIsRegistered(testPathNormed);
  
              assertHDFSPathIdsContain(hdfsPathIds, hdfsPathId);
          }
      }
  
      private void assertHDFSPathIdsContain(List<AtlasObjectId> hdfsPathObjectIds, String hdfsPathId) {
          Set<String> hdfsPathGuids = new HashSet<>();
  
          for (AtlasObjectId hdfsPathObjectId : hdfsPathObjectIds) {
              hdfsPathGuids.add(hdfsPathObjectId.getGuid());
          }
  
          assertTrue(hdfsPathGuids.contains(hdfsPathId));
      }
  
      protected String assertHDFSPathIsRegistered(String path) throws Exception {
          LOG.debug("Searching for hdfs path {}", path);
          // ATLAS-2444 HDFS name node federation adds the cluster name to the qualifiedName
          if (path.startsWith("hdfs://")) {
              String pathWithCluster = path + "@" + CLUSTER_NAME;
              return assertEntityIsRegistered(HDFS_PATH, REFERENCEABLE_ATTRIBUTE_NAME, pathWithCluster, null);
          } else {
              return assertEntityIsRegistered(HDFS_PATH, REFERENCEABLE_ATTRIBUTE_NAME, path, null);
          }
      }
  
      protected String assertDatabaseIsRegistered(String dbName) throws Exception {
          return assertDatabaseIsRegistered(dbName, null);
      }
  
      protected String assertDatabaseIsRegistered(String dbName, AssertPredicate assertPredicate) throws Exception {
          LOG.debug("Searching for database: {}", dbName);
  
          String dbQualifiedName = HiveMetaStoreBridge.getDBQualifiedName(CLUSTER_NAME, dbName);
  
          return assertEntityIsRegistered(HIVE_DB.getName(), REFERENCEABLE_ATTRIBUTE_NAME, dbQualifiedName, assertPredicate);
      }
  
      public void assertDatabaseIsNotRegistered(String dbName) throws Exception {
          LOG.debug("Searching for database {}", dbName);
          String dbQualifiedName = HiveMetaStoreBridge.getDBQualifiedName(CLUSTER_NAME, dbName);
          assertEntityIsNotRegistered(HIVE_DB.getName(), ATTRIBUTE_QUALIFIED_NAME, dbQualifiedName);
      }
  
      protected void assertEntityIsNotRegistered(final String typeName, final String property, final String value) throws Exception {
          // wait for sufficient time before checking if entity is not available.
          long waitTime = 10000;
          LOG.debug("Waiting for {} msecs, before asserting entity is not registered.", waitTime);
          Thread.sleep(waitTime);
  
          try {
              atlasClientV2.getEntityByAttribute(typeName, Collections.singletonMap(property, value));
  
              fail(String.format("Entity was not supposed to exist for typeName = %s, attributeName = %s, attributeValue = %s", typeName, property, value));
          } catch (AtlasServiceException e) {
              if (e.getStatus() == NOT_FOUND) {
                  return;
              }
          }
      }
  
      protected AtlasEntity getAtlasEntityByType(String type, String id) throws Exception {
          AtlasEntity atlasEntity = null;
          AtlasEntity.AtlasEntityWithExtInfo atlasEntityWithExtInfoForProcess = atlasClientV2.getEntityByAttribute(type,
                  Collections.singletonMap(AtlasClient.GUID, id));
          atlasEntity = atlasEntityWithExtInfoForProcess.getEntity();
          return atlasEntity;
      }
  
  
      public static class HiveEventContext {
          private Set<ReadEntity> inputs;
          private Set<WriteEntity> outputs;
  
          private String user;
          private UserGroupInformation ugi;
          private HiveOperation operation;
          private HookContext.HookType hookType;
          private JSONObject jsonPlan;
          private String queryId;
          private String queryStr;
          private Long queryStartTime;
  
          public Map<String, List<ColumnLineageUtils.HiveColumnLineageInfo>> lineageInfo;
  
          private List<HookNotification> messages = new ArrayList<>();
  
          public void setInputs(Set<ReadEntity> inputs) {
              this.inputs = inputs;
          }
  
          public void setOutputs(Set<WriteEntity> outputs) {
              this.outputs = outputs;
          }
  
          public void setUser(String user) {
              this.user = user;
          }
  
          public void setUgi(UserGroupInformation ugi) {
              this.ugi = ugi;
          }
  
          public void setOperation(HiveOperation operation) {
              this.operation = operation;
          }
  
          public void setHookType(HookContext.HookType hookType) {
              this.hookType = hookType;
          }
  
          public void setQueryId(String queryId) {
              this.queryId = queryId;
          }
  
          public void setQueryStr(String queryStr) {
              this.queryStr = queryStr;
          }
  
          public void setQueryStartTime(Long queryStartTime) {
              this.queryStartTime = queryStartTime;
          }
  
          public void setLineageInfo(LineageInfo lineageInfo){
              try {
                  this.lineageInfo = ColumnLineageUtils.buildLineageMap(lineageInfo);
                  LOG.debug("Column Lineage Map => {} ", this.lineageInfo.entrySet());
              }catch (Throwable e){
                  LOG.warn("Column Lineage Map build failed with exception {}", e);
              }
          }
  
          public Set<ReadEntity> getInputs() {
              return inputs;
          }
  
          public Set<WriteEntity> getOutputs() {
              return outputs;
          }
  
          public String getUser() {
              return user;
          }
  
          public UserGroupInformation getUgi() {
              return ugi;
          }
  
          public HiveOperation getOperation() {
              return operation;
          }
  
          public HookContext.HookType getHookType() {
              return hookType;
          }
  
          public String getQueryId() {
              return queryId;
          }
  
          public String getQueryStr() {
              return queryStr;
          }
  
          public Long getQueryStartTime() {
              return queryStartTime;
          }
  
          public void addMessage(HookNotification message) {
              messages.add(message);
          }
  
          public List<HookNotification> getMessages() {
              return messages;
          }
      }
  
  
      @VisibleForTesting
      protected static String getProcessQualifiedName(HiveMetaStoreBridge dgiBridge, HiveEventContext eventContext,
                                            final SortedSet<ReadEntity> sortedHiveInputs,
                                            final SortedSet<WriteEntity> sortedHiveOutputs,
                                            SortedMap<ReadEntity, AtlasEntity> hiveInputsMap,
                                            SortedMap<WriteEntity, AtlasEntity> hiveOutputsMap) throws HiveException {
          HiveOperation op = eventContext.getOperation();
          if (isCreateOp(eventContext)) {
              Entity entity = getEntityByType(sortedHiveOutputs, Entity.Type.TABLE);
  
              if (entity != null) {
                  Table outTable = entity.getTable();
                  //refresh table
                  outTable = dgiBridge.getHiveClient().getTable(outTable.getDbName(), outTable.getTableName());
                  return HiveMetaStoreBridge.getTableProcessQualifiedName(dgiBridge.getMetadataNamespace(), outTable);
              }
          }
  
          StringBuilder buffer = new StringBuilder(op.getOperationName());
  
          boolean ignoreHDFSPathsinQFName = ignoreHDFSPathsinQFName(op, sortedHiveInputs, sortedHiveOutputs);
          if ( ignoreHDFSPathsinQFName && LOG.isDebugEnabled()) {
              LOG.debug("Ignoring HDFS paths in qualifiedName for {} {} ", op, eventContext.getQueryStr());
          }
  
          addInputs(dgiBridge, op, sortedHiveInputs, buffer, hiveInputsMap, ignoreHDFSPathsinQFName);
          buffer.append(IO_SEP);
          addOutputs(dgiBridge, op, sortedHiveOutputs, buffer, hiveOutputsMap, ignoreHDFSPathsinQFName);
          LOG.info("Setting process qualified name to {}", buffer);
          return buffer.toString();
      }
  
      protected static Entity getEntityByType(Set<? extends Entity> entities, Entity.Type entityType) {
          for (Entity entity : entities) {
              if (entity.getType() == entityType) {
                  return entity;
              }
          }
          return null;
      }
  
  
      protected static boolean ignoreHDFSPathsinQFName(final HiveOperation op, final Set<ReadEntity> inputs, final Set<WriteEntity> outputs) {
          switch (op) {
              case LOAD:
              case IMPORT:
                  return isPartitionBasedQuery(outputs);
              case EXPORT:
                  return isPartitionBasedQuery(inputs);
              case QUERY:
                  return true;
          }
          return false;
      }
  
      protected static boolean isPartitionBasedQuery(Set<? extends Entity> entities) {
          for (Entity entity : entities) {
              if (Entity.Type.PARTITION.equals(entity.getType())) {
                  return true;
              }
          }
          return false;
      }
  
      protected static boolean isCreateOp(HiveEventContext hiveEvent) {
          return HiveOperation.CREATETABLE.equals(hiveEvent.getOperation())
                  || HiveOperation.CREATEVIEW.equals(hiveEvent.getOperation())
                  || HiveOperation.ALTERVIEW_AS.equals(hiveEvent.getOperation())
                  || HiveOperation.ALTERTABLE_LOCATION.equals(hiveEvent.getOperation())
                  || HiveOperation.CREATETABLE_AS_SELECT.equals(hiveEvent.getOperation());
      }
  
      protected static void addInputs(HiveMetaStoreBridge hiveBridge, HiveOperation op, SortedSet<ReadEntity> sortedInputs, StringBuilder buffer, final Map<ReadEntity, AtlasEntity> refs, final boolean ignoreHDFSPathsInQFName) throws HiveException {
          if (refs != null) {
              if (sortedInputs != null) {
                  Set<String> dataSetsProcessed = new LinkedHashSet<>();
                  for (Entity input : sortedInputs) {
  
                      if (!dataSetsProcessed.contains(input.getName().toLowerCase())) {
                          //HiveOperation.QUERY type encompasses INSERT, INSERT_OVERWRITE, UPDATE, DELETE, PATH_WRITE operations
                          if (ignoreHDFSPathsInQFName &&
                                  (Entity.Type.DFS_DIR.equals(input.getType()) || Entity.Type.LOCAL_DIR.equals(input.getType()))) {
                              LOG.debug("Skipping dfs dir input addition to process qualified name {} ", input.getName());
                          } else if (refs.containsKey(input)) {
                              if ( input.getType() == Entity.Type.PARTITION || input.getType() == Entity.Type.TABLE) {
                                  Table inputTable = refreshTable(hiveBridge, input.getTable().getDbName(), input.getTable().getTableName());
  
                                  if (inputTable != null) {
                                      addDataset(buffer, refs.get(input), HiveMetaStoreBridge.getTableCreatedTime(inputTable));
                                  }
                              } else {
                                  addDataset(buffer, refs.get(input));
                              }
                          }
  
                          dataSetsProcessed.add(input.getName().toLowerCase());
                      }
                  }
  
              }
          }
      }
  
      protected static void addDataset(StringBuilder buffer, AtlasEntity ref, final long createTime) {
          addDataset(buffer, ref);
          buffer.append(SEP);
          buffer.append(createTime);
      }
  
      protected static void addDataset(StringBuilder buffer, AtlasEntity ref) {
          buffer.append(SEP);
          String dataSetQlfdName = (String) ref.getAttribute(AtlasClient.REFERENCEABLE_ATTRIBUTE_NAME);
          // '/' breaks query parsing on ATLAS
          buffer.append(dataSetQlfdName.toLowerCase().replaceAll("/", ""));
      }
  
      protected static void addOutputs(HiveMetaStoreBridge hiveBridge, HiveOperation op, SortedSet<WriteEntity> sortedOutputs, StringBuilder buffer, final Map<WriteEntity, AtlasEntity> refs, final boolean ignoreHDFSPathsInQFName) throws HiveException {
          if (refs != null) {
              Set<String> dataSetsProcessed = new LinkedHashSet<>();
              if (sortedOutputs != null) {
                  for (WriteEntity output : sortedOutputs) {
                      final Entity entity = output;
                      if (!dataSetsProcessed.contains(output.getName().toLowerCase())) {
                          if (ignoreHDFSPathsInQFName &&
                                  (Entity.Type.DFS_DIR.equals(output.getType()) || Entity.Type.LOCAL_DIR.equals(output.getType()))) {
                              LOG.debug("Skipping dfs dir output addition to process qualified name {} ", output.getName());
                          } else if (refs.containsKey(output)) {
                              //HiveOperation.QUERY type encompasses INSERT, INSERT_OVERWRITE, UPDATE, DELETE, PATH_WRITE operations
                              if (addQueryType(op, (WriteEntity) entity)) {
                                  buffer.append(SEP);
                                  buffer.append(((WriteEntity) entity).getWriteType().name());
                              }
  
                              if ( output.getType() == Entity.Type.PARTITION || output.getType() == Entity.Type.TABLE) {
                                  Table outputTable = refreshTable(hiveBridge, output.getTable().getDbName(), output.getTable().getTableName());
  
                                  if (outputTable != null) {
                                      addDataset(buffer, refs.get(output), HiveMetaStoreBridge.getTableCreatedTime(outputTable));
                                  }
                              } else {
                                  addDataset(buffer, refs.get(output));
                              }
                          }
  
                          dataSetsProcessed.add(output.getName().toLowerCase());
                      }
                  }
              }
          }
      }
  
      protected static Table refreshTable(HiveMetaStoreBridge dgiBridge, String dbName, String tableName) {
          try {
              return dgiBridge.getHiveClient().getTable(dbName, tableName);
          } catch (HiveException excp) { // this might be the case for temp tables
              LOG.warn("failed to get details for table {}.{}. Ignoring. {}: {}", dbName, tableName, excp.getClass().getCanonicalName(), excp.getMessage());
          }
  
          return null;
      }
  
      protected static boolean addQueryType(HiveOperation op, WriteEntity entity) {
          if (entity.getWriteType() != null && HiveOperation.QUERY.equals(op)) {
              switch (entity.getWriteType()) {
                  case INSERT:
                  case INSERT_OVERWRITE:
                  case UPDATE:
                  case DELETE:
                      return true;
                  case PATH_WRITE:
                      //Add query type only for DFS paths and ignore local paths since they are not added as outputs
                      if ( !Entity.Type.LOCAL_DIR.equals(entity.getType())) {
                          return true;
                      }
                      break;
                  default:
              }
          }
          return false;
      }
  
  
      @VisibleForTesting
      protected static final class EntityComparator implements Comparator<Entity> {
          @Override
          public int compare(Entity o1, Entity o2) {
              String s1 = o1.getName();
              String s2 = o2.getName();
              if (s1 == null || s2 == null){
                  s1 = o1.getD().toString();
                  s2 = o2.getD().toString();
              }
              return s1.toLowerCase().compareTo(s2.toLowerCase());
          }
      }
  
      @VisibleForTesting
      protected static final Comparator<Entity> entityComparator = new EntityComparator();
  
      protected AtlasObjectId toAtlasObjectId(Object obj) {
          final AtlasObjectId ret;
  
          if (obj instanceof AtlasObjectId) {
              ret = (AtlasObjectId) obj;
          } else if (obj instanceof Map) {
              ret = new AtlasObjectId((Map) obj);
          } else if (obj != null) {
              ret = new AtlasObjectId(obj.toString()); // guid
          } else {
              ret = null;
          }
  
          return ret;
      }
  
      protected List<AtlasObjectId> toAtlasObjectIdList(Object obj) {
          final List<AtlasObjectId> ret;
  
          if (obj instanceof Collection) {
              Collection coll = (Collection) obj;
  
              ret = new ArrayList<>(coll.size());
  
              for (Object item : coll) {
                  AtlasObjectId objId = toAtlasObjectId(item);
  
                  if (objId != null) {
                      ret.add(objId);
                  }
              }
          } else {
              AtlasObjectId objId = toAtlasObjectId(obj);
  
              if (objId != null) {
                  ret = new ArrayList<>(1);
  
                  ret.add(objId);
              } else {
                  ret = null;
              }
          }
  
          return ret;
      }
  
      protected AtlasStruct toAtlasStruct(Object obj) {
          final AtlasStruct ret;
  
          if (obj instanceof AtlasStruct) {
              ret = (AtlasStruct) obj;
          } else if (obj instanceof Map) {
              ret = new AtlasStruct((Map) obj);
          } else {
              ret = null;
          }
  
          return ret;
      }
  
      protected List<AtlasStruct> toAtlasStructList(Object obj) {
          final List<AtlasStruct> ret;
  
          if (obj instanceof Collection) {
              Collection coll = (Collection) obj;
  
              ret = new ArrayList<>(coll.size());
  
              for (Object item : coll) {
                  AtlasStruct struct = toAtlasStruct(item);
  
                  if (struct != null) {
                      ret.add(struct);
                  }
              }
          } else {
              AtlasStruct struct = toAtlasStruct(obj);
  
              if (struct != null) {
                  ret = new ArrayList<>(1);
  
                  ret.add(struct);
              } else {
                  ret = null;
              }
          }
  
          return ret;
      }
  }
  