E-MapReduce:Kudu connector

Schema emulation disabled (default)

With schema emulation disabled, all Kudu tables appear in the default schema.

Query the orders table using its fully qualified name:

SELECT * FROM kudu.default.orders;

If you set kudu as the catalog and default as the schema, the query simplifies to:

SELECT * FROM orders;

If a table name contains special characters, wrap it in double quotation marks:

SELECT * FROM kudu.default."special.table!";

Quick example: create and query a table

1. Create a table named users in the default schema:

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

   When creating a table, specify the primary key, column encoding or compression format, and partition information (hash or range).

2. Inspect the table schema:

   DESCRIBE kudu.default.users;

   Expected output:

      Column   |  Type   |                      Extra                      | Comment
   ------------+---------+-------------------------------------------------+---------
    user_id    | integer | primary_key, encoding=auto, compression=default |
    first_name | varchar | nullable, encoding=auto, compression=default    |
    last_name  | varchar | nullable, encoding=auto, compression=default    |
   (3 rows)

3. Insert rows:

   INSERT INTO kudu.default.users VALUES (1, 'Donald', 'Duck'), (2, 'Mickey', 'Mouse');

4. Query the data:

   SELECT * FROM kudu.default.users;

Schema emulation enabled

Enable schema emulation by setting kudu.schema-emulation.enabled=true in etc/catalog/kudu.properties. Kudu tables are then mapped to Trino schemas based on their names.

Mapping with an empty prefix (kudu.schema-emulation.prefix=)

Kudu does not support schemas. Trino creates a special table named $schemas to manage them.

Mapping with the standard prefix (kudu.schema-emulation.prefix=presto::)

With the standard prefix, Trino creates a special table named presto::$schemas to manage schemas.

Trino to Kudu

When writing data from Trino to Kudu, the following type mappings apply:

The following Trino types are not supported: TIME, JSON, TIME WITH TIMEZONE, TIMESTAMP WITH TIME ZONE, INTERVAL YEAR TO MONTH, INTERVAL DAY TO SECOND, ARRAY, MAP, IPADDRESS.

Column properties

Specify column properties in the WITH clause:

Example with explicit encoding and compression:

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 (...);

Partition design

Kudu supports hash partitions and range partitions. A table must have at least one partition of either type.

Define hash partitions

Specify partition_by_hash_columns (partition key columns) and partition_by_hash_buckets (number of buckets). Partition key columns must be a subset of primary key columns.

One hash partition group:

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

col1 and col2 are the hash partition key columns. Rows are distributed across 4 buckets.

Two independent hash partition groups:

CREATE TABLE mytable (
  col1 varchar WITH (primary_key=true),
  col2 varchar WITH (primary_key=true),
  ...
) WITH (
  partition_by_hash_columns = ARRAY['col1'],
  partition_by_hash_buckets = 2,
  partition_by_second_hash_columns = ARRAY['col2'],
  partition_by_second_hash_buckets = 3
);

The first group distributes rows by col1 into 2 buckets; the second distributes by col2 into 3 buckets. The total number of partitions is 6 (2 x 3).

Define range partitions

Use partition_by_range_columns to specify range partition columns and range_partitions to define initial partition bounds.

CREATE TABLE events (
  rack varchar WITH (primary_key=true),
  machine varchar WITH (primary_key=true),
  event_time timestamp WITH (primary_key=true),
  ...
) WITH (
  partition_by_hash_columns = ARRAY['rack'],
  partition_by_hash_buckets = 2,
  partition_by_second_hash_columns = ARRAY['machine'],
  partition_by_second_hash_buckets = 3,
  partition_by_range_columns = ARRAY['event_time'],
  range_partitions = '[{"lower": null, "upper": "2018-01-01T00:00:00"},
                       {"lower": "2018-01-01T00:00:00", "upper": null}]'
);

This table has two hash partition groups and one range partition on event_time, split at 2018-01-01T00:00:00.

Manage range partitions

Add or drop range partitions on an existing table using stored procedures:

-- Add a range partition
CALL kudu.system.add_range_partition(<schema_name>, <table_name>, <range_partition_as_json_string>)

-- Drop a range partition
CALL kudu.system.drop_range_partition(<schema_name>, <table_name>, <range_partition_as_json_string>)

Parameters:

JSON value formats by data type:

Example: Add a range partition to the events table in the myschema schema, covering records from 2018-01-01 to 2018-06-01:

CALL kudu.system.add_range_partition('myschema', 'events', '{"lower": "2018-01-01", "upper": "2018-06-01"}')

The lower bound "2018-01-01" is interpreted as 2018-01-01T00:00:00.000.

To view all existing range partitions on a table, run SHOW CREATE TABLE. The range_partitions property in the output lists the current partition bounds.