public StatData doQueryAuditData(DataSource dataSource, String startTime, String endTime,
            String inlongGroupId, String inlongStreamId, String auditId,
            String auditTag, boolean distinct) {
        List<StatData> result = new LinkedList<>();
        String querySQL = distinct
                ? Configuration.getInstance().get(KEY_RECONCILIATION_DISTINCT_SQL, DEFAULT_RECONCILIATION_DISTINCT_SQL)
                : Configuration.getInstance().get(KEY_RECONCILIATION_SQL, DEFAULT_RECONCILIATION_SQL);
        List<String> paramList = new ArrayList<>();
        if (WILDCARD_STREAM_ID.equals(inlongStreamId)) {
            querySQL = AuditUtils.removeStreamIdCondition(querySQL);
            paramList.add(startTime);
            paramList.add(endTime);
            paramList.add(auditId);
            paramList.add(inlongGroupId);
        } else {
            paramList.add(startTime);
            paramList.add(endTime);
            paramList.add(auditId);
            paramList.add(inlongGroupId);
            paramList.add(inlongStreamId);
        }
        paramList.add(auditTag);

        try (Connection connection = dataSource.getConnection();
                PreparedStatement pstat = connection.prepareStatement(querySQL)) {
            for (int i = 0; i < paramList.size(); i++) {
                pstat.setString(i + 1, paramList.get(i));
            }

            try (ResultSet resultSet = pstat.executeQuery()) {
                while (resultSet.next()) {
                    StatData data = new StatData();
                    data.setAuditVersion(resultSet.getLong(COLUMN_AUDIT_VERSION));
                    data.setCount(resultSet.getLong(COLUMN_CNT));
                    data.setLogTs(startTime);
                    data.setInlongGroupId(inlongGroupId);
                    data.setInlongStreamId(inlongStreamId);
                    data.setAuditId(auditId);
                    data.setAuditTag(auditTag);
                    result.add(data);
                }
            } catch (SQLException sqlException) {
                LOGGER.error("Query ips has SQL exception!, datasource={} ", dataSource, sqlException);
            }
        } catch (Exception exception) {
            LOGGER.error("Query audit data has exception! ", exception);
        }
        return AuditUtils.getMaxAuditVersionAuditData(result);
    }
Do query report ips.

@param dataSource
@param startTime
@param endTime
@param inlongGroupId
@param inlongStreamId
@param auditId
@return
inlong-audit/audit-service/src/main/java/org/apache/inlong/audit/service/cache/RealTimeQuery.java