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/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 64f87629e [test] add more test case for schema evolution with schema
included in records (#4284)
64f87629e is described below
commit 64f87629e265b370b6e6088a831fbbbbbe37b030
Author: Fantasy-Jay <[email protected]>
AuthorDate: Tue Oct 8 16:35:20 2024 +0800
[test] add more test case for schema evolution with schema included in
records (#4284)
---
.../kafka/KafkaDebeziumSyncTableActionITCase.java | 8 +-
.../cdc/kafka/KafkaSyncTableActionITCase.java | 121 ++++++++++++++++++++-
.../schema/schemaevolution/debezium-data-1.txt | 3 +-
.../schema/schemaevolution/debezium-data-2.txt | 3 +-
.../schema/schemaevolution/debezium-data-3.txt | 5 +-
.../schema/schemaevolution/debezium-data-4.txt | 2 +-
.../schema/schemaevolution/debezium-data-5.txt | 19 ++++
...um-data-4.txt => debezium-data-with-delete.txt} | 0
8 files changed, 150 insertions(+), 11 deletions(-)
diff --git
a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/kafka/KafkaDebeziumSyncTableActionITCase.java
b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/kafka/KafkaDebeziumSyncTableActionITCase.java
index 04dfb3769..430598753 100644
---
a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/kafka/KafkaDebeziumSyncTableActionITCase.java
+++
b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/kafka/KafkaDebeziumSyncTableActionITCase.java
@@ -106,10 +106,16 @@ public class KafkaDebeziumSyncTableActionITCase extends
KafkaSyncTableActionITCa
@Test
@Timeout(60)
- public void testSchemaIncludeRecord1() throws Exception {
+ public void testSchemaIncludeRecord() throws Exception {
testSchemaIncludeRecord(DEBEZIUM);
}
+ @Test
+ @Timeout(60)
+ public void testSchemaEvolutionWithSchemaIncludeRecord() throws Exception {
+
runSingleTableSchemaEvolutionWithSchemaIncludeRecord("schemaevolution",
DEBEZIUM);
+ }
+
@Test
@Timeout(60)
public void testAllTypesWithSchema() throws Exception {
diff --git
a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/kafka/KafkaSyncTableActionITCase.java
b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/kafka/KafkaSyncTableActionITCase.java
index 117994835..f5b6bb592 100644
---
a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/kafka/KafkaSyncTableActionITCase.java
+++
b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/kafka/KafkaSyncTableActionITCase.java
@@ -28,7 +28,9 @@ import org.apache.paimon.types.DataType;
import org.apache.paimon.types.DataTypes;
import org.apache.paimon.types.RowType;
+import org.apache.flink.api.common.JobStatus;
import org.apache.flink.configuration.Configuration;
+import org.apache.flink.core.execution.JobClient;
import java.util.ArrayList;
import java.util.Arrays;
@@ -136,6 +138,120 @@ public class KafkaSyncTableActionITCase extends
KafkaActionITCaseBase {
waitForResult(expected, table, rowType, primaryKeys);
}
+ protected void runSingleTableSchemaEvolutionWithSchemaIncludeRecord(
+ String sourceDir, String format) throws Exception {
+ final String topic = "schema_evolution";
+ createTestTopic(topic, 1, 1);
+ writeRecordsToKafka(
+ topic, "kafka/%s/table/schema/%s/%s-data-1.txt", format,
sourceDir, format);
+
+ Map<String, String> kafkaConfig = getBasicKafkaConfig();
+ kafkaConfig.put(VALUE_FORMAT.key(), format + "-json");
+ kafkaConfig.put(TOPIC.key(), topic);
+ KafkaSyncTableAction action =
+ syncTableActionBuilder(kafkaConfig)
+ .withPrimaryKeys("id")
+ .withTableConfig(getBasicTableConfig())
+ .build();
+ JobClient jobClient = runActionWithDefaultEnv(action);
+
+ FileStoreTable table = getFileStoreTable(tableName);
+
+ RowType rowType =
+ RowType.of(
+ new DataType[] {
+ DataTypes.INT().notNull(),
+ DataTypes.STRING(),
+ DataTypes.STRING(),
+ DataTypes.DOUBLE()
+ },
+ new String[] {"id", "name", "description", "weight"});
+ List<String> primaryKeys = Collections.singletonList("id");
+ List<String> expected =
+ Collections.singletonList("+I[101, scooter, Small 2-wheel
scooter, 3.14]");
+ waitForResult(expected, table, rowType, primaryKeys);
+
+ // add column
+ writeRecordsToKafka(
+ topic, "kafka/%s/table/schema/%s/%s-data-2.txt", format,
sourceDir, format);
+
+ rowType =
+ RowType.of(
+ new DataType[] {
+ DataTypes.INT().notNull(),
+ DataTypes.STRING(),
+ DataTypes.STRING(),
+ DataTypes.DOUBLE(),
+ DataTypes.INT()
+ },
+ new String[] {"id", "name", "description", "weight",
"age"});
+ expected =
+ Arrays.asList(
+ "+I[101, scooter, Small 2-wheel scooter, 3.14, NULL]",
+ "+I[103, 12-pack drill bits, 12-pack of drill bits
with sizes ranging from #40 to #3, 0.8, 18]");
+ waitForResult(expected, table, rowType, primaryKeys);
+
+ // column type promotion (int32 -> int64)
+ writeRecordsToKafka(
+ topic, "kafka/%s/table/schema/%s/%s-data-3.txt", format,
sourceDir, format);
+
+ rowType =
+ RowType.of(
+ new DataType[] {
+ DataTypes.INT().notNull(),
+ DataTypes.STRING(),
+ DataTypes.STRING(),
+ DataTypes.DOUBLE(),
+ DataTypes.BIGINT()
+ },
+ new String[] {"id", "name", "description", "weight",
"age"});
+ expected =
+ Arrays.asList(
+ "+I[101, scooter, Small 2-wheel scooter, 3.14, NULL]",
+ "+I[103, 12-pack drill bits, 12-pack of drill bits
with sizes ranging from #40 to #3, 0.8, 18]",
+ "+I[104, hammer, 12oz carpenter's hammer, 0.75, 24]");
+ waitForResult(expected, table, rowType, primaryKeys);
+
+ // column type changed ignore (int64 -> int32)
+ writeRecordsToKafka(
+ topic, "kafka/%s/table/schema/%s/%s-data-4.txt", format,
sourceDir, format);
+
+ rowType =
+ RowType.of(
+ new DataType[] {
+ DataTypes.INT().notNull(),
+ DataTypes.STRING(),
+ DataTypes.STRING(),
+ DataTypes.DOUBLE(),
+ DataTypes.BIGINT()
+ },
+ new String[] {"id", "name", "description", "weight",
"age"});
+ expected =
+ Arrays.asList(
+ "+I[101, scooter, Small 2-wheel scooter, 3.14, NULL]",
+ "+I[103, 12-pack drill bits, 12-pack of drill bits
with sizes ranging from #40 to #3, 0.8, 18]",
+ "+I[104, hammer, 12oz carpenter's hammer, 0.75, 24]",
+ "+I[105, hammer, 14oz carpenter's hammer, 0.875, 24]");
+ waitForResult(expected, table, rowType, primaryKeys);
+
+ // column type covert exception (int64 -> string)
+ writeRecordsToKafka(
+ topic, "kafka/%s/table/schema/%s/%s-data-5.txt", format,
sourceDir, format);
+
+ while (true) {
+ JobStatus status = jobClient.getJobStatus().get();
+ if (status != JobStatus.RUNNING) {
+ assertThatThrownBy(() ->
jobClient.getJobExecutionResult().get())
+ .satisfies(
+ anyCauseMatches(
+ UnsupportedOperationException.class,
+ "Cannot convert field age from type
BIGINT to STRING of Paimon table"));
+ break;
+ }
+ Thread.sleep(1000);
+ }
+ }
+
public void testNotSupportFormat(String format) throws Exception {
final String topic = "not_support";
createTestTopic(topic, 1, 1);
@@ -478,7 +594,10 @@ public class KafkaSyncTableActionITCase extends
KafkaActionITCaseBase {
final String topic = "test_kafka_schema";
createTestTopic(topic, 1, 1);
writeRecordsToKafka(
- topic, "kafka/%s/table/schema/schemaevolution/%s-data-4.txt",
format, format);
+ topic,
+
"kafka/%s/table/schema/schemaevolution/%s-data-with-delete.txt",
+ format,
+ format);
Configuration kafkaConfig =
Configuration.fromMap(getBasicKafkaConfig());
kafkaConfig.setString(VALUE_FORMAT.key(), format + "-json");
diff --git
a/paimon-flink/paimon-flink-cdc/src/test/resources/kafka/debezium/table/schema/schemaevolution/debezium-data-1.txt
b/paimon-flink/paimon-flink-cdc/src/test/resources/kafka/debezium/table/schema/schemaevolution/debezium-data-1.txt
index 3b34b6f8f..7dd068d00 100644
---
a/paimon-flink/paimon-flink-cdc/src/test/resources/kafka/debezium/table/schema/schemaevolution/debezium-data-1.txt
+++
b/paimon-flink/paimon-flink-cdc/src/test/resources/kafka/debezium/table/schema/schemaevolution/debezium-data-1.txt
@@ -16,5 +16,4 @@
* limitations under the License.
*/
-{"schema":null, "payload":{"before": null, "after": {"id": 101, "name":
"scooter", "description": "Small 2-wheel scooter", "weight": 3.14}, "source":
{"version": "1.9.7.Final", "connector": "mysql", "name": "mysql_binlog_source",
"ts_ms": 1596684883000, "snapshot": "false", "db": "test", "sequence": null,
"table": "product", "server_id": 0, "gtid": null, "file": "", "pos": 0, "row":
0, "thread": null, "query": null}, "op": "c", "ts_ms": 1596684883000,
"transaction": null}}
-{"schema":null, "payload":{"before": null, "after": {"id": 102, "name": "car
battery", "description": "12V car battery", "weight": 8.1}, "source":
{"version": "1.9.7.Final", "connector": "mysql", "name": "mysql_binlog_source",
"ts_ms": 1596684883000, "snapshot": "false", "db": "test", "sequence": null,
"table": "product", "server_id": 0, "gtid": null, "file": "", "pos": 0, "row":
0, "thread": null, "query": null}, "op": "c", "ts_ms": 1596684883000,
"transaction": null}}
+{"schema":{"type":"struct","fields":[{"type":"struct","fields":[{"type":"int32","optional":false,"field":"id"},{"type":"string","optional":false,"field":"name"},{"type":"string","optional":true,"field":"description"},{"type":"double","optional":true,"field":"weight"}],"optional":true,"name":"dbserver1.inventory.products.Value","field":"after"},{"type":"struct","fields":[{"type":"string","optional":false,"field":"version"},{"type":"string","optional":false,"field":"connector"},{"type":"st
[...]
\ No newline at end of file
diff --git
a/paimon-flink/paimon-flink-cdc/src/test/resources/kafka/debezium/table/schema/schemaevolution/debezium-data-2.txt
b/paimon-flink/paimon-flink-cdc/src/test/resources/kafka/debezium/table/schema/schemaevolution/debezium-data-2.txt
index 3b0571d72..0eec06df3 100644
---
a/paimon-flink/paimon-flink-cdc/src/test/resources/kafka/debezium/table/schema/schemaevolution/debezium-data-2.txt
+++
b/paimon-flink/paimon-flink-cdc/src/test/resources/kafka/debezium/table/schema/schemaevolution/debezium-data-2.txt
@@ -16,5 +16,4 @@
* limitations under the License.
*/
-{"schema":null, "payload":{"before": null, "after": {"id": 103, "name":
"12-pack drill bits", "description": "12-pack of drill bits with sizes ranging
from #40 to #3", "weight": 0.8, "age": 18}, "source": {"version":
"1.9.7.Final", "connector": "mysql", "name": "mysql_binlog_source", "ts_ms":
1596684883000, "snapshot": "false", "db": "test", "sequence": null, "table":
"product", "server_id": 0, "gtid": null, "file": "", "pos": 0, "row": 0,
"thread": null, "query": null}, "op": "c", "ts_m [...]
-{"schema":null, "payload":{"before": null, "after": {"id": 104, "name":
"hammer", "description": "12oz carpenter's hammer", "weight": 0.75, "age": 24},
"source": {"version": "1.9.7.Final", "connector": "mysql", "name":
"mysql_binlog_source", "ts_ms": 1596684883000, "snapshot": "false", "db":
"test", "sequence": null, "table": "product", "server_id": 0, "gtid": null,
"file": "", "pos": 0, "row": 0, "thread": null, "query": null}, "op": "c",
"ts_ms": 1596684883000, "transaction": null}}
+{"schema":{"type":"struct","fields":[{"type":"struct","fields":[{"type":"int32","optional":false,"field":"id"},{"type":"string","optional":false,"field":"name"},{"type":"string","optional":true,"field":"description"},{"type":"double","optional":true,"field":"weight"},{"type":"int32","optional":true,"field":"age"}],"optional":true,"name":"dbserver1.inventory.products.Value","field":"after"},{"type":"struct","fields":[{"type":"string","optional":false,"field":"version"},{"type":"string","o
[...]
diff --git
a/paimon-flink/paimon-flink-cdc/src/test/resources/kafka/debezium/table/schema/schemaevolution/debezium-data-3.txt
b/paimon-flink/paimon-flink-cdc/src/test/resources/kafka/debezium/table/schema/schemaevolution/debezium-data-3.txt
index ea99eb71f..754015d7d 100644
---
a/paimon-flink/paimon-flink-cdc/src/test/resources/kafka/debezium/table/schema/schemaevolution/debezium-data-3.txt
+++
b/paimon-flink/paimon-flink-cdc/src/test/resources/kafka/debezium/table/schema/schemaevolution/debezium-data-3.txt
@@ -16,7 +16,4 @@
* limitations under the License.
*/
-{"schema":null, "payload":{"before": null, "after": {"id": 105, "name":
"hammer", "description": "14oz carpenter's hammer", "weight": 0.875, "address":
"Shanghai"}, "source": {"version": "1.9.7.Final", "connector": "mysql", "name":
"mysql_binlog_source", "ts_ms": 1596684883000, "snapshot": "false", "db":
"test", "sequence": null, "table": "product", "server_id": 0, "gtid": null,
"file": "", "pos": 0, "row": 0, "thread": null, "query": null}, "op": "c",
"ts_ms": 1596684883000, "transactio [...]
-{"schema":null, "payload":{"before": {"id": 101, "name": "scooter",
"description": "Small 2-wheel scooter", "weight": 3.14}, "after": null,
"source": {"version": "1.9.7.Final", "connector": "mysql", "name":
"mysql_binlog_source", "ts_ms": 1596684883000, "snapshot": "false", "db":
"test", "sequence": null, "table": "product", "server_id": 0, "gtid": null,
"file": "", "pos": 0, "row": 0, "thread": null, "query": null}, "op": "d",
"ts_ms": 1596684883000, "transaction": null}}
-{"schema":null, "payload":{"before": {"address": "Shanghai"}, "after": {"id":
105, "name": "hammer", "description": "14oz carpenter's hammer", "weight":
0.875, "address": "Beijing"}, "source": {"version": "1.9.7.Final", "connector":
"mysql", "name": "mysql_binlog_source", "ts_ms": 1596684906000, "snapshot":
"false", "db": "test", "sequence": null, "table": "product", "server_id": 0,
"gtid": null, "file": "", "pos": 0, "row": 0, "thread": null, "query": null},
"op": "u", "ts_ms": 15966849 [...]
-{"schema":null, "payload":{"before": null, "after": {"id": 107, "name":
"rocks", "description": "box of assorted rocks", "weight": 5.3}, "source":
{"version": "1.9.7.Final", "connector": "mysql", "name": "mysql_binlog_source",
"ts_ms": 1596684883000, "snapshot": "false", "db": "test", "sequence": null,
"table": "product", "server_id": 0, "gtid": null, "file": "", "pos": 0, "row":
0, "thread": null, "query": null}, "op": "c", "ts_ms": 1596684883000,
"transaction": null}}
+{"schema":{"type":"struct","fields":[{"type":"struct","fields":[{"type":"int32","optional":false,"field":"id"},{"type":"string","optional":false,"field":"name"},{"type":"string","optional":true,"field":"description"},{"type":"double","optional":true,"field":"weight"},{"type":"int64","optional":true,"field":"age"}],"optional":true,"name":"dbserver1.inventory.products.Value","field":"after"},{"type":"struct","fields":[{"type":"string","optional":false,"field":"version"},{"type":"string","o
[...]
\ No newline at end of file
diff --git
a/paimon-flink/paimon-flink-cdc/src/test/resources/kafka/debezium/table/schema/schemaevolution/debezium-data-4.txt
b/paimon-flink/paimon-flink-cdc/src/test/resources/kafka/debezium/table/schema/schemaevolution/debezium-data-4.txt
index 911c9ce0a..154f41dd9 100644
---
a/paimon-flink/paimon-flink-cdc/src/test/resources/kafka/debezium/table/schema/schemaevolution/debezium-data-4.txt
+++
b/paimon-flink/paimon-flink-cdc/src/test/resources/kafka/debezium/table/schema/schemaevolution/debezium-data-4.txt
@@ -16,4 +16,4 @@
* limitations under the License.
*/
-{"schema":null, "payload":{"before": {"id": 101, "name": "scooter",
"description": "Small 2-wheel scooter", "weight": 3.14}, "after": null,
"source": {"version": "1.9.7.Final", "connector": "mysql", "name":
"mysql_binlog_source", "ts_ms": 1596684883000, "snapshot": "false", "db":
"test", "sequence": null, "table": "product", "server_id": 0, "gtid": null,
"file": "", "pos": 0, "row": 0, "thread": null, "query": null}, "op": "d",
"ts_ms": 1596684883000, "transaction": null}}
\ No newline at end of file
+{"schema":{"type":"struct","fields":[{"type":"struct","fields":[{"type":"int32","optional":false,"field":"id"},{"type":"string","optional":false,"field":"name"},{"type":"string","optional":true,"field":"description"},{"type":"double","optional":true,"field":"weight"},{"type":"int32","optional":true,"field":"age"}],"optional":true,"name":"dbserver1.inventory.products.Value","field":"after"},{"type":"struct","fields":[{"type":"string","optional":false,"field":"version"},{"type":"string","o
[...]
\ No newline at end of file
diff --git
a/paimon-flink/paimon-flink-cdc/src/test/resources/kafka/debezium/table/schema/schemaevolution/debezium-data-5.txt
b/paimon-flink/paimon-flink-cdc/src/test/resources/kafka/debezium/table/schema/schemaevolution/debezium-data-5.txt
new file mode 100644
index 000000000..fde64c550
--- /dev/null
+++
b/paimon-flink/paimon-flink-cdc/src/test/resources/kafka/debezium/table/schema/schemaevolution/debezium-data-5.txt
@@ -0,0 +1,19 @@
+/*
+ * 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.
+ */
+
+{"schema":{"type":"struct","fields":[{"type":"struct","fields":[{"type":"int32","optional":false,"field":"id"},{"type":"string","optional":false,"field":"name"},{"type":"string","optional":true,"field":"description"},{"type":"double","optional":true,"field":"weight"},{"type":"string","optional":true,"field":"age"}],"optional":true,"name":"dbserver1.inventory.products.Value","field":"after"},{"type":"struct","fields":[{"type":"string","optional":false,"field":"version"},{"type":"string","
[...]
\ No newline at end of file
diff --git
a/paimon-flink/paimon-flink-cdc/src/test/resources/kafka/debezium/table/schema/schemaevolution/debezium-data-4.txt
b/paimon-flink/paimon-flink-cdc/src/test/resources/kafka/debezium/table/schema/schemaevolution/debezium-data-with-delete.txt
similarity index 100%
copy from
paimon-flink/paimon-flink-cdc/src/test/resources/kafka/debezium/table/schema/schemaevolution/debezium-data-4.txt
copy to
paimon-flink/paimon-flink-cdc/src/test/resources/kafka/debezium/table/schema/schemaevolution/debezium-data-with-delete.txt
