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 0de0d44f CASSANALYTICS-204: CDC Avro schemas have invalid or colliding
record names for UDTs and collections (#253)
0de0d44f is described below
commit 0de0d44f471aa162b920e0bab241d67b9ac1bb9b
Author: Jyothsna konisa <[email protected]>
AuthorDate: Mon Oct 19 12:00:16 2026 -0700
CASSANALYTICS-204: CDC Avro schemas have invalid or colliding record names
for UDTs and collections (#253)
Patch by Jyothsna Konisa; Reviewed by Yifan Cai for CASSANALYTICS-204
---
CHANGES.txt | 1 +
.../cdc/avro/CqlToAvroSchemaConverterTest.java | 100 ++++++++++++++++++++-
.../CqlToAvroSchemaConverterImplementation.java | 65 ++++++++++----
3 files changed, 145 insertions(+), 21 deletions(-)
diff --git a/CHANGES.txt b/CHANGES.txt
index 1198a67e..02c2799f 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,5 +1,6 @@
0.5.0
-----
+ * CDC Avro schemas have invalid or colliding record names for UDTs and
collections (CASSANALYTICS-204)
* 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)
diff --git
a/cassandra-analytics-cdc-codec/src/test/java/org/apache/cassandra/cdc/avro/CqlToAvroSchemaConverterTest.java
b/cassandra-analytics-cdc-codec/src/test/java/org/apache/cassandra/cdc/avro/CqlToAvroSchemaConverterTest.java
index 4a12f943..fb309fbb 100644
---
a/cassandra-analytics-cdc-codec/src/test/java/org/apache/cassandra/cdc/avro/CqlToAvroSchemaConverterTest.java
+++
b/cassandra-analytics-cdc-codec/src/test/java/org/apache/cassandra/cdc/avro/CqlToAvroSchemaConverterTest.java
@@ -25,6 +25,8 @@ import java.net.URL;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Paths;
+import java.util.ArrayList;
+import java.util.List;
import java.util.Objects;
import java.util.function.Consumer;
@@ -165,8 +167,10 @@ public class CqlToAvroSchemaConverterTest
final Schema innerUdt = type.getElementType();
assertThat(innerUdt.getType()).as("Inner UDT type should be
RECORD").isEqualTo(RECORD);
- assertThat(AvroSchemas.isFrozen(innerUdt)).as("Inner UDT should be
frozen").isTrue();
-
assertThat(AvroSchemas.cqlType(AvroSchemas.unwrapNullable(innerUdt))).as("Inner
UDT CQL type").isEqualTo("frozen<udt_2>");
+ // UDT records only carry properties that do not depend on the
use; frozen is part of the list's CQL type
+ assertThat(innerUdt.getFullName()).as("Inner UDT full
name").isEqualTo("a.udt_2");
+ assertThat(AvroSchemas.isFrozen(innerUdt)).as("Inner UDT should
not carry the frozen flag").isFalse();
+
assertThat(AvroSchemas.cqlType(AvroSchemas.unwrapNullable(innerUdt))).as("Inner
UDT CQL type").isEqualTo("udt_2");
Schema id = innerUdt.getField("id").schema();
assertThat(id).as("Inner UDT id field should not be
null").isNotNull();
@@ -182,6 +186,98 @@ public class CqlToAvroSchemaConverterTest
assertThat(readCqlType(schema, "c2")).as("CQL type for 'c2'
field").isEqualTo("frozen<list<frozen<udt_2>>>");
}
+ @Test
+ public void testRecordNamesAreValidAndUnique()
+ {
+ String address = "CREATE TYPE address (city text);";
+ String phone = "CREATE TYPE phone (number int);";
+ String person = "CREATE TYPE person (home frozen<address>, mobile
frozen<phone>);";
+ String tagged = "CREATE TYPE tagged (tags frozen<map<text, int>>);";
+ String createStmt = decorateTable("CREATE TABLE %s (pk int PRIMARY
KEY, u frozen<address>, l list<frozen<address>>, " +
+ "s set<frozen<address>>, p
frozen<person>, m map<frozen<address>, frozen<phone>>, " +
+ "v map<int, frozen<address>>, t
map<text, text>, n map<int, frozen<map<int, int>>>, " +
+ "k map<frozen<map<int, text>>,
frozen<map<int, int>>>, g frozen<tagged>, w address, " +
+ "c list<frozen<map<text, text>>>, cs
set<frozen<map<text, text>>>);");
+ Schema schema = cqlToAvroSchemaConverter.convert("a", createStmt,
setOf(address, phone, person, tagged));
+ String ns = schema.getNamespace();
+
+ // UDT records are named <keyspace>.<udt> wherever the UDT is used
+ assertThat(columnType(schema,
"u").getFullName()).isEqualTo("a.address");
+ assertThat(columnType(schema,
"w").getFullName()).isEqualTo("a.address");
+ assertThat(columnType(schema,
"l").getElementType().getFullName()).isEqualTo("a.address");
+ assertThat(columnType(schema,
"s").getElementType().getFullName()).isEqualTo("a.address");
+ Schema personRecord = columnType(schema, "p");
+ assertThat(personRecord.getFullName()).isEqualTo("a.person");
+
assertThat(personRecord.getField("home").schema().getFullName()).isEqualTo("a.address");
+
assertThat(personRecord.getField("mobile").schema().getFullName()).isEqualTo("a.phone");
+ Schema mapEntry = columnType(schema, "m").getElementType();
+
assertThat(mapEntry.getField("key").schema().getFullName()).isEqualTo("a.address");
+
assertThat(mapEntry.getField("value").schema().getFullName()).isEqualTo("a.phone");
+ assertThat(columnType(schema,
"v").getElementType().getField("value").schema().getFullName()).isEqualTo("a.address");
+ // each UDT is defined once and referenced by name elsewhere
+ String json = schema.toString();
+ assertThat(countOccurrences(json,
"\"name\":\"address\"")).isEqualTo(1);
+ assertThat(countOccurrences(json, "\"name\":\"phone\"")).isEqualTo(1);
+ assertThat(countOccurrences(json, "\"name\":\"person\"")).isEqualTo(1);
+
+ // map entry records are named by their position; key and value are
added when both contain map entry records
+ assertThat(mapEntry.getFullName()).isEqualTo(ns +
".m.array_map.array_map");
+ assertThat(columnType(schema,
"t").getElementType().getFullName()).isEqualTo(ns + ".t.array_map.array_map");
+ assertThat(columnType(schema,
"n").getElementType().getField("value").schema().getElementType().getFullName())
+ .isEqualTo(ns + ".n.array_map.array_map.array_map");
+ Schema mapsEntry = columnType(schema, "k").getElementType();
+
assertThat(mapsEntry.getField("key").schema().getElementType().getFullName()).isEqualTo(ns
+ ".k.array_map.key.array_map.array_map");
+
assertThat(mapsEntry.getField("value").schema().getElementType().getFullName()).isEqualTo(ns
+ ".k.array_map.value.array_map.array_map");
+ // list and set elements use the collection's namespace
+ assertThat(columnType(schema,
"c").getElementType().getElementType().getFullName()).isEqualTo(ns +
".c.array_map.array_map");
+ assertThat(columnType(schema,
"cs").getElementType().getElementType().getFullName()).isEqualTo(ns +
".cs.array_map.array_map");
+ // records nested in UDT fields are named by the UDT
+ assertThat(columnType(schema,
"g").getField("tags").schema().getElementType().getFullName())
+ .isEqualTo("a.tagged.record_udt.tags.array_map.array_map");
+
+ // all names are valid, and the schema is unchanged after a JSON round
trip, which fails if two different
+ // records share a full name
+ List<Schema> records = new ArrayList<>();
+ collectRecords(schema, records);
+ assertThat(records).extracting(Schema::getFullName)
+ .allMatch(name ->
name.matches("[A-Za-z_][A-Za-z0-9_]*(\\.[A-Za-z_][A-Za-z0-9_]*)*"));
+ assertThat(new
Schema.Parser().parse(schema.toString())).isEqualTo(schema);
+ }
+
+ private static int countOccurrences(String text, String token)
+ {
+ int count = 0;
+ for (int index = text.indexOf(token); index >= 0; index =
text.indexOf(token, index + token.length()))
+ {
+ count++;
+ }
+ return count;
+ }
+
+ private static Schema columnType(Schema schema, String column)
+ {
+ return AvroSchemas.unwrapNullable(schema.getField(column).schema());
+ }
+
+ private static void collectRecords(Schema schema, List<Schema> records)
+ {
+ switch (schema.getType())
+ {
+ case RECORD:
+ records.add(schema);
+ schema.getFields().forEach(field ->
collectRecords(field.schema(), records));
+ break;
+ case ARRAY:
+ collectRecords(schema.getElementType(), records);
+ break;
+ case UNION:
+ schema.getTypes().forEach(type -> collectRecords(type,
records));
+ break;
+ default:
+ break;
+ }
+ }
+
@Test
public void testNestedMapType()
{
diff --git
a/cassandra-four-zero-avro-converter/src/main/java/org/apache/cassandra/cdc/avro/CqlToAvroSchemaConverterImplementation.java
b/cassandra-four-zero-avro-converter/src/main/java/org/apache/cassandra/cdc/avro/CqlToAvroSchemaConverterImplementation.java
index 0e4d5257..225e095d 100644
---
a/cassandra-four-zero-avro-converter/src/main/java/org/apache/cassandra/cdc/avro/CqlToAvroSchemaConverterImplementation.java
+++
b/cassandra-four-zero-avro-converter/src/main/java/org/apache/cassandra/cdc/avro/CqlToAvroSchemaConverterImplementation.java
@@ -185,36 +185,40 @@ public class CqlToAvroSchemaConverterImplementation
implements CqlToAvroSchemaCo
AbstractType<?> cqlType = columnCqlType.isReversed() ?
((ReversedType<?>) columnCqlType).baseType : columnCqlType;
Schema result;
- // If it's a collection
+ // If it's a collection. List and set elements use the collection's
namespace: a collection has a single element
+ // type, and Avro names cannot contain the CQL type (which is kept in
the cqlType property instead).
if (cqlType instanceof ListType)
{
- String collectionNamespace = namespace + '.' +
cqlType.asCQL3Type().toString();
result = SchemaBuilder.array()
- .items(schemaFrom(((ListType<?>)
cqlType).getElementsType(), collectionNamespace));
+ .items(schemaFrom(((ListType<?>)
cqlType).getElementsType(), namespace));
}
else if (cqlType instanceof SetType)
{
- String collectionNamespace = namespace + '.' +
cqlType.asCQL3Type().toString();
result = SchemaBuilder.array()
- .items(schemaFrom(((SetType<?>)
cqlType).getElementsType(), collectionNamespace));
+ .items(schemaFrom(((SetType<?>)
cqlType).getElementsType(), namespace));
AvroSchemas.flagArrayAsSet(result);
}
else if (cqlType instanceof MapType)
{
String arrayBasedMapName = AvroConstants.ARRAY_BASED_MAP_NAME;
String childNamespace = namespace + '.' + arrayBasedMapName;
+ MapType<?, ?> cqlMap = (MapType<?, ?>) cqlType;
+ // Map entry records nested in both the key and the value are
namespaced by key and value so they do not
+ // share a full name. Other maps keep their namespace, so their
existing schemas do not change.
+ boolean bothContainMaps = containsMap(cqlMap.getKeysType()) &&
containsMap(cqlMap.getValuesType());
+ String keyNamespace = bothContainMaps ? childNamespace + '.' +
ARRAY_BASED_MAP_KEY_NAME : childNamespace;
+ String valueNamespace = bothContainMaps ? childNamespace + '.' +
ARRAY_BASED_MAP_VALUE_NAME : childNamespace;
// create a new avro schema, array of records, to simulate a map.
SchemaBuilder.FieldAssembler<Schema> keyValue =
SchemaBuilder.builder()
.record(arrayBasedMapName)
.namespace(childNamespace)
.fields();
- MapType<?, ?> cqlMap = (MapType<?, ?>) cqlType;
// map keys and values are not nullable
keyValue.name(ARRAY_BASED_MAP_KEY_NAME)
- .type(schemaFrom(cqlMap.getKeysType(), childNamespace))
+ .type(schemaFrom(cqlMap.getKeysType(), keyNamespace))
.noDefault();
keyValue.name(ARRAY_BASED_MAP_VALUE_NAME)
- .type(schemaFrom(cqlMap.getValuesType(), childNamespace))
+ .type(schemaFrom(cqlMap.getValuesType(), valueNamespace))
.noDefault();
result = SchemaBuilder.builder()
.array()
@@ -223,25 +227,28 @@ public class CqlToAvroSchemaConverterImplementation
implements CqlToAvroSchemaCo
}
else if (cqlType instanceof UserType)
{
+ // A UDT record is named <keyspace>.<udt> wherever the UDT is
used, so Avro defines it once in the schema
+ // and references it by name elsewhere. Records nested in its
fields are namespaced by the UDT, not by its
+ // position; the record_udt segment keeps them apart from the
records of a table named like the UDT.
UserType userType = (UserType) cqlType;
- String recordBasedUdt = AvroConstants.RECORD_BASED_UDT_NAME;
- String childNamespace = namespace + '.' + recordBasedUdt;
- Schema[] ar = userType.fieldTypes().stream().map(udtType ->
schemaFrom(udtType, childNamespace)).toArray(Schema[]::new);
-
- // create a new avro schema, a single record, to represent the udt.
+ String udtName = userType.getNameAsString();
+ String fieldNamespace = userType.keyspace + '.' + udtName + '.' +
AvroConstants.RECORD_BASED_UDT_NAME;
SchemaBuilder.FieldAssembler<Schema> udtValue =
SchemaBuilder.builder()
-
.record(recordBasedUdt)
-
.namespace(childNamespace)
+
.record(udtName)
+
.namespace(userType.keyspace)
.fields();
-
- for (int i = 0; i < ar.length; i++)
+ for (int i = 0; i < userType.size(); i++)
{
- udtValue.name(userType.fieldNameAsString(i))
- .type(ar[i])
+ String fieldName = userType.fieldNameAsString(i);
+ udtValue.name(fieldName)
+ .type(schemaFrom(userType.fieldType(i), fieldNamespace
+ '.' + fieldName))
.noDefault();
}
result = udtValue.endRecord();
AvroSchemas.flagAsUdt(result);
+ // the record is shared by all uses of the UDT, so properties that
depend on the use (frozen, reversed) are not set
+ AvroSchemas.flagCqlType(result, udtName);
+ return result;
}
else
{
@@ -261,6 +268,26 @@ public class CqlToAvroSchemaConverterImplementation
implements CqlToAvroSchemaCo
return result;
}
+ /**
+ * @return whether the type is a map, or a list or set of maps (maps
nested in UDTs are namespaced by the UDT)
+ */
+ private static boolean containsMap(AbstractType<?> cqlType)
+ {
+ if (cqlType instanceof MapType)
+ {
+ return true;
+ }
+ if (cqlType instanceof ListType)
+ {
+ return containsMap(((ListType<?>) cqlType).getElementsType());
+ }
+ if (cqlType instanceof SetType)
+ {
+ return containsMap(((SetType<?>) cqlType).getElementsType());
+ }
+ return false;
+ }
+
/**
* Following link contains details about all cql types in cassandra.
* https://cassandra.apache.org/doc/latest/cassandra/cql/types.html
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
