Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

[fix] Fix garbled of table or column comments contain Chinese characters(#401) #403

Merged
merged 2 commits into from
Jun 18, 2024
Merged

[fix] Fix garbled of table or column comments contain Chinese characters(#401) #403

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
a5ccb92
[fix] Fix garbled of table or column comments contain Chinese charact…
qg-lin Jun 13, 2024
e9585a6
[fix] Fix garbled of table or column comments contain Chinese charact…
qg-lin Jun 17, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
41 changes: 37 additions & 4 deletions flink-doris-connector/src/main/java/org/apache/doris/flink/cfg/DorisOptions.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,7 @@ public class DorisOptions extends DorisConnectionOptions {
private static final long serialVersionUID = 1L;

private String tableIdentifier;
private String charsetEncoding = "UTF-8";

public DorisOptions(String fenodes, String username, String password, String tableIdentifier) {
super(fenodes, username, password);
Expand All @@ -50,9 +51,11 @@ public DorisOptions(
String password,
String tableIdentifier,
String jdbcUrl,
boolean redirect) {
boolean redirect,
String charsetEncoding) {
JNSimba marked this conversation as resolved.
Show resolved Hide resolved
super(fenodes, beNodes, username, password, jdbcUrl, redirect);
this.tableIdentifier = tableIdentifier;
this.charsetEncoding = charsetEncoding;
}

public String getTableIdentifier() {
Expand All @@ -63,6 +66,14 @@ public void setTableIdentifier(String tableIdentifier) {
this.tableIdentifier = tableIdentifier;
}

public String getCharsetEncoding() {
return charsetEncoding;
}

public void setCharsetEncoding(String charsetEncoding) {
this.charsetEncoding = charsetEncoding;
}

public static Builder builder() {
return new Builder();
}
Expand All @@ -82,13 +93,21 @@ public boolean equals(Object o) {
&& Objects.equals(username, that.username)
&& Objects.equals(password, that.password)
&& Objects.equals(jdbcUrl, that.jdbcUrl)
&& Objects.equals(benodes, that.benodes);
&& Objects.equals(benodes, that.benodes)
&& Objects.equals(charsetEncoding, that.charsetEncoding);
}

@Override
public int hashCode() {
return Objects.hash(
fenodes, username, password, jdbcUrl, benodes, autoRedirect, tableIdentifier);
fenodes,
username,
password,
jdbcUrl,
benodes,
autoRedirect,
tableIdentifier,
charsetEncoding);
}

/** Builder of {@link DorisOptions}. */
Expand All @@ -100,6 +119,7 @@ public static class Builder {
private String password;
private boolean autoRedirect = true;
private String tableIdentifier;
private String charsetEncoding = "UTF-8";

/** required, tableIdentifier. */
public Builder setTableIdentifier(String tableIdentifier) {
Expand Down Expand Up @@ -142,12 +162,25 @@ public Builder setAutoRedirect(boolean autoRedirect) {
return this;
}

/** optional, Charset encoding for Http client, default UTF-8. */
public Builder setCharsetEncoding(String charsetEncoding) {
this.charsetEncoding = charsetEncoding;
return this;
}

public DorisOptions build() {
checkNotNull(fenodes, "No fenodes supplied.");
// multi table load, don't need check
// checkNotNull(tableIdentifier, "No tableIdentifier supplied.");
return new DorisOptions(
fenodes, benodes, username, password, tableIdentifier, jdbcUrl, autoRedirect);
fenodes,
benodes,
username,
password,
tableIdentifier,
jdbcUrl,
autoRedirect,
charsetEncoding);
}
}
}
13 changes: 10 additions & 3 deletions ...doris-connector/src/main/java/org/apache/doris/flink/sink/schema/SchemaChangeManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -133,7 +133,10 @@ public boolean checkSchemaChange(String database, String table, Map<String, Obje
table);
HttpGetWithEntity httpGet = new HttpGetWithEntity(requestUrl);
httpGet.setHeader(HttpHeaders.AUTHORIZATION, authHeader());
httpGet.setEntity(new StringEntity(objectMapper.writeValueAsString(params)));
httpGet.setEntity(
new StringEntity(
objectMapper.writeValueAsString(params),
dorisOptions.getCharsetEncoding()));
String responseEntity = "";
Map<String, Object> responseMap = handleResponse(httpGet, responseEntity);
return handleSchemaChange(responseMap, responseEntity);
Expand Down Expand Up @@ -173,8 +176,12 @@ public HttpPost buildHttpPost(String ddl, String database)
database);
HttpPost httpPost = new HttpPost(requestUrl);
httpPost.setHeader(HttpHeaders.AUTHORIZATION, authHeader());
httpPost.setHeader(HttpHeaders.CONTENT_TYPE, "application/json");
httpPost.setEntity(new StringEntity(objectMapper.writeValueAsString(param)));
httpPost.setHeader(
HttpHeaders.CONTENT_TYPE,
String.format("application/json;charset=%s", dorisOptions.getCharsetEncoding()));
httpPost.setEntity(
new StringEntity(
objectMapper.writeValueAsString(param), dorisOptions.getCharsetEncoding()));
return httpPost;
}

Expand Down
57 changes: 57 additions & 0 deletions ...doris-connector/src/test/java/org/apache/doris/flink/sink/schema/SchemaManagerITCase.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,8 +28,11 @@
import java.io.IOException;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.HashMap;
import java.util.Map;

public class SchemaManagerITCase extends DorisTestBase {

Expand Down Expand Up @@ -82,6 +85,60 @@ public void testAddColumn() throws SQLException, IOException, IllegalArgumentExc
Assert.assertTrue(exists);
}

@Test
public void testAddColumnWithChineseComment()
throws SQLException, IOException, IllegalArgumentException {
String addColumnTbls = "add_column";
initDorisSchemaChangeTable(addColumnTbls);

// add a column by UTF-8 encoding
String addColumnName = "col_with_comment1";
String chineseComment = "中文注释1";
addColumnWithChineseCommentAndAssert(addColumnTbls, addColumnName, chineseComment, true);

// change charset encoding to US-ASCII would cause garbled of Chinese.
options.setCharsetEncoding("US-ASCII");
addColumnName = "col_with_comment2";
chineseComment = "中文注释2";
addColumnWithChineseCommentAndAssert(addColumnTbls, addColumnName, chineseComment, false);
}

private void addColumnWithChineseCommentAndAssert(
String tableName, String addColumnName, String chineseComment, boolean assertFlag)
throws SQLException, IOException, IllegalArgumentException {
FieldSchema field = new FieldSchema(addColumnName, "string", chineseComment);
schemaChangeManager.addColumn(DATABASE, tableName, field);
boolean exists = schemaChangeManager.addColumn(DATABASE, tableName, field);
Assert.assertTrue(exists);

exists = schemaChangeManager.checkColumnExists(DATABASE, tableName, addColumnName);
Assert.assertTrue(exists);

// check Chinese comment
Map<String, String> columnComments = getColumnComments(tableName);
if (assertFlag) {
Assert.assertEquals(columnComments.get(addColumnName), chineseComment);
} else {
Assert.assertNotEquals(columnComments.get(addColumnName), chineseComment);
}
}

private Map<String, String> getColumnComments(String table) throws SQLException {
Map<String, String> columnCommentsMap = new HashMap<>();
try (Connection connection =
DriverManager.getConnection(
String.format(URL, DORIS_CONTAINER.getHost()), USERNAME, PASSWORD)) {
ResultSet columns = connection.getMetaData().getColumns(null, DATABASE, table, null);

while (columns.next()) {
String columnName = columns.getString("COLUMN_NAME");
String comment = columns.getString("REMARKS");
columnCommentsMap.put(columnName, comment);
}
}
return columnCommentsMap;
}

@Test
public void testDropColumn() throws SQLException, IOException, IllegalArgumentException {
String dropColumnTbls = "drop_column";
Expand Down
6 changes: 6 additions & 0 deletions ...k-doris-connector/src/test/java/org/apache/doris/flink/sink/schema/SchemaManagerTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -132,6 +132,12 @@ public void testAddColumn() {
Assert.assertEquals(
"ALTER TABLE `test`.`test_flink` ADD COLUMN `col` int DEFAULT current_timestamp COMMENT 'comment \"\\'sdf\\''",
addColumnDDL);

field = new FieldSchema("col", "int", "current_timestamp", "中文注释");
addColumnDDL = SchemaChangeHelper.buildAddColumnDDL("test.test_flink", field);
Assert.assertEquals(
"ALTER TABLE `test`.`test_flink` ADD COLUMN `col` int DEFAULT current_timestamp COMMENT '中文注释'",
addColumnDDL);
}

@Test
Expand Down
Loading