You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@paimon.apache.org by lz...@apache.org on 2023/05/12 09:04:53 UTC
[incubator-paimon] branch master updated: [flink] Support 'substring' for cdc computed column (#1140)
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 91f29adfc [flink] Support 'substring' for cdc computed column (#1140)
91f29adfc is described below
commit 91f29adfc1560c9ec6f971856034d1a0b90f1c0e
Author: yuzelin <33...@users.noreply.github.com>
AuthorDate: Fri May 12 17:04:49 2023 +0800
[flink] Support 'substring' for cdc computed column (#1140)
---
docs/content/how-to/cdc-ingestion.md | 2 +
.../paimon/flink/action/cdc/mysql/Expression.java | 83 +++++++++++++++++++++-
.../flink/action/cdc/mysql/MySqlActionUtils.java | 15 ++--
.../cdc/mysql/MySqlSyncTableActionITCase.java | 16 +++--
4 files changed, 99 insertions(+), 17 deletions(-)
diff --git a/docs/content/how-to/cdc-ingestion.md b/docs/content/how-to/cdc-ingestion.md
index b86cc3c60..40627e2b9 100644
--- a/docs/content/how-to/cdc-ingestion.md
+++ b/docs/content/how-to/cdc-ingestion.md
@@ -59,6 +59,8 @@ To use this feature through `flink run`, run the following shell command.
* `--primary-keys` are the primary keys for Paimon table. If there are multiple primary keys, connect them with comma, for example `buyer_id,seller_id`.
* `--computed-column` are the definitions of computed columns. The argument field is from MySQL table field name. Supported expressions are:
* year(date-column): Extract year from a DATE, DATETIME or TIMESTAMP. Output is an INT value represent the year.
+ * substring(column,beginInclusive): Get column.substring(beginInclusive). Output is a STRING.
+ * substring(column,beginInclusive,endExclusive): Get column.substring(beginInclusive,endExclusive). Output is a STRING.
* `--mysql-conf` is the configuration for Flink CDC MySQL table sources. Each configuration should be specified in the format `key=value`. `hostname`, `username`, `password`, `database-name` and `table-name` are required configurations, others are optional. See its [document](https://ververica.github.io/flink-cdc-connectors/master/content/connectors/mysql-cdc.html#connector-options) for a complete list of configurations.
* `--catalog-conf` is the configuration for Paimon catalog. Each configuration should be specified in the format `key=value`. See [here]({{< ref "maintenance/configurations" >}}) for a complete list of catalog configurations.
* `--table-conf` is the configuration for Paimon table sink. Each configuration should be specified in the format `key=value`. See [here]({{< ref "maintenance/configurations" >}}) for a complete list of table configurations.
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/cdc/mysql/Expression.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/cdc/mysql/Expression.java
index 70ab5994d..45a974dbd 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/cdc/mysql/Expression.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/cdc/mysql/Expression.java
@@ -26,13 +26,15 @@ import javax.annotation.Nullable;
import java.io.Serializable;
import java.time.LocalDateTime;
-import java.util.Collections;
+import java.util.Arrays;
import java.util.List;
+import static org.apache.paimon.utils.Preconditions.checkArgument;
+
/** Produce a computation result for computed column. */
public interface Expression extends Serializable {
- List<String> SUPPORTED_EXPRESSION = Collections.singletonList("year");
+ List<String> SUPPORTED_EXPRESSION = Arrays.asList("year", "substring");
/** Return name of referenced field. */
String fieldReference();
@@ -44,10 +46,12 @@ public interface Expression extends Serializable {
String eval(String input);
static Expression create(
- String exprName, String fieldReference, DataType fieldType, @Nullable String literal) {
+ String exprName, String fieldReference, DataType fieldType, String... literals) {
switch (exprName) {
case "year":
return year(fieldReference);
+ case "substring":
+ return substring(fieldReference, literals);
// TODO: support more expression
default:
throw new UnsupportedOperationException(
@@ -61,6 +65,36 @@ public interface Expression extends Serializable {
return new YearComputer(fieldReference);
}
+ static Expression substring(String fieldReference, String... literals) {
+ checkArgument(
+ literals.length == 1 || literals.length == 2,
+ String.format(
+ "'substring' expression supports one or two arguments, but found '%s'.",
+ literals.length));
+ int beginInclusive;
+ Integer endExclusive;
+ try {
+ beginInclusive = Integer.parseInt(literals[0]);
+ endExclusive = literals.length == 1 ? null : Integer.parseInt(literals[1]);
+ } catch (NumberFormatException e) {
+ throw new RuntimeException(
+ String.format(
+ "The index arguments '%s' contain non integer value.",
+ Arrays.toString(literals)),
+ e);
+ }
+ checkArgument(
+ beginInclusive >= 0,
+ "begin index argument (%s) of 'substring' must be >= 0.",
+ beginInclusive);
+ checkArgument(
+ endExclusive == null || endExclusive > beginInclusive,
+ "end index (%s) must be larger than begin index (%s).",
+ endExclusive,
+ beginInclusive);
+ return new Substring(fieldReference, beginInclusive, endExclusive);
+ }
+
/** Compute year from a time input. */
final class YearComputer implements Expression {
@@ -88,4 +122,47 @@ public interface Expression extends Serializable {
return String.valueOf(localDateTime.getYear());
}
}
+
+ /** Get substring using {@link String#substring}. */
+ final class Substring implements Expression {
+
+ private static final long serialVersionUID = 1L;
+
+ private final String fieldReference;
+ private final int beginInclusive;
+ @Nullable private final Integer endExclusive;
+
+ private Substring(
+ String fieldReference, int beginInclusive, @Nullable Integer endExclusive) {
+ this.fieldReference = fieldReference;
+ this.beginInclusive = beginInclusive;
+ this.endExclusive = endExclusive;
+ }
+
+ @Override
+ public String fieldReference() {
+ return fieldReference;
+ }
+
+ @Override
+ public DataType outputType() {
+ return DataTypes.STRING();
+ }
+
+ @Override
+ public String eval(String input) {
+ try {
+ if (endExclusive == null) {
+ return input.substring(beginInclusive);
+ } else {
+ return input.substring(beginInclusive, endExclusive);
+ }
+ } catch (StringIndexOutOfBoundsException e) {
+ throw new RuntimeException(
+ String.format(
+ "Cannot get substring from '%s' because the indexes are out of range. Begin index: %s, end index: %s.",
+ input, beginInclusive, endExclusive));
+ }
+ }
+ }
}
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlActionUtils.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlActionUtils.java
index 61078207c..ad47b8e80 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlActionUtils.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlActionUtils.java
@@ -41,6 +41,7 @@ import org.apache.kafka.connect.json.JsonConverterConfig;
import java.sql.Connection;
import java.sql.DriverManager;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.List;
@@ -297,15 +298,11 @@ class MySqlActionUtils {
String exprName = expression.substring(0, left);
String[] args = expression.substring(left + 1, right).split(",");
- for (int i = 0; i < args.length; i++) {
- args[i] = args[i].trim();
- }
- checkArgument(
- args.length >= 1 && args.length <= 2,
- "Currently, computed column only supports one or two arguments.");
+ checkArgument(args.length >= 1, "Computed column needs at least one argument.");
- String fieldReference = args[0];
- String literal = args.length == 2 ? args[1] : null;
+ String fieldReference = args[0].trim();
+ String[] literals =
+ Arrays.stream(args).skip(1).map(String::trim).toArray(String[]::new);
checkArgument(
typeMapping.containsKey(fieldReference),
String.format(
@@ -319,7 +316,7 @@ class MySqlActionUtils {
exprName,
fieldReference,
typeMapping.get(fieldReference),
- literal)));
+ literals)));
}
return computedColumns;
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncTableActionITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncTableActionITCase.java
index 02f201cde..4f135df39 100644
--- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncTableActionITCase.java
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncTableActionITCase.java
@@ -786,7 +786,9 @@ public class MySqlSyncTableActionITCase extends MySqlActionITCaseBase {
Arrays.asList(
"_year_date=year(_date)",
"_year_datetime=year(_datetime)",
- "_year_timestamp=year(_timestamp)");
+ "_year_timestamp=year(_timestamp)",
+ "_substring_date1=substring(_date,2)",
+ "_substring_date2=substring(_timestamp,5,10)");
MySqlSyncTableAction action =
new MySqlSyncTableAction(
@@ -825,7 +827,9 @@ public class MySqlSyncTableActionITCase extends MySqlActionITCaseBase {
DataTypes.TIMESTAMP(0),
DataTypes.INT().notNull(),
DataTypes.INT(),
- DataTypes.INT()
+ DataTypes.INT(),
+ DataTypes.STRING(),
+ DataTypes.STRING()
},
new String[] {
"pk",
@@ -834,12 +838,14 @@ public class MySqlSyncTableActionITCase extends MySqlActionITCaseBase {
"_timestamp",
"_year_date",
"_year_datetime",
- "_year_timestamp"
+ "_year_timestamp",
+ "_substring_date1",
+ "_substring_date2"
});
List<String> expected =
Arrays.asList(
- "+I[1, 19439, 2022-01-01T14:30, 2021-09-15T15:00:10, 2023, 2022, 2021]",
- "+I[2, 19439, NULL, NULL, 2023, NULL, NULL]");
+ "+I[1, 19439, 2022-01-01T14:30, 2021-09-15T15:00:10, 2023, 2022, 2021, 23-03-23, 09-15]",
+ "+I[2, 19439, NULL, NULL, 2023, NULL, NULL, 23-03-23, NULL]");
waitForResult(expected, table, rowType, Arrays.asList("pk", "_year_date"));
}
}