public void query(String startTime, String endTime, String auditId) {
            try (Connection connection = dataSource.getConnection();
                    PreparedStatement pstat = connection.prepareStatement(sourceConfig.getQuerySql())) {
                if (connection.isClosed()) {
                    createDataSource();
                }
                pstat.setString(1, startTime);
                pstat.setString(2, endTime);
                pstat.setString(3, auditId);
                if (sourceConfig.isNeedJoin()) {
                    pstat.setString(4, startTime);
                    pstat.setString(5, endTime);
                    pstat.setString(6, auditId);
                }
                try (ResultSet resultSet = pstat.executeQuery()) {
                    while (resultSet.next()) {
                        StatData data = new StatData();
                        data.setLogTs(startTime);
                        data.setInlongGroupId(resultSet.getString(1));
                        data.setInlongStreamId(resultSet.getString(2));
                        data.setAuditId(resultSet.getString(3));
                        String auditTag = resultSet.getString(4);
                        data.setAuditTag(StringUtils.isBlank(auditTag) ? DEFAULT_AUDIT_TAG : auditTag);
                        long count = resultSet.getLong(5);
                        data.setCount(count);
                        data.setSize(resultSet.getLong(6));
                        data.setDelay(CacheUtils.calculateAverageDelay(count, resultSet.getLong(7)));
                        dataQueue.push(data);
                    }
                } catch (SQLException sqlException) {
                    LOGGER.error("Query has SQL exception! ", sqlException);
                }
            } catch (Exception exception) {
                LOGGER.error("Query has exception! ", exception);
            }
        }
Query

@param startTime
@param endTime
@param auditId
inlong-audit/audit-service/src/main/java/org/apache/inlong/audit/service/source/JdbcSource.java