[Bug] [oracle logminer connector] oracle logminer connector 不支持 float、timestamp、blob、clob 、decimal、number、text数据类型

[LiDianLEC]

Search before asking

• [X] I had searched in the issues and found no similar issues.

What happened

[Bug] [oracle logminer] oracle logminer 不支持 float、timestamp、blob、clob 、decimal、number、text数据类型

What you expected to happen

[Bug] [oracle logminer] oracle logminer 不支持 float、timestamp、blob、clob 、decimal、number、text数据类型

How to reproduce

[Bug] [oracle logminer] oracle logminer 不支持 float、timestamp、blob、clob 、decimal、number、text数据类型

Anything else

[Bug] [oracle logminer] oracle logminer 不支持 float、timestamp、blob、clob 、decimal、number、text数据类型

Version

master

Are you willing to submit PR?

• [X] Yes I am willing to submit a PR!

Code of Conduct

• [X] I agree to follow this project's Code of Conduct

[FlechazoW]

有具体报错吗？我看代码里是对这些类型有做支持的。

[LiDianLEC]

logminer到oracle，blob类型报错

• 报错信息
• 报错方法

[FlechazoW]

能提供下你的任务脚本嘛？

[LiDianLEC]

• CHUNJUN JSON

{
  "job":{
    "content":[
      {
        "reader":{
          "parameter":{
            "password":"*",
            "splitUpdate":true,
            "cat":"update,insert,delete",
            "jdbcUrl":"jdbc:oracle:thin:@//*",
            "column":[
              {
                "name": "B2",
                "type": "BLOB"
              }
            ],
            "readPosition":"current",
            "pavingData":true,
            "table":[
              "ROMA_LOGMINER.SS1"
            ],
            "username":"roma_logminer"
          },
          "name":"oraclelogminersource",
          "table":{
            "tableName":"sourceTable"
          }
        },
        "transformer":{
          "transformSql":"SELECT B2 FROM sourceTable"
        },
        "writer":{
          "parameter":{
            "password":"*",
            "column":[
              {
                "index": 0,
                "name": "B2",
                "type": "BLOB"
              }
            ],
            "connection":[
              {
                "schema":"ROMA_LOGMINER",
                "jdbcUrl":"jdbc:oracle:thin:@10.141.202.*",
                "table":[
                  "TT1"
                ]
              }
            ],
            "writeMode":"insert",
            "batchSize":1024,
            "username":"roma_logminer"
          },
          "name":"oraclewriter",
          "flushIntervalMills":10,
          "batchSize":1024,
          "table":{
            "tableName":"sinkTable"
          }
        }
      }
    ],
    "setting":{
      "restore":{
        "isStream":true,
        "isRestore":true
      },
      "log":{
        "path":"",
        "level":"debug",
        "pattern":"",
        "isLogger":false
      },
      "errorLimit":{

      },
      "speed":{
        "readerChannel":1,
        "writerChannel":1,
        "bytes":0
      }
    }
  }
}

• ORACLE SQL

CREATE TABLE "ROMA_LOGMINER"."SS1" (
  "ID" NUMBER(10,0),
  "F2" FLOAT(32),
  "B2" BLOB,
  "V2" VARCHAR2(255),
  "DT" DATE,
  "C2" CLOB,
  "TT" TIMESTAMP
)

CREATE TABLE "ROMA_LOGMINER"."TT1" (
  "ID" NUMBER(10,0),
  "F2" FLOAT(32),
  "B2" BLOB,
  "V2" VARCHAR2(255),
  "DT" DATE,
  "C2" CLOB,
  "TT" TIMESTAMP
)

[LiDianLEC]

logminer到oracle，clob类型报错

• 报错信息
• CHUNJUN JSON

{
  "job":{
    "content":[
      {
        "reader":{
          "parameter":{
            "password":"*",
            "splitUpdate":true,
            "cat":"update,insert,delete",
            "jdbcUrl":"jdbc:oracle:thin:@//10.141.*",
            "column":[
              {
                "name": "C2",
                "type": "CLOB"
              }
            ],
            "readPosition":"current",
            "pavingData":true,
            "table":[
              "ROMA_LOGMINER.SS1"
            ],
            "username":"roma_logminer"
          },
          "name":"oraclelogminersource",
          "table":{
            "tableName":"sourceTable"
          }
        },
        "transformer":{
          "transformSql":"SELECT C2 FROM sourceTable"
        },
        "writer":{
          "parameter":{
            "password":"*",
            "column":[
              {
                "index": 0,
                "name": "C2",
                "type": "CLOB"
              }
            ],
            "connection":[
              {
                "schema":"ROMA_LOGMINER",
                "jdbcUrl":"jdbc:oracle:thin:@10.141.*",
                "table":[
                  "TT1"
                ]
              }
            ],
            "writeMode":"insert",
            "batchSize":1024,
            "username":"roma_logminer"
          },
          "name":"oraclewriter",
          "flushIntervalMills":10,
          "batchSize":1024,
          "table":{
            "tableName":"sinkTable"
          }
        }
      }
    ],
    "setting":{
      "restore":{
        "isStream":true,
        "isRestore":true
      },
      "log":{
        "path":"",
        "level":"debug",
        "pattern":"",
        "isLogger":false
      },
      "errorLimit":{

      },
      "speed":{
        "readerChannel":1,
        "writerChannel":1,
        "bytes":0
      }
    }
  }
}

[LiDianLEC]

• ORACLE SQL

insert into "ROMA_LOGMINER"."SS1" (ID,F2,V2,DT,TT,C2) VALUES (19,9.99,'VVVV',TO_DATE('2022-09-01 12:22:22', 'yyyy-mm-dd HH:mi:ss'),TO_TIMESTAMP('2022-09-01 12:22:22', 'yyyy-mm-dd HH:mi:ss'),'CCCC')

[LiDianLEC]

2. logminer - mysql

1.1. 数据重复消费两次

1.2. 类型转换失败

1.2.1. NUMBER -> BIGINT

1.2.2. FLOAT -> DOUBLE

1.2.3. DATE -> DATE

• SQL

-- ORACLE
CREATE TABLE "ROMA_LOGMINER"."SS1" (
  "ID" NUMBER(10,0),
  "F2" FLOAT(32),
  "B2" BLOB,
  "V2" VARCHAR2(255),
  "DT" DATE,
  "C2" CLOB,
  "TT" TIMESTAMP
)

insert into "ROMA_LOGMINER"."SS1" (ID,F2,V2,DT,TT) VALUES (1,100.1,'VVVV',TO_DATE('2022-09-01 12:22:22', 'yyyy-mm-dd HH:mi:ss'),TO_TIMESTAMP('2022-09-01 12:22:22', 'yyyy-mm-dd HH:mi:ss'))

-- MYSQL
CREATE TABLE `tt1` (
  `ID` bigint(20) DEFAULT NULL,
  `F2` double DEFAULT NULL,
  `B2` blob,
  `V2` varchar(255) DEFAULT NULL,
  `DT` date DEFAULT NULL,
  `C2` text,
  `TT` timestamp NULL DEFAULT NULL
) ENGINE=InnoDB DEFAULT CHARSET=utf8;

• ORACLE插入数据

• MYSQL实时同步数据

• 错误信息

  • 消费两次数据
  • number->bigint类型同步错误
  • float->double类型同屋错误
  • date->date类型转换错误

• CHUNJUN JSON

{
  "job":{
    "content":[
      {
        "reader":{
          "parameter":{
            "password":"*",
            "splitUpdate":true,
            "cat":"update,insert,delete",
            "jdbcUrl":"jdbc:oracle:thin:@//10.141*d",
            "column":[
              {
                "name": "ID",
                "type": "NUMBER"
              },
              {
                "name": "F2",
                "type": "FLOAT"
              },
              {
                "name": "V2",
                "type": "VARCHAR"
              },
              {
                "name": "DT",
                "type": "DATE"
              },
              {
                "name": "TT",
                "type": "TIMESTAMP"
              }
            ],
            "readPosition":"current",
            "pavingData":true,
            "table":[
              "ROMA_LOGMINER.SS1"
            ],
            "username":"ro*ner"
          },
          "name":"oraclelogminersource",
          "table":{
            "tableName":"sourceTable"
          }
        },
        "transformer":{
          "transformSql":"SELECT ID,F2,V2,DT,TT FROM sourceTable"
        },
        "writer":{
          "parameter":{
            "password":"S*VE",
            "column":[
              {
                "index": 0,
                "name": "ID",
                "type": "BIGINT"
              },
              {
                "index": 1,
                "name": "F2",
                "type": "DOUBLE"
              },
              {
                "index": 2,
                "name": "V2",
                "type": "VARCHAR"
              },
              {
                "index": 3,
                "name": "DT",
                "type": "DATE"
              },
              {
                "index": 4,
                "name": "TT",
                "type": "TIMESTAMP"
              }
            ],
            "connection":[
              {
                "schema":"live",
                "jdbcUrl":"jdbc:mysql://10.1*e?characterEncoding=utf8&allowLoadLocalInfile=false&autoDeserialize=false&allowLocalInfile=false&allowUrlInLocalInfile=false",
                "table":[
                  "TT1"
                ]
              }
            ],
            "writeMode":"update",
            "username":"***"
          },
          "name":"mysqlwriter",
          "flushIntervalMills":10,
          "batchSize":1024,
          "table":{
            "tableName":"sinkTable"
          }
        }
      }
    ],
    "setting":{
      "restore":{
        "isStream":true,
        "isRestore":true
      },
      "log":{
        "path":"",
        "level":"debug",
        "pattern":"",
        "isLogger":false
      },
      "errorLimit":{

      },
      "speed":{
        "readerChannel":1,
        "writerChannel":1,
        "bytes":0
      }
    }
  }
}

[LiDianLEC]

2. logminer - mysql

1.1. 数据重复消费两次

1.2. 类型转换失败

1.2.1. NUMBER -> BIGINT

1.2.2. FLOAT -> DOUBLE

1.2.3. DATE -> DATE

• SQL

-- ORACLE
CREATE TABLE "ROMA_LOGMINER"."SS1" (
  "ID" NUMBER(10,0),
  "F2" FLOAT(32),
  "B2" BLOB,
  "V2" VARCHAR2(255),
  "DT" DATE,
  "C2" CLOB,
  "TT" TIMESTAMP
)

insert into "ROMA_LOGMINER"."SS1" (ID,F2,V2,DT,TT) VALUES (1,100.1,'VVVV',TO_DATE('2022-09-01 12:22:22', 'yyyy-mm-dd HH:mi:ss'),TO_TIMESTAMP('2022-09-01 12:22:22', 'yyyy-mm-dd HH:mi:ss'))

-- MYSQL
CREATE TABLE `tt1` (
  `ID` bigint(20) DEFAULT NULL,
  `F2` double DEFAULT NULL,
  `B2` blob,
  `V2` varchar(255) DEFAULT NULL,
  `DT` date DEFAULT NULL,
  `C2` text,
  `TT` timestamp NULL DEFAULT NULL
) ENGINE=InnoDB DEFAULT CHARSET=utf8;

• ORACLE插入数据

• MYSQL实时同步数据

• 错误信息

  • 消费两次数据
  • number->bigint类型同步错误
  • float->double类型同屋错误
  • date->date类型转换错误

• CHUNJUN JSON

{
  "job":{
    "content":[
      {
        "reader":{
          "parameter":{
            "password":"*",
            "splitUpdate":true,
            "cat":"update,insert,delete",
            "jdbcUrl":"jdbc:oracle:thin:@//10.141*d",
            "column":[
              {
                "name": "ID",
                "type": "NUMBER"
              },
              {
                "name": "F2",
                "type": "FLOAT"
              },
              {
                "name": "V2",
                "type": "VARCHAR"
              },
              {
                "name": "DT",
                "type": "DATE"
              },
              {
                "name": "TT",
                "type": "TIMESTAMP"
              }
            ],
            "readPosition":"current",
            "pavingData":true,
            "table":[
              "ROMA_LOGMINER.SS1"
            ],
            "username":"ro*ner"
          },
          "name":"oraclelogminersource",
          "table":{
            "tableName":"sourceTable"
          }
        },
        "transformer":{
          "transformSql":"SELECT ID,F2,V2,DT,TT FROM sourceTable"
        },
        "writer":{
          "parameter":{
            "password":"S*VE",
            "column":[
              {
                "index": 0,
                "name": "ID",
                "type": "BIGINT"
              },
              {
                "index": 1,
                "name": "F2",
                "type": "DOUBLE"
              },
              {
                "index": 2,
                "name": "V2",
                "type": "VARCHAR"
              },
              {
                "index": 3,
                "name": "DT",
                "type": "DATE"
              },
              {
                "index": 4,
                "name": "TT",
                "type": "TIMESTAMP"
              }
            ],
            "connection":[
              {
                "schema":"live",
                "jdbcUrl":"jdbc:mysql://10.1*e?characterEncoding=utf8&allowLoadLocalInfile=false&autoDeserialize=false&allowLocalInfile=false&allowUrlInLocalInfile=false",
                "table":[
                  "TT1"
                ]
              }
            ],
            "writeMode":"update",
            "username":"***"
          },
          "name":"mysqlwriter",
          "flushIntervalMills":10,
          "batchSize":1024,
          "table":{
            "tableName":"sinkTable"
          }
        }
      }
    ],
    "setting":{
      "restore":{
        "isStream":true,
        "isRestore":true
      },
      "log":{
        "path":"",
        "level":"debug",
        "pattern":"",
        "isLogger":false
      },
      "errorLimit":{

      },
      "speed":{
        "readerChannel":1,
        "writerChannel":1,
        "bytes":0
      }
    }
  }
}