yuxiqian commented on code in PR #4488:
URL: https://github.com/apache/flink-cdc/pull/4488#discussion_r3663764197
##########
flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/functions/impl/StringFunctions.java:
##########
@@ -176,6 +197,193 @@ public static String lower(String str) {
return str.toLowerCase();
}
+ public static String overlay(String str, String replacement, Number start)
{
+ if (replacement == null) {
+ return null;
+ }
+ return overlay(str, replacement, start, replacement.length());
+ }
+
+ public static String overlay(String str, String replacement, Number start,
Number length) {
+ if (str == null || replacement == null || start == null || length ==
null) {
+ return null;
+ }
+ int startPosition = start.intValue();
+ int len = length.intValue();
+ if (startPosition <= 0 || startPosition > str.length()) {
+ return str;
+ }
+
+ StringBuilder builder = new StringBuilder();
+ builder.append(str, 0, startPosition - 1);
+ builder.append(replacement);
+ if (startPosition + len <= str.length() && len > 0) {
+ builder.append(str.substring(startPosition - 1 + len));
+ }
+ return builder.toString();
+ }
+
+ public static Integer position(String seek, String str) {
+ return position(seek, str, 1);
+ }
+
+ public static Integer position(String seek, String str, Number from) {
+ if (seek == null || str == null || from == null) {
+ return null;
+ }
+ if (seek.isEmpty()) {
+ return 1;
+ }
+ int fromCodePoint = Math.max(from.intValue() - 1, 0);
+ int codePointLength = str.codePointCount(0, str.length());
+ if (fromCodePoint > codePointLength) {
+ return 0;
+ }
+ int fromIndex = str.offsetByCodePoints(0, fromCodePoint);
+ int index = str.indexOf(seek, fromIndex);
+ return index < 0 ? 0 : str.codePointCount(0, index) + 1;
+ }
+
+ public static Integer instr(String str, String subString) {
+ if (str == null || subString == null) {
+ return null;
+ }
+ int index = str.indexOf(subString);
+ return index < 0 ? 0 : str.codePointCount(0, index) + 1;
+ }
+
+ public static Integer locate(String seek, String str) {
+ return position(seek, str);
+ }
+
+ public static Integer locate(String seek, String str, Number from) {
+ return position(seek, str, from);
+ }
+
+ public static String ltrim(String str) {
+ return ltrim(str, " ");
+ }
+
+ public static String ltrim(String str, String trimStr) {
+ return trim(str, trimStr, true, false);
+ }
+
+ public static String rtrim(String str) {
+ return rtrim(str, " ");
+ }
+
+ public static String rtrim(String str, String trimStr) {
+ return trim(str, trimStr, false, true);
+ }
+
+ public static String btrim(String str) {
+ return btrim(str, " ");
+ }
+
+ public static String btrim(String str, String trimStr) {
+ return trim(str, trimStr, true, true);
+ }
+
+ public static String lpad(String base, Number len, String pad) {
+ return pad(base, len, pad, true);
+ }
+
+ public static String rpad(String base, Number len, String pad) {
+ return pad(base, len, pad, false);
+ }
+
+ public static String replace(String str, String oldStr, String
replacement) {
+ if (str == null || oldStr == null || replacement == null) {
+ return null;
+ }
+ return str.replace(oldStr, replacement);
+ }
+
+ public static String repeat(String str, Number repeat) {
+ if (str == null || repeat == null) {
+ return null;
+ }
+ int count = repeat.intValue();
+ if (count <= 0) {
+ return "";
+ }
+ return str.repeat(count);
+ }
+
+ public static String left(String str, Number length) {
+ if (str == null || length == null) {
+ return null;
+ }
+ int len = length.intValue();
+ if (len <= 0) {
+ return "";
+ }
+ int codePointLength = str.codePointCount(0, str.length());
+ if (len >= codePointLength) {
+ return str;
+ }
+ return str.substring(0, str.offsetByCodePoints(0, len));
+ }
+
+ public static String right(String str, Number length) {
+ if (str == null || length == null) {
+ return null;
+ }
+ int len = length.intValue();
+ if (len <= 0) {
+ return "";
+ }
+ int codePointLength = str.codePointCount(0, str.length());
+ if (len >= codePointLength) {
+ return str;
+ }
+ return str.substring(str.offsetByCodePoints(0, codePointLength - len));
+ }
+
+ public static Boolean startswith(String str, String prefix) {
+ if (str == null || prefix == null) {
+ return null;
+ }
+ return str.startsWith(prefix);
+ }
+
+ public static Boolean startswith(byte[] bytes, byte[] prefix) {
+ if (bytes == null || prefix == null) {
+ return null;
+ }
+ return matchesAt(bytes, prefix, 0);
+ }
+
+ public static Boolean endswith(String str, String suffix) {
+ if (str == null || suffix == null) {
+ return null;
+ }
+ return str.endsWith(suffix);
+ }
+
+ public static Boolean endswith(byte[] bytes, byte[] suffix) {
+ if (bytes == null || suffix == null) {
+ return null;
+ }
+ return matchesAt(bytes, suffix, bytes.length - suffix.length);
+ }
+
+ public static String toBase64(String str) {
+ if (str == null) {
+ return null;
+ }
+ return
Base64.getEncoder().encodeToString(str.getBytes(StandardCharsets.UTF_8));
+ }
+
+ public static String fromBase64(String str) {
+ if (str == null) {
+ return null;
+ }
+ return new String(
+
Base64.getDecoder().decode(str.getBytes(StandardCharsets.UTF_8)),
+ StandardCharsets.UTF_8);
Review Comment:
If decoded bytes are not valid UTF-8 string, it will be replaced by
`U+FFFD`. Maybe we should provide a variant function for base64 string <->
bytes.
##########
flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/parser/TransformParser.java:
##########
@@ -196,12 +196,33 @@ public static SqlSelect parseSelect(String statement) {
throw new ParseException("Statements can not be parsed.", e);
}
if (sqlNode instanceof SqlSelect) {
+ replaceParserBoundStringOperators(sqlNode);
return (SqlSelect) sqlNode;
} else {
throw new ParseException("Only select statements can be parsed.");
}
}
+ private static void replaceParserBoundStringOperators(SqlNode sqlNode) {
+ if (sqlNode instanceof SqlBasicCall) {
+ SqlBasicCall call = (SqlBasicCall) sqlNode;
+ if ("OVERLAY".equalsIgnoreCase(call.getOperator().getName())) {
+ call.setOperator(TransformSqlOperatorTable.OVERLAY);
+ } else if (call.getKind() == SqlKind.POSITION) {
+ call.setOperator(TransformSqlOperatorTable.POSITION);
+ }
Review Comment:
This would be unnecessary if we use Calcite standard
`SqlStdOperatorTable.OVERLAY` and `.POSITION`.
##########
docs/content.zh/docs/core-concept/transform.md:
##########
@@ -177,10 +177,28 @@ Flink CDC 使用 [Calcite](https://calcite.apache.org/)
来解析表达式并且
| UPPER(string) | upper(string)
| 返回大写形式的字符串。
|
| LOWER(string) | lower(string)
| 返回小写形式的字符串。
|
| TRIM(string1) | trim('BOTH',string1)
| 返回去除两端空格的字符串。
|
+| LTRIM(string[, trimString]) | ltrim(string[,
trimString]) | 返回去除开头 trimString 字符后的字符串，默认去除空格。
|
+| RTRIM(string[, trimString]) | rtrim(string[,
trimString]) | 返回去除末尾 trimString 字符后的字符串，默认去除空格。
|
+| BTRIM(string[, trimString]) | btrim(string[,
trimString]) | 返回去除开头和末尾 trimString 字符后的字符串，默认去除空格。
|
| REGEXP_REPLACE(string1, string2, string3) | regexpReplace(string1,
string2, string3) | 返回将 STRING1 中所有匹配正则表达式 STRING2 的子串替换为 STRING3
后的字符串。例如，'foobar'.regexpReplace('oo\|ar', '') 返回 "fb"。 |
| SUBSTR(string, integer1[, integer2]) |
substr(string,integer1,integer2) | 返回 STRING 从位置 integer1 开始、长度为
integer2（默认到末尾）的子串。
|
| SUBSTRING(string FROM integer1 [ FOR integer2 ]) |
substring(string,integer1,integer2) | 返回 STRING 从位置 integer1 开始、长度为
integer2（默认到末尾）的子串。
|
+| OVERLAY(string1 PLACING string2 FROM integer1 [FOR integer2]) |
overlay(string1, string2, integer1[, integer2]) | 从位置 integer1 开始，用 STRING2 替换
STRING1 的子串，替换长度默认为 STRING2 的长度。
|
+| POSITION(string1 IN string2) |
position(string1, string2) | 返回 STRING1 在 STRING2
中第一次出现的位置。起始位置为 1，未找到时返回 0。
|
+| LOCATE(string1, string2[, integer]) | locate(string1,
string2[, integer]) | 返回 STRING1 在 STRING2 中第一次出现的位置，可指定从 integer
开始查找。起始位置为 1，未找到时返回 0。 |
+| INSTR(string1, string2) | instr(string1,
string2) | 返回 STRING2 在 STRING1 中第一次出现的位置。起始位置为 1，未找到时返回
0。 |
| CONCAT(string1, string2,…) | concat(string1,
string2,…) | 返回连接 string1、string2、… 后的字符串。例如，CONCAT('AA', 'BB',
'CC') 返回 'AABBCC'。 |
+| CONCAT_WS(separator, string1, string2,...) |
concatWs(separator, string1, string2,...) | 使用分隔符连接 string1、string2、...
后返回字符串。NULL 字符串参数会被跳过。
|
+| LPAD(string1, integer, string2) | lpad(string1,
integer, string2) | 返回使用 STRING2 左填充 STRING1 至 integer
个字符后的字符串。如果 STRING1 更长，则截断到 integer 个字符。 |
+| RPAD(string1, integer, string2) | rpad(string1,
integer, string2) | 返回使用 STRING2 右填充 STRING1 至 integer
个字符后的字符串。如果 STRING1 更长，则截断到 integer 个字符。 |
+| REPLACE(string1, string2, string3) |
replace(string1, string2, string3) | 返回将 STRING1 中所有 STRING2 替换为
STRING3 后的字符串。
|
+| REPEAT(string, integer) | repeat(string,
integer) | 返回将 STRING 重复 integer 次后的字符串。
|
+| LEFT(string, integer) | left(string,
integer) | 返回 STRING 最左侧 integer 个字符。
|
+| RIGHT(string, integer) | right(string,
integer) | 返回 STRING 最右侧 integer 个字符。
|
+| STARTSWITH(string1, string2) |
startswith(string1, string2) | 返回 STRING1 是否以 STRING2
开头，支持字符字符串和二进制字符串。
|
+| ENDSWITH(string1, string2) |
endswith(string1, string2) | 返回 STRING1 是否以 STRING2
结尾，支持字符字符串和二进制字符串。
|
+| TO_BASE64(string) |
toBase64(string) | 将 STRING 编码为 base64 字符串。
|
+| FROM_BASE64(string) |
fromBase64(string) | 解码 base64 字符串。
|
Review Comment:
Ditto, supporting FROM/TO_BASE64 with generic binary would be useful.
##########
flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/parser/metadata/TransformSqlOperatorTable.java:
##########
@@ -174,6 +242,38 @@ public void lookupOperatorOverloads(
public static final SqlFunction UPPER = SqlStdOperatorTable.UPPER;
public static final SqlFunction LOWER = SqlStdOperatorTable.LOWER;
public static final SqlFunction TRIM = SqlStdOperatorTable.TRIM;
+ public static final SqlFunction LTRIM =
Review Comment:
We may use definitions in SqlStdOperatorTable if possible.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
