Managing schema automatically

Author: BinaryIgor

README.md

@@ -12,7 +12,7 @@ For scalability details, see [benchmarks](/benchmarks/README.md).
 
 ## How it works
 
-We just need to have a few tables (postgres syntax, schema managed by EventSQL):
+We just need to have a few tables (postgres syntax, schema managed fully by EventSQL):
 
 ```sql
 CREATE TABLE topic (
@@ -33,17 +33,15 @@ CREATE TABLE consumer (
   PRIMARY KEY (topic, name, partition)
 );
 
-CREATE TABLE event (
-  topic TEXT NOT NULL,
-  id BIGSERIAL NOT NULL,
+CREATE TABLE {topic}_event (
+  id BIGSERIAL PRIMARY KEY,
   partition SMALLINT NOT NULL,
   key TEXT,
   value BYTEA NOT NULL,
   buffered_at TIMESTAMP NOT NULL,
   created_at TIMESTAMP NOT NULL DEFAULT NOW(),
-  metadata JSON NOT NULL,
-  PRIMARY KEY (topic, id)
-) PARTITION BY LIST (topic);
+  metadata JSON NOT NULL
+);
 
 -- Same schema as event, just not partitioned. --
 -- It is used to handle eventual consistency of auto increment; --
@@ -52,8 +50,8 @@ CREATE TABLE event (
 -- they are then moved to event table in bulk, by a single, serialized writer; --
 -- because there is only one writer, it fixes eventual consistency issue --
 CREATE TABLE event_buffer (
-  topic TEXT NOT NULL,
   id BIGSERIAL PRIMARY KEY,
+  topic TEXT NOT NULL,
   partition SMALLINT NOT NULL,
   key TEXT,
   value BYTEA NOT NULL,
@@ -63,12 +61,12 @@ CREATE TABLE event_buffer (
 -- Used to lock single event_buffer to event writer; --
 -- there cannot be more than one record of this table! --
 CREATE TABLE event_buffer_lock (
-  created_at TIMESTAMP NOT NULL DEFAULT NOW()
+  id TEXT PRIMARY KEY
 );
-INSERT INTO event_buffer_lock VALUES (DEFAULT);
+INSERT INTO event_buffer_lock VALUES ('singleton-lock');
 ```
 
-To consume messages, we just need to periodically (every one to a few seconds) do:
+To consume events, we just need to periodically (every one to a few seconds) do:
 
 ```sql
 BEGIN;
@@ -77,8 +75,8 @@ SELECT * FROM consumer
 WHERE topic = :topic AND name = :c_name 
 FOR UPDATE SKIP LOCKED;
 
-SELECT * FROM event
-WHERE topic = :topic AND (:last_event_id IS NULL OR id > :last_event_id)
+SELECT * FROM {topic}_event
+WHERE (:last_event_id IS NULL OR id > :last_event_id)
 ORDER BY id LIMIT :limit;
 
 (process events)
@@ -105,8 +103,8 @@ SELECT * FROM consumer
 WHERE topic = :topic AND name = :c_name AND partition = 0 
 FOR UPDATE SKIP LOCKED;
 
-SELECT * FROM event
-WHERE topic = :topic AND partition = 0 AND (:last_event_id IS NULL OR id > :last_event_id)
+SELECT * FROM {topic}_event
+WHERE partition = 0 AND (:last_event_id IS NULL OR id > :last_event_id)
 ORDER BY id LIMIT :limit;
 
 (process events)
@@ -142,6 +140,9 @@ ver shardedEventSQL = new EventSQL(dataSources, EventSQLDialect.POSTGRES);
 
 Sharded version works in the same vain - it just assumes that topics and consumers are hosted on multiple dbs.
 
+Required tables are managed automatically by the library, but if you want to customize their schema a bit, you can provide your own `EventSQLRegistry.TableManager` implementation.
+See `EventSQLRegistry` for details.
+
 ### Topics and Consumers
 
 Having `EventSQL` instance, we can register topics and their consumers:
@@ -248,7 +249,7 @@ var consumers = eventSQL.consumers();
 consumers.startConsumer("txt_topic", "single-consumer", event -> {
   // handle single event
 });
-// with more frequent polling - by default it is 0.5 second
+// with more frequent polling - by default it is 1 second
 consumers.startConsumer("txt_topic", "single-consumer-customized", event -> {
   // handle single event
 }, Duration.ofMillis(100));

TODO.md

@@ -4,7 +4,4 @@
 * compact topics - unique key
 * join, aka streams
 * more elaborate definitions change support - especially around partitions growth & shrinkage
-* JavaDocs
-* Support schemas init in registry - why require schemas from users, if it is always the same (keeping dbs diffs in mind)?
-* using JOOQ tradeoffs?
-* full MySQL/MariaDB support
+* JavaDocs

benchmarks/events-db/init_db.sql

@@ -2,39 +2,3 @@ CREATE DATABASE events;
 
 CREATE USER events WITH password 'events';
 ALTER DATABASE events OWNER TO events;
-
-\c events;
-SET ROLE events;
-
-CREATE TABLE topic (
-  name TEXT PRIMARY KEY,
-  partitions SMALLINT NOT NULL,
-  created_at TIMESTAMP NOT NULL DEFAULT NOW()
-);
-
-CREATE TABLE consumer (
-  topic TEXT NOT NULL,
-  name TEXT NOT NULL,
-  partition SMALLINT NOT NULL,
-  first_event_id BIGINT,
-  last_event_id BIGINT,
-  last_consumption_at TIMESTAMP,
-  consumed_events BIGINT NOT NULL,
-  created_at TIMESTAMP NOT NULL DEFAULT NOW(),
-  PRIMARY KEY (topic, name, partition)
-);
-
-CREATE TABLE event_buffer (
-  topic TEXT NOT NULL,
-  id BIGSERIAL PRIMARY KEY,
-  partition SMALLINT NOT NULL,
-  key TEXT,
-  value BYTEA NOT NULL,
-  created_at TIMESTAMP NOT NULL DEFAULT NOW(),
-  metadata JSON NOT NULL
-);
-
-CREATE TABLE event_buffer_lock (
-  created_at TIMESTAMP NOT NULL DEFAULT NOW()
-);
-INSERT INTO event_buffer_lock VALUES (DEFAULT);

src/main/java/com/binaryigor/eventsql/EventSQLConsumers.java

@@ -7,7 +7,7 @@
 
 public interface EventSQLConsumers {
 
-    Duration DEFAULT_POLLING_DELAY = Duration.ofMillis(500);
+    Duration DEFAULT_POLLING_DELAY = Duration.ofSeconds(1);
     int DEFAULT_IN_MEMORY_EVENTS = 10;
 
     void startConsumer(String topic, String name,

src/main/java/com/binaryigor/eventsql/internal/sql/SQLConsumerRepository.java

@@ -7,6 +7,7 @@
 import org.jooq.Field;
 import org.jooq.Table;
 import org.jooq.impl.DSL;
+import org.jooq.impl.SQLDataType;
 
 import java.sql.Timestamp;
 import java.time.Instant;
@@ -40,7 +41,7 @@ public void createTable() {
                 .column(PARTITION, PARTITION.getDataType().notNull())
                 .column(FIRST_EVENT_ID)
                 .column(LAST_EVENT_ID)
-                .column(LAST_CONSUMPTION_AT)
+                .column(LAST_CONSUMPTION_AT, SQLDataType.TIMESTAMP)
                 .column(CONSUMED_EVENTS, CONSUMED_EVENTS.getDataType().notNull())
                 .column(CREATED_AT, CREATED_AT.getDataType().notNull().defaultValue(DSL.now()))
                 .constraint(DSL.constraint().primaryKey(TOPIC, NAME, PARTITION))
@@ -61,7 +62,8 @@ public void saveAll(Collection<Consumer> consumers) {
                 .insertInto(CONSUMER)
                 .columns(TOPIC, NAME, PARTITION, FIRST_EVENT_ID, LAST_EVENT_ID, LAST_CONSUMPTION_AT, CONSUMED_EVENTS);
 
-        consumers.forEach(c -> insert.values(c.topic(), c.name(), (short) c.partition(), c.firstEventId(), c.lastEventId(), c.lastConsumptionAt(), c.consumedEvents()));
+        consumers.forEach(c -> insert.values(c.topic(), c.name(), (short) c.partition(),
+                c.firstEventId(), c.lastEventId(), c.lastConsumptionAt(), c.consumedEvents()));
 
         insert.onConflict(TOPIC, NAME, PARTITION)
                 .doUpdate()