Implementation:Risingwavelabs Risingwave CassandraUtil

Property	Value
File	java/connector-node/risingwave-sink-cassandra/src/main/java/com/risingwave/connector/CassandraUtil.java
Language	Java
Lines	174
Category	Utility
Package	com.risingwave.connector

Overview

CassandraUtil is a utility class providing static helper methods for the Cassandra sink connector. It handles three primary responsibilities: mapping RisingWave data types to corresponding Cassandra protocol types, validating schema compatibility between RisingWave table columns and Cassandra table metadata, and converting row values from RisingWave types to Cassandra-compatible Java objects. The class ensures type safety when sinking data from RisingWave to Cassandra.

Code Reference

Source Location

java/connector-node/risingwave-sink-cassandra/src/main/java/com/risingwave/connector/CassandraUtil.java

Signature

public class CassandraUtil {
    private static int getCorrespondingCassandraType(DataType dataType);
    public static void checkSchema(List<ColumnDesc> columnDescs, Map<CqlIdentifier, ColumnMetadata> cassandraColumnDescMap);
    public static void checkPrimaryKey(List<ColumnMetadata> cassandraColumnMetadatas, List<String> columnMetadatas);
    public static Object convertRow(Object value, TypeName typeName);
    public static String convertCQLIdentifiers(String identifier);
}

Imports

import com.datastax.oss.driver.api.core.CqlIdentifier;
import com.datastax.oss.driver.api.core.data.CqlDuration;
import com.datastax.oss.driver.api.core.metadata.schema.ColumnMetadata;
import com.risingwave.connector.api.ColumnDesc;
import com.risingwave.proto.Data.DataType;
import com.risingwave.proto.Data.DataType.TypeName;
import io.grpc.Status;
import java.nio.ByteBuffer;
import java.time.OffsetDateTime;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.stream.Collectors;

I/O Contract

Type Mapping (getCorrespondingCassandraType)

RisingWave Type	Cassandra Protocol Type
INT16	SMALLINT
INT32	INT
INT64	BIGINT
FLOAT	FLOAT
DOUBLE	DOUBLE
BOOLEAN	BOOLEAN
VARCHAR	VARCHAR
DECIMAL	DECIMAL
TIMESTAMPTZ	TIMESTAMP
DATE	DATE
TIME	TIME
BYTEA	BLOB
INTERVAL	DURATION
TIMESTAMP	(unsupported - throws exception)
LIST, STRUCT	(unimplemented - throws exception)

checkSchema

• Input: A list of RisingWave ColumnDesc objects and a map of Cassandra ColumnMetadata keyed by CqlIdentifier.
• Output: Void on success; throws gRPC FAILED_PRECONDITION if column count, column names, or data types do not match.

checkPrimaryKey

• Input: A list of Cassandra ColumnMetadata for primary key columns and a list of RisingWave primary key column names.
• Output: Void on success; throws gRPC FAILED_PRECONDITION if primary key length or column names do not match.

convertRow

• Input: A Java Object value and a RisingWave TypeName enum.
• Output: The value converted to a Cassandra-compatible Java type (e.g., OffsetDateTime to Instant, String to CqlDuration, byte[] to ByteBuffer).

convertCQLIdentifiers

• Input: A plain string identifier.
• Output: The identifier wrapped in double quotes for CQL safety.

Usage Examples

// Validate schema compatibility
CassandraUtil.checkSchema(tableSchema.getColumnDescs(), tableMetadata.getColumns());

// Validate primary keys for upsert sinks
CassandraUtil.checkPrimaryKey(tableMetadata.getPrimaryKey(), tableSchema.getPrimaryKeys());

// Convert a value for insertion
Object cassandraValue = CassandraUtil.convertRow(myValue, TypeName.TIMESTAMPTZ);

// Quote a CQL identifier
String quotedTable = CassandraUtil.convertCQLIdentifiers("my_table");
// Returns: "\"my_table\""

Related Pages

• CassandraConfig - Uses convertCQLIdentifiers for table name processing
• CassandraFactory - Calls checkSchema and checkPrimaryKey during validation