yifan-c commented on code in PR #252:
URL:
https://github.com/apache/cassandra-analytics/pull/252#discussion_r4262872419
##########
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:
The following is out of the scope of this PR – `typeLookup` should ensure
that return value is non-null or throw in its API contract. It is currently
missing from the `Function` interface. We should have a dedicated interface to
define the 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:
should `deserializeTuple` also take `TypeConverter` as an input for the
nested cases? For example, collections in tuple.
##########
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:
How about creating a dedicated class?
--
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]
