From 3d778be1dc4fe7043652768b0a883878184125c7 Mon Sep 17 00:00:00 2001 From: Arnav Balyan Date: Sat, 29 Aug 2026 15:23:32 +0530 Subject: [PATCH] update --- .../debezium/DebeziumJsonRecordParser.java | 39 ++++++++ .../DebeziumJsonRecordParserTest.java | 91 +++++++++++++++++++ 2 files changed, 130 insertions(+) create mode 100644 paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/format/debezium/DebeziumJsonRecordParserTest.java diff --git a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/format/debezium/DebeziumJsonRecordParser.java b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/format/debezium/DebeziumJsonRecordParser.java index 14dd00ee3d43..9551973f35a8 100644 --- a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/format/debezium/DebeziumJsonRecordParser.java +++ b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/format/debezium/DebeziumJsonRecordParser.java @@ -218,6 +218,45 @@ protected Map extractRowData(JsonNode record, CdcSchema.Builder return resultMap; } + @Override + protected List extractPrimaryKeys() { + List primaryKeys = super.extractPrimaryKeys(); + if (!primaryKeys.isEmpty()) { + return primaryKeys; + } + + Object key = currentRecord.getKey(); + if (!(key instanceof JsonNode)) { + return Collections.emptyList(); + } + + JsonNode keyNode = (JsonNode) key; + JsonNode keySchema = keyNode.get(FIELD_SCHEMA); + if (!isNull(keySchema) + && keySchema.isObject() + && keySchema.has("fields") + && keySchema.get("fields").isArray() + && keyNode.has(FIELD_PAYLOAD)) { + ArrayNode fields = getNodeAs(keySchema, "fields", ArrayNode.class); + List fieldNames = new ArrayList<>(fields.size()); + for (JsonNode field : fields) { + String fieldName = getString(field, "field"); + if (fieldName != null) { + fieldNames.add(fieldName); + } + } + return fieldNames; + } + + if (!keyNode.isObject()) { + return Collections.emptyList(); + } + + List fieldNames = new ArrayList<>(); + keyNode.fieldNames().forEachRemaining(fieldNames::add); + return fieldNames; + } + @Override protected String primaryField() { return FIELD_PRIMARY; diff --git a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/format/debezium/DebeziumJsonRecordParserTest.java b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/format/debezium/DebeziumJsonRecordParserTest.java new file mode 100644 index 000000000000..747c56d5b55a --- /dev/null +++ b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/format/debezium/DebeziumJsonRecordParserTest.java @@ -0,0 +1,91 @@ +/* + * 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.format.debezium; + +import org.apache.paimon.flink.action.cdc.CdcSourceRecord; +import org.apache.paimon.flink.action.cdc.TypeMapping; +import org.apache.paimon.schema.Schema; + +import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.databind.JsonNode; +import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.databind.ObjectMapper; + +import org.junit.jupiter.api.Test; + +import java.util.Collections; + +import static org.assertj.core.api.Assertions.assertThat; + +/** Tests for {@link DebeziumJsonRecordParser}. */ +public class DebeziumJsonRecordParserTest { + + private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper(); + + @Test + public void testPrimaryKeysFromSchemaEnabledKey() throws Exception { + JsonNode key = + OBJECT_MAPPER.readTree( + "{\"schema\":{\"type\":\"struct\",\"fields\":[" + + "{\"type\":\"int64\",\"optional\":false,\"field\":\"id\"}," + + "{\"type\":\"string\",\"optional\":false,\"field\":\"tenant\"}]}," + + "\"payload\":{\"id\":1,\"tenant\":\"A\"}}"); + + assertPrimaryKeys(key, value(null), "id", "tenant"); + } + + @Test + public void testPrimaryKeysFromSchemaLessKey() throws Exception { + JsonNode key = OBJECT_MAPPER.readTree("{\"id\":1,\"tenant\":\"A\"}"); + + assertPrimaryKeys(key, value(null), "id", "tenant"); + } + + @Test + public void testPrimaryKeysFromValueTakePrecedence() throws Exception { + JsonNode key = OBJECT_MAPPER.readTree("{\"id\":1}"); + + assertPrimaryKeys(key, value("[\"tenant\"]"), "tenant"); + } + + @Test + public void testEmptyPrimaryKeysFallBackToKey() throws Exception { + JsonNode key = OBJECT_MAPPER.readTree("{\"id\":1}"); + + assertPrimaryKeys(key, value("[]"), "id"); + } + + private static JsonNode value(String primaryKeys) throws Exception { + String primaryKeyField = primaryKeys == null ? "" : "\"pkNames\":" + primaryKeys + ","; + return OBJECT_MAPPER.readTree( + "{" + + primaryKeyField + + "\"before\":null," + + "\"after\":{\"id\":1,\"tenant\":\"A\",\"name\":\"Alice\"}," + + "\"source\":{\"db\":\"test\",\"table\":\"users\"}," + + "\"op\":\"c\"}"); + } + + private static void assertPrimaryKeys(JsonNode key, JsonNode value, String... primaryKeys) { + DebeziumJsonRecordParser parser = + new DebeziumJsonRecordParser(TypeMapping.defaultMapping(), Collections.emptyList()); + Schema schema = parser.buildSchema(new CdcSourceRecord("users", key, value)); + + assertThat(schema).isNotNull(); + assertThat(schema.primaryKeys()).containsExactly(primaryKeys); + } +}