Refactor event ingestion into conversion + pipeline queues with MySQL source support by gengdy1545 · Pull Request #21 · AntiO2/pixels-sink

gengdy1545 · 2026-07-30T06:36:19Z

Summary

Refactor sink ingestion so sources own protocol conversion and publish only canonical RowChangeEvent / TransactionMetadata into pipeline queues.
Replace the provider hierarchy with pipeline-owned BlockingBoundedQueue for backpressure and ordered delivery.
Add MySQL Debezium source support (adapter + config) alongside Postgres, and unify logging on Log4j2 to remove competing bindings.

Changes

Source conversion & pipelines

Move Kafka / Engine / Storage deserialization and Debezium envelope handling into conversion (JSON/Avro/Struct converters, envelope normalizer, source adapters).
Introduce TablePipeline / TransactionPipeline / TablePipelineManager so processors consume canonical events only.
Add KafkaRowSource / KafkaTransactionSource and rework PixelsDebeziumConsumer to convert at the source boundary.
Add MySQL source adapter and conf/pixels-sink.mysql.properties.

Provider removal

Remove EventProvider and per-source provider subclasses.
Pipelines own BlockingBoundedQueue for bounded, blocking handoff between sources and processors.
Update docs (overview, usage, configuration) to match the new architecture.

Logging

Standardize on Log4j2; exclude competing SLF4J/Log4j bindings in pom.xml.
Refresh log4j2.properties / jvm.conf and drop unused logging.properties.

…Queue

AntiO2 · 2026-07-30T07:02:07Z

+    public static RecordType classify(SourceRecord sourceRecord, String transactionTopic)
+    {
+        if (sourceRecord == null || sourceRecord.value() == null)
+        {
+            return RecordType.TOMBSTONE;
+        }
+        if (!(sourceRecord.value() instanceof Struct value))
+        {
+            return RecordType.UNKNOWN_CONTROL;
+        }
+
+        if (transactionTopic != null && transactionTopic.equals(sourceRecord.topic()))
+        {
+            String status = DebeziumRecordUtil.getStringSafely(value, "status");
+            String id = DebeziumRecordUtil.getStringSafely(value, "id");
+            return (status.equals("BEGIN") || status.equals("END")) && !id.isBlank()
+                    ? RecordType.TRANSACTION
+                    : RecordType.UNKNOWN_CONTROL;
+        }
+
+        return isRowChange(value) ? RecordType.ROW : RecordType.UNKNOWN_CONTROL;
+    }
+
+    private static boolean isRowChange(Struct value)
+    {
+        String op = DebeziumRecordUtil.getStringSafely(value, "op").toLowerCase(Locale.ROOT);
+        if (!(op.equals("c") || op.equals("u") || op.equals("d") || op.equals("r")))
+        {
+            return false;
+        }
+
+        Object sourceObject = DebeziumRecordUtil.getFieldSafely(value, "source");
+        if (!(sourceObject instanceof Struct source) ||
+                DebeziumRecordUtil.getStringSafely(source, "db").isBlank() ||
+                DebeziumRecordUtil.getStringSafely(source, "table").isBlank())
+        {
+            return false;
+        }
+
+        Object before = DebeziumRecordUtil.getFieldSafely(value, "before");
+        Object after = DebeziumRecordUtil.getFieldSafely(value, "after");
+        return switch (op)
+        {
+            case "c", "r" -> after instanceof Struct;
+            case "u" -> before instanceof Struct && after instanceof Struct;
+            case "d" -> before instanceof Struct;
+            default -> false;
+        };
+    }
+
+    private void logSourceRecord(SourceRecord record, RecordType recordType)
+    {
+        if (!LOGGER.isDebugEnabled())
+        {
+            return;
+        }
+        Struct value = record.value() instanceof Struct struct ? struct : null;
+        Struct source = value == null ? null :
+                asStruct(DebeziumRecordUtil.getFieldSafely(value, "source"));
+        Struct transaction = value == null ? null :
+                asStruct(DebeziumRecordUtil.getFieldSafely(value, "transaction"));
+        String rawTransactionId = recordType == RecordType.TRANSACTION
+                ? DebeziumRecordUtil.getStringSafely(value, "id")
+                : DebeziumRecordUtil.getStringSafely(transaction, "id");
+        String canonicalTransactionId =
+                connectorAdapter.normalizeTransactionId(rawTransactionId);
+
+        LOGGER.debug("Debezium SourceRecord topic={}, sourcePartition={}, sourceOffset={}, " +
+                        "keySchema={}, key={}, valueSchema={}, category={}, transaction.id={}, " +
+                        "source.gtid={}, source.file={}, source.pos={}, source.row={}",
+                record.topic(), record.sourcePartition(), record.sourceOffset(),
+                schemaName(record.keySchema()), record.key(), schemaName(record.valueSchema()),
+                recordType, canonicalTransactionId,
+                DebeziumRecordUtil.getStringSafely(source, "gtid"),
+                DebeziumRecordUtil.getStringSafely(source, "file"),
+                DebeziumRecordUtil.getStringSafely(source, "pos"),
+                DebeziumRecordUtil.getStringSafely(source, "row"));
+    }
+
+    private static Struct asStruct(Object value)
+    {
+        return value instanceof Struct struct ? struct : null;
+    }
+
+    private static String schemaName(org.apache.kafka.connect.data.Schema schema)


这里应该也要保留流水线。

AntiO2 · 2026-07-30T07:05:42Z

这个是测试什么的

AntiO2 · 2026-07-30T07:08:01Z

        return (dateToString);
    }

-    public static String convertTimestampToString(Date date)


连MySQL和PG确认下CDC能不能正确解析。我记得Debezium Date、Time、Timestamp，包括不同精度的时间，有点复杂。

AntiO2 · 2026-07-30T07:10:37Z

还有测试一下Decimal的解析

AntiO2 · 2026-07-30T07:22:58Z

@codex review

chatgpt-codex-connector

💡 Codex Review

Here are some automated review suggestions for this pull request.

Reviewed commit: aa78cf9286

ℹ️ About Codex in GitHub

Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you

Open a pull request for review
Mark a draft as ready
Comment "@codex review".

If Codex has suggestions, it will comment; otherwise it will react with 👍.

Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".

chatgpt-codex-connector · 2026-07-30T07:26:58Z

+        tablePipelineManager.close();
+        transactionPipeline.close();


Drain pipeline queues before closing storage playback

When sink.datasource=storage runs without looping, clean() waits only for the file-consumer threads, which merely enqueue converted records into the new table and transaction pipelines. These calls then close queues that explicitly discard pending values, so any backlog not yet written when file playback finishes is silently lost; wait for the pipeline processors to drain before closing them.

Useful? React with 👍 / 👎.

chatgpt-codex-connector · 2026-07-30T07:26:58Z

+            if (adapter == null)
+            {
+                throw new IllegalStateException(
+                        "No Debezium source adapter configured for " + topic);


Configure the adapter before decoding transaction metadata

With the packaged Kafka configuration (src/main/resources/pixels-sink.properties), debezium.connector.class is absent while this JSON transaction deserializer now requires an adapter. The first transaction record therefore throws here, and KafkaTransactionSource catches that exception outside its polling loop and terminates permanently, preventing transactional rows from being committed. Either supply/infer the adapter for Kafka transaction messages or reject the incomplete configuration at startup.

Useful? React with 👍 / 👎.

…optional

dannygeng added 4 commits July 20, 2026 17:39

feat: Refactor sink source conversion and event pipelines

7c4d525

refactor: Replace event providers with pipeline-owned BlockingBounded…

aebc87c

…Queue

fix: Unify logging on Log4j2 and eliminate competing bindings

d0bff50

feat: Add debug logs for Retina batch flush send and ack

aa78cf9

AntiO2 reviewed Jul 30, 2026

View reviewed changes

Comment thread pom.xml

AntiO2 reviewed Jul 30, 2026

View reviewed changes

Comment thread src/main/java/io/pixelsdb/pixels/sink/source/storage/LegacySinkStorageSource.java

AntiO2 reviewed Jul 30, 2026

View reviewed changes

Comment thread src/main/java/io/pixelsdb/pixels/sink/source/storage/AbstractSinkStorageSource.java

AntiO2 reviewed Jul 30, 2026

View reviewed changes

chatgpt-codex-connector Bot reviewed Jul 30, 2026

View reviewed changes

fix: Represent null Debezium columns as empty bytes instead of isNull

963f836

gengdy1545 mentioned this pull request Jul 30, 2026

[Issue #13] Integrate Retina and add Debezium Engine as a data source. pixelsdb/pixels-sink#14

Open

gengdy1545 added 3 commits July 31, 2026 10:20

refactor: rename freshness config to sink.query.* and make Hive JDBC …

846166b

…optional

feat: Add sink.storage.mode to select stream or memory storage source

2b9ae21

perf: parallelize storage source decoding and add close/abort shutdown

05e3682

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Refactor event ingestion into conversion + pipeline queues with MySQL source support - #21

Refactor event ingestion into conversion + pipeline queues with MySQL source support#21
gengdy1545 wants to merge 8 commits into
AntiO2:masterfrom
gengdy1545:feature/mysqlSource

gengdy1545 commented Jul 30, 2026

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

AntiO2 Jul 30, 2026

Uh oh!

AntiO2 Jul 30, 2026

Uh oh!

AntiO2 Jul 30, 2026

Uh oh!

AntiO2 commented Jul 30, 2026

Uh oh!

AntiO2 commented Jul 30, 2026

Uh oh!

chatgpt-codex-connector Bot left a comment

Uh oh!

chatgpt-codex-connector Bot Jul 30, 2026

Uh oh!

chatgpt-codex-connector Bot Jul 30, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

Conversation

gengdy1545 commented Jul 30, 2026

Summary

Changes

Source conversion & pipelines

Provider removal

Logging

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

AntiO2 Jul 30, 2026

Choose a reason for hiding this comment

Uh oh!

AntiO2 Jul 30, 2026

Choose a reason for hiding this comment

Uh oh!

AntiO2 Jul 30, 2026

Choose a reason for hiding this comment

Uh oh!

AntiO2 commented Jul 30, 2026

Uh oh!

AntiO2 commented Jul 30, 2026

Uh oh!

chatgpt-codex-connector Bot left a comment

Choose a reason for hiding this comment

💡 Codex Review

Uh oh!

chatgpt-codex-connector Bot Jul 30, 2026

Choose a reason for hiding this comment

Uh oh!

chatgpt-codex-connector Bot Jul 30, 2026

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants