Chemy1220
/
examples
forked from opengaussexamples/examples
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

!16 MySQL到openGauss迁移工具--openGauss赛道2021 

Merge pull request !16 from osinfra/master
This commit is contained in:
opengauss-bot 2021-09-11 08:33:03 +00:00 committed by Gitee
parent 1be96e981f 3b2c627e5a
commit e2c2076829
34 changed files with 9596 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

43
MySQL2openGauss/bin/run.sh Normal file
View File

@ -0,0 +1,43 @@
#!/bin/bash
cd `dirname $0`
cd ..
BASEDIR=`pwd`
# set env
export MYSQL_TO_OPENGAUSS_HOME=$BASEDIR
export LD_LIBRARY_PATH=$BASEDIR/lib/unixODBC/unixODBC-2.3.9/lib:$BASEDIR/unixODBC-2.3.9/lib:$BASEDIR/lib:$LD_LIBRARY_PATH
export ODBCSYSINI=$BASEDIR/conf
export ODBCINI=$BASEDIR/conf/odbc.ini
export CREATE_TABLE_REPLACE_KV=$BASEDIR/conf/createtablerkv.prop
source $BASEDIR/conf/mysql2opengauss.prop
source $BASEDIR/conf/metrics.prop
# odbc.ini
sed "s:OPENGAUSS_SERVERNAME:$opengauss_host:g" $BASEDIR/conf/odbc.ini.template > $BASEDIR/conf/odbc.ini
sed -i "s:OPENGAUSS_PORT:$opengauss_port:g" $BASEDIR/conf/odbc.ini
sed -i "s:OPENGAUSS_USERNAME:$opengauss_username:g" $BASEDIR/conf/odbc.ini
sed -i "s:OPENGAUSS_PASSWORD:$opengauss_password:g" $BASEDIR/conf/odbc.ini
sed -i "s:OPENGAUSS_DATABASE:$opengauss_database:g" $BASEDIR/conf/odbc.ini
sed -i "s:METRICS_SERVERNAME:$metrics_host:g" $BASEDIR/conf/odbc.ini
sed -i "s:METRICS_PORT:$metrics_port:g" $BASEDIR/conf/odbc.ini
sed -i "s:METRICS_USERNAME:$metrics_username:g" $BASEDIR/conf/odbc.ini
sed -i "s:METRICS_PASSWORD:$metrics_password:g" $BASEDIR/conf/odbc.ini
sed -i "s:METRICS_DATABASE:$metrics_database:g" $BASEDIR/conf/odbc.ini
export metrics_dbtype
# odbcinst.ini
sed "s:BASEDIR:$BASEDIR:g" $BASEDIR/conf/odbcinst.ini.template > $BASEDIR/conf/odbcinst.ini
# run
cd bin
./mysqltoopengauss
if [ "$?" == "0" ];
then
echo success.
else
echo failed to migrate mysql data!!!
fi

54
MySQL2openGauss/build_aarch64.sh Normal file
View File

@ -0,0 +1,54 @@
#!/bin/bash
cd `dirname $0`
BASEDIR=`pwd`
build_error () {
echo build error !!!
exit 1
}
# unixODBC
cd lib/unixODBC
tar -xf unixODBC-2.3.9-aarch64.tar.gz
cd $BASEDIR
# mysql
cd mysqlsrc
tar -xf mysql-boost-5.7.27.tar.gz
cd mysql-5.7.27
ln -s $BASEDIR/lib/unixODBC unixODBC
rm -f ./client/CMakeLists.txt
cp -f $BASEDIR/cmake/mysql-5.7.27/client/CMakeLists.txt ./client/
cp -f $BASEDIR/src/* ./client/
cmake . -DCMAKE_INSTALL_PREFIX=$BASEDIR/mysql5727 -DWITH_BOOST=$BASEDIR/mysqlsrc/mysql-5.7.27/boost/boost_1_59_0
make -j2
# ignore error and make again
make
if [ "$?" != "0" ]; then
build_error
fi
rm -f $BASEDIR/bin/mysqltoopengauss
cp -f $BASEDIR/mysqlsrc/mysql-5.7.27/client/mysqltoopengauss $BASEDIR/bin/
cd $BASEDIR
# libs
cd lib
tar -xf depends-aarch64.tar.gz
cd $BASEDIR
# pack
cd bin
chmod +x *
cd ..
rm -rf output/mysqltoopengauss
mkdir -p output/mysqltoopengauss
cp -rf bin output/mysqltoopengauss
cp -rf conf output/mysqltoopengauss
cp -rf lib output/mysqltoopengauss
rm -rf output/mysqltoopengauss/lib/unixODBC/*.tar.gz
rm -rf output/mysqltoopengauss/lib/*.tar.gz
cd output
chmod +x mysqltoopengauss/bin/*
tar -czf mysqltoopengauss.tar.gz mysqltoopengauss
rm -rf mysqltoopengauss
echo mysqltoopengauss is pack to `pwd`/mysqltoopengauss.tar.gz

54
MySQL2openGauss/build_x86.sh Normal file
View File

@ -0,0 +1,54 @@
#!/bin/bash
cd `dirname $0`
BASEDIR=`pwd`
build_error () {
echo build error !!!
exit 1
}
# unixODBC
cd lib/unixODBC
tar -xf unixODBC-2.3.9.tar.gz
cd $BASEDIR
# mysql
cd mysqlsrc
tar -xf mysql-boost-5.7.27.tar.gz
cd mysql-5.7.27
ln -s $BASEDIR/lib/unixODBC unixODBC
rm -f ./client/CMakeLists.txt
cp -f $BASEDIR/cmake/mysql-5.7.27/client/CMakeLists.txt ./client/
cp -f $BASEDIR/src/* ./client/
cmake . -DCMAKE_INSTALL_PREFIX=$BASEDIR/mysql5727 -DWITH_BOOST=$BASEDIR/mysqlsrc/mysql-5.7.27/boost/boost_1_59_0
make -j2
# ignore error and make again
make
if [ "$?" != "0" ]; then
build_error
fi
rm -f $BASEDIR/bin/mysqltoopengauss
cp -f $BASEDIR/mysqlsrc/mysql-5.7.27/client/mysqltoopengauss $BASEDIR/bin/
cd $BASEDIR
# libs
cd lib
tar -xf depends.tar.gz
cd $BASEDIR
# pack
cd bin
chmod +x *
cd ..
rm -rf output/mysqltoopengauss
mkdir -p output/mysqltoopengauss
cp -rf bin output/mysqltoopengauss
cp -rf conf output/mysqltoopengauss
cp -rf lib output/mysqltoopengauss
rm -rf output/mysqltoopengauss/lib/unixODBC/*.tar.gz
rm -rf output/mysqltoopengauss/lib/*.tar.gz
cd output
chmod +x mysqltoopengauss/bin/*
tar -czf mysqltoopengauss.tar.gz mysqltoopengauss
rm -rf mysqltoopengauss
echo mysqltoopengauss is pack to `pwd`/mysqltoopengauss.tar.gz

143
MySQL2openGauss/cmake/mysql-5.7.27/client/CMakeLists.txt Normal file
View File

@ -0,0 +1,143 @@
# Copyright (c) 2006, 2018, Oracle and/or its affiliates. All rights reserved.
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; version 2 of the License.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA
INCLUDE_DIRECTORIES(
${CMAKE_SOURCE_DIR}/client
${CMAKE_SOURCE_DIR}/include
${CMAKE_SOURCE_DIR}/mysys_ssl
${LZ4_INCLUDE_DIR}
${SSL_INCLUDE_DIRS}
${CMAKE_SOURCE_DIR}/libmysql
${CMAKE_SOURCE_DIR}/libbinlogevents/include
${CMAKE_SOURCE_DIR}/regex
${CMAKE_SOURCE_DIR}/strings
${EDITLINE_INCLUDE_DIR}
${CMAKE_CURRENT_BINARY_DIR}
)
INCLUDE(${MYSQL_CMAKE_SCRIPT_DIR}/compile_flags.cmake)
## Subdirectory with common client code.
ADD_SUBDIRECTORY(base)
## Subdirectory for mysqlpump code.
ADD_SUBDIRECTORY(dump)
## We will need libeay32.dll and ssleay32.dll when running client executables.
COPY_OPENSSL_DLLS(copy_openssl_client)
INCLUDE(${MYSQL_CMAKE_SCRIPT_DIR}/compile_flags.cmake)
ADD_DEFINITIONS(${SSL_DEFINES})
MYSQL_ADD_EXECUTABLE(mysql completion_hash.cc mysql.cc readline.cc ../sql-common/sql_string.cc)
TARGET_LINK_LIBRARIES(mysql mysqlclient)
IF(UNIX)
TARGET_LINK_LIBRARIES(mysql ${EDITLINE_LIBRARY})
ENDIF(UNIX)
IF(NOT WITHOUT_SERVER)
MYSQL_ADD_EXECUTABLE(mysql_upgrade
upgrade/program.cc
)
ADD_COMPILE_FLAGS(
upgrade/program.cc COMPILE_FLAGS -I${BOOST_PATCHES_DIR} -I${BOOST_INCLUDE_DIR}
)
TARGET_LINK_LIBRARIES(mysql_upgrade mysqlclient client_base mysqlcheck_core)
ADD_DEPENDENCIES(mysql_upgrade GenFixPrivs GenSysSchema)
ENDIF()
MYSQL_ADD_EXECUTABLE(mysqltest mysqltest.cc COMPONENT Test)
SET_SOURCE_FILES_PROPERTIES(mysqltest.cc PROPERTIES COMPILE_FLAGS "-DTHREADS")
TARGET_LINK_LIBRARIES(mysqltest mysqlclient regex)
ADD_DEPENDENCIES(mysqltest GenError)
ADD_CONVENIENCE_LIBRARY(mysqlcheck_core check/mysqlcheck_core.cc)
TARGET_LINK_LIBRARIES(mysqlcheck_core mysqlclient)
MYSQL_ADD_EXECUTABLE(mysqlcheck check/mysqlcheck.cc)
TARGET_LINK_LIBRARIES(mysqlcheck mysqlcheck_core)
MYSQL_ADD_EXECUTABLE(mysqldump mysqldump.c ../sql-common/my_user.c)
TARGET_LINK_LIBRARIES(mysqldump mysqlclient)
LINK_DIRECTORIES(${CMAKE_SOURCE_DIR}/unixODBC/unixODBC-2.3.9/lib)
MYSQL_ADD_EXECUTABLE(mysqltoopengauss mysqltoopengauss.cpp ProcOdbc.cpp ProcMetrics.cpp ../sql-common/my_user.c)
ADD_COMPILE_FLAGS(
ProcOdbc.cpp
COMPILE_FLAGS "-I${CMAKE_SOURCE_DIR}/unixODBC/unixODBC-2.3.9/include"
)
TARGET_LINK_LIBRARIES(mysqltoopengauss mysqlclient odbc)
MYSQL_ADD_EXECUTABLE(mysqlimport mysqlimport.c)
SET_SOURCE_FILES_PROPERTIES(mysqlimport.c PROPERTIES COMPILE_FLAGS "-DTHREADS")
TARGET_LINK_LIBRARIES(mysqlimport mysqlclient)
MYSQL_ADD_EXECUTABLE(mysqlshow mysqlshow.c)
TARGET_LINK_LIBRARIES(mysqlshow mysqlclient)
MYSQL_ADD_EXECUTABLE(mysql_plugin mysql_plugin.c)
TARGET_LINK_LIBRARIES(mysql_plugin mysqlclient)
MYSQL_ADD_EXECUTABLE(mysqlbinlog mysqlbinlog.cc)
ADD_COMPILE_FLAGS(
mysqlbinlog.cc
COMPILE_FLAGS "-I${CMAKE_SOURCE_DIR}/sql" "-DHAVE_REPLICATION" "-DDISABLE_PSI_MUTEX"
)
TARGET_LINK_LIBRARIES(mysqlbinlog mysqlclient binlogevents_static)
MYSQL_ADD_EXECUTABLE(mysqladmin mysqladmin.cc)
TARGET_LINK_LIBRARIES(mysqladmin mysqlclient)
MYSQL_ADD_EXECUTABLE(mysqlslap mysqlslap.cc)
SET_SOURCE_FILES_PROPERTIES(mysqlslap.cc PROPERTIES COMPILE_FLAGS "-DTHREADS")
TARGET_LINK_LIBRARIES(mysqlslap mysqlclient)
MYSQL_ADD_EXECUTABLE(mysql_config_editor mysql_config_editor.cc)
TARGET_LINK_LIBRARIES(mysql_config_editor mysqlclient)
MYSQL_ADD_EXECUTABLE(mysql_secure_installation mysql_secure_installation.cc)
TARGET_LINK_LIBRARIES(mysql_secure_installation mysqlclient)
IF(UNIX AND NOT WITHOUT_SERVER)
MYSQL_ADD_EXECUTABLE(mysql_install_db
mysql_install_db.cc auth_utils.cc path.cc logger.cc)
TARGET_LINK_LIBRARIES(mysql_install_db mysqlclient)
ADD_COMPILE_FLAGS(
auth_utils.cc mysql_install_db.cc
COMPILE_FLAGS "-I${CMAKE_SOURCE_DIR}/sql/auth"
)
ADD_DEPENDENCIES(mysql_install_db GenBootstrapPriv GenSysSchema)
ENDIF()
MYSQL_ADD_EXECUTABLE(mysql_ssl_rsa_setup mysql_ssl_rsa_setup.cc path.cc logger.cc)
TARGET_LINK_LIBRARIES(mysql_ssl_rsa_setup mysys mysys_ssl)
SET_TARGET_PROPERTIES(mysql_ssl_rsa_setup PROPERTIES LINKER_LANGUAGE CXX)
# "WIN32" also covers 64 bit. "echo" is used in some files below "mysql-test/".
IF(WIN32)
MYSQL_ADD_EXECUTABLE(echo echo.c)
ENDIF(WIN32)
SET_TARGET_PROPERTIES (
mysql_plugin
mysqlcheck
mysqldump
mysqltoopengauss
mysqlimport
mysqlshow
mysqlslap
PROPERTIES HAS_CXX TRUE)
ADD_DEFINITIONS(-DHAVE_DLOPEN)

4
MySQL2openGauss/conf/createfunctionrkv.prop Normal file
View File

@ -0,0 +1,4 @@
) CHARSET latin1_bin=)
) CHARSET latin1=)
) CHARSET utf8_bin=)
) CHARSET utf8=)

26
MySQL2openGauss/conf/createfunctiontkv.prop Normal file
View File

@ -0,0 +1,26 @@
RETURNS longtext=RETURNS text
RETURNS tinyint(=RETURNS integer(
RETURNS bigint(=RETURNS integer(
RETURNS datetime(3)=RETURNS timestamp
RETURNS datetime=RETURNS timestamp
RETURNS tinytext=RETURNS text
RETURNS longblob=RETURNS bytea
RETURNS mediumtext=RETURNS text
RETURNS mediumint(=RETURNS integer(
RETURNS smallint(=RETURNS integer(
RETURNS tinyblob=RETURNS bytea
RETURNS mediumblob=RETURNS bytea
RETURNS varbinary(=RETURNS bytea(
RETURNS binary(=RETURNS bytea(
RETURNS decimal(=RETURNS numeric(
RETURNS double=RETURNS double precision
RETURNS float=RETURNS double precision
RETURNS bit(=RETURNS integer(
RETURNS year(=RETURNS integer(
RETURNS geometrycollection=RETURNS bytea
RETURNS geometry=RETURNS point
RETURNS linestring=RETURNS path
RETURNS multipoint=RETURNS bytea
RETURNS multilinestring=RETURNS bytea
RETURNS multipolygon=RETURNS bytea
RETURNS int(=RETURNS integer(

45
MySQL2openGauss/conf/createtablerkv.prop Normal file
View File

@ -0,0 +1,45 @@
CHARACTER SET utf8 COLLATE utf8_bin=
CHARACTER SET utf8=
COLLATE utf8_bin=
COLLATE utf8_general_ci=
CHARACTER SET latin1 COLLATE latin1_bin=
CHARACTER SET latin1=
COLLATE latin1_bin=
AUTO_INCREMENT=
" longtext=" text
tinyint(=integer(
bigint(=integer(
datetime(3)=timestamp without time zone
" datetime=" timestamp without time zone
" timestamp,=" timestamp without time zone,
" timestamp NULL DEFAULT NULL,=" timestamp without time zone NULL,
" timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP,=" timestamp without time zone NOT NULL,
" timestamp NOT NULL DEFAULT '0000-00-00 00:00:00',=" timestamp without time zone NOT NULL,
" tinytext=" text
" longblob=" bytea
" mediumtext=" text
mediumint(=integer(
smallint(=integer(
" time,=" time without time zone,
" time NOT NULL,=" time without time zone NOT NULL,
" time DEFAULT NULL,=" time without time zone DEFAULT NULL,
" tinyblob=" bytea
" mediumblob=" bytea
varbinary(=bytea(
binary(=bytea(
decimal(=numeric(
ON UPDATE CURRENT_TIMESTAMP=
" double,=" double precision,
" double DEFAULT=" double precision DEFAULT
" double NOT NULL=" double precision NOT NULL
" float=" double precision
bit(=integer(
year(=integer(
" geometrycollection=" bytea
" geometry=" point
" linestring=" path
" multipoint=" bytea
" multilinestring=" bytea
" multipolygon=" bytea
int(=integer(

6
MySQL2openGauss/conf/metrics.prop Normal file
View File

@ -0,0 +1,6 @@
metrics_host=127.0.0.1
metrics_port=3306
metrics_username=root
metrics_password=password
metrics_database=metricsdb
metrics_dbtype=mysql

11
MySQL2openGauss/conf/mysql2opengauss.prop Normal file
View File

@ -0,0 +1,11 @@
mysql_host=127.0.0.1
mysql_port=3306
mysql_username=username
mysql_password=password
mysql_database=database
opengauss_host=127.0.0.1
opengauss_port=5432
opengauss_username=opengauss
opengauss_password=password
opengauss_database=database
loglevel=4

28
MySQL2openGauss/conf/odbc.ini.template Normal file
View File

@ -0,0 +1,28 @@
[opengauss]
Driver=opengauss
Servername=OPENGAUSS_SERVERNAME
Database=OPENGAUSS_DATABASE
Username=OPENGAUSS_USERNAME
Password=OPENGAUSS_PASSWORD
Port=OPENGAUSS_PORT
Sslmode=disable
[metrics_mysql]
Driver=mysql
SERVER=METRICS_SERVERNAME
PORT=METRICS_PORT
USER=METRICS_USERNAME
Password=METRICS_PASSWORD
Database=METRICS_DATABASE
OPTION=3
Threading=1
ClientCharSet=UTF8
[metrics_opengauss]
Driver=opengauss
Servername=METRICS_SERVERNAME
Database=METRICS_DATABASE
Username=METRICS_USERNAME
Password=METRICS_PASSWORD
Port=METRICS_PORT
Sslmode=disable

7
MySQL2openGauss/conf/odbcinst.ini.template Normal file
View File

@ -0,0 +1,7 @@
[opengauss]
Driver64=BASEDIR/lib/psqlodbcw.so
setup=BASEDIR/lib/psqlodbcw.so
[mysql]
Driver64=BASEDIR/lib/libmyodbc5w.so
setup=BASEDIR/lib/libmyodbc5S.so

1293
MySQL2openGauss/grafana/samplemodel.json Normal file
View File

File diff suppressed because it is too large Load Diff

BIN
MySQL2openGauss/lib/depends-aarch64.tar.gz Normal file
View File

Binary file not shown.

1
MySQL2openGauss/lib/depends-aarch64.tar.gz.txt Normal file
View File

@ -0,0 +1 @@
把depends-aarch64.tar.gz放到此目录

BIN
MySQL2openGauss/lib/depends.tar.gz Normal file
View File

Binary file not shown.

1
MySQL2openGauss/lib/depends.tar.gz.txt Normal file
View File

@ -0,0 +1 @@
把depends.tar.gz放到此目录

BIN
MySQL2openGauss/lib/libmyodbc5S.so Normal file
View File

Binary file not shown.

BIN
MySQL2openGauss/lib/libmyodbc5w.so Normal file
View File

Binary file not shown.

BIN
MySQL2openGauss/lib/unixODBC/unixODBC-2.3.9-aarch64.tar.gz Normal file
View File

Binary file not shown.

1
MySQL2openGauss/lib/unixODBC/unixODBC-2.3.9-aarch64.tar.gz.txt Normal file
View File

@ -0,0 +1 @@
把unixODBC-2.3.9-aarch64.tar.gz放到此目录

BIN
MySQL2openGauss/lib/unixODBC/unixODBC-2.3.9.tar.gz Normal file
View File

Binary file not shown.

1
MySQL2openGauss/lib/unixODBC/unixODBC-2.3.9.tar.gz.txt Normal file
View File

@ -0,0 +1 @@
把unixODBC-2.3.9.tar.gz放到此目录

BIN
MySQL2openGauss/mysql2opengauss.pdf Normal file
View File

Binary file not shown.

BIN
MySQL2openGauss/mysqlsrc/mysql-boost-5.7.27.tar.gz Normal file
View File

Binary file not shown.

1
MySQL2openGauss/mysqlsrc/mysql-boost-5.7.27.tar.gz.txt Normal file
View File

@ -0,0 +1 @@
把mysql-boost-5.7.27.tar.gz放到此目录

BIN
MySQL2openGauss/mysqltoopengauss.centos7.x86_64.tar.gz Normal file
View File

Binary file not shown.

BIN
MySQL2openGauss/mysqltoopengauss.openeuler.aarch64.tar.gz Normal file
View File

Binary file not shown.

100
MySQL2openGauss/sql/metrics_mysql.sql Normal file
View File

@ -0,0 +1,100 @@
CREATE TABLE `db_info_metrics` (
`procpid` int(11) NOT NULL,
`proctime` datetime(3) NOT NULL,
`srcaddr` varchar(255),
`dstaddr` varchar(255),
`srcdb` varchar(255),
`dstdb` varchar(255),
`starttime` datetime(3),
`endtime` datetime(3),
`consumetime` int(15),
`tablecount` int(11),
`viewcount` int(11),
`procedurecount` int(11),
`functioncount` int(11),
`totalcount` int(15),
`rightcount` int(15),
`errorcount` int(15),
`insertrightcount` int(15),
`inserterrorcount` int(15),
PRIMARY KEY (`procpid`, `proctime`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8;
CREATE TABLE `time_db_info_metrics` (
`updatetime` datetime(3) NOT NULL,
`procpid` int(11) NOT NULL,
`proctime` datetime(3) NOT NULL,
`srcaddr` varchar(255),
`dstaddr` varchar(255),
`srcdb` varchar(255),
`dstdb` varchar(255),
`starttime` datetime(3),
`endtime` datetime(3),
`consumetime` int(15),
`tablecount` int(11),
`viewcount` int(11),
`procedurecount` int(11),
`functioncount` int(11),
`totalcount` int(15),
`rightcount` int(15),
`errorcount` int(15),
`insertrightcount` int(15),
`inserterrorcount` int(15),
PRIMARY KEY (`updatetime`, `procpid`, `proctime`, `endtime`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8;
CREATE TABLE `table_info_metrics` (
`procpid` int(11) NOT NULL,
`proctime` datetime(3) NOT NULL,
`tablename` varchar(255) NOT NULL,
`srcaddr` varchar(255),
`dstaddr` varchar(255),
`srcdb` varchar(255),
`dstdb` varchar(255),
`starttime` datetime(3),
`endtime` datetime(3),
`consumetime` int(15),
`totalcount` int(15),
`rightcount` int(15),
`errorcount` int(15),
`insertrightcount` int(15),
`inserterrorcount` int(15),
PRIMARY KEY (`procpid`, `proctime`, `tablename`, `starttime`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8;
CREATE TABLE `time_table_info_metrics` (
`updatetime` datetime(3) NOT NULL,
`procpid` int(11) NOT NULL,
`proctime` datetime(3) NOT NULL,
`tablename` varchar(255) NOT NULL,
`srcaddr` varchar(255),
`dstaddr` varchar(255),
`srcdb` varchar(255),
`dstdb` varchar(255),
`starttime` datetime(3) NOT NULL,
`endtime` datetime(3),
`consumetime` int(15),
`totalcount` int(15),
`rightcount` int(15),
`errorcount` int(15),
`insertrightcount` int(15),
`inserterrorcount` int(15),
PRIMARY KEY (`updatetime`, `procpid`, `proctime`, `tablename`, `starttime`, `endtime`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8;
CREATE TABLE `sql_info_metrics` (
`procpid` int(11) NOT NULL,
`proctime` datetime(3) NOT NULL,
`name` varchar(255) NOT NULL,
`type` varchar(255) NOT NULL,
`srcaddr` varchar(255),
`dstaddr` varchar(255),
`srcdb` varchar(255),
`dstdb` varchar(255),
`starttime` datetime(3),
`endtime` datetime(3),
`consumetime` int(15),
`iserror` int(1),
`errorinfo` varchar(255),
`sqldata` mediumtext
) ENGINE=InnoDB DEFAULT CHARSET=utf8;

100
MySQL2openGauss/sql/metrics_opengauss.sql Normal file
View File

@ -0,0 +1,100 @@
CREATE TABLE "db_info_metrics" (
"procpid" integer(11) NOT NULL,
"proctime" timestamp without time zone NOT NULL,
"srcaddr" varchar(255),
"dstaddr" varchar(255),
"srcdb" varchar(255),
"dstdb" varchar(255),
"starttime" timestamp without time zone,
"endtime" timestamp without time zone,
"consumetime" integer(15),
"tablecount" integer(11),
"viewcount" integer(11),
"procedurecount" integer(11),
"functioncount" integer(11),
"totalcount" integer(15),
"rightcount" integer(15),
"errorcount" integer(15),
"insertrightcount" integer(15),
"inserterrorcount" integer(15),
PRIMARY KEY ("procpid", "proctime")
);
CREATE TABLE "time_db_info_metrics" (
"updatetime" timestamp without time zone NOT NULL,
"procpid" integer(11) NOT NULL,
"proctime" timestamp without time zone NOT NULL,
"srcaddr" varchar(255),
"dstaddr" varchar(255),
"srcdb" varchar(255),
"dstdb" varchar(255),
"starttime" timestamp without time zone,
"endtime" timestamp without time zone,
"consumetime" integer(15),
"tablecount" integer(11),
"viewcount" integer(11),
"procedurecount" integer(11),
"functioncount" integer(11),
"totalcount" integer(15),
"rightcount" integer(15),
"errorcount" integer(15),
"insertrightcount" integer(15),
"inserterrorcount" integer(15),
PRIMARY KEY ("updatetime", "procpid", "proctime", "endtime")
);
CREATE TABLE "table_info_metrics" (
"procpid" integer(11) NOT NULL,
"proctime" timestamp without time zone NOT NULL,
"tablename" varchar(255) NOT NULL,
"srcaddr" varchar(255),
"dstaddr" varchar(255),
"srcdb" varchar(255),
"dstdb" varchar(255),
"starttime" timestamp without time zone,
"endtime" timestamp without time zone,
"consumetime" integer(15),
"totalcount" integer(15),
"rightcount" integer(15),
"errorcount" integer(15),
"insertrightcount" integer(15),
"inserterrorcount" integer(15),
PRIMARY KEY ("procpid", "proctime", "tablename", "starttime")
);
CREATE TABLE "time_table_info_metrics" (
"updatetime" timestamp without time zone NOT NULL,
"procpid" integer(11) NOT NULL,
"proctime" timestamp without time zone NOT NULL,
"tablename" varchar(255) NOT NULL,
"srcaddr" varchar(255),
"dstaddr" varchar(255),
"srcdb" varchar(255),
"dstdb" varchar(255),
"starttime" timestamp without time zone NOT NULL,
"endtime" timestamp without time zone,
"consumetime" integer(15),
"totalcount" integer(15),
"rightcount" integer(15),
"errorcount" integer(15),
"insertrightcount" integer(15),
"inserterrorcount" integer(15),
PRIMARY KEY ("updatetime", "procpid", "proctime", "tablename", "starttime", "endtime")
);
CREATE TABLE "sql_info_metrics" (
"procpid" integer(11) NOT NULL,
"proctime" timestamp without time zone NOT NULL,
"name" varchar(255) NOT NULL,
"type" varchar(255) NOT NULL,
"srcaddr" varchar(255),
"dstaddr" varchar(255),
"srcdb" varchar(255),
"dstdb" varchar(255),
"starttime" timestamp without time zone,
"endtime" timestamp without time zone,
"consumetime" integer(15),
"iserror" integer(1),
"errorinfo" varchar(255),
"sqldata" text
);

734
MySQL2openGauss/src/ProcMetrics.cpp Normal file
View File

@ -0,0 +1,734 @@
#include <stdio.h>
#include <stdlib.h>
#ifdef WIN32
#include <sys/timeb.h>
#else
#include <sys/time.h>
#endif
#include <string.h>
#include <unistd.h>
#include <fstream>
#include <iostream>
#include <list>
#include <map>
#include "ProcMetrics.h"
#define OPENGAUSS_DBTYPE "opengauss"
#define MYSQL_DBTYPE "mysql"
#define OPENGAUSS_DB 1
#define MYSQL_DB 2
#define INSERT_KEY "),("
#define INSERT_KEY_LEN 3
namespace DMODBC {
static int sg_dbtype;
static int countInsertSql(const char* sql) {
int ret = 1;
sql = strstr(sql, INSERT_KEY);
while (sql) {
ret++;
sql = strstr(sql + INSERT_KEY_LEN, INSERT_KEY);
}
return ret;
}
Synchronized::Synchronized(Mutex& m) : mutex_(&m) { mutex_->lock(); }
Synchronized::~Synchronized() { mutex_->unlock(); }
Mutex::Mutex() {
#ifdef WIN32
InitializeCriticalSection(&lock_);
#else
pthread_mutex_init(&lock_, NULL);
#endif
}
Mutex::~Mutex() {
#ifdef WIN32
DeleteCriticalSection(&lock_);
#else
pthread_mutex_destroy(&lock_);
#endif
}
void Mutex::lock() {
#ifdef WIN32
EnterCriticalSection(&lock_);
#else
pthread_mutex_lock(&lock_);
#endif
}
void Mutex::unlock() {
#ifdef WIN32
LeaveCriticalSection(&lock_);
#else
pthread_mutex_unlock(&lock_);
#endif
}
MetricInfo::~MetricInfo() {}
SqlInfo::SqlInfo()
: baseinfo_(NULL), starttime_(0), endtime_(0), error_(false) {}
SqlInfo::~SqlInfo() {}
std::string SqlInfo::toSql() {
std::string ret;
if (error_) {
char* tmp;
int len = 1024;
if (sql_.size() > errordetail_.size()) {
if (len < (int)sql_.size() * 2) {
len = (int)sql_.size() * 2;
}
} else {
if (len < (int)errordetail_.size() * 2) {
len = (int)errordetail_.size() * 2;
}
}
tmp = (char*)malloc(len);
switch (sg_dbtype) {
case OPENGAUSS_DB:
ret =
"insert into sql_info_metrics(procpid, proctime, name, type, "
"srcaddr, "
"dstaddr, srcdb, "
"dstdb, starttime, endtime, consumetime, iserror, errorinfo, "
"sqldata) "
"values(";
sprintf(tmp,
"%d, TO_TIMESTAMP(%d), '%s', '%s', '%s', '%s', '%s', '%s', "
"TO_TIMESTAMP(%d), "
"TO_TIMESTAMP(%d), %d, %d, '",
baseinfo_->procpid, (int)(baseinfo_->proctime / 1000),
name_.c_str(), type_.c_str(), baseinfo_->srcaddr.c_str(),
baseinfo_->dstaddr.c_str(), baseinfo_->srcdb.c_str(),
baseinfo_->dstdb.c_str(), (int)(starttime_ / 1000),
(int)(endtime_ / 1000), (int)(endtime_ - starttime_),
error_ ? 1 : 0);
break;
case MYSQL_DB:
ret =
"insert into sql_info_metrics(procpid, proctime, name, type, "
"srcaddr, "
"dstaddr, srcdb, "
"dstdb, starttime, endtime, consumetime, iserror, errorinfo, "
"sqldata) "
"values(";
sprintf(tmp,
"%d, FROM_UNIXTIME(%d, '%%Y-%%m-%%d %%H-%%i-%%s'), '%s', '%s', "
"'%s', "
"'%s', '%s', '%s', "
"FROM_UNIXTIME(%d, '%%Y-%%m-%%d %%H-%%i-%%s'), "
"FROM_UNIXTIME(%d, '%%Y-%%m-%%d %%H-%%i-%%s'), %d, %d, '",
baseinfo_->procpid, (int)(baseinfo_->proctime / 1000),
name_.c_str(), type_.c_str(), baseinfo_->srcaddr.c_str(),
baseinfo_->dstaddr.c_str(), baseinfo_->srcdb.c_str(),
baseinfo_->dstdb.c_str(), (int)(starttime_ / 1000),
(int)(endtime_ / 1000), (int)(endtime_ - starttime_),
error_ ? 1 : 0);
break;
default:
// 不支持其他类型
return "";
}
ret += tmp;
formatSql(errordetail_.c_str(), errordetail_.size(), tmp);
ret += tmp;
ret += "', '";
formatSql(sql_.c_str(), sql_.size(), tmp);
ret += tmp;
ret += "')";
free(tmp);
}
return ret;
}
void SqlInfo::formatSql(const char* in, int inlen, char* out) {
int outindex = 0;
for (int index = 0; index < inlen; index++) {
if (in[index] == '\'') {
out[outindex] = '\'';
out[outindex + 1] = '\'';
outindex += 2;
} else {
out[outindex] = in[index];
outindex++;
}
}
out[outindex] = '\0';
}
TableInfo::TableInfo()
: baseinfo_(NULL),
starttime_(0),
endtime_(0),
rightcount_(0),
errorcount_(0),
insertrightcount_(0),
inserterrorcount_(0) {}
TableInfo::~TableInfo() {}
std::string TableInfo::toSql() {
std::string ret;
char tmp[1024];
switch (sg_dbtype) {
case OPENGAUSS_DB:
sprintf(tmp,
"%d, TO_TIMESTAMP(%d), '%s', '%s', '%s', '%s', '%s', "
"TO_TIMESTAMP(%d), "
"TO_TIMESTAMP(%d), %d, %d, %d, %d, %d, %d",
baseinfo_->procpid, (int)(baseinfo_->proctime / 1000),
table_.c_str(), baseinfo_->srcaddr.c_str(),
baseinfo_->dstaddr.c_str(), baseinfo_->srcdb.c_str(),
baseinfo_->dstdb.c_str(), (int)(starttime_ / 1000),
(int)(endtime_ / 1000), (int)(endtime_ - starttime_),
(rightcount_ + errorcount_), rightcount_, errorcount_,
insertrightcount_, inserterrorcount_);
break;
case MYSQL_DB:
sprintf(
tmp,
"%d, FROM_UNIXTIME(%d, '%%Y-%%m-%%d %%H-%%i-%%s'), '%s', '%s', '%s', "
"'%s', '%s', "
"FROM_UNIXTIME(%d, '%%Y-%%m-%%d %%H-%%i-%%s'), "
"FROM_UNIXTIME(%d, '%%Y-%%m-%%d %%H-%%i-%%s'), %d, %d, %d, %d, %d, "
"%d",
baseinfo_->procpid, (int)(baseinfo_->proctime / 1000), table_.c_str(),
baseinfo_->srcaddr.c_str(), baseinfo_->dstaddr.c_str(),
baseinfo_->srcdb.c_str(), baseinfo_->dstdb.c_str(),
(int)(starttime_ / 1000), (int)(endtime_ / 1000),
(int)(endtime_ - starttime_), (rightcount_ + errorcount_),
rightcount_, errorcount_, insertrightcount_, inserterrorcount_);
break;
default:
// 不支持其他类型
return "";
}
ret =
"insert into table_info_metrics(procpid, proctime, tablename, srcaddr, "
"dstaddr, srcdb, "
"dstdb, starttime, endtime, consumetime, totalcount, rightcount, "
"errorcount, insertrightcount, inserterrorcount) values(";
ret += tmp;
ret += ")";
return ret;
}
TimeTableInfo::TimeTableInfo(long long updatetime, TableInfo* tableInfo) {
baseinfo_ = tableInfo->baseinfo_;
updatetime_ = updatetime;
table_ = tableInfo->table_;
starttime_ = tableInfo->starttime_;
endtime_ = tableInfo->endtime_;
rightcount_ = tableInfo->rightcount_;
errorcount_ = tableInfo->errorcount_;
insertrightcount_ = tableInfo->insertrightcount_;
inserterrorcount_ = tableInfo->inserterrorcount_;
}
TimeTableInfo::TimeTableInfo(TableInfo* tableInfo) {
baseinfo_ = tableInfo->baseinfo_;
updatetime_ = ProcMetrics::getCurrentTime();
table_ = tableInfo->table_;
starttime_ = tableInfo->starttime_;
endtime_ = tableInfo->endtime_;
rightcount_ = tableInfo->rightcount_;
errorcount_ = tableInfo->errorcount_;
insertrightcount_ = tableInfo->insertrightcount_;
inserterrorcount_ = tableInfo->inserterrorcount_;
}
TimeTableInfo::~TimeTableInfo() {}
std::string TimeTableInfo::toSql() {
std::string ret;
char tmp[1024];
switch (sg_dbtype) {
case OPENGAUSS_DB:
sprintf(tmp,
"TO_TIMESTAMP(%d), %d, TO_TIMESTAMP(%d), '%s', '%s', '%s', '%s', "
"'%s', "
"TO_TIMESTAMP(%d), "
"TO_TIMESTAMP(%d), %d, %d, %d, %d, %d, %d",
(int)(updatetime_ / 1000), baseinfo_->procpid,
(int)(baseinfo_->proctime / 1000), table_.c_str(),
baseinfo_->srcaddr.c_str(), baseinfo_->dstaddr.c_str(),
baseinfo_->srcdb.c_str(), baseinfo_->dstdb.c_str(),
(int)(starttime_ / 1000), (int)(endtime_ / 1000),
(int)(updatetime_ - starttime_), (rightcount_ + errorcount_),
rightcount_, errorcount_, insertrightcount_, inserterrorcount_);
break;
case MYSQL_DB:
sprintf(tmp,
"FROM_UNIXTIME(%d, '%%Y-%%m-%%d %%H-%%i-%%s'), %d, "
"FROM_UNIXTIME(%d, '%%Y-%%m-%%d %%H-%%i-%%s'), '%s', '%s', '%s', "
"'%s', '%s', "
"FROM_UNIXTIME(%d, '%%Y-%%m-%%d %%H-%%i-%%s'), "
"FROM_UNIXTIME(%d, '%%Y-%%m-%%d %%H-%%i-%%s'), %d, %d, %d, %d, "
"%d, %d",
(int)(updatetime_ / 1000), baseinfo_->procpid,
(int)(baseinfo_->proctime / 1000), table_.c_str(),
baseinfo_->srcaddr.c_str(), baseinfo_->dstaddr.c_str(),
baseinfo_->srcdb.c_str(), baseinfo_->dstdb.c_str(),
(int)(starttime_ / 1000), (int)(endtime_ / 1000),
(int)(updatetime_ - starttime_), (rightcount_ + errorcount_),
rightcount_, errorcount_, insertrightcount_, inserterrorcount_);
break;
default:
// 不支持其他类型
return "";
}
ret =
"insert into time_table_info_metrics(updatetime, procpid, proctime, "
"tablename, "
"srcaddr, "
"dstaddr, srcdb, "
"dstdb, starttime, endtime, consumetime, totalcount, rightcount, "
"errorcount, insertrightcount, inserterrorcount) values(";
ret += tmp;
ret += ")";
return ret;
}
DBInfo::DBInfo()
: baseinfo_(NULL),
starttime_(0),
endtime_(0),
tablecount_(0),
viewcount_(0),
procedurecount_(0),
functioncount_(0),
rightcount_(0),
errorcount_(0),
insertrightcount_(0),
inserterrorcount_(0) {}
DBInfo::~DBInfo() {}
std::string DBInfo::toSql() {
std::string ret;
char tmp[1024];
switch (sg_dbtype) {
case OPENGAUSS_DB:
sprintf(tmp,
"%d, TO_TIMESTAMP(%d), '%s', '%s', '%s', '%s', TO_TIMESTAMP(%d), "
"TO_TIMESTAMP(%d), %d, %d, %d, %d, %d, %d, %d, %d, %d, %d",
baseinfo_->procpid, (int)(baseinfo_->proctime / 1000),
baseinfo_->srcaddr.c_str(), baseinfo_->dstaddr.c_str(),
baseinfo_->srcdb.c_str(), baseinfo_->dstdb.c_str(),
(int)(starttime_ / 1000), (int)(endtime_ / 1000),
(int)(endtime_ - starttime_), tablecount_, viewcount_,
procedurecount_, functioncount_, (rightcount_ + errorcount_),
rightcount_, errorcount_, insertrightcount_, inserterrorcount_);
break;
case MYSQL_DB:
sprintf(
tmp,
"%d, FROM_UNIXTIME(%d, '%%Y-%%m-%%d %%H-%%i-%%s'), '%s', '%s', '%s', "
"'%s', FROM_UNIXTIME(%d, '%%Y-%%m-%%d %%H-%%i-%%s'), "
"FROM_UNIXTIME(%d, '%%Y-%%m-%%d %%H-%%i-%%s'), %d, %d, %d, %d, %d, "
"%d, %d, %d, %d, %d",
baseinfo_->procpid, (int)(baseinfo_->proctime / 1000),
baseinfo_->srcaddr.c_str(), baseinfo_->dstaddr.c_str(),
baseinfo_->srcdb.c_str(), baseinfo_->dstdb.c_str(),
(int)(starttime_ / 1000), (int)(endtime_ / 1000),
(int)(endtime_ - starttime_), tablecount_, viewcount_,
procedurecount_, functioncount_, (rightcount_ + errorcount_),
rightcount_, errorcount_, insertrightcount_, inserterrorcount_);
break;
default:
// 不支持其他类型
return "";
}
ret =
"insert into db_info_metrics(procpid, proctime, srcaddr, dstaddr, srcdb, "
"dstdb, starttime, endtime, consumetime, tablecount, viewcount, "
"procedurecount, functioncount, totalcount, rightcount, "
"errorcount, insertrightcount, inserterrorcount) values(";
ret += tmp;
ret += ")";
return ret;
}
TimeDBInfo::TimeDBInfo(DBInfo* dbInfo) {
baseinfo_ = dbInfo->baseinfo_;
updatetime_ = ProcMetrics::getCurrentTime();
starttime_ = dbInfo->starttime_;
endtime_ = dbInfo->endtime_;
tablecount_ = dbInfo->tablecount_;
viewcount_ = dbInfo->viewcount_;
procedurecount_ = dbInfo->procedurecount_;
functioncount_ = dbInfo->functioncount_;
rightcount_ = dbInfo->rightcount_;
errorcount_ = dbInfo->errorcount_;
insertrightcount_ = dbInfo->insertrightcount_;
inserterrorcount_ = dbInfo->inserterrorcount_;
}
TimeDBInfo::TimeDBInfo(long long updatetime, DBInfo* dbInfo) {
baseinfo_ = dbInfo->baseinfo_;
updatetime_ = updatetime;
starttime_ = dbInfo->starttime_;
endtime_ = dbInfo->endtime_;
tablecount_ = dbInfo->tablecount_;
viewcount_ = dbInfo->viewcount_;
procedurecount_ = dbInfo->procedurecount_;
functioncount_ = dbInfo->functioncount_;
rightcount_ = dbInfo->rightcount_;
errorcount_ = dbInfo->errorcount_;
insertrightcount_ = dbInfo->insertrightcount_;
inserterrorcount_ = dbInfo->inserterrorcount_;
}
TimeDBInfo::~TimeDBInfo() {}
std::string TimeDBInfo::toSql() {
std::string ret;
char tmp[1024];
switch (sg_dbtype) {
case OPENGAUSS_DB:
sprintf(tmp,
"TO_TIMESTAMP(%d), %d, TO_TIMESTAMP(%d), '%s', '%s', '%s', '%s', "
"TO_TIMESTAMP(%d), "
"TO_TIMESTAMP(%d), %d, %d, %d, %d, %d, %d, %d, %d, %d, %d",
(int)(updatetime_ / 1000), baseinfo_->procpid,
(int)(baseinfo_->proctime / 1000), baseinfo_->srcaddr.c_str(),
baseinfo_->dstaddr.c_str(), baseinfo_->srcdb.c_str(),
baseinfo_->dstdb.c_str(), (int)(starttime_ / 1000),
(int)(endtime_ / 1000), (int)(updatetime_ - starttime_),
tablecount_, viewcount_, procedurecount_, functioncount_,
(rightcount_ + errorcount_), rightcount_, errorcount_,
insertrightcount_, inserterrorcount_);
break;
case MYSQL_DB:
sprintf(tmp,
"FROM_UNIXTIME(%d, '%%Y-%%m-%%d %%H-%%i-%%s'), %d, "
"FROM_UNIXTIME(%d, '%%Y-%%m-%%d %%H-%%i-%%s'), '%s', '%s', '%s', "
"'%s', FROM_UNIXTIME(%d, '%%Y-%%m-%%d %%H-%%i-%%s'), "
"FROM_UNIXTIME(%d, '%%Y-%%m-%%d %%H-%%i-%%s'), %d, %d, %d, %d, "
"%d, %d, %d, %d, %d, %d",
(int)(updatetime_ / 1000), baseinfo_->procpid,
(int)(baseinfo_->proctime / 1000), baseinfo_->srcaddr.c_str(),
baseinfo_->dstaddr.c_str(), baseinfo_->srcdb.c_str(),
baseinfo_->dstdb.c_str(), (int)(starttime_ / 1000),
(int)(endtime_ / 1000), (int)(updatetime_ - starttime_),
tablecount_, viewcount_, procedurecount_, functioncount_,
(rightcount_ + errorcount_), rightcount_, errorcount_,
insertrightcount_, inserterrorcount_);
break;
default:
// 不支持其他类型
return "";
}
ret =
"insert into time_db_info_metrics(updatetime, procpid, proctime, "
"srcaddr, dstaddr, srcdb, "
"dstdb, starttime, endtime, consumetime, tablecount, viewcount, "
"procedurecount, functioncount, totalcount, rightcount, "
"errorcount, insertrightcount, inserterrorcount) values(";
ret += tmp;
ret += ")";
return ret;
}
MetricInfoSeq::MetricInfoSeq(int maxsize) : maxsize_(maxsize) {}
MetricInfoSeq::~MetricInfoSeq() { destroy(); }
int MetricInfoSeq::getSize() {
Synchronized sync(lock_);
return metrics_.size();
}
std::list<MetricInfo*> MetricInfoSeq::pollAllMetrics() {
std::list<MetricInfo*> ret;
Synchronized sync(lock_);
ret = metrics_;
metrics_ = std::list<MetricInfo*>();
return ret;
}
bool MetricInfoSeq::add(MetricInfo* info) {
bool ret = true;
Synchronized sync(lock_);
if ((maxsize_ <= 0) || ((int)(metrics_.size()) < maxsize_)) {
metrics_.push_back(info);
} else {
// 达到最大数组容量
ret = false;
}
return ret;
}
void MetricInfoSeq::destroy() {
Synchronized sync(lock_);
for (std::list<MetricInfo*>::iterator iter = metrics_.begin();
iter != metrics_.end(); iter++) {
delete (*iter);
}
metrics_.clear();
}
static void* ProcThread(void* arg) {
ProcMetrics* procMetrics = (ProcMetrics*)arg;
procMetrics->saveMetricsToDb();
return NULL;
}
ProcMetrics::ProcMetrics(const char* metricDatasource, const char* srcaddr,
const char* dstaddr, const char* srcdb,
const char* dstdb, int seqmaxsize, const char* dbtype)
: startingDB_(NULL), running_(true), timeDB_(false), timeTable_(false) {
baseDBInfo_.procpid = getPid();
baseDBInfo_.proctime = getCurrentTime();
baseDBInfo_.srcaddr = srcaddr;
baseDBInfo_.dstaddr = dstaddr;
baseDBInfo_.srcdb = srcdb;
baseDBInfo_.dstdb = dstdb;
metricInfoSeq_ = new MetricInfoSeq(seqmaxsize);
procOdbc_ = new DMODBC::ProcOdbc(metricDatasource);
procOdbc_->setLoglevel(5);
if (strcmp(OPENGAUSS_DBTYPE, dbtype) == 0) {
sg_dbtype = OPENGAUSS_DB;
} else if (strcmp(MYSQL_DBTYPE, dbtype) == 0) {
sg_dbtype = MYSQL_DB;
} else {
printf("unsupport db type: %s\n", dbtype);
}
if (!procOdbc_->connect()) {
printf("cannot connect to the datasource %s\n", metricDatasource);
} else {
pthread_create(&savingDatasThread_, NULL, ProcThread, this);
}
}
ProcMetrics::~ProcMetrics() {
while (hasFinishedInfos()) {
sleep(1);
}
running_ = false;
pthread_join(savingDatasThread_, NULL);
if (startingDB_) {
delete startingDB_;
startingDB_ = NULL;
}
if (metricInfoSeq_) {
delete metricInfoSeq_;
metricInfoSeq_ = NULL;
}
if (procOdbc_) {
procOdbc_->disConnect();
delete procOdbc_;
procOdbc_ = NULL;
}
}
void ProcMetrics::startDB() {
if (startingDB_) {
delete startingDB_;
}
startingDB_ = new DBInfo();
startingDB_->baseinfo_ = &baseDBInfo_;
startingDB_->starttime_ = getCurrentTime();
}
void ProcMetrics::startTable(const char* table) {
startingTable_ = new TableInfo();
startingTable_->baseinfo_ = &baseDBInfo_;
startingTable_->table_ = table;
startingTable_->starttime_ = getCurrentTime();
}
void ProcMetrics::startSql(const char* name, const char* type,
const char* sql) {
startingSql_ = new SqlInfo();
startingSql_->baseinfo_ = &baseDBInfo_;
startingSql_->name_ = name;
startingSql_->type_ = type;
startingSql_->sql_ = sql;
startingSql_->starttime_ = getCurrentTime();
}
void ProcMetrics::endSql(const char* name, const char* type, const char* sql,
bool error, const char* errordetail, bool isinsert) {
if (startingSql_) {
startingSql_->error_ = error;
startingSql_->errordetail_ = errordetail;
startingSql_->endtime_ = getCurrentTime();
{
Synchronized sync(lock_);
finishedInfos_.push_back(startingSql_);
}
startingSql_ = NULL;
}
int insertcount = 0;
if (isinsert) {
insertcount = countInsertSql(sql);
}
if (startingTable_) {
if (error) {
startingTable_->errorcount_++;
if (insertcount) {
startingTable_->inserterrorcount_ += insertcount;
}
} else {
startingTable_->rightcount_++;
if (insertcount) {
startingTable_->insertrightcount_ += insertcount;
}
}
if (timeTable_) {
{
Synchronized sync(lock_);
finishedInfos_.push_back(new TimeTableInfo(startingTable_));
}
timeTable_ = false;
}
}
if (startingDB_) {
if (error) {
startingDB_->errorcount_++;
if (insertcount) {
startingDB_->inserterrorcount_ += insertcount;
}
} else {
startingDB_->rightcount_++;
if (insertcount) {
startingDB_->insertrightcount_ += insertcount;
}
}
if (timeDB_) {
{
Synchronized sync(lock_);
finishedInfos_.push_back(new TimeDBInfo(startingDB_));
}
timeDB_ = false;
}
}
}
void ProcMetrics::endTable(const char* table) {
if (startingTable_) {
startingTable_->endtime_ = getCurrentTime();
{
Synchronized sync(lock_);
finishedInfos_.push_back(startingTable_);
finishedInfos_.push_back(
new TimeTableInfo(startingTable_->endtime_, startingTable_));
}
startingTable_ = NULL;
}
}
void ProcMetrics::endDB() {
if (startingDB_) {
startingDB_->endtime_ = getCurrentTime();
{
Synchronized sync(lock_);
finishedInfos_.push_back(startingDB_);
finishedInfos_.push_back(
new TimeDBInfo(startingDB_->endtime_, startingDB_));
}
startingDB_ = NULL;
}
}
void ProcMetrics::addTablecount(int count) {
if (startingDB_) {
startingDB_->tablecount_ += count;
}
}
void ProcMetrics::addViewcount(int count) {
if (startingDB_) {
startingDB_->viewcount_ += count;
}
}
void ProcMetrics::addProcedurecount(int count) {
if (startingDB_) {
startingDB_->procedurecount_ += count;
}
}
void ProcMetrics::addFunctioncount(int count) {
if (startingDB_) {
startingDB_->functioncount_ += count;
}
}
int ProcMetrics::getPid() {
#ifdef WIN32
return (int)GetCurrentProcessId();
#else
return (int)getpid();
#endif
}
long long ProcMetrics::getCurrentTime() {
#ifdef WIN32
timeb timebuffer;
ftime(&timebuffer);
return (long long)(timebuffer.time * 1000ULL + timebuffer.millitm);
#else
timeval t;
gettimeofday(&t, 0);
return (long long)(t.tv_sec * 1000ULL + t.tv_usec / 1000);
#endif
}
std::list<MetricInfo*> ProcMetrics::clearAndGetFinishedInfos() {
std::list<MetricInfo*> ret;
Synchronized sync(lock_);
ret = finishedInfos_;
finishedInfos_ = std::list<MetricInfo*>();
return ret;
}
bool ProcMetrics::hasFinishedInfos() {
bool ret = true;
Synchronized sync(lock_);
if (finishedInfos_.empty()) {
ret = false;
}
return ret;
}
void ProcMetrics::saveMetricsToDb() {
std::string str;
while (running_) {
timeDB_ = true;
timeTable_ = true;
std::list<MetricInfo*> metricInfos = clearAndGetFinishedInfos();
for (std::list<MetricInfo*>::iterator iter = metricInfos.begin();
iter != metricInfos.end(); iter++) {
str = (*iter)->toSql();
if (!str.empty()) {
procOdbc_->executeGeneralSql(str.c_str(), NULL);
}
delete *iter;
}
sleep(2);
}
}
}; // namespace DMODBC

220
MySQL2openGauss/src/ProcMetrics.h Normal file
View File

@ -0,0 +1,220 @@
#ifndef _PROCMETRICS_H_
#define _PROCMETRICS_H_
#include <list>
#include <map>
#include <string>
#ifdef WIN32
#include <Windows.h>
#else
#include <pthread.h>
#endif
#include "ProcOdbc.h"
namespace DMODBC {
struct BaseDBInfo {
int procpid;
long long proctime;
std::string srcaddr;
std::string dstaddr;
std::string srcdb;
std::string dstdb;
};
class Mutex {
public:
Mutex();
virtual ~Mutex();
void lock();
void unlock();
private:
#ifdef WIN32
CRITICAL_SECTION lock_;
#else
pthread_mutex_t lock_;
#endif
};
class Synchronized {
public:
Synchronized(Mutex& m);
virtual ~Synchronized();
private:
Mutex* mutex_;
};
class MetricInfo {
public:
virtual ~MetricInfo();
virtual std::string toSql() = 0;
};
class SqlInfo : public MetricInfo {
public:
SqlInfo();
virtual ~SqlInfo();
virtual std::string toSql();
BaseDBInfo* baseinfo_;
std::string name_;
std::string type_;
std::string sql_;
long long starttime_;
long long endtime_;
bool error_;
std::string errordetail_;
private:
void formatSql(const char* in, int inlen, char* out);
};
class TableInfo : public MetricInfo {
public:
TableInfo();
virtual ~TableInfo();
virtual std::string toSql();
BaseDBInfo* baseinfo_;
std::string table_;
long long starttime_;
long long endtime_;
int rightcount_;
int errorcount_;
int insertrightcount_;
int inserterrorcount_;
};
class TimeTableInfo : public MetricInfo {
public:
TimeTableInfo(TableInfo* tableInfo);
TimeTableInfo(long long updatetime_, TableInfo* tableInfo);
virtual ~TimeTableInfo();
virtual std::string toSql();
BaseDBInfo* baseinfo_;
long long updatetime_;
std::string table_;
long long starttime_;
long long endtime_;
int rightcount_;
int errorcount_;
int insertrightcount_;
int inserterrorcount_;
};
class DBInfo : public MetricInfo {
public:
DBInfo();
virtual ~DBInfo();
virtual std::string toSql();
BaseDBInfo* baseinfo_;
long long starttime_;
long long endtime_;
int tablecount_;
int viewcount_;
int procedurecount_;
int functioncount_;
int rightcount_;
int errorcount_;
int insertrightcount_;
int inserterrorcount_;
};
class TimeDBInfo : public MetricInfo {
public:
TimeDBInfo(DBInfo* dbInfo);
TimeDBInfo(long long updatetime, DBInfo* dbInfo);
virtual ~TimeDBInfo();
virtual std::string toSql();
BaseDBInfo* baseinfo_;
long long updatetime_;
long long starttime_;
long long endtime_;
int tablecount_;
int viewcount_;
int procedurecount_;
int functioncount_;
int rightcount_;
int errorcount_;
int insertrightcount_;
int inserterrorcount_;
};
class MetricInfoSeq {
public:
MetricInfoSeq(int maxsize);
virtual ~MetricInfoSeq();
int getSize();
std::list<MetricInfo*> pollAllMetrics();
bool add(MetricInfo* info);
private:
void destroy();
private:
int maxsize_;
Mutex lock_;
std::list<MetricInfo*> metrics_;
};
class ProcMetrics {
public:
ProcMetrics(const char* metricDatasource, const char* srcaddr,
const char* dstaddr, const char* srcdb, const char* dstdb,
int seqmaxsize, const char* dbtype);
virtual ~ProcMetrics();
void startDB();
void startTable(const char* table);
void startSql(const char* name, const char* type, const char* sql);
void endSql(const char* name, const char* type, const char* sql, bool error,
const char* errordetail, bool isinsert);
void endTable(const char* table);
void endDB();
void addTablecount(int count);
void addViewcount(int count);
void addProcedurecount(int count);
void addFunctioncount(int count);
void saveMetricsToDb();
static long long getCurrentTime();
private:
int getPid();
std::list<MetricInfo*> clearAndGetFinishedInfos();
bool hasFinishedInfos();
private:
ProcOdbc* procOdbc_;
BaseDBInfo baseDBInfo_;
DBInfo* startingDB_;
TableInfo* startingTable_;
SqlInfo* startingSql_;
int size_;
MetricInfoSeq* metricInfoSeq_;
Mutex lock_;
std::list<MetricInfo*> finishedInfos_;
bool running_;
bool timeDB_;
bool timeTable_;
pthread_t savingDatasThread_;
};
}; // namespace DMODBC
#endif

993
MySQL2openGauss/src/ProcOdbc.cpp Normal file
View File

@ -0,0 +1,993 @@
#include <sqlext.h>
#include <stdio.h>
#include <stdlib.h>
#ifdef WIN32
#include <windows.h>
#endif
#include <string.h>
#include <unistd.h>
#include <fstream>
#include <iostream>
#include <list>
#include <map>
#include "ProcOdbc.h"
#define SQLK_CREATE_TABLE "CREATE TABLE "
#define SQLK_UNIQUE_KEY "UNIQUE KEY "
#define SQLK_PRIMARY_KEY "PRIMARY KEY "
#define SQLK_FOREIGN_KEY "FOREIGN KEY "
#define SQLK_KEY "KEY "
#define SQLK_COMMENT " COMMENT '"
#define SQLK_BYTEA_START " bytea("
#define SQLK_BYTEA_END ")"
#define SQLK_BYTEA " bytea"
#define SQLK_SET_START " set("
#define SQLK_SET_END ")"
#define SQLK_TEXT " text"
#define SQLK_PROCEDURE "PROCEDURE"
#define SQLK_PROCEDURE_DEF_END ")"
#define SQLK_FUNCTION "FUNCTION"
#define SQLK_FUNCTION_DEF_END "\n"
#define SQLK_INSERT_INTO "\nINSERT INTO \""
#define SQLK_DOUBLE_BRACKET " double("
#define SQLK_DOUBLE_BRACKET_END ")"
#define DATA_NOT_SEND_ERROR "could not send data"
#define NO_CONNECTION_ERROR "no connection"
#define NULL_STRING_ERROR "violates not-null constraint"
#define TABLE_ALREADY_EXIST_ERROR "\" already exists;"
#define INFO_LOG_LEVEL 3
#define WARN_LOG_LEVEL 4
#define ERROR_LOG_LEVEL 5
namespace DMODBC {
struct ReplaceKV {
std::string key;
std::string value;
};
struct DBInfo {
SQLHENV V_OD_Env; // Handle ODBC environment
SQLHSTMT V_OD_hstmt; // Handle statement
SQLHDBC V_OD_hdbc; // Handle connection
char sqlattr[100];
SQLINTEGER V_OD_erg;
SQLINTEGER V_OD_buffer;
SQLINTEGER V_OD_err;
SQLINTEGER V_OD_id;
};
static void printDBError(SQLSMALLINT type, SQLHDBC hdbc,
struct DBError *dbError) {
char status[16];
int buflen = 1024;
char *msg = (char *)malloc(buflen);
msg[0] = '\0';
SQLSMALLINT sqlmsglen;
SQLINTEGER sqlerr;
SQLGetDiagRec(type, hdbc, 1, (SQLCHAR *)status, &sqlerr, (SQLCHAR *)msg,
buflen - 1, &sqlmsglen);
if (buflen < sqlmsglen) {
free(msg);
buflen = sqlmsglen;
msg = (char *)malloc(buflen);
msg[0] = '\0';
SQLGetDiagRec(type, hdbc, 1, (SQLCHAR *)status, &sqlerr, (SQLCHAR *)msg,
buflen - 1, &sqlmsglen);
}
printf("%s: %s\n", status, msg);
if (dbError != NULL) {
dbError->error = status;
dbError->error += ": ";
dbError->error += msg;
if (strstr(msg, TABLE_ALREADY_EXIST_ERROR)) {
dbError->alreadyExistError = true;
} else {
dbError->alreadyExistError = false;
}
}
free(msg);
}
static bool isConnectionError(SQLSMALLINT type, SQLHDBC hdbc) {
bool ret = false;
char status[16];
int buflen = 1024;
char *msg = (char *)malloc(buflen);
msg[0] = '\0';
SQLSMALLINT sqlmsglen;
SQLINTEGER sqlerr;
SQLGetDiagRec(type, hdbc, 1, (SQLCHAR *)status, &sqlerr, (SQLCHAR *)msg,
buflen - 1, &sqlmsglen);
if (buflen < sqlmsglen) {
free(msg);
buflen = sqlmsglen;
msg = (char *)malloc(buflen);
msg[0] = '\0';
SQLGetDiagRec(type, hdbc, 1, (SQLCHAR *)status, &sqlerr, (SQLCHAR *)msg,
buflen - 1, &sqlmsglen);
}
if (strstr(msg, DATA_NOT_SEND_ERROR) || strstr(msg, NO_CONNECTION_ERROR)) {
ret = true;
}
free(msg);
return ret;
}
static bool isNullStringError(SQLSMALLINT type, SQLHDBC hdbc) {
bool ret = false;
char status[16];
int buflen = 1024;
char *msg = (char *)malloc(buflen);
msg[0] = '\0';
SQLSMALLINT sqlmsglen;
SQLINTEGER sqlerr;
SQLGetDiagRec(type, hdbc, 1, (SQLCHAR *)status, &sqlerr, (SQLCHAR *)msg,
buflen - 1, &sqlmsglen);
if (buflen < sqlmsglen) {
free(msg);
buflen = sqlmsglen;
msg = (char *)malloc(buflen);
msg[0] = '\0';
SQLGetDiagRec(type, hdbc, 1, (SQLCHAR *)status, &sqlerr, (SQLCHAR *)msg,
buflen - 1, &sqlmsglen);
}
if (strstr(msg, NULL_STRING_ERROR)) {
ret = true;
}
free(msg);
return ret;
}
static void replace(std::string &src, const char *orgstr, const char *newstr) {
std::string::size_type pos = src.find(orgstr);
while (pos != std::string::npos) {
src.replace(pos, strlen(orgstr), newstr);
pos = src.find(orgstr);
}
}
static bool replaceonce(std::string &src, const char *orgstr,
const char *newstr) {
bool ret = false;
std::string::size_type pos = src.find(orgstr);
if (pos != std::string::npos) {
src.replace(pos, strlen(orgstr), newstr);
ret = true;
}
return ret;
}
static std::string trim(const std::string &s) {
static const char *whiteSpace = " \t\r\n";
if (s.empty()) {
return s;
}
std::string::size_type b = s.find_first_not_of(whiteSpace);
if (std::string::npos == b) {
return "";
}
std::string::size_type e = s.find_last_not_of(whiteSpace);
return std::string(s, b, e - b + 1);
}
static int getLine(std::istream &in, std::string &strLine) {
strLine = "";
int count = 0;
bool lineFlag = false;
char ch;
do {
ch = in.get();
if (!in.good()) {
if (count > 0) {
return count;
} else {
return -1;
}
}
switch (ch) {
case '\r':
break;
case '\n':
lineFlag = true;
break;
default:
strLine += ch;
count++;
break;
}
} while (!lineFlag);
return count;
}
static void loadKvList(std::istream &in, std::list<ReplaceKV> &replaceKvList) {
std::string fullLine, command;
std::string leftSide, rightSide;
std::string::size_type length;
while (getLine(in, fullLine) != -1) {
length = fullLine.find('#');
if (std::string::npos == length) {
command = fullLine;
} else if (length > 0) {
command = fullLine.substr(0, length);
} else {
continue;
}
command = trim(command);
length = command.find('=');
if (std::string::npos != length) {
if ((length + 1) != command.size()) {
leftSide = trim(command.substr(0, length));
rightSide = trim(command.substr(length + 1));
} else {
leftSide = trim(command.substr(0, length));
rightSide = "";
}
} else {
continue;
}
ReplaceKV replaceKv;
replaceKv.key = leftSide;
replaceKv.value = rightSide;
replaceKvList.push_back(replaceKv);
}
}
static void loadKvList(const char *kvListFile,
std::list<ReplaceKV> &replaceKvList) {
std::ifstream inputStream;
inputStream.open(kvListFile);
if (!inputStream.fail()) {
loadKvList(inputStream, replaceKvList);
inputStream.close();
} else {
printf("cannot open kv file %s\n", kvListFile);
}
}
static void loadKvMap(std::istream &in,
std::map<std::string, std::string> &replaceKvMap) {
std::string fullLine, command;
std::string leftSide, rightSide;
std::string::size_type length;
while (getLine(in, fullLine) != -1) {
length = fullLine.find('#');
if (std::string::npos == length) {
command = fullLine;
} else if (length > 0) {
command = fullLine.substr(0, length);
} else {
continue;
}
command = trim(command);
length = command.find('=');
if (std::string::npos != length) {
if ((length + 1) != command.size()) {
leftSide = trim(command.substr(0, length));
rightSide = trim(command.substr(length + 1, command.size() - length));
} else {
leftSide = trim(command.substr(0, length));
}
} else {
continue;
}
replaceKvMap[leftSide] = rightSide;
}
}
static void loadKvMap(const char *kvListFile,
std::map<std::string, std::string> &replaceKvMap) {
std::ifstream inputStream;
inputStream.open(kvListFile);
if (!inputStream.fail()) {
loadKvMap(inputStream, replaceKvMap);
inputStream.close();
} else {
printf("cannot open kv file %s\n", kvListFile);
}
}
static std::list<ReplaceKV> getCreateTableReplaceKV() {
std::list<ReplaceKV> ret;
std::string replaceKvFile;
const char *homedir = getenv(MYSQL_TO_OPENGAUSS_HOME_DIR);
if (!homedir) {
return ret;
}
replaceKvFile = homedir;
replaceKvFile += "/conf/createtablerkv.prop";
loadKvList(replaceKvFile.c_str(), ret);
return ret;
}
static std::list<ReplaceKV> getCreateFunctionReplaceKV() {
std::list<ReplaceKV> ret;
std::string replaceKvFile;
const char *homedir = getenv(MYSQL_TO_OPENGAUSS_HOME_DIR);
if (!homedir) {
return ret;
}
replaceKvFile = homedir;
replaceKvFile += "/conf/createfunctionrkv.prop";
loadKvList(replaceKvFile.c_str(), ret);
return ret;
}
static std::list<ReplaceKV> getCreateFunctionTypeKV() {
std::list<ReplaceKV> ret;
std::string replaceKvFile;
const char *homedir = getenv(MYSQL_TO_OPENGAUSS_HOME_DIR);
if (!homedir) {
return ret;
}
replaceKvFile = homedir;
replaceKvFile += "/conf/createfunctiontkv.prop";
loadKvList(replaceKvFile.c_str(), ret);
return ret;
}
ProcOdbc::ProcOdbc(const char *datasource)
: isconn_(false), loglevel_(INFO_LOG_LEVEL) {
strcpy(datasource_, datasource);
dbinfo_ = new DBInfo;
currentArgc_ = 4;
currentArgv_ = (char **)malloc(sizeof(char *) * currentArgc_);
currentArgv_[0] = (char *)malloc(256);
strcpy(currentArgv_[0], "mysqltoopengauss");
currentArgv_[1] = (char *)malloc(256);
strcpy(currentArgv_[1], "--compatible=postgresql");
currentArgv_[2] = (char *)malloc(256);
strcpy(currentArgv_[2], "--default-character-set=utf8");
currentArgv_[3] = (char *)malloc(256);
currentArgv_[3][0] = '\0';
}
ProcOdbc::~ProcOdbc() {
disConnect();
delete (DBInfo *)dbinfo_;
freeCurrentArgv();
}
struct MysqlInfo ProcOdbc::getMysqlInfo() {
std::string confFile;
std::map<std::string, std::string> confMap;
struct MysqlInfo ret;
const char *homedir = getenv(MYSQL_TO_OPENGAUSS_HOME_DIR);
if (!homedir) {
return ret;
}
confFile = homedir;
confFile += "/conf/mysql2opengauss.prop";
loadKvMap(confFile.c_str(), confMap);
ret.host = confMap["mysql_host"];
ret.port = atoi(confMap["mysql_port"].c_str());
ret.username = confMap["mysql_username"];
ret.password = confMap["mysql_password"];
ret.database = confMap["mysql_database"];
ret.loglevel = atoi(confMap["loglevel"].c_str());
ret.dsthost = confMap["opengauss_host"];
ret.dstport = atoi(confMap["opengauss_port"].c_str());
ret.dstdatabase = confMap["opengauss_database"];
printf("from MySQL %s:%d database %s to openGauss %s:%d database %s\n",
confMap["mysql_host"].c_str(), atoi(confMap["mysql_port"].c_str()),
confMap["mysql_database"].c_str(), confMap["opengauss_host"].c_str(),
atoi(confMap["opengauss_port"].c_str()),
confMap["opengauss_database"].c_str());
return ret;
}
bool ProcOdbc::isConnected() { return isconn_; }
bool ProcOdbc::connect() {
if (isconn_) {
// 已经连接
return isconn_;
}
DBInfo *dbinfo = (DBInfo *)dbinfo_;
// 1. 申请环境句柄
dbinfo->V_OD_erg =
SQLAllocHandle(SQL_HANDLE_ENV, SQL_NULL_HANDLE, &dbinfo->V_OD_Env);
if ((dbinfo->V_OD_erg != SQL_SUCCESS) &&
(dbinfo->V_OD_erg != SQL_SUCCESS_WITH_INFO)) {
printf("Error AllocHandle SQL_HANDLE_ENV\n");
return isconn_;
}
// 2. 设置环境属性（版本信息）
SQLSetEnvAttr(dbinfo->V_OD_Env, SQL_ATTR_ODBC_VERSION, (void *)SQL_OV_ODBC3,
0);
// 3. 申请连接句柄
dbinfo->V_OD_erg =
SQLAllocHandle(SQL_HANDLE_DBC, dbinfo->V_OD_Env, &dbinfo->V_OD_hdbc);
if ((dbinfo->V_OD_erg != SQL_SUCCESS) &&
(dbinfo->V_OD_erg != SQL_SUCCESS_WITH_INFO)) {
SQLFreeHandle(SQL_HANDLE_ENV, dbinfo->V_OD_Env);
printf("Error AllocHandle SQL_HANDLE_DBC\n");
printDBError(SQL_HANDLE_ENV, dbinfo->V_OD_Env, NULL);
return isconn_;
}
// 4. 设置连接属性
SQLSetConnectAttr(dbinfo->V_OD_hdbc, SQL_ATTR_AUTOCOMMIT,
(SQLPOINTER)SQL_AUTOCOMMIT_ON, 0);
// 5.
// odbc.ini文件中已经配置了用户名密码，那么这里可以留空（""）；但是不建议这么做，因为一旦odbc.ini权限管理不善，将导致数据库用户密码泄露。
printf("datasource: %s\n", datasource_);
dbinfo->V_OD_erg =
SQLConnect(dbinfo->V_OD_hdbc, (SQLCHAR *)datasource_, SQL_NTS,
(SQLCHAR *)"", SQL_NTS, (SQLCHAR *)"", SQL_NTS);
if ((dbinfo->V_OD_erg != SQL_SUCCESS) &&
(dbinfo->V_OD_erg != SQL_SUCCESS_WITH_INFO)) {
printf("Error SQLConnect %d\n", (int)(dbinfo->V_OD_erg));
printDBError(SQL_HANDLE_DBC, dbinfo->V_OD_hdbc, NULL);
SQLFreeHandle(SQL_HANDLE_ENV, dbinfo->V_OD_Env);
return isconn_;
}
printf("Connected !\n");
// 6. 设置语句属性
SQLSetStmtAttr(dbinfo->V_OD_hstmt, SQL_ATTR_QUERY_TIMEOUT, (SQLPOINTER *)3,
0);
// 7. 申请语句句柄
SQLAllocHandle(SQL_HANDLE_STMT, dbinfo->V_OD_hdbc, &dbinfo->V_OD_hstmt);
isconn_ = true;
return isconn_;
}
void ProcOdbc::disConnect() {
if (!isconn_) {
// 没有连接，直接返回
return;
}
isconn_ = false;
DBInfo *dbinfo = (DBInfo *)dbinfo_;
// 16. 断开数据源连接并释放句柄资源
SQLFreeHandle(SQL_HANDLE_STMT, dbinfo->V_OD_hstmt);
SQLDisconnect(dbinfo->V_OD_hdbc);
SQLFreeHandle(SQL_HANDLE_DBC, dbinfo->V_OD_hdbc);
SQLFreeHandle(SQL_HANDLE_ENV, dbinfo->V_OD_Env);
}
bool ProcOdbc::reConnect() {
disConnect();
return connect();
}
std::list<std::string> ProcOdbc::procCreateTableSql(const char *sql,
const char *incrementkey) {
std::string retsql = sql;
std::string ik;
if (incrementkey != NULL) {
ik = incrementkey;
}
replaceCreateTableSql(retsql);
return procCreateTableSqlLine(retsql, ik);
}
std::string ProcOdbc::procProcedureFunctionSql(const char *sql,
bool &isfunction) {
static std::list<ReplaceKV> funrkv;
static std::list<ReplaceKV> funtkv;
std::string ret;
std::string sqlstr = sql;
replace(sqlstr, "`", "\"");
replace(sqlstr, "\r", "");
std::string::size_type pos = sqlstr.find(SQLK_PROCEDURE);
std::string::size_type pos2 = sqlstr.find(SQLK_PROCEDURE_DEF_END, pos + 1);
if ((pos != std::string::npos) && (pos2 != std::string::npos)) {
isfunction = false;
ret = "CREATE ";
ret += sqlstr.substr(pos, pos2 - pos + 1);
ret += " IS ";
ret += sqlstr.substr(pos2 + 1);
} else {
isfunction = true;
if (funrkv.empty()) {
funrkv = getCreateFunctionReplaceKV();
funtkv = getCreateFunctionTypeKV();
}
// 替换一般信息，只换一次
for (std::list<ReplaceKV>::iterator iter = funrkv.begin();
iter != funrkv.end(); iter++) {
if (replaceonce(sqlstr, iter->key.c_str(), iter->value.c_str())) {
break;
}
}
// 替换类型，只换一次
for (std::list<ReplaceKV>::iterator iter = funtkv.begin();
iter != funtkv.end(); iter++) {
if (replaceonce(sqlstr, iter->key.c_str(), iter->value.c_str())) {
break;
}
}
pos = sqlstr.find(SQLK_FUNCTION);
pos2 = sqlstr.find(SQLK_FUNCTION_DEF_END, pos + 1);
if ((pos != std::string::npos) && (pos2 != std::string::npos)) {
ret = "CREATE ";
ret += sqlstr.substr(pos, pos2 - pos + 1);
ret += " AS $$ ";
ret += sqlstr.substr(pos2 + 1);
ret += "; $$ LANGUAGE plpgsql";
}
}
return ret;
}
std::string &ProcOdbc::replaceCreateTableSql(std::string &sql) {
static std::list<ReplaceKV> replaceKVs;
if (replaceKVs.empty()) {
replaceKVs = getCreateTableReplaceKV();
}
for (std::list<ReplaceKV>::iterator iter = replaceKVs.begin();
iter != replaceKVs.end(); iter++) {
replace(sql, iter->key.c_str(), iter->value.c_str());
}
formatByteaAtCreateTableSql(sql);
formatSetAtCreateTableSql(sql);
return sql;
}
std::list<std::string> ProcOdbc::procCreateTableSqlLine(
const std::string &sql, const std::string &incrementkey) {
std::list<std::string> ret;
std::string table;
std::string outputsql;
std::string tmp;
std::string tmp2;
std::string::size_type tmppos;
std::string::size_type tmppos2;
std::string::size_type tmppos3;
std::string::size_type tmppos4;
std::string::size_type pos = 0;
std::string::size_type nextpos;
bool iskeyline;
int createTableLen = strlen(SQLK_CREATE_TABLE);
int uniqueKeyLen = strlen(SQLK_UNIQUE_KEY);
int keyLen = strlen(SQLK_KEY);
int commentLen = strlen(SQLK_COMMENT);
int doubleBracketLen = strlen(SQLK_DOUBLE_BRACKET);
nextpos = sql.find('\n', pos);
while (nextpos != std::string::npos) {
iskeyline = false;
tmp = sql.substr(pos, nextpos - pos);
// CREATE TABLE
tmppos = tmp.find(SQLK_CREATE_TABLE);
if (tmppos != std::string::npos) {
tmppos2 = tmp.find('(');
if (tmppos2 != std::string::npos) {
table = tmp.substr(tmppos + createTableLen,
tmppos2 - tmppos - createTableLen - 1);
}
iskeyline = true;
} else {
// double bracket
tmppos = tmp.find(SQLK_DOUBLE_BRACKET);
if (tmppos != std::string::npos) {
tmppos2 = tmp.find(SQLK_DOUBLE_BRACKET_END, tmppos + doubleBracketLen);
if (tmppos2 != std::string::npos) {
tmp2 = tmp.substr(0, tmppos);
tmp2 += " double precision ";
tmp2 += tmp.substr(tmppos2 + strlen(SQLK_DOUBLE_BRACKET_END));
tmp = tmp2;
}
}
// COMMENT
tmppos = tmp.find(SQLK_COMMENT);
if (tmppos != std::string::npos) {
tmppos2 = tmp.find('\'', tmppos + commentLen);
tmppos3 = tmp.find('"');
tmppos4 = tmp.find('"', tmppos3 + 2);
if ((tmppos2 != std::string::npos) && (tmppos3 != std::string::npos) &&
(tmppos4 != std::string::npos)) {
tmp2 = "COMMENT ON COLUMN ";
tmp2 += table + ".";
tmp2 += tmp.substr(tmppos3, tmppos4 - tmppos3 + 1);
tmp2 += " IS ";
tmp2 += tmp.substr(tmppos + commentLen - 1,
tmppos2 + 2 - tmppos - commentLen);
ret.push_back(tmp2);
}
tmp = tmp.substr(0, tmppos) + ",";
} else {
// UNIQUE KEY
tmppos = tmp.find(SQLK_UNIQUE_KEY);
if (tmppos != std::string::npos) {
tmppos2 = tmp.find('(');
if (tmppos2 != std::string::npos) {
tmp2 = " CONSTRAINT";
tmp2 += " \"";
tmp2 += table.substr(1, table.size() - 2);
tmp2 += "_";
tmp2 += tmp.substr(tmppos + uniqueKeyLen + 1,
tmppos2 - tmppos - uniqueKeyLen - 1);
tmp2 += "UNIQUE ";
tmp2 += tmp.substr(tmppos2);
tmp = tmp2;
}
iskeyline = true;
} else {
// PRIMARY KEY
tmppos = tmp.find(SQLK_PRIMARY_KEY);
if (tmppos != std::string::npos) {
// PRIMARY KEY不需要处理
iskeyline = true;
} else {
// FOREIGN KEY
tmppos = tmp.find(SQLK_FOREIGN_KEY);
if (tmppos != std::string::npos) {
// FOREIGN KEY不需要处理
iskeyline = true;
} else {
// KEY
tmppos = tmp.find(SQLK_KEY);
if (tmppos != std::string::npos) {
tmppos2 = tmp.find('(');
if (tmppos2 != std::string::npos) {
tmppos3 = tmp.find(')', tmppos2);
if (tmppos3 != std::string::npos) {
tmp2 = "CREATE INDEX";
tmp2 += " \"";
tmp2 += table.substr(1, table.size() - 2);
tmp2 += "_";
tmp2 += tmp.substr(tmppos + keyLen + 1,
tmppos2 - tmppos - keyLen - 2);
tmp2 += " ON ";
tmp2 += table;
tmp2 += tmp.substr(tmppos2, tmppos3 - tmppos2 + 1);
ret.push_back(tmp2);
tmp = "";
}
}
iskeyline = true;
}
}
}
}
}
}
if (!tmp.empty()) {
if (!iskeyline && !incrementkey.empty()) {
if (tmp.find(incrementkey) != std::string::npos) {
// 自增主键
tmppos = tmp.rfind(",");
if (tmppos != std::string::npos) {
tmp = tmp.substr(0, tmppos) + " default nextval('seq_" +
table.substr(1, table.size() - 2) + "_" +
incrementkey.substr(1, incrementkey.size() - 2) + "'),";
} else {
tmp = tmp.substr(0, tmppos) + " default nextval('seq_" +
table.substr(1, table.size() - 2) + "_" +
incrementkey.substr(1, incrementkey.size() - 2) + "')";
}
}
}
outputsql += tmp + "\n";
}
pos = nextpos + 1;
nextpos = sql.find('\n', pos);
}
if (outputsql.at(outputsql.size() - 2) == ',') {
outputsql = outputsql.substr(0, outputsql.size() - 2) + "\n";
}
outputsql += sql.substr(pos);
ret.push_front(outputsql);
if (!incrementkey.empty()) {
// 如果有自增主键，增加自增主键创建语句
outputsql = "create SEQUENCE seq_";
outputsql += table.substr(1, table.size() - 2) + "_" +
incrementkey.substr(1, incrementkey.size() - 2);
outputsql += " cache 100";
ret.push_front(outputsql);
outputsql = "alter SEQUENCE seq_";
outputsql += table.substr(1, table.size() - 2) + "_" +
incrementkey.substr(1, incrementkey.size() - 2);
outputsql += " OWNED by ";
outputsql += table;
outputsql += ".";
outputsql += incrementkey;
ret.push_back(outputsql);
}
return ret;
}
std::string &ProcOdbc::formatByteaAtCreateTableSql(std::string &sql) {
std::string::size_type pos1 = sql.find(SQLK_BYTEA_START);
std::string::size_type pos2;
while (pos1 != std::string::npos) {
pos2 = sql.find(SQLK_BYTEA_END, pos1 + 1);
if (pos2 == std::string::npos) {
break;
}
sql.replace(pos1, pos2 - pos1 + strlen(SQLK_BYTEA_END), SQLK_BYTEA);
pos1 = sql.find(SQLK_BYTEA_START);
}
return sql;
}
std::string &ProcOdbc::formatSetAtCreateTableSql(std::string &sql) {
std::string::size_type pos1 = sql.find(SQLK_SET_START);
std::string::size_type pos2;
while (pos1 != std::string::npos) {
pos2 = sql.find(SQLK_SET_END, pos1 + 1);
if (pos2 == std::string::npos) {
break;
}
sql.replace(pos1, pos2 - pos1 + strlen(SQLK_SET_END), SQLK_TEXT);
pos1 = sql.find(SQLK_SET_START);
}
return sql;
}
char **ProcOdbc::getCurrentArgv() { return currentArgv_; }
int ProcOdbc::getCurrentArgc() { return currentArgc_ - 1; }
void ProcOdbc::setLoglevel(int loglevel) { loglevel_ = loglevel; };
bool ProcOdbc::executeCreateTableSql(const char *sql, DBError *dbError) {
bool ret = true;
DBInfo *dbinfo = (DBInfo *)dbinfo_;
// 直接执行SQL语句。
if (loglevel_ <= INFO_LOG_LEVEL) {
printf("start: execute sql %s\n", sql);
}
int execret = execSqlStably(sql);
if ((execret != SQL_SUCCESS) && (execret != SQL_SUCCESS_WITH_INFO)) {
printf("finished error(%d): execute sql %s\n", (int)execret, sql);
printDBError(SQL_HANDLE_STMT, dbinfo->V_OD_hstmt, dbError);
ret = false;
// if (dbError) {
// if (!dbError->alreadyExistError) {
// errcreatetablesqls_.push_back(sql);
// }
//}
} else {
if (loglevel_ <= INFO_LOG_LEVEL) {
printf("finished successfully: execute sql %s\n", sql);
}
}
return ret;
}
bool ProcOdbc::executeInsertSql(const char *sql, DBError *dbError) {
bool ret = true;
int insertIntoLen = strlen(SQLK_INSERT_INTO);
std::string sqlstr = sql;
std::string insertSql;
// 拆分成多条sql语句
std::string::size_type pos = 0;
std::string::size_type pos2 = sqlstr.find(SQLK_INSERT_INTO, insertIntoLen);
while (true) {
if (pos2 == std::string::npos) {
insertSql = sqlstr.substr(pos);
ret = executeInsertSql(insertSql.c_str(), "' '", dbError);
break;
} else {
insertSql = sqlstr.substr(pos, pos2 - pos);
ret = executeInsertSql(insertSql.c_str(), "' '", dbError);
}
pos = pos2 + 1;
pos2 = sqlstr.find(SQLK_INSERT_INTO, pos + insertIntoLen);
}
return ret;
}
bool ProcOdbc::executeInsertSql(const char *sql, const char *replaceNullStr,
DBError *dbError) {
bool ret = true;
std::string sqlstr;
DBInfo *dbinfo = (DBInfo *)dbinfo_;
// 直接执行SQL语句。
if (loglevel_ <= INFO_LOG_LEVEL) {
printf("start: execute sql %s\n", sql);
}
int execret = execSqlStably(sql);
if ((execret != SQL_SUCCESS) && (execret != SQL_SUCCESS_WITH_INFO)) {
if (isNullStringError(SQL_HANDLE_STMT, dbinfo->V_OD_hstmt)) {
// 非空字符串约束，把''改成replaceNullStr，并重试一次
printDBError(SQL_HANDLE_STMT, dbinfo->V_OD_hstmt, dbError);
printf("replace '' by %s and retry\n", replaceNullStr);
sqlstr = sql;
replace(sqlstr, "''", replaceNullStr);
execret = execSqlStably(sqlstr.c_str());
}
}
if ((execret != SQL_SUCCESS) && (execret != SQL_SUCCESS_WITH_INFO)) {
printf("finished error(%d): execute sql %s\n", (int)execret, sql);
printDBError(SQL_HANDLE_STMT, dbinfo->V_OD_hstmt, dbError);
ret = false;
} else {
if (loglevel_ <= INFO_LOG_LEVEL) {
printf("finished successfully: execute sql %s\n", sql);
}
}
return ret;
}
bool ProcOdbc::executeCreateViewSql(const char *sql, DBError *dbError) {
bool ret = true;
DBInfo *dbinfo = (DBInfo *)dbinfo_;
// 直接执行SQL语句。
if (loglevel_ <= INFO_LOG_LEVEL) {
printf("start: execute sql %s\n", sql);
}
int execret = execSqlStably(sql);
if ((execret != SQL_SUCCESS) && (execret != SQL_SUCCESS_WITH_INFO)) {
printf("finished error(%d): execute sql %s\n", (int)execret, sql);
printDBError(SQL_HANDLE_STMT, dbinfo->V_OD_hstmt, dbError);
ret = false;
} else {
if (loglevel_ <= INFO_LOG_LEVEL) {
printf("finished successfully: execute sql %s\n", sql);
}
}
return ret;
}
bool ProcOdbc::executeGeneralSql(const char *sql, DBError *dbError) {
bool ret = true;
DBInfo *dbinfo = (DBInfo *)dbinfo_;
// 直接执行SQL语句。
if (loglevel_ <= INFO_LOG_LEVEL) {
printf("start: execute sql %s\n", sql);
}
int execret = execSqlStably(sql);
if ((execret != SQL_SUCCESS) && (execret != SQL_SUCCESS_WITH_INFO)) {
printf("finished error(%d): execute sql %s\n", (int)execret, sql);
printDBError(SQL_HANDLE_STMT, dbinfo->V_OD_hstmt, dbError);
ret = false;
} else {
if (loglevel_ <= INFO_LOG_LEVEL) {
printf("finished successfully: execute sql %s\n", sql);
}
}
return ret;
}
int ProcOdbc::execSqlStably(const char *sql) {
int retry = 30;
DBInfo *dbinfo = (DBInfo *)dbinfo_;
SQLINTEGER execret;
while (retry-- > 0) {
execret = SQLExecDirect(dbinfo->V_OD_hstmt, (SQLCHAR *)sql, SQL_NTS);
if ((execret != SQL_SUCCESS) && (execret != SQL_SUCCESS_WITH_INFO)) {
if (isConnectionError(SQL_HANDLE_STMT, dbinfo->V_OD_hstmt)) {
// 连接错误，重新连接，并重试
printDBError(SQL_HANDLE_STMT, dbinfo->V_OD_hstmt, NULL);
if (retry > 0) {
printf("reconnect and retry(%d)...", retry);
sleep(2);
}
reConnect();
} else {
// 其他错误，退出循环
break;
}
} else {
// 执行成功，退出循环
break;
}
}
return (int)execret;
}
bool ProcOdbc::isBinaryBlobChar(unsigned char ch) {
return (ch <= 31) || (ch >= 127) /*&& (ch <= 255)*/ || (ch == 39) ||
(ch == 92);
}
std::list<std::string> ProcOdbc::clearAndGetErrCreateTableSqls() {
std::list<std::string> tmp;
std::list<std::string> ret = errcreatetablesqls_;
errcreatetablesqls_ = tmp;
return ret;
}
bool ProcOdbc::executeLeftErrorSqls() {
bool ret = true;
if (!executeLeftErrorCreateTableSqls()) {
ret = false;
}
return ret;
}
bool ProcOdbc::executeLeftErrorCreateTableSqls() {
std::list<std::string> sqls = clearAndGetErrCreateTableSqls();
int retry = 1000;
struct DBError dbError;
while (sqls.size() > 0) {
// 有错误sql时不断执行
printf("retry...\n");
for (std::list<std::string>::iterator iter = sqls.begin();
iter != sqls.end(); iter++) {
printf("retry sql %s\n", iter->c_str());
executeCreateTableSql(iter->c_str(), &dbError);
}
if (--retry == 0) {
if (hasErrorCreateTableSqls()) {
// 还仍然有错误
printErrorCreateTableSqls();
}
break;
}
sqls = clearAndGetErrCreateTableSqls();
}
return !hasErrorCreateTableSqls();
}
bool ProcOdbc::hasErrorSqls() { return hasErrorCreateTableSqls(); }
bool ProcOdbc::hasErrorCreateTableSqls() {
return errcreatetablesqls_.size() > 0;
}
void ProcOdbc::printErrorSqls() { printErrorCreateTableSqls(); }
void ProcOdbc::printErrorCreateTableSqls() {
if (errcreatetablesqls_.size() > 0) {
printf("error sql count: %d\n", (int)errcreatetablesqls_.size());
for (std::list<std::string>::iterator iter = errcreatetablesqls_.begin();
iter != errcreatetablesqls_.end(); iter++) {
printf(" %s\n", iter->c_str());
}
}
}
void ProcOdbc::freeCurrentArgv() {
for (int index = 0; index < currentArgc_; index++) {
free(currentArgv_[index]);
}
free(currentArgv_);
}
}; // namespace DMODBC

84
MySQL2openGauss/src/ProcOdbc.h Normal file
View File

@ -0,0 +1,84 @@
#ifndef _PROCODBC_H_
#define _PROCODBC_H_
#include <list>
#include <string>
#define MYSQL_TO_OPENGAUSS_HOME_DIR "MYSQL_TO_OPENGAUSS_HOME"
namespace DMODBC {
struct DBError {
bool alreadyExistError;
std::string error;
};
struct MysqlInfo {
std::string host;
int port;
std::string username;
std::string password;
std::string database;
std::string dsthost;
int dstport;
std::string dstdatabase;
int loglevel;
};
class ProcOdbc {
public:
ProcOdbc(const char* datasource);
virtual ~ProcOdbc();
static struct MysqlInfo getMysqlInfo();
bool isConnected();
bool connect();
void disConnect();
bool reConnect();
std::list<std::string> procCreateTableSql(const char* sql,
const char* incrementkey);
std::string procProcedureFunctionSql(const char* sql, bool& isfunction);
bool executeCreateTableSql(const char* sql, DBError* dbError);
bool executeInsertSql(const char* sql, DBError* dbError);
bool executeCreateViewSql(const char* sql, DBError* dbError);
bool executeGeneralSql(const char* sql, DBError* dbError);
std::list<std::string> clearAndGetErrCreateTableSqls();
bool executeLeftErrorSqls();
bool hasErrorSqls();
void printErrorSqls();
char** getCurrentArgv();
int getCurrentArgc();
void setLoglevel(int loglevel);
private:
std::string& replaceCreateTableSql(std::string& sql);
std::list<std::string> procCreateTableSqlLine(
const std::string& sql, const std::string& incrementkey);
std::string& formatByteaAtCreateTableSql(std::string& sql);
std::string& formatSetAtCreateTableSql(std::string& sql);
bool executeLeftErrorCreateTableSqls();
bool hasErrorCreateTableSqls();
void printErrorCreateTableSqls();
void freeCurrentArgv();
bool executeInsertSql(const char* sql, const char* replaceNullStr, DBError* dbError);
int execSqlStably(const char* sql);
static bool isBinaryBlobChar(unsigned char ch);
private:
char datasource_[256];
bool isconn_;
void* dbinfo_;
char** currentArgv_;
int currentArgc_;
std::list<std::string> errcreatetablesqls_;
int loglevel_;
};
}; // namespace DMODBC
#endif

5646
MySQL2openGauss/src/mysqltoopengauss.cpp Normal file
View File

File diff suppressed because it is too large Load Diff