@Test
public void testBuildQuery() {
    final SignalBasedIncrementalSnapshotChangeEventSource<TableId> source = new SignalBasedIncrementalSnapshotChangeEventSource<>(
            config(), new JdbcConnection(config().getConfig(), config -> null), null, null, SnapshotProgressListener.NO_OP, DataChangeEventListener.NO_OP);
    final IncrementalSnapshotContext<TableId> context = new IncrementalSnapshotContext<>();
    source.setContext(context);
    final Column pk1 = Column.editor().name("pk1").create();
    final Column pk2 = Column.editor().name("pk2").create();
    final Column val1 = Column.editor().name("val1").create();
    final Column val2 = Column.editor().name("val2").create();
    final Table table = Table.editor().tableId(new TableId(null, "s1", "table1")).addColumn(pk1).addColumn(pk2)
            .addColumn(val1).addColumn(val2).setPrimaryKeyNames("pk1", "pk2").create();
    Assertions.assertThat(source.buildChunkQuery(table)).isEqualTo("SELECT * FROM \"s1\".\"table1\" ORDER BY pk1, pk2 LIMIT 1024");
    context.nextChunkPosition(new Object[]{ 1, 5 });
    context.maximumKey(new Object[]{ 10, 50 });
    Assertions.assertThat(source.buildChunkQuery(table)).isEqualTo(
            "SELECT * FROM \"s1\".\"table1\" WHERE pk1 >= ? AND pk2 >= ? AND NOT (pk1 = ? AND pk2 = ?) AND pk1 <= ? AND pk2 <= ? ORDER BY pk1, pk2 LIMIT 1024");
}

pk1 pk2
1   101
2   105
3   110
4   1    
5   2
6   3