[martin-g]

Note: This PR is based on top of #753

Add a Cassandra table provider

Adds datafusion-table-providers-cassandra, a DataFusion TableProvider for Apache Cassandra. It is built on the ScyllaDB Rust driver (scylla 1.9), so it also works with ScyllaDB. The crate is exposed from datafusion-table-providers behind the new cassandra feature.

Features

Connection pool (CassandraConnectionPool)

• Wraps one shared driver Session, which manages its own connections. CassandraConnection is a cheap handle to it.
• Parameters:
  • hosts: comma-separated contact points, optionally as host:port. host is accepted as an alias.
  • port: defaults to 9042.
  • keyspace: the keyspace unqualified table names are resolved in.
  • user / password
  • local_datacenter: the datacenter to prefer when routing requests.
  • connection_timeout_ms
• CassandraConnectionPool::from_session builds a pool from an existing driver session.
• Supports the predefined TableReference -> SchemaRef map, which skips schema inference.
• Lists keyspaces (schemas()) and tables (tables()) from system_schema.
• A missing table returns an UndefinedTable error.

Reading (CassandraTableFactory::table_provider)

• The schema comes from the metadata of a prepared SELECT *, so no table data is read to infer it.
• Table names are case-sensitive and accepted as table or keyspace.table. Identifiers are always quoted.
• Scans run as CassandraExec. The results are streamed from Cassandra's pages and converted to Arrow record batches.
• EXPLAIN shows the CQL that is run, e.g. CassandraExec cql=SELECT "name" FROM "ks"."users" WHERE "id" IN (1, 3) LIMIT 10.

Pushdown

CQL has no joins or aggregations, and without ALLOW FILTERING it can only filter on key columns. So only these are pushed down to Cassandra:

• Projection. SELECT count(*) selects a single key column.
• LIMIT. LIMIT 0 doesn't query Cassandra at all.
• Filters on key columns that make up a valid CQL WHERE clause. These are reported as TableProviderFilterPushDown::Exact:
  • = or IN on every partition key column. A short IN list that DataFusion rewrote to a = 1 OR a = 2 is also accepted.
  • then = on the leading clustering columns, and optionally a range (<, <=, >, >=) on the next one.
  • A filter is pushed down only for types that Cassandra compares the same way as DataFusion: integers, text/varchar/ascii, boolean, blob, date, timestamp and time. uuid/timeuuid support only =, with the canonical lowercase value.
  • Literals Cassandra would reject, such as an empty partition key or non-ASCII text for an ascii column, are left to DataFusion.
• All other filters are Unsupported and evaluated by DataFusion: non-key columns, partial partition keys, !=, NOT IN, duplicate restrictions on the same column.
• If scan() receives filters that together don't form a valid WHERE clause, it checks them again. DataFusion decides on filters separately in each optimizer pass, so this can happen. The filters that can't be pushed down are applied by a FilterExec over the scan.

Type mapping

CQL	Arrow
ascii, text/varchar	Utf8
tinyint, smallint, int, bigint, counter	Int8, Int16, Int32, Int64
float, double	Float32, Float64
boolean	Boolean
blob	Binary
date	Date32
time	Time64(Nanosecond)
timestamp	Timestamp(Millisecond, "UTC")
duration	Interval(MonthDayNano)
decimal, varint, uuid, timeuuid, inet	Utf8 (no exact Arrow equivalent)
list<T>, set<T>	List
map<K, V>	Map
vector<T, n>	FixedSizeList
tuple<...>, user-defined types	Struct

All fields are nullable, because Cassandra returns null for any regular column without a value.

Writing (CassandraTableFactory::read_write_table_provider)

• INSERT INTO runs a prepared CQL INSERT per row, with up to 64 statements in flight. As in CQL, inserting a row whose primary key already exists replaces that row (an upsert).
• INSERT OVERWRITE truncates the table first. The truncation happens only after the INSERT was prepared successfully, so an invalid insert leaves the table untouched.
• Arrow values are converted to the CQL types of the target columns, including collections, tuples and user-defined types. A UUID can also be written from FixedSizeBinary(16).
• counter columns are rejected, because they can only be changed with UPDATE.
• Cassandra has no transactions: rows written before a failure are kept.

Federation (federation feature, enabled by default)

• CassandraTable implements SQLExecutor, and table_provider() returns a FederatedTableProviderAdaptor.
• CQL can't run the joins, aggregations and arbitrary filters that federation would push down. So a federated sub-plan is narrowed to its table scans: they keep their projection, key-column filters and LIMIT, and DataFusion evaluates the rest.
• Joins are never pushed down (JoinPushDown::Disallow).

Testing

• Unit tests cover parameter parsing, CQL generation, the filter pushdown rules and literal formatting, type mapping and federation plan rewriting.
• Docker-based integration tests (core/tests/cassandra) cover:
  • reading all supported types
  • projection, filters and limits
  • key-column filter pushdown, including compound partition keys, clustering ranges and the FilterExec fallback
  • count(*) and EXPLAIN
  • federation and joins
  • keyspace and table listing, and missing tables
  • pools with a predefined schema
  • INSERT INTO / INSERT OVERWRITE
• The integration tests run against cassandra:5.0 by default. Set CASSANDRA_DOCKER_IMAGE to use another image. ScyllaDB images (scylladb/scylla) are configured through entrypoint arguments instead of environment variables. The test container starts as a single node that doesn't wait for gossip.
• CI:
  • the Cassandra crate is added to the leaf-crate build matrix
  • cassandra:5.0 runs in the integration-test job
  • a new ScyllaDB Integration Tests job runs the same tests against scylladb/scylla:2026.3.3
• Added an example (cargo run -p datafusion-table-providers --example cassandra --features cassandra) and a README section.

🤖 Generated with Claude Code