Как добавить только новые строки SQL в Elasticsearch — Logstash — Kibana?

Я пытаюсь добавить некоторые данные из SQL-запроса в свой Elasticsearch, и в настоящее время он работает как вход Logstash jdbc следующим образом:

 jdbc {
        type => "database_log"
        jdbc_connection_string => "jdbc:sqlserver://(redacted)"
        jdbc_user => (redacted)
        jdbc_password => (redacted)
        jdbc_driver_library => "D:\ELK_56\sqljdbc4.jar"
        jdbc_driver_class => "com.microsoft.sqlserver.jdbc.SQLServerDriver"
        schedule => "0 * * * *"
        statement => "select user_id,First_Name from user_login"

Теперь проблема в том, что, как и каждый час, когда выполняется ввод, он просто добавляет весь результат запроса, около 124 тыс. строк. Мое намерение состоит в том, чтобы просто добавить новые строки (которых около 3-4 в час), которые появились с момента последнего запроса, и избежать дублирования.

Я не могу просто решить эту проблему, добавив в запрос предложение WHERE, поскольку в этой конкретной таблице нет столбца даты или какого-либо такого идентификатора.

Есть ли способ заставить сам logstash или elasticsearch вычесть предыдущий результат из последнего и просто добавить новые строки?

Спасибо за ваш ответ.