[PLUGIN-1813] CloudSQLMySQL Escape column names

[psainics]

CloudSQLMySQL Escape column names

Jira : PLUGIN-1813

Description

If column name is a MySQL reserved keyword say insert the query will fail due to incorrect sql syntax.
This PR fixes this issue by escaping the column names using back ticks.

Similar column escapes is used in other plugins. Ref CloudSQLPostgreSQLSink

Code change

• Modified CloudSQLMySQLSink.java
• Modified MysqlSink.java

Unit Tests

• Added CloudSQLMySQLSinkTest.java
• 
• Added MysqlSinkTest.java
• 

[itsankit-google]

I know this will fix one part of the problem but now this turns into a backward incompatible change because let's say if MySQL by default convert all identifiers to lowercase and if a cx has an existing pipeline which has any column name in uppercase, it will start failing.

The right thing to do would be as described in JIRA to introduce a new config named Field Mapping which can map any CDAP field in the schema to a different name and will be passed as quoted.

[psainics]

I know this will fix one part of the problem but now this turns into a backward incompatible change because let's say if MySQL by default convert all identifiers to lowercase and if a cx has an existing pipeline which has any column name in uppercase, it will start failing.

The right thing to do would be as described in JIRA to introduce a new config named Field Mapping which can map any CDAP field in the schema to a different name and will be passed as quoted.

Mysql Col names are not case sensitive.

[itsankit-google]

Please do the same changes for MySQL sink as well.

[itsankit-google]

I know this will fix one part of the problem but now this turns into a backward incompatible change because let's say if MySQL by default convert all identifiers to lowercase and if a cx has an existing pipeline which has any column name in uppercase, it will start failing.
The right thing to do would be as described in JIRA to introduce a new config named Field Mapping which can map any CDAP field in the schema to a different name and will be passed as quoted.

Mysql Col names are not case sensitivity.

Discussed offline, to address the larger problem in future.

[itsankit-google]

why this change?

[psainics]

Ref:

database-plugins/oracle-plugin/src/main/java/io/cdap/plugin/oracle/OracleSinkDBRecord.java

Lines 44 to 52 in acd47fa

	@Override
	protected void insertOperation(PreparedStatement stmt) throws SQLException {
	for (int fieldIndex = 0; fieldIndex < columnTypes.size(); fieldIndex++) {
	ColumnType columnType = columnTypes.get(fieldIndex);
	// Get the field from the schema using the column name with ignoring case.
	Schema.Field field = record.getSchema().getField(columnType.getName(), true);
	writeToDB(stmt, field, fieldIndex);
	}
	}

This is done to get the cols values from schema with ignoring case.

If cols in CDAP schema are in different case from the cols in the mysql DB, ignoring case is required else we will get null values.