Support for collections 🗃️ #187

nikagra · 2025-12-09T14:52:15Z

Follow up to #21 which in turn is a continuation of #12

This pull request introduces support for collection types (LIST, SET, MAP) and User Defined Types (UDT) in the connector, both in the connector's logic and its configuration. It adds a new configuration option for how non-frozen collections are represented (currently only supporting "delta" mode), and updates the documentation to describe these changes. Additionally, the PR refactors schema and translation logic to handle the new types and cleans up configuration defaults.

Support for collections and UDTs:

Added support for both frozen and non-frozen collections (LIST, SET, MAP) and UDT columns, including their representation in Kafka Connect schemas and JSON output. The connector now translates these types from ScyllaDB to Kafka Connect records, handling complex nested structures. [1] [2] [3]
Introduced logic to handle non-frozen collections using a new "delta" mode, where changes to collections are represented as a struct with mode and elements fields, allowing for granular tracking of modifications, overwrites, and removals. [1] [2]

Connector configuration enhancements:

Added a new configuration option scylla.collections.mode (with enum type CollectionsMode), allowing users to specify how non-frozen collections are represented. The default is "delta" mode, and this is now parsed and exposed via the connector config. [1] [2] [3]
Cleaned up configuration defaults for several pooling and retry options, removing unnecessary .optional() calls and ensuring default values are always set. [1] [2] [3] [4] [5] [6] [7]

Documentation updates:

Updated the README.md to document the new support for collections and UDTs, including detailed descriptions of the JSON representation for both frozen and non-frozen collections, and the meaning of the new configuration option. [1] [2]

Codebase refactoring and type safety:

Refactored schema and translation logic to use the new Field class instead of Cell, and updated methods to support recursive translation of nested types. Improved type safety and code clarity in the process. [1] [2] [3] [4]

New enums for clarity:

Introduced new enums CollectionOperation and CollectionsMode to clearly represent collection change operations and configuration modes, improving code readability and maintainability. [1] [2]

Relevant issues:

Known limitation: non-frozen collections []/{} vs NULL on INSERT #201

Copilot

Pull request overview

This PR adds comprehensive support for Scylla collection types (LIST, SET, MAP, TUPLE, UDT) to the Debezium CDC connector. The implementation enables both frozen and non-frozen collection handling, with frozen collections being fully supported and non-frozen collections supported through a "delta" mode configuration option.

Key changes:

Extended schema computation to recursively handle collection types with proper Kafka Connect schema mappings
Implemented recursive field translation logic to convert Scylla collection data to Kafka format
Added configuration option for collection modes (currently only DELTA mode)

Reviewed changes

Copilot reviewed 5 out of 5 changed files in this pull request and generated 6 comments.

Show a summary per file

File	Description
ScyllaSchema.java	Adds recursive schema computation for collections, updates `isSupportedColumnSchema` to detect frozen vs non-frozen collections using CDC column patterns
ScyllaChangeRecordEmitter.java	Refactors from Cell to Field API, implements recursive `translateFieldToKafka` for all collection types
ScyllaConnectorConfig.java	Adds `COLLECTIONS_MODE` configuration field with DELTA as default value
CollectionsMode.java	New enum defining collection handling modes (currently only DELTA)
README.md	Documents collection support including frozen/non-frozen formats and delta mode behavior

Comments suppressed due to low confidence (1)

src/main/java/com/scylladb/cdc/debezium/connector/ScyllaChangeRecordEmitter.java:199

Variable keyStruct may be null at this access because of this null argument.
Variable keyStruct may be null at this access because of this null argument.

        keyStruct.put(cdef.getColumnName(), value);

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

src/main/java/com/scylladb/cdc/debezium/connector/ScyllaSchema.java

src/main/java/com/scylladb/cdc/debezium/connector/ScyllaChangeRecordEmitter.java

README.md

src/main/java/com/scylladb/cdc/debezium/connector/ScyllaChangeRecordEmitter.java

Copilot

Pull request overview

Copilot reviewed 9 out of 9 changed files in this pull request and generated 9 comments.

Copilot · 2025-12-22T19:51:22Z

src/main/java/com/scylladb/cdc/debezium/connector/ScyllaChangeRecordEmitter.java

+      if (cdef.getBaseTableColumnKind() == ColumnKind.PARTITION_KEY
+          || cdef.getBaseTableColumnKind() == ColumnKind.CLUSTERING_KEY) {


Inconsistent use of getBaseTableColumnKind() and ColumnKind here compared to getBaseTableColumnType() and ColumnType used throughout ScyllaSchema.java. For consistency and maintainability, the same API should be used across the codebase. Check if getBaseTableColumnKind() is the newer API and update ScyllaSchema.java accordingly, or revert to using getBaseTableColumnType() here.

Copilot · 2025-12-22T19:51:22Z

src/main/java/com/scylladb/cdc/debezium/connector/ScyllaChangeRecordEmitter.java

-        }
+        keyStruct.put(cdef.getColumnName(), value);
      } else {
+        Schema cellSchema = schema.cellSchema(cdef.getColumnName());


Potential NullPointerException: schema.cellSchema(cdef.getColumnName()) can return null if the column is not found in the cellSchemas map. The code immediately attempts to access cellSchema.field(ScyllaSchema.CELL_VALUE) on line 216 without checking for null. Add a null check after line 208 or ensure that cellSchema will always be present for supported columns.

Suggested change

Schema cellSchema = schema.cellSchema(cdef.getColumnName());

Schema cellSchema = schema.cellSchema(cdef.getColumnName());

if (cellSchema == null) {

// No schema available for this column; skip to avoid NullPointerException.

continue;

}

Copilot · 2025-12-22T19:51:22Z

src/main/java/com/scylladb/cdc/debezium/connector/ScyllaConnectorConfig.java

+          .withWidth(ConfigDef.Width.SHORT)
+          .withImportance(ConfigDef.Importance.MEDIUM)
+          .withDescription(
+              "How to represent non-frozen collections. Currently, only 'delta' mode is supported - in the future support for more modes may be added. 'Delta' mode: change in collection is represented as a struct with 2 fields, 'mode' and 'elements'. 'mode' describes what type of change happened (modifying collection, overwriting collection), 'elements' contains added/removed elements.");


The description for scylla.collections.mode is quite lengthy for a configuration field. Consider breaking it into multiple sentences or shortening it for better readability. For example: "Specifies how non-frozen collections are represented. Currently only 'delta' mode is supported. In delta mode, collection changes are represented as a struct with 'mode' and 'elements' fields, where 'mode' indicates the type of change (modify or overwrite) and 'elements' contains the added or removed elements."

Suggested change

"How to represent non-frozen collections. Currently, only 'delta' mode is supported - in the future support for more modes may be added. 'Delta' mode: change in collection is represented as a struct with 2 fields, 'mode' and 'elements'. 'mode' describes what type of change happened (modifying collection, overwriting collection), 'elements' contains added/removed elements.");

"Specifies how non-frozen collections are represented. Currently, only 'delta' mode is supported. "

+ "In delta mode, collection changes are represented as a struct with 'mode' and 'elements' fields, "

+ "where 'mode' indicates the type of change (modify or overwrite) and 'elements' contains the added or removed elements.");

Copilot · 2025-12-22T19:51:22Z

src/main/java/com/scylladb/cdc/debezium/connector/ScyllaConnectorConfig.java

+  /*
+   * Scylla CDC Source Connector relies on heartbeats to move the offset,
+   * because the offset determines if the generation ended, therefore HEARTBEAT_INTERVAL
+   * should be positive (0 would disable heartbeats) and a default value is changed
+   * (previously 0).
+   */
+


This comment block about HEARTBEAT_INTERVAL appears to be misplaced. It's positioned between the COLLECTIONS_MODE field (lines 119-127) and the USER field (lines 136-142), but the actual CUSTOM_HEARTBEAT_INTERVAL field is defined much later at line 249. Either remove this duplicate comment (since the same comment appears at line 243-247 right before the field definition) or move it to immediately precede the CUSTOM_HEARTBEAT_INTERVAL field definition.

Suggested change

/*

* Scylla CDC Source Connector relies on heartbeats to move the offset,

* because the offset determines if the generation ended, therefore HEARTBEAT_INTERVAL

* should be positive (0 would disable heartbeats) and a default value is changed

* (previously 0).

*/

Copilot · 2025-12-22T19:51:23Z

src/main/java/com/scylladb/cdc/debezium/connector/ScyllaChangeRecordEmitter.java

+            deletedKeys.forEach(
+                (key) -> {
+                  addedElements.put(key, null);
+                });


The lambda expression can be simplified. The explicit parameter parentheses and braces are unnecessary for a single-statement lambda with a single parameter. Consider simplifying to: deletedKeys.forEach(key -> addedElements.put(key, null));

Suggested change

deletedKeys.forEach(

(key) -> {

addedElements.put(key, null);

});

deletedKeys.forEach(key -> addedElements.put(key, null));

Copilot · 2025-12-22T19:51:23Z

src/main/java/com/scylladb/cdc/debezium/connector/ScyllaChangeRecordEmitter.java

+          Short index = -1;
+          for (Map.Entry<String, Field> element : elementsMap.entrySet()) {
+            index++;
+            if ((!element.getValue().isNull()) || deletedKeys.contains(index)) {


The initialization of index to -1 followed by immediate pre-increment appears confusing. Consider initializing to 0 and using post-increment, or add a comment explaining why -1 is used (e.g., "Start at -1 because we increment before processing each element").

Suggested change

Short index = -1;

for (Map.Entry<String, Field> element : elementsMap.entrySet()) {

index++;

if ((!element.getValue().isNull()) || deletedKeys.contains(index)) {

Short index = 0;

for (Map.Entry<String, Field> element : elementsMap.entrySet()) {

if ((!element.getValue().isNull()) || deletedKeys.contains(index++)) {

Copilot · 2025-12-22T19:51:23Z

README.md

+Each non-frozen collection column is represented as a struct, with fields `mode` and `elements`. This struct will be stored in "Cell" described previously.
+`mode` can be:
+- `MODIFY` - elements were added or deleted.
+- `OVERWRITE` - whole content of collection was removed, and new elements were added. If no elements were added (meaning the collection was just removed), this mode won't be used - instead, whole struct (stored in `field` value of "Cell" struct, as mentioned previously) will be null.


Typo: "field" should be "value" for consistency. The text reads "stored in field value of 'Cell' struct" but should be "stored in value field of 'Cell' struct" to match the actual field name and maintain consistency with other references.

Suggested change

- `OVERWRITE` - whole content of collection was removed, and new elements were added. If no elements were added (meaning the collection was just removed), this mode won't be used - instead, whole struct (stored in `field` value of "Cell" struct, as mentioned previously) will be null.

- `OVERWRITE` - whole content of collection was removed, and new elements were added. If no elements were added (meaning the collection was just removed), this mode won't be used - instead, whole struct (stored in `value` field of "Cell" struct, as mentioned previously) will be null.

Copilot · 2025-12-22T19:51:23Z

src/test/java/com/scylladb/cdc/debezium/connector/KafkaUtils.java

+   * Parses a Debezium-style JSON envelope and extracts a collection of strings from the given field
+   * under the {@code after} section. Expects the structure:


The Javadoc is misleading. It states "extracts a collection of strings" but the method is generic and uses a Function<JsonNode, T> mapper to extract elements of any type T, not just strings. Update the documentation to reflect this, for example: "Parses a Debezium-style JSON envelope and extracts a list of elements from the given field under the {@code after} section, applying the provided mapper function to each element."

Suggested change

* Parses a Debezium-style JSON envelope and extracts a collection of strings from the given field

* under the {@code after} section. Expects the structure:

* Parses a Debezium-style JSON envelope and extracts a list of elements from the given field

* under the {@code after} section, applying the provided mapper function to each element.

* Expects the structure:

Copilot · 2025-12-22T19:51:24Z

src/main/java/com/scylladb/cdc/debezium/connector/ScyllaChangeRecordEmitter.java

-        if (keyStruct != null) {
-          keyStruct.put(cdef.getColumnName(), value);
-        }
+        keyStruct.put(cdef.getColumnName(), value);


Variable keyStruct may be null at this access because of this null argument.
Variable keyStruct may be null at this access because of this null argument.

Suggested change

keyStruct.put(cdef.getColumnName(), value);

if (keyStruct != null) {

keyStruct.put(cdef.getColumnName(), value);

}

Add support for including frozen lists in generated changes. Made necessary changes to support nested data types.

Add support for including frozen sets in generated changes.

Add support for including frozen maps in generated changes.

Add support for including tuples in generated changes. For a tuple, a Kafka Connect struct is created with "tuple_member_*" for each member of a tuple (as they can have different data types inside).

Add support for including frozen UDTs in generated changes.

…ed error handling and configuration options. Fix formatting issues. Co-authored-by: Copilot <[email protected]>

…g connector registration in integration tests.

…egistration

…ests for list, set, and map collections

…ns and update integration tests for list, set, and map types

…epresentation and nested field handling

dkropachev · 2025-12-20T15:39:46Z

src/main/java/com/scylladb/cdc/debezium/connector/ScyllaSchema.java


-  protected static boolean isSupportedColumnSchema(ChangeSchema.ColumnDefinition cdef) {
+  protected static boolean isSupportedColumnSchema(
+      ChangeSchema changeSchema, ChangeSchema.ColumnDefinition cdef) {


don't seemse like changeSchema is used at all

dkropachev · 2025-12-20T16:18:03Z

src/main/java/com/scylladb/cdc/debezium/connector/CollectionsMode.java

+package com.scylladb.cdc.debezium.connector;
+
+public enum CollectionsMode {
+  DELTA,


It needs some explanation.
Also I feel like we should not have it, and instead have one configuration that does it for whole data.
For instance, we already have experimental.preimages.enabled that makes it create full data record if pre-mage is turned on, so probably we should follow that one, instead of having a sperate config.

…ions and add example message structures

nikagra force-pushed the feat/non-frozen-collections branch from a4c715a to 1821b94 Compare December 9, 2025 17:20

nikagra changed the title ~~Feat/non frozen collections~~ Support for collections 🗃️ Dec 9, 2025

nikagra force-pushed the feat/non-frozen-collections branch from 3afb149 to d4fe276 Compare December 9, 2025 18:36

nikagra requested a review from Copilot December 10, 2025 12:08

Copilot started reviewing on behalf of nikagra December 10, 2025 12:08 View session

Copilot AI reviewed Dec 10, 2025

View reviewed changes

nikagra force-pushed the feat/non-frozen-collections branch 3 times, most recently from 9022a4b to a5aa1d3 Compare December 16, 2025 16:45

nikagra force-pushed the feat/non-frozen-collections branch from a5aa1d3 to 58485b6 Compare December 17, 2025 20:01

mykaul mentioned this pull request Dec 18, 2025

feature request: support for collection types (LIST, SET, MAP) and UDT #9

Open

nikagra force-pushed the feat/non-frozen-collections branch 6 times, most recently from 15f0b41 to 6f80092 Compare December 22, 2025 19:28

nikagra marked this pull request as ready for review December 22, 2025 19:45

nikagra requested review from Copilot and dkropachev December 22, 2025 19:45

Copilot started reviewing on behalf of nikagra December 22, 2025 19:46 View session

Copilot AI reviewed Dec 22, 2025

View reviewed changes

avelanarius and others added 8 commits December 29, 2025 18:45

Add support for frozen lists

64098e6

Add support for including frozen lists in generated changes. Made necessary changes to support nested data types.

Add support for frozen sets

f60264b

Add support for including frozen sets in generated changes.

Add support for frozen maps

28fe616

Add support for including frozen maps in generated changes.

Add support for tuples

ed2fb50

Add support for including tuples in generated changes. For a tuple, a Kafka Connect struct is created with "tuple_member_*" for each member of a tuple (as they can have different data types inside).

Add support for frozen UDTs

16a9ddc

Add support for including frozen UDTs in generated changes.

Fix: allow null sets/maps/lists

8f0b939

Add option with collections representation

a011784

Avoid generating schemas multiple times

562c479

Lorak-mmk and others added 8 commits December 29, 2025 18:45

Support for non-frozen collections

1ae90b9

Describe collections format in README.md

5fa91c2

Enhance README and code to support non-frozen collections with improv…

3b535ba

…ed error handling and configuration options. Fix formatting issues. Co-authored-by: Copilot <[email protected]>

Reverting changes omitted during rebase. Improve error handling durin…

e602349

…g connector registration in integration tests.

Integration tests to support frozen collections including edge cases.

d8889c9

Add integration test for edge cases with frozen collections

5698b26

Add integration tests for UDT and complex types, refactor connector r…

845ff61

…egistration

Refactor JSON extraction logic in KafkaUtils and update integration t…

428e5c7

…ests for list, set, and map collections

nikagra force-pushed the feat/non-frozen-collections branch from 6f80092 to 428e5c7 Compare December 29, 2025 17:45

nikagra added 2 commits December 30, 2025 18:57

Enhance support for non-frozen collections: refactor schema definitio…

9a66e60

…ns and update integration tests for list, set, and map types

Update documentation for non-frozen collections: clarify delta mode r…

3949238

…epresentation and nested field handling

nikagra force-pushed the feat/non-frozen-collections branch from 0606283 to 3949238 Compare December 30, 2025 17:57

dkropachev reviewed Dec 30, 2025

View reviewed changes

Refactor integration tests for non-frozen collections: enhance assert…

d568447

…ions and add example message structures

		if (cdef.getBaseTableColumnKind() == ColumnKind.PARTITION_KEY
		\|\| cdef.getBaseTableColumnKind() == ColumnKind.CLUSTERING_KEY) {

-        Schema cellSchema = schema.cellSchema(cdef.getColumnName());
+        Schema cellSchema = schema.cellSchema(cdef.getColumnName());
+        if (cellSchema == null) {
+          // No schema available for this column; skip to avoid NullPointerException.
+          continue;
+        }

-              "How to represent non-frozen collections. Currently, only 'delta' mode is supported - in the future support for more modes may be added. 'Delta' mode: change in collection is represented as a struct with 2 fields, 'mode' and 'elements'. 'mode' describes what type of change happened (modifying collection, overwriting collection), 'elements' contains added/removed elements.");
+              "Specifies how non-frozen collections are represented. Currently, only 'delta' mode is supported. "
+                  + "In delta mode, collection changes are represented as a struct with 'mode' and 'elements' fields, "
+                  + "where 'mode' indicates the type of change (modify or overwrite) and 'elements' contains the added or removed elements.");

	- `OVERWRITE` - whole content of collection was removed, and new elements were added. If no elements were added (meaning the collection was just removed), this mode won't be used - instead, whole struct (stored in `field` value of "Cell" struct, as mentioned previously) will be null.
	- `OVERWRITE` - whole content of collection was removed, and new elements were added. If no elements were added (meaning the collection was just removed), this mode won't be used - instead, whole struct (stored in `value` field of "Cell" struct, as mentioned previously) will be null.

		* Parses a Debezium-style JSON envelope and extracts a collection of strings from the given field
		* under the {@code after} section. Expects the structure:

-   * Parses a Debezium-style JSON envelope and extracts a collection of strings from the given field
-   * under the {@code after} section. Expects the structure:
+   * Parses a Debezium-style JSON envelope and extracts a list of elements from the given field
+   * under the {@code after} section, applying the provided mapper function to each element.
+   * Expects the structure:

Support for collections 🗃️ #187

Are you sure you want to change the base?

Support for collections 🗃️ #187

Uh oh!

Conversation

nikagra commented Dec 9, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Relevant issues:

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Reviewed changes

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Uh oh!

Copilot AI Dec 22, 2025

Choose a reason for hiding this comment

Uh oh!

Copilot AI Dec 22, 2025

Choose a reason for hiding this comment

Uh oh!

Copilot AI Dec 22, 2025

Choose a reason for hiding this comment

Uh oh!

Copilot AI Dec 22, 2025

Choose a reason for hiding this comment

Uh oh!

Copilot AI Dec 22, 2025

Choose a reason for hiding this comment

Uh oh!

Copilot AI Dec 22, 2025

Choose a reason for hiding this comment

Uh oh!

Copilot AI Dec 22, 2025

Choose a reason for hiding this comment

Uh oh!

Copilot AI Dec 22, 2025

Choose a reason for hiding this comment

Uh oh!

Copilot AI Dec 22, 2025

Choose a reason for hiding this comment

Uh oh!

dkropachev Dec 20, 2025

Choose a reason for hiding this comment

Uh oh!

dkropachev Dec 20, 2025

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

nikagra commented Dec 9, 2025 •

edited

Loading