Better schema

Author: BinaryIgor

README.md

@@ -43,11 +43,11 @@ CREATE TABLE {topic}_event (
   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 (
   id BIGSERIAL PRIMARY KEY,
@@ -58,12 +58,11 @@ CREATE TABLE event_buffer (
   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 (
-  id TEXT PRIMARY KEY
+  topic TEXT PRIMARY KEY
 );
-INSERT INTO event_buffer_lock VALUES ('singleton-lock');
 ```
 
 To consume events, we just need to periodically (every one to a few seconds) do:
@@ -128,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;
 
@@ -140,7 +138,7 @@ 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.
+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

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

@@ -16,12 +16,12 @@ public interface EventSQLRegistry {
 
     List<ConsumerDefinition> listConsumers();
 
-    void configureTableManager(TableManager tableManager);
+    void configureTablesManager(TablesManager tablesManager);
 
-    TableManager tableManager();
+    TablesManager tablesManager();
 
     // all methods must be idempotent
-    interface TableManager {
+    interface TablesManager {
 
         void prepareTopicTable();
 

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,7 +17,7 @@ public class DefaultEventSQLRegistry implements EventSQLRegistry {
     private final EventRepository eventRepository;
     private final ConsumerRepository consumerRepository;
     private final Transactions transactions;
-    private TableManager tableManager;
+    private TablesManager tablesManager;
 
     public DefaultEventSQLRegistry(TopicRepository topicRepository,
                                    EventRepository eventRepository,
@@ -27,14 +27,14 @@ public DefaultEventSQLRegistry(TopicRepository topicRepository,
         this.eventRepository = eventRepository;
         this.consumerRepository = consumerRepository;
         this.transactions = transactions;
-        this.tableManager = new DefaultTableManager(topicRepository, consumerRepository, eventRepository);
+        this.tablesManager = new DefaultTablesManager(topicRepository, consumerRepository, eventRepository);
 
         registerTables();
     }
 
     private void registerTables() {
-        tableManager.prepareTopicTable();
-        tableManager.prepareConsumerTable();
+        tablesManager.prepareTopicTable();
+        tablesManager.prepareConsumerTable();
     }
 
     // TODO: support more complex modifications
@@ -49,7 +49,7 @@ public EventSQLRegistry registerTopic(TopicDefinition topic) {
         if (currentTopicDefinitionOpt.isEmpty()) {
             transactions.execute(() -> {
                 topicRepository.save(topic);
-                tableManager.prepareEventTable(topic.name());
+                tablesManager.prepareEventTable(topic.name());
             });
             return this;
         }
@@ -81,7 +81,7 @@ public EventSQLRegistry unregisterTopic(String topic) {
             }
 
             topicRepository.delete(topic);
-            tableManager.dropEventTable(topic);
+            tablesManager.dropEventTable(topic);
         });
 
         return this;
@@ -101,7 +101,7 @@ public EventSQLRegistry registerConsumer(ConsumerDefinition consumer) {
                     .formatted(topic.name(), consumer.name()));
         }
 
-        tableManager.prepareConsumerTable();
+        tablesManager.prepareConsumerTable();
 
         var currentConsumers = consumerRepository.allOf(consumer.topic(), consumer.name());
         if (consumerDefinitionHaveNotChanged(currentConsumers, topic, consumer)) {
@@ -152,13 +152,13 @@ public List<ConsumerDefinition> listConsumers() {
     }
 
     @Override
-    public void configureTableManager(TableManager tableManager) {
-        this.tableManager = tableManager;
+    public void configureTablesManager(TablesManager tablesManager) {
+        this.tablesManager = tablesManager;
     }
 
     @Override
-    public TableManager tableManager() {
-        return tableManager;
+    public TablesManager tablesManager() {
+        return tablesManager;
     }
 
     private TopicDefinition findTopicDefinition(String topic) {

…entsql/internal/DefaultTableManager.java …ntsql/internal/DefaultTablesManager.javasrc/main/java/com/binaryigor/eventsql/internal/DefaultTableManager.java renamed to src/main/java/com/binaryigor/eventsql/internal/DefaultTablesManager.java src/main/java/com/binaryigor/eventsql/internal/DefaultTableManager.java renamed to src/main/java/com/binaryigor/eventsql/internal/DefaultTablesManager.java

@@ -2,13 +2,13 @@
 
 import com.binaryigor.eventsql.EventSQLRegistry;
 
-public class DefaultTableManager implements EventSQLRegistry.TableManager {
+public class DefaultTablesManager implements EventSQLRegistry.TablesManager {
 
     private final TopicRepository topicRepository;
     private final ConsumerRepository consumerRepository;
     private final EventRepository eventRepository;
 
-    public DefaultTableManager(TopicRepository topicRepository, ConsumerRepository consumerRepository, EventRepository eventRepository) {
+    public DefaultTablesManager(TopicRepository topicRepository, ConsumerRepository consumerRepository, EventRepository eventRepository) {
         this.topicRepository = topicRepository;
         this.consumerRepository = consumerRepository;
         this.eventRepository = eventRepository;
@@ -26,7 +26,6 @@ public void prepareConsumerTable() {
 
     @Override
     public void prepareEventTable(String topic) {
-        eventRepository.prepareBufferLock();
         eventRepository.createBuffer();
         eventRepository.createPartition(topic);
     }

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

@@ -9,8 +9,6 @@ public interface EventRepository {
 
     void createBuffer();
 
-    void prepareBufferLock();
-
     void createPartition(String topic);
 
     void dropPartition(String topic);
@@ -19,7 +17,7 @@ public interface EventRepository {
 
     void createAll(Collection<EventInput> events);
 
-    int flushBuffer(int toFlush);
+    int flushBuffer(Collection<String> topics, int toFlush);
 
     List<Event> nextEvents(String topic, Long lastId, int limit);
 

src/main/java/com/binaryigor/eventsql/internal/sharded/ShardedEventSQLRegistry.java

@@ -49,12 +49,12 @@ public List<ConsumerDefinition> listConsumers() {
     }
 
     @Override
-    public void configureTableManager(TableManager tableManager) {
-        registries.forEach(r -> r.configureTableManager(tableManager));
+    public void configureTablesManager(TablesManager tablesManager) {
+        registries.forEach(r -> r.configureTablesManager(tablesManager));
     }
 
     @Override
-    public TableManager tableManager() {
-        return registries.getFirst().tableManager();
+    public TablesManager tablesManager() {
+        return registries.getFirst().tablesManager();
     }
 }