You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@shardingsphere.apache.org by GitBox <gi...@apache.org> on 2022/12/06 03:30:42 UTC
[GitHub] [shardingsphere] sandynz commented on a diff in pull request #22685: Adding xid and csn parse for openGauss's logical replication
sandynz commented on code in PR #22685:
URL: https://github.com/apache/shardingsphere/pull/22685#discussion_r1040395649
##########
kernel/data-pipeline/dialect/opengauss/src/main/java/org/apache/shardingsphere/data/pipeline/opengauss/ingest/wal/decode/MppdbDecodingPlugin.java:
##########
@@ -59,30 +68,46 @@ public final class MppdbDecodingPlugin implements DecodingPlugin {
private final BaseTimestampUtils timestampUtils;
+ private final boolean decodeWithXid;
+
+ public MppdbDecodingPlugin(final BaseTimestampUtils timestampUtils) {
+ this.timestampUtils = timestampUtils;
+ decodeWithXid = false;
+ }
+
@Override
public AbstractWALEvent decode(final ByteBuffer data, final BaseLogSequenceNumber logSequenceNumber) {
AbstractWALEvent result;
- char eventType = readOneChar(data);
- result = '{' == eventType ? readTableEvent(readMppData(data)) : new PlaceholderEvent();
+ String dataText = StandardCharsets.UTF_8.decode(data).toString();
Review Comment:
Is character set always UTF-8?
##########
kernel/data-pipeline/dialect/opengauss/src/main/java/org/apache/shardingsphere/data/pipeline/opengauss/ingest/OpenGaussWALDumper.java:
##########
@@ -99,6 +113,23 @@ private PgConnection getReplicationConnectionUnwrap() throws SQLException {
return logicalReplication.createConnection((StandardPipelineDataSourceConfiguration) dumperConfig.getDataSourceConfig()).unwrap(PgConnection.class);
}
+ private void processEventWithXid(final AbstractWALEvent event) {
+ if (event instanceof AbstractRowEvent) {
+ rowEvents.add((AbstractRowEvent) event);
+ return;
+ }
+ if (event instanceof BeginXidEvent) {
+ rowEvents.clear();
+ }
+ if (event instanceof CommitXidEvent) {
+ for (AbstractRowEvent each : rowEvents) {
+ // TODO if multiple threads are pushing data in a channel at the same time, the order of the data will be incorrect
+ channel.pushRecord(walEventConverter.convert(each));
+ }
+ }
+ channel.pushRecord(walEventConverter.convert(event));
Review Comment:
Is it required to put begin and commit TX events into channel?
##########
kernel/data-pipeline/dialect/postgresql/src/test/java/org/apache/shardingsphere/data/pipeline/postgresql/ingest/wal/WALEventConverterTest.java:
##########
@@ -89,6 +94,24 @@ private void initTableData(final DumperConfiguration dumperConfig) {
}
}
+ @Test
+ public void assertConvertBeginXidEvent() {
+ BeginXidEvent beginXidEvent = new BeginXidEvent(100);
+ beginXidEvent.setLogSequenceNumber(new PostgreSQLLogSequenceNumber(LogSequenceNumber.valueOf("0/0")));
+ Record record = walEventConverter.convert(beginXidEvent);
+ assertTrue(record instanceof PlaceholderRecord);
+ assertThat(((WALPosition) record.getPosition()).getLogSequenceNumber().asLong(), is(0L));
+ }
+
+ @Test
+ public void assertConvertCommitXidEvent() {
+ CommitXidEvent commitXidEvent = new CommitXidEvent(1, 3468L);
+ commitXidEvent.setLogSequenceNumber(new PostgreSQLLogSequenceNumber(LogSequenceNumber.valueOf("0/0")));
Review Comment:
Could we use more meaningful LSN? Since the default LSN value might be 0 too (in openGauss driver LogSequenceNumber), the unit test might not take effect
##########
kernel/data-pipeline/dialect/postgresql/src/main/java/org/apache/shardingsphere/data/pipeline/postgresql/ingest/wal/event/CommitXidEvent.java:
##########
@@ -0,0 +1,33 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.shardingsphere.data.pipeline.postgresql.ingest.wal.event;
+
+import lombok.Getter;
+import lombok.RequiredArgsConstructor;
+
+/**
+ * Commit xid event.
+ */
+@RequiredArgsConstructor
+@Getter
+public final class CommitXidEvent extends AbstractWALEvent {
+
+ private final long xid;
+
+ private final Long csn;
Review Comment:
Why type of `xid` is `long`, but type of `csn` is `Long`
##########
kernel/data-pipeline/dialect/opengauss/src/main/java/org/apache/shardingsphere/data/pipeline/opengauss/ingest/OpenGaussWALDumper.java:
##########
@@ -33,14 +33,19 @@
import org.apache.shardingsphere.data.pipeline.postgresql.ingest.wal.WALEventConverter;
import org.apache.shardingsphere.data.pipeline.postgresql.ingest.wal.WALPosition;
import org.apache.shardingsphere.data.pipeline.postgresql.ingest.wal.decode.DecodingPlugin;
+import org.apache.shardingsphere.data.pipeline.postgresql.ingest.wal.event.AbstractRowEvent;
import org.apache.shardingsphere.data.pipeline.postgresql.ingest.wal.event.AbstractWALEvent;
+import org.apache.shardingsphere.data.pipeline.postgresql.ingest.wal.event.BeginXidEvent;
+import org.apache.shardingsphere.data.pipeline.postgresql.ingest.wal.event.CommitXidEvent;
Review Comment:
`Xid` could be `TX` in class name, means begin or commit transaction.
And also in variable name ane method name.
##########
kernel/data-pipeline/dialect/opengauss/src/main/java/org/apache/shardingsphere/data/pipeline/opengauss/ingest/OpenGaussWALDumper.java:
##########
@@ -99,6 +113,23 @@ private PgConnection getReplicationConnectionUnwrap() throws SQLException {
return logicalReplication.createConnection((StandardPipelineDataSourceConfiguration) dumperConfig.getDataSourceConfig()).unwrap(PgConnection.class);
}
+ private void processEventWithXid(final AbstractWALEvent event) {
+ if (event instanceof AbstractRowEvent) {
+ rowEvents.add((AbstractRowEvent) event);
+ return;
+ }
+ if (event instanceof BeginXidEvent) {
+ rowEvents.clear();
+ }
+ if (event instanceof CommitXidEvent) {
+ for (AbstractRowEvent each : rowEvents) {
+ // TODO if multiple threads are pushing data in a channel at the same time, the order of the data will be incorrect
Review Comment:
Every `channel` instance could be used only by current OpenGaussWALDumper instance, it's the current logic. Then use merge-order on diferent channels.
To reuse more current code.
##########
kernel/data-pipeline/dialect/opengauss/src/main/java/org/apache/shardingsphere/data/pipeline/opengauss/ingest/wal/decode/MppdbDecodingPlugin.java:
##########
@@ -59,30 +68,46 @@ public final class MppdbDecodingPlugin implements DecodingPlugin {
private final BaseTimestampUtils timestampUtils;
+ private final boolean decodeWithXid;
+
+ public MppdbDecodingPlugin(final BaseTimestampUtils timestampUtils) {
+ this.timestampUtils = timestampUtils;
+ decodeWithXid = false;
+ }
+
@Override
public AbstractWALEvent decode(final ByteBuffer data, final BaseLogSequenceNumber logSequenceNumber) {
AbstractWALEvent result;
- char eventType = readOneChar(data);
- result = '{' == eventType ? readTableEvent(readMppData(data)) : new PlaceholderEvent();
+ String dataText = StandardCharsets.UTF_8.decode(data).toString();
+ if (decodeWithXid) {
+ result = decodeDataWithXid(dataText);
+ } else {
+ result = decodeDataIgnoreXid(dataText);
+ }
result.setLogSequenceNumber(logSequenceNumber);
return result;
}
- private char readOneChar(final ByteBuffer data) {
- return (char) data.get();
+ private AbstractWALEvent decodeDataWithXid(final String dataText) {
+ AbstractWALEvent result;
+ if (dataText.startsWith("{")) {
+ result = readTableEvent(dataText);
+ return result;
+ }
+ Matcher beginXidMatcher = PATTERN_BEGIN_XID.matcher(dataText);
+ Matcher commitXidMatcher = PATTERN_COMMIT_XID.matcher(dataText);
Review Comment:
It might hurt performance when pattern match frequently here, it's better to use streaming parsing at one time
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@shardingsphere.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org