This is an automated email from the ASF dual-hosted git repository.
jyothsnakonisa pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/cassandra-analytics.git
The following commit(s) were added to refs/heads/trunk by this push:
new 6ca4f1cd CASSANALYTICS-203: Fix UDT column deserialization in CDC
(#252)
6ca4f1cd is described below
commit 6ca4f1cdb04e68826abebed1a6a16bda55b4044b
Author: Jyothsna konisa <[email protected]>
AuthorDate: Wed Oct 14 16:40:54 2026 -0700
CASSANALYTICS-203: Fix UDT column deserialization in CDC (#252)
Patch by Jyothsna Konisa; Reviewed by Yifan Cai for CASSANALYTICS-203
---
CHANGES.txt | 1 +
.../java/org/apache/cassandra/cdc/CdcLogMode.java | 3 +-
.../cdc/avro/AvroBaseRecordTransformer.java | 2 +-
.../apache/cassandra/cdc/avro/CdcEventUtils.java | 2 +-
.../cassandra/cdc/avro/JavaTypeConverter.java | 140 +++++++++++
.../cassandra/cdc/json/AvroJsonTransformer.java | 5 +-
.../cassandra/cdc/avro/AvroUdtTransformerTest.java | 255 +++++++++++++++++++++
.../cassandra/cdc/avro/JavaTypeConverterTest.java | 86 +++++++
8 files changed, 489 insertions(+), 5 deletions(-)
diff --git a/CHANGES.txt b/CHANGES.txt
index 092926ae..1198a67e 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,5 +1,6 @@
0.5.0
-----
+ * Fix UDT column deserialization in CDC (CASSANALYTICS-203)
* Skip CDC-irrelevant mutations on deserialization failure (CASSANALYTICS-196)
* Opening one keyspace instance rebuilds every keyspace instance in the
Cassandra 6.0 bridge (CASSANALYTICS-202)
* CDC logs NPE for deleted column values (CASSANALYTICS-178)
diff --git
a/cassandra-analytics-cdc-codec/src/main/java/org/apache/cassandra/cdc/CdcLogMode.java
b/cassandra-analytics-cdc-codec/src/main/java/org/apache/cassandra/cdc/CdcLogMode.java
index e89e2738..b6aea613 100644
---
a/cassandra-analytics-cdc-codec/src/main/java/org/apache/cassandra/cdc/CdcLogMode.java
+++
b/cassandra-analytics-cdc-codec/src/main/java/org/apache/cassandra/cdc/CdcLogMode.java
@@ -28,6 +28,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.cdc.api.KeyspaceTypeKey;
+import org.apache.cassandra.cdc.avro.JavaTypeConverter;
import org.apache.cassandra.cdc.msg.CdcEvent;
import org.apache.cassandra.cdc.msg.Value;
import org.apache.cassandra.spark.data.CqlField;
@@ -166,7 +167,7 @@ public enum CdcLogMode implements CdcLogger
try
{
CqlField.CqlType type =
typeLookup.apply(KeyspaceTypeKey.of(fieldValue.keyspace,
fieldValue.columnType));
- Object javaValue = type.deserializeToJavaType(value);
+ Object javaValue = JavaTypeConverter.deserialize(type, value);
return String.format("[%s : %s]", fieldValue.columnName,
javaValue);
}
catch (Throwable t)
diff --git
a/cassandra-analytics-cdc-codec/src/main/java/org/apache/cassandra/cdc/avro/AvroBaseRecordTransformer.java
b/cassandra-analytics-cdc-codec/src/main/java/org/apache/cassandra/cdc/avro/AvroBaseRecordTransformer.java
index 7c0ac173..c4cbd702 100644
---
a/cassandra-analytics-cdc-codec/src/main/java/org/apache/cassandra/cdc/avro/AvroBaseRecordTransformer.java
+++
b/cassandra-analytics-cdc-codec/src/main/java/org/apache/cassandra/cdc/avro/AvroBaseRecordTransformer.java
@@ -78,7 +78,7 @@ extends CdcEventAvroEncoder
GenericData.Record update = new GenericData.Record(tableSchema);
CqlField.CqlType type =
typeLookup.apply(KeyspaceTypeKey.of(event.keyspace, field.columnType));
- Object javaValue = type.deserializeToJavaType(fieldValue);
+ Object javaValue = JavaTypeConverter.deserialize(type, fieldValue);
update.put(field.columnName, AvroDataUtils.toAvro(javaValue,
column.schema()));
return ByteBuffer.wrap(encode(store.getWriter(event.keyspace + '.'
+ event.table, null), update));
};
diff --git
a/cassandra-analytics-cdc-codec/src/main/java/org/apache/cassandra/cdc/avro/CdcEventUtils.java
b/cassandra-analytics-cdc-codec/src/main/java/org/apache/cassandra/cdc/avro/CdcEventUtils.java
index fef91ab6..7c54df4c 100644
---
a/cassandra-analytics-cdc-codec/src/main/java/org/apache/cassandra/cdc/avro/CdcEventUtils.java
+++
b/cassandra-analytics-cdc-codec/src/main/java/org/apache/cassandra/cdc/avro/CdcEventUtils.java
@@ -323,7 +323,7 @@ public final class CdcEventUtils
"Encountered an unknown field during
event encoding. " +
"Field: %s. Avro schema: %s",
field.columnName, tableSchema.getFullName());
CqlField.CqlType type =
typeLookup.apply(KeyspaceTypeKey.of(event.keyspace, field.columnType));
- Object javaValue = type.deserializeToJavaType(value);
+ Object javaValue = JavaTypeConverter.deserialize(type, value);
update.put(field.columnName, AvroDataUtils.toAvro(javaValue,
column.schema()));
}
return new UpdatedEvent(update, truncatedFields);
diff --git
a/cassandra-analytics-cdc-codec/src/main/java/org/apache/cassandra/cdc/avro/JavaTypeConverter.java
b/cassandra-analytics-cdc-codec/src/main/java/org/apache/cassandra/cdc/avro/JavaTypeConverter.java
new file mode 100644
index 00000000..35715ad7
--- /dev/null
+++
b/cassandra-analytics-cdc-codec/src/main/java/org/apache/cassandra/cdc/avro/JavaTypeConverter.java
@@ -0,0 +1,140 @@
+/*
+ * 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.cassandra.cdc.avro;
+
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.LinkedHashMap;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import org.apache.avro.Schema;
+import org.apache.cassandra.spark.data.CqlField;
+import org.apache.cassandra.spark.data.TypeConverter;
+import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * {@link TypeConverter} that converts CQL values into the Java values
expected by
+ * {@link AvroDataUtils#toAvro(Object, Schema)}.
+ * <p>
+ * Cassandra serializers leave UDTs as raw bytes, including inside collections
and tuples. This converter
+ * deserializes them into {@code Map<String, Object>} keyed by field name with
{@link CqlField.CqlUdt#deserializeUdt},
+ * and converts the elements of collections and tuples recursively. All other
values are returned unchanged.
+ */
+public final class JavaTypeConverter implements TypeConverter
+{
+ public static final JavaTypeConverter INSTANCE = new JavaTypeConverter();
+
+ private JavaTypeConverter()
+ {
+ }
+
+ /**
+ * Deserializes a column value into the Java value expected by {@link
AvroDataUtils#toAvro(Object, Schema)}.
+ *
+ * @param type the CQL type of the column
+ * @param value the serialized column value
+ * @return the deserialized value, or null if the value is null
+ */
+ @Nullable
+ public static Object deserialize(CqlField.CqlType type, @Nullable
ByteBuffer value)
+ {
+ if (value == null)
+ {
+ return null;
+ }
+ // only types containing UDTs need conversion; avoid the overhead for
everything else
+ if (type.udts().isEmpty())
+ {
+ return type.deserializeToJavaType(value);
+ }
+ return type.deserializeToType(INSTANCE, value);
+ }
+
+ @Override
+ @SuppressWarnings("unchecked")
+ public Object convert(CqlField.CqlType cqlType, @NotNull Object value,
boolean isFrozen)
+ {
+ switch (cqlType.internalType())
+ {
+ case Frozen:
+ return convert(((CqlField.CqlFrozen) cqlType).inner(), value,
true);
+ case Udt:
+ return value instanceof ByteBuffer
+ ? ((CqlField.CqlUdt) cqlType).deserializeUdt(this,
(ByteBuffer) value, isFrozen)
+ : value;
+ case Tuple:
+ return convertTuple((CqlField.CqlTuple) cqlType, value);
+ case List:
+ CqlField.CqlType listElementType = ((CqlField.CqlCollection)
cqlType).type();
+ List<Object> list = new ArrayList<>(((Collection<Object>)
value).size());
+ for (Object element : (Collection<Object>) value)
+ {
+ list.add(convertNullable(listElementType, element));
+ }
+ return list;
+ case Set:
+ CqlField.CqlType setElementType = ((CqlField.CqlCollection)
cqlType).type();
+ Set<Object> set = new LinkedHashSet<>();
+ for (Object element : (Collection<Object>) value)
+ {
+ set.add(convertNullable(setElementType, element));
+ }
+ return set;
+ case Map:
+ CqlField.CqlMap mapType = (CqlField.CqlMap) cqlType;
+ Map<Object, Object> map = new LinkedHashMap<>();
+ for (Map.Entry<Object, Object> entry : ((Map<Object, Object>)
value).entrySet())
+ {
+ map.put(convertNullable(mapType.keyType(), entry.getKey()),
+ convertNullable(mapType.valueType(),
entry.getValue()));
+ }
+ return map;
+ default:
+ return value;
+ }
+ }
+
+ /**
+ * Tuple elements are deserialized by {@link
CqlField.CqlTuple#deserializeTuple} without a converter, so each
+ * element is converted here, which also handles collections and UDTs
nested in the tuple.
+ */
+ private Object[] convertTuple(CqlField.CqlTuple tuple, Object value)
+ {
+ Object[] elements = value instanceof ByteBuffer
+ ? tuple.deserializeTuple((ByteBuffer) value, true)
+ : (Object[]) value;
+ Object[] result = new Object[elements.length];
+ for (int index = 0; index < elements.length; index++)
+ {
+ result[index] = convertNullable(tuple.type(index),
elements[index]);
+ }
+ return result;
+ }
+
+ private Object convertNullable(CqlField.CqlType cqlType, @Nullable Object
value)
+ {
+ return value == null ? null : convert(cqlType, value, true);
+ }
+}
diff --git
a/cassandra-analytics-cdc-codec/src/main/java/org/apache/cassandra/cdc/json/AvroJsonTransformer.java
b/cassandra-analytics-cdc-codec/src/main/java/org/apache/cassandra/cdc/json/AvroJsonTransformer.java
index 270e6c9c..718d9bd6 100644
---
a/cassandra-analytics-cdc-codec/src/main/java/org/apache/cassandra/cdc/json/AvroJsonTransformer.java
+++
b/cassandra-analytics-cdc-codec/src/main/java/org/apache/cassandra/cdc/json/AvroJsonTransformer.java
@@ -28,6 +28,7 @@ import org.apache.cassandra.cdc.avro.CdcEventAvroEncoder;
import org.apache.cassandra.cdc.api.KeyspaceTypeKey;
import org.apache.cassandra.cdc.avro.AvroDataUtils;
import org.apache.cassandra.cdc.avro.CdcEventUtils;
+import org.apache.cassandra.cdc.avro.JavaTypeConverter;
import org.apache.cassandra.cdc.msg.CdcEvent;
import org.apache.cassandra.cdc.msg.Value;
import org.apache.cassandra.cdc.schemastore.SchemaStore;
@@ -51,7 +52,7 @@ public class AvroJsonTransformer extends CdcEventAvroEncoder
Schema tableSchema = store.getSchema(event.keyspace + '.' +
event.table, null);
GenericData.Record update = new GenericData.Record(tableSchema);
CqlField.CqlType type =
typeLookup.apply(KeyspaceTypeKey.of(event.keyspace, field.columnType));
- Object javaValue = type.deserializeToJavaType(field.getValue());
+ Object javaValue = JavaTypeConverter.deserialize(type,
field.getValue());
update.put(field.columnName, AvroDataUtils.toAvro(javaValue,
tableSchema.getField(field.columnName).schema()));
return ByteBuffer.wrap(encode(store.getWriter(event.keyspace + '.'
+ event.table, null), update));
};
@@ -67,7 +68,7 @@ public class AvroJsonTransformer extends CdcEventAvroEncoder
CdcEventUtils.updatedFields(event)
.forEach(field -> {
CqlField.CqlType type =
typeLookup.apply(KeyspaceTypeKey.of(event.keyspace, field.columnType));
- Object javaValue =
type.deserializeToJavaType(field.getValue());
+ Object javaValue =
JavaTypeConverter.deserialize(type, field.getValue());
update.put(field.columnName,
AvroDataUtils.toAvro(javaValue,
tableSchema.getField(field.columnName).schema()));
});
return update;
diff --git
a/cassandra-analytics-cdc-codec/src/test/java/org/apache/cassandra/cdc/avro/AvroUdtTransformerTest.java
b/cassandra-analytics-cdc-codec/src/test/java/org/apache/cassandra/cdc/avro/AvroUdtTransformerTest.java
new file mode 100644
index 00000000..5d582f1e
--- /dev/null
+++
b/cassandra-analytics-cdc-codec/src/test/java/org/apache/cassandra/cdc/avro/AvroUdtTransformerTest.java
@@ -0,0 +1,255 @@
+/*
+ * 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.cassandra.cdc.avro;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
+import java.util.Collections;
+import java.util.List;
+import java.util.UUID;
+import java.util.concurrent.TimeUnit;
+
+import org.junit.jupiter.api.Test;
+
+import org.apache.avro.Schema;
+import org.apache.avro.generic.GenericData;
+import org.apache.avro.generic.GenericDatumReader;
+import org.apache.avro.generic.GenericDatumWriter;
+import org.apache.avro.generic.GenericRecord;
+import org.apache.avro.io.DecoderFactory;
+import org.apache.cassandra.bridge.CassandraBridgeImplementation;
+import org.apache.cassandra.cdc.msg.CdcEvent;
+import org.apache.cassandra.cdc.msg.CdcEventBuilder;
+import org.apache.cassandra.cdc.msg.Value;
+import org.apache.cassandra.cdc.schemastore.SchemaStore;
+import org.apache.cassandra.spark.data.CqlTable;
+import org.apache.cassandra.spark.data.ReplicationFactor;
+import org.apache.cassandra.spark.data.partitioner.Partitioner;
+
+import static org.apache.cassandra.spark.utils.ArrayUtils.listOf;
+import static org.apache.cassandra.spark.utils.ArrayUtils.setOf;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Tests that CDC events with UDT columns are transformed into Avro records,
i.e. UDT values are
+ * published as nested Avro records instead of raw bytes.
+ */
+public class AvroUdtTransformerTest
+{
+ private static final String KEYSPACE = "avro_udt_transformer_ks";
+ private static final ReplicationFactor REPLICATION_FACTOR =
+ new
ReplicationFactor(ReplicationFactor.ReplicationStrategy.SimpleStrategy,
Collections.singletonMap("replication_factor", 1));
+
+ private final CassandraBridgeImplementation bridge = new
CassandraBridgeImplementation();
+
+ @Test
+ public void testListOfFrozenUdts() throws IOException
+ {
+ CqlTable table = registerTable("CREATE TABLE " + KEYSPACE + ".task (id
uuid PRIMARY KEY, name text, " +
+ "results list<frozen<task_result>>);",
+ "CREATE TYPE " + KEYSPACE +
".task_result (status text, message text);");
+ UUID id = UUID.randomUUID();
+ CdcEvent event = insert(table, id,
+ Value.of(KEYSPACE, "name", "text",
text("test-task")),
+ Value.of(KEYSPACE, "results",
"list<frozen<task_result>>",
+ list(fields(text("SUCCESS"),
text("completed")))));
+
+ GenericRecord payload = transformAndEncode(table, event);
+
+ assertThat(payload.get("id")).hasToString(id.toString());
+ assertThat(payload.get("name")).hasToString("test-task");
+ List<?> results = (List<?>) payload.get("results");
+ assertThat(results).hasSize(1);
+ GenericRecord result = (GenericRecord) results.get(0);
+ assertThat(result.get("status")).hasToString("SUCCESS");
+ assertThat(result.get("message")).hasToString("completed");
+ }
+
+ @Test
+ public void testListOfFrozenUdtsWithNestedUdt() throws IOException
+ {
+ CqlTable table = registerTable("CREATE TABLE " + KEYSPACE + ".command
(id uuid PRIMARY KEY, command_class text, " +
+ "execution_results
list<frozen<execution_result>>);",
+ "CREATE TYPE " + KEYSPACE +
".transient_event (id uuid, event_class text);",
+ "CREATE TYPE " + KEYSPACE +
".execution_result " +
+ "(domain_event frozen<transient_event>,
status text);");
+ UUID id = UUID.randomUUID();
+ UUID eventId = UUID.randomUUID();
+ CdcEvent event = insert(table, id,
+ Value.of(KEYSPACE, "command_class", "text",
text("TestCommand")),
+ Value.of(KEYSPACE, "execution_results",
"list<frozen<execution_result>>",
+ list(fields(fields(uuid(eventId),
text("TestEvent")), text("SUCCESS")))));
+
+ GenericRecord payload = transformAndEncode(table, event);
+
+ GenericRecord executionResult = (GenericRecord) ((List<?>)
payload.get("execution_results")).get(0);
+ GenericRecord domainEvent = (GenericRecord)
executionResult.get("domain_event");
+ assertThat(domainEvent.get("id")).hasToString(eventId.toString());
+ assertThat(domainEvent.get("event_class")).hasToString("TestEvent");
+ assertThat(executionResult.get("status")).hasToString("SUCCESS");
+ }
+
+ @Test
+ public void testFrozenUdtColumn() throws IOException
+ {
+ CqlTable table = registerTable("CREATE TABLE " + KEYSPACE + ".single
(id uuid PRIMARY KEY, result frozen<single_result>);",
+ "CREATE TYPE " + KEYSPACE +
".single_result (status text, code int);");
+ CdcEvent event = insert(table, UUID.randomUUID(),
+ Value.of(KEYSPACE, "result",
"frozen<single_result>",
+ fields(text("SUCCESS"),
ByteBuffer.allocate(4).putInt(0, 200))));
+
+ GenericRecord result = (GenericRecord) transformAndEncode(table,
event).get("result");
+
+ assertThat(result.get("status")).hasToString("SUCCESS");
+ assertThat(result.get("code")).isEqualTo(200);
+ }
+
+ @Test
+ public void testNonFrozenUdtColumn() throws IOException
+ {
+ CqlTable table = registerTable("CREATE TABLE " + KEYSPACE +
".multi_cell (id uuid PRIMARY KEY, result multi_cell_result);",
+ "CREATE TYPE " + KEYSPACE +
".multi_cell_result (status text, code int);");
+ // multi-cell UDT cells are packed with the cell count first
+ CdcEvent event = insert(table, UUID.randomUUID(),
+ Value.of(KEYSPACE, "result",
"multi_cell_result",
+ list(text("SUCCESS"),
ByteBuffer.allocate(4).putInt(0, 200))));
+
+ GenericRecord result = (GenericRecord) transformAndEncode(table,
event).get("result");
+
+ assertThat(result.get("status")).hasToString("SUCCESS");
+ assertThat(result.get("code")).isEqualTo(200);
+ }
+
+ /**
+ * Parses the table and registers its UDTs with the types used for type
lookup, as the CDC bridge does
+ */
+ private CqlTable registerTable(String createTable, String... createTypes)
+ {
+ CqlTable table = bridge.buildSchema(createTable, KEYSPACE,
REPLICATION_FACTOR, Partitioner.Murmur3Partitioner, setOf(createTypes));
+ table.udts().forEach(udt ->
bridge.cassandraTypes().updateUDTs(KEYSPACE, udt));
+ return table;
+ }
+
+ private static CdcEvent insert(CqlTable table, UUID id, Value...
valueColumns)
+ {
+ CdcEventBuilder builder = CdcEventBuilder.of(CdcEvent.Kind.INSERT,
KEYSPACE, table.table());
+ builder.setPartitionKeys(listOf(Value.of(KEYSPACE, "id", "uuid",
uuid(id))));
+ builder.setValueColumns(listOf(valueColumns));
+
builder.setMaxTimestampMicros(TimeUnit.MILLISECONDS.toMicros(System.currentTimeMillis()));
+ return builder.build();
+ }
+
+ /**
+ * Transforms the event, encodes and decodes the full CDC record, and
returns the decoded payload
+ */
+ private GenericRecord transformAndEncode(CqlTable table, CdcEvent event)
throws IOException
+ {
+ Schema tableSchema = new
CqlToAvroSchemaConverterImplementation(bridge).convert(table);
+ AvroGenericRecordTransformer transformer =
+ new AvroGenericRecordTransformer(new
SingleTableSchemaStore(tableSchema),
+ key ->
bridge.cassandraTypes().parseType(key.keyspace, key.type),
+ null);
+
+ GenericData.Record record = transformer.transform(event);
+
+ // encoding fails with a ClassCastException if UDT values are not Avro
records
+ byte[] encoded = AvroDataUtils.encode(new
GenericDatumWriter<>(record.getSchema()), record, null);
+ GenericRecord decoded = new
GenericDatumReader<GenericRecord>(record.getSchema())
+ .read(null,
DecoderFactory.get().binaryDecoder(encoded, null));
+ return (GenericRecord) decoded.get(AvroConstants.PAYLOAD_KEY);
+ }
+
+ private static ByteBuffer text(String value)
+ {
+ return ByteBuffer.wrap(value.getBytes(StandardCharsets.UTF_8));
+ }
+
+ private static ByteBuffer uuid(UUID value)
+ {
+ return ByteBuffer.allocate(16)
+ .putLong(0, value.getMostSignificantBits())
+ .putLong(8, value.getLeastSignificantBits());
+ }
+
+ // frozen UDT encoding: [int length][bytes] per field
+ private static ByteBuffer fields(ByteBuffer... values)
+ {
+ return concat(false, values);
+ }
+
+ // collection encoding: [int count] followed by [int length][bytes] per
element
+ private static ByteBuffer list(ByteBuffer... elements)
+ {
+ return concat(true, elements);
+ }
+
+ private static ByteBuffer concat(boolean withCount, ByteBuffer... values)
+ {
+ int size = withCount ? Integer.BYTES : 0;
+ for (ByteBuffer value : values)
+ {
+ size += Integer.BYTES + value.remaining();
+ }
+ ByteBuffer result = ByteBuffer.allocate(size);
+ if (withCount)
+ {
+ result.putInt(values.length);
+ }
+ for (ByteBuffer value : values)
+ {
+ result.putInt(value.remaining());
+ result.put(value.duplicate());
+ }
+ result.flip();
+ return result;
+ }
+
+ /**
+ * {@link SchemaStore} serving the Avro schema of a single table
+ */
+ private static class SingleTableSchemaStore implements SchemaStore
+ {
+ private final Schema schema;
+
+ SingleTableSchemaStore(Schema schema)
+ {
+ this.schema = schema;
+ }
+
+ @Override
+ public Schema getSchema(String namespace, String name)
+ {
+ return schema;
+ }
+
+ @Override
+ public GenericDatumWriter<GenericRecord> getWriter(String namespace,
String name)
+ {
+ return new GenericDatumWriter<>(schema);
+ }
+
+ @Override
+ public GenericDatumReader<GenericRecord> getReader(String namespace,
String name)
+ {
+ return new GenericDatumReader<>(schema);
+ }
+ }
+}
diff --git
a/cassandra-analytics-cdc-codec/src/test/java/org/apache/cassandra/cdc/avro/JavaTypeConverterTest.java
b/cassandra-analytics-cdc-codec/src/test/java/org/apache/cassandra/cdc/avro/JavaTypeConverterTest.java
new file mode 100644
index 00000000..75adeb0d
--- /dev/null
+++
b/cassandra-analytics-cdc-codec/src/test/java/org/apache/cassandra/cdc/avro/JavaTypeConverterTest.java
@@ -0,0 +1,86 @@
+/*
+ * 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.cassandra.cdc.avro;
+
+import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
+import java.util.Collections;
+import java.util.Map;
+
+import org.junit.jupiter.api.Test;
+
+import org.apache.cassandra.bridge.CassandraBridgeImplementation;
+import org.apache.cassandra.spark.data.CqlField;
+import org.apache.cassandra.spark.data.CqlTable;
+import org.apache.cassandra.spark.data.ReplicationFactor;
+import org.apache.cassandra.spark.data.partitioner.Partitioner;
+
+import static org.apache.cassandra.spark.utils.ArrayUtils.setOf;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Tests for {@link JavaTypeConverter}
+ */
+public class JavaTypeConverterTest
+{
+ private static final String KEYSPACE = "java_type_converter_ks";
+ private static final ReplicationFactor REPLICATION_FACTOR =
+ new
ReplicationFactor(ReplicationFactor.ReplicationStrategy.SimpleStrategy,
Collections.singletonMap("replication_factor", 1));
+
+ @Test
+ public void testUdtInCollectionInTuple()
+ {
+ // tuple elements are deserialized without a converter, so nested
collections and UDTs must still be converted
+ CqlTable table = new CassandraBridgeImplementation()
+ .buildSchema("CREATE TABLE " + KEYSPACE + ".t (id int
PRIMARY KEY, " +
+ "c frozen<tuple<int,
frozen<list<frozen<address>>>>>);",
+ KEYSPACE, REPLICATION_FACTOR,
Partitioner.Murmur3Partitioner,
+ setOf("CREATE TYPE " + KEYSPACE +
".address (city text);"));
+ CqlField.CqlType type = table.getField("c").type();
+ ByteBuffer address = concat(false,
ByteBuffer.wrap("Paris".getBytes(StandardCharsets.UTF_8)));
+ ByteBuffer serialized = concat(false, ByteBuffer.allocate(4).putInt(0,
7), concat(true, address));
+
+ Object[] value = (Object[]) JavaTypeConverter.deserialize(type,
serialized);
+
+ assertThat(value).containsExactly(7,
Collections.singletonList(Map.of("city", "Paris")));
+ }
+
+ // values encoded as [int length][bytes], optionally prefixed by the
number of values (collections)
+ private static ByteBuffer concat(boolean withCount, ByteBuffer... values)
+ {
+ int size = withCount ? Integer.BYTES : 0;
+ for (ByteBuffer value : values)
+ {
+ size += Integer.BYTES + value.remaining();
+ }
+ ByteBuffer result = ByteBuffer.allocate(size);
+ if (withCount)
+ {
+ result.putInt(values.length);
+ }
+ for (ByteBuffer value : values)
+ {
+ result.putInt(value.remaining());
+ result.put(value.duplicate());
+ }
+ result.flip();
+ return result;
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
