Joins

Join Execution Types

Two types of joins are available, depending on need:

• Native Joins
• SQL Joins

Join Distribution Types

There are two types of join distribution:

1. Node-Local Joins - each node executes its portion of the overall join operation; faster, but join distribution criteria must be met
2. Distributed Joins - data is redistributed across the nodes so that node-local joins can be performed; slower and requiring more memory, but with no distribution restrictions

Creating a Join View

To create a join view, the /create/jointable endpoint requires four parameters:

1. the name of the join view to create
2. the list of member tables to be joined with optional aliases (TableA as A, TableB as B, ..., TableZ)
3. the list of columns and/or column expressions selected from the joined tables with optional aliases (TableA.id, B.id as B_id, ..., TableY.total + TableZ.total as YZ_total)
4. the SQL-style expression by which the tables can be joined and, optionally, filtered (TableA.b_id = B.id and ... A.z_id = TableZ.id)

Given a table clause of TableA as A, the a_id column of that table can be accessed via any of the following means:

• unique name, assuming the name is unique across the tables being joined (a_id)
• name with table prefix (TableA.a_id)
• name with table alias (A.a_id)

1
2
3
4
5
6

kinetica.create_join_table(
    join_table_name = 'example.customer_order_item',
    table_names = ['example.customer as c','example.orders as o','example.lineitem as l'],
    column_names = ['c_name','o_orderstatus','l_partkey','l_extendedprice','l_quantity','l_orderkey'],
    expressions = ['c_custkey = o_custkey','o_orderkey = l_orderkey']
)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

SELECT
    c_name,
    o_orderstatus,
    l_partkey,
    l_extendedprice,
    l_quantity
FROM
    example.customer c,
    example.orders o,
    example.lineitem l
WHERE
    c_custkey = o_custkey AND
    o_orderkey = l_orderkey

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

SELECT
    c_name,
    o_orderstatus,
    l_partkey,
    l_extendedprice,
    l_quantity
FROM
    example.customer c
INNER JOIN
    example.orders o ON c_custkey = o_custkey
INNER JOIN
    example.lineitem l ON o_orderkey = l_orderkey

1
2
3
4
5
6

kinetica.create_join_table(
    join_table_name = 'example.order_lineitem',
    table_names = ['example.orders as o','example.lineitem as l'],
    column_names = ['o_orderkey','l_partkey','l_extendedprice','l_quantity'],
    expressions = ['left join o, l on (o_orderkey = l_orderkey)']
)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

SELECT
    o_orderkey,
    l_partkey,
    l_extendedprice,
    l_quantity
FROM
    example.orders o
LEFT JOIN
    example.lineitem l
ON
    o_orderkey = l_orderkey

Operating on a Join View

Since join views are read-only, no inserts, updates, or deletes can be issued against them. Additionally, the following data processing endpoints cannot be applied to a join view.

• /aggregate/convexhull
• /aggregate/kmeans
• /aggregate/minmax/geometry
• /filter/bygeometry
• /filter/byrange
• /filter/bystring
• /filter/byvalue

1

join_view = gpudb.GPUdbTable(name = 'example.customer_order_item', db = kinetica)

1
2
3
4

join_view.filter(
    view_name = 'example.customer_order_item_multiple',
    expression = 'l_quantity > 1'
)

1
2
3
4
5
6
7
8
9

records = join_view.aggregate_group_by(
    column_names = [
        'c_name',
        'SUM(l_quantity) as total_items',
        'SUM(l_quantity * l_extendedprice) as total_spent'
    ],
    offset = 0,
    options = {'sort_by':'key'}
)['data']

1
2
3
4

revenue = join_view.aggregate_statistics(
    column_name = 'l_quantity * l_extendedprice',
    stats = 'sum'
)['stats']['sum']

1
2
3
4
5

records = join_view.get_records_by_column(
    column_names = ['c_name','o_orderstatus','l_partkey','l_extendedprice','l_quantity'],
    offset = 0,
    limit = gpudb.GPUdb.END_OF_SET
)

SQL Conversion Example

This example shows how the TPC-H benchmark's Query 5 SQL statement can be implemented in the Kinetica Python interface. The SQL for Query 5 is:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

SELECT
    n_name,
    SUM(l_extendedprice * (1 - l_discount)) AS revenue
FROM
    example.customer,
    example.orders,
    example.lineitem,
    example.supplier,
    example.nation,
    example.region
WHERE
    c_custkey = o_custkey AND
    l_orderkey = o_orderkey AND
    l_suppkey = s_suppkey AND
    c_nationkey = s_nationkey AND
    s_nationkey = n_nationkey AND
    n_regionkey = r_regionkey AND
    r_name = 'ASIA' AND
    o_orderdate >= '1994-01-01' AND
    o_orderdate < DATE('1994-01-01') + INTERVAL '1' YEAR
GROUP BY
    n_name
ORDER BY
    revenue DESC

To implement the above statement, a join view needs to be created with all of the tables listed in the FROM clause, and an expression, based on the WHERE clause, which binds the tables together and applies the necessary filters:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

kinetica.create_join_table(
    join_table_name = 'example.tpch_query_5_join_filter',
    table_names = [
        'example.customer',
        'example.orders',
        'example.lineitem',
        'example.supplier',
        'example.nation',
        'example.region'
    ],
    column_names = ['n_name','l_extendedprice * (1 - l_discount) as effective_price'],
    expressions = [
        'c_custkey = o_custkey',
        'l_orderkey = o_orderkey',
        'l_suppkey = s_suppkey',
        'c_nationkey = s_nationkey',
        's_nationkey = n_nationkey',
        'n_regionkey = r_regionkey',
        'r_name = \'ASIA\'',
        'o_orderdate >= \'1994-01-01\'',
        "o_orderdate < TIMESTAMPADD(YEAR, 1, DATE('1994-01-01'))"
    ]
)

Neither the table names nor the column names of the join view need to be aliased, as all column names are unique across all tables in this example. However, the derived column expression still requires an alias. Note also that columns used in the expressions list do not need to appear in the column_names list.

This join view could also have been done in two steps, a join and a filter:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

kinetica.create_join_table(
    join_table_name = 'example.tpch_query_5_join',
    table_names = [
        'example.customer',
        'example.orders',
        'example.lineitem',
        'example.supplier',
        'example.nation',
        'example.region'
    ],
    column_names = [
        'n_name',
        'l_extendedprice * (1 - l_discount) as effective_price',
        'r_name',
        'o_orderdate',
        'l_orderkey'
    ],
    expressions = [
        'c_custkey = o_custkey',
        'l_orderkey = o_orderkey',
        'l_suppkey = s_suppkey',
        'c_nationkey = s_nationkey',
        's_nationkey = n_nationkey',
        'n_regionkey = r_regionkey'
    ]
)

The join view can then have the WHERE clause filters applied in a separate step:

1
2
3
4
5
6

join_view.filter(
     view_name = 'example.tpch_query_5_join_filter',
     expression = 'r_name = \'ASIA\' and ' \
                  "o_orderdate >= '1994-01-01' and " \
                  "o_orderdate < TIMESTAMPADD(YEAR, 1, DATE('1994-01-01'))"
)

Note that the columns used in the filter's expression must appear in the join's column_names list, so that they are available for subsequent filter calls to use.

Regardless of whether the join/filter was done in one step or two, the results need to be grouped and summed to match the TPC-H query. The following call is given the directive to sort_by the value, which will sort on the aggregation column:

1
2
3
4
5
6
7
8

records = join_view.aggregate_group_by(
    column_names = [
        'n_name',
        'SUM(effective_price) as revenue'
    ],
    offset = 0,
    options = {"sort_by":"value", "sort_order":"descending"}
)['data']

The results can be displayed by calling:

1
2

for record in zip(records["n_name"], records["revenue"]):
    print("{:<13s} {:7.2f}".format(record[0], record[1]))

Performance Optimization

In order to improve the performance of a join operation, the columns used to make the relation should be indexed.

When a join is performed along a foreign key relation, it is indexed automatically as a function of the foreign key itself--no additional setup is required.

When a join is performed along a relation where no foreign key exists, the indexing must be done as a separate configuration step. Both columns, on either side of the join relation, should have a column index applied. An error may result if a join is attempted across a relation where only one side of the relation is indexed.

When a composite equi-join is performed, only one index will be used. For example: you have two tables, a and b, that each have the same columns, x and y. You also have an attribute index on each column. If you use the expression a.x == b.x and a.y == b.y, only the first expression, a.x == b.x, will use the index.

Limitations & Cautions

• Joins cannot be used to render geospatial tracks; read more about geospatial objects here.
• While a node-local join may include any number of replicated tables and sharded tables, all sharded tables must be joined by equating their respective shard key columns.
• A sharded table can be left (outer) joined to a replicated table via node-local join, but a replicated table can only be left (outer) joined to a sharded table via distributed join.
• When a data set is cleared (dropped) all the join views created from the data set are also automatically cleared.
• Records cannot be added to a join view, nor can records be updated or deleted.
• A join view is transient, by default, and will expire after the default TTL setting.
• A join view is not persisted, and will not survive a database restart.

Memory Implications

Joins have a variable memory footprint depending on the type of join. Kinetica will attempt to create the most efficient join as possible. The amount of memory a join requires depends on the type of join and the source data sets being joined together.

• A primary key join occupies the least amount of memory in Kinetica. This type is created from a join between two data sets with a primary key - foreign key relationship, allowing for an index on the lookup between the two data sets.
• A bitmask join occupies the second least amount of memory in Kinetica. This type results from a join between two data sets with no filtering of either table.
• An equijoin typically occupies the third least amount of memory in Kinetica. This type results from a join between two data sets containing an equality operator, returning only the records with equivalent values for the specified columns.
• A predicate join can be the most expensive or one of the cheapest for your memory; it entirely depends on how sparse the join is. A predicate join comprises a list of indices, or matched pairs of values, from the source data sets. This type results from a join between two data sets that are filtered prior to joining them.

Any column expression used in the column list for a join will be evaluated when the join is queried, requiring no persistent memory to hold the calculated values. While the memory footprint is small, the performance impact could be large, as the expression will need to be re-evaluated each time the join is queried.

Operations on Underlying Tables

Join views are affected by data modifications on underlying tables in different ways, based on the type of data modification.

• inserts - if a record is added to an underlying table, it will not appear in queries of the join view, whether or not the applied filter would include it
• updates - if a record that already contributes to a join view record set is modified in its underlying table, it will appear with the modified values when the join view is re-queried; if a record that does not contribute to a join view record set is modified in its underlying table in a way that would make it part of the join view record set, it will not appear when the join view is re-queried
• deletes - if a record that already contributes to a join view record set is deleted from its underlying table, it will still appear when the join view is re-queried

Overall, data manipulation within the tables supporting a join view will not cause the number of records returned by the join view to change, though updates to records already part of the join view record set will be reflected in the data values retrieved from the join view.