oracular (3) Cassandra::Client.3pm.gz
NAME
Cassandra::Client - Perl library for accessing Cassandra using its binary network protocol
VERSION
version 0.21
DESCRIPTION
"Cassandra::Client" is a Perl library giving its users access to the Cassandra database, through the
native protocol. Both synchronous and asynchronous querying is supported, through various common calling
styles.
EXAMPLE
use Cassandra::Client;
my $client= Cassandra::Client->new(
contact_points => [ '127.0.0.1', '192.168.0.1' ],
username => "my_user",
password => "my_password",
keyspace => "my_keyspace",
);
$client->connect;
$client->each_page("SELECT id, column FROM my_table WHERE id=?", [ 5 ], undef, sub {
for my $row (@{shift->rows}) {
my ($id, $column)= @$row;
say "$id: $column";
}
});
METHODS
Cassandra::Client->new(%options)
Create a new "Cassandra::Client" instance, with the given options.
contact_points
Required. Arrayref of seed hosts to use when connecting. Specify more than one for increased
reliability. This array is shuffled before use, so that random hosts are picked from the array.
keyspace
Default keyspace to use on all underlying connections. Can be overridden by querying for specific
keyspaces, eg "SELECT * FROM system.peers".
anyevent
Should our internal event loop be based on AnyEvent, or should we just use our own? A true value
means enable AnyEvent. Needed for promises to work.
port
Port number to use. Defaults to 9042.
cql_version
CQL version to use. Defaults to the version the server is running. Override only if your client
has specific CQL requirements.
compression
Compression method to use. Defaults to the best available version, based on server and client
support. Possible values are "snappy", "lz4", and "none".
default_consistency
Default consistency level to use. Defaults to "one". Can be overridden on a query basis as well,
by passing a "consistency" attribute.
default_idempotency
Default value of the "idempotent" query attribute that indicates if a write query may be retried
without harm. It defaults to false.
max_page_size
Default max page size to pass to the server. This defaults to 5000. Note that large values can
cause trouble on Cassandra. Can be overridden by passing "page_size" in query attributes.
max_connections
Maximum amount of connections to keep open in the Cassandra connection pool. Defaults to 2 for
historical reasons, raise this if appropriate.
timer_granularity
Timer granularity used for timeouts. Defaults to 0.1 (100ms). Change this if you're setting
timeouts to values lower than a second.
request_timeout
Maximum time to wait for a query, in seconds. Defaults to 11.
warmup
Whether to connect to the full cluster in connect(), or delay that until queries come in.
protocol_version
Cassandra protocol version to use. Currently defaults to 4, can also be set to 3 for
compatibility with older versions of Cassandra.
$client->batch($queries[, $attributes])
Run one or more queries, in a batch, on Cassandra. Queries must be specified as an arrayref of
"[$query, \@bind]" pairs.
Defaults to a logged batch, which can be overridden by passing "logged", "unlogged" or "counter" as
the "batch_type" attribute.
$client->batch([
[ "INSERT INTO my_table (a, b) VALUES (?, ?)", [ $row1_a, $row1_b ] ],
[ "INSERT INTO my_table (a, b) VALUES (?, ?)", [ $row2_a, $row2_b ] ],
], { batch_type => "unlogged" });
$client->execute($query[, $bound_parameters[, $attributes]])
Executes a single query on Cassandra, and fetch the results (if any).
For queries that have large amounts of result rows and end up spanning multiple pages, "each_page" is
the function you need. "execute" does not handle pagination, and may end up missing rows unless
pagination is implemented by its user through the "page" attribute.
$client->execute(
"UPDATE my_table SET column=:new_column WHERE id=:id",
{ new_column => 2, id => 5 },
{ consistency => "quorum" },
);
The "idempotent" attribute indicates that the query is idempotent and may be retried without harm.
$client->each_page($query, $bound_parameters, $attributes, $page_callback)
Executes a query and invokes $page_callback with each page of the results, represented as
Cassandra::Client::ResultSet objects.
# Downloads the entire table from the database, even if it's terabytes in size
$client->each_page( "SELECT id, column FROM my_table", undef, undef, sub {
my $page= shift;
for my $row (@{$page->rows}) {
say $row->[0];
}
});
$client->prepare($query)
Prepares a query on the server. "execute" and "each_page" already do this internally, so this method
is only useful for preloading purposes (and to check whether queries even compile, I guess).
$client->shutdown()
Disconnect all connections and abort all current queries. After this, the "Cassandra::Client" object
considers itself shut down and must be reconstructed with new().
$client->wait_for_schema_agreement()
Wait until all nodes agree on the schema version. Useful after changing table or keyspace
definitions.
(A?)SYNCHRONOUS
It's up to the user to choose which calling style to use: synchronous, asynchronous with promises, or
through returned coderefs.
Synchronous
All "Cassandra::Client" methods are available as synchronous methods by using their normal names. For
example, "$client->connect();" will block until the client has connected. Similarly,
"$client->execute($query)" will wait for the query response. These are arguably not the fastest variants
(there's no parallelism in queries) but certainly the most convenient.
my $client= Cassandra::Client->new( ... );
$client->connect;
$client->execute("INSERT INTO my_table (id, value) VALUES (?, ?) USING TTL ?",
[ 1, "test", 86400 ],
{ consistency => "quorum" });
Promises
"Cassandra::Client" methods are also available as promises (see perldoc AnyEvent::XSPromises). This
integrates well with other libraries that deal with promises or asynchronous callbacks. Note that for
promises to work, "AnyEvent" is required, and needs to be enabled by passing "anyevent => 1" to
"Cassandra::Client->new()".
Promise variants are available by prefixing method names with "async_", eg. "async_connect",
"async_execute", etc. The usual result of the method is passed to the promise's success handler, or to
the failure handler if there was an error.
# Asynchronously pages through the result set, processing data as it comes in.
my $promise= $client->async_each_page("SELECT id, column FROM my_table WHERE id=?", [ 5 ], undef, sub {
for my $row (@{shift->rows}) {
my ($id, $column)= @$row;
say "Row: $id $column";
}
})->then(sub {
say "We finished paging through all the rows";
}, sub {
my $error= shift;
});
Promises normally get resolved from event loops, so for this to work you need one. Normally you would
deal with that by collecting all your promises and then waiting for that :
use AnyEvent::XSPromises qw/collect/;
use AnyEvent;
my @promises= ( ... ); # See other examples
my $condvar= AnyEvent->condvar;
collect(@promises)->then(sub {
$condvar->send;
}, sub {
my $error= shift;
warn "Unhandled error! $error";
$condvar->send;
});
$condvar->recv; # Wait for the promsie to resolve or fail
How you integrate this into your infrastructure is of course up to you, and beyond the scope of the
"Cassandra::Client" documentation.
Coderefs
These are the simplest form of asynchronous querying in "Cassandra::Client". Instead of dealing with
complex callback resolution, the client simply returns a coderef that, once invoked, returns what the
original method would have retruned.
The variants are available by prefixing method names with "future_", eg. "future_connect",
"future_execute", etc. These methods return a coderef.
my $coderef= $client->future_execute("INSERT INTO table (id, value) VALUES (?, ?), [ $id, $value ]);
# Do other things
...
# Wait for the query to finish
$coderef->();
Upon errors, the coderef will die, just like the synchronous methods would. Because of this, invoking the
coderef immediately after getting it is equal to using the synchronous methods :
# This :
$client->connect;
# Is the same as :
$client->future_connect->();
When used properly, coderefs can give a modest performance boost, but their real value is in the ease of
use compared to promises.
CAVEATS, BUGS, TODO
• Thread support is untested. Use at your own risk.
• The "timestamp" format is implemented naively by returning milliseconds since the UNIX epoch. In Perl
you get this number through "time() * 1000". Trying to save times as "DateTime" objects or strings
will not work, and will likely result in warnings and unexpected behavior.
AUTHOR
Tom van der Woerdt <tvdw@cpan.org>
COPYRIGHT AND LICENSE
This software is copyright (c) 2023 by Tom van der Woerdt.
This is free software; you can redistribute it and/or modify it under the same terms as the Perl 5
programming language system itself.