This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/dev/pr-12839-2be65da8caceb528790e1ff794b4b842e55b69eb in repository https://gitbox.apache.org/repos/asf/seatunnel.git
commit 68c686438f31fcbaa376081af6a4536caa227f12 Author: Goutam Adwant <[email protected]> AuthorDate: Sun Oct 18 06:02:54 2026 +0000 [Fix][Connector-V2] Add opt-in collision-safe Easysearch and Typesense IDs (#12839) --- docs/en/connectors/sink/Easysearch.md | 7 ++ docs/en/connectors/sink/Typesense.md | 7 ++ docs/zh/connectors/sink/Easysearch.md | 7 ++ docs/zh/connectors/sink/Typesense.md | 7 ++ .../easysearch/config/EasysearchSinkOptions.java | 7 ++ .../seatunnel/easysearch/config/KeyEncoding.java} | 26 ++----- .../seatunnel/easysearch/dto/IndexInfo.java | 3 + .../serialize/EasysearchRowSerializer.java | 5 +- .../easysearch/serialize/KeyExtractor.java | 25 ++++-- .../easysearch/sink/EasysearchSinkFactory.java | 1 + .../easysearch/serialize/KeyExtractorTest.java | 89 ++++++++++++++++++++++ .../KeyEncoding.java} | 26 ++----- .../typesense/config/TypesenseSinkOptions.java | 7 ++ .../seatunnel/typesense/dto/CollectionInfo.java | 3 + .../typesense/serialize/KeyExtractor.java | 25 ++++-- .../serialize/sink/TypesenseRowSerializer.java | 10 +-- .../typesense/sink/TypesenseSinkFactory.java | 1 + .../serializer/TypesenseRowSerializerTest.java | 50 ++++++++++++ .../e2e/connector/easysearch/EasysearchIT.java | 21 +++++ .../easysearch_composite_key_encoding.conf | 50 ++++++++++++ .../e2e/connector/typesense/TypesenseIT.java | 9 +++ .../fake_to_typesense_composite_key_encoding.conf | 47 ++++++++++++ 22 files changed, 375 insertions(+), 58 deletions(-) diff --git a/docs/en/connectors/sink/Easysearch.md b/docs/en/connectors/sink/Easysearch.md index 671c95838e..15071472df 100644 --- a/docs/en/connectors/sink/Easysearch.md +++ b/docs/en/connectors/sink/Easysearch.md @@ -52,6 +52,7 @@ Engine Supported | index | string | yes | - | | primary_keys | list | no | | | key_delimiter | string | no | `_` | +| key_encoding | enum | no | LEGACY | | username | string | no | | | password | string | no | | | max_retry_count | int | no | 3 | @@ -82,6 +83,12 @@ Primary key fields used to generate the document `_id`. Configure this option wh Delimiter for composite keys ("_" by default), e.g., "$" would result in document `_id` "KEY1$KEY2$KEY3". +### key_encoding [enum] + +`LEGACY` (default) keeps existing document IDs. With multiple `primary_keys`, values containing `key_delimiter` can produce the same ID. Set `LENGTH_PREFIXED` to encode each formatted value as `<length>:<value>`; for example, `["1_1", "2"]` becomes `3:1_11:2` and `["1", "1_2"]` becomes `1:13:1_2`. This mode ignores `key_delimiter` and leaves single-key IDs unchanged. + +Changing this option for an existing composite-key index changes its document IDs. Reindex or migrate existing documents before switching; otherwise subsequent writes can create duplicates instead of updating the old IDs. + ### username [string] security username diff --git a/docs/en/connectors/sink/Typesense.md b/docs/en/connectors/sink/Typesense.md index dcc538dafb..527efefd62 100644 --- a/docs/en/connectors/sink/Typesense.md +++ b/docs/en/connectors/sink/Typesense.md @@ -33,6 +33,7 @@ or more primary key fields. | data_save_mode | string | Yes | APPEND_DATA | How to handle existing documents before writing. | | primary_keys | array | No | - | Source fields used to build the Typesense document `id`. | | key_delimiter | string | No | `_` | Delimiter used when `primary_keys` contains more than one field. | +| key_encoding | enum | No | LEGACY | Optional collision-safe encoding for composite document IDs. | | api_key | string | Yes | - | Typesense API key. | | max_retry_count | int | No | 3 | Maximum retry count for one bulk request. | | max_batch_size | int | No | 10 | Maximum number of documents sent in one bulk request. | @@ -56,6 +57,12 @@ the connector joins their values with `key_delimiter`. Without `primary_keys`, T Sets the delimiter for composite keys (default is `_`). +### key_encoding [enum] + +`LEGACY` (default) preserves existing IDs, but composite keys can collide when a value contains `key_delimiter`. `LENGTH_PREFIXED` encodes each formatted value as `<length>:<value>`: `["1_1", "2"]` becomes `3:1_11:2`, while `["1", "1_2"]` becomes `1:13:1_2`. It ignores `key_delimiter` and does not change single-key IDs. + +Switching an existing composite-key collection changes document IDs. Migrate or recreate existing documents first; otherwise later writes can create duplicates instead of updating them. + ### api_key [string] The `api_key` for secure access to Typesense. Treat this value as a secret and prefer passing it via a job secret or environment variable when running on shared infrastructure. diff --git a/docs/zh/connectors/sink/Easysearch.md b/docs/zh/connectors/sink/Easysearch.md index 53f77d6531..109217a7c1 100644 --- a/docs/zh/connectors/sink/Easysearch.md +++ b/docs/zh/connectors/sink/Easysearch.md @@ -51,6 +51,7 @@ import ChangeLog from '../changelog/connector-easysearch.md'; | index | string | 是 | - | | primary_keys | list | 否 | - | | key_delimiter | string | 否 | `_` | +| key_encoding | enum | 否 | LEGACY | | username | string | 否 | - | | password | string | 否 | - | | max_retry_count | int | 否 | 3 | @@ -81,6 +82,12 @@ import ChangeLog from '../changelog/connector-easysearch.md'; 复合键的分隔符 (默认为"_" ), 例如, "$" 将导致文档 `_id` "KEY1$KEY2$KEY3". +### key_encoding [enum] + +`LEGACY`（默认）保持现有文档 ID 不变。配置多个 `primary_keys` 时，字段值包含 `key_delimiter` 可能导致不同记录生成相同 ID。设置为 `LENGTH_PREFIXED` 后，每个格式化后的字段值按 `<长度>:<值>` 编码，例如 `["1_1", "2"]` 生成 `3:1_11:2`，`["1", "1_2"]` 生成 `1:13:1_2`。该模式忽略 `key_delimiter`，单字段主键的 ID 不变。 + +对已有复合键索引切换此选项会改变文档 ID。切换前应重建索引或迁移已有文档，否则后续写入可能创建重复文档，而不是更新原文档。 + ### username [string] 安全用户名 diff --git a/docs/zh/connectors/sink/Typesense.md b/docs/zh/connectors/sink/Typesense.md index c9217c8dd3..d5da897b62 100644 --- a/docs/zh/connectors/sink/Typesense.md +++ b/docs/zh/connectors/sink/Typesense.md @@ -32,6 +32,7 @@ import ChangeLog from '../changelog/connector-typesense.md'; | data_save_mode | string | 是 | APPEND_DATA | 写入前如何处理目标 collection 中已有文档。 | | primary_keys | array | 否 | - | 用于生成 Typesense 文档 `id` 的源字段。 | | key_delimiter | string | 否 | `_` | `primary_keys` 配置多个字段时使用的拼接分隔符。 | +| key_encoding | enum | 否 | LEGACY | 可选的复合文档 ID 防冲突编码方式。 | | api_key | string | 是 | - | Typesense API Key。 | | max_retry_count | int | 否 | 3 | 单个批量请求的最大重试次数。 | | max_batch_size | int | 否 | 10 | 单个批量请求最多写入的文档数量。 | @@ -54,6 +55,12 @@ Typesense 的访问地址，格式为 `host:port`，例如：`["typesense-01:810 设定复合键的分隔符（默认为 `_`）。 +### key_encoding [enum] + +`LEGACY`（默认）保持现有 ID 不变，但字段值包含 `key_delimiter` 时复合键可能发生 ID 冲突。`LENGTH_PREFIXED` 将每个格式化后的字段值按 `<长度>:<值>` 编码：`["1_1", "2"]` 生成 `3:1_11:2`，`["1", "1_2"]` 生成 `1:13:1_2`。该模式忽略 `key_delimiter`，单字段主键的 ID 不变。 + +对已有复合键 collection 切换此选项会改变文档 ID。切换前应迁移或重新创建已有文档，否则后续写入可能创建重复文档，而不是更新原文档。 + ### api_key [string] Typesense 安全认证的 `api_key`。请把它当作敏感凭据处理；在共享环境运行时，建议通过作业密钥或环境变量注入。 diff --git a/seatunnel-connectors-v2/connector-easysearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/easysearch/config/EasysearchSinkOptions.java b/seatunnel-connectors-v2/connector-easysearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/easysearch/config/EasysearchSinkOptions.java index a7d0535b2f..fc504e1930 100644 --- a/seatunnel-connectors-v2/connector-easysearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/easysearch/config/EasysearchSinkOptions.java +++ b/seatunnel-connectors-v2/connector-easysearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/easysearch/config/EasysearchSinkOptions.java @@ -44,6 +44,13 @@ public class EasysearchSinkOptions extends EasysearchSinkCommonOptions { .withDescription( "Delimiter for composite keys (\"_\" by default), e.g., \"$\" would result in document `_id` \"KEY1$KEY2$KEY3\"."); + public static final Option<KeyEncoding> KEY_ENCODING = + Options.key("key_encoding") + .enumType(KeyEncoding.class) + .defaultValue(KeyEncoding.LEGACY) + .withDescription( + "LEGACY preserves existing document IDs; LENGTH_PREFIXED avoids composite-key delimiter collisions."); + public static final Option<Integer> MAX_BATCH_SIZE = Options.key("max_batch_size") .intType() diff --git a/seatunnel-connectors-v2/connector-typesense/src/main/java/org/apache/seatunnel/connectors/seatunnel/typesense/dto/CollectionInfo.java b/seatunnel-connectors-v2/connector-easysearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/easysearch/config/KeyEncoding.java similarity index 51% copy from seatunnel-connectors-v2/connector-typesense/src/main/java/org/apache/seatunnel/connectors/seatunnel/typesense/dto/CollectionInfo.java copy to seatunnel-connectors-v2/connector-easysearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/easysearch/config/KeyEncoding.java index e2ab88c832..2654ae752f 100644 --- a/seatunnel-connectors-v2/connector-typesense/src/main/java/org/apache/seatunnel/connectors/seatunnel/typesense/dto/CollectionInfo.java +++ b/seatunnel-connectors-v2/connector-easysearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/easysearch/config/KeyEncoding.java @@ -15,26 +15,10 @@ * limitations under the License. */ -package org.apache.seatunnel.connectors.seatunnel.typesense.dto; +package org.apache.seatunnel.connectors.seatunnel.easysearch.config; -import org.apache.seatunnel.api.configuration.ReadonlyConfig; -import org.apache.seatunnel.connectors.seatunnel.typesense.config.TypesenseSinkOptions; - -import lombok.Data; - -@Data -public class CollectionInfo { - - private String collection; - private String type; - private String[] primaryKeys; - private String keyDelimiter; - - public CollectionInfo(String collection, ReadonlyConfig config) { - this.collection = collection; - if (config.getOptional(TypesenseSinkOptions.PRIMARY_KEYS).isPresent()) { - primaryKeys = config.get(TypesenseSinkOptions.PRIMARY_KEYS).toArray(new String[0]); - } - keyDelimiter = config.get(TypesenseSinkOptions.KEY_DELIMITER); - } +/** Encoding of composite document IDs. Existing indexes use {@link #LEGACY}. */ +public enum KeyEncoding { + LEGACY, + LENGTH_PREFIXED } diff --git a/seatunnel-connectors-v2/connector-easysearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/easysearch/dto/IndexInfo.java b/seatunnel-connectors-v2/connector-easysearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/easysearch/dto/IndexInfo.java index 8da4a0854d..961b876f64 100644 --- a/seatunnel-connectors-v2/connector-easysearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/easysearch/dto/IndexInfo.java +++ b/seatunnel-connectors-v2/connector-easysearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/easysearch/dto/IndexInfo.java @@ -20,6 +20,7 @@ package org.apache.seatunnel.connectors.seatunnel.easysearch.dto; import org.apache.seatunnel.api.configuration.ReadonlyConfig; import org.apache.seatunnel.connectors.seatunnel.easysearch.config.EasysearchSinkCommonOptions; import org.apache.seatunnel.connectors.seatunnel.easysearch.config.EasysearchSinkOptions; +import org.apache.seatunnel.connectors.seatunnel.easysearch.config.KeyEncoding; import lombok.Data; @@ -30,6 +31,7 @@ public class IndexInfo { private String index; private String[] primaryKeys; private String keyDelimiter; + private KeyEncoding keyEncoding; public IndexInfo(ReadonlyConfig pluginConfig) { index = pluginConfig.get(EasysearchSinkCommonOptions.INDEX); @@ -38,5 +40,6 @@ public class IndexInfo { pluginConfig.get(EasysearchSinkOptions.PRIMARY_KEYS).toArray(new String[0]); } keyDelimiter = pluginConfig.get(EasysearchSinkOptions.KEY_DELIMITER); + keyEncoding = pluginConfig.get(EasysearchSinkOptions.KEY_ENCODING); } } diff --git a/seatunnel-connectors-v2/connector-easysearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/easysearch/serialize/EasysearchRowSerializer.java b/seatunnel-connectors-v2/connector-easysearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/easysearch/serialize/EasysearchRowSerializer.java index e57ad1fd51..c106fe531e 100644 --- a/seatunnel-connectors-v2/connector-easysearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/easysearch/serialize/EasysearchRowSerializer.java +++ b/seatunnel-connectors-v2/connector-easysearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/easysearch/serialize/EasysearchRowSerializer.java @@ -51,7 +51,10 @@ public class EasysearchRowSerializer implements SeaTunnelRowSerializer { this.seaTunnelRowType = seaTunnelRowType; this.keyExtractor = KeyExtractor.createKeyExtractor( - seaTunnelRowType, indexInfo.getPrimaryKeys(), indexInfo.getKeyDelimiter()); + seaTunnelRowType, + indexInfo.getPrimaryKeys(), + indexInfo.getKeyDelimiter(), + indexInfo.getKeyEncoding()); } @Override diff --git a/seatunnel-connectors-v2/connector-easysearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/easysearch/serialize/KeyExtractor.java b/seatunnel-connectors-v2/connector-easysearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/easysearch/serialize/KeyExtractor.java index c732d4288b..bf8ab7f396 100644 --- a/seatunnel-connectors-v2/connector-easysearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/easysearch/serialize/KeyExtractor.java +++ b/seatunnel-connectors-v2/connector-easysearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/easysearch/serialize/KeyExtractor.java @@ -20,6 +20,7 @@ package org.apache.seatunnel.connectors.seatunnel.easysearch.serialize; import org.apache.seatunnel.api.table.type.SeaTunnelDataType; import org.apache.seatunnel.api.table.type.SeaTunnelRow; import org.apache.seatunnel.api.table.type.SeaTunnelRowType; +import org.apache.seatunnel.connectors.seatunnel.easysearch.config.KeyEncoding; import org.apache.seatunnel.connectors.seatunnel.easysearch.exception.EasysearchConnectorException; import lombok.AllArgsConstructor; @@ -38,9 +39,18 @@ import static org.apache.seatunnel.connectors.seatunnel.easysearch.exception.Eas public class KeyExtractor implements Function<SeaTunnelRow, String>, Serializable { private final FieldFormatter[] fieldFormatters; private final String keyDelimiter; + private final KeyEncoding keyEncoding; public static Function<SeaTunnelRow, String> createKeyExtractor( SeaTunnelRowType rowType, String[] primaryKeys, String keyDelimiter) { + return createKeyExtractor(rowType, primaryKeys, keyDelimiter, KeyEncoding.LEGACY); + } + + public static Function<SeaTunnelRow, String> createKeyExtractor( + SeaTunnelRowType rowType, + String[] primaryKeys, + String keyDelimiter, + KeyEncoding keyEncoding) { if (primaryKeys == null) { return row -> null; } @@ -52,7 +62,8 @@ public class KeyExtractor implements Function<SeaTunnelRow, String>, Serializabl FieldFormatter fieldFormatter = createFieldFormatter(fieldIndex, fieldType); fieldFormatters.add(fieldFormatter); } - return new KeyExtractor(fieldFormatters.toArray(new FieldFormatter[0]), keyDelimiter); + return new KeyExtractor( + fieldFormatters.toArray(new FieldFormatter[0]), keyDelimiter, keyEncoding); } private static FieldFormatter createFieldFormatter( @@ -83,11 +94,15 @@ public class KeyExtractor implements Function<SeaTunnelRow, String>, Serializabl public String apply(SeaTunnelRow row) { StringBuilder builder = new StringBuilder(); for (int i = 0; i < fieldFormatters.length; i++) { - if (i > 0) { - builder.append(keyDelimiter); - } String value = fieldFormatters[i].format(row); - builder.append(value); + if (keyEncoding == KeyEncoding.LENGTH_PREFIXED && fieldFormatters.length > 1) { + builder.append(value.length()).append(':').append(value); + } else { + if (i > 0) { + builder.append(keyDelimiter); + } + builder.append(value); + } } return builder.toString(); } diff --git a/seatunnel-connectors-v2/connector-easysearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/easysearch/sink/EasysearchSinkFactory.java b/seatunnel-connectors-v2/connector-easysearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/easysearch/sink/EasysearchSinkFactory.java index ec38d23166..23fd6dfacf 100644 --- a/seatunnel-connectors-v2/connector-easysearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/easysearch/sink/EasysearchSinkFactory.java +++ b/seatunnel-connectors-v2/connector-easysearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/easysearch/sink/EasysearchSinkFactory.java @@ -54,6 +54,7 @@ public class EasysearchSinkFactory implements TableSinkFactory { EasysearchSinkOptions.PASSWORD, EasysearchSinkOptions.PRIMARY_KEYS, EasysearchSinkOptions.KEY_DELIMITER, + EasysearchSinkOptions.KEY_ENCODING, EasysearchSinkOptions.TLS_VERIFY_CERTIFICATE, EasysearchSinkOptions.TLS_VERIFY_HOSTNAME, EasysearchSinkOptions.TLS_KEY_STORE_PATH, diff --git a/seatunnel-connectors-v2/connector-easysearch/src/test/java/org/apache/seatunnel/connectors/seatunnel/easysearch/serialize/KeyExtractorTest.java b/seatunnel-connectors-v2/connector-easysearch/src/test/java/org/apache/seatunnel/connectors/seatunnel/easysearch/serialize/KeyExtractorTest.java new file mode 100644 index 0000000000..34875cbbc4 --- /dev/null +++ b/seatunnel-connectors-v2/connector-easysearch/src/test/java/org/apache/seatunnel/connectors/seatunnel/easysearch/serialize/KeyExtractorTest.java @@ -0,0 +1,89 @@ +/* + * 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.seatunnel.connectors.seatunnel.easysearch.serialize; + +import org.apache.seatunnel.api.configuration.ReadonlyConfig; +import org.apache.seatunnel.api.table.type.BasicType; +import org.apache.seatunnel.api.table.type.RowKind; +import org.apache.seatunnel.api.table.type.SeaTunnelDataType; +import org.apache.seatunnel.api.table.type.SeaTunnelRow; +import org.apache.seatunnel.api.table.type.SeaTunnelRowType; +import org.apache.seatunnel.connectors.seatunnel.easysearch.config.EasysearchSinkOptions; +import org.apache.seatunnel.connectors.seatunnel.easysearch.config.KeyEncoding; +import org.apache.seatunnel.connectors.seatunnel.easysearch.dto.IndexInfo; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +import java.util.Arrays; +import java.util.HashMap; +import java.util.Map; +import java.util.function.Function; + +class KeyExtractorTest { + private static final SeaTunnelRowType ROW_TYPE = + new SeaTunnelRowType( + new String[] {"first", "second"}, + new SeaTunnelDataType<?>[] {BasicType.STRING_TYPE, BasicType.STRING_TYPE}); + + @Test + void lengthPrefixedIdsSeparateDelimiterCollisions() { + Function<SeaTunnelRow, String> legacy = + KeyExtractor.createKeyExtractor(ROW_TYPE, new String[] {"first", "second"}, "_"); + Function<SeaTunnelRow, String> safe = + KeyExtractor.createKeyExtractor( + ROW_TYPE, + new String[] {"first", "second"}, + "_", + KeyEncoding.LENGTH_PREFIXED); + SeaTunnelRow left = new SeaTunnelRow(new Object[] {"1_1", "2"}); + SeaTunnelRow right = new SeaTunnelRow(new Object[] {"1", "1_2"}); + + Assertions.assertEquals("1_1_2", legacy.apply(left)); + Assertions.assertEquals(legacy.apply(left), legacy.apply(right)); + Assertions.assertEquals("3:1_11:2", safe.apply(left)); + Assertions.assertEquals("1:13:1_2", safe.apply(right)); + } + + @Test + void oneKeyKeepsItsExistingIdInBothModes() { + SeaTunnelRow row = new SeaTunnelRow(new Object[] {"1_1", "2"}); + Assertions.assertEquals( + "1_1", + KeyExtractor.createKeyExtractor( + ROW_TYPE, new String[] {"first"}, "_", KeyEncoding.LENGTH_PREFIXED) + .apply(row)); + } + + @Test + void serializerUsesTheSameSafeIdForUpsertAndDelete() { + Map<String, Object> options = new HashMap<>(); + options.put(EasysearchSinkOptions.INDEX.key(), "id_probe"); + options.put(EasysearchSinkOptions.PRIMARY_KEYS.key(), Arrays.asList("first", "second")); + options.put(EasysearchSinkOptions.KEY_ENCODING.key(), "LENGTH_PREFIXED"); + EasysearchRowSerializer serializer = + new EasysearchRowSerializer( + new IndexInfo(ReadonlyConfig.fromMap(options)), ROW_TYPE); + SeaTunnelRow row = new SeaTunnelRow(new Object[] {"1_1", "2"}); + + row.setRowKind(RowKind.INSERT); + Assertions.assertTrue(serializer.serializeRow(row).contains("3:1_11:2")); + row.setRowKind(RowKind.DELETE); + Assertions.assertTrue(serializer.serializeRow(row).contains("3:1_11:2")); + } +} diff --git a/seatunnel-connectors-v2/connector-typesense/src/main/java/org/apache/seatunnel/connectors/seatunnel/typesense/dto/CollectionInfo.java b/seatunnel-connectors-v2/connector-typesense/src/main/java/org/apache/seatunnel/connectors/seatunnel/typesense/config/KeyEncoding.java similarity index 51% copy from seatunnel-connectors-v2/connector-typesense/src/main/java/org/apache/seatunnel/connectors/seatunnel/typesense/dto/CollectionInfo.java copy to seatunnel-connectors-v2/connector-typesense/src/main/java/org/apache/seatunnel/connectors/seatunnel/typesense/config/KeyEncoding.java index e2ab88c832..ee6d067c4f 100644 --- a/seatunnel-connectors-v2/connector-typesense/src/main/java/org/apache/seatunnel/connectors/seatunnel/typesense/dto/CollectionInfo.java +++ b/seatunnel-connectors-v2/connector-typesense/src/main/java/org/apache/seatunnel/connectors/seatunnel/typesense/config/KeyEncoding.java @@ -15,26 +15,10 @@ * limitations under the License. */ -package org.apache.seatunnel.connectors.seatunnel.typesense.dto; +package org.apache.seatunnel.connectors.seatunnel.typesense.config; -import org.apache.seatunnel.api.configuration.ReadonlyConfig; -import org.apache.seatunnel.connectors.seatunnel.typesense.config.TypesenseSinkOptions; - -import lombok.Data; - -@Data -public class CollectionInfo { - - private String collection; - private String type; - private String[] primaryKeys; - private String keyDelimiter; - - public CollectionInfo(String collection, ReadonlyConfig config) { - this.collection = collection; - if (config.getOptional(TypesenseSinkOptions.PRIMARY_KEYS).isPresent()) { - primaryKeys = config.get(TypesenseSinkOptions.PRIMARY_KEYS).toArray(new String[0]); - } - keyDelimiter = config.get(TypesenseSinkOptions.KEY_DELIMITER); - } +/** Encoding of composite document IDs. Existing indexes use {@link #LEGACY}. */ +public enum KeyEncoding { + LEGACY, + LENGTH_PREFIXED } diff --git a/seatunnel-connectors-v2/connector-typesense/src/main/java/org/apache/seatunnel/connectors/seatunnel/typesense/config/TypesenseSinkOptions.java b/seatunnel-connectors-v2/connector-typesense/src/main/java/org/apache/seatunnel/connectors/seatunnel/typesense/config/TypesenseSinkOptions.java index 7b87fbfb82..8e9d141f6d 100644 --- a/seatunnel-connectors-v2/connector-typesense/src/main/java/org/apache/seatunnel/connectors/seatunnel/typesense/config/TypesenseSinkOptions.java +++ b/seatunnel-connectors-v2/connector-typesense/src/main/java/org/apache/seatunnel/connectors/seatunnel/typesense/config/TypesenseSinkOptions.java @@ -44,6 +44,13 @@ public class TypesenseSinkOptions extends TypesenseBaseOptions { .withDescription( "Delimiter for composite keys (\"_\" by default), e.g., \"$\" would result in document `id` \"KEY1$KEY2$KEY3\"."); + public static final Option<KeyEncoding> KEY_ENCODING = + Options.key("key_encoding") + .enumType(KeyEncoding.class) + .defaultValue(KeyEncoding.LEGACY) + .withDescription( + "LEGACY preserves existing document IDs; LENGTH_PREFIXED avoids composite-key delimiter collisions."); + public static final Option<Integer> MAX_BATCH_SIZE = Options.key("max_batch_size") .intType() diff --git a/seatunnel-connectors-v2/connector-typesense/src/main/java/org/apache/seatunnel/connectors/seatunnel/typesense/dto/CollectionInfo.java b/seatunnel-connectors-v2/connector-typesense/src/main/java/org/apache/seatunnel/connectors/seatunnel/typesense/dto/CollectionInfo.java index e2ab88c832..803a1a7530 100644 --- a/seatunnel-connectors-v2/connector-typesense/src/main/java/org/apache/seatunnel/connectors/seatunnel/typesense/dto/CollectionInfo.java +++ b/seatunnel-connectors-v2/connector-typesense/src/main/java/org/apache/seatunnel/connectors/seatunnel/typesense/dto/CollectionInfo.java @@ -18,6 +18,7 @@ package org.apache.seatunnel.connectors.seatunnel.typesense.dto; import org.apache.seatunnel.api.configuration.ReadonlyConfig; +import org.apache.seatunnel.connectors.seatunnel.typesense.config.KeyEncoding; import org.apache.seatunnel.connectors.seatunnel.typesense.config.TypesenseSinkOptions; import lombok.Data; @@ -29,6 +30,7 @@ public class CollectionInfo { private String type; private String[] primaryKeys; private String keyDelimiter; + private KeyEncoding keyEncoding; public CollectionInfo(String collection, ReadonlyConfig config) { this.collection = collection; @@ -36,5 +38,6 @@ public class CollectionInfo { primaryKeys = config.get(TypesenseSinkOptions.PRIMARY_KEYS).toArray(new String[0]); } keyDelimiter = config.get(TypesenseSinkOptions.KEY_DELIMITER); + keyEncoding = config.get(TypesenseSinkOptions.KEY_ENCODING); } } diff --git a/seatunnel-connectors-v2/connector-typesense/src/main/java/org/apache/seatunnel/connectors/seatunnel/typesense/serialize/KeyExtractor.java b/seatunnel-connectors-v2/connector-typesense/src/main/java/org/apache/seatunnel/connectors/seatunnel/typesense/serialize/KeyExtractor.java index 5dc56f0bb7..a8cbce8a93 100644 --- a/seatunnel-connectors-v2/connector-typesense/src/main/java/org/apache/seatunnel/connectors/seatunnel/typesense/serialize/KeyExtractor.java +++ b/seatunnel-connectors-v2/connector-typesense/src/main/java/org/apache/seatunnel/connectors/seatunnel/typesense/serialize/KeyExtractor.java @@ -21,6 +21,7 @@ import org.apache.seatunnel.api.table.type.SeaTunnelDataType; import org.apache.seatunnel.api.table.type.SeaTunnelRow; import org.apache.seatunnel.api.table.type.SeaTunnelRowType; import org.apache.seatunnel.common.exception.CommonErrorCodeDeprecated; +import org.apache.seatunnel.connectors.seatunnel.typesense.config.KeyEncoding; import org.apache.seatunnel.connectors.seatunnel.typesense.exception.TypesenseConnectorException; import lombok.AllArgsConstructor; @@ -37,22 +38,35 @@ import java.util.function.Function; public class KeyExtractor implements Function<SeaTunnelRow, String>, Serializable { private final FieldFormatter[] fieldFormatters; private final String keyDelimiter; + private final KeyEncoding keyEncoding; @Override public String apply(SeaTunnelRow row) { StringBuilder builder = new StringBuilder(); for (int i = 0; i < fieldFormatters.length; i++) { - if (i > 0) { - builder.append(keyDelimiter); - } String value = fieldFormatters[i].format(row); - builder.append(value); + if (keyEncoding == KeyEncoding.LENGTH_PREFIXED && fieldFormatters.length > 1) { + builder.append(value.length()).append(':').append(value); + } else { + if (i > 0) { + builder.append(keyDelimiter); + } + builder.append(value); + } } return builder.toString(); } public static Function<SeaTunnelRow, String> createKeyExtractor( SeaTunnelRowType rowType, String[] primaryKeys, String keyDelimiter) { + return createKeyExtractor(rowType, primaryKeys, keyDelimiter, KeyEncoding.LEGACY); + } + + public static Function<SeaTunnelRow, String> createKeyExtractor( + SeaTunnelRowType rowType, + String[] primaryKeys, + String keyDelimiter, + KeyEncoding keyEncoding) { if (primaryKeys == null) { return row -> null; } @@ -64,7 +78,8 @@ public class KeyExtractor implements Function<SeaTunnelRow, String>, Serializabl FieldFormatter fieldFormatter = createFieldFormatter(fieldIndex, fieldType); fieldFormatters.add(fieldFormatter); } - return new KeyExtractor(fieldFormatters.toArray(new FieldFormatter[0]), keyDelimiter); + return new KeyExtractor( + fieldFormatters.toArray(new FieldFormatter[0]), keyDelimiter, keyEncoding); } private static FieldFormatter createFieldFormatter( diff --git a/seatunnel-connectors-v2/connector-typesense/src/main/java/org/apache/seatunnel/connectors/seatunnel/typesense/serialize/sink/TypesenseRowSerializer.java b/seatunnel-connectors-v2/connector-typesense/src/main/java/org/apache/seatunnel/connectors/seatunnel/typesense/serialize/sink/TypesenseRowSerializer.java index 67e0ae17f8..6ea7e48e2f 100644 --- a/seatunnel-connectors-v2/connector-typesense/src/main/java/org/apache/seatunnel/connectors/seatunnel/typesense/serialize/sink/TypesenseRowSerializer.java +++ b/seatunnel-connectors-v2/connector-typesense/src/main/java/org/apache/seatunnel/connectors/seatunnel/typesense/serialize/sink/TypesenseRowSerializer.java @@ -48,7 +48,8 @@ public class TypesenseRowSerializer implements SeaTunnelRowSerializer { KeyExtractor.createKeyExtractor( seaTunnelRowType, collectionInfo.getPrimaryKeys(), - collectionInfo.getKeyDelimiter()); + collectionInfo.getKeyDelimiter(), + collectionInfo.getKeyEncoding()); } @Override @@ -70,12 +71,11 @@ public class TypesenseRowSerializer implements SeaTunnelRowSerializer { @Override public String serializeRowForDelete(SeaTunnelRow row) { String key = keyExtractor.apply(row); - Map<String, Object> document = toDocumentMap(row, seaTunnelRowType); - String id = document.get("id").toString(); if (StringUtils.isNotBlank(key)) { - id = key; + return key; } - return id; + Map<String, Object> document = toDocumentMap(row, seaTunnelRowType); + return document.get("id").toString(); } private Map<String, Object> toDocumentMap(SeaTunnelRow row, SeaTunnelRowType rowType) { diff --git a/seatunnel-connectors-v2/connector-typesense/src/main/java/org/apache/seatunnel/connectors/seatunnel/typesense/sink/TypesenseSinkFactory.java b/seatunnel-connectors-v2/connector-typesense/src/main/java/org/apache/seatunnel/connectors/seatunnel/typesense/sink/TypesenseSinkFactory.java index ecd78ffa4e..cf08d17da0 100644 --- a/seatunnel-connectors-v2/connector-typesense/src/main/java/org/apache/seatunnel/connectors/seatunnel/typesense/sink/TypesenseSinkFactory.java +++ b/seatunnel-connectors-v2/connector-typesense/src/main/java/org/apache/seatunnel/connectors/seatunnel/typesense/sink/TypesenseSinkFactory.java @@ -54,6 +54,7 @@ public class TypesenseSinkFactory implements TableSinkFactory { TypesenseSinkOptions.SCHEMA_SAVE_MODE, TypesenseSinkOptions.DATA_SAVE_MODE) .optional(TypesenseSinkOptions.PRIMARY_KEYS) .optional(TypesenseSinkOptions.KEY_DELIMITER) + .optional(TypesenseSinkOptions.KEY_ENCODING) .optional(TypesenseSinkOptions.MAX_BATCH_SIZE) .optional(TypesenseSinkOptions.MAX_RETRY_COUNT) .optional(SinkConnectorCommonOptions.MULTI_TABLE_SINK_REPLICA) diff --git a/seatunnel-connectors-v2/connector-typesense/src/test/java/org/apache/seatunnel/connectors/seatunnel/typesense/serializer/TypesenseRowSerializerTest.java b/seatunnel-connectors-v2/connector-typesense/src/test/java/org/apache/seatunnel/connectors/seatunnel/typesense/serializer/TypesenseRowSerializerTest.java index cdc7fe66c8..481e64a238 100644 --- a/seatunnel-connectors-v2/connector-typesense/src/test/java/org/apache/seatunnel/connectors/seatunnel/typesense/serializer/TypesenseRowSerializerTest.java +++ b/seatunnel-connectors-v2/connector-typesense/src/test/java/org/apache/seatunnel/connectors/seatunnel/typesense/serializer/TypesenseRowSerializerTest.java @@ -62,4 +62,54 @@ public class TypesenseRowSerializerTest { String data = "{\"name\":\"jack\",\"id\":\"0001\"}"; Assertions.assertEquals(typesenseRowSerializer.serializeRow(row), data); } + + @Test + public void testCompositeKeyEncodingPreservesLegacyIdsByDefault() { + SeaTunnelRowType schema = + new SeaTunnelRowType( + new String[] {"first", "second"}, + new SeaTunnelDataType[] {STRING_TYPE, STRING_TYPE}); + SeaTunnelRow left = new SeaTunnelRow(new Object[] {"1_1", "2"}); + SeaTunnelRow right = new SeaTunnelRow(new Object[] {"1", "1_2"}); + TypesenseRowSerializer legacy = serializer(schema, null); + TypesenseRowSerializer safe = serializer(schema, "LENGTH_PREFIXED"); + + Assertions.assertEquals("1_1_2", legacy.serializeRowForDelete(left)); + Assertions.assertEquals( + legacy.serializeRowForDelete(left), legacy.serializeRowForDelete(right)); + Assertions.assertEquals("3:1_11:2", safe.serializeRowForDelete(left)); + Assertions.assertEquals("1:13:1_2", safe.serializeRowForDelete(right)); + Assertions.assertTrue(safe.serializeRow(left).contains("\"id\":\"3:1_11:2\"")); + Assertions.assertTrue(safe.serializeRow(right).contains("\"id\":\"1:13:1_2\"")); + } + + @Test + public void testSingleKeyEncodingKeepsTheExistingId() { + SeaTunnelRowType schema = + new SeaTunnelRowType( + new String[] {"first", "second"}, + new SeaTunnelDataType[] {STRING_TYPE, STRING_TYPE}); + Map<String, Object> confMap = new HashMap<>(); + confMap.put(TypesenseBaseOptions.COLLECTION.key(), "id_probe"); + confMap.put(TypesenseSinkOptions.PRIMARY_KEYS.key(), Arrays.asList("first")); + confMap.put(TypesenseSinkOptions.KEY_ENCODING.key(), "LENGTH_PREFIXED"); + TypesenseRowSerializer serializer = + new TypesenseRowSerializer( + new CollectionInfo("id_probe", ReadonlyConfig.fromMap(confMap)), schema); + + Assertions.assertEquals( + "1_1", + serializer.serializeRowForDelete(new SeaTunnelRow(new Object[] {"1_1", "2"}))); + } + + private static TypesenseRowSerializer serializer(SeaTunnelRowType schema, String encoding) { + Map<String, Object> confMap = new HashMap<>(); + confMap.put(TypesenseBaseOptions.COLLECTION.key(), "id_probe"); + confMap.put(TypesenseSinkOptions.PRIMARY_KEYS.key(), Arrays.asList("first", "second")); + if (encoding != null) { + confMap.put(TypesenseSinkOptions.KEY_ENCODING.key(), encoding); + } + return new TypesenseRowSerializer( + new CollectionInfo("id_probe", ReadonlyConfig.fromMap(confMap)), schema); + } } diff --git a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-easysearch-e2e/src/test/java/org/apache/seatunnel/e2e/connector/easysearch/EasysearchIT.java b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-easysearch-e2e/src/test/java/org/apache/seatunnel/e2e/connector/easysearch/EasysearchIT.java index 8c279e65b7..5414057407 100644 --- a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-easysearch-e2e/src/test/java/org/apache/seatunnel/e2e/connector/easysearch/EasysearchIT.java +++ b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-easysearch-e2e/src/test/java/org/apache/seatunnel/e2e/connector/easysearch/EasysearchIT.java @@ -166,6 +166,27 @@ public class EasysearchIT extends TestSuiteBase implements TestResource { mapTestDatasetForDSL(), readSinkData())); } + @TestTemplate + public void testCompositeKeyEncodingAvoidsDocumentIdCollisions(TestContainer container) + throws Exception { + Container.ExecResult result = + container.executeJob("/easysearch/easysearch_composite_key_encoding.conf"); + Assertions.assertEquals(0, result.getExitCode()); + + Awaitility.given() + .ignoreExceptions() + .pollInterval(1L, TimeUnit.SECONDS) + .atMost(INDEX_REFRESH_TIMEOUT_SECONDS, TimeUnit.SECONDS) + .untilAsserted( + () -> + Assertions.assertEquals( + 2L, + easysearchClient + .getIndexDocsCount("st_id_encoding") + .get(0) + .getDocsCount())); + } + @TestTemplate public void testEasysearchWithSaveMode(TestContainer container) throws IOException, InterruptedException { diff --git a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-easysearch-e2e/src/test/resources/easysearch/easysearch_composite_key_encoding.conf b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-easysearch-e2e/src/test/resources/easysearch/easysearch_composite_key_encoding.conf new file mode 100644 index 0000000000..6044166315 --- /dev/null +++ b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-easysearch-e2e/src/test/resources/easysearch/easysearch_composite_key_encoding.conf @@ -0,0 +1,50 @@ +# +# 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. +# + +env { + parallelism = 1 + job.mode = "BATCH" +} + +source { + FakeSource { + schema { + fields { + first = string + second = string + marker = string + } + } + rows = [ + { kind = INSERT, fields = ["1_1", "2", "left"] }, + { kind = INSERT, fields = ["1", "1_2", "right"] } + ] + } +} + +sink { + Easysearch { + hosts = ["https://e2e_easysearch:9200";] + username = "admin" + password = "admin" + tls_verify_certificate = false + tls_verify_hostname = false + index = "st_id_encoding" + primary_keys = ["first", "second"] + key_encoding = "LENGTH_PREFIXED" + } +} diff --git a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-typesense-e2e/src/test/java/org/apache/seatunnel/e2e/connector/typesense/TypesenseIT.java b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-typesense-e2e/src/test/java/org/apache/seatunnel/e2e/connector/typesense/TypesenseIT.java index db1e6bc26c..c442aa9404 100644 --- a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-typesense-e2e/src/test/java/org/apache/seatunnel/e2e/connector/typesense/TypesenseIT.java +++ b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-typesense-e2e/src/test/java/org/apache/seatunnel/e2e/connector/typesense/TypesenseIT.java @@ -130,6 +130,15 @@ public class TypesenseIT extends TestSuiteBase implements TestResource { Assertions.assertEquals(typesenseClient.search(sinkCollection, null, 0).getFound(), 5); } + @TestTemplate + public void testCompositeKeyEncodingAvoidsDocumentIdCollisions(TestContainer container) + throws Exception { + Container.ExecResult result = + container.executeJob("/fake_to_typesense_composite_key_encoding.conf"); + Assertions.assertEquals(0, result.getExitCode()); + Assertions.assertEquals(2, typesenseClient.search(sinkCollection, null, 0).getFound()); + } + @TestTemplate public void testFakeToTypesenseWithRecreateSchema(TestContainer container) throws Exception { List<Field> fields = new ArrayList<>(); diff --git a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-typesense-e2e/src/test/resources/fake_to_typesense_composite_key_encoding.conf b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-typesense-e2e/src/test/resources/fake_to_typesense_composite_key_encoding.conf new file mode 100644 index 0000000000..9cd03af206 --- /dev/null +++ b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-typesense-e2e/src/test/resources/fake_to_typesense_composite_key_encoding.conf @@ -0,0 +1,47 @@ +# +# 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. +# + +env { + parallelism = 1 + job.mode = "BATCH" +} + +source { + FakeSource { + schema { + fields { + first = string + second = string + marker = string + } + } + rows = [ + { kind = INSERT, fields = ["1_1", "2", "left"] }, + { kind = INSERT, fields = ["1", "1_2", "right"] } + ] + } +} + +sink { + Typesense { + hosts = ["e2e_typesense:8108"] + collection = "typesense_test_collection" + api_key = "xyz" + primary_keys = ["first", "second"] + key_encoding = "LENGTH_PREFIXED" + } +}
