[Issue #13] Integrate Retina and add Debezium Engine as a data source.

.gitignore

@@ -22,7 +22,6 @@
 # virtual machine crash logs, see http://www.java.com/en/download/help/error_hotspot.xml
 hs_err_pid*
 
-.idea
 build
 target
 .classpath*
@@ -48,3 +47,21 @@ resources/*.xml
 *.so
 *.o
 .vscode
+.idea
+
+data/
+tmp/
+
+!requirements.txt
+freshness*.png
+rate*.png
+resulti7i/
+result1k2_2/
+result_lance
+cluster*.png
+pixels-sink.out
+/*.png
+/*.csv
+*logs/*.out
+*logs/*.png
+*logs/**/*.png

README.md

@@ -1,4 +1,50 @@
 # pixels-sink
-The data sink for Pixels. It ingests data into Pixels in a streaming manner.
 
-> This project is under development.
+Pixels Sink is the data sink service for Pixels. It ingests debezium-format change events from multiple sources (Debezium engine, Kafka, or storage files), converts them into Pixels events, and writes them to a configured sink (Retina, CSV, Proto, Flink, or none).
+
+This project is under active development.
+
+## Docs
+
+- Architecture and pipeline overview: [docs/overview.md](docs/overview.md)
+- Transaction handling: [docs/transaction.md](docs/transaction.md)
+- Usage guide: [docs/usage.md](docs/usage.md)
+- Configuration reference: [docs/configuration.md](docs/configuration.md)
+- Local dev environment (Docker): [develop/README.md](develop/README.md)
+
+## Quick Start
+
+### Requirements
+- [Pixels](https://github.com/pixelsdb/pixels)
+- Java 17
+- Maven 3.9+
+- Source and sink dependencies based on your configuration
+  - Kafka broker if `sink.datasource=kafka`
+  - Debezium + database access if `sink.datasource=engine`
+  - Retina service if `sink.mode=retina`
+  - Trino if freshness checking is enabled and uses Trino
+
+### Build
+```bash
+mvn -q -DskipTests package
+```
+
+### Run (Script)
+```bash
+./pixels-sink [config.properties]
+```
+
+The script reads `conf/jvm.conf` and uses a properties file configured inside `./pixels-sink` by default. If you pass a path, it overrides the default.
+
+### Run (IDE)
+- Main class: `io.pixelsdb.pixels.sink.PixelsSinkApp`
+- Program arguments: `-c conf/pixels-sink.aws.properties`
+
+## Configuration
+- Sample configs are in `conf/`.
+- Start with `conf/pixels-sink.aws.properties` and adjust.
+- See [docs/configuration.md](docs/configuration.md) for a full key reference and guidance.
+
+## Monitoring
+- Enable Prometheus metrics with `sink.monitor.enable=true`.
+- Metrics endpoint listens on `sink.monitor.port` (default `9464`).

conf/jvm.conf

@@ -0,0 +1,26 @@
+-server
+-XX:+AlwaysPreTouch
+-Dfile.encoding=UTF-8
+-Duser.timezone=UTC
+
+-Xms8g
+-Xmx60g
+
+-XX:+UseG1GC
+-XX:MaxGCPauseMillis=200
+-XX:InitiatingHeapOccupancyPercent=35
+-XX:+ParallelRefProcEnabled
+-XX:+UnlockExperimentalVMOptions
+-XX:+TrustFinalNonStaticFields
+-XX:+DisableExplicitGC
+
+-Xss512k
+
+
+-XX:+HeapDumpOnOutOfMemoryError
+-XX:HeapDumpPath=/var/log/java/java_heapdump.hprof
+-XX:+ExitOnOutOfMemoryError
+
+-agentlib:jdwp=transport=dt_socket,server=y,suspend=n,address=*:10086
+--add-opens=java.base/sun.nio.ch=ALL-UNNAMED
+--add-opens=java.base/java.nio=ALL-UNNAMED

conf/pixels-sink.aws.properties

@@ -0,0 +1,118 @@
+# engine | kafka | storage
+sink.datasource=storage
+# -1 means no limit, Only implement in retina sink mode yet:
+sink.datasource.rate.limit=200000
+sink.monitor.report.file=/home/ubuntu/pixels-sink/result1k2_feb/rate_8192tile_3.csv
+sink.monitor.freshness.file=/home/ubuntu/pixels-sink/result1k2_feb/fresh_8192tile_3.csv
+# rate limiter implement: guava or semaphore
+sink.datasource.rate.limit.type=guava
+# Sink Config: retina | csv | proto | flink | none
+sink.mode=retina
+sink.retina.client=1
+# in flight rpc
+sink.retina.rpc.limit=100000
+sink.retina.trans.limit=10000000
+sink.retina.log.queue=false
+sink.retina.trans.request.batch=true
+sink.commit.method=async
+sink.retina.trans.request.batch.size=100
+## batch or single or record, batch is recommend. record is faster, but doesn't have ACID feature
+sink.trans.mode=batch
+sink.monitor.report.enable=true
+# trino for freshness query
+trino.url=jdbc:trino://realtime-pixels-coordinator:8080/pixels/pixels_bench
+# trino.url=jdbc:trino://realtime-pixels-coordinator:8080/pixels/pixels_bench_sf10x
+trino.user=pixels
+trino.password=password
+trino.parallel=1
+# row or txn or embed
+sink.monitor.freshness.level=embed
+sink.monitor.freshness.verbose=true
+sink.monitor.freshness.embed.warmup=10
+sink.monitor.freshness.embed.static=false
+sink.monitor.freshness.embed.snapshot=true
+sink.monitor.freshness.embed.tablelist=loantrans
+sink.monitor.freshness.embed.delay=10
+# sink.monitor.freshness.embed.tablelist=savingaccount
+sink.monitor.freshness.timestamp=true
+sink.storage.loop=true
+# Kafka Config
+bootstrap.servers=realtime-kafka-2:29092
+group.id=3078
+auto.offset.reset=earliest
+key.deserializer=org.apache.kafka.common.serialization.StringDeserializer
+#value.deserializer=io.pixelsdb.pixels.writer.deserializer.RowChangeEventAvroDeserializer
+value.deserializer=io.pixelsdb.pixels.sink.event.deserializer.RowChangeEventJsonDeserializer
+# Topic & Database Config
+topic.prefix=postgresql.oltp_server
+consumer.capture_database=pixels_bench_sf1x
+consumer.include_tables=
+sink.csv.path=./data
+sink.csv.enable_header=false
+## Retina Config
+sink.retina.embedded=false
+# stub or stream
+sink.retina.mode=stream
+#writer.retina.mode=stub
+sink.remote.host=localhost
+sink.remote.port=29422
+sink.timeout.ms=5000
+sink.flush.interval.ms=5
+sink.flush.batch.size=100
+sink.max.retries=3
+## writer commit
+# sync or async
+sink.commit.batch.size=100
+sink.commit.batch.worker=4
+sink.commit.batch.delay=10
+## Proto Config
+sink.proto.dir=file:///home/ubuntu/disk2/hybench/
+sink.proto.data=hybench1000_4
+# sink.proto.data=hybench100_3
+# sink.proto.data=hybench100_4
+# sink.proto.data=hybench10_10
+sink.proto.maxRecords=100000
+## Flink Config
+sink.flink.server.port=9091
+## Schema Registry
+sink.registry.url=http://localhost:8080/apis/registry/v2
+# Transaction Config
+transaction.topic.suffix=transaction
+#transaction.topic.value.deserializer=io.pixelsdb.pixels.writer.deserializer.TransactionAvroMessageDeserializer
+transaction.topic.value.deserializer=io.pixelsdb.pixels.sink.event.deserializer.TransactionJsonMessageDeserializer
+sink.trans.batch.size=100
+
+# Sink Metrics
+sink.monitor.enable=true
+sink.monitor.port=9464
+sink.monitor.report.interval=1000
+sink.monitor.freshness.interval=1000
+
+# Interact with other rpc
+sink.rpc.enable=true
+sink.rpc.mock.delay=20
+# debezium engine config
+debezium.name=testEngine
+debezium.connector.class=io.debezium.connector.postgresql.PostgresConnector
+debezium.provide.transaction.metadata=true
+debezium.offset.storage=org.apache.kafka.connect.storage.FileOffsetBackingStore
+debezium.offset.storage.file.filename=/tmp/offsets.dat
+debezium.offset.flush.interval.ms=60000
+debezium.schema.history.internal=io.debezium.storage.file.history.FileSchemaHistory
+debezium.schema.history.internal.file.filename=/tmp/schemahistory.dat
+debezium.database.hostname=realtime-pg-2
+debezium.database.port=5432
+debezium.database.user=pixels
+debezium.database.password=pixels_realtime_crud
+debezium.database.dbname=pixels_bench
+debezium.plugin.name=pgoutput
+debezium.database.server.id=1
+debezium.schema.include.list=public
+debezium.snapshot.mode=never
+debezium.key.converter=org.apache.kafka.connect.json.JsonConverter
+debezium.value.converter=org.apache.kafka.connect.json.JsonConverter
+debezium.topic.prefix=postgresql.oltp_server
+debezium.transforms=topicRouting
+debezium.transforms.topicRouting.type=org.apache.kafka.connect.transforms.RegexRouter
+debezium.transforms.topicRouting.regex=postgresql\\.oltp_server\\.public\\.(.*)
+debezium.transforms.topicRouting.replacement=postgresql.oltp_server.pixels_bench_sf10x.$1

conf/pixels-sink.ch.properties

@@ -0,0 +1,115 @@
+# engine | kafka | storage
+sink.datasource=storage
+sink.mode=retina
+#sink.datasource=engine
+#sink.mode=proto
+sink.proto.data=CH10K_2
+# -1 means no limit, Only implement in retina sink mode yet:
+sink.datasource.rate.limit=80000
+sink.monitor.report.file=/home/ubuntu/pixels-sink/result_ch/rate_100K.csv
+sink.monitor.freshness.file=/home/ubuntu/pixels-sink/result_ch/fresh_100K.csv
+# rate limiter implement: guava or semaphore
+sink.datasource.rate.limit.type=guava
+sink.retina.client=1
+# in flight rpc
+sink.retina.rpc.limit=1000000
+sink.retina.trans.limit=1000000
+sink.retina.log.queue=false
+sink.retina.trans.request.batch=false
+sink.commit.method=sync
+sink.retina.trans.request.batch.size=1000
+## batch or single or record, batch is recommend. record is faster, but doesn't have ACID feature
+sink.trans.mode=batch
+# sink.trans.mode=record
+sink.monitor.report.enable=true
+# trino for freshness query
+trino.url=jdbc:trino://realtime-pixels-coordinator:8080/pixels/pixels_bench
+trino.user=pixels
+trino.password=password
+trino.parallel=1
+# row or txn or embed
+sink.monitor.freshness.level=embed
+sink.monitor.freshness.verbose=true
+sink.monitor.freshness.embed.warmup=10
+sink.monitor.freshness.embed.static=false
+sink.monitor.freshness.embed.snapshot=true
+sink.monitor.freshness.embed.tablelist=stock
+sink.monitor.freshness.embed.delay=10
+sink.monitor.freshness.timestamp=true
+sink.storage.loop=true
+# Kafka Config
+bootstrap.servers=realtime-kafka-2:29092
+group.id=3078
+auto.offset.reset=earliest
+key.deserializer=org.apache.kafka.common.serialization.StringDeserializer
+#value.deserializer=io.pixelsdb.pixels.writer.deserializer.RowChangeEventAvroDeserializer
+value.deserializer=io.pixelsdb.pixels.sink.event.deserializer.RowChangeEventJsonDeserializer
+# Topic & Database Config
+topic.prefix=postgresql.oltp_server
+consumer.capture_database=pixels_bench_sf1x
+consumer.include_tables=
+sink.csv.path=./data
+sink.csv.enable_header=false
+## Retina Config
+sink.retina.embedded=false
+# stub or stream
+sink.retina.mode=stream
+#writer.retina.mode=stub
+sink.remote.host=localhost
+sink.remote.port=29422
+sink.timeout.ms=5000
+sink.flush.interval.ms=10
+sink.flush.batch.size=200
+sink.max.retries=3
+## writer commit
+# sync or async
+sink.commit.batch.size=1000
+sink.commit.batch.worker=8
+sink.commit.batch.delay=1000
+## Proto Config
+sink.proto.dir=file:///home/ubuntu/disk2/chbench/
+sink.proto.maxRecords=100000
+## Flink Config
+sink.flink.server.port=9091
+## Schema Registry
+sink.registry.url=http://localhost:8080/apis/registry/v2
+# Transaction Config
+transaction.topic.suffix=transaction
+#transaction.topic.value.deserializer=io.pixelsdb.pixels.writer.deserializer.TransactionAvroMessageDeserializer
+transaction.topic.value.deserializer=io.pixelsdb.pixels.sink.event.deserializer.TransactionJsonMessageDeserializer
+sink.trans.batch.size=100
+
+# Sink Metrics
+sink.monitor.enable=true
+sink.monitor.port=9464
+sink.monitor.report.interval=1000
+sink.monitor.freshness.interval=1000
+
+# Interact with other rpc
+sink.rpc.enable=true
+sink.rpc.mock.delay=20
+# debezium engine config
+debezium.name=testEngine
+debezium.connector.class=io.debezium.connector.postgresql.PostgresConnector
+debezium.provide.transaction.metadata=true
+debezium.offset.storage=org.apache.kafka.connect.storage.FileOffsetBackingStore
+debezium.offset.storage.file.filename=/tmp/offsets.dat
+debezium.offset.flush.interval.ms=60000
+debezium.schema.history.internal=io.debezium.storage.file.history.FileSchemaHistory
+debezium.schema.history.internal.file.filename=/tmp/schemahistory.dat
+debezium.database.hostname=realtime-pg-2
+debezium.database.port=5432
+debezium.database.user=pixels
+debezium.database.password=pixels_realtime_crud
+debezium.database.dbname=pixels_bench
+debezium.plugin.name=pgoutput
+debezium.database.server.id=1
+debezium.schema.include.list=tpcch
+debezium.snapshot.mode=never
+debezium.key.converter=org.apache.kafka.connect.json.JsonConverter
+debezium.value.converter=org.apache.kafka.connect.json.JsonConverter
+debezium.topic.prefix=postgresql.oltp_server
+debezium.transforms=topicRouting
+debezium.transforms.topicRouting.type=org.apache.kafka.connect.transforms.RegexRouter
+debezium.transforms.topicRouting.regex=postgresql\\.oltp_server\\.public\\.(.*)
+debezium.transforms.topicRouting.replacement=postgresql.oltp_server.pixels_bench_sf10x.$1

conf/pixels-sink.flink.properties

@@ -0,0 +1,45 @@
+# engine | kafka | storage
+sink.datasource=storage
+# -1 means no limit, Only implement in retina sink mode yet
+sink.datasource.rate.limit=50000
+# Sink Config: retina | csv | proto | flink | none
+sink.mode=flink
+sink.commit.batch.size=20
+## batch or single or record, batch is recommend. record is faster, but doesn't have ACID feature
+sink.trans.mode=batch
+sink.monitor.report.enable=true
+sink.monitor.report.file=/home/ubuntu/pixels-sink/result_lance/rate_test.csv
+sink.monitor.freshness.file=/home/ubuntu/pixels-sink/result_lance/fresh_test.csv
+# trino for freshness query
+trino.url=jdbc:trino://realtime-pixels-coordinator:8080/lance/default
+trino.user=pixels
+trino.password=password
+trino.parallel=1
+# row or txn or embed
+sink.monitor.freshness.level=embed
+sink.monitor.freshness.embed.warmup=10
+sink.monitor.freshness.embed.static=false
+sink.monitor.freshness.embed.snapshot=false
+sink.monitor.freshness.embed.tablelist=stock
+sink.monitor.freshness.verbose=true
+sink.monitor.freshness.timestamp=true
+sink.storage.loop=true
+
+sink.remote.host=localhost
+sink.remote.port=29422
+sink.timeout.ms=5000
+sink.flush.interval.ms=50
+sink.flush.batch.size=10
+sink.max.retries=3
+
+## Proto Config
+# sink.proto.data=hybench1000_4
+sink.proto.data=CH10K_2
+## Flink Config
+sink.flink.server.port=9091
+
+# Sink Metrics
+sink.monitor.enable=true
+sink.monitor.port=9465
+sink.monitor.report.interval=10000
+sink.monitor.freshness.interval=1000