This is an automated email from the ASF dual-hosted git repository. leonard pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/flink-cdc.git
commit c92903016dee22e67bcea9d418b257f90738025b Author: wenmo <[email protected]> AuthorDate: Wed Mar 20 20:03:49 2024 +0800 [cdc-common] Introduce partitionKeys into schema and add related util classes --- .../flink/cdc/common/event/DataChangeEvent.java | 28 +++++ .../org/apache/flink/cdc/common/schema/Schema.java | 60 +++++++++- .../flink/cdc/common/utils/DateTimeUtils.java | 122 +++++++++++++++++++++ .../apache/flink/cdc/common/utils/SchemaUtils.java | 43 ++------ .../apache/flink/cdc/common/utils/StringUtils.java | 26 +++++ .../flink/cdc/common/utils/ThreadLocalCache.java | 87 +++++++++++++++ .../flink/cdc/common/utils/StringUtilsTest.java} | 29 ++--- 7 files changed, 340 insertions(+), 55 deletions(-) diff --git a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/event/DataChangeEvent.java b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/event/DataChangeEvent.java index 1f0289056..da4d454dd 100644 --- a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/event/DataChangeEvent.java +++ b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/event/DataChangeEvent.java @@ -139,6 +139,34 @@ public class DataChangeEvent implements ChangeEvent, Serializable { return new DataChangeEvent(tableId, null, after, OperationType.REPLACE, meta); } + /** + * Updates the before of a {@link DataChangeEvent} instance that describes the event with meta + * info. + */ + public static DataChangeEvent projectBefore( + DataChangeEvent dataChangeEvent, RecordData projectedBefore) { + return new DataChangeEvent( + dataChangeEvent.tableId, + projectedBefore, + dataChangeEvent.after, + dataChangeEvent.op, + dataChangeEvent.meta); + } + + /** + * Updates the after of a {@link DataChangeEvent} instance that describes the event with meta + * info. + */ + public static DataChangeEvent projectAfter( + DataChangeEvent dataChangeEvent, RecordData projectedAfter) { + return new DataChangeEvent( + dataChangeEvent.tableId, + dataChangeEvent.before, + projectedAfter, + dataChangeEvent.op, + dataChangeEvent.meta); + } + @Override public boolean equals(Object o) { if (this == o) { diff --git a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/schema/Schema.java b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/schema/Schema.java index 31110cfd5..62657be88 100644 --- a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/schema/Schema.java +++ b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/schema/Schema.java @@ -51,6 +51,8 @@ public class Schema implements Serializable { private final List<String> primaryKeys; + private final List<String> partitionKeys; + private final Map<String, String> options; private final @Nullable String comment; @@ -67,6 +69,20 @@ public class Schema implements Serializable { this.primaryKeys = primaryKeys; this.options = options; this.comment = comment; + this.partitionKeys = new ArrayList<>(); + } + + private Schema( + List<Column> columns, + List<String> primaryKeys, + List<String> partitionKeys, + Map<String, String> options, + @Nullable String comment) { + this.columns = columns; + this.primaryKeys = primaryKeys; + this.partitionKeys = partitionKeys; + this.options = options; + this.comment = comment; } /** Returns the number of columns of this schema. */ @@ -96,6 +112,11 @@ public class Schema implements Serializable { return primaryKeys; } + /** Returns the partition keys of the table or data collection. */ + public List<String> partitionKeys() { + return partitionKeys; + } + /** Returns the options of the table or data collection. */ public Map<String, String> options() { return options; @@ -139,7 +160,12 @@ public class Schema implements Serializable { /** Returns a copy of the schema with a replaced list of {@Column}. */ public Schema copy(List<Column> columns) { - return new Schema(columns, new ArrayList<>(primaryKeys), new HashMap<>(options), comment); + return new Schema( + columns, + new ArrayList<>(primaryKeys), + new ArrayList<>(partitionKeys), + new HashMap<>(options), + comment); } @Override @@ -153,13 +179,14 @@ public class Schema implements Serializable { Schema schema = (Schema) o; return Objects.equals(columns, schema.columns) && Objects.equals(primaryKeys, schema.primaryKeys) + && Objects.equals(partitionKeys, schema.partitionKeys) && Objects.equals(options, schema.options) && Objects.equals(comment, schema.comment); } @Override public int hashCode() { - return Objects.hash(columns, primaryKeys, options, comment); + return Objects.hash(columns, primaryKeys, partitionKeys, options, comment); } // ----------------------------------------------------------------------------------- @@ -200,6 +227,9 @@ public class Schema implements Serializable { } sb.append("}"); sb.append(", primaryKeys=").append(String.join(";", primaryKeys)); + if (!partitionKeys.isEmpty()) { + sb.append(", partitionKeys=").append(String.join(";", partitionKeys)); + } sb.append(", options=").append(describeOptions()); return sb.toString(); @@ -213,6 +243,7 @@ public class Schema implements Serializable { private List<Column> columns; private List<String> primaryKeys; + private List<String> partitionKeys; private final Map<String, String> options; private @Nullable String comment; @@ -221,11 +252,34 @@ public class Schema implements Serializable { public Builder() { this.primaryKeys = new ArrayList<>(); + this.partitionKeys = new ArrayList<>(); this.options = new HashMap<>(); this.columns = new ArrayList<>(); this.columnNames = new HashSet<>(); } + /** + * Declares a partition key constraint for a set of given columns. Partition key uniquely + * identify a row in a table. Neither of columns in a partition can be nullable. + * + * @param columnNames columns that form a unique primary key + */ + public Builder partitionKey(String... columnNames) { + this.partitionKeys = Arrays.asList(columnNames); + return this; + } + + /** + * Declares a partition key constraint for a set of given columns. Partition key uniquely + * identify a row in a table. Neither of columns in a partition can be nullable. + * + * @param columnNames columns that form a unique primary key + */ + public Builder partitionKey(List<String> columnNames) { + this.partitionKeys = new ArrayList<>(columnNames); + return this; + } + /** Adopts all fields of the given row as physical columns of the schema. */ public Builder fromRowDataType(DataType type) { Preconditions.checkNotNull(type, "Data type must not be null."); @@ -367,7 +421,7 @@ public class Schema implements Serializable { /** Returns an instance of a {@link Schema}. */ public Schema build() { - return new Schema(columns, primaryKeys, options, comment); + return new Schema(columns, primaryKeys, partitionKeys, options, comment); } } } diff --git a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/utils/DateTimeUtils.java b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/utils/DateTimeUtils.java new file mode 100644 index 000000000..9a009df62 --- /dev/null +++ b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/utils/DateTimeUtils.java @@ -0,0 +1,122 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.cdc.common.utils; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.text.ParseException; +import java.text.SimpleDateFormat; +import java.time.Instant; +import java.time.ZoneId; +import java.time.ZonedDateTime; +import java.util.Date; +import java.util.TimeZone; + +/** Utility functions for datetime types: date, time, timestamp. */ +public class DateTimeUtils { + + private static final Logger LOG = LoggerFactory.getLogger(DateTimeUtils.class); + + /** The julian date of the epoch, 1970-01-01. */ + public static final int EPOCH_JULIAN = 2440588; + + /** + * The number of milliseconds in a day. + * + * <p>This is the modulo 'mask' used when converting TIMESTAMP values to DATE and TIME values. + */ + public static final long MILLIS_PER_DAY = 86400000L; // = 24 * 60 * 60 * 1000 + + /** + * A ThreadLocal cache map for SimpleDateFormat, because SimpleDateFormat is not thread-safe. + * (string_format) => formatter + */ + private static final ThreadLocalCache<String, SimpleDateFormat> FORMATTER_CACHE = + ThreadLocalCache.of(SimpleDateFormat::new); + + // -------------------------------------------------------------------------------------------- + // TIMESTAMP to DATE/TIME utils + // -------------------------------------------------------------------------------------------- + + /** + * Get date from a timestamp. + * + * @param ts the timestamp in milliseconds. + * @return the date in days. + */ + public static int timestampMillisToDate(long ts) { + int days = (int) (ts / MILLIS_PER_DAY); + if (days < 0) { + days = days - 1; + } + return days; + } + + /** + * Get time from a timestamp. + * + * @param ts the timestamp in milliseconds. + * @return the time in milliseconds. + */ + public static int timestampMillisToTime(long ts) { + return (int) (ts % MILLIS_PER_DAY); + } + + // -------------------------------------------------------------------------------------------- + // Parsing functions + // -------------------------------------------------------------------------------------------- + /** Returns the epoch days since 1970-01-01. */ + public static int parseDate(String dateStr, String fromFormat) { + // It is OK to use UTC, we just want get the epoch days + // TODO use offset, better performance + long ts = internalParseTimestampMillis(dateStr, fromFormat, TimeZone.getTimeZone("UTC")); + ZoneId zoneId = ZoneId.of("UTC"); + Instant instant = Instant.ofEpochMilli(ts); + ZonedDateTime zdt = ZonedDateTime.ofInstant(instant, zoneId); + return ymdToUnixDate(zdt.getYear(), zdt.getMonthValue(), zdt.getDayOfMonth()); + } + + private static long internalParseTimestampMillis(String dateStr, String format, TimeZone tz) { + SimpleDateFormat formatter = FORMATTER_CACHE.get(format); + formatter.setTimeZone(tz); + try { + Date date = formatter.parse(dateStr); + return date.getTime(); + } catch (ParseException e) { + LOG.error( + String.format( + "Exception when parsing datetime string '%s' in format '%s'", + dateStr, format), + e); + return Long.MIN_VALUE; + } + } + + private static int ymdToUnixDate(int year, int month, int day) { + final int julian = ymdToJulian(year, month, day); + return julian - EPOCH_JULIAN; + } + + private static int ymdToJulian(int year, int month, int day) { + int a = (14 - month) / 12; + int y = year + 4800 - a; + int m = month + 12 * a - 3; + return day + (153 * m + 2) / 5 + 365 * y + y / 4 - y / 100 + y / 400 - 32045; + } +} diff --git a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/utils/SchemaUtils.java b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/utils/SchemaUtils.java index 1d8fa1e80..107e63d82 100644 --- a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/utils/SchemaUtils.java +++ b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/utils/SchemaUtils.java @@ -41,9 +41,17 @@ public class SchemaUtils { * RecordData. */ public static List<RecordData.FieldGetter> createFieldGetters(Schema schema) { - List<RecordData.FieldGetter> fieldGetters = new ArrayList<>(schema.getColumns().size()); - for (int i = 0; i < schema.getColumns().size(); i++) { - fieldGetters.add(RecordData.createFieldGetter(schema.getColumns().get(i).getType(), i)); + return createFieldGetters(schema.getColumns()); + } + + /** + * create a list of {@link RecordData.FieldGetter} from given {@link Column} to get Object from + * RecordData. + */ + public static List<RecordData.FieldGetter> createFieldGetters(List<Column> columns) { + List<RecordData.FieldGetter> fieldGetters = new ArrayList<>(columns.size()); + for (int i = 0; i < columns.size(); i++) { + fieldGetters.add(RecordData.createFieldGetter(columns.get(i).getType(), i)); } return fieldGetters; } @@ -69,11 +77,6 @@ public class SchemaUtils { private static Schema applyAddColumnEvent(AddColumnEvent event, Schema oldSchema) { LinkedList<Column> columns = new LinkedList<>(oldSchema.getColumns()); for (AddColumnEvent.ColumnWithPosition columnWithPosition : event.getAddedColumns()) { - if (columns.contains(columnWithPosition.getAddColumn())) { - throw new IllegalArgumentException( - columnWithPosition.getAddColumn().getName() - + " of AddColumnEvent is already existed"); - } switch (columnWithPosition.getPosition()) { case FIRST: { @@ -123,14 +126,6 @@ public class SchemaUtils { } private static Schema applyDropColumnEvent(DropColumnEvent event, Schema oldSchema) { - event.getDroppedColumnNames() - .forEach( - column -> { - if (!oldSchema.getColumn(column).isPresent()) { - throw new IllegalArgumentException( - column + " of DropColumnEvent is not existed"); - } - }); List<Column> columns = oldSchema.getColumns().stream() .filter( @@ -141,14 +136,6 @@ public class SchemaUtils { } private static Schema applyRenameColumnEvent(RenameColumnEvent event, Schema oldSchema) { - event.getNameMapping() - .forEach( - (name, newName) -> { - if (!oldSchema.getColumn(name).isPresent()) { - throw new IllegalArgumentException( - name + " of RenameColumnEvent is not existed"); - } - }); List<Column> columns = new ArrayList<>(); oldSchema .getColumns() @@ -165,14 +152,6 @@ public class SchemaUtils { } private static Schema applyAlterColumnTypeEvent(AlterColumnTypeEvent event, Schema oldSchema) { - event.getTypeMapping() - .forEach( - (name, newType) -> { - if (!oldSchema.getColumn(name).isPresent()) { - throw new IllegalArgumentException( - name + " of AlterColumnTypeEvent is not existed"); - } - }); List<Column> columns = new ArrayList<>(); oldSchema .getColumns() diff --git a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/utils/StringUtils.java b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/utils/StringUtils.java index 7e539e48c..8728ac804 100644 --- a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/utils/StringUtils.java +++ b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/utils/StringUtils.java @@ -39,4 +39,30 @@ public class StringUtils { } return true; } + + /** + * Convert underline and space case to camel case. + * + * @param str The string to convert + * @return String which is camel case. + */ + public static String convertToCamelCase(String str) { + StringBuilder result = new StringBuilder(); + if (isNullOrWhitespaceOnly(str)) { + return ""; + } else if (!str.contains("_") && !str.contains(" ")) { + return str.toLowerCase(); + } + String[] camels = str.split("[_| ]"); + for (String camel : camels) { + if (camel.isEmpty()) { + continue; + } + result.append(camel.substring(0, 1).toUpperCase()); + result.append(camel.substring(1).toLowerCase()); + } + StringBuilder ret = new StringBuilder(result.substring(0, 1).toLowerCase()); + ret.append(result.substring(1, result.toString().length())); + return ret.toString(); + } } diff --git a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/utils/ThreadLocalCache.java b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/utils/ThreadLocalCache.java new file mode 100644 index 000000000..f4b8101ed --- /dev/null +++ b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/utils/ThreadLocalCache.java @@ -0,0 +1,87 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.cdc.common.utils; + +import org.apache.flink.annotation.Internal; + +import java.util.LinkedHashMap; +import java.util.Map; +import java.util.function.Function; + +/** + * Provides a thread local cache with a maximum cache size per thread. + * + * <p>Note: Values must not be null. + */ +@Internal +public abstract class ThreadLocalCache<K, V> { + + private static final int DEFAULT_CACHE_SIZE = 64; + + private final ThreadLocal<BoundedMap<K, V>> cache = new ThreadLocal<>(); + private final int maxSizePerThread; + + protected ThreadLocalCache() { + this(DEFAULT_CACHE_SIZE); + } + + protected ThreadLocalCache(int maxSizePerThread) { + this.maxSizePerThread = maxSizePerThread; + } + + public V get(K key) { + BoundedMap<K, V> map = cache.get(); + if (map == null) { + map = new BoundedMap<>(maxSizePerThread); + cache.set(map); + } + V value = map.get(key); + if (value == null) { + value = getNewInstance(key); + map.put(key, value); + } + return value; + } + + public abstract V getNewInstance(K key); + + private static class BoundedMap<K, V> extends LinkedHashMap<K, V> { + + private static final long serialVersionUID = 1L; + + private final int maxSize; + + private BoundedMap(int maxSize) { + this.maxSize = maxSize; + } + + @Override + protected boolean removeEldestEntry(Map.Entry<K, V> eldest) { + return this.size() > maxSize; + } + } + + public static <K, V> ThreadLocalCache<K, V> of(Function<K, V> creator) { + return new ThreadLocalCache<K, V>() { + @Override + public V getNewInstance(K key) { + return creator.apply(key); + } + }; + } +} diff --git a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/utils/StringUtils.java b/flink-cdc-common/src/test/java/org/apache/flink/cdc/common/utils/StringUtilsTest.java similarity index 53% copy from flink-cdc-common/src/main/java/org/apache/flink/cdc/common/utils/StringUtils.java copy to flink-cdc-common/src/test/java/org/apache/flink/cdc/common/utils/StringUtilsTest.java index 7e539e48c..38c7c89d0 100644 --- a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/utils/StringUtils.java +++ b/flink-cdc-common/src/test/java/org/apache/flink/cdc/common/utils/StringUtilsTest.java @@ -17,26 +17,15 @@ package org.apache.flink.cdc.common.utils; -/** Utility class to convert objects into strings in vice-versa. */ -public class StringUtils { - /** - * Checks if the string is null, empty, or contains only whitespace characters. A whitespace - * character is defined via {@link Character#isWhitespace(char)}. - * - * @param str The string to check - * @return True, if the string is null or blank, false otherwise. - */ - public static boolean isNullOrWhitespaceOnly(String str) { - if (str == null || str.isEmpty()) { - return true; - } +import org.junit.Assert; +import org.junit.Test; - final int len = str.length(); - for (int i = 0; i < len; i++) { - if (!Character.isWhitespace(str.charAt(i))) { - return false; - } - } - return true; +/** A test for the {@link StringUtils}. */ +public class StringUtilsTest { + @Test + public void testConvertToCamelCase() { + String str = "AA_BB CC"; + String camelCaseStr = StringUtils.convertToCamelCase(str); + Assert.assertEquals("aaBbCc", camelCaseStr); } }
