5.10. Kudu Connector

    Connector is compatible with all Apache Kudu versions starting from 1.0.

    If the connector uses features that are not available on the target server, an error will be returned.Apache Kudu 1.8.0 is currently used for testing.

    Configuration

    To configure the Kudu connector, create a catalog properties file with the following contents,replacing the properties as appropriate:

    Apache Kudu does not support schemas, i.e. namespaces for tables.The connector can optionally emulate schemas by table naming conventions.

    The emulation of schemas is disabled by default.In this case all Kudu tables are part of the default schema.

    For example, a Kudu table named orders can be queried in Prestowith SELECT FROM kudu.default.orders or simple with SELECT FROM ordersif catalog and schema are set to kudu and default respectively.

    Table names can contain any characters in Kudu. In this case, use double quotes.E.g. To query a Kudu table named special.table! use SELECT * FROM kudu.default."special.table!".

    Example

    • Create a users table in the default schema with
    1. CREATE TABLE kudu.default.users (  user_id int WITH (primary_key = true),  first_name varchar,  last_name varchar) WITH (  partition_by_hash_columns = ARRAY['user_id'],  partition_by_hash_buckets = 2);

    On creating a Kudu table you must/can specify addition information aboutthe primary key, encoding, and compression of columns and hash or rangepartitioning. Details see in sectionCreate Table.

    • The table can be described using
    1. DESCRIBE kudu.default.users;

    You should get something like

    1.    Column   |  Type   |                      Extra                      | Comment
    2. ------------+---------+-------------------------------------------------+---------
    3.  user_id    | integer | primary_key, encoding=auto, compression=default |
    4.  first_name | varchar | nullable, encoding=auto, compression=default    |
    5.  last_name  | varchar | nullable, encoding=auto, compression=default    |
    6. (3 rows)
    • Insert some data with
    1. INSERT INTO kudu.default.users VALUES (1, 'Donald', 'Duck'), (2, 'Mickey', 'Mouse');
    • Select the inserted data

    If schema emulation has been enabled in the connector properties, i.e. etc/catalog/kudu.properties,tables are mapped to schemas depending on some conventions.

    • With kudu.schema-emulation.enabled=true and kudu.schema-emulation.prefix=,the mapping works like:

    Kudu Table NamePresto Qualified Nameorderskudu.default.orderspart1.part2kudu.part1.part2x.y.zkudu.x."y.z"

    As schemas are not directly supported by Kudu, a special table named$schemas is created for managing the schemas.

    • With kudu.schema-emulation.enabled=true and kudu.schema-emulation.prefix=presto::,the mapping works like:

    Kudu Table NamePresto Qualified Nameorderskudu.default.orderspart1.part2kudu.default."part1.part2"x.y.zkudu.default."x.y.z"presto::part1.part2kudu.part1.part2presto:x.y.zkudu.x."y.z"

    As schemas are not directly supported by Kudu, a special table namedpresto::$schemas is created for managing the schemas.

    Data Type Mapping

    The data types of Presto and Kudu are mapped as far as possible:

    ALTER SCHEMA … RENAME TO … is not supported.

    Create Table

    On creating a Kudu Table you need to provide the columns and their types, ofcourse, but Kudu needs information about partitioning and optionallyfor column encoding and compression.

      The primary key consists of user_id and event_name, the table is partitioned intofive partitions by hash values of the column user_id, and the number_of_replicas isexplicitly set to 3.

      The primary key columns must always be the first columns of the column list.All columns used in partitions must be part of the primary key.

      The table property number_of_replicas is optional. It defines thenumber of tablet replicas and must be an odd number. If it is not specified,the default replication factor from the Kudu master configuration is used.

      Kudu supports two different kinds of partitioning: hash and range partitioning.Hash partitioning distributes rows by hash value into one of many buckets.Range partitions distributes rows using a totally-ordered range partition key.The concrete range partitions must be created explicitly.Kudu also supports multi-level partitioning. A table must have at least onepartitioning (either hash or range). It can have at most one range partitioning,but multiple hash partitioning ‘levels’.

      For more details see .

      Besides column name and type, you can specify some more properties of a column.

      Example

      1. CREATE TABLE mytable (  name varchar WITH (primary_key = true, encoding = 'dictionary', compression = 'snappy'),  index bigint WITH (nullable = true, encoding = 'runlength', compression = 'lz4'),  comment varchar WITH (nullable = true, encoding = 'plain', compression = 'default'),   …) WITH (…);

      A table must have at least one partitioning (either hash or range).It can have at most one range partitioning, but multiple hash partitioning ‘levels’.For more details see Apache Kudu documentation:

      If you create a Kudu table in Presto, the partitioning design is given byseveral table properties.

      Hash partitioning

      You can provide the first hash partition group with two table properties:

      The partition_by_hash_columns defines the column(s) belonging to thepartition group and partition_by_hash_buckets the number of partitions tosplit the hash values range into. All partition columns must be part of theprimary key.

      Example:

      1. CREATE TABLE mytable (  col1 varchar WITH (primary_key=true),  col2 varchar WITH (primary_key=true),  …) WITH (  partition_by_hash_columns = ARRAY['col1', 'col2'],  partition_by_hash_buckets = 4)

      This defines a hash partitioning with the columns col1 and col2distributed over 4 partitions.

      To define two separate hash partition groups use also the second pairof table properties named partition_by_second_hash_columns andpartition_by_second_hash_buckets.

      Example:

      This defines a two-level hash partitioning with the first hash partition groupover the column col1 distributed over 2 buckets and the secondhash partition group over the column col2 distributed over 3 buckets.As a result you have table with 2 x 3 = 6 partitions.

      Range partitioning

      You can provide at most one range partitioning in Apache Kudu. The columnsare defined with the table property partition_by_range_columns.The ranges themselves are given either in thetable property range_partitions on creating the table.Or alternatively, the procedures kudu.system.add_range_partition andkudu.system.drop_range_partition can be used to manage rangepartitions for existing tables. For both ways see below for moredetails.

      Example:

      With the range_partitions table property you specify the concreterange partitions to be created. The range partition definition itselfmust be given in the table property partition_design separately.

      Example:

      1. CREATE TABLE events (  serialno varchar WITH (primary_key = true),  event_time timestamp WITH (primary_key = true),  message varchar) WITH (  partition_by_hash_columns = ARRAY['serialno'],  partition_by_hash_buckets = 4,  partition_by_range_columns = ARRAY['event_time'],  range_partitions = '[{"lower": null, "upper": "2017-01-01T00:00:00"},                       {"lower": "2017-01-01T00:00:00", "upper": "2017-07-01T00:00:00"},                       {"lower": "2017-07-01T00:00:00", "upper": "2018-01-01T00:00:00"}]');

      This creates a table with a hash partition on column serialno with 4buckets and range partitioning on column event_time. Additionallythree range partitions are created:

      1. for all event_times before the year 2017 (lower bound = null means it is unbound)
      2. for the first half of the year 2017
      3. for the second half the year 2017

      This means any try to add rows with event_time of year 2018 or greater will fail, as no partition is defined.The next section shows how to define a new range partition for an existing table.

      Managing range partitions

      For existing tables, there are procedures to add and drop a rangepartition.

      • adding a range partition
      1. CALL kudu.system.add_range_partition(<schema>, <table>, <range_partition_as_json_string>),
      • dropping a range partition
      1. CALL kudu.system.drop_range_partition(<schema>, <table>, <range_partition_as_json_string>)

      • <schema>: schema of the table

      • <table>: table names

      • <range_partition_as_json_string>: lower and upper bound of therange partition as json string in the form'{"lower": <value>, "upper": <value>}', or if the range partitionhas multiple columns:'{"lower": [<value_col1>,…], "upper": [<value_col1>,…]}'. Theconcrete literal for lower and upper bound values are depending onthe column types.

      Examples:

      Presto Data TypeJSON string exampleBIGINT‘{“lower”: 0, “upper”: 1000000}’SMALLINT‘{“lower”: 10, “upper”: null}’VARCHAR‘{“lower”: “A”, “upper”: “M”}’TIMESTAMP‘{“lower”: “2018-02-01T00:00:00.000”,“upper”: “2018-02-01T12:00:00.000”}’BOOLEAN‘{“lower”: false, “upper”: true}’VARBINARYvalues encoded as base64 strings

      To specified an unbounded bound, use the value null.

      Example:

      This would add a range partition for a table events in the schemamyschema with the lower bound 2018-01-01 (more exactly2018-01-01T00:00:00.000) and the upper bound 2018-07-01.

      Use the sql statement SHOW CREATE TABLE to query the existingrange partitions (they are shown in the table propertyrange_partitions).

      Adding a column to an existing table uses the SQL statement ALTER TABLE … ADD COLUMN ….You can specify the same column properties as on creating a table.

      Example:

      See also .

      Known limitations

      • Only lower case table and column names in Kudu are supported