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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
