in cassandra-four-zero-avro-converter/src/main/java/org/apache/cassandra/cdc/avro/CqlToAvroSchemaConverterImplementation.java [271:370]
private static Schema convertLiteralType(AbstractType<?> cqlType, String namespace)
{
Schema result;
if (cqlType instanceof AsciiType)
{
//ascii
result = SchemaBuilder.builder(namespace).stringType();
}
else if (cqlType instanceof LongType)
{
//big int
result = SchemaBuilder.builder(namespace).longType();
}
else if (cqlType instanceof BytesType)
{
//blob
result = SchemaBuilder.builder(namespace).bytesType();
}
else if (cqlType instanceof BooleanType)
{
// boolean
result = SchemaBuilder.builder(namespace).booleanType();
}
else if (cqlType instanceof SimpleDateType)
{
// Date with logical type date
result = SchemaBuilder.builder(namespace).intType();
LogicalTypes.date().addToSchema(result);
}
else if (cqlType instanceof DecimalType)
{
// fixed with logical type decimal
result = SchemaBuilder.builder(namespace).fixed(".fixed").size(16);
LogicalTypes.decimal(38, 19).addToSchema(result);
}
else if (cqlType instanceof DoubleType)
{
// double
result = SchemaBuilder.builder(namespace).doubleType();
}
else if (cqlType instanceof FloatType)
{
// Float
result = SchemaBuilder.builder(namespace).floatType();
}
else if (cqlType instanceof InetAddressType)
{
// Inet address
result = SchemaBuilder.builder(namespace).bytesType();
new LogicalType(INET_NAME).addToSchema(result);
}
else if (cqlType instanceof Int32Type || cqlType instanceof ShortType)
{
// int 32 & smallint
result = SchemaBuilder.builder(namespace).intType();
}
else if (cqlType instanceof UTF8Type)
{
// text & varchar
result = SchemaBuilder.builder(namespace).stringType();
}
else if (cqlType instanceof TimeType)
{
// time
result = SchemaBuilder.builder(namespace).longType();
}
else if (cqlType instanceof TimestampType)
{
// timestamp
result = SchemaBuilder.builder(namespace).longType();
LogicalTypes.timestampMicros().addToSchema(result);
}
else if (cqlType instanceof TimeUUIDType || cqlType instanceof UUIDType)
{
// timeuuid, uuid
result = SchemaBuilder.builder(namespace).stringType();
LogicalTypes.uuid().addToSchema(result);
}
else if (cqlType instanceof ByteType)
{
// tinyint
result = SchemaBuilder.builder(namespace).intType();
}
else if (cqlType instanceof IntegerType)
{
// varint
result = SchemaBuilder.builder(namespace).fixed(".fixed").size(16);
LogicalTypes.decimal(38, 0).addToSchema(result);
}
else if (cqlType instanceof DurationType || cqlType instanceof EmptyType || cqlType instanceof CounterColumnType)
{
throw new UnsupportedOperationException("Unsupported Cql data type " + cqlType.asCQL3Type());
// not supported
}
else
{
throw new RuntimeException("Unknown Cql datatype " + cqlType.asCQL3Type());
}
return result;
}