Pipeline support parallel decoding plugin of opengauss

.github/workflows/e2e-operation.yml

@@ -60,12 +60,12 @@ jobs:
       fail-fast: false
       matrix:
         operation: [ transaction, pipeline, showprocesslist ]
-        image: [ { type: "it.docker.mysql.version", version: "mysql:5.7" }, { type: "it.docker.postgresql.version", version: "postgres:12-alpine" }, { type: "it.docker.opengauss.version", version: "enmotech/opengauss:2.1.0" } ]
+        image: [ { type: "it.docker.mysql.version", version: "mysql:5.7" }, { type: "it.docker.postgresql.version", version: "postgres:12-alpine" }, { type: "it.docker.opengauss.version", version: "enmotech/opengauss:2.1.0,enmotech/opengauss:3.1.0" } ]
         exclude:
           - operation: showprocesslist
             image: { type: "it.docker.postgresql.version", version: "postgres:12-alpine" }
           - operation: showprocesslist
-            image: { type: "it.docker.opengauss.version", version: "enmotech/opengauss:2.1.0" }
+            image: { type: "it.docker.opengauss.version", version: "enmotech/opengauss:2.1.0,enmotech/opengauss:3.1.0" }
     steps:
       - env:
           changed_operations: ${{ needs.detect-changed-files.outputs.changed_operations }}

kernel/data-pipeline/dialect/opengauss/src/main/java/org/apache/shardingsphere/data/pipeline/opengauss/ingest/OpenGaussWALDumper.java

@@ -19,15 +19,15 @@
 
 import lombok.SneakyThrows;
 import lombok.extern.slf4j.Slf4j;
-import org.apache.shardingsphere.data.pipeline.core.ingest.dumper.incremental.IncrementalDumperContext;
 import org.apache.shardingsphere.data.pipeline.api.type.StandardPipelineDataSourceConfiguration;
-import org.apache.shardingsphere.data.pipeline.core.execute.AbstractPipelineLifecycleRunnable;
 import org.apache.shardingsphere.data.pipeline.core.channel.PipelineChannel;
+import org.apache.shardingsphere.data.pipeline.core.exception.IngestException;
+import org.apache.shardingsphere.data.pipeline.core.execute.AbstractPipelineLifecycleRunnable;
 import org.apache.shardingsphere.data.pipeline.core.ingest.dumper.incremental.IncrementalDumper;
+import org.apache.shardingsphere.data.pipeline.core.ingest.dumper.incremental.IncrementalDumperContext;
 import org.apache.shardingsphere.data.pipeline.core.ingest.position.IngestPosition;
 import org.apache.shardingsphere.data.pipeline.core.ingest.record.Record;
 import org.apache.shardingsphere.data.pipeline.core.metadata.loader.PipelineTableMetaDataLoader;
-import org.apache.shardingsphere.data.pipeline.core.exception.IngestException;
 import org.apache.shardingsphere.data.pipeline.opengauss.ingest.wal.OpenGaussLogicalReplication;
 import org.apache.shardingsphere.data.pipeline.opengauss.ingest.wal.decode.MppdbDecodingPlugin;
 import org.apache.shardingsphere.data.pipeline.opengauss.ingest.wal.decode.OpenGaussLogSequenceNumber;
@@ -46,12 +46,18 @@
 import org.opengauss.replication.PGReplicationStream;
 
 import java.nio.ByteBuffer;
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.ResultSet;
 import java.sql.SQLException;
+import java.sql.Statement;
 import java.util.Collections;
 import java.util.LinkedList;
 import java.util.List;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReference;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
 
 /**
  * WAL dumper of openGauss.
@@ -60,6 +66,10 @@
 @Slf4j
 public final class OpenGaussWALDumper extends AbstractPipelineLifecycleRunnable implements IncrementalDumper {
 
+    private static final Pattern VERSION_PATTERN = Pattern.compile("^\\(openGauss (\\d)");
+
+    private static final int DEFAULT_VERSION = 2;
+
     private final IncrementalDumperContext dumperContext;
 
     private final AtomicReference<WALPosition> walPosition;
@@ -74,6 +84,8 @@ public final class OpenGaussWALDumper extends AbstractPipelineLifecycleRunnable
 
     private List<AbstractRowEvent> rowEvents = new LinkedList<>();
 
+    private final AtomicReference<Long> currentCsn = new AtomicReference<>();
+
     public OpenGaussWALDumper(final IncrementalDumperContext dumperContext, final IngestPosition position,
                               final PipelineChannel channel, final PipelineTableMetaDataLoader metaDataLoader) {
         ShardingSpherePreconditions.checkState(StandardPipelineDataSourceConfiguration.class.equals(dumperContext.getCommonContext().getDataSourceConfig().getClass()),
@@ -110,10 +122,11 @@ protected void runBlocking() {
     @SneakyThrows(InterruptedException.class)
     private void dump() throws SQLException {
         PGReplicationStream stream = null;
+        int majorVersion = getMajorVersion();
         try (PgConnection connection = getReplicationConnectionUnwrap()) {
             stream = logicalReplication.createReplicationStream(connection, walPosition.get().getLogSequenceNumber(),
-                    OpenGaussIngestPositionManager.getUniqueSlotName(connection, dumperContext.getJobId()));
-            DecodingPlugin decodingPlugin = new MppdbDecodingPlugin(new OpenGaussTimestampUtils(connection.getTimestampUtils()), decodeWithTX);
+                    OpenGaussIngestPositionManager.getUniqueSlotName(connection, dumperContext.getJobId()), majorVersion);
+            DecodingPlugin decodingPlugin = new MppdbDecodingPlugin(new OpenGaussTimestampUtils(connection.getTimestampUtils()), decodeWithTX, majorVersion >= 3);
             while (isRunning()) {
                 ByteBuffer message = stream.readPending();
                 if (null == message) {
@@ -122,7 +135,7 @@ private void dump() throws SQLException {
                 }
                 AbstractWALEvent event = decodingPlugin.decode(message, new OpenGaussLogSequenceNumber(stream.getLastReceiveLSN()));
                 if (decodeWithTX) {
-                    processEventWithTX(event);
+                    processEventWithTX(event, majorVersion);
                 } else {
                     processEventIgnoreTX(event);
                 }
@@ -138,28 +151,61 @@ private void dump() throws SQLException {
         }
     }
 
+    private int getMajorVersion() throws SQLException {
+        StandardPipelineDataSourceConfiguration dataSourceConfig = (StandardPipelineDataSourceConfiguration) dumperContext.getCommonContext().getDataSourceConfig();
+        try (
+                Connection connection = DriverManager.getConnection(dataSourceConfig.getUrl(), dataSourceConfig.getUsername(), dataSourceConfig.getPassword());
+                Statement statement = connection.createStatement();
+                ResultSet resultSet = statement.executeQuery("SELECT version()")) {
+            resultSet.next();
+            String versionText = resultSet.getString(1);
+            return parseMajorVersion(versionText);
+        }
+    }
+
+    private int parseMajorVersion(final String versionText) {
+        Matcher matcher = VERSION_PATTERN.matcher(versionText);
+        boolean isFind = matcher.find();
+        log.info("openGauss major version={}, `select version()`={}", isFind ? matcher.group(1) : DEFAULT_VERSION, versionText);
+        if (isFind) {
+            return Integer.parseInt(matcher.group(1));
+        }
+        return DEFAULT_VERSION;
+    }
+
     private PgConnection getReplicationConnectionUnwrap() throws SQLException {
         return logicalReplication.createConnection((StandardPipelineDataSourceConfiguration) dumperContext.getCommonContext().getDataSourceConfig()).unwrap(PgConnection.class);
     }
 
-    private void processEventWithTX(final AbstractWALEvent event) {
+    private void processEventWithTX(final AbstractWALEvent event, final int majorVersion) {
         if (event instanceof BeginTXEvent) {
+            if (majorVersion < 3) {
+                return;
+            }
+            if (!rowEvents.isEmpty()) {
+                log.warn("Commit event parse have problem, there still has uncommitted row events size={}, ", rowEvents.size());
+            }
+            currentCsn.set(((BeginTXEvent) event).getCsn());
             return;
         }
         if (event instanceof AbstractRowEvent) {
-            rowEvents.add((AbstractRowEvent) event);
+            AbstractRowEvent rowEvent = (AbstractRowEvent) event;
+            rowEvent.setCsn(currentCsn.get());
+            rowEvents.add(rowEvent);
             return;
         }
         if (event instanceof CommitTXEvent) {
-            Long csn = ((CommitTXEvent) event).getCsn();
             List<Record> records = new LinkedList<>();
             for (AbstractRowEvent each : rowEvents) {
-                each.setCsn(csn);
+                if (majorVersion < 3) {
+                    each.setCsn(((CommitTXEvent) event).getCsn());
+                }
                 records.add(walEventConverter.convert(each));
             }
             records.add(walEventConverter.convert(event));
             channel.push(records);
             rowEvents = new LinkedList<>();
+            currentCsn.set(null);
         }
     }
 

kernel/data-pipeline/dialect/opengauss/src/main/java/org/apache/shardingsphere/data/pipeline/opengauss/ingest/wal/OpenGaussLogicalReplication.java

@@ -28,6 +28,7 @@
 import org.opengauss.jdbc.PgConnection;
 import org.opengauss.replication.LogSequenceNumber;
 import org.opengauss.replication.PGReplicationStream;
+import org.opengauss.replication.fluent.logical.ChainedLogicalStreamBuilder;
 
 import java.sql.Connection;
 import java.sql.DriverManager;
@@ -86,17 +87,26 @@ private Connection tryConnectingToHAPort(final String jdbcUrl, final Properties
      * @param connection connection
      * @param startPosition start position
      * @param slotName slot name
+     * @param majorVersion version
      * @return replication stream
      * @throws SQLException SQL exception
      */
-    public PGReplicationStream createReplicationStream(final PgConnection connection, final BaseLogSequenceNumber startPosition, final String slotName) throws SQLException {
-        return connection.getReplicationAPI()
+    public PGReplicationStream createReplicationStream(final PgConnection connection, final BaseLogSequenceNumber startPosition, final String slotName,
+                                                       final int majorVersion) throws SQLException {
+        ChainedLogicalStreamBuilder logicalStreamBuilder = connection.getReplicationAPI()
                 .replicationStream()
                 .logical()
                 .withSlotName(slotName)
                 .withSlotOption("include-xids", true)
                 .withSlotOption("skip-empty-xacts", true)
-                .withStartPosition((LogSequenceNumber) startPosition.get())
+                .withStartPosition((LogSequenceNumber) startPosition.get());
+        if (majorVersion < 3) {
+            return logicalStreamBuilder.start();
+        }
+        return logicalStreamBuilder
+                .withSlotOption("parallel-decode-num", 10)
+                .withSlotOption("decode-style", "j")
+                .withSlotOption("sending-batch", 0)
                 .start();
     }
 }

kernel/data-pipeline/dialect/opengauss/src/main/java/org/apache/shardingsphere/data/pipeline/opengauss/ingest/wal/decode/MppdbDecodingPlugin.java

@@ -57,9 +57,7 @@ public final class MppdbDecodingPlugin implements DecodingPlugin {
 
     private final boolean decodeWithTX;
 
-    public MppdbDecodingPlugin(final BaseTimestampUtils timestampUtils) {
-        this(timestampUtils, false);
-    }
+    private final boolean decodeParallelly;
 
     @Override
     public AbstractWALEvent decode(final ByteBuffer data, final BaseLogSequenceNumber logSequenceNumber) {
@@ -77,10 +75,18 @@ public AbstractWALEvent decode(final ByteBuffer data, final BaseLogSequenceNumbe
     }
 
     private AbstractWALEvent decodeDataWithTX(final String dataText) {
+        if (decodeParallelly) {
+            return decodeParallelly(dataText);
+        } else {
+            return decodeSerially(dataText);
+        }
+    }
+
+    private AbstractWALEvent decodeSerially(final String dataText) {
         AbstractWALEvent result = new PlaceholderEvent();
         if (dataText.startsWith("BEGIN")) {
             int beginIndex = dataText.indexOf("BEGIN") + "BEGIN".length() + 1;
-            result = new BeginTXEvent(Long.parseLong(dataText.substring(beginIndex)));
+            result = new BeginTXEvent(Long.parseLong(dataText.substring(beginIndex)), null);
         } else if (dataText.startsWith("COMMIT")) {
             int commitBeginIndex = dataText.indexOf("COMMIT") + "COMMIT".length() + 1;
             int csnBeginIndex = dataText.indexOf("CSN") + "CSN".length() + 1;
@@ -91,6 +97,22 @@ private AbstractWALEvent decodeDataWithTX(final String dataText) {
         return result;
     }
 
+    private AbstractWALEvent decodeParallelly(final String dataText) {
+        AbstractWALEvent result = new PlaceholderEvent();
+        if (dataText.startsWith("BEGIN")) {
+            int beginIndex = dataText.indexOf("CSN:") + "CSN:".length() + 1;
+            int firstLsnIndex = dataText.indexOf("first_lsn");
+            long csn = firstLsnIndex > 0 ? Long.parseLong(dataText.substring(beginIndex, firstLsnIndex - 1)) : 0L;
+            result = new BeginTXEvent(null, csn);
+        } else if (dataText.startsWith("commit") || dataText.startsWith("COMMIT")) {
+            int beginIndex = dataText.indexOf("xid:") + "xid:".length() + 1;
+            result = new CommitTXEvent(Long.parseLong(dataText.substring(beginIndex)), null);
+        } else if (dataText.startsWith("{")) {
+            result = readTableEvent(dataText);
+        }
+        return result;
+    }
+
     private AbstractWALEvent decodeDataIgnoreTX(final String dataText) {
         return dataText.startsWith("{") ? readTableEvent(dataText) : new PlaceholderEvent();
     }

kernel/data-pipeline/dialect/opengauss/src/test/java/org/apache/shardingsphere/data/pipeline/opengauss/ingest/OpenGaussWALDumperTest.java

@@ -0,0 +1,41 @@
+/*
+ * 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.opengauss.ingest;
+
+import org.apache.shardingsphere.infra.util.reflection.ReflectionUtils;
+import org.junit.jupiter.api.Test;
+
+import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.mockito.Mockito.mock;
+
+class OpenGaussWALDumperTest {
+
+    @Test
+    void assertGetVersion() throws NoSuchMethodException {
+        OpenGaussWALDumper dumper = mock(OpenGaussWALDumper.class);
+        int version = ReflectionUtils.invokeMethod(OpenGaussWALDumper.class.getDeclaredMethod("parseMajorVersion", String.class), dumper,
+                "(openGauss 3.1.0 build ) compiled at 2023-02-17 16:13:51 commit 0 last mr   on x86_64-unknown-linux-gnu, compiled by g++ (GCC) 7.3.0, 64-bit");
+        assertThat(version, is(3));
+        OpenGaussWALDumper mock = mock(OpenGaussWALDumper.class);
+        version = ReflectionUtils.invokeMethod(OpenGaussWALDumper.class.getDeclaredMethod("parseMajorVersion", String.class), mock, "(openGauss 5.0.1 build )");
+        assertThat(version, is(5));
+        version = ReflectionUtils.invokeMethod(OpenGaussWALDumper.class.getDeclaredMethod("parseMajorVersion", String.class), mock, "not match");
+        assertThat(version, is(2));
+    }
+}