Optimizing Hive on Tez Query with OR in LEFT JOIN

A strategic guide to rewriting and optimizing your query for improved performance

Key Highlights

Simplify OR Conditions: Break the join condition into separate, simpler joins using UNION or subqueries.
Leverage Tez and CBO: Enable Tez execution engine’s optimizations and Cost-Based Optimization (CBO) for more efficient query planning.
Optimize Data Storage: Use efficient file formats (like ORC), partitioning, bucketing, and vectorized execution to further enhance performance.

Introduction

When designing and optimizing queries in Hive on Tez, particularly those with complex join conditions involving OR logic, performance can degrade significantly if the query plan is suboptimal. A direct LEFT JOIN with an OR condition—such as t1.a = t2.a OR t1.b = t2.b—forces the system to evaluate all combinations, leading to resource-intensive operations and potential scalability issues with large datasets.

The goal here is to rewrite and optimize the query so that it returns the same results as the original query, but with an execution plan that takes advantage of Hive and Tez’s advanced features. This guide presents approaches like splitting the join logic using UNION ALL, employing Cost-Based Optimization (CBO), and ensuring that the tables are stored in optimized file formats such as ORC.

Optimization Strategies

1. Rewriting the Query Using UNION ALL

One of the most effective methods for eliminating the performance pitfalls of an OR condition in a join is to break the query into two separate LEFT JOINs that target each condition independently. By combining the results through UNION ALL (or UNION to eliminate duplicates, if necessary), you allow Hive to optimize each join separately.

Optimized Query Structure

The idea is to perform one LEFT JOIN where t1.a = t2.a and another where t1.b = t2.b. However, care must be taken to avoid duplicate rows for cases where the data might satisfy both conditions.


-- Optimized Query using UNION ALL
SELECT t1.a, t1.b, t2_a.a, t2_a.b
FROM t1
LEFT JOIN t2 AS t2_a
  ON t1.a = t2_a.a
  
UNION ALL

SELECT t1.a, t1.b, t2_b.a, t2_b.b
FROM t1
LEFT JOIN t2 AS t2_b
  ON t1.b = t2_b.b
WHERE NOT EXISTS (
  SELECT 1
  FROM t2 AS t2_a
  WHERE t1.a = t2_a.a
);

This approach ensures that the join based on t1.a = t2.a is handled first. In the second part, the WHERE NOT EXISTS clause filters out rows already captured by the first join to prevent duplication. This query structure can reduce the computational overhead typically associated with OR conditions.

2. Utilizing Subqueries for Pre-filtering

In certain situations, leveraging subqueries to pre-select matching rows from the join table can be an effective strategy. This method reduces the data volume that needs to be processed during the join.

For example, you can filter the second table (t2) based on relevant values from t1 before performing the join. Although this may sometimes be less efficient than using UNION ALL, it provides an alternative for cases with complex filtering requirements.

Example Scenario


SELECT t1.a, t1.b, t2.a, t2.b
FROM t1
LEFT JOIN (
  SELECT DISTINCT a, b
  FROM t2
  WHERE a IN (SELECT a FROM t1)
     OR b IN (SELECT b FROM t1)
) t2 ON t1.a = t2.a OR t1.b = t2.b

This subquery pre-selects rows from t2 that are likely to join with t1 based on either column condition. This method can reduce the amount of data scanned during the join operation.

3. Harnessing Hive and Tez Optimizations

Beyond rewriting the query, ensuring that your Hive and Tez environments are optimally configured is crucial. Several settings can be adjusted to leverage advanced query planning:

Cost-Based Optimization (CBO)

Enabling Cost-Based Optimization allows Hive to analyze table statistics and generate more efficient execution plans. To enable CBO, run the following commands:


SET hive.cbo.enable=true;
SET hive.compute.query.using.stats=true;
SET hive.stats.fetch.column.stats=true;
SET hive.stats.fetch.partition.stats=true;

Additionally, compute statistics for your tables to ensure the optimizer has the necessary data:


ANALYZE TABLE t1 COMPUTE STATISTICS;
ANALYZE TABLE t2 COMPUTE STATISTICS;

Tez-Specific Parameters

Tweaking Tez parameters can improve resource allocation and processing efficiency:

Container Size: Adjust hive.tez.container.size to allocate sufficient memory per Tez container.
Task Memory: Set tez.task.resource.memory to ensure each task has enough memory to process data batches.
Grouping Size: Configure tez.grouping.min-size to optimize task grouping based on input size.

Vectorized Execution

Hive’s vectorized execution processes batches of rows (typically 1024 rows at a time) rather than row by row. Enabling vectorization can significantly reduce execution time:


SET hive.vectorized.execution.enabled=true;
SET hive.vectorized.execution.reduce.enabled=true;

Optimizing Data Storage

File Format Conversion to ORC

Using ORC (Optimized Row Columnar) format is a standard approach in Hive to improve query performance. ORC offers benefits such as:

Improved input/output (I/O) performance due to columnar storage.
Predicate pushdown which minimizes the data scanned.
Efficient compression options to reduce storage space and speed up data reading.

To convert your tables to ORC format, use commands similar to:


ALTER TABLE t1 CONVERT TO FILE FORMAT ORC;
ALTER TABLE t2 CONVERT TO FILE FORMAT ORC;

Partitioning and Bucketing

For large datasets, partitioning tables based on frequently queried columns, such as the join keys, minimizes the amount of data read during query execution. Bucketing further segments partitions, allowing for more evenly distributed data processing.

Implementing well-designed partitioning and bucketing schemes can lead to significant performance gains and a more balanced workload distribution.

Additional Tools for Performance Monitoring

Beyond optimizing the query structure and configuration, monitoring the query’s performance is key to identifying potential bottlenecks. Tools such as Ambari or Cloudera Manager provide real-time insights into query execution on the Tez engine, allowing administrators to:

Monitor the number of tasks and their durations.
Identify data skew issues that might affect individual reducer performance.
Adjust parallel execution parameters based on the observed workload.

Practical Performance Comparison

Aspect	Original Query (OR Condition)	Optimized Query (UNION ALL)
Join Operation	Single LEFT JOIN with combined OR condition	Two separate LEFT JOINs with UNION ALL
Execution Plan	Potential Cartesian product effect	Optimized, efficient join processing
Performance Impact	Higher resource consumption and slower execution	Reduced row comparisons and better resource usage
Data Accuracy	Same results as intended but inefficient	Same results with improved performance

Putting It All Together

The final optimized query integrates the key ideas presented: splitting the OR join condition into separate LEFT JOIN operations combined with UNION ALL (or UNION for duplicate elimination), enabling advanced Tez and Hive configurations, and ensuring that your data storage is optimized with ORC, partitioning, and bucketing.

Here is the consolidated version:


-- Set Tez execution engine and enable CBO
SET hive.execution.engine=tez;
SET hive.cbo.enable=true;
SET hive.compute.query.using.stats=true;
SET hive.stats.fetch.column.stats=true;
SET hive.stats.fetch.partition.stats=true;

-- Enable vectorization
SET hive.vectorized.execution.enabled=true;
SET hive.vectorized.execution.reduce.enabled=true;

-- Collect necessary statistics for better optimization
ANALYZE TABLE t1 COMPUTE STATISTICS;
ANALYZE TABLE t2 COMPUTE STATISTICS;

-- Optimized Query Combining Two LEFT JOINs using UNION ALL
SELECT t1.a, t1.b, t2_a.a, t2_a.b
FROM t1
LEFT JOIN t2 AS t2_a
  ON t1.a = t2_a.a

UNION ALL

SELECT t1.a, t1.b, t2_b.a, t2_b.b
FROM t1
LEFT JOIN t2 AS t2_b
  ON t1.b = t2_b.b
WHERE NOT EXISTS (
  SELECT 1
  FROM t2 AS t2_a
  WHERE t1.a = t2_a.a
);

By adopting this approach, the query is restructured to enable more efficient processing while preserving the intended result set. The use of basic join arithmetic combined with Hive and Tez’s optimization capabilities significantly improves execution speed, especially in large-scale data environments.