You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@paimon.apache.org by "MonsterChenzhuo (via GitHub)" <gi...@apache.org> on 2023/08/15 16:23:23 UTC
[GitHub] [incubator-paimon] MonsterChenzhuo opened a new pull request, #1827: [flink]Optimize kafka format
MonsterChenzhuo opened a new pull request, #1827:
URL: https://github.com/apache/incubator-paimon/pull/1827
<!-- Please specify the module before the PR name: [core] ... or [flink] ... -->
### Purpose
subtask issue: #1826
<!-- What is the purpose of the change -->
### Tests
<!-- List UT and IT cases to verify this change -->
### API and Format
<!-- Does this change affect API or storage format -->
### Documentation
<!-- Does this change introduce a new feature -->
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-paimon] MonsterChenzhuo commented on a diff in pull request #1827: [flink]Supported ogg format and optimize format scalability
Posted by "MonsterChenzhuo (via GitHub)" <gi...@apache.org>.
MonsterChenzhuo commented on code in PR #1827:
URL: https://github.com/apache/incubator-paimon/pull/1827#discussion_r1301266248
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/cdc/kafka/KafkaSchema.java:
##########
@@ -117,16 +120,13 @@ public static KafkaSchema getKafkaSchema(Configuration kafkaConfig, String topic
ConsumerRecords<String, String> records =
consumer.poll(Duration.ofMillis(POLL_TIMEOUT_MILLIS));
for (ConsumerRecord<String, String> record : records) {
- String format = kafkaConfig.get(KafkaConnectorOptions.VALUE_FORMAT);
- if ("canal-json".equals(format)) {
- CanalRecordParser parser = new CanalRecordParser(true, Collections.emptyList());
- KafkaSchema kafkaSchema = parser.getKafkaSchema(record.value());
- if (kafkaSchema != null) {
- return kafkaSchema;
- }
- } else {
- throw new UnsupportedOperationException(
- "This format: " + format + " is not support.");
+ DataFormat format = getDataFormat(kafkaConfig);
+ RecordParser recordParser =
+ format.createParser(
+ true, new TableNameConverter(true), Collections.emptyList());
Review Comment:
done.
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-paimon] yuzelin commented on a diff in pull request #1827: [flink]Supported ogg format and optimize format scalability
Posted by "yuzelin (via GitHub)" <gi...@apache.org>.
yuzelin commented on code in PR #1827:
URL: https://github.com/apache/incubator-paimon/pull/1827#discussion_r1299903152
##########
paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/cdc/kafka/KafkaActionITCaseBase.java:
##########
@@ -369,4 +372,19 @@ protected FileStoreTable getFileStoreTable(String tableName) throws Exception {
Identifier identifier = Identifier.create(database, tableName);
return (FileStoreTable) catalog().getTable(identifier);
}
+
+ protected void waitTablesCreated(String... tables) throws Exception {
+ CommonTestUtils.waitUtil(
+ () -> {
+ try {
+ List<String> existed = catalog().listTables(database);
+ return existed.containsAll(Arrays.asList(tables));
+ } catch (Catalog.DatabaseNotExistException e) {
+ throw new RuntimeException(e);
+ }
+ },
+ Duration.ofSeconds(50),
+ Duration.ofMillis(100),
+ "Failed to wait tables to be created in 5 seconds.");
Review Comment:
Error message doesn't match the code.
##########
docs/content/how-to/cdc-ingestion.md:
##########
@@ -294,11 +294,15 @@ If a message in a Kafka topic is a change event captured from another database u
</tr>
<tr>
<td><a href="https://nightlies.apache.org/flink/flink-docs-release-1.16/zh/docs/connectors/table/formats/ogg/">OGG CDC</a></td>
- <td>False</td>
+ <td>True</td>
</tr>
</tbody>
</table>
+**Note that**
+
+In Oracle GoldenGate, the data format synchronized to Kafka does not include field data type information. As a result, Paimon sets the data type for all fields to "String" by default.
+
Review Comment:
Use
```
{{< hint info >}}
information here
{{< /hint >}}
```
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/cdc/kafka/KafkaSchema.java:
##########
@@ -117,16 +120,13 @@ public static KafkaSchema getKafkaSchema(Configuration kafkaConfig, String topic
ConsumerRecords<String, String> records =
consumer.poll(Duration.ofMillis(POLL_TIMEOUT_MILLIS));
for (ConsumerRecord<String, String> record : records) {
- String format = kafkaConfig.get(KafkaConnectorOptions.VALUE_FORMAT);
- if ("canal-json".equals(format)) {
- CanalRecordParser parser = new CanalRecordParser(true, Collections.emptyList());
- KafkaSchema kafkaSchema = parser.getKafkaSchema(record.value());
- if (kafkaSchema != null) {
- return kafkaSchema;
- }
- } else {
- throw new UnsupportedOperationException(
- "This format: " + format + " is not support.");
+ DataFormat format = getDataFormat(kafkaConfig);
+ RecordParser recordParser =
+ format.createParser(
+ true, new TableNameConverter(true), Collections.emptyList());
Review Comment:
This part can be moved before `while (true)`
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/cdc/kafka/formats/ogg/OggRecordParser.java:
##########
@@ -0,0 +1,218 @@
+/*
+ * 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.paimon.flink.action.cdc.kafka.formats.ogg;
+
+import org.apache.paimon.flink.action.cdc.ComputedColumn;
+import org.apache.paimon.flink.action.cdc.TableNameConverter;
+import org.apache.paimon.flink.action.cdc.kafka.KafkaSchema;
+import org.apache.paimon.flink.action.cdc.kafka.formats.RecordParser;
+import org.apache.paimon.flink.sink.cdc.CdcRecord;
+import org.apache.paimon.flink.sink.cdc.RichCdcMultiplexRecord;
+import org.apache.paimon.types.DataType;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.RowKind;
+
+import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.core.JsonProcessingException;
+import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.core.type.TypeReference;
+import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.databind.JsonNode;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import static org.apache.paimon.utils.Preconditions.checkNotNull;
+
+/**
+ * Implementation of {@link RecordParser} for parsing messages in the Ogg format.
+ *
+ * <p>This parser handles records in the Ogg format and extracts relevant information to produce
+ * {@link RichCdcMultiplexRecord} objects.
+ */
+public class OggRecordParser extends RecordParser {
+
+ private static final String FIELD_PRIMARY_KEYS = "primary_keys";
+ private static final String FIELD_BEFORE = "before";
+ private static final String FIELD_AFTER = "after";
+ private static final String FIELD_TYPE = "op_type";
+ private static final String OP_UPDATE = "U";
+ private static final String OP_INSERT = "I";
+ private static final String OP_DELETE = "D";
+ private final List<ComputedColumn> computedColumns;
+
+ public OggRecordParser(
+ boolean caseSensitive,
+ TableNameConverter tableNameConverter,
+ List<ComputedColumn> computedColumns) {
+ super(tableNameConverter, caseSensitive);
+ this.computedColumns = computedColumns;
+ }
+
+ @Override
+ public List<RichCdcMultiplexRecord> extractRecords() {
+ List<RichCdcMultiplexRecord> records = new ArrayList<>();
+ String type = extractString(FIELD_TYPE);
+ switch (type) {
+ case OP_UPDATE:
+ processRecord(root.get(FIELD_BEFORE), RowKind.UPDATE_BEFORE, records);
+ processRecord(root.get(FIELD_AFTER), RowKind.UPDATE_AFTER, records);
Review Comment:
CanalRecordParser and MySqlDebeziumJsonEventParser set the records to -D and +I. I think here should stay consistent.
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/cdc/kafka/formats/ogg/OggRecordParser.java:
##########
@@ -0,0 +1,218 @@
+/*
+ * 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.paimon.flink.action.cdc.kafka.formats.ogg;
+
+import org.apache.paimon.flink.action.cdc.ComputedColumn;
+import org.apache.paimon.flink.action.cdc.TableNameConverter;
+import org.apache.paimon.flink.action.cdc.kafka.KafkaSchema;
+import org.apache.paimon.flink.action.cdc.kafka.formats.RecordParser;
+import org.apache.paimon.flink.sink.cdc.CdcRecord;
+import org.apache.paimon.flink.sink.cdc.RichCdcMultiplexRecord;
+import org.apache.paimon.types.DataType;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.RowKind;
+
+import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.core.JsonProcessingException;
+import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.core.type.TypeReference;
+import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.databind.JsonNode;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import static org.apache.paimon.utils.Preconditions.checkNotNull;
+
+/**
+ * Implementation of {@link RecordParser} for parsing messages in the Ogg format.
+ *
+ * <p>This parser handles records in the Ogg format and extracts relevant information to produce
+ * {@link RichCdcMultiplexRecord} objects.
+ */
+public class OggRecordParser extends RecordParser {
+
+ private static final String FIELD_PRIMARY_KEYS = "primary_keys";
+ private static final String FIELD_BEFORE = "before";
+ private static final String FIELD_AFTER = "after";
+ private static final String FIELD_TYPE = "op_type";
+ private static final String OP_UPDATE = "U";
+ private static final String OP_INSERT = "I";
+ private static final String OP_DELETE = "D";
+ private final List<ComputedColumn> computedColumns;
+
+ public OggRecordParser(
+ boolean caseSensitive,
+ TableNameConverter tableNameConverter,
+ List<ComputedColumn> computedColumns) {
+ super(tableNameConverter, caseSensitive);
+ this.computedColumns = computedColumns;
+ }
+
+ @Override
+ public List<RichCdcMultiplexRecord> extractRecords() {
+ List<RichCdcMultiplexRecord> records = new ArrayList<>();
+ String type = extractString(FIELD_TYPE);
+ switch (type) {
+ case OP_UPDATE:
+ processRecord(root.get(FIELD_BEFORE), RowKind.UPDATE_BEFORE, records);
+ processRecord(root.get(FIELD_AFTER), RowKind.UPDATE_AFTER, records);
+ break;
+ case OP_INSERT:
+ processRecord(root.get(FIELD_AFTER), RowKind.INSERT, records);
+ break;
+ case OP_DELETE:
+ processRecord(root.get(FIELD_BEFORE), RowKind.DELETE, records);
+ break;
+ default:
+ throw new UnsupportedOperationException("Unknown record type: " + type);
+ }
+ return records;
+ }
+
+ private void processRecord(
+ JsonNode jsonNode, RowKind rowKind, List<RichCdcMultiplexRecord> records) {
+ LinkedHashMap<String, DataType> paimonFieldTypes = new LinkedHashMap<>();
+ Map<String, String> rowData = extractRow(jsonNode, paimonFieldTypes);
+ rowData = caseSensitive ? rowData : keyCaseInsensitive(rowData);
+ records.add(createRecord(rowKind, rowData, paimonFieldTypes));
+ }
+
+ private RichCdcMultiplexRecord createRecord(
+ RowKind rowKind,
+ Map<String, String> data,
+ LinkedHashMap<String, DataType> paimonFieldTypes) {
+ return new RichCdcMultiplexRecord(
+ databaseName,
+ tableName,
+ paimonFieldTypes,
+ extractPrimaryKeys(FIELD_PRIMARY_KEYS),
+ new CdcRecord(rowKind, data));
+ }
+
+ @Override
+ public KafkaSchema getKafkaSchema(String record) {
+ try {
+ root = OBJECT_MAPPER.readValue(record, JsonNode.class);
+ } catch (JsonProcessingException e) {
+ throw new RuntimeException("Error processing JSON: " + record, e);
+ }
+ validateFormat();
+
+ LinkedHashMap<String, DataType> paimonFieldTypes = extractFieldTypesFromOracleType();
+
+ return new KafkaSchema(
+ databaseName, tableName, paimonFieldTypes, extractPrimaryKeys(FIELD_PRIMARY_KEYS));
+ }
+
+ @Override
+ protected void validateFormat() {
+ String errorMessageTemplate =
+ "Didn't find '%s' node in json. Please make sure your topic's format is correct.";
+
+ checkNotNull(root.get(FIELD_TABLE), errorMessageTemplate, FIELD_TABLE);
+ checkNotNull(root.get(FIELD_TYPE), errorMessageTemplate, FIELD_TYPE);
+ checkNotNull(root.get(FIELD_PRIMARY_KEYS), errorMessageTemplate, FIELD_PRIMARY_KEYS);
+
+ String fieldType = root.get(FIELD_TYPE).asText();
+
+ switch (fieldType) {
+ case OP_UPDATE:
+ checkNotNull(root.get(FIELD_AFTER), errorMessageTemplate, FIELD_AFTER);
+ checkNotNull(root.get(FIELD_BEFORE), errorMessageTemplate, FIELD_BEFORE);
+ break;
+ case OP_INSERT:
+ checkNotNull(root.get(FIELD_AFTER), errorMessageTemplate, FIELD_AFTER);
+ break;
+ case OP_DELETE:
+ checkNotNull(root.get(FIELD_BEFORE), errorMessageTemplate, FIELD_BEFORE);
+ checkNotNull(root.get(FIELD_AFTER), errorMessageTemplate, "null");
+ break;
+ }
+ }
+
+ @Override
+ protected String extractString(String key) {
+ if (key.equals(FIELD_TABLE)) {
+ extractDatabaseAndTableNames();
+ return tableName;
+ } else if (key.equals(FIELD_DATABASE)) {
+ extractDatabaseAndTableNames();
+ return databaseName;
+ }
+ return root.get(key) != null ? root.get(key).asText() : null;
+ }
+
+ private void extractDatabaseAndTableNames() {
+ JsonNode tableNode = root.get(FIELD_TABLE);
+ if (tableNode != null) {
+ String[] dbt = tableNode.asText().split("\\.", 2); // Limit split to 2 parts
+ if (dbt.length == 2) {
+ databaseName = dbt[0];
+ tableName = dbt[1];
+ }
+ }
+ }
+
+ private LinkedHashMap<String, DataType> extractFieldTypesFromOracleType() {
+ LinkedHashMap<String, DataType> fieldTypes = new LinkedHashMap<>();
+
+ JsonNode record = root.get(FIELD_AFTER);
+ Map<String, Object> linkedHashMap =
+ OBJECT_MAPPER.convertValue(
+ record, new TypeReference<LinkedHashMap<String, Object>>() {});
+ if (linkedHashMap == null) {
+ return new LinkedHashMap<>();
+ }
+
+ Set<String> keySet = linkedHashMap.keySet();
+ String[] columns = keySet.toArray(new String[0]);
+ for (String column : columns) {
+ fieldTypes.put(toFieldName(column), DataTypes.STRING());
+ }
+ return fieldTypes;
+ }
+
+ private Map<String, String> extractRow(
+ JsonNode record, LinkedHashMap<String, DataType> paimonFieldTypes) {
+ Map<String, String> linkedHashMap =
+ OBJECT_MAPPER.convertValue(
+ record, new TypeReference<LinkedHashMap<String, String>>() {});
+ if (linkedHashMap == null) {
+ return new HashMap<>();
+ }
+
+ Map<String, String> resultMap = new HashMap<>();
+ for (Map.Entry<String, String> entry : linkedHashMap.entrySet()) {
+ String key = entry.getKey();
+ String value = entry.getValue();
+ paimonFieldTypes.put(key, DataTypes.STRING());
+ resultMap.put(key, value);
+ }
+
+ // generate values for computed columns
+ for (ComputedColumn computedColumn : computedColumns) {
+ resultMap.put(
+ computedColumn.columnName(),
+ computedColumn.eval(resultMap.get(computedColumn.fieldReference())));
Review Comment:
Computed column value is evaluated, should pad the paimonFieldTypes?
I also noticed that CanalRecordParser `extractRecords` has the same problem, please fix it.
##########
paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/cdc/kafka/KafkaOggSyncDatabaseActionITCase.java:
##########
@@ -0,0 +1,575 @@
+/*
+ * 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.paimon.flink.action.cdc.kafka;
+
+import org.apache.paimon.catalog.Catalog;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.schema.Schema;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.types.DataType;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.RowType;
+
+import org.apache.flink.api.common.restartstrategy.RestartStrategies;
+import org.apache.flink.core.execution.JobClient;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
+
+import javax.annotation.Nullable;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.ThreadLocalRandom;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** IT cases for {@link KafkaSyncDatabaseAction}. */
+public class KafkaOggSyncDatabaseActionITCase extends KafkaActionITCaseBase {
+
+ @Test
+ @Timeout(60)
+ public void testSchemaEvolutionMultiTopic() throws Exception {
+
+ final String topic1 = "schema_evolution_0";
+ final String topic2 = "schema_evolution_1";
+ boolean writeOne = false;
+ int fileCount = 2;
+ List<String> topics = Arrays.asList(topic1, topic2);
+ topics.forEach(
+ topic -> {
+ createTestTopic(topic, 1, 1);
+ });
+
+ // ---------- Write the ogg json into Kafka -------------------
+
+ for (int i = 0; i < fileCount; i++) {
+ try {
+ writeRecordsToKafka(
+ topics.get(i),
+ readLines(
+ "kafka/ogg/database/schemaevolution/topic"
+ + i
+ + "/ogg-data-1.txt"));
+ } catch (Exception e) {
+ throw new Exception("Failed to write ogg data to Kafka.", e);
+ }
+ }
+
+ Map<String, String> kafkaConfig = getBasicKafkaConfig();
+ kafkaConfig.put("value.format", "ogg-json");
+ kafkaConfig.put("topic", String.join(";", topics));
+
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+ env.setParallelism(2);
+ env.enableCheckpointing(1000);
+ env.setRestartStrategy(RestartStrategies.noRestart());
+
+ ThreadLocalRandom random = ThreadLocalRandom.current();
+ Map<String, String> tableConfig = new HashMap<>();
+ tableConfig.put("bucket", String.valueOf(random.nextInt(3) + 1));
+ tableConfig.put("sink.parallelism", String.valueOf(random.nextInt(3) + 1));
+ KafkaSyncDatabaseAction action =
+ new KafkaSyncDatabaseAction(
+ kafkaConfig, warehouse, database, Collections.emptyMap(), tableConfig);
+ action.build(env);
+ JobClient client = env.executeAsync();
+ waitJobRunning(client);
+
+ testSchemaEvolutionImpl(topics, writeOne, fileCount);
+ }
+
+ @Test
+ @Timeout(60)
+ public void testSchemaEvolutionOneTopic() throws Exception {
+
+ final String topic = "schema_evolution";
+ boolean writeOne = true;
+ int fileCount = 2;
+ List<String> topics = Collections.singletonList(topic);
+ topics.forEach(
+ t -> {
+ createTestTopic(t, 1, 1);
+ });
+
+ // ---------- Write the ogg json into Kafka -------------------
+
+ for (int i = 0; i < fileCount; i++) {
+ try {
+ writeRecordsToKafka(
+ topics.get(0),
+ readLines(
+ "kafka/ogg/database/schemaevolution/topic"
+ + i
+ + "/ogg-data-1.txt"));
+ } catch (Exception e) {
+ throw new Exception("Failed to write ogg data to Kafka.", e);
+ }
+ }
+
+ Map<String, String> kafkaConfig = getBasicKafkaConfig();
+ kafkaConfig.put("value.format", "ogg-json");
+ kafkaConfig.put("topic", String.join(";", topics));
+
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+ env.setParallelism(2);
+ env.enableCheckpointing(1000);
+ env.setRestartStrategy(RestartStrategies.noRestart());
+
+ ThreadLocalRandom random = ThreadLocalRandom.current();
+ Map<String, String> tableConfig = new HashMap<>();
+ tableConfig.put("bucket", String.valueOf(random.nextInt(3) + 1));
+ tableConfig.put("sink.parallelism", String.valueOf(random.nextInt(3) + 1));
+ KafkaSyncDatabaseAction action =
+ new KafkaSyncDatabaseAction(
+ kafkaConfig, warehouse, database, Collections.emptyMap(), tableConfig);
+ action.build(env);
+ JobClient client = env.executeAsync();
+ waitJobRunning(client);
+
+ testSchemaEvolutionImpl(topics, writeOne, fileCount);
+ }
+
+ private void testSchemaEvolutionImpl(List<String> topics, boolean writeOne, int fileCount)
+ throws Exception {
+ waitTablesCreated("T1", "T2");
+
+ FileStoreTable table1 = getFileStoreTable("T1");
+ FileStoreTable table2 = getFileStoreTable("T2");
+
+ RowType rowType1 =
+ RowType.of(
+ new DataType[] {
+ DataTypes.STRING().notNull(),
+ DataTypes.STRING(),
+ DataTypes.STRING(),
+ DataTypes.STRING()
+ },
+ new String[] {"id", "name", "description", "weight"});
+ List<String> primaryKeys1 = Collections.singletonList("id");
+ List<String> expected =
+ Arrays.asList(
+ "+I[101, scooter, Small 2-wheel scooter, 3.140000104904175]",
+ "+I[102, car battery, 12V car battery, 8.100000381469727]");
+ waitForResult(expected, table1, rowType1, primaryKeys1);
+
+ RowType rowType2 =
+ RowType.of(
+ new DataType[] {
+ DataTypes.STRING().notNull(),
+ DataTypes.STRING(),
+ DataTypes.STRING(),
+ DataTypes.STRING()
+ },
+ new String[] {"id", "name", "description", "weight"});
+ List<String> primaryKeys2 = Collections.singletonList("id");
+ List<String> expected2 =
+ Arrays.asList(
+ "+I[103, 12-pack drill bits, 12-pack of drill bits with sizes ranging from #40 to #3, 0.800000011920929]",
+ "+I[104, hammer, 12oz carpenter's hammer, 0.75]");
+ waitForResult(expected2, table2, rowType2, primaryKeys2);
+
+ for (int i = 0; i < fileCount; i++) {
+ try {
+ writeRecordsToKafka(
+ writeOne ? topics.get(0) : topics.get(i),
+ readLines(
+ "kafka/ogg/database/schemaevolution/topic"
+ + i
+ + "/ogg-data-2.txt"));
+ } catch (Exception e) {
+ throw new Exception("Failed to write ogg data to Kafka.", e);
+ }
+ }
+
+ rowType1 =
+ RowType.of(
+ new DataType[] {
+ DataTypes.STRING().notNull(),
+ DataTypes.STRING(),
+ DataTypes.STRING(),
+ DataTypes.STRING(),
+ DataTypes.STRING()
+ },
+ new String[] {"id", "name", "description", "weight", "age"});
+ expected =
+ Arrays.asList(
+ "+I[101, scooter, Small 2-wheel scooter, 3.140000104904175, NULL]",
+ "+I[102, car battery, 12V car battery, 8.100000381469727, NULL]",
+ "+I[103, 12-pack drill bits, 12-pack of drill bits with sizes ranging from #40 to #3, 0.800000011920929, 19]",
+ "+I[104, hammer, 12oz carpenter's hammer, 0.75, 25]");
+ waitForResult(expected, table1, rowType1, primaryKeys1);
+
+ rowType2 =
+ RowType.of(
+ new DataType[] {
+ DataTypes.STRING().notNull(),
+ DataTypes.STRING(),
+ DataTypes.STRING(),
+ DataTypes.STRING(),
+ DataTypes.STRING()
+ },
+ new String[] {"id", "name", "description", "weight", "address"});
+ expected =
+ Arrays.asList(
+ "+I[103, 12-pack drill bits, 12-pack of drill bits with sizes ranging from #40 to #3, 0.800000011920929, Beijing]",
+ "+I[104, hammer, 12oz carpenter's hammer, 0.75, Shanghai]");
+ waitForResult(expected, table2, rowType2, primaryKeys2);
+ }
+
+ @Test
+ public void testTopicIsEmpty() {
+
+ Map<String, String> kafkaConfig = getBasicKafkaConfig();
+ kafkaConfig.put("value.format", "ogg-json");
+
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+ KafkaSyncDatabaseAction action =
+ new KafkaSyncDatabaseAction(
+ kafkaConfig,
+ warehouse,
+ database,
+ Collections.emptyMap(),
+ Collections.emptyMap());
+
+ assertThatThrownBy(() -> action.build(env))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessage("kafka-conf [topic] must be specified.");
Review Comment:
It's better to use
```
assertThatThrownBy(() -> action.build(env))
.satisfies(
AssertionUtils.anyCauseMatches(....));
```
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-paimon] yuzelin commented on pull request #1827: [flink]Supported ogg format and optimize format scalability
Posted by "yuzelin (via GitHub)" <gi...@apache.org>.
yuzelin commented on PR #1827:
URL: https://github.com/apache/incubator-paimon/pull/1827#issuecomment-1689191887
LGTM. cc @JingsongLi
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-paimon] MonsterChenzhuo commented on a diff in pull request #1827: [flink]Supported ogg format and optimize format scalability
Posted by "MonsterChenzhuo (via GitHub)" <gi...@apache.org>.
MonsterChenzhuo commented on code in PR #1827:
URL: https://github.com/apache/incubator-paimon/pull/1827#discussion_r1301313268
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/cdc/kafka/formats/ogg/OggRecordParser.java:
##########
@@ -0,0 +1,218 @@
+/*
+ * 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.paimon.flink.action.cdc.kafka.formats.ogg;
+
+import org.apache.paimon.flink.action.cdc.ComputedColumn;
+import org.apache.paimon.flink.action.cdc.TableNameConverter;
+import org.apache.paimon.flink.action.cdc.kafka.KafkaSchema;
+import org.apache.paimon.flink.action.cdc.kafka.formats.RecordParser;
+import org.apache.paimon.flink.sink.cdc.CdcRecord;
+import org.apache.paimon.flink.sink.cdc.RichCdcMultiplexRecord;
+import org.apache.paimon.types.DataType;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.RowKind;
+
+import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.core.JsonProcessingException;
+import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.core.type.TypeReference;
+import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.databind.JsonNode;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import static org.apache.paimon.utils.Preconditions.checkNotNull;
+
+/**
+ * Implementation of {@link RecordParser} for parsing messages in the Ogg format.
+ *
+ * <p>This parser handles records in the Ogg format and extracts relevant information to produce
+ * {@link RichCdcMultiplexRecord} objects.
+ */
+public class OggRecordParser extends RecordParser {
+
+ private static final String FIELD_PRIMARY_KEYS = "primary_keys";
+ private static final String FIELD_BEFORE = "before";
+ private static final String FIELD_AFTER = "after";
+ private static final String FIELD_TYPE = "op_type";
+ private static final String OP_UPDATE = "U";
+ private static final String OP_INSERT = "I";
+ private static final String OP_DELETE = "D";
+ private final List<ComputedColumn> computedColumns;
+
+ public OggRecordParser(
+ boolean caseSensitive,
+ TableNameConverter tableNameConverter,
+ List<ComputedColumn> computedColumns) {
+ super(tableNameConverter, caseSensitive);
+ this.computedColumns = computedColumns;
+ }
+
+ @Override
+ public List<RichCdcMultiplexRecord> extractRecords() {
+ List<RichCdcMultiplexRecord> records = new ArrayList<>();
+ String type = extractString(FIELD_TYPE);
+ switch (type) {
+ case OP_UPDATE:
+ processRecord(root.get(FIELD_BEFORE), RowKind.UPDATE_BEFORE, records);
+ processRecord(root.get(FIELD_AFTER), RowKind.UPDATE_AFTER, records);
+ break;
+ case OP_INSERT:
+ processRecord(root.get(FIELD_AFTER), RowKind.INSERT, records);
+ break;
+ case OP_DELETE:
+ processRecord(root.get(FIELD_BEFORE), RowKind.DELETE, records);
+ break;
+ default:
+ throw new UnsupportedOperationException("Unknown record type: " + type);
+ }
+ return records;
+ }
+
+ private void processRecord(
+ JsonNode jsonNode, RowKind rowKind, List<RichCdcMultiplexRecord> records) {
+ LinkedHashMap<String, DataType> paimonFieldTypes = new LinkedHashMap<>();
+ Map<String, String> rowData = extractRow(jsonNode, paimonFieldTypes);
+ rowData = caseSensitive ? rowData : keyCaseInsensitive(rowData);
+ records.add(createRecord(rowKind, rowData, paimonFieldTypes));
+ }
+
+ private RichCdcMultiplexRecord createRecord(
+ RowKind rowKind,
+ Map<String, String> data,
+ LinkedHashMap<String, DataType> paimonFieldTypes) {
+ return new RichCdcMultiplexRecord(
+ databaseName,
+ tableName,
+ paimonFieldTypes,
+ extractPrimaryKeys(FIELD_PRIMARY_KEYS),
+ new CdcRecord(rowKind, data));
+ }
+
+ @Override
+ public KafkaSchema getKafkaSchema(String record) {
+ try {
+ root = OBJECT_MAPPER.readValue(record, JsonNode.class);
+ } catch (JsonProcessingException e) {
+ throw new RuntimeException("Error processing JSON: " + record, e);
+ }
+ validateFormat();
+
+ LinkedHashMap<String, DataType> paimonFieldTypes = extractFieldTypesFromOracleType();
+
+ return new KafkaSchema(
+ databaseName, tableName, paimonFieldTypes, extractPrimaryKeys(FIELD_PRIMARY_KEYS));
+ }
+
+ @Override
+ protected void validateFormat() {
+ String errorMessageTemplate =
+ "Didn't find '%s' node in json. Please make sure your topic's format is correct.";
+
+ checkNotNull(root.get(FIELD_TABLE), errorMessageTemplate, FIELD_TABLE);
+ checkNotNull(root.get(FIELD_TYPE), errorMessageTemplate, FIELD_TYPE);
+ checkNotNull(root.get(FIELD_PRIMARY_KEYS), errorMessageTemplate, FIELD_PRIMARY_KEYS);
+
+ String fieldType = root.get(FIELD_TYPE).asText();
+
+ switch (fieldType) {
+ case OP_UPDATE:
+ checkNotNull(root.get(FIELD_AFTER), errorMessageTemplate, FIELD_AFTER);
+ checkNotNull(root.get(FIELD_BEFORE), errorMessageTemplate, FIELD_BEFORE);
+ break;
+ case OP_INSERT:
+ checkNotNull(root.get(FIELD_AFTER), errorMessageTemplate, FIELD_AFTER);
+ break;
+ case OP_DELETE:
+ checkNotNull(root.get(FIELD_BEFORE), errorMessageTemplate, FIELD_BEFORE);
+ checkNotNull(root.get(FIELD_AFTER), errorMessageTemplate, "null");
+ break;
+ }
+ }
+
+ @Override
+ protected String extractString(String key) {
+ if (key.equals(FIELD_TABLE)) {
+ extractDatabaseAndTableNames();
+ return tableName;
+ } else if (key.equals(FIELD_DATABASE)) {
+ extractDatabaseAndTableNames();
+ return databaseName;
+ }
+ return root.get(key) != null ? root.get(key).asText() : null;
+ }
+
+ private void extractDatabaseAndTableNames() {
+ JsonNode tableNode = root.get(FIELD_TABLE);
+ if (tableNode != null) {
+ String[] dbt = tableNode.asText().split("\\.", 2); // Limit split to 2 parts
+ if (dbt.length == 2) {
+ databaseName = dbt[0];
+ tableName = dbt[1];
+ }
+ }
+ }
+
+ private LinkedHashMap<String, DataType> extractFieldTypesFromOracleType() {
+ LinkedHashMap<String, DataType> fieldTypes = new LinkedHashMap<>();
+
+ JsonNode record = root.get(FIELD_AFTER);
+ Map<String, Object> linkedHashMap =
+ OBJECT_MAPPER.convertValue(
+ record, new TypeReference<LinkedHashMap<String, Object>>() {});
+ if (linkedHashMap == null) {
+ return new LinkedHashMap<>();
+ }
+
+ Set<String> keySet = linkedHashMap.keySet();
+ String[] columns = keySet.toArray(new String[0]);
+ for (String column : columns) {
+ fieldTypes.put(toFieldName(column), DataTypes.STRING());
+ }
+ return fieldTypes;
+ }
+
+ private Map<String, String> extractRow(
+ JsonNode record, LinkedHashMap<String, DataType> paimonFieldTypes) {
+ Map<String, String> linkedHashMap =
+ OBJECT_MAPPER.convertValue(
+ record, new TypeReference<LinkedHashMap<String, String>>() {});
+ if (linkedHashMap == null) {
+ return new HashMap<>();
+ }
+
+ Map<String, String> resultMap = new HashMap<>();
+ for (Map.Entry<String, String> entry : linkedHashMap.entrySet()) {
+ String key = entry.getKey();
+ String value = entry.getValue();
+ paimonFieldTypes.put(key, DataTypes.STRING());
+ resultMap.put(key, value);
+ }
+
+ // generate values for computed columns
+ for (ComputedColumn computedColumn : computedColumns) {
+ resultMap.put(
+ computedColumn.columnName(),
+ computedColumn.eval(resultMap.get(computedColumn.fieldReference())));
Review Comment:
done.
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-paimon] MonsterChenzhuo commented on pull request #1827: [flink]Supported ogg format and optimize format scalability
Posted by "MonsterChenzhuo (via GitHub)" <gi...@apache.org>.
MonsterChenzhuo commented on PR #1827:
URL: https://github.com/apache/incubator-paimon/pull/1827#issuecomment-1687989243
@JingsongLi @yuzelin PTAL,thanks.
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-paimon] MonsterChenzhuo commented on pull request #1827: [flink]Supported ogg format and optimize format scalability
Posted by "MonsterChenzhuo (via GitHub)" <gi...@apache.org>.
MonsterChenzhuo commented on PR #1827:
URL: https://github.com/apache/incubator-paimon/pull/1827#issuecomment-1686417695
> very well job,ogg-json does not contain field type information, all types are string, whether this does not meet user expectation? but there doesn't seem to be a better solution
In my personal experience, string types work for most situations. However, I have a follow up plan for your concern.
I plan to implement it later:
1. In the whole library synchronization scenario, allow users to create tables in Paimon first. Synchronization will be based on the table created by the user for type mapping.
2. Table level synchronization. This will allow users to pass data type parameters. This approach is similar to FlinkSQL.
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-paimon] JingsongLi merged pull request #1827: [flink]Supported ogg format and optimize format scalability
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi merged PR #1827:
URL: https://github.com/apache/incubator-paimon/pull/1827
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-paimon] MonsterChenzhuo commented on a diff in pull request #1827: [WIP][flink]Supported ogg format and optimize format scalability
Posted by "MonsterChenzhuo (via GitHub)" <gi...@apache.org>.
MonsterChenzhuo commented on code in PR #1827:
URL: https://github.com/apache/incubator-paimon/pull/1827#discussion_r1297228706
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/cdc/kafka/formats/DataFormat.java:
##########
@@ -0,0 +1,66 @@
+/*
+ * 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.paimon.flink.action.cdc.kafka.formats;
+
+import org.apache.paimon.flink.action.cdc.ComputedColumn;
+import org.apache.paimon.flink.action.cdc.TableNameConverter;
+import org.apache.paimon.flink.action.cdc.kafka.formats.canal.CanalRecordParser;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.streaming.connectors.kafka.table.KafkaConnectorOptions;
+
+import java.util.List;
+
+/** supported formats. */
+public enum DataFormat {
+ CANAL_JSON(CanalRecordParser::new);
+ // Add more data formats here if needed
+
+ private final DataFormatParser parser;
+
+ DataFormat(DataFormatParser parser) {
+ this.parser = parser;
+ }
+
+ public RecordParser createParser(
+ boolean caseSensitive,
+ TableNameConverter tableNameConverter,
+ List<ComputedColumn> computedColumns) {
+ return parser.createParser(caseSensitive, tableNameConverter, computedColumns);
+ }
+
+ public static DataFormat getDataFormat(Configuration kafkaConfig) {
+ String formatStr = kafkaConfig.get(KafkaConnectorOptions.VALUE_FORMAT);
+ try {
+ return DataFormat.valueOf(formatStr.replace("-", "_").toUpperCase());
+ } catch (Exception e) {
+ throw new UnsupportedOperationException(
+ "This format: " + formatStr + " is not support.");
+ }
+ }
+
+ /** Data format formats. */
+ @FunctionalInterface
+ public interface DataFormatParser {
Review Comment:
done.
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-paimon] yuzelin commented on a diff in pull request #1827: [flink]Optimize kafka format
Posted by "yuzelin (via GitHub)" <gi...@apache.org>.
yuzelin commented on code in PR #1827:
URL: https://github.com/apache/incubator-paimon/pull/1827#discussion_r1295362957
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/cdc/kafka/formats/DataFormat.java:
##########
@@ -0,0 +1,66 @@
+/*
+ * 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.paimon.flink.action.cdc.kafka.formats;
+
+import org.apache.paimon.flink.action.cdc.ComputedColumn;
+import org.apache.paimon.flink.action.cdc.TableNameConverter;
+import org.apache.paimon.flink.action.cdc.kafka.formats.canal.CanalRecordParser;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.streaming.connectors.kafka.table.KafkaConnectorOptions;
+
+import java.util.List;
+
+/** supported formats. */
+public enum DataFormat {
+ CANAL_JSON(CanalRecordParser::new);
+ // Add more data formats here if needed
+
+ private final DataFormatParser parser;
+
+ DataFormat(DataFormatParser parser) {
+ this.parser = parser;
+ }
+
+ public RecordParser createParser(
+ boolean caseSensitive,
+ TableNameConverter tableNameConverter,
+ List<ComputedColumn> computedColumns) {
+ return parser.createParser(caseSensitive, tableNameConverter, computedColumns);
+ }
+
+ public static DataFormat getDataFormat(Configuration kafkaConfig) {
+ String formatStr = kafkaConfig.get(KafkaConnectorOptions.VALUE_FORMAT);
+ try {
+ return DataFormat.valueOf(formatStr.replace("-", "_").toUpperCase());
+ } catch (Exception e) {
+ throw new UnsupportedOperationException(
+ "This format: " + formatStr + " is not support.");
+ }
+ }
+
+ /** Data format formats. */
+ @FunctionalInterface
+ public interface DataFormatParser {
Review Comment:
I think this interface can be named `RecordParserFactory`.
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-paimon] MonsterChenzhuo commented on pull request #1827: [flink]Supported ogg format and optimize format scalability
Posted by "MonsterChenzhuo (via GitHub)" <gi...@apache.org>.
MonsterChenzhuo commented on PR #1827:
URL: https://github.com/apache/incubator-paimon/pull/1827#issuecomment-1689201505
> LGTM. cc @JingsongLi
@yuzelin
The problem of ci is a bit of a headache。I want to migrate the content of CDC into the paimon-flink/paimon-flink-cdc-e2e module in order to address the timeout issue of the utitcase-flink.yml CI process. Do you have any suggestions?
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-paimon] MonsterChenzhuo commented on a diff in pull request #1827: [flink]Supported ogg format and optimize format scalability
Posted by "MonsterChenzhuo (via GitHub)" <gi...@apache.org>.
MonsterChenzhuo commented on code in PR #1827:
URL: https://github.com/apache/incubator-paimon/pull/1827#discussion_r1301282678
##########
paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/cdc/kafka/KafkaOggSyncDatabaseActionITCase.java:
##########
@@ -0,0 +1,575 @@
+/*
+ * 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.paimon.flink.action.cdc.kafka;
+
+import org.apache.paimon.catalog.Catalog;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.schema.Schema;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.types.DataType;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.RowType;
+
+import org.apache.flink.api.common.restartstrategy.RestartStrategies;
+import org.apache.flink.core.execution.JobClient;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
+
+import javax.annotation.Nullable;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.ThreadLocalRandom;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** IT cases for {@link KafkaSyncDatabaseAction}. */
+public class KafkaOggSyncDatabaseActionITCase extends KafkaActionITCaseBase {
+
+ @Test
+ @Timeout(60)
+ public void testSchemaEvolutionMultiTopic() throws Exception {
+
+ final String topic1 = "schema_evolution_0";
+ final String topic2 = "schema_evolution_1";
+ boolean writeOne = false;
+ int fileCount = 2;
+ List<String> topics = Arrays.asList(topic1, topic2);
+ topics.forEach(
+ topic -> {
+ createTestTopic(topic, 1, 1);
+ });
+
+ // ---------- Write the ogg json into Kafka -------------------
+
+ for (int i = 0; i < fileCount; i++) {
+ try {
+ writeRecordsToKafka(
+ topics.get(i),
+ readLines(
+ "kafka/ogg/database/schemaevolution/topic"
+ + i
+ + "/ogg-data-1.txt"));
+ } catch (Exception e) {
+ throw new Exception("Failed to write ogg data to Kafka.", e);
+ }
+ }
+
+ Map<String, String> kafkaConfig = getBasicKafkaConfig();
+ kafkaConfig.put("value.format", "ogg-json");
+ kafkaConfig.put("topic", String.join(";", topics));
+
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+ env.setParallelism(2);
+ env.enableCheckpointing(1000);
+ env.setRestartStrategy(RestartStrategies.noRestart());
+
+ ThreadLocalRandom random = ThreadLocalRandom.current();
+ Map<String, String> tableConfig = new HashMap<>();
+ tableConfig.put("bucket", String.valueOf(random.nextInt(3) + 1));
+ tableConfig.put("sink.parallelism", String.valueOf(random.nextInt(3) + 1));
+ KafkaSyncDatabaseAction action =
+ new KafkaSyncDatabaseAction(
+ kafkaConfig, warehouse, database, Collections.emptyMap(), tableConfig);
+ action.build(env);
+ JobClient client = env.executeAsync();
+ waitJobRunning(client);
+
+ testSchemaEvolutionImpl(topics, writeOne, fileCount);
+ }
+
+ @Test
+ @Timeout(60)
+ public void testSchemaEvolutionOneTopic() throws Exception {
+
+ final String topic = "schema_evolution";
+ boolean writeOne = true;
+ int fileCount = 2;
+ List<String> topics = Collections.singletonList(topic);
+ topics.forEach(
+ t -> {
+ createTestTopic(t, 1, 1);
+ });
+
+ // ---------- Write the ogg json into Kafka -------------------
+
+ for (int i = 0; i < fileCount; i++) {
+ try {
+ writeRecordsToKafka(
+ topics.get(0),
+ readLines(
+ "kafka/ogg/database/schemaevolution/topic"
+ + i
+ + "/ogg-data-1.txt"));
+ } catch (Exception e) {
+ throw new Exception("Failed to write ogg data to Kafka.", e);
+ }
+ }
+
+ Map<String, String> kafkaConfig = getBasicKafkaConfig();
+ kafkaConfig.put("value.format", "ogg-json");
+ kafkaConfig.put("topic", String.join(";", topics));
+
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+ env.setParallelism(2);
+ env.enableCheckpointing(1000);
+ env.setRestartStrategy(RestartStrategies.noRestart());
+
+ ThreadLocalRandom random = ThreadLocalRandom.current();
+ Map<String, String> tableConfig = new HashMap<>();
+ tableConfig.put("bucket", String.valueOf(random.nextInt(3) + 1));
+ tableConfig.put("sink.parallelism", String.valueOf(random.nextInt(3) + 1));
+ KafkaSyncDatabaseAction action =
+ new KafkaSyncDatabaseAction(
+ kafkaConfig, warehouse, database, Collections.emptyMap(), tableConfig);
+ action.build(env);
+ JobClient client = env.executeAsync();
+ waitJobRunning(client);
+
+ testSchemaEvolutionImpl(topics, writeOne, fileCount);
+ }
+
+ private void testSchemaEvolutionImpl(List<String> topics, boolean writeOne, int fileCount)
+ throws Exception {
+ waitTablesCreated("T1", "T2");
+
+ FileStoreTable table1 = getFileStoreTable("T1");
+ FileStoreTable table2 = getFileStoreTable("T2");
+
+ RowType rowType1 =
+ RowType.of(
+ new DataType[] {
+ DataTypes.STRING().notNull(),
+ DataTypes.STRING(),
+ DataTypes.STRING(),
+ DataTypes.STRING()
+ },
+ new String[] {"id", "name", "description", "weight"});
+ List<String> primaryKeys1 = Collections.singletonList("id");
+ List<String> expected =
+ Arrays.asList(
+ "+I[101, scooter, Small 2-wheel scooter, 3.140000104904175]",
+ "+I[102, car battery, 12V car battery, 8.100000381469727]");
+ waitForResult(expected, table1, rowType1, primaryKeys1);
+
+ RowType rowType2 =
+ RowType.of(
+ new DataType[] {
+ DataTypes.STRING().notNull(),
+ DataTypes.STRING(),
+ DataTypes.STRING(),
+ DataTypes.STRING()
+ },
+ new String[] {"id", "name", "description", "weight"});
+ List<String> primaryKeys2 = Collections.singletonList("id");
+ List<String> expected2 =
+ Arrays.asList(
+ "+I[103, 12-pack drill bits, 12-pack of drill bits with sizes ranging from #40 to #3, 0.800000011920929]",
+ "+I[104, hammer, 12oz carpenter's hammer, 0.75]");
+ waitForResult(expected2, table2, rowType2, primaryKeys2);
+
+ for (int i = 0; i < fileCount; i++) {
+ try {
+ writeRecordsToKafka(
+ writeOne ? topics.get(0) : topics.get(i),
+ readLines(
+ "kafka/ogg/database/schemaevolution/topic"
+ + i
+ + "/ogg-data-2.txt"));
+ } catch (Exception e) {
+ throw new Exception("Failed to write ogg data to Kafka.", e);
+ }
+ }
+
+ rowType1 =
+ RowType.of(
+ new DataType[] {
+ DataTypes.STRING().notNull(),
+ DataTypes.STRING(),
+ DataTypes.STRING(),
+ DataTypes.STRING(),
+ DataTypes.STRING()
+ },
+ new String[] {"id", "name", "description", "weight", "age"});
+ expected =
+ Arrays.asList(
+ "+I[101, scooter, Small 2-wheel scooter, 3.140000104904175, NULL]",
+ "+I[102, car battery, 12V car battery, 8.100000381469727, NULL]",
+ "+I[103, 12-pack drill bits, 12-pack of drill bits with sizes ranging from #40 to #3, 0.800000011920929, 19]",
+ "+I[104, hammer, 12oz carpenter's hammer, 0.75, 25]");
+ waitForResult(expected, table1, rowType1, primaryKeys1);
+
+ rowType2 =
+ RowType.of(
+ new DataType[] {
+ DataTypes.STRING().notNull(),
+ DataTypes.STRING(),
+ DataTypes.STRING(),
+ DataTypes.STRING(),
+ DataTypes.STRING()
+ },
+ new String[] {"id", "name", "description", "weight", "address"});
+ expected =
+ Arrays.asList(
+ "+I[103, 12-pack drill bits, 12-pack of drill bits with sizes ranging from #40 to #3, 0.800000011920929, Beijing]",
+ "+I[104, hammer, 12oz carpenter's hammer, 0.75, Shanghai]");
+ waitForResult(expected, table2, rowType2, primaryKeys2);
+ }
+
+ @Test
+ public void testTopicIsEmpty() {
+
+ Map<String, String> kafkaConfig = getBasicKafkaConfig();
+ kafkaConfig.put("value.format", "ogg-json");
+
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+ KafkaSyncDatabaseAction action =
+ new KafkaSyncDatabaseAction(
+ kafkaConfig,
+ warehouse,
+ database,
+ Collections.emptyMap(),
+ Collections.emptyMap());
+
+ assertThatThrownBy(() -> action.build(env))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessage("kafka-conf [topic] must be specified.");
Review Comment:
done.
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-paimon] Alibaba-HZY commented on pull request #1827: [flink]Supported ogg format and optimize format scalability
Posted by "Alibaba-HZY (via GitHub)" <gi...@apache.org>.
Alibaba-HZY commented on PR #1827:
URL: https://github.com/apache/incubator-paimon/pull/1827#issuecomment-1686255902
very well job,ogg-json does not contain field type information, all types are string, whether this does not meet user expectation? but there doesn't seem to be a better solution
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-paimon] MonsterChenzhuo commented on a diff in pull request #1827: [flink]Supported ogg format and optimize format scalability
Posted by "MonsterChenzhuo (via GitHub)" <gi...@apache.org>.
MonsterChenzhuo commented on code in PR #1827:
URL: https://github.com/apache/incubator-paimon/pull/1827#discussion_r1301259994
##########
docs/content/how-to/cdc-ingestion.md:
##########
@@ -294,11 +294,15 @@ If a message in a Kafka topic is a change event captured from another database u
</tr>
<tr>
<td><a href="https://nightlies.apache.org/flink/flink-docs-release-1.16/zh/docs/connectors/table/formats/ogg/">OGG CDC</a></td>
- <td>False</td>
+ <td>True</td>
</tr>
</tbody>
</table>
+**Note that**
+
+In Oracle GoldenGate, the data format synchronized to Kafka does not include field data type information. As a result, Paimon sets the data type for all fields to "String" by default.
+
Review Comment:
done.
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-paimon] MonsterChenzhuo commented on a diff in pull request #1827: [flink]Supported ogg format and optimize format scalability
Posted by "MonsterChenzhuo (via GitHub)" <gi...@apache.org>.
MonsterChenzhuo commented on code in PR #1827:
URL: https://github.com/apache/incubator-paimon/pull/1827#discussion_r1301290534
##########
paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/cdc/kafka/KafkaActionITCaseBase.java:
##########
@@ -369,4 +372,19 @@ protected FileStoreTable getFileStoreTable(String tableName) throws Exception {
Identifier identifier = Identifier.create(database, tableName);
return (FileStoreTable) catalog().getTable(identifier);
}
+
+ protected void waitTablesCreated(String... tables) throws Exception {
+ CommonTestUtils.waitUtil(
+ () -> {
+ try {
+ List<String> existed = catalog().listTables(database);
+ return existed.containsAll(Arrays.asList(tables));
+ } catch (Catalog.DatabaseNotExistException e) {
+ throw new RuntimeException(e);
+ }
+ },
+ Duration.ofSeconds(50),
+ Duration.ofMillis(100),
+ "Failed to wait tables to be created in 5 seconds.");
Review Comment:
done.
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-paimon] MonsterChenzhuo commented on pull request #1827: [flink]Optimize kafka format
Posted by "MonsterChenzhuo (via GitHub)" <gi...@apache.org>.
MonsterChenzhuo commented on PR #1827:
URL: https://github.com/apache/incubator-paimon/pull/1827#issuecomment-1680064641
> Can you try to implement a parser for debezium? We should take a look if this refactor is OK for other formats.
Sure, I can do that.
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-paimon] yuzelin commented on pull request #1827: [flink]Supported ogg format and optimize format scalability
Posted by "yuzelin (via GitHub)" <gi...@apache.org>.
yuzelin commented on PR #1827:
URL: https://github.com/apache/incubator-paimon/pull/1827#issuecomment-1689230012
How about use syntax `mvn -T 1C -B clean install -pl 'org.apache.paimon:paimon-flink-common' -Dtest='!org.apache.paimon.flink.action.*'` in flink tests? I think this can skip the action tests. Then we can use `mvn test -Dxxx` to specify the action tests. WDYT?
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-paimon] MonsterChenzhuo commented on pull request #1827: [flink]Supported ogg format and optimize format scalability
Posted by "MonsterChenzhuo (via GitHub)" <gi...@apache.org>.
MonsterChenzhuo commented on PR #1827:
URL: https://github.com/apache/incubator-paimon/pull/1827#issuecomment-1687486261
> Can you try to implement a parser for debezium? We should take a look if this refactor is OK for other formats.
> Can you try to implement a parser for debezium? We should take a look if this refactor is OK for other formats.
Currently, the pr has chosen ogg for its implementation. After the merge, I will continue to push for debezium support.
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-paimon] MonsterChenzhuo commented on a diff in pull request #1827: [flink]Supported ogg format and optimize format scalability
Posted by "MonsterChenzhuo (via GitHub)" <gi...@apache.org>.
MonsterChenzhuo commented on code in PR #1827:
URL: https://github.com/apache/incubator-paimon/pull/1827#discussion_r1301314788
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/cdc/kafka/formats/ogg/OggRecordParser.java:
##########
@@ -0,0 +1,218 @@
+/*
+ * 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.paimon.flink.action.cdc.kafka.formats.ogg;
+
+import org.apache.paimon.flink.action.cdc.ComputedColumn;
+import org.apache.paimon.flink.action.cdc.TableNameConverter;
+import org.apache.paimon.flink.action.cdc.kafka.KafkaSchema;
+import org.apache.paimon.flink.action.cdc.kafka.formats.RecordParser;
+import org.apache.paimon.flink.sink.cdc.CdcRecord;
+import org.apache.paimon.flink.sink.cdc.RichCdcMultiplexRecord;
+import org.apache.paimon.types.DataType;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.RowKind;
+
+import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.core.JsonProcessingException;
+import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.core.type.TypeReference;
+import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.databind.JsonNode;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import static org.apache.paimon.utils.Preconditions.checkNotNull;
+
+/**
+ * Implementation of {@link RecordParser} for parsing messages in the Ogg format.
+ *
+ * <p>This parser handles records in the Ogg format and extracts relevant information to produce
+ * {@link RichCdcMultiplexRecord} objects.
+ */
+public class OggRecordParser extends RecordParser {
+
+ private static final String FIELD_PRIMARY_KEYS = "primary_keys";
+ private static final String FIELD_BEFORE = "before";
+ private static final String FIELD_AFTER = "after";
+ private static final String FIELD_TYPE = "op_type";
+ private static final String OP_UPDATE = "U";
+ private static final String OP_INSERT = "I";
+ private static final String OP_DELETE = "D";
+ private final List<ComputedColumn> computedColumns;
+
+ public OggRecordParser(
+ boolean caseSensitive,
+ TableNameConverter tableNameConverter,
+ List<ComputedColumn> computedColumns) {
+ super(tableNameConverter, caseSensitive);
+ this.computedColumns = computedColumns;
+ }
+
+ @Override
+ public List<RichCdcMultiplexRecord> extractRecords() {
+ List<RichCdcMultiplexRecord> records = new ArrayList<>();
+ String type = extractString(FIELD_TYPE);
+ switch (type) {
+ case OP_UPDATE:
+ processRecord(root.get(FIELD_BEFORE), RowKind.UPDATE_BEFORE, records);
+ processRecord(root.get(FIELD_AFTER), RowKind.UPDATE_AFTER, records);
Review Comment:
done.
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-paimon] MonsterChenzhuo commented on pull request #1827: [flink]Supported ogg format and optimize format scalability
Posted by "MonsterChenzhuo (via GitHub)" <gi...@apache.org>.
MonsterChenzhuo commented on PR #1827:
URL: https://github.com/apache/incubator-paimon/pull/1827#issuecomment-1685982185
> Very thanks to @MonsterChenzhuo .
>
> This is what we want to do! We must do some abstraction for log sync.
>
> 1. Abstraction for format: canal debezium and more...
>
> 2. Abstraction for log system: kafka pulsar..
>
>
> This is the first step.
@JingsongLi @yuzelin PTAL,thanks.
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org