private void fetchMoreEntries() throws SQLException, AuditLogEntryException {
    LongRange idsToRead = getIdsToRead();
    // No more entries to read
    if (idsToRead.getMaximumLong() == 0) {
      return;
    }
    // TODO: Remove left outer join and command type filter once the
    // exchange partition bug is fixed in HIVE-12215
    String queryFormatString = "SELECT a.id, a.create_time, "
        + "command_type, command, name, category, "
        + "type, serialized_object "
        + "FROM %s a LEFT OUTER JOIN %s b on a.id = b.audit_log_id "
        + "WHERE a.id >= ? AND a.id <= ? "
        + "AND (command_type IS NULL OR command_type "
        + "NOT IN('SHOWTABLES', 'SHOWPARTITIONS', 'SWITCHDATABASE')) "
        + "ORDER BY id "
        // Get read locks on the specified rows to prevent skipping of rows that haven't committed
        // yet, but have an ID between idsToRead. For example, one transaction starts and
        // inserts id = 1, but another transaction starts, inserts, and commits i=2 before the
        // first transaction commits. Locking can also be done with serializable isolation level.
        + "LOCK IN SHARE MODE";
    String query = String.format(queryFormatString,
        auditLogTableName, outputObjectsTableName,
        idsToRead.getMinimumLong(), idsToRead.getMaximumLong());
    Connection connection = dbConnectionFactory.getConnection();
    PreparedStatement ps = connection.prepareStatement(query);
    int index = 1;
    ps.setLong(index++, idsToRead.getMinimumLong());
    ps.setLong(index++, idsToRead.getMaximumLong());
    ResultSet rs = ps.executeQuery();
    long id = -1;
    Timestamp createTime = null;
    HiveOperation commandType = null;
    String command = null;
    String objectName;
    String objectCategory;
    String objectType;
    String objectSerialized;
    long previouslyReadId = -1;
    Timestamp previouslyReadTs = null;
    HiveOperation previousCommandType = null;
    String previousCommand = null;
    // For a given audit log ID, the join would have produced multiple rows
    // for each ID. Each row contains a single output. Group all the rows
    // and the outputs into a AuditLogEntry.
    // For a given audit log ID, these accumulate the outputs from the
    // different rows.
    List<String> outputDirectories = new LinkedList<>();
    List<Table> outputTables = new LinkedList<>();
    List<NamedPartition> outputPartitions = new LinkedList<>();
    List<Table> referenceTables = new LinkedList<>();
    Table inputTable = null;
    NamedPartition renameFromPartition = null;
    while (rs.next()) {
      id = rs.getLong("id");
      createTime = rs.getTimestamp("create_time");
      // Invalid operations are returned as null
      String commandTypeString = rs.getString("command_type");
      commandType = convertToHiveOperation(commandTypeString);
      if (commandType == null) {
        LOG.debug(String.format("Invalid operation %s in audit log id: %s", commandTypeString, id));
      }
      command = rs.getString("command");
      objectName = rs.getString("name");
      objectCategory = rs.getString("category");
      objectType = rs.getString("type");
      objectSerialized = rs.getString("serialized_object");
      if (previouslyReadId != -1 && id != previouslyReadId) {
        lastReadId = previouslyReadId;
        // This means that all the outputs for a given audit log entry
        // has been read.
        AuditLogEntry entry = new AuditLogEntry(
            previouslyReadId,
            previouslyReadTs,
            previousCommandType,
            previousCommand,
            outputDirectories,
            referenceTables,
            outputTables,
            outputPartitions,
            inputTable,
            renameFromPartition);
        auditLogEntries.add(entry);
        // Reset these accumulated values
        outputDirectories = new LinkedList<>();
        referenceTables = new LinkedList<>();
        outputTables = new LinkedList<>();
        outputPartitions = new LinkedList<>();
        renameFromPartition = null;
        inputTable = null;
      }
      previouslyReadId = id;
      previouslyReadTs = createTime;
      previousCommandType = commandType;
      previousCommand = command;
      if ("DIRECTORY".equals(objectType)) {
        outputDirectories.add(objectName);
      } else if ("TABLE".equals(objectType)) {
        Table table = new Table();
        try {
          ReplicationUtils.deserializeObject(objectSerialized, table);
        } catch (MetadataException e) {
          throw new AuditLogEntryException(e);
        }
        ReplicationUtils.normalizeNames(table);
        if ("OUTPUT".equals(objectCategory)) {
          outputTables.add(table);
        } else if ("REFERENCE_TABLE".equals(objectCategory)) {
          referenceTables.add(table);
        } else if ("RENAME_FROM".equals(objectCategory) || "INPUT".equals(objectCategory)) {
          inputTable = table;
        } else {
          throw new RuntimeException("Unhandled category: " + objectCategory);
        }
      } else if ("PARTITION".equals(objectType)) {
        Partition partition = new Partition();
        try {
          ReplicationUtils.deserializeObject(objectSerialized, partition);
        } catch (MetadataException e) {
          throw new AuditLogEntryException(e);
        }
        ReplicationUtils.normalizeNames(partition);
        String partitionName = getPartitionNameFromOutputCol(objectName);
        NamedPartition namedPartition = new NamedPartition(partitionName, partition);
        if ("OUTPUT".equals(objectCategory)) {
          outputPartitions.add(namedPartition);
        } else if ("RENAME_FROM".equals(objectCategory) || "INPUT".equals(objectCategory)) {
          renameFromPartition = namedPartition;
        } else {
          throw new RuntimeException("Unhandled category: " + objectCategory);
        }
      } else if ("DFS_DIR".equals(objectType)) {
        outputDirectories.add(objectName);
      } else if ("LOCAL_DIR".equals(objectType)) {
        outputDirectories.add(objectName);
      } else if ("DATABASE".equals(objectType)) {
        // Currently, nothing is done with DB's
      } else if (objectType == null) {
        // This will happen for queries that don't have any output
        // objects. This can be removed a long with the OUTER aspect
        // of the join above once the bug with exchange partitions is
        // fixed.
        LOG.debug("No output objects");
      } else {
        throw new RuntimeException("Unhandled output type: " + objectType);
      }
    }
    // This is the case where we read to the end of the table.
    if (id != -1) {
      AuditLogEntry entry = new AuditLogEntry(
          id,
          createTime,
          commandType,
          command,
          outputDirectories,
          referenceTables,
          outputTables,
          outputPartitions,
          inputTable,
          renameFromPartition);
      auditLogEntries.add(entry);
    }
    // Note: if we constantly get empty results (i.e. no valid entries
    // because all the commands got filtered out), then the lastReadId won't
    // be updated for a while.
    lastReadId = idsToRead.getMaximumLong();
    return;
  }
private void fetchMoreEntries() throws SQLException, AuditLogEntryException {
    LongRange idsToRead = getIdsToRead();
   
    if (idsToRead.getMaximumLong() == 0) {
      return;
    }
   
   
    String queryFormatString = "SELECT a.id, a.create_time, "
        + "command_type, command, name, category, "
        + "type, serialized_object "
        + "FROM %s a LEFT OUTER JOIN %s b on a.id = b.audit_log_id "
        + "WHERE a.id >= ? AND a.id <= ? "
        + "AND (command_type IS NULL OR command_type "
        + "NOT IN('SHOWTABLES', 'SHOWPARTITIONS', 'SWITCHDATABASE')) "
        + "ORDER BY id "
       
       
       
       
        + "LOCK IN SHARE MODE";
    String query = String.format(queryFormatString,
        auditLogTableName, outputObjectsTableName,
        idsToRead.getMinimumLong(), idsToRead.getMaximumLong());
    Connection connection = dbConnectionFactory.getConnection();
    PreparedStatement ps = connection.prepareStatement(query);
    int index = 1;
    ps.setLong(index++, idsToRead.getMinimumLong());
    ps.setLong(index++, idsToRead.getMaximumLong());
    ResultSet rs = ps.executeQuery();
    long id = -1;
    Timestamp createTime = null;
    HiveOperation commandType = null;
    String command = null;
    String objectName;
    String objectCategory;
    String objectType;
    String objectSerialized;
    long previouslyReadId = -1;
    Timestamp previouslyReadTs = null;
    HiveOperation previousCommandType = null;
    String previousCommand = null;
   
   
   
   
   
    List<String> outputDirectories = new LinkedList<>();
    List<Table> outputTables = new LinkedList<>();
    List<NamedPartition> outputPartitions = new LinkedList<>();
    List<Table> referenceTables = new LinkedList<>();
    Table inputTable = null;
    NamedPartition renameFromPartition = null;
    while (rs.next()) {
      id = rs.getLong("id");
      createTime = rs.getTimestamp("create_time");
     
      String commandTypeString = rs.getString("command_type");
      commandType = convertToHiveOperation(commandTypeString);
      if (commandType == null) {
        LOG.debug(String.format("Invalid operation %s in audit log id: %s", commandTypeString, id));
      }
      command = rs.getString("command");
      objectName = rs.getString("name");
      objectCategory = rs.getString("category");
      objectType = rs.getString("type");
      objectSerialized = rs.getString("serialized_object");
      if (previouslyReadId != -1 && id != previouslyReadId) {
        lastReadId = previouslyReadId;
       
       
        AuditLogEntry entry = new AuditLogEntry(
            previouslyReadId,
            previouslyReadTs,
            previousCommandType,
            previousCommand,
            outputDirectories,
            referenceTables,
            outputTables,
            outputPartitions,
            inputTable,
            renameFromPartition);
        auditLogEntries.add(entry);
       
        outputDirectories = new LinkedList<>();
        referenceTables = new LinkedList<>();
        outputTables = new LinkedList<>();
        outputPartitions = new LinkedList<>();
        renameFromPartition = null;
        inputTable = null;
      }
      previouslyReadId = id;
      previouslyReadTs = createTime;
      previousCommandType = commandType;
      previousCommand = command;
      if ("DIRECTORY".equals(objectType)) {
        outputDirectories.add(objectName);
      } else if ("TABLE".equals(objectType)) {
        Table table = new Table();
        try {
          ReplicationUtils.deserializeObject(objectSerialized, table);
        } catch (MetadataException e) {
          throw new AuditLogEntryException(e);
        }
        ReplicationUtils.normalizeNames(table);
        if ("OUTPUT".equals(objectCategory)) {
          outputTables.add(table);
        } else if ("REFERENCE_TABLE".equals(objectCategory)) {
          referenceTables.add(table);
        } else if ("RENAME_FROM".equals(objectCategory) || "INPUT".equals(objectCategory)) {
          inputTable = table;
        } else {
          throw new RuntimeException("Unhandled category: " + objectCategory);
        }
      } else if ("PARTITION".equals(objectType)) {
        Partition partition = new Partition();
        try {
          ReplicationUtils.deserializeObject(objectSerialized, partition);
        } catch (MetadataException e) {
          throw new AuditLogEntryException(e);
        }
        ReplicationUtils.normalizeNames(partition);
        String partitionName = getPartitionNameFromOutputCol(objectName);
        NamedPartition namedPartition = new NamedPartition(partitionName, partition);
        if ("OUTPUT".equals(objectCategory)) {
          outputPartitions.add(namedPartition);
        } else if ("RENAME_FROM".equals(objectCategory) || "INPUT".equals(objectCategory)) {
          renameFromPartition = namedPartition;
        } else {
          throw new RuntimeException("Unhandled category: " + objectCategory);
        }
      } else if ("DFS_DIR".equals(objectType)) {
        outputDirectories.add(objectName);
      } else if ("LOCAL_DIR".equals(objectType)) {
        outputDirectories.add(objectName);
      } else if ("DATABASE".equals(objectType)) {
       
      } else if (objectType == null) {
       
       
       
       
        LOG.debug("No output objects");
      } else {
        throw new RuntimeException("Unhandled output type: " + objectType);
      }
    }
   
    if (id != -1) {
      AuditLogEntry entry = new AuditLogEntry(
          id,
          createTime,
          commandType,
          command,
          outputDirectories,
          referenceTables,
          outputTables,
          outputPartitions,
          inputTable,
          renameFromPartition);
      auditLogEntries.add(entry);
    }
   
   
   
    lastReadId = idsToRead.getMaximumLong();
    return;
  }
private void fetchmoreentries() throws sqlexception, auditlogentryexception { longrange idstoread = getidstoread(); if (idstoread.getmaximumlong() == 0) { return; } string queryformatstring = "select a.id, a.create_time, " + "command_type, command, name, category, " + "type, serialized_object " + "from %s a left outer join %s b on a.id = b.audit_log_id " + "where a.id >= ? and a.id <= ? " + "and (command_type is null or command_type " + "not in('showtables', 'showpartitions', 'switchdatabase')) " + "order by id " + "lock in share mode"; string query = string.format(queryformatstring, auditlogtablename, outputobjectstablename, idstoread.getminimumlong(), idstoread.getmaximumlong()); connection connection = dbconnectionfactory.getconnection(); preparedstatement ps = connection.preparestatement(query); int index = 1; ps.setlong(index++, idstoread.getminimumlong()); ps.setlong(index++, idstoread.getmaximumlong()); resultset rs = ps.executequery(); long id = -1; timestamp createtime = null; hiveoperation commandtype = null; string command = null; string objectname; string objectcategory; string objecttype; string objectserialized; long previouslyreadid = -1; timestamp previouslyreadts = null; hiveoperation previouscommandtype = null; string previouscommand = null; list<string> outputdirectories = new linkedlist<>(); list<table> outputtables = new linkedlist<>(); list<namedpartition> outputpartitions = new linkedlist<>(); list<table> referencetables = new linkedlist<>(); table inputtable = null; namedpartition renamefrompartition = null; while (rs.next()) { id = rs.getlong("id"); createtime = rs.gettimestamp("create_time"); string commandtypestring = rs.getstring("command_type"); commandtype = converttohiveoperation(commandtypestring); if (commandtype == null) { log.debug(string.format("invalid operation %s in audit log id: %s", commandtypestring, id)); } command = rs.getstring("command"); objectname = rs.getstring("name"); objectcategory = rs.getstring("category"); objecttype = rs.getstring("type"); objectserialized = rs.getstring("serialized_object"); if (previouslyreadid != -1 && id != previouslyreadid) { lastreadid = previouslyreadid; auditlogentry entry = new auditlogentry( previouslyreadid, previouslyreadts, previouscommandtype, previouscommand, outputdirectories, referencetables, outputtables, outputpartitions, inputtable, renamefrompartition); auditlogentries.add(entry); outputdirectories = new linkedlist<>(); referencetables = new linkedlist<>(); outputtables = new linkedlist<>(); outputpartitions = new linkedlist<>(); renamefrompartition = null; inputtable = null; } previouslyreadid = id; previouslyreadts = createtime; previouscommandtype = commandtype; previouscommand = command; if ("directory".equals(objecttype)) { outputdirectories.add(objectname); } else if ("table".equals(objecttype)) { table table = new table(); try { replicationutils.deserializeobject(objectserialized, table); } catch (metadataexception e) { throw new auditlogentryexception(e); } replicationutils.normalizenames(table); if ("output".equals(objectcategory)) { outputtables.add(table); } else if ("reference_table".equals(objectcategory)) { referencetables.add(table); } else if ("rename_from".equals(objectcategory) || "input".equals(objectcategory)) { inputtable = table; } else { throw new runtimeexception("unhandled category: " + objectcategory); } } else if ("partition".equals(objecttype)) { partition partition = new partition(); try { replicationutils.deserializeobject(objectserialized, partition); } catch (metadataexception e) { throw new auditlogentryexception(e); } replicationutils.normalizenames(partition); string partitionname = getpartitionnamefromoutputcol(objectname); namedpartition namedpartition = new namedpartition(partitionname, partition); if ("output".equals(objectcategory)) { outputpartitions.add(namedpartition); } else if ("rename_from".equals(objectcategory) || "input".equals(objectcategory)) { renamefrompartition = namedpartition; } else { throw new runtimeexception("unhandled category: " + objectcategory); } } else if ("dfs_dir".equals(objecttype)) { outputdirectories.add(objectname); } else if ("local_dir".equals(objecttype)) { outputdirectories.add(objectname); } else if ("database".equals(objecttype)) { } else if (objecttype == null) { log.debug("no output objects"); } else { throw new runtimeexception("unhandled output type: " + objecttype); } } if (id != -1) { auditlogentry entry = new auditlogentry( id, createtime, commandtype, command, outputdirectories, referencetables, outputtables, outputpartitions, inputtable, renamefrompartition); auditlogentries.add(entry); } lastreadid = idstoread.getmaximumlong(); return; }