You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cassandra.apache.org by "jyothsnakonisa (via GitHub)" <gi...@apache.org> on 2026/10/13 21:52:57 UTC
Re: [PR] CASSANALYTICS-203: Fix UDT column deserialization in CDC [cassandra-analytics]
jyothsnakonisa commented on code in PR #252:
URL: https://github.com/apache/cassandra-analytics/pull/252#discussion_r4264429756
##########
cassandra-analytics-cdc-codec/src/main/java/org/apache/cassandra/cdc/avro/AvroBaseRecordTransformer.java:
##########
@@ -78,7 +78,7 @@ public GenericData.Record transform(CdcEvent event)
GenericData.Record update = new GenericData.Record(tableSchema);
CqlField.CqlType type = typeLookup.apply(KeyspaceTypeKey.of(event.keyspace, field.columnType));
- Object javaValue = type.deserializeToJavaType(fieldValue);
Review Comment:
Agreed. I'll file a follow-up JIRA for a dedicated interface with a non-null contract.
##########
cassandra-analytics-cdc-codec/src/main/java/org/apache/cassandra/cdc/avro/AvroDataUtils.java:
##########
@@ -56,6 +64,95 @@ private AvroDataUtils()
private static final Conversions.DecimalConversion decimalConversions = new Conversions.DecimalConversion();
+ /**
+ * Converts the values produced by {@link CqlField.CqlType#deserializeToType} into the Java values expected by
+ * {@link #toAvro(Object, Schema)}. Cassandra serializers leave UDTs as raw bytes, including inside collections
+ * and tuples, so they are deserialized into {@code Map<String, Object>} keyed by field name with
+ * {@link CqlField.CqlUdt#deserializeUdt}. All other values are returned unchanged.
+ */
+ private static final TypeConverter CDC_TYPE_CONVERTER = new TypeConverter()
+ {
+ @Override
+ @SuppressWarnings("unchecked")
+ public Object convert(CqlField.CqlType cqlType, Object value, boolean isFrozen)
+ {
+ if (value == null)
+ {
+ return null;
+ }
+ 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:
+ CqlField.CqlTuple tuple = (CqlField.CqlTuple) cqlType;
+ Object[] elements = value instanceof ByteBuffer
+ ? tuple.deserializeTuple((ByteBuffer) value, true)
Review Comment:
Not needed. Unlike `deserializeUdt`, the tuple branch converts each element with its own type right after `deserializeTuple`, so collections in a tuple are still converted (same as `SparkTuple`). Covered by `JavaTypeConverterTest#testUdtInCollectionInTuple`.
##########
cassandra-analytics-cdc-codec/src/main/java/org/apache/cassandra/cdc/avro/AvroDataUtils.java:
##########
@@ -56,6 +64,95 @@ private AvroDataUtils()
private static final Conversions.DecimalConversion decimalConversions = new Conversions.DecimalConversion();
+ /**
+ * Converts the values produced by {@link CqlField.CqlType#deserializeToType} into the Java values expected by
+ * {@link #toAvro(Object, Schema)}. Cassandra serializers leave UDTs as raw bytes, including inside collections
+ * and tuples, so they are deserialized into {@code Map<String, Object>} keyed by field name with
+ * {@link CqlField.CqlUdt#deserializeUdt}. All other values are returned unchanged.
+ */
+ private static final TypeConverter CDC_TYPE_CONVERTER = new TypeConverter()
Review Comment:
Done. Moved the converter into a new `JavaTypeConverter`; `AvroDataUtils` is unchanged from trunk.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org