[Feature-2932][Pipeline] Flink CDC pipeline supports transform #2937

aiwenmo · 2023-12-27T16:38:40Z

This is a draft pull request, and the functionality has not been completed yet.
Please discuss the details, implementation, and optimization under this PR.
I will complete the remaining code. thx

aiwenmo · 2023-12-28T01:27:50Z

What is the more appropriate way to obtain the field name and data type of dataChangeEvent. after()?

aiwenmo · 2023-12-28T10:18:16Z

Change the parameter definition of transform to the following:

transform:
  - source-table: mydb.app_order_.*
    projection: id, order_id, TO_UPPER(product_name)
    filter: id > 10 AND order_id > 100
    description: project fields from source table
  - source-table: mydb.web_order_.*
    projection: CONCAT(id, order_id) as uniq_id, *
    filter: uniq_id > 10
    description: add new uniq_id for each row

lvyanquan · 2024-01-01T02:31:08Z

...-cdc-runtime/src/main/java/com/ververica/cdc/runtime/operators/transform/FilterFunction.java

+        TableId tableId = dataChangeEvent.tableId();
+        BinaryRecordData after = (BinaryRecordData) dataChangeEvent.after();
+        if(after == null){
+            return false;


If we filter out all the deletion event, then the data of target table and source table will be inconsistent.

thx. I will fix it.

I have fixed it.

lvyanquan

Thanks for this great contribution. Left some comments, we need to clearly declare support functions and data types.
What‘s more, we also need to consider the situation of restarting from a failure.

lvyanquan · 2024-01-01T13:30:26Z

flink-cdc-composer/src/main/java/com/ververica/cdc/composer/definition/TransformDef.java

@@ -16,9 +16,78 @@

 package com.ververica.cdc.composer.definition;

+import java.util.Objects;
+import java.util.Optional;
+
 /**
 * Definition of transformation.
 *
 * <p>Transformation will be implemented later, therefore we left the class blank.


this comment can be removed.

lvyanquan · 2024-01-01T14:54:23Z

...-runtime/src/main/java/com/ververica/cdc/runtime/operators/transform/ProjectionFunction.java

+                new ArrayList<>(
+                        Arrays.asList(
+                                new Column[createTableEvent.getSchema().getColumns().size()]));
+        Collections.copy(sourceColumn, createTableEvent.getSchema().getColumns());


Can we simply use createTableEvent.getSchema().getColumns() to build sourceColumn?

lvyanquan · 2024-01-01T14:56:26Z

...-runtime/src/main/java/com/ververica/cdc/runtime/operators/transform/ProjectionFunction.java

+                Projector projector = route.f1;
+                BinaryRecordData data =
+                        projector.generateRecordData(after, sourceColumnMap.get(tableId));
+                return DataChangeEvent.setAfter(dataChangeEvent, data);


We should modify before and after record here, because sink like Kafka, pulsar need to use both to build debezium json format.

lvyanquan · 2024-01-01T14:57:45Z

...-runtime/src/main/java/com/ververica/cdc/runtime/operators/transform/ProjectionFunction.java

+        // skip delete event
+        if (after == null) {
+            return event;
+        }


Delete event should be processed to be consistent with source table.

lvyanquan · 2024-01-01T15:04:18Z

...-cdc-runtime/src/main/java/com/ververica/cdc/runtime/operators/transform/FilterFunction.java

+        private final List<Tuple3<String, String, String>> filterRules = new ArrayList<>();
+
+        public Builder addFilter(String tableInclusions, String projection, String filter) {
+            filterRules.add(Tuple3.of(tableInclusions, projection, filter));


Looks like projection is useless in FilterFunction, can we remove it?

lvyanquan · 2024-01-01T15:07:10Z

...-cdc-runtime/src/main/java/com/ververica/cdc/runtime/operators/transform/FilterFunction.java

+        return true;
+    }
+
+    private void transformCreateTableEvent(CreateTableEvent createTableEvent) {


CreateTableEvent is not transformed here, We should use a more appropriate method name.

lvyanquan · 2024-01-02T01:29:01Z

...time/src/test/java/com/ververica/cdc/runtime/operators/transform/ProjectionFunctionTest.java

+                new CreateTableEvent(CUSTOMERS_TABLEID, CUSTOMERS_SCHEMA);
+        ProjectionFunction transform =
+                ProjectionFunction.newBuilder()
+                        .addProjection(CUSTOMERS_TABLEID.identifier(), "*, col1 + col2 col12")


A doc or more tests is necessary to show what kind of Projection we support after this pr. For example, can we support CAST Function now?

lvyanquan · 2024-01-02T01:39:12Z

flink-cdc-runtime/src/main/java/com/ververica/cdc/runtime/operators/transform/Projector.java

+        for (int i = 0; i < columns.size(); i++) {
+            originalValueMap.put(
+                    columns.get(i).getName(),
+                    fromDataType(after.getString(i), columns.get(i).getType()));


There are more data types in BinaryRecordData, so we may need to keep a list of FieldGetter to get the correct objects, like createFieldGetter.

lvyanquan · 2024-01-02T02:05:52Z

flink-cdc-runtime/src/main/java/com/ververica/cdc/runtime/operators/transform/Projector.java

+                return DecimalData.fromBigDecimal(
+                        bigDecimalValue, bigDecimalValue.precision(), bigDecimalValue.scale());
+            default:
+                return BinaryStringData.fromString(value.toString());


We need to support more types, such as some time types and float types.

aiwenmo · 2024-01-02T02:36:13Z

@lvyanquan Thanks for your review. I will fix these issues tonight.

aiwenmo · 2024-01-03T01:03:27Z

There are only two bugs that have not been fixed yet. I will fix them again today.
1.A doc or more tests is necessary to show what kind of Projection we support after this pr. For example, can we support CAST Function now?
2.We need to support more types, such as some time types and float types.

PatrickRen

@aiwenmo Thanks for the great work! I looked through the PR and left some comments. I'm not quite familiar with Calcite so please correct me if I'm wrong

PatrickRen · 2024-01-05T09:47:57Z

flink-cdc-common/src/main/java/com/ververica/cdc/common/utils/SchemaUtils.java

+     * 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) {


Looks like the logic here is the same as createFieldGetters(Schema). What about letting createFieldGetters(Schema) reuse this one:

public static List<RecordData.FieldGetter> createFieldGetters(Schema schema) { return createFieldGetters(schema.getColumns()); }

thx. I will optimize it.

PatrickRen · 2024-01-05T10:11:36Z

...cdc-runtime/src/main/java/com/ververica/cdc/runtime/operators/transform/ColumnTransform.java

+    private final DataType dataType;
+    private final SqlBasicCall transform;
+    private final JexlExpression expression;
+    private static final Engine jexlEngine = new Engine();


I'm a rookie of Calcite and JEXL so please correct me if I'm wrong. It looks like projection and filter expressions are interpreted as SQL syntax first and parsed by Calcite in Projector, then evaluated by JEXL syntax during runtime. Are these two syntaxes compatible to each other?

The original intention of using Calcite is to provide parsing capability for Flink function calls. JEXL is a temporary solution that can handle most scenarios, and I will add some unit tests to validate it. Of course, if you have any other good suggestions, we can discuss them together.

PatrickRen · 2024-01-05T10:18:56Z

...-cdc-runtime/src/main/java/com/ververica/cdc/runtime/operators/transform/FilterFunction.java

+        return true;
+    }
+
+    private void catchSchema(CreateTableEvent createTableEvent) {


Do you mean cache?

PatrickRen · 2024-01-05T10:35:27Z

...-composer/src/main/java/com/ververica/cdc/composer/flink/translator/TransformTranslator.java

+        DataStream<Event> projectionOutput =
+                input.map(projectionFunctionBuilder.build(), new EventTypeInfo())
+                        .name("Transform:Projection");
+        return projectionOutput.filter(filterFunctionBuilder.build()).name("Transform:Filter");


Ideally the order of filter and projection can be adjusted dynamically. If there's no filter on computed columns, filter can be moved before projection. It's totally acceptable to me to hard-code the order for now and we can optimize it in the future.

We can design new parameters to determine the execution order of projection and filter。

PatrickRen · 2024-01-05T10:37:12Z

flink-cdc-common/src/main/java/com/ververica/cdc/common/event/DataChangeEvent.java

+     * Updates the before of a {@link DataChangeEvent} instance that describes the event with meta
+     * info.
+     */
+    public static DataChangeEvent setBefore(DataChangeEvent dataChangeEvent, RecordData before) {


What about resetBefore/After?

It's a good idea.

PatrickRen · 2024-01-05T10:38:09Z

...-values/src/main/java/com/ververica/cdc/connectors/values/source/ValuesDataSourceHelper.java

+        // rename column
+        /*Map<String, String> nameMapping = new HashMap<>();
+        nameMapping.put("col2", "newCol2");
+        nameMapping.put("col3", "newCol3");
+        RenameColumnEvent renameColumnEvent = new RenameColumnEvent(TABLE_1, nameMapping);
+        split1.add(renameColumnEvent);*/
+
+        // drop column
+        /*DropColumnEvent dropColumnEvent =
+            new DropColumnEvent(
+                TABLE_1,
+                Collections.singletonList(
+                    Column.physicalColumn("newCol2", DataTypes.STRING())));
+        split1.add(dropColumnEvent);*/
+
+        // delete
+        /*split1.add(
+        DataChangeEvent.deleteEvent(
+            TABLE_1,
+            generator.generate(
+                new Object[] {
+                    BinaryStringData.fromString("1"),
+                    BinaryStringData.fromString("1")
+                })));*/
+
+        // update
+        /*split1.add(
+        DataChangeEvent.updateEvent(
+            TABLE_1,
+            generator.generate(
+                new Object[] {
+                    BinaryStringData.fromString("2"),
+                    BinaryStringData.fromString("")
+                }),
+            generator.generate(
+                new Object[] {
+                    BinaryStringData.fromString("2"),
+                    BinaryStringData.fromString("x")
+                })));*/


I want to add unit tests for various scenarios, so this code has not been deleted. Next, I will add unit tests over the weekend.

lvyanquan

Left some comments.
I have two points to discuss:
The first one is the compatibility between JEXL and calcite, One reference is that Flink is using Janino, would it be better.
The second one is how to determine the type of compute column, if we want to support operations of two types, like int value + long value.

lvyanquan · 2024-01-08T07:14:13Z

...-composer/src/main/java/com/ververica/cdc/composer/flink/translator/TransformTranslator.java

+            if (transform.isValidFilter()) {
+                containFilter = true;
+                filterFunctionBuilder.addFilter(
+                        transform.getSourceTable(), transform.getFilter().get());


It seems somewhat inconsistent, do we still need to set Filter to optional?

Yes. Filter is optional.

lvyanquan · 2024-01-08T08:54:53Z

flink-cdc-runtime/src/main/java/com/ververica/cdc/runtime/operators/transform/Projector.java

+        return schemaChangeEvent;
+    }
+
+    public BinaryRecordData generateRecordData(BinaryRecordData after, List<Column> columns) {


BinaryRecordData recordData will be better.

lvyanquan · 2024-01-08T09:16:13Z

...-cdc-runtime/src/main/java/com/ververica/cdc/runtime/operators/transform/FilterFunction.java

+            return true;
+        }
+        if (!(event instanceof DataChangeEvent)) {
+            return true;


DataChangeEvent of delete should also apply filter.

lvyanquan · 2024-01-08T09:18:49Z

flink-cdc-runtime/src/main/java/com/ververica/cdc/runtime/operators/transform/RowFilter.java

+        for (int i = 0; i < columns.size(); i++) {
+            jexlContext.set(
+                    columns.get(i).getName(),
+                    fromDataType(after.getString(i), columns.get(i).getType()));


We should keep a sourceColumnMap like what ProjectionFunction do to get the correct values.

lvyanquan · 2024-01-08T11:25:33Z

flink-cdc-runtime/src/main/java/com/ververica/cdc/runtime/parser/FlinkSqlParser.java

+                        }
+                    }
+                    columnTransformList.add(
+                            ColumnTransform.of(columnName, DataTypes.STRING(), transform));


Set all column to string type will cause inaccuracies. We need to add type inference, or simply extract one data type from relevant columns.

lvyanquan · 2024-01-24T15:32:19Z

flink-cdc-runtime/src/main/java/com/ververica/cdc/runtime/operators/transform/Projector.java

+        return recordDataGenerator;
+    }
+
+    public boolean isVaild() {


lvyanquan · 2024-01-24T15:33:27Z

flink-cdc-runtime/src/main/java/com/ververica/cdc/runtime/operators/transform/RowFilter.java

+        try {
+            return (Boolean) expressionEvaluator.evaluate(params.toArray());
+        } catch (InvocationTargetException e) {
+            e.printStackTrace();


lvyanquan · 2024-01-25T12:27:04Z

...ntime/src/main/java/com/ververica/cdc/runtime/operators/transform/TransformDataFunction.java

+import java.util.stream.Collectors;
+
+/** A data process function that applies user-defined transform logics. */
+public class TransformDataFunction extends AbstractStreamOperator<Event>


TransformDataOperator?

lvyanquan · 2024-01-26T06:14:36Z

...src/test/java/com/ververica/cdc/runtime/operators/transform/TransformSchemaFunctionTest.java

+                    .physicalColumn("col1", DataTypes.STRING())
+                    .physicalColumn("col2", DataTypes.STRING())
+                    .physicalColumn("col3", DataTypes.STRING())
+                    .physicalColumn("col12", DataTypes.STRING())


actual column orders should be like the following?

.physicalColumn("col1", DataTypes.STRING()) .physicalColumn("col2", DataTypes.STRING()) .physicalColumn("col12", DataTypes.STRING()) .physicalColumn("col3", DataTypes.STRING())

lvyanquan · 2024-01-26T06:38:54Z

...ime/src/main/java/com/ververica/cdc/runtime/operators/transform/TransformSchemaFunction.java

+        }
+    }
+
+    private TransformSchemaFunction(List<Tuple2<String, String>> transformRules) {


Just a discussion, should we chain this operator with Source?

lvyanquan · 2024-01-26T06:56:48Z

flink-cdc-runtime/src/main/java/com/ververica/cdc/runtime/parser/FlinkSqlParser.java

+        SqlNode validateSqlNode = validator.validate(sqlNode);
+        HepProgramBuilder builder = new HepProgramBuilder();
+        HepPlanner planner = new HepPlanner(builder.build());
+        RelOptCluster cluster = RelOptCluster.create(planner, new RexBuilder(factory));


Looks like some classes are unrelated with columns and sqlNode, can we reuse them?

lvyanquan · 2024-01-26T07:21:37Z

...ime/src/main/java/com/ververica/cdc/runtime/operators/transform/TransformSchemaFunction.java

+        } else if (event instanceof DataChangeEvent) {
+            Optional<DataChangeEvent> dataChangeEventOptional =
+                    applyDataChangeEvent(((DataChangeEvent) event));
+            if (dataChangeEventOptional.isPresent()) {


In which condition will dataChangeEvent be empty, as we don't do filter here?

lvyanquan · 2024-01-26T07:55:11Z

flink-cdc-runtime/src/main/java/com/ververica/cdc/runtime/operators/transform/RowFilter.java

+            if (columnNames.contains(columns.get(i).getName())) {
+                params.add(
+                        DataTypeConverter.convertToOriginal(
+                                after.getString(i), columns.get(i).getType()));


tableInfo.getFieldGetters()[i].getFieldOrNull(after)

lvyanquan · 2024-01-29T07:32:55Z

Hi @aiwenmo , I met Invalid signature file digest for Manifest main when testing this transform in E2e environment.
Can you help to fix it when building flink-cdc-dist jar?

lvyanquan · 2024-01-29T07:48:38Z

...-composer/src/main/java/com/ververica/cdc/composer/flink/translator/TransformTranslator.java

+                transformFunctionBuilder.addTransform(
+                        transform.getSourceTable(),
+                        transform.getProjection().get(),
+                        transform.getFilter().get());


Though transform.isValidProjection() is true, transform.getFilter().get() will still be possible to get NoPointException.

lvyanquan · 2024-01-29T09:35:23Z

...cdc-runtime/src/main/java/com/ververica/cdc/runtime/operators/transform/ColumnTransform.java

+            params.add(tableInfo.getTableName());
+        }
+
+        ExpressionEvaluator expressionEvaluator =


We should cache expressionEvaluator like what RowFilter do.

lvyanquan · 2024-01-29T09:55:41Z

flink-cdc-runtime/src/main/java/com/ververica/cdc/runtime/operators/transform/RowFilter.java

+    private final String expression;
+    private final String scriptExpression;
+    private final List<String> columnNames;
+    private ExpressionEvaluator expressionEvaluator;


Got A different class with the same name was previously loaded by 'app'. (org.codehaus.janino.ExpressionEvaluator is in unnamed module of loader 'app') error when submitting job to session.

leonardBang

Thanks @aiwenmo and @lvyanquan for the great work, I only left two comments

leonardBang · 2024-03-05T11:36:25Z

...me/src/main/java/com/ververica/cdc/runtime/operators/transform/TransformFilterProcessor.java

+                TransformExpressionCompiler.compileExpression(transformExpressionKey);
+        try {
+            return (Boolean)
+                    expressionEvaluator.evaluate(generateParams(after, tableInfo, transformFilter));


Could one TransformFilterProcessor instance holds a expressionEvaluator which built in construction? and thus we can offer a method like private boolean process( BinaryRecordData after). In this way, we can avoid compile expression per record.

leonardBang · 2024-03-05T11:40:48Z

...rc/main/java/com/ververica/cdc/runtime/operators/transform/TransformProjectionProcessor.java

+    }
+
+    public static BinaryRecordData processData(
+            BinaryRecordData after, TableInfo tableInfo, TransformProjection transformProjection) {


The static method let the class looks like a utils, could we instantiate a TransformProjectionProcessor transformProjectionProcessor = new TransformProjectionProcessor(xx, xx) and then call transformProjectionProcessor.process() ?

leonardBang

Thanks @aiwenmo for the update, I think this PR is ready to merge, I only left two minor comments

leonardBang · 2024-03-06T13:25:14Z

flink-cdc-common/src/main/java/com/ververica/cdc/common/utils/ThreadLocalCache.java

+
+    private static class BoundedMap<K, V> extends LinkedHashMap<K, V> {
+
+        private static final long serialVersionUID = -211630219014422361L;


leonardBang · 2024-03-06T13:26:04Z

flink-cdc-common/src/main/java/com/ververica/cdc/common/utils/DateTimeUtils.java

@@ -0,0 +1,120 @@
+/*
+ * Copyright 2023 Ververica Inc.


please rebase to fix the copyright.

… classes

…ator to support transformation

This closes apache#2937.

leonardBang

Thanks @aiwenmo for the great work, LGTM although I think we can improve the test coverage as well as the document, but I think we can improve them later.

aiwenmo · 2024-04-02T14:19:26Z

Thanks @aiwenmo for the great work, LGTM although I think we can improve the test coverage as well as the document, but I think we can improve them later.

Thanks @leonardBang , I will improve them in the new pull request.

This closes apache#2937.

github-actions bot added values-pipeline-connector composer common runtime labels Dec 27, 2023

aiwenmo marked this pull request as draft December 28, 2023 01:08

aiwenmo mentioned this pull request Dec 28, 2023

[Feature][Pipeline] Flink CDC pipeline supports transform #2932

Closed

2 tasks

github-actions bot added the cli label Dec 28, 2023

lvyanquan reviewed Jan 1, 2024

View reviewed changes

lvyanquan reviewed Jan 2, 2024

View reviewed changes

PatrickRen reviewed Jan 5, 2024

View reviewed changes

aiwenmo marked this pull request as ready for review January 6, 2024 13:24

lvyanquan reviewed Jan 8, 2024

View reviewed changes

lvyanquan reviewed Jan 26, 2024

View reviewed changes

lvyanquan reviewed Jan 29, 2024

View reviewed changes

github-actions bot added docs Improvements or additions to documentation mongodb-cdc-connector build mysql-cdc-connector base oracle-cdc-connector postgres-cdc-connector sqlserver-cdc-connector e2e-tests mysql-pipeline-connector labels Jan 29, 2024

leonardBang reviewed Mar 5, 2024

View reviewed changes

leonardBang reviewed Mar 6, 2024

View reviewed changes

github-actions bot added docs Improvements or additions to documentation mongodb-cdc-connector mysql-cdc-connector base oracle-cdc-connector postgres-cdc-connector sqlserver-cdc-connector labels Mar 6, 2024

aiwenmo force-pushed the Feature-2932 branch from 0815970 to b7c421d Compare March 7, 2024 16:07

github-actions bot added db2-cdc-connector and removed docs Improvements or additions to documentation mongodb-cdc-connector mysql-cdc-connector base oracle-cdc-connector postgres-cdc-connector sqlserver-cdc-connector labels Mar 7, 2024

lvyanquan mentioned this pull request Mar 19, 2024

Flink CDC 3.1.0 Plan #2861

Closed

22 tasks

aiwenmo force-pushed the Feature-2932 branch from 277cdd7 to a05aa65 Compare March 20, 2024 12:45

aiwenmo and others added 5 commits April 2, 2024 19:43

[cdc-composer] Introduce transform definition and parser

05f238c

[cdc-common] Introduce partitionKeys into schema and add related util…

8dc407b

… classes

[cdc-runtime] Introduce TransformSchemaOperator and TransformDataOper…

bb1a1e1

…ator to support transformation

[build] Optimize pom to solve the CI error

38f172d

[minor][cdc-common] Improve the java doc of translators

094500e

This closes apache#2937.

leonardBang force-pushed the Feature-2932 branch from 472d430 to 094500e Compare April 2, 2024 11:59

leonardBang approved these changes Apr 2, 2024

View reviewed changes

leonardBang merged commit 1d05abf into apache:master Apr 2, 2024
13 checks passed

wuzhenhua01 pushed a commit to wuzhenhua01/flink-cdc-connectors that referenced this pull request Aug 4, 2024

[minor][cdc-common] Improve the java doc of translators

7e38135

This closes apache#2937.


		private static class BoundedMap<K, V> extends LinkedHashMap<K, V> {

		private static final long serialVersionUID = -211630219014422361L;

[Feature-2932][Pipeline] Flink CDC pipeline supports transform #2937

[Feature-2932][Pipeline] Flink CDC pipeline supports transform #2937

Conversation

aiwenmo commented Dec 27, 2023

aiwenmo commented Dec 28, 2023

aiwenmo commented Dec 28, 2023

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

lvyanquan left a comment • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

lvyanquan Jan 1, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

lvyanquan Jan 2, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

aiwenmo commented Jan 2, 2024

aiwenmo commented Jan 3, 2024

PatrickRen left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

lvyanquan left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

lvyanquan commented Jan 29, 2024

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

leonardBang left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

leonardBang left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

leonardBang left a comment

Choose a reason for hiding this comment

aiwenmo commented Apr 2, 2024

lvyanquan left a comment •

edited

Loading

lvyanquan Jan 1, 2024 •

edited

Loading

lvyanquan Jan 2, 2024 •

edited

Loading