-
Notifications
You must be signed in to change notification settings - Fork 52
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
Showing
6 changed files
with
218 additions
and
93 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
23 changes: 23 additions & 0 deletions
23
src/main/java/io/debezium/connector/db2/platform/Db2PlatformAdapter.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,23 @@ | ||
| /* | ||
| * Copyright Debezium Authors. | ||
| * | ||
| * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0 | ||
| */ | ||
| package io.debezium.connector.db2.platform; | ||
|
|
||
| /** | ||
| * Implementation details differing between Db2 flavours | ||
| * | ||
| * @author Jiri Pechanec | ||
| */ | ||
| public interface Db2PlatformAdapter { | ||
|
|
||
| String getMaxLsnQuery(); | ||
|
|
||
| String getAllChangesForTableQuery(); | ||
|
|
||
| String getListOfCdcEnabledTablesQuery(); | ||
|
|
||
| String getListOfNewCdcEnabledTablesQuery(); | ||
|
|
||
| } |
72 changes: 72 additions & 0 deletions
72
src/main/java/io/debezium/connector/db2/platform/LuwPlatform.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,72 @@ | ||
| /* | ||
| * Copyright Debezium Authors. | ||
| * | ||
| * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0 | ||
| */ | ||
| package io.debezium.connector.db2.platform; | ||
|
|
||
| import io.debezium.connector.db2.Db2ConnectorConfig; | ||
|
|
||
| /** | ||
| * Implementation details for LUW Platform (Linux, Unix, Windows) | ||
| * | ||
| * @author Jiri Pechanec | ||
| */ | ||
| public class LuwPlatform implements Db2PlatformAdapter { | ||
|
|
||
| private final String getMaxLsn; | ||
| private final String getAllChangesForTable; | ||
| private final String getListOfCdcEnabledTables; | ||
| private final String getListOfNewCdcEnabledTables; | ||
|
|
||
| public LuwPlatform(Db2ConnectorConfig connectorConfig) { | ||
|
|
||
| this.getMaxLsn = "SELECT max(t.SYNCHPOINT) FROM ( SELECT CD_NEW_SYNCHPOINT AS SYNCHPOINT FROM " + connectorConfig.getCdcControlSchema() | ||
| + ".IBMSNAP_REGISTER UNION ALL SELECT SYNCHPOINT AS SYNCHPOINT FROM " + connectorConfig.getCdcControlSchema() + ".IBMSNAP_REGISTER) t"; | ||
|
|
||
| this.getAllChangesForTable = "SELECT " | ||
| + "CASE " | ||
| + "WHEN IBMSNAP_OPERATION = 'D' AND (LEAD(cdc.IBMSNAP_OPERATION,1,'X') OVER (PARTITION BY cdc.IBMSNAP_COMMITSEQ ORDER BY cdc.IBMSNAP_INTENTSEQ)) ='I' THEN 3 " | ||
| + "WHEN IBMSNAP_OPERATION = 'I' AND (LAG(cdc.IBMSNAP_OPERATION,1,'X') OVER (PARTITION BY cdc.IBMSNAP_COMMITSEQ ORDER BY cdc.IBMSNAP_INTENTSEQ)) ='D' THEN 4 " | ||
| + "WHEN IBMSNAP_OPERATION = 'D' THEN 1 " | ||
| + "WHEN IBMSNAP_OPERATION = 'I' THEN 2 " | ||
| + "END " | ||
| + "OPCODE," | ||
| + "cdc.* " | ||
| + "FROM " + connectorConfig.getCdcChangeTablesSchema() + ".# cdc WHERE IBMSNAP_COMMITSEQ >= ? AND IBMSNAP_COMMITSEQ <= ? " | ||
| + "order by IBMSNAP_COMMITSEQ, IBMSNAP_INTENTSEQ"; | ||
|
|
||
| this.getListOfCdcEnabledTables = "select r.SOURCE_OWNER, r.SOURCE_TABLE, r.CD_OWNER, r.CD_TABLE, r.CD_NEW_SYNCHPOINT, r.CD_OLD_SYNCHPOINT, t.TBSPACEID, t.TABLEID , CAST((t.TBSPACEID * 65536 + t.TABLEID )AS INTEGER )from " | ||
| + connectorConfig.getCdcControlSchema() | ||
| + ".IBMSNAP_REGISTER r left JOIN SYSCAT.TABLES t ON r.SOURCE_OWNER = t.TABSCHEMA AND r.SOURCE_TABLE = t.TABNAME WHERE r.SOURCE_OWNER <> ''"; | ||
|
|
||
| // No new Tables 1=0 | ||
| this.getListOfNewCdcEnabledTables = "select CAST((t.TBSPACEID * 65536 + t.TABLEID )AS INTEGER ) AS OBJECTID, " + | ||
| " CD_OWNER CONCAT '.' CONCAT CD_TABLE, " + | ||
| " CD_NEW_SYNCHPOINT, " + | ||
| " CD_OLD_SYNCHPOINT " + | ||
| "from " + connectorConfig.getCdcControlSchema() | ||
| + ".IBMSNAP_REGISTER r left JOIN SYSCAT.TABLES t ON r.SOURCE_OWNER = t.TABSCHEMA AND r.SOURCE_TABLE = t.TABNAME " + | ||
| "WHERE r.SOURCE_OWNER <> '' AND CD_NEW_SYNCHPOINT > ? AND (CD_OLD_SYNCHPOINT < ? OR CD_OLD_SYNCHPOINT IS NULL)"; | ||
| } | ||
|
|
||
| @Override | ||
| public String getMaxLsnQuery() { | ||
| return getMaxLsn; | ||
| } | ||
|
|
||
| @Override | ||
| public String getAllChangesForTableQuery() { | ||
| return getAllChangesForTable; | ||
| } | ||
|
|
||
| @Override | ||
| public String getListOfCdcEnabledTablesQuery() { | ||
| return getListOfCdcEnabledTables; | ||
| } | ||
|
|
||
| @Override | ||
| public String getListOfNewCdcEnabledTablesQuery() { | ||
| return getListOfNewCdcEnabledTables; | ||
| } | ||
| } |
Oops, something went wrong.