Guidelines for Partitioning Joins

It is recommended that you partition joins using only HASH or CUSTOM partitioning methods as the ROUNDROBIN partitioning method typically produces inconsistent join results. You can specify a partitioning method for only a subset of the join inputs with data from the other join inputs being broadcast to all parallel join instances. For example, if a join element has three input windows but you only specify HASH partitioning for Input1, then the other two inputs (Input2 and Input3) are broadcast by default.

Choosing a Partitioning Key

When using the HASH partitioning method, it is recommended that you use a hash key for each join input that is a subset of the join key for this input. For example, for join window InputWindow1, the join conditions are InputWindow1.key1 and InputWindow1.key2. The join keys for InputWindow1 are InputWindow1.key1 and InputWindow1.key2, so it is recommended that you choose a hash key for InputWindow that is either InputWindow1.key1 or InputWindow1.key2. Using different logic can break the original join semantics and produce unexpected results.

Similarly, when using the CUSTOM partitioning method, design the partitioning logic so that it is in line with the join keys. For example, for the join window below, you can only specify a partition method on W1 or W2. For W1, meaningful hash keys are (Key1W1), (Key2W1), and (Key1W1, Key2W1). For W2, meaningful hash keys are (Key1W2), (Key2W2), and (Key1W2, Key2W2). If you use HASH partitioning for both W1 and W2, then meaningful hash key pairs are (Key1W1) - (Key1W2), (Key2W1) - (Key2W2), and (Key1W1, Key2W1) - (Key1W2, Key2W2).

CREATE INPUT WINDOW W1 SCHEMA (Key1W1 integer, Key2W1 string, Val1W1 integer, Val2W1 string) PRIMARY KEY (Key1W1, Key2W1);
CREATE INPUT WINDOW W2 SCHEMA (Key1W2 integer, Key2W2 string, Val1W2 integer, Val2W2 string) PRIMARY KEY (Key1W2, Key2W2);

CREATE OUTPUT WINDOW OW1
PRIMARY KEY (Key1W2, Key2W2)
AS SELECT W1.*, W2.*
FROM W1 INNER JOIN W2 ON W1.Key1W1 = W2.Key1W2 AND W1.Key2W1 = W2.Key2W2;

Effects of Partitioning on Outer Joins

Although partitioning is generally supported for all types of joins (inner, outer, left, right, and full), partitioning the outer side of an outer join may change the original join semantics and lead to different results than a nonpartitioned scenario.

Here is an example of a left outer join:

CREATE INPUT WINDOW W1 SCHEMA (KeyW1 integer, ValW1 integer) PRIMARY KEY (KeyW1);
CREATE INPUT WINDOW W2 SCHEMA (KeyW2 integer, ValW2 integer) PRIMARY KEY (KeyW2); 

CREATE OUTPUT WINDOW OW1
PRIMARY KEY (KeyW1)
PARTITION
    BY W2 HASH (KeyW2)
PARTITIONS 2
AS SELECT W1.*, W2.*
FROM W1 OUTER JOIN W2 ON W1.KeyW1 = W2.KeyW2;

The input data of W1 is: <W1 ESP_OPS="i" KeyW1=”1”, ValW1=”100”>

The input data of W2 is: <W2 ESP_OPS="i" KeyW2=”1”, ValW2=”200”>

Given this input data for W1 and W2, the join result in a non-partitioned case would be <W1 ESP_OPS="i" KeyW1=”1”, ValW1=”100”, KeyW2=”1”, ValW2=”200”>.

However, because the default partitioning method for W1 is broadcast and W1 is the outer side of the given left outer join, record <W1> is sent to both instances of the join in the partitioned case. The join instance, which also receives the record from W2, then produces the result record <OW1 ESP_OPS="i" KeyW1=”1”, ValW1=”100”, KeyW2=”1”, ValW2=”200”>. The other instance also produces the result record <OW1 ESP_OPS="i" KeyW1=”1”, ValW1=”100”>. Since W1.KeyW1 is the primary key of the join window, depending on which of the two results arrive last at the final implicit union node, the record contained in the final join window can be either <OW1 ESP_OPS="i" KeyW1=”1”, ValW1=”100”, KeyW2=”1”, ValW2=”200”> or <OW1 ESP_OPS="i" KeyW1=”1”, ValW1=”100”>.

Similar scenarios can also occur for right and full joins.