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.

$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.

    my $client= Cassandra::Client->new( ... );
    $client->connect;
    $client->execute("INSERT INTO my_table (id, value) VALUES (?, ?) USING TTL ?",
        [ 1, "test", 86400 ],
        { consistency => "quorum" });

    # 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;
    });

    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

    my $coderef= $client->future_execute("INSERT INTO table (id, value) VALUES (?, ?), [ $id, $value ]);
    # Do other things
    ...
    # Wait for the query to finish
    $coderef->();

    # This :
    $client->connect;
    # Is the same as :
    $client->future_connect->();

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.