in cassandra-four-zero-bridge/src/main/java/org/apache/cassandra/cdc/msg/FourZeroCdcEventBuilder.java [197:249]
private void addColumn(List<Value> holder, ColumnData cd)
{
ColumnMetadata columnMetadata = cd.column();
String columnName = columnMetadata.name.toCQLString();
if (columnMetadata.isComplex()) // multi-cell column
{
ComplexColumnData complex = (ComplexColumnData) cd;
DeletionTime deletionTime = complex.complexDeletion();
if (deletionTime.isLive())
{
// the complex data is live, but there could be element deletion inside.
if (complex.column().type instanceof ListType)
{
// In the case of unfrozen lists, it reads the value from C*
readFromCassandra(holder, complex);
}
else
{
processComplexData(holder, complex);
}
}
else if (complex.cellsCount() > 0)
{
// The condition, complex data is not live && cellCount > 0, indicates that a new value is set to the column.
// The CQL operation could be either insert or update the column.
// Since the new value is in the mutation already, reading from C* can be skipped
processComplexData(holder, complex);
}
else // the entire multi-cell collection/UDT is deleted.
{
kind = CdcEvent.Kind.DELETE;
updateMaxTimestamp(deletionTime.markedForDeleteAt());
holder.add(makeValue(null, complex.column()));
}
}
else // simple column
{
Cell<?> cell = (Cell<?>) cd;
updateMaxTimestamp(cell.timestamp());
if (cell.isTombstone())
{
holder.add(makeValue(null, cell.column()));
}
else
{
holder.add(makeValue(cell.buffer(), cell.column()));
if (cell.isExpiring())
{
setTTL(cell.ttl(), cell.localDeletionTime());
}
}
}
}