Support to replicate vanilla PostgreSQL tables

config.example.toml

@@ -77,17 +77,26 @@ sink.type = 'stdout'
 topic.namingstrategy.type = 'debezium'
 topic.prefix = 'timescaledb'
 
-timescaledb.hypertables.excludes = []
+timescaledb.hypertables.excludes = ['pgcatalog.*']
 timescaledb.hypertables.includes = ['public.test']
 timescaledb.events.read = true
 timescaledb.events.insert = true
 timescaledb.events.update = true
 timescaledb.events.delete = true
 timescaledb.events.truncate = true
-timescaledb.events.message = false
+timescaledb.events.message = false #deprecated: see postgresql.events.message
 timescaledb.events.compression = false
 timescaledb.events.decompression = false
 
+postgresql.tables.excludes = ['pgcatalog.*']
+postgresql.tables.includes = ['public.*']
+postgresql.events.read = true
+postgresql.events.insert = true
+postgresql.events.update = true
+postgresql.events.delete = true
+postgresql.events.truncate = true
+postgresql.events.message = false
+
 logging.level = 'info'
 logging.outputs.console.enabled = true
 logging.outputs.file.enabled = false

config.example.yml

@@ -18,6 +18,19 @@ postgresql:
 #      timeout: 60
 #      maxSize: 100000
 
+  tables:
+    excludes:
+      - 'pg_catalog.*'
+    includes:
+      - 'public.*'
+  events:
+    read: true
+    insert: true
+    update: true
+    delete: true
+    truncate: true
+    message: true
+
 stateStorage:
   type: file
   file:
@@ -119,7 +132,7 @@ timescaledb:
     update: true
     delete: true
     truncate: true
-    message: true
+    message: false #deprecated: see postgresql\events\message
     compression: false
     decompression: false
 

internal/eventing/eventemitting/eventemitter.go

@@ -169,10 +169,11 @@ type eventEmitterEventHandler struct {
 }
 
 func (e *eventEmitterEventHandler) OnReadEvent(
-	lsn pgtypes.LSN, hypertable *systemcatalog.Hypertable, _ *systemcatalog.Chunk, newValues map[string]any,
+	lsn pgtypes.LSN, table schema.TableAlike,
+	_ *systemcatalog.Chunk, newValues map[string]any,
 ) error {
 
-	cnValues, err := e.convertValues(hypertable, newValues)
+	cnValues, err := e.convertValues(table, newValues)
 	if err != nil {
 		return err
 	}
@@ -186,7 +187,7 @@ func (e *eventEmitterEventHandler) OnReadEvent(
 		},
 	}
 
-	return e.emit0(xld, true, hypertable,
+	return e.emit0(xld, true, table,
 		func(stream stream.Stream) (schema.Struct, error) {
 			return stream.Key(newValues)
 		},
@@ -197,15 +198,16 @@ func (e *eventEmitterEventHandler) OnReadEvent(
 }
 
 func (e *eventEmitterEventHandler) OnInsertEvent(
-	xld pgtypes.XLogData, hypertable *systemcatalog.Hypertable, _ *systemcatalog.Chunk, newValues map[string]any,
+	xld pgtypes.XLogData, table schema.TableAlike,
+	_ *systemcatalog.Chunk, newValues map[string]any,
 ) error {
 
-	cnValues, err := e.convertValues(hypertable, newValues)
+	cnValues, err := e.convertValues(table, newValues)
 	if err != nil {
 		return err
 	}
 
-	return e.emit(xld, hypertable,
+	return e.emit(xld, table,
 		func(stream stream.Stream) (schema.Struct, error) {
 			return stream.Key(newValues)
 		},
@@ -216,20 +218,20 @@ func (e *eventEmitterEventHandler) OnInsertEvent(
 }
 
 func (e *eventEmitterEventHandler) OnUpdateEvent(
-	xld pgtypes.XLogData, hypertable *systemcatalog.Hypertable,
+	xld pgtypes.XLogData, table schema.TableAlike,
 	_ *systemcatalog.Chunk, oldValues, newValues map[string]any,
 ) error {
 
-	coValues, err := e.convertValues(hypertable, oldValues)
+	coValues, err := e.convertValues(table, oldValues)
 	if err != nil {
 		return err
 	}
-	cnValues, err := e.convertValues(hypertable, newValues)
+	cnValues, err := e.convertValues(table, newValues)
 	if err != nil {
 		return err
 	}
 
-	return e.emit(xld, hypertable,
+	return e.emit(xld, table,
 		func(stream stream.Stream) (schema.Struct, error) {
 			return stream.Key(newValues)
 		},
@@ -240,16 +242,16 @@ func (e *eventEmitterEventHandler) OnUpdateEvent(
 }
 
 func (e *eventEmitterEventHandler) OnDeleteEvent(
-	xld pgtypes.XLogData, hypertable *systemcatalog.Hypertable,
+	xld pgtypes.XLogData, table schema.TableAlike,
 	_ *systemcatalog.Chunk, oldValues map[string]any, tombstone bool,
 ) error {
 
-	coValues, err := e.convertValues(hypertable, oldValues)
+	coValues, err := e.convertValues(table, oldValues)
 	if err != nil {
 		return err
 	}
 
-	return e.emit(xld, hypertable,
+	return e.emit(xld, table,
 		func(stream stream.Stream) (schema.Struct, error) {
 			return stream.Key(oldValues)
 		},
@@ -260,10 +262,10 @@ func (e *eventEmitterEventHandler) OnDeleteEvent(
 }
 
 func (e *eventEmitterEventHandler) OnTruncateEvent(
-	xld pgtypes.XLogData, hypertable *systemcatalog.Hypertable,
+	xld pgtypes.XLogData, table schema.TableAlike,
 ) error {
 
-	return e.emit(xld, hypertable,
+	return e.emit(xld, table,
 		func(stream stream.Stream) (schema.Struct, error) {
 			return nil, nil
 		},
@@ -355,15 +357,15 @@ func (e *eventEmitterEventHandler) OnTransactionFinishedEvent(
 }
 
 func (e *eventEmitterEventHandler) emit(
-	xld pgtypes.XLogData, hypertable *systemcatalog.Hypertable,
+	xld pgtypes.XLogData, table schema.TableAlike,
 	keyFactory keyFactoryFn, payloadFactory payloadFactoryFn,
 ) error {
 
-	return e.emit0(xld, false, hypertable, keyFactory, payloadFactory)
+	return e.emit0(xld, false, table, keyFactory, payloadFactory)
 }
 
 func (e *eventEmitterEventHandler) emit0(
-	xld pgtypes.XLogData, snapshot bool, hypertable *systemcatalog.Hypertable,
+	xld pgtypes.XLogData, snapshot bool, hypertable schema.TableAlike,
 	keyFactory keyFactoryFn, payloadFactory payloadFactoryFn,
 ) error {
 
@@ -452,10 +454,10 @@ func (e *eventEmitterEventHandler) timescaleEventKey(
 }
 
 func (e *eventEmitterEventHandler) convertValues(
-	hypertable *systemcatalog.Hypertable, values map[string]any,
+	table schema.TableAlike, values map[string]any,
 ) (map[string]any, error) {
 
-	return e.convertColumnValues(hypertable.TableColumns(), values)
+	return e.convertColumnValues(table.TableColumns(), values)
 }
 
 func (e *eventEmitterEventHandler) convertColumnValues(

internal/eventing/eventfiltering/eventfilter.go

@@ -29,19 +29,19 @@ import (
 
 type EventFilter interface {
 	Evaluate(
-		hypertable *systemcatalog.Hypertable, key, value schema.Struct,
+		table schema.TableAlike, key, value schema.Struct,
 	) (bool, error)
 }
 
 type eventFilterFunc func(
-	hypertable *systemcatalog.Hypertable, key, value schema.Struct,
+	table schema.TableAlike, key, value schema.Struct,
 ) (bool, error)
 
 func (eff eventFilterFunc) Evaluate(
-	hypertable *systemcatalog.Hypertable, key, value schema.Struct,
+	table schema.TableAlike, key, value schema.Struct,
 ) (bool, error) {
 
-	return eff(hypertable, key, value)
+	return eff(table, key, value)
 }
 
 func NewEventFilter(
@@ -60,8 +60,8 @@ func NewEventFilter(
 			defaultValue = *def.DefaultValue
 		}
 
-		if def.Hypertables != nil {
-			tf, err := tablefiltering.NewTableFilter(def.Hypertables.Excludes, def.Hypertables.Includes, true)
+		if def.Tables != nil {
+			tf, err := tablefiltering.NewTableFilter(def.Tables.Excludes, def.Tables.Includes, true)
 			if err != nil {
 				return nil, err
 			}
@@ -86,7 +86,7 @@ func NewEventFilter(
 }
 
 var acceptAllFilter eventFilterFunc = func(
-	_ *systemcatalog.Hypertable, _, _ schema.Struct,
+	_ schema.TableAlike, _, _ schema.Struct,
 ) (bool, error) {
 
 	return true, nil
@@ -98,11 +98,11 @@ var compositeFilter = func(
 
 	return eventFilterFunc(
 		func(
-			hypertable *systemcatalog.Hypertable, key, value schema.Struct,
+			table schema.TableAlike, key, value schema.Struct,
 		) (bool, error) {
 
 			for i, tableFilter := range tableFilters {
-				if hypertable == nil || tableFilter.Enabled(hypertable) {
+				if table == nil || tableFilter.Enabled(table) {
 					success, err := filters[i].evaluate(key, value)
 					if err != nil {
 						return false, err
@@ -153,23 +153,23 @@ func (f *eventFilter) evaluate(
 
 type tableFilter interface {
 	Enabled(
-		hypertable *systemcatalog.Hypertable,
+		table systemcatalog.SystemEntity,
 	) bool
 }
 
 type tableFilterFunc func(
-	hypertable *systemcatalog.Hypertable,
+	table systemcatalog.SystemEntity,
 ) bool
 
 func (tff tableFilterFunc) Enabled(
-	hypertable *systemcatalog.Hypertable,
+	table systemcatalog.SystemEntity,
 ) bool {
 
-	return tff(hypertable)
+	return tff(table)
 }
 
 var acceptAllTableFilter tableFilterFunc = func(
-	*systemcatalog.Hypertable,
+	systemcatalog.SystemEntity,
 ) bool {
 
 	return true