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 02:34:39 UTC
[GitHub] [shardingsphere] azexcy opened a new pull request, #22685: Adding xid and csn parse for openGauss's logical replication
azexcy opened a new pull request, #22685:
URL: https://github.com/apache/shardingsphere/pull/22685
Changes proposed in this pull request:
- Adding xid and csn parse for openGauss's logical replication
- Add unit test
---
Before committing this PR, I'm sure that I have checked the following options:
- [ ] My code follows the [code of conduct](https://shardingsphere.apache.org/community/en/involved/conduct/code/) of this project.
- [ ] I have self-reviewed the commit code.
- [ ] I have (or in comment I request) added corresponding labels for the pull request.
- [ ] I have passed maven check locally : `./mvnw clean install -B -T1C -Dmaven.javadoc.skip -Dmaven.jacoco.skip -e`.
- [ ] I have made corresponding changes to the documentation.
- [ ] I have added corresponding unit tests for my changes.
--
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
[GitHub] [shardingsphere] sandynz commented on a diff in pull request #22685: Adding xid and csn parse for openGauss's logical replication
Posted by GitBox <gi...@apache.org>.
sandynz commented on code in PR #22685:
URL: https://github.com/apache/shardingsphere/pull/22685#discussion_r1040607440
##########
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:
OK. Since TX events are ignored before, if it's added this time, then it might need to update related events consumption
--
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
[GitHub] [shardingsphere] azexcy commented on a diff in pull request #22685: Adding xid and csn parse for openGauss's logical replication
Posted by GitBox <gi...@apache.org>.
azexcy commented on code in PR #22685:
URL: https://github.com/apache/shardingsphere/pull/22685#discussion_r1040622074
##########
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:
org.opengauss.core.v3.QueryExecutorImpl#startCopy
```
Utils.encodeUTF8(statementName);
```
seem the client always use the utf-8, and tested succeed at local when contain chinese characters
--
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
[GitHub] [shardingsphere] sandynz commented on a diff in pull request #22685: Adding xid and csn parse for openGauss's logical replication
Posted by GitBox <gi...@apache.org>.
sandynz commented on code in PR #22685:
URL: https://github.com/apache/shardingsphere/pull/22685#discussion_r1040525102
##########
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:
OK
--
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
[GitHub] [shardingsphere] codecov-commenter commented on pull request #22685: Adding xid and csn parse for openGauss's logical replication
Posted by GitBox <gi...@apache.org>.
codecov-commenter commented on PR #22685:
URL: https://github.com/apache/shardingsphere/pull/22685#issuecomment-1338670087
# [Codecov](https://codecov.io/gh/apache/shardingsphere/pull/22685?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) Report
> Merging [#22685](https://codecov.io/gh/apache/shardingsphere/pull/22685?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (378b5a8) into [master](https://codecov.io/gh/apache/shardingsphere/commit/3f2bc933be88f4ba97c426b23ddb723e1b5f4c11?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (3f2bc93) will **decrease** coverage by `0.13%`.
> The diff coverage is `22.17%`.
```diff
@@ Coverage Diff @@
## master #22685 +/- ##
============================================
- Coverage 49.49% 49.36% -0.14%
- Complexity 2439 2441 +2
============================================
Files 4094 4108 +14
Lines 57204 57461 +257
Branches 9800 9848 +48
============================================
+ Hits 28315 28367 +52
- Misses 26452 26649 +197
- Partials 2437 2445 +8
```
| [Impacted Files](https://codecov.io/gh/apache/shardingsphere/pull/22685?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | Coverage Δ | |
|---|---|---|
| [...dingsphere/data/pipeline/cdc/client/CDCClient.java](https://codecov.io/gh/apache/shardingsphere/pull/22685/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-a2VybmVsL2RhdGEtcGlwZWxpbmUvY2RjL2NsaWVudC9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvc2hhcmRpbmdzcGhlcmUvZGF0YS9waXBlbGluZS9jZGMvY2xpZW50L0NEQ0NsaWVudC5qYXZh) | `0.00% <0.00%> (ø)` | |
| [...peline/cdc/client/handler/LoginRequestHandler.java](https://codecov.io/gh/apache/shardingsphere/pull/22685/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-a2VybmVsL2RhdGEtcGlwZWxpbmUvY2RjL2NsaWVudC9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvc2hhcmRpbmdzcGhlcmUvZGF0YS9waXBlbGluZS9jZGMvY2xpZW50L2hhbmRsZXIvTG9naW5SZXF1ZXN0SGFuZGxlci5qYXZh) | `0.00% <0.00%> (ø)` | |
| [...cdc/client/handler/SubscriptionRequestHandler.java](https://codecov.io/gh/apache/shardingsphere/pull/22685/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-a2VybmVsL2RhdGEtcGlwZWxpbmUvY2RjL2NsaWVudC9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvc2hhcmRpbmdzcGhlcmUvZGF0YS9waXBlbGluZS9jZGMvY2xpZW50L2hhbmRsZXIvU3Vic2NyaXB0aW9uUmVxdWVzdEhhbmRsZXIuamF2YQ==) | `0.00% <0.00%> (ø)` | |
| [...e/data/pipeline/cdc/client/util/RequestIdUtil.java](https://codecov.io/gh/apache/shardingsphere/pull/22685/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-a2VybmVsL2RhdGEtcGlwZWxpbmUvY2RjL2NsaWVudC9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvc2hhcmRpbmdzcGhlcmUvZGF0YS9waXBlbGluZS9jZGMvY2xpZW50L3V0aWwvUmVxdWVzdElkVXRpbC5qYXZh) | `0.00% <0.00%> (ø)` | |
| [...ata/pipeline/cdc/constant/CDCConnectionStatus.java](https://codecov.io/gh/apache/shardingsphere/pull/22685/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-a2VybmVsL2RhdGEtcGlwZWxpbmUvY2RjL2NvcmUvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL3NoYXJkaW5nc3BoZXJlL2RhdGEvcGlwZWxpbmUvY2RjL2NvbnN0YW50L0NEQ0Nvbm5lY3Rpb25TdGF0dXMuamF2YQ==) | `0.00% <0.00%> (ø)` | |
| [...a/pipeline/cdc/generator/CDCResponseGenerator.java](https://codecov.io/gh/apache/shardingsphere/pull/22685/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-a2VybmVsL2RhdGEtcGlwZWxpbmUvY2RjL2NvcmUvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL3NoYXJkaW5nc3BoZXJlL2RhdGEvcGlwZWxpbmUvY2RjL2dlbmVyYXRvci9DRENSZXNwb25zZUdlbmVyYXRvci5qYXZh) | `0.00% <0.00%> (ø)` | |
| [...shardingsphere/data/pipeline/cdc/rule/CDCRule.java](https://codecov.io/gh/apache/shardingsphere/pull/22685/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-a2VybmVsL2RhdGEtcGlwZWxpbmUvY2RjL2NvcmUvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL3NoYXJkaW5nc3BoZXJlL2RhdGEvcGlwZWxpbmUvY2RjL3J1bGUvQ0RDUnVsZS5qYXZh) | `0.00% <0.00%> (ø)` | |
| [...data/pipeline/cdc/rule/builder/CDCRuleBuilder.java](https://codecov.io/gh/apache/shardingsphere/pull/22685/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-a2VybmVsL2RhdGEtcGlwZWxpbmUvY2RjL2NvcmUvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL3NoYXJkaW5nc3BoZXJlL2RhdGEvcGlwZWxpbmUvY2RjL3J1bGUvYnVpbGRlci9DRENSdWxlQnVpbGRlci5qYXZh) | `0.00% <0.00%> (ø)` | |
| [...line/cdc/yaml/config/YamlCDCRuleConfiguration.java](https://codecov.io/gh/apache/shardingsphere/pull/22685/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-a2VybmVsL2RhdGEtcGlwZWxpbmUvY2RjL2NvcmUvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL3NoYXJkaW5nc3BoZXJlL2RhdGEvcGlwZWxpbmUvY2RjL3lhbWwvY29uZmlnL1lhbWxDRENSdWxlQ29uZmlndXJhdGlvbi5qYXZh) | `0.00% <0.00%> (ø)` | |
| [.../yaml/swapper/YamlCDCRuleConfigurationSwapper.java](https://codecov.io/gh/apache/shardingsphere/pull/22685/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-a2VybmVsL2RhdGEtcGlwZWxpbmUvY2RjL2NvcmUvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL3NoYXJkaW5nc3BoZXJlL2RhdGEvcGlwZWxpbmUvY2RjL3lhbWwvc3dhcHBlci9ZYW1sQ0RDUnVsZUNvbmZpZ3VyYXRpb25Td2FwcGVyLmphdmE=) | `0.00% <0.00%> (ø)` | |
| ... and [14 more](https://codecov.io/gh/apache/shardingsphere/pull/22685/diff?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | |
:mega: We’re building smart automated test selection to slash your CI/CD build times. [Learn more](https://about.codecov.io/iterative-testing/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation)
--
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
[GitHub] [shardingsphere] azexcy commented on a diff in pull request #22685: Adding xid and csn parse for openGauss's logical replication
Posted by GitBox <gi...@apache.org>.
azexcy commented on code in PR #22685:
URL: https://github.com/apache/shardingsphere/pull/22685#discussion_r1040622074
##########
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:
org.opengauss.core.v3.QueryExecutorImpl#startCopy
--
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
[GitHub] [shardingsphere] sandynz commented on a diff in pull request #22685: Adding xid and csn parse for openGauss's logical replication
Posted by GitBox <gi...@apache.org>.
sandynz commented on code in PR #22685:
URL: https://github.com/apache/shardingsphere/pull/22685#discussion_r1040618686
##########
kernel/data-pipeline/dialect/opengauss/src/main/java/org/apache/shardingsphere/data/pipeline/opengauss/ingest/wal/decode/MppdbDecodingPlugin.java:
##########
@@ -59,30 +61,45 @@ public final class MppdbDecodingPlugin implements DecodingPlugin {
private final BaseTimestampUtils timestampUtils;
+ private final boolean decodeWithTX;
+
+ public MppdbDecodingPlugin(final BaseTimestampUtils timestampUtils) {
+ this.timestampUtils = timestampUtils;
+ decodeWithTX = false;
+ }
+
@Override
public AbstractWALEvent decode(final ByteBuffer data, final BaseLogSequenceNumber logSequenceNumber) {
AbstractWALEvent result;
- char eventType = readOneChar(data);
- result = '{' == eventType ? readTableEvent(readMppData(data)) : new PlaceholderEvent();
+ byte[] bytes = new byte[data.remaining()];
+ data.get(bytes);
+ String dataText = new String(bytes);
Review Comment:
`new String(bytes)` will use `Charset.defaultCharset()`, is it the same as plugin data?
--
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
[GitHub] [shardingsphere] sandynz commented on a diff in pull request #22685: Adding xid and csn parse for openGauss's logical replication
Posted by GitBox <gi...@apache.org>.
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
[GitHub] [shardingsphere] azexcy commented on a diff in pull request #22685: Adding xid and csn parse for openGauss's logical replication
Posted by GitBox <gi...@apache.org>.
azexcy commented on code in PR #22685:
URL: https://github.com/apache/shardingsphere/pull/22685#discussion_r1040513701
##########
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:
Because PostgreSQL don't have csn now, its always null now
--
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
[GitHub] [shardingsphere] azexcy commented on a diff in pull request #22685: Adding xid and csn parse for openGauss's logical replication
Posted by GitBox <gi...@apache.org>.
azexcy commented on code in PR #22685:
URL: https://github.com/apache/shardingsphere/pull/22685#discussion_r1040595748
##########
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:
Improved, reading only the first character would cause the word to be split, which is not easy to understand, so I use substring instead of pattern
--
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
[GitHub] [shardingsphere] TeslaCN commented on pull request #22685: Adding xid and csn parse for openGauss's logical replication
Posted by GitBox <gi...@apache.org>.
TeslaCN commented on PR #22685:
URL: https://github.com/apache/shardingsphere/pull/22685#issuecomment-1338646854
Hi @azexcy
Is this required in 5.3.0? Or we could move it to 5.3.1.
--
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
[GitHub] [shardingsphere] azexcy commented on a diff in pull request #22685: Adding xid and csn parse for openGauss's logical replication
Posted by GitBox <gi...@apache.org>.
azexcy commented on code in PR #22685:
URL: https://github.com/apache/shardingsphere/pull/22685#discussion_r1040555721
##########
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:
The lsn example like
```
lsn | xid | data
-----------+------+--------------------------------------------------------
0/19EA2C0 | 1045 | BEGIN 1045
0/19EA2C0 | 1045 | table public.t: INSERT: id[integer]:1 name[text]:51459cbc211647e7b31c8720
0/19EA300 | 1045 | table public.t: INSERT: id[integer]:2 name[text]:51459cbc211647e7b31c8720
0/19EA340 | 1045 | table public.t: INSERT: id[integer]:3 name[text]:51459cbc211647e7b31c8720
0/19EA380 | 1045 | table public.t: INSERT: id[integer]:4 name[text]:51459cbc211647e7b31c8720
0/19EA3C0 | 1045 | table public.t: INSERT: id[integer]:5 name[text]:51459cbc211647e7b31c8720
0/19EA400 | 1045 | table public.t: INSERT: id[integer]:6 name[text]:51459cbc211647e7b31c8720
0/19EA440 | 1045 | table public.t: INSERT: id[integer]:7 name[text]:51459cbc211647e7b31c8720
0/19EA480 | 1045 | table public.t: INSERT: id[integer]:8 name[text]:51459cbc211647e7b31c8720
0/19EA4C0 | 1045 | table public.t: INSERT: id[integer]:9 name[text]:51459cbc211647e7b31c8720
0/19EA500 | 1045 | table public.t: INSERT: id[integer]:10 name[text]:51459cbc211647e7b31c8720
0/19EA5B0 | 1045 | COMMIT 1045
(13 rows)
```
maybe we can ignore begin TX event, but the commit event should put into channel, for persist progress
<img width="1169" alt="image" src="https://user-images.githubusercontent.com/101622833/205838999-bcaebf9c-f64b-415e-a277-363735907c43.png">
--
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
[GitHub] [shardingsphere] sandynz merged pull request #22685: Adding xid and csn parse for openGauss's logical replication
Posted by GitBox <gi...@apache.org>.
sandynz merged PR #22685:
URL: https://github.com/apache/shardingsphere/pull/22685
--
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