No partitions approach

BinaryIgor · BinaryIgor · commit 2ccbc507df6a · 2025-05-26T13:20:25.000+02:00
diff --git a/README.md b/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 by EventSQL):
 
 ```sql
 CREATE TABLE topic (
diff --git a/benchmarks/events-db/init_db.sql b/benchmarks/events-db/init_db.sql
@@ -24,18 +24,6 @@ CREATE TABLE consumer (
   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,
diff --git a/benchmarks/runner/src/main/java/com/binaryigor/eventsql/benchmarks/EventSQLBenchmarksRunner.java b/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()) {
diff --git a/src/main/java/com/binaryigor/eventsql/EventSQL.java b/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);
 
diff --git a/src/main/java/com/binaryigor/eventsql/EventSQLRegistry.java b/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 configureTableManager(TableManager tableManager);
+    
+    TableManager tableManager();
+
+    // all methods must be idempotent
+    interface TableManager {
+        
+        void prepareTopicTable();
+        
+        void prepareConsumerTable();
+        
+        void prepareEventTable(String topic);
+        
+        void dropEventTable(String topic);
+    }
 }
diff --git a/src/main/java/com/binaryigor/eventsql/internal/ConsumerRepository.java b/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);
diff --git a/src/main/java/com/binaryigor/eventsql/internal/DefaultEventSQLRegistry.java b/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 TableManager tableManager;
 
     public DefaultEventSQLRegistry(TopicRepository topicRepository,
                                    EventRepository eventRepository,
@@ -26,6 +27,14 @@ public DefaultEventSQLRegistry(TopicRepository topicRepository,
         this.eventRepository = eventRepository;
         this.consumerRepository = consumerRepository;
         this.transactions = transactions;
+        this.tableManager = new DefaultTableManager(topicRepository, consumerRepository, eventRepository);
+
+        registerTables();
+    }
+
+    private void registerTables() {
+        tableManager.prepareTopicTable();
+        tableManager.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());
+                tableManager.prepareEventTable(topic.name());
             });
             return this;
         }
@@ -72,7 +81,7 @@ public EventSQLRegistry unregisterTopic(String topic) {
             }
 
             topicRepository.delete(topic);
-            eventRepository.deletePartition(topic);
+            tableManager.dropEventTable(topic);
         });
 
         return this;
@@ -92,6 +101,8 @@ public EventSQLRegistry registerConsumer(ConsumerDefinition consumer) {
                     .formatted(topic.name(), consumer.name()));
         }
 
+        tableManager.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 configureTableManager(TableManager tableManager) {
+        this.tableManager = tableManager;
+    }
+
+    @Override
+    public TableManager tableManager() {
+        return tableManager;
+    }
+
     private TopicDefinition findTopicDefinition(String topic) {
         return topicRepository.ofName(topic)
                 .orElseThrow(() -> new IllegalArgumentException("%s topic doesn't exist".formatted(topic)));
diff --git a/src/main/java/com/binaryigor/eventsql/internal/DefaultTableManager.java b/src/main/java/com/binaryigor/eventsql/internal/DefaultTableManager.java
@@ -0,0 +1,38 @@
+package com.binaryigor.eventsql.internal;
+
+import com.binaryigor.eventsql.EventSQLRegistry;
+
+public class DefaultTableManager implements EventSQLRegistry.TableManager {
+
+    private final TopicRepository topicRepository;
+    private final ConsumerRepository consumerRepository;
+    private final EventRepository eventRepository;
+
+    public DefaultTableManager(TopicRepository topicRepository, ConsumerRepository consumerRepository, EventRepository eventRepository) {
+        this.topicRepository = topicRepository;
+        this.consumerRepository = consumerRepository;
+        this.eventRepository = eventRepository;
+    }
+
+    @Override
+    public void prepareTopicTable() {
+        topicRepository.createTable();
+    }
+
+    @Override
+    public void prepareConsumerTable() {
+        consumerRepository.createTable();
+    }
+
+    @Override
+    public void prepareEventTable(String topic) {
+        eventRepository.prepareBufferLock();
+        eventRepository.createBuffer();
+        eventRepository.createPartition(topic);
+    }
+
+    @Override
+    public void dropEventTable(String topic) {
+        eventRepository.dropPartition(topic);
+    }
+}
diff --git a/src/main/java/com/binaryigor/eventsql/internal/EventRepository.java b/src/main/java/com/binaryigor/eventsql/internal/EventRepository.java
@@ -7,9 +7,13 @@
 
 public interface EventRepository {
 
+    void createBuffer();
+
+    void prepareBufferLock();
+
     void createPartition(String topic);
 
-    void deletePartition(String topic);
+    void dropPartition(String topic);
 
     void create(EventInput event);
 
diff --git a/src/main/java/com/binaryigor/eventsql/internal/TopicRepository.java b/src/main/java/com/binaryigor/eventsql/internal/TopicRepository.java
@@ -7,6 +7,8 @@
 
 public interface TopicRepository {
 
+    void createTable();
+
     void save(TopicDefinition topic);
 
     Optional<TopicDefinition> ofName(String name);
diff --git a/src/main/java/com/binaryigor/eventsql/internal/sharded/ShardedEventSQLRegistry.java b/src/main/java/com/binaryigor/eventsql/internal/sharded/ShardedEventSQLRegistry.java
@@ -47,4 +47,14 @@ public EventSQLRegistry unregisterConsumer(String topic, String name) {
     public List<ConsumerDefinition> listConsumers() {
         return registries.getFirst().listConsumers();
     }
+
+    @Override
+    public void configureTableManager(TableManager tableManager) {
+        registries.forEach(r -> r.configureTableManager(tableManager));
+    }
+
+    @Override
+    public TableManager tableManager() {
+        return registries.getFirst().tableManager();
+    }
 }
diff --git a/src/main/java/com/binaryigor/eventsql/internal/sql/SQLConsumerRepository.java b/src/main/java/com/binaryigor/eventsql/internal/sql/SQLConsumerRepository.java
@@ -8,6 +8,7 @@
 import org.jooq.Table;
 import org.jooq.impl.DSL;
 
+import java.sql.Timestamp;
 import java.time.Instant;
 import java.util.Collection;
 import java.util.List;
@@ -23,12 +24,29 @@ public class SQLConsumerRepository implements ConsumerRepository {
     private static final Field<Long> LAST_EVENT_ID = DSL.field("last_event_id", Long.class);
     private static final Field<Instant> LAST_CONSUMPTION_AT = DSL.field("last_consumption_at", Instant.class);
     private static final Field<Long> CONSUMED_EVENTS = DSL.field("consumed_events", Long.class);
+    private static final Field<Timestamp> CREATED_AT = DSL.field("created_at", Timestamp.class);
     private final DSLContextProvider contextProvider;
 
     public SQLConsumerRepository(DSLContextProvider contextProvider) {
         this.contextProvider = contextProvider;
     }
 
+    @Override
+    public void createTable() {
+        contextProvider.get()
+                .createTableIfNotExists(CONSUMER)
+                .column(TOPIC, TOPIC.getDataType().notNull())
+                .column(NAME, NAME.getDataType().notNull())
+                .column(PARTITION, PARTITION.getDataType().notNull())
+                .column(FIRST_EVENT_ID)
+                .column(LAST_EVENT_ID)
+                .column(LAST_CONSUMPTION_AT)
+                .column(CONSUMED_EVENTS, CONSUMED_EVENTS.getDataType().notNull())
+                .column(CREATED_AT, CREATED_AT.getDataType().notNull().defaultValue(DSL.now()))
+                .constraint(DSL.constraint().primaryKey(TOPIC, NAME, PARTITION))
+                .execute();
+    }
+
     @Override
     public void save(Consumer consumer) {
         saveAll(List.of(consumer));
diff --git a/src/main/java/com/binaryigor/eventsql/internal/sql/SQLEventRepository.java b/src/main/java/com/binaryigor/eventsql/internal/sql/SQLEventRepository.java
diff --git a/src/main/java/com/binaryigor/eventsql/internal/sql/SQLTopicRepository.java b/src/main/java/com/binaryigor/eventsql/internal/sql/SQLTopicRepository.java
diff --git a/src/test/java/com/binaryigor/eventsql/EventSQLRegistryTest.java b/src/test/java/com/binaryigor/eventsql/EventSQLRegistryTest.java
diff --git a/src/test/java/com/binaryigor/eventsql/test/IntegrationTest.java b/src/test/java/com/binaryigor/eventsql/test/IntegrationTest.java
diff --git a/src/test/java/com/binaryigor/eventsql/test/ShardedIntegrationTest.java b/src/test/java/com/binaryigor/eventsql/test/ShardedIntegrationTest.java

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