You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@shardingsphere.apache.org by "azexcy (via GitHub)" <gi...@apache.org> on 2024/03/15 03:44:30 UTC
[PR] Pipeline support parallel decoding plugin of opengauss [shardingsphere]
azexcy opened a new pull request, #30498:
URL: https://github.com/apache/shardingsphere/pull/30498
Parallel decoder enabled by default after opengauss 3
Changes proposed in this pull request:
- Pipeline support parallel decoding plugin of opengauss
---
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
Re: [PR] Pipeline support parallel decoding plugin of opengauss [shardingsphere]
Posted by "azexcy (via GitHub)" <gi...@apache.org>.
azexcy commented on code in PR #30498:
URL: https://github.com/apache/shardingsphere/pull/30498#discussion_r1527839285
##########
kernel/data-pipeline/dialect/opengauss/src/test/java/org/apache/shardingsphere/data/pipeline/opengauss/ingest/wal/decode/MppdbDecodingPluginTest.java:
##########
@@ -237,21 +238,41 @@ void assertDecodeWithXid() {
tableData.setColumnsVal(new String[]{"'7D'"});
List<String> dataList = Arrays.asList("BEGIN 1", JsonUtils.toJsonString(tableData), JsonUtils.toJsonString(tableData),
"COMMIT 1 (at 2022-10-27 04:19:39.476261+00) CSN 3468");
- MppdbDecodingPlugin mppdbDecodingPlugin = new MppdbDecodingPlugin(null, true);
+ MppdbDecodingPlugin mppdbDecodingPlugin = new MppdbDecodingPlugin(null, true, 2);
List<AbstractWALEvent> expectedEvent = new LinkedList<>();
for (String each : dataList) {
expectedEvent.add(mppdbDecodingPlugin.decode(ByteBuffer.wrap(each.getBytes()), logSequenceNumber));
}
assertThat(expectedEvent.size(), is(4));
AbstractWALEvent actualFirstEvent = expectedEvent.get(0);
assertInstanceOf(BeginTXEvent.class, actualFirstEvent);
- assertThat(((BeginTXEvent) actualFirstEvent).getXid(), is(1L));
AbstractWALEvent actualLastEvent = expectedEvent.get(expectedEvent.size() - 1);
assertInstanceOf(CommitTXEvent.class, actualLastEvent);
assertThat(((CommitTXEvent) actualLastEvent).getCsn(), is(3468L));
assertThat(((CommitTXEvent) actualLastEvent).getXid(), is(1L));
}
+ @Test
+ void assertParallelDecodeWithTx() {
+ MppTableData tableData = new MppTableData();
+ tableData.setTableName("public.test");
+ tableData.setOpType("INSERT");
+ tableData.setColumnsName(new String[]{"data"});
+ tableData.setColumnsType(new String[]{"raw"});
+ tableData.setColumnsVal(new String[]{"'7D'"});
+ List<String> dataList = Arrays.asList("BEGIN CSN: 951909 first_lsn: 5/59825858", JsonUtils.toJsonString(tableData), JsonUtils.toJsonString(tableData), "commit xid: 1006076");
Review Comment:
Additional compatibility when dealing with begin event
--
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
Re: [PR] Pipeline support parallel decoding plugin of opengauss [shardingsphere]
Posted by "azexcy (via GitHub)" <gi...@apache.org>.
azexcy commented on code in PR #30498:
URL: https://github.com/apache/shardingsphere/pull/30498#discussion_r1527759236
##########
kernel/data-pipeline/dialect/opengauss/src/test/java/org/apache/shardingsphere/data/pipeline/opengauss/ingest/wal/decode/MppdbDecodingPluginTest.java:
##########
@@ -237,21 +238,41 @@ void assertDecodeWithXid() {
tableData.setColumnsVal(new String[]{"'7D'"});
List<String> dataList = Arrays.asList("BEGIN 1", JsonUtils.toJsonString(tableData), JsonUtils.toJsonString(tableData),
"COMMIT 1 (at 2022-10-27 04:19:39.476261+00) CSN 3468");
- MppdbDecodingPlugin mppdbDecodingPlugin = new MppdbDecodingPlugin(null, true);
+ MppdbDecodingPlugin mppdbDecodingPlugin = new MppdbDecodingPlugin(null, true, 2);
List<AbstractWALEvent> expectedEvent = new LinkedList<>();
for (String each : dataList) {
expectedEvent.add(mppdbDecodingPlugin.decode(ByteBuffer.wrap(each.getBytes()), logSequenceNumber));
}
assertThat(expectedEvent.size(), is(4));
AbstractWALEvent actualFirstEvent = expectedEvent.get(0);
assertInstanceOf(BeginTXEvent.class, actualFirstEvent);
- assertThat(((BeginTXEvent) actualFirstEvent).getXid(), is(1L));
AbstractWALEvent actualLastEvent = expectedEvent.get(expectedEvent.size() - 1);
assertInstanceOf(CommitTXEvent.class, actualLastEvent);
assertThat(((CommitTXEvent) actualLastEvent).getCsn(), is(3468L));
assertThat(((CommitTXEvent) actualLastEvent).getXid(), is(1L));
}
+ @Test
+ void assertParallelDecodeWithTx() {
+ MppTableData tableData = new MppTableData();
+ tableData.setTableName("public.test");
+ tableData.setOpType("INSERT");
+ tableData.setColumnsName(new String[]{"data"});
+ tableData.setColumnsType(new String[]{"raw"});
+ tableData.setColumnsVal(new String[]{"'7D'"});
+ List<String> dataList = Arrays.asList("BEGIN CSN: 951909 first_lsn: 5/59825858", JsonUtils.toJsonString(tableData), JsonUtils.toJsonString(tableData), "commit xid: 1006076");
+ MppdbDecodingPlugin mppdbDecodingPlugin = new MppdbDecodingPlugin(null, true, 3);
+ List<AbstractWALEvent> expectedEvents = new LinkedList<>();
+ for (String each : dataList) {
+ expectedEvents.add(mppdbDecodingPlugin.decode(ByteBuffer.wrap(each.getBytes()), logSequenceNumber));
+ }
+ assertThat(expectedEvents.size(), is(4));
+ assertInstanceOf(BeginTXEvent.class, expectedEvents.get(0));
+ assertThat(((BeginTXEvent) expectedEvents.get(0)).getCsn(), is(951909L));
+ assertThat(((CommitTXEvent) expectedEvents.get(3)).getXid(), is(1006076L));
+ assertNull(((CommitTXEvent) expectedEvents.get(3)).getCsn());
Review Comment:
AbstractRowEvent set CSN at OpenGaussWALDumper, so at `decode` method , CSN is always null
--
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
Re: [PR] Pipeline support parallel decoding plugin of opengauss [shardingsphere]
Posted by "azexcy (via GitHub)" <gi...@apache.org>.
azexcy commented on PR #30498:
URL: https://github.com/apache/shardingsphere/pull/30498#issuecomment-1999161601
Default mpp_decoding
```
BEGIN 1004573
{"table_name":"public.t_sales","op_type":"INSERT","columns_name":["id","goods_name","sales_date","sales_volume","sales_store"],"columns_type":["integer","character","timestamp without time zone","integer","character"],"columns_val":["1","'aq '","'2024-03-14 20:30:58'","2","'d2 '"],"old_keys_name":[],"old_keys_type":[],"old_keys_val":[]}
COMMIT 1004573 (at 2024-03-14 20:31:12.098179+08) CSN 951909
```
Parallel decoding plugin
```
BEGIN CSN: 951909 first_lsn: 5/59825858
{"table_name":"public.t_sales","op_type":"INSERT","columns_name":["id","goods_name","sales_date","sales_volume","sales_store"],"columns_type":["integer","character","timestamp without time zone","integer","character"],"columns_val":["1","'aq '","'2024-03-14 20:30:58'","2","'d2 '"],"old_keys_name":[],"old_keys_type":[],"old_keys_val":[]}
commit xid: 1004573
```
--
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
Re: [PR] Pipeline support parallel decoding plugin of opengauss [shardingsphere]
Posted by "azexcy (via GitHub)" <gi...@apache.org>.
azexcy commented on code in PR #30498:
URL: https://github.com/apache/shardingsphere/pull/30498#discussion_r1527710375
##########
kernel/data-pipeline/dialect/opengauss/src/test/java/org/apache/shardingsphere/data/pipeline/opengauss/ingest/wal/decode/MppdbDecodingPluginTest.java:
##########
@@ -237,21 +238,41 @@ void assertDecodeWithXid() {
tableData.setColumnsVal(new String[]{"'7D'"});
List<String> dataList = Arrays.asList("BEGIN 1", JsonUtils.toJsonString(tableData), JsonUtils.toJsonString(tableData),
"COMMIT 1 (at 2022-10-27 04:19:39.476261+00) CSN 3468");
- MppdbDecodingPlugin mppdbDecodingPlugin = new MppdbDecodingPlugin(null, true);
+ MppdbDecodingPlugin mppdbDecodingPlugin = new MppdbDecodingPlugin(null, true, 2);
List<AbstractWALEvent> expectedEvent = new LinkedList<>();
for (String each : dataList) {
expectedEvent.add(mppdbDecodingPlugin.decode(ByteBuffer.wrap(each.getBytes()), logSequenceNumber));
}
assertThat(expectedEvent.size(), is(4));
AbstractWALEvent actualFirstEvent = expectedEvent.get(0);
assertInstanceOf(BeginTXEvent.class, actualFirstEvent);
- assertThat(((BeginTXEvent) actualFirstEvent).getXid(), is(1L));
AbstractWALEvent actualLastEvent = expectedEvent.get(expectedEvent.size() - 1);
assertInstanceOf(CommitTXEvent.class, actualLastEvent);
assertThat(((CommitTXEvent) actualLastEvent).getCsn(), is(3468L));
assertThat(((CommitTXEvent) actualLastEvent).getXid(), is(1L));
}
+ @Test
+ void assertParallelDecodeWithTx() {
+ MppTableData tableData = new MppTableData();
+ tableData.setTableName("public.test");
+ tableData.setOpType("INSERT");
+ tableData.setColumnsName(new String[]{"data"});
+ tableData.setColumnsType(new String[]{"raw"});
+ tableData.setColumnsVal(new String[]{"'7D'"});
+ List<String> dataList = Arrays.asList("BEGIN CSN: 951909 first_lsn: 5/59825858", JsonUtils.toJsonString(tableData), JsonUtils.toJsonString(tableData), "commit xid: 1006076");
Review Comment:
I test `opengauss 3.x` and `5.x`, prarallel decoding always return lowercase at commit. So here's a direct lowercase judgment
--
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
Re: [PR] Pipeline support parallel decoding plugin of opengauss [shardingsphere]
Posted by "sandynz (via GitHub)" <gi...@apache.org>.
sandynz merged PR #30498:
URL: https://github.com/apache/shardingsphere/pull/30498
--
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
Re: [PR] Pipeline support parallel decoding plugin of opengauss [shardingsphere]
Posted by "azexcy (via GitHub)" <gi...@apache.org>.
azexcy commented on code in PR #30498:
URL: https://github.com/apache/shardingsphere/pull/30498#discussion_r1527710375
##########
kernel/data-pipeline/dialect/opengauss/src/test/java/org/apache/shardingsphere/data/pipeline/opengauss/ingest/wal/decode/MppdbDecodingPluginTest.java:
##########
@@ -237,21 +238,41 @@ void assertDecodeWithXid() {
tableData.setColumnsVal(new String[]{"'7D'"});
List<String> dataList = Arrays.asList("BEGIN 1", JsonUtils.toJsonString(tableData), JsonUtils.toJsonString(tableData),
"COMMIT 1 (at 2022-10-27 04:19:39.476261+00) CSN 3468");
- MppdbDecodingPlugin mppdbDecodingPlugin = new MppdbDecodingPlugin(null, true);
+ MppdbDecodingPlugin mppdbDecodingPlugin = new MppdbDecodingPlugin(null, true, 2);
List<AbstractWALEvent> expectedEvent = new LinkedList<>();
for (String each : dataList) {
expectedEvent.add(mppdbDecodingPlugin.decode(ByteBuffer.wrap(each.getBytes()), logSequenceNumber));
}
assertThat(expectedEvent.size(), is(4));
AbstractWALEvent actualFirstEvent = expectedEvent.get(0);
assertInstanceOf(BeginTXEvent.class, actualFirstEvent);
- assertThat(((BeginTXEvent) actualFirstEvent).getXid(), is(1L));
AbstractWALEvent actualLastEvent = expectedEvent.get(expectedEvent.size() - 1);
assertInstanceOf(CommitTXEvent.class, actualLastEvent);
assertThat(((CommitTXEvent) actualLastEvent).getCsn(), is(3468L));
assertThat(((CommitTXEvent) actualLastEvent).getXid(), is(1L));
}
+ @Test
+ void assertParallelDecodeWithTx() {
+ MppTableData tableData = new MppTableData();
+ tableData.setTableName("public.test");
+ tableData.setOpType("INSERT");
+ tableData.setColumnsName(new String[]{"data"});
+ tableData.setColumnsType(new String[]{"raw"});
+ tableData.setColumnsVal(new String[]{"'7D'"});
+ List<String> dataList = Arrays.asList("BEGIN CSN: 951909 first_lsn: 5/59825858", JsonUtils.toJsonString(tableData), JsonUtils.toJsonString(tableData), "commit xid: 1006076");
Review Comment:
I test `opengauss 3.x` and `5.x`, prarallel decoding always return lowercase at commit.
<img width="1084" alt="image" src="https://github.com/apache/shardingsphere/assets/101622833/4b1b241f-f387-4005-b8ec-4c03a8c6cb8b">
--
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
Re: [PR] Pipeline support parallel decoding plugin of opengauss [shardingsphere]
Posted by "sandynz (via GitHub)" <gi...@apache.org>.
sandynz commented on code in PR #30498:
URL: https://github.com/apache/shardingsphere/pull/30498#discussion_r1527914474
##########
kernel/data-pipeline/dialect/opengauss/src/main/java/org/apache/shardingsphere/data/pipeline/opengauss/ingest/wal/decode/MppdbDecodingPlugin.java:
##########
@@ -91,6 +97,21 @@ private AbstractWALEvent decodeDataWithTX(final String dataText) {
return result;
}
+ private AbstractWALEvent decodeVersionGreaterThan3(final String dataText) {
+ AbstractWALEvent result = new PlaceholderEvent();
+ if (dataText.startsWith("BEGIN")) {
+ int beginIndex = dataText.indexOf("CSN:") + "CSN:".length() + 1;
+ int endIndex = dataText.indexOf("first_lsn") - 1;
+ result = new BeginTXEvent(null, Long.parseLong(dataText.substring(beginIndex, endIndex)));
Review Comment:
Need to check whether there's `first_lsn` or not
##########
kernel/data-pipeline/dialect/opengauss/src/main/java/org/apache/shardingsphere/data/pipeline/opengauss/ingest/OpenGaussWALDumper.java:
##########
@@ -60,6 +67,10 @@
@Slf4j
public final class OpenGaussWALDumper extends AbstractPipelineLifecycleRunnable implements IncrementalDumper {
+ private static final Pattern VERSION_PATTERN = Pattern.compile("(\\d+)");
Review Comment:
Seems `openGauss ` prefix could not be removed in VERSION_PATTERN
##########
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 int majorVersion;
Review Comment:
Could we replace `int majorVersion` to `boolean decodeParallelly`, it's error-prone to to transfer majorVersion (e.g. 2, 3 or any other integer) on invocation.
##########
kernel/data-pipeline/dialect/opengauss/src/main/java/org/apache/shardingsphere/data/pipeline/opengauss/ingest/wal/decode/MppdbDecodingPlugin.java:
##########
@@ -91,6 +97,21 @@ private AbstractWALEvent decodeDataWithTX(final String dataText) {
return result;
}
+ private AbstractWALEvent decodeVersionGreaterThan3(final String dataText) {
+ AbstractWALEvent result = new PlaceholderEvent();
+ if (dataText.startsWith("BEGIN")) {
+ int beginIndex = dataText.indexOf("CSN:") + "CSN:".length() + 1;
+ int endIndex = dataText.indexOf("first_lsn") - 1;
+ result = new BeginTXEvent(null, Long.parseLong(dataText.substring(beginIndex, endIndex)));
+ } else if (dataText.startsWith("commit")) {
Review Comment:
It's better to compatible with `COMMIT`, since it's different as `BEGIN`, though it's defined as `commit` in document.
e.g. Use `dataText.startsWith("commit") || dataText.startsWith("COMMIT")`
##########
kernel/data-pipeline/dialect/opengauss/src/main/java/org/apache/shardingsphere/data/pipeline/opengauss/ingest/OpenGaussWALDumper.java:
##########
@@ -138,28 +152,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);
+ if (matcher.find()) {
+ log.info("openGauss major version={}, `select version()`={}", matcher.group(1), versionText);
+ return Integer.parseInt(matcher.group(1));
+ }
Review Comment:
No matter `matcher.find()` return true or not, print the log
##########
kernel/data-pipeline/dialect/opengauss/src/main/java/org/apache/shardingsphere/data/pipeline/opengauss/ingest/wal/decode/MppdbDecodingPlugin.java:
##########
@@ -77,10 +75,18 @@ public AbstractWALEvent decode(final ByteBuffer data, final BaseLogSequenceNumbe
}
private AbstractWALEvent decodeDataWithTX(final String dataText) {
+ if (majorVersion < 3) {
+ return decodeVersionLessThan3(dataText);
+ } else {
+ return decodeVersionGreaterThan3(dataText);
+ }
Review Comment:
`decodeVersionLessThan3` could be `decodeSerially`, `decodeVersionGreaterThan3` could be `decodeParallelly`
##########
kernel/data-pipeline/dialect/opengauss/src/main/java/org/apache/shardingsphere/data/pipeline/opengauss/ingest/OpenGaussWALDumper.java:
##########
@@ -138,28 +152,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);
+ if (matcher.find()) {
+ log.info("openGauss major version={}, `select version()`={}", matcher.group(1), versionText);
+ 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()) {
+ channel.push(rowEvents.stream().map(walEventConverter::convert).collect(Collectors.toList()));
+ rowEvents = new LinkedList<>();
+ }
Review Comment:
`rowEvents` belongs to previous transaction, and BeginTXEvent belongs to next transaction, looks it should not occur if CommitTXEvent parsing works.
Could we remove this block of code and just `log.warn`?
--
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
Re: [PR] Pipeline support parallel decoding plugin of opengauss [shardingsphere]
Posted by "sandynz (via GitHub)" <gi...@apache.org>.
sandynz commented on code in PR #30498:
URL: https://github.com/apache/shardingsphere/pull/30498#discussion_r1527021665
##########
kernel/data-pipeline/dialect/opengauss/src/main/java/org/apache/shardingsphere/data/pipeline/opengauss/ingest/wal/decode/MppdbDecodingPlugin.java:
##########
@@ -78,15 +76,28 @@ public AbstractWALEvent decode(final ByteBuffer data, final BaseLogSequenceNumbe
private AbstractWALEvent decodeDataWithTX(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)));
- } else if (dataText.startsWith("COMMIT")) {
- int commitBeginIndex = dataText.indexOf("COMMIT") + "COMMIT".length() + 1;
- int csnBeginIndex = dataText.indexOf("CSN") + "CSN".length() + 1;
- result = new CommitTXEvent(Long.parseLong(dataText.substring(commitBeginIndex, dataText.indexOf(' ', commitBeginIndex))), Long.parseLong(dataText.substring(csnBeginIndex)));
- } else if (dataText.startsWith("{")) {
- result = readTableEvent(dataText);
+ if (majorVersion <= 2) {
+ if (dataText.startsWith("BEGIN")) {
+ int beginIndex = dataText.indexOf("BEGIN") + "BEGIN".length() + 1;
+ 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;
+ result = new CommitTXEvent(Long.parseLong(dataText.substring(commitBeginIndex, dataText.indexOf(' ', commitBeginIndex))), Long.parseLong(dataText.substring(csnBeginIndex)));
+ } else if (dataText.startsWith("{")) {
+ result = readTableEvent(dataText);
+ }
+ } else {
+ if (dataText.startsWith("BEGIN")) {
+ int beginIndex = dataText.indexOf("CSN:") + "CSN:".length() + 1;
+ int endIndex = dataText.indexOf("first_lsn") - 1;
+ result = new BeginTXEvent(null, Long.parseLong(dataText.substring(beginIndex, endIndex)));
+ } else if (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);
+ }
Review Comment:
1, Could we extract 2 methods, one is for openGauss 2, and another is for openGauss 3+
2, `majorVersion <= 2` could be `majorVersion < 3`, use the same expression as other places
##########
kernel/data-pipeline/dialect/opengauss/src/test/java/org/apache/shardingsphere/data/pipeline/opengauss/ingest/wal/decode/MppdbDecodingPluginTest.java:
##########
@@ -237,21 +238,41 @@ void assertDecodeWithXid() {
tableData.setColumnsVal(new String[]{"'7D'"});
List<String> dataList = Arrays.asList("BEGIN 1", JsonUtils.toJsonString(tableData), JsonUtils.toJsonString(tableData),
"COMMIT 1 (at 2022-10-27 04:19:39.476261+00) CSN 3468");
- MppdbDecodingPlugin mppdbDecodingPlugin = new MppdbDecodingPlugin(null, true);
+ MppdbDecodingPlugin mppdbDecodingPlugin = new MppdbDecodingPlugin(null, true, 2);
List<AbstractWALEvent> expectedEvent = new LinkedList<>();
for (String each : dataList) {
expectedEvent.add(mppdbDecodingPlugin.decode(ByteBuffer.wrap(each.getBytes()), logSequenceNumber));
}
assertThat(expectedEvent.size(), is(4));
AbstractWALEvent actualFirstEvent = expectedEvent.get(0);
assertInstanceOf(BeginTXEvent.class, actualFirstEvent);
- assertThat(((BeginTXEvent) actualFirstEvent).getXid(), is(1L));
AbstractWALEvent actualLastEvent = expectedEvent.get(expectedEvent.size() - 1);
assertInstanceOf(CommitTXEvent.class, actualLastEvent);
assertThat(((CommitTXEvent) actualLastEvent).getCsn(), is(3468L));
assertThat(((CommitTXEvent) actualLastEvent).getXid(), is(1L));
}
+ @Test
+ void assertParallelDecodeWithTx() {
+ MppTableData tableData = new MppTableData();
+ tableData.setTableName("public.test");
+ tableData.setOpType("INSERT");
+ tableData.setColumnsName(new String[]{"data"});
+ tableData.setColumnsType(new String[]{"raw"});
+ tableData.setColumnsVal(new String[]{"'7D'"});
+ List<String> dataList = Arrays.asList("BEGIN CSN: 951909 first_lsn: 5/59825858", JsonUtils.toJsonString(tableData), JsonUtils.toJsonString(tableData), "commit xid: 1006076");
+ MppdbDecodingPlugin mppdbDecodingPlugin = new MppdbDecodingPlugin(null, true, 3);
+ List<AbstractWALEvent> expectedEvents = new LinkedList<>();
Review Comment:
Looks `expectedEvents` is actual events
##########
kernel/data-pipeline/dialect/opengauss/src/main/java/org/apache/shardingsphere/data/pipeline/opengauss/ingest/OpenGaussWALDumper.java:
##########
@@ -60,6 +66,10 @@
@Slf4j
public final class OpenGaussWALDumper extends AbstractPipelineLifecycleRunnable implements IncrementalDumper {
+ private static final Pattern VERSION_PATTERN = Pattern.compile("(openGauss) (\\d+)");
Review Comment:
Looks parentheses of `(openGauss)` is not required
##########
kernel/data-pipeline/dialect/opengauss/src/test/java/org/apache/shardingsphere/data/pipeline/opengauss/ingest/OpenGaussWALDumperTest.java:
##########
@@ -0,0 +1,57 @@
+/*
+ * 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.data.pipeline.api.type.StandardPipelineDataSourceConfiguration;
+import org.apache.shardingsphere.data.pipeline.core.ingest.dumper.DumperCommonContext;
+import org.apache.shardingsphere.data.pipeline.core.ingest.dumper.incremental.IncrementalDumperContext;
+import org.apache.shardingsphere.data.pipeline.core.ingest.dumper.mapper.ActualAndLogicTableNameMapper;
+import org.apache.shardingsphere.data.pipeline.core.ingest.dumper.mapper.TableAndSchemaNameMapper;
+import org.apache.shardingsphere.infra.util.reflection.ReflectionUtils;
+import org.junit.jupiter.api.Test;
+
+import java.sql.Connection;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.Statement;
+
+import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+class OpenGaussWALDumperTest {
+
+ @Test
+ void assertGetVersion() throws NoSuchMethodException, SQLException {
+ StandardPipelineDataSourceConfiguration mockDataSourceConfig = new StandardPipelineDataSourceConfiguration("jdbc:mysql://127.0.0.1:3306/test", "root", "root");
+ IncrementalDumperContext dumperContext = new IncrementalDumperContext(new DumperCommonContext("", mockDataSourceConfig, mock(ActualAndLogicTableNameMapper.class),
+ mock(TableAndSchemaNameMapper.class)), "1", true);
+ OpenGaussWALDumper dumper = new OpenGaussWALDumper(dumperContext, null, null, null);
+ Connection connection = mock(Connection.class);
+ Statement statement = mock(Statement.class);
+ ResultSet resultSet = mock(ResultSet.class);
+ when(connection.createStatement()).thenReturn(statement);
+ when(statement.executeQuery(anyString())).thenReturn(resultSet);
+ when(resultSet.next()).thenReturn(true);
+ when(resultSet.getString(1)).thenReturn("(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");
+ int version = ReflectionUtils.invokeMethod(dumper.getClass().getDeclaredMethod("getMajorVersion", Connection.class), dumper, connection);
+ assertThat(version, is(3));
+ }
Review Comment:
Looks this unit test just assert version parsing. Could we extract `parseMajorVersion` in `OpenGaussWALDumper`, then this unit test could be simplified by removing all mocks.
##########
kernel/data-pipeline/dialect/opengauss/src/main/java/org/apache/shardingsphere/data/pipeline/opengauss/ingest/OpenGaussWALDumper.java:
##########
@@ -110,10 +122,15 @@ protected void runBlocking() {
@SneakyThrows(InterruptedException.class)
private void dump() throws SQLException {
PGReplicationStream stream = null;
+ StandardPipelineDataSourceConfiguration dataSourceConfig = (StandardPipelineDataSourceConfiguration) dumperContext.getCommonContext().getDataSourceConfig();
+ int majorVersion;
+ try (Connection connection = DriverManager.getConnection(dataSourceConfig.getUrl(), dataSourceConfig.getUsername(), dataSourceConfig.getPassword())) {
+ majorVersion = getMajorVersion(connection);
+ }
Review Comment:
Could we merge these lines of code into `getMajorVersion` method?
##########
kernel/data-pipeline/dialect/opengauss/src/main/java/org/apache/shardingsphere/data/pipeline/opengauss/ingest/OpenGaussWALDumper.java:
##########
@@ -138,28 +155,50 @@ private void dump() throws SQLException {
}
}
+ private int getMajorVersion(final Connection connection) throws SQLException {
+ try (
+ Statement statement = connection.createStatement();
+ ResultSet resultSet = statement.executeQuery("SELECT version()")) {
+ resultSet.next();
+ String versionText = resultSet.getString(1);
+ log.info("openGauss select version() result={}", versionText);
+ Matcher matcher = VERSION_PATTERN.matcher(versionText);
+ if (matcher.find()) {
+ return Integer.parseInt(matcher.group(2));
+ }
Review Comment:
Could we print the parsed version? And merg previous log.
##########
kernel/data-pipeline/dialect/opengauss/src/test/java/org/apache/shardingsphere/data/pipeline/opengauss/ingest/wal/decode/MppdbDecodingPluginTest.java:
##########
@@ -237,21 +238,41 @@ void assertDecodeWithXid() {
tableData.setColumnsVal(new String[]{"'7D'"});
List<String> dataList = Arrays.asList("BEGIN 1", JsonUtils.toJsonString(tableData), JsonUtils.toJsonString(tableData),
"COMMIT 1 (at 2022-10-27 04:19:39.476261+00) CSN 3468");
- MppdbDecodingPlugin mppdbDecodingPlugin = new MppdbDecodingPlugin(null, true);
+ MppdbDecodingPlugin mppdbDecodingPlugin = new MppdbDecodingPlugin(null, true, 2);
List<AbstractWALEvent> expectedEvent = new LinkedList<>();
for (String each : dataList) {
expectedEvent.add(mppdbDecodingPlugin.decode(ByteBuffer.wrap(each.getBytes()), logSequenceNumber));
}
assertThat(expectedEvent.size(), is(4));
AbstractWALEvent actualFirstEvent = expectedEvent.get(0);
assertInstanceOf(BeginTXEvent.class, actualFirstEvent);
- assertThat(((BeginTXEvent) actualFirstEvent).getXid(), is(1L));
AbstractWALEvent actualLastEvent = expectedEvent.get(expectedEvent.size() - 1);
assertInstanceOf(CommitTXEvent.class, actualLastEvent);
assertThat(((CommitTXEvent) actualLastEvent).getCsn(), is(3468L));
assertThat(((CommitTXEvent) actualLastEvent).getXid(), is(1L));
}
+ @Test
+ void assertParallelDecodeWithTx() {
+ MppTableData tableData = new MppTableData();
+ tableData.setTableName("public.test");
+ tableData.setOpType("INSERT");
+ tableData.setColumnsName(new String[]{"data"});
+ tableData.setColumnsType(new String[]{"raw"});
+ tableData.setColumnsVal(new String[]{"'7D'"});
+ List<String> dataList = Arrays.asList("BEGIN CSN: 951909 first_lsn: 5/59825858", JsonUtils.toJsonString(tableData), JsonUtils.toJsonString(tableData), "commit xid: 1006076");
+ MppdbDecodingPlugin mppdbDecodingPlugin = new MppdbDecodingPlugin(null, true, 3);
+ List<AbstractWALEvent> expectedEvents = new LinkedList<>();
+ for (String each : dataList) {
+ expectedEvents.add(mppdbDecodingPlugin.decode(ByteBuffer.wrap(each.getBytes()), logSequenceNumber));
+ }
+ assertThat(expectedEvents.size(), is(4));
+ assertInstanceOf(BeginTXEvent.class, expectedEvents.get(0));
+ assertThat(((BeginTXEvent) expectedEvents.get(0)).getCsn(), is(951909L));
+ assertThat(((CommitTXEvent) expectedEvents.get(3)).getXid(), is(1006076L));
+ assertNull(((CommitTXEvent) expectedEvents.get(3)).getCsn());
Review Comment:
Data record could be asserted. Including data and CSN.
##########
kernel/data-pipeline/dialect/opengauss/src/test/java/org/apache/shardingsphere/data/pipeline/opengauss/ingest/wal/decode/MppdbDecodingPluginTest.java:
##########
@@ -237,21 +238,41 @@ void assertDecodeWithXid() {
tableData.setColumnsVal(new String[]{"'7D'"});
List<String> dataList = Arrays.asList("BEGIN 1", JsonUtils.toJsonString(tableData), JsonUtils.toJsonString(tableData),
"COMMIT 1 (at 2022-10-27 04:19:39.476261+00) CSN 3468");
- MppdbDecodingPlugin mppdbDecodingPlugin = new MppdbDecodingPlugin(null, true);
+ MppdbDecodingPlugin mppdbDecodingPlugin = new MppdbDecodingPlugin(null, true, 2);
List<AbstractWALEvent> expectedEvent = new LinkedList<>();
for (String each : dataList) {
expectedEvent.add(mppdbDecodingPlugin.decode(ByteBuffer.wrap(each.getBytes()), logSequenceNumber));
}
assertThat(expectedEvent.size(), is(4));
AbstractWALEvent actualFirstEvent = expectedEvent.get(0);
assertInstanceOf(BeginTXEvent.class, actualFirstEvent);
- assertThat(((BeginTXEvent) actualFirstEvent).getXid(), is(1L));
AbstractWALEvent actualLastEvent = expectedEvent.get(expectedEvent.size() - 1);
assertInstanceOf(CommitTXEvent.class, actualLastEvent);
assertThat(((CommitTXEvent) actualLastEvent).getCsn(), is(3468L));
assertThat(((CommitTXEvent) actualLastEvent).getXid(), is(1L));
}
+ @Test
+ void assertParallelDecodeWithTx() {
+ MppTableData tableData = new MppTableData();
+ tableData.setTableName("public.test");
+ tableData.setOpType("INSERT");
+ tableData.setColumnsName(new String[]{"data"});
+ tableData.setColumnsType(new String[]{"raw"});
+ tableData.setColumnsVal(new String[]{"'7D'"});
+ List<String> dataList = Arrays.asList("BEGIN CSN: 951909 first_lsn: 5/59825858", JsonUtils.toJsonString(tableData), JsonUtils.toJsonString(tableData), "commit xid: 1006076");
Review Comment:
`commit` is different as `BEGIN`, is it case sensitive?
--
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