Update

Author: ZorTik

asql-core/src/main/java/me/zort/sqllib/internal/query/DeleteQuery.java

@@ -4,6 +4,7 @@
 import me.zort.sqllib.api.Executive;
 import me.zort.sqllib.SQLDatabaseConnection;
 import me.zort.sqllib.internal.query.part.LimitStatement;
+import me.zort.sqllib.internal.query.part.OffsetStatement;
 import org.jetbrains.annotations.Nullable;
 
 import java.util.ArrayList;
@@ -35,6 +36,12 @@ public DeleteQuery from(String table) {
     return this;
   }
 
+  @Override
+  public DeleteQuery offset(int offset) {
+    then(new OffsetStatement<>(this, new ArrayList<>(), offset));
+    return this;
+  }
+
   public DeleteQuery limit(int limit) {
     then(new LimitStatement<>(this, new ArrayList<>(), limit));
     return this;

asql-core/src/main/java/me/zort/sqllib/internal/query/Limitable.java

@@ -2,6 +2,7 @@
 
 public interface Limitable<P extends QueryNode<?> & Limitable<P>> { // P = self
 
+  P offset(int offset);
   P limit(int limit);
 
 }

asql-core/src/main/java/me/zort/sqllib/internal/query/SelectQuery.java

@@ -4,14 +4,16 @@
 import me.zort.sqllib.api.Executive;
 import me.zort.sqllib.SQLDatabaseConnection;
 import me.zort.sqllib.internal.query.part.LimitStatement;
+import me.zort.sqllib.internal.query.part.OffsetStatement;
 import org.jetbrains.annotations.Nullable;
 
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
 import java.util.Objects;
 
-public class SelectQuery extends QueryNode<QueryNode<?>> implements Executive, Conditional<SelectQuery>, Limitable<SelectQuery>, ResultSetAware {
+public class SelectQuery extends QueryNode<QueryNode<?>> implements
+        Executive, Conditional<SelectQuery>, Limitable<SelectQuery>, ResultSetAware {
 
   private final List<String> cols;
   private String table;
@@ -39,6 +41,12 @@ public SelectQuery from(String table) {
     return this;
   }
 
+  @Override
+  public SelectQuery offset(int offset) {
+    then(new OffsetStatement<>(this, new ArrayList<>(), offset));
+    return this;
+  }
+
   public SelectQuery limit(int limit) {
     then(new LimitStatement<>(this, new ArrayList<>(), limit));
     return this;

asql-core/src/main/java/me/zort/sqllib/internal/query/part/OffsetStatement.java

@@ -0,0 +1,30 @@
+package me.zort.sqllib.internal.query.part;
+
+import me.zort.sqllib.internal.query.QueryDetails;
+import me.zort.sqllib.internal.query.QueryNode;
+import me.zort.sqllib.internal.query.ResultSetAware;
+import org.jetbrains.annotations.Nullable;
+
+import java.util.HashMap;
+import java.util.List;
+
+public class OffsetStatement<P extends QueryNode<?>> extends QueryNode<P> implements ResultSetAware {
+
+  private final int offset;
+
+  public OffsetStatement(@Nullable P parent, List<QueryNode<?>> initial, int offset) {
+    super(parent, initial, Integer.MAX_VALUE);
+    this.offset = offset;
+  }
+
+  @Override
+  public QueryDetails buildQueryDetails() {
+    return new QueryDetails(" OFFSET " + Math.max(offset, 0), new HashMap<>());
+  }
+
+  @SuppressWarnings("unchecked")
+  @Override
+  public OffsetStatement<P> then(String part) {
+    return (OffsetStatement<P>) super.then(part);
+  }
+}

asql-debezium/.gitignore

@@ -4,6 +4,8 @@ build/
 !**/src/main/**/build/
 !**/src/test/**/build/
 
+main/java/me/zort/sqllib/debezium/Test.java
+
 ### IntelliJ IDEA ###
 .idea/modules.xml
 .idea/jarRepositories.xml

asql-debezium/build.gradle

@@ -17,6 +17,14 @@ dependencies {
     implementation 'io.debezium:debezium-connector-mysql:1.9.6.Final'
     implementation 'com.google.guava:guava:31.0-jre'
     implementation project(':api')
+    implementation project(':shared')
     implementation project(':core')
     annotationProcessor 'org.projectlombok:lombok:1.18.24'
+}
+
+task runWithJavaExec(type: JavaExec) {
+    group = "Execution"
+    description = "Run the main class with JavaExecTask"
+    classpath = sourceSets.main.runtimeClasspath
+    main = "me.zort.sqllib.debezium.Test"
 }

asql-debezium/settings.gradle

@@ -1,5 +1,7 @@
 rootProject.name = 'asql-debezium'
 include ":api"
 include ":core"
+include ":shared"
 project(":api").projectDir = file("../asql-api")
-project(":core").projectDir = file("../asql-core")
+project(":core").projectDir = file("../asql-core")
+project(":shared").projectDir = file("../asql-shared")

…sqllib/debezium/ASQLDebeziumService.java …sqllib/debezium/ASQLDebeziumWatcher.javaasql-debezium/src/main/java/me/zort/sqllib/debezium/ASQLDebeziumService.java renamed to asql-debezium/src/main/java/me/zort/sqllib/debezium/ASQLDebeziumWatcher.java asql-debezium/src/main/java/me/zort/sqllib/debezium/ASQLDebeziumService.java renamed to asql-debezium/src/main/java/me/zort/sqllib/debezium/ASQLDebeziumWatcher.java

@@ -11,15 +11,17 @@
 import me.zort.sqllib.debezium.builder.EntityFilterBuilder;
 import org.jetbrains.annotations.NotNull;
 
+import java.io.IOException;
 import java.lang.reflect.AnnotatedElement;
 import java.net.URI;
-import java.sql.Connection;
+import java.net.URISyntaxException;
+import java.sql.DatabaseMetaData;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
 import java.util.function.Consumer;
 import java.util.function.Function;
 
@@ -40,31 +42,46 @@
  * @author ZorTik
  */
 @Beta
-public final class ASQLDebeziumService implements
-        DebeziumEngine.ChangeConsumer<ChangeEvent<String, String>>, Runnable {
+public final class ASQLDebeziumWatcher
+        implements DebeziumEngine.ChangeConsumer<ChangeEvent<String, String>> {
 
   @SneakyThrows
-  public static @NotNull Builder configure(@NotNull SQLDatabaseConnection connection) {
+  public static @NotNull Builder configure(
+          @NotNull SQLDatabaseConnection connection, String password
+  ) {
     if (!(connection instanceof SQLDatabaseConnectionImpl)) {
       throw new IllegalArgumentException("Connection does not contain options!");
     } else if (!connection.isConnected()) {
       throw new IllegalArgumentException("Connection is not connected!");
     }
-    Connection rawConnection = connection.getConnection();
-    URI uri = new URI(rawConnection.getMetaData().getURL());
+    DatabaseMetaData rawConnectionMeta = connection.getConnection().getMetaData();
+    URI uri = new URI(rawConnectionMeta.getURL());
+    return configure(
+            uri.getHost(), uri.getPort(), rawConnectionMeta.getUserName(), password
+    );
+  }
+
+  public static @NotNull Builder configure(
+          @NotNull String hostname,
+          int port,
+          @NotNull String username,
+          @NotNull String password
+  ) throws URISyntaxException {
     Configuration.Builder configBuilder = Configuration.create()
-            .with("database.hostname", uri.getHost())
-            .with("database.port", uri.getPort());
-    // TODO: Build configuration builder from raw connection details
+            .with("database.hostname", hostname)
+            .with("database.port", String.valueOf(port))
+            .with("database.user", username)
+            .with("database.password", password);
     return new Builder(configBuilder);
   }
 
   private final DebeziumEngine<ChangeEvent<String, String>> engine;
   private final Map<RecordFilter, Consumer<ChangeEvent<String, String>>> handlers;
+  private boolean running = false;
 
-  private ASQLDebeziumService(DebeziumEngine.Builder<ChangeEvent<String, String>> builder) {
+  private ASQLDebeziumWatcher(DebeziumEngine.Builder<ChangeEvent<String, String>> builder) {
     this.engine = builder.notifying(this).build();
-    this.handlers = null;
+    this.handlers = new ConcurrentHashMap<>();
   }
 
   @Override
@@ -85,9 +102,26 @@ public void handleBatch(
     committer.markBatchFinished();
   }
 
-  @Override
-  public void run() {
-    engine.run();
+  public void start(ExecutorService executor) {
+    if (running) {
+      throw new IllegalStateException("Service is already running!");
+    }
+    executor.submit(engine);
+    running = true;
+  }
+
+  public void stop() {
+    try {
+      engine.close();
+    } catch (IOException e) {
+      throw new RuntimeException(e);
+    } finally {
+      running = false;
+    }
+  }
+
+  public @NotNull CompletableFuture<ChangeEvent<String, String>> awaitChange() {
+    return awaitChange(RecordFilter.any());
   }
 
   /**
@@ -116,7 +150,13 @@ public static class Builder {
     private Builder(Configuration.Builder initialConfig) {
       this.config = initialConfig;
       edit(builder -> builder
-              .with("name", "Asql-Debezium-" + (++serviceCount)));
+              .with("name", "Asql-Debezium-" + (++serviceCount))
+              .with("offset.storage", "org.apache.kafka.connect.storage.FileOffsetBackingStore")
+              .with("offset.storage.file.filename", System.getProperty("user.dir") + "/offsets.dat")
+              .with("server.id", serviceCount)
+              .with("database.history", "io.debezium.relational.history.FileDatabaseHistory")
+              .with("io.debezium.relational.history.FileDatabaseHistory", System.getProperty("user.dir") + "/dbhistory.dat")
+              .with("offset.flush.interval.ms", 1000));
     }
 
     public @NotNull Builder edit(
@@ -130,10 +170,19 @@ private Builder(Configuration.Builder initialConfig) {
       return edit(builder -> builder.with("connector.class", type.getClassName()));
     }
 
-    public @NotNull ASQLDebeziumService build() {
+    public @NotNull ASQLDebeziumWatcher build() {
+      Configuration configuration = config.build();
+      assertProperty(configuration, "connector.class");
+      assertProperty(configuration, "database.hostname");
       DebeziumEngine.Builder<ChangeEvent<String, String>> builder = DebeziumEngine.create(Json.class)
-              .using(config.build().asProperties());
-      return new ASQLDebeziumService(builder);
+              .using(configuration.asProperties());
+      return new ASQLDebeziumWatcher(builder);
+    }
+
+    private static void assertProperty(Configuration config, String name) {
+      if (!config.hasKey(name)) {
+        throw new IllegalArgumentException("Configuration requires property " + name);
+      }
     }
   }
 

asql-debezium/src/main/java/me/zort/sqllib/debezium/RecordFilter.java

@@ -1,6 +1,8 @@
 package me.zort.sqllib.debezium;
 
+import io.debezium.data.Envelope;
 import io.debezium.engine.ChangeEvent;
+import me.zort.sqllib.debezium.filter.EnvelopeOpRecordFilter;
 import me.zort.sqllib.mapping.annotation.Table;
 import me.zort.sqllib.util.ParameterPair;
 import org.jetbrains.annotations.NotNull;
@@ -10,11 +12,11 @@
 
 /**
  * Tests a record if the provided record should be notified
- * to the registered consumer in {@link ASQLDebeziumService}.
+ * to the registered consumer in {@link ASQLDebeziumWatcher}.
  *
  * @author ZorTik
  */
-public final class RecordFilter {
+public class RecordFilter {
 
   private static final long DEFAULT_EXPIRATION = 10000L;
 
@@ -31,39 +33,46 @@ public boolean test(ChangeEvent<String, String> record) {
     return testFunction.test(record);
   }
 
-  public void setExpireAfter(long expireAfter) {
+  public final void setExpireAfter(long expireAfter) {
     if (isRegistered()) {
       throw new IllegalStateException("Filter expiration cannot be changed after it has been registered!");
     }
     this.expireAfter = expireAfter;
   }
 
-  public boolean isRegistered() {
+  public final boolean isRegistered() {
     return expireAt != -1;
   }
 
-  public boolean expired() {
+  public final boolean expired() {
     return System.currentTimeMillis() >= expireAt;
   }
 
-  void markRegistered() {
+  final void markRegistered() {
     if (isRegistered()) {
       throw new IllegalStateException("Filter has already been registered!");
     }
     expireAt = System.currentTimeMillis() + expireAfter;
   }
 
+  public static @NotNull RecordFilter any() {
+    return new RecordFilter(record -> true);
+  }
+
   public static @NotNull RecordFilter table(AnnotatedElement element, ParameterPair... parameters) {
     final String table = Table.Util.getFromContext(element, parameters);
     return new RecordFilter(record -> record.destination().equals(table));
   }
 
-  public static @NotNull RecordFilter column(String column) {
-    return new RecordFilter(record -> record.key().equals(column));
+  public static @NotNull RecordFilter columnChanged(String column) {
+    return new EnvelopeOpRecordFilter(record -> record.key().equals(column), Envelope.Operation.UPDATE);
   }
 
-  public static @NotNull RecordFilter column(String column, String value) {
-    return new RecordFilter(record -> record.key().equals(column) && record.value().equals(value));
+  public static @NotNull RecordFilter columnChanged(String column, String value) {
+    return new EnvelopeOpRecordFilter(
+            record -> record.key().equals(column) && record.value().equals(value),
+            Envelope.Operation.UPDATE
+    );
   }
 
   public static @NotNull RecordFilter join(RecordFilter... filters) {

asql-debezium/src/main/java/me/zort/sqllib/debezium/RecordFilterBuilder.java

@@ -10,18 +10,18 @@
 
 /**
  * Provides easy to use API for building and executing filters
- * in {@link ASQLDebeziumService}.
+ * in {@link ASQLDebeziumWatcher}.
  *
  * @see RecordFilter
- * @see ASQLDebeziumService
+ * @see ASQLDebeziumWatcher
  * @author ZorTik
  */
 public class RecordFilterBuilder {
 
-  private final ASQLDebeziumService service;
+  private final ASQLDebeziumWatcher service;
   private RecordFilter filter;
 
-  public RecordFilterBuilder(ASQLDebeziumService service) {
+  public RecordFilterBuilder(ASQLDebeziumWatcher service) {
     this.service = service;
     this.filter = null;
   }