Merge pull request #7 from BinaryIgor/no-partitions-approach

No partitions approach

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):
+We just need to have a few tables (postgres syntax, schema managed fully by EventSQL):
 
 ```sql
 CREATE TABLE topic (
@@ -33,42 +33,39 @@ 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. --
+-- Same schema as event, just not partitioned (by topic). --
 -- It is used to handle eventual consistency of auto increment; --
 -- there is no guarantee that record of id 2 is visible after id 1 record. --
 -- Events are first inserted to the event_buffer; --
--- they are then moved to event table in bulk, by a single, serialized writer; --
+-- they are then moved to the {topic}_event table in bulk, by a single, serialized writer (per topic); --
 -- 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,
   created_at TIMESTAMP NOT NULL DEFAULT NOW(),
   metadata JSON NOT NULL
 );
--- Used to lock single event_buffer to event writer; --
--- there cannot be more than one record of this table! --
+CREATE INDEX event_buffer_topic_id ON event_buffer (topic, id);
+-- Used to lock single (per topic) event_buffer to {topic}_event writer --
 CREATE TABLE event_buffer_lock (
-  created_at TIMESTAMP NOT NULL DEFAULT NOW()
+  topic TEXT PRIMARY KEY
 );
-INSERT INTO event_buffer_lock VALUES (DEFAULT);
 ```
 
-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 +74,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 +102,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)
@@ -130,9 +127,8 @@ them:
 ```java
 
 import com.binaryigor.eventsql.EventSQL;
-// dialect of your events backend - POSTGRES, MYSQL, MARIADB and so on;
-// as of now, only POSTGRES has fully tested support;
-// should also work with others but some things - event table partition management for example - works only with Postgres, for others it must be managed manually
+// dialect of your events backend - POSTGRES, MYSQL, MARIADB;
+// as of now, only POSTGRES has fully tested support, but should work on others as well
 import com.binaryigor.eventsql.EventSQLDialect;
 import javax.sql.DataSource;
 
@@ -142,6 +138,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.TablesManager` implementation.
+See `EventSQLRegistry` for details.
+
 ### Topics and Consumers
 
 Having `EventSQL` instance, we can register topics and their consumers:
@@ -248,7 +247,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,51 +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 (
-  topic TEXT NOT NULL,
-  id BIGSERIAL NOT NULL,
-  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);
-
-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);

benchmarks/runner/src/main/java/com/binaryigor/eventsql/benchmarks/EventSQLBenchmarksRunner.java

@@ -128,7 +128,7 @@ static ConsumerDefinition consumerDefinition(EventSQL eventSQL) {
     }
 
     static EventTableStats eventTableStats(DataSource source) {
-        return executeQuery(source, "SELECT partition, MAX(id) FROM event WHERE topic = '%s' GROUP BY partition"
+        return executeQuery(source, "SELECT partition, MAX(id) FROM %s_event GROUP BY partition"
                 .formatted(TEST_TOPIC), r -> {
             var lastIdsPerPartition = new HashMap<Integer, Long>();
             while (r.next()) {

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

@@ -69,7 +69,7 @@ public EventSQL(Collection<DataSource> dataSources, EventSQLDialect sqlDialect,
 
             var topicRepository = new SQLTopicRepository(transactions);
             var consumerRepository = new SQLConsumerRepository(transactions);
-            var eventRepository = new SQLEventRepository(transactions, transactions, sqlDialect);
+            var eventRepository = new SQLEventRepository(transactions, transactions);
 
             var registry = new DefaultEventSQLRegistry(topicRepository, eventRepository, consumerRepository, transactions);
 

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/EventSQLRegistry.java

@@ -15,4 +15,20 @@ public interface EventSQLRegistry {
     EventSQLRegistry unregisterConsumer(String topic, String name);
 
     List<ConsumerDefinition> listConsumers();
+    
+    void configureTablesManager(TablesManager tablesManager);
+    
+    TablesManager tablesManager();
+
+    // all methods must be idempotent
+    interface TablesManager {
+        
+        void prepareTopicTable();
+        
+        void prepareConsumerTable();
+        
+        void prepareEventTable(String topic);
+        
+        void dropEventTable(String topic);
+    }
 }

src/main/java/com/binaryigor/eventsql/internal/ConsumerRepository.java

@@ -6,6 +6,8 @@
 
 public interface ConsumerRepository {
 
+    void createTable();
+
     void save(Consumer consumer);
 
     void saveAll(Collection<Consumer> consumers);

src/main/java/com/binaryigor/eventsql/internal/DefaultEventSQLPublisher.java

@@ -8,7 +8,10 @@
 
 import java.time.Duration;
 import java.util.Collection;
+import java.util.Collections;
 import java.util.List;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
@@ -24,6 +27,7 @@ public class DefaultEventSQLPublisher implements EventSQLPublisher {
     private final EventRepository eventRepository;
     private Partitioner partitioner;
     private final AtomicBoolean published = new AtomicBoolean(false);
+    private final Set<String> publishedTopics = Collections.newSetFromMap(new ConcurrentHashMap<>());
     private final AtomicBoolean flushPublishBufferThreadSet = new AtomicBoolean(false);
     private volatile Thread flushPublishBufferThread;
     private final int flushPublishBufferSize;
@@ -67,6 +71,8 @@ private void publish(String topicName, Collection<EventPublication> publications
         eventRepository.createAll(toCreateEvents);
 
         published.set(true);
+        publishedTopics.add(topicName);
+
         var shouldSetFlushPublishBufferThread = !flushPublishBufferThreadSet.compareAndExchange(false, true);
         if (shouldSetFlushPublishBufferThread) {
             startFlushPublishBufferThread();
@@ -82,7 +88,7 @@ private void startFlushPublishBufferThread() {
                         Thread.sleep(flushPublishBufferDelay);
                     }
                     if (moreToFlush || published.getAndSet(false)) {
-                        var flushed = eventRepository.flushBuffer(flushPublishBufferSize);
+                        var flushed = eventRepository.flushBuffer(publishedTopics, flushPublishBufferSize);
                         moreToFlush = flushed > flushPublishBufferSize;
                     }
                 } catch (Exception e) {

src/main/java/com/binaryigor/eventsql/internal/DefaultEventSQLRegistry.java

@@ -17,6 +17,7 @@ public class DefaultEventSQLRegistry implements EventSQLRegistry {
     private final EventRepository eventRepository;
     private final ConsumerRepository consumerRepository;
     private final Transactions transactions;
+    private TablesManager tablesManager;
 
     public DefaultEventSQLRegistry(TopicRepository topicRepository,
                                    EventRepository eventRepository,
@@ -26,6 +27,14 @@ public DefaultEventSQLRegistry(TopicRepository topicRepository,
         this.eventRepository = eventRepository;
         this.consumerRepository = consumerRepository;
         this.transactions = transactions;
+        this.tablesManager = new DefaultTablesManager(topicRepository, consumerRepository, eventRepository);
+
+        registerTables();
+    }
+
+    private void registerTables() {
+        tablesManager.prepareTopicTable();
+        tablesManager.prepareConsumerTable();
     }
 
     // TODO: support more complex modifications
@@ -40,7 +49,7 @@ public EventSQLRegistry registerTopic(TopicDefinition topic) {
         if (currentTopicDefinitionOpt.isEmpty()) {
             transactions.execute(() -> {
                 topicRepository.save(topic);
-                eventRepository.createPartition(topic.name());
+                tablesManager.prepareEventTable(topic.name());
             });
             return this;
         }
@@ -72,7 +81,7 @@ public EventSQLRegistry unregisterTopic(String topic) {
             }
 
             topicRepository.delete(topic);
-            eventRepository.deletePartition(topic);
+            tablesManager.dropEventTable(topic);
         });
 
         return this;
@@ -92,6 +101,8 @@ public EventSQLRegistry registerConsumer(ConsumerDefinition consumer) {
                     .formatted(topic.name(), consumer.name()));
         }
 
+        tablesManager.prepareConsumerTable();
+
         var currentConsumers = consumerRepository.allOf(consumer.topic(), consumer.name());
         if (consumerDefinitionHaveNotChanged(currentConsumers, topic, consumer)) {
             return this;
@@ -140,6 +151,16 @@ public List<ConsumerDefinition> listConsumers() {
                 .toList();
     }
 
+    @Override
+    public void configureTablesManager(TablesManager tablesManager) {
+        this.tablesManager = tablesManager;
+    }
+
+    @Override
+    public TablesManager tablesManager() {
+        return tablesManager;
+    }
+
     private TopicDefinition findTopicDefinition(String topic) {
         return topicRepository.ofName(topic)
                 .orElseThrow(() -> new IllegalArgumentException("%s topic doesn't exist".formatted(topic)));