-- Partitioned table: booking_date is the partition column
-- booking_date does NOT appear in the regular column list!
CREATE TABLE bookings (
    booking_id  STRING,
    passenger   STRING,
    flight_code STRING,
    amount      DOUBLE
)
PARTITIONED BY (booking_date STRING)  -- partition key goes here
STORED AS ORC;

-- This creates directory structure:
-- /user/hive/warehouse/bookings/booking_date=2024-01-15/part-00000.orc
-- /user/hive/warehouse/bookings/booking_date=2024-01-16/part-00000.orc
-- ...

-- STATIC PARTITIONING: you manually specify the partition value
INSERT INTO bookings PARTITION (booking_date='2024-01-15')
SELECT booking_id, passenger, flight_code, amount
FROM bookings_raw
WHERE booking_date = '2024-01-15';
-- Loads ONLY one partition at a time
-- SAFE: you control exactly which partition is written
-- Use for: loading historical data one date at a time

-- DYNAMIC PARTITIONING: Hive reads the partition value from the data itself
SET hive.exec.dynamic.partition=true;
SET hive.exec.dynamic.partition.mode=nonstrict;
-- nonstrict = allow ALL partitions to be dynamic (at least 1 static needed for strict mode)

INSERT INTO bookings PARTITION (booking_date)
SELECT booking_id, passenger, flight_code, amount, booking_date  -- booking_date last!
FROM bookings_raw;
-- Hive reads booking_date column from each row, creates partition automatically
-- ⚠️ Can create HUNDREDS of partitions in one query → many small files problem!
-- ⚠️ If booking_date has 1000 unique dates → 1000 partitions × reducers = MANY files

-- Limit max dynamic partitions (safety net)
SET hive.exec.max.dynamic.partitions=1000;
SET hive.exec.max.dynamic.partitions.pernode=256;
-- If query would create more than 1000 partitions → FAIL (prevents cluster overload)

-- Merge small output files after insert
SET hive.merge.mapredfiles=true;
SET hive.merge.mapfiles=true;
SET hive.merge.smallfiles.avgsize=128000000;  -- 128 MB target
SET hive.merge.size.per.task=256000000;       -- 256 MB max
-- After INSERT, Hive runs a merge job to combine small output files into larger ones
-- This prevents small files inside partitions!

-- THE SCENARIO: Files added directly to HDFS (bypassing Hive)
-- Example: Sqoop wrote data to /user/hive/warehouse/bookings/booking_date=2024-02-01/
-- But Hive Metastore doesn't know this partition exists!

-- SYMPTOM:
SELECT * FROM bookings WHERE booking_date = '2024-02-01';
-- Returns: 0 rows (Metastore says this partition doesn't exist!)

-- FIX:
MSCK REPAIR TABLE bookings;
-- "MSCK" = MetaStore Check
-- What it does: scans HDFS for partition directories, adds missing partitions to Metastore
-- After this: Hive knows about the new partition → query works!

-- FASTER ALTERNATIVE (add one specific partition):
ALTER TABLE bookings ADD PARTITION (booking_date='2024-02-01')
LOCATION '/user/hive/warehouse/bookings/booking_date=2024-02-01/';

-- ⚠️ MSCK REPAIR can be SLOW on tables with THOUSANDS of partitions
-- (scans all of HDFS under the table path)
-- For large tables: use ALTER TABLE ADD PARTITION instead!

-- Create bucketed table: divide by booking_id into 32 buckets
CREATE TABLE bookings_bucketed (
    booking_id  STRING,
    passenger   STRING,
    amount      DOUBLE
)
CLUSTERED BY (booking_id) INTO 32 BUCKETS
STORED AS ORC;

-- Insert data — Hive automatically routes rows to correct bucket via hash(booking_id) % 32
INSERT INTO bookings_bucketed SELECT * FROM bookings_raw;

-- Bucket JOIN optimization (two tables bucketed by SAME column in SAME number of buckets)
-- Bucket 1 from table A joins with Bucket 1 from table B → NO SHUFFLE!
SET hive.optimize.bucketmapjoin=true;
SET hive.optimize.bucketmapjoin.sortedmerge=true;

-- Efficient SAMPLING (get 10% of data exactly):
SELECT * FROM bookings_bucketed TABLESAMPLE(BUCKET 1 OUT OF 10 ON booking_id);
-- Returns exactly 1/10 of rows = 10% sample (not random, but consistent)

-- Create ORC table with compression:
CREATE TABLE bookings_orc (
    booking_id STRING, passenger STRING, amount DOUBLE
)
STORED AS ORC
TBLPROPERTIES ("orc.compress"="SNAPPY");
-- SNAPPY = fast (good for hot frequently-read data)
-- ZLIB = more compressed (good for cold archive data)

-- Create Parquet table:
CREATE TABLE bookings_parquet (
    booking_id STRING, passenger STRING, amount DOUBLE
)
STORED AS PARQUET;

-- Convert existing Text table to ORC (HUGE performance win):
CREATE TABLE bookings_orc STORED AS ORC AS
SELECT * FROM bookings_text;
-- This alone can make queries 10-50x faster!

-- DEFAULT in older clusters: MapReduce (slow)
-- SET this at the START of every Hive session:
SET hive.execution.engine=tez;  -- Or 'mr' for MapReduce (don't use)

-- WHY TEZ IS FASTER:
-- MapReduce: Chain of MR jobs → each writes to HDFS between stages
-- SELECT country, COUNT(*), SUM(amount) GROUP BY country ORDER BY country
-- MapReduce: Job1 (GROUP BY) → write HDFS → Job2 (ORDER BY) → write HDFS
-- Tez:       One DAG: GROUP BY → ORDER BY (in memory, no intermediate HDFS write!)
-- Result: 5-10x faster for multi-stage queries

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

-- WHAT IT DOES:
-- Default: process 1 row at a time (row-by-row)
-- Vectorized: process 1024 rows at once (batch)
-- Uses CPU SIMD instructions (process multiple values in parallel on the CPU)
-- Result: 2-5x faster for column-heavy aggregations (SUM, AVG, COUNT)
-- Works BEST with ORC format (which stores data column-by-column)
-- ⚠️ Only works with ORC (not Text or CSV tables)

-- Enable Cost-Based Optimizer
SET hive.cbo.enable=true;
SET hive.compute.query.using.stats=true;
SET hive.stats.fetch.column.stats=true;

-- CBO needs TABLE STATISTICS to work:
ANALYZE TABLE bookings COMPUTE STATISTICS;           -- row count, file sizes
ANALYZE TABLE bookings COMPUTE STATISTICS FOR COLUMNS booking_id, country, amount;
-- Column stats: min, max, distinct count, nulls
-- CBO uses these stats to choose optimal join order and join type

-- WITHOUT stats: Hive guesses → wrong join order → shuffle join for a small table
-- WITH stats:    CBO says "passengers table has 10k rows → use MAP JOIN, no shuffle"

-- MAP JOIN = load small table into EVERY mapper's RAM, join locally (NO SHUFFLE!)
-- Auto map join (CBO decides based on table size):
SET hive.auto.convert.join=true;
SET hive.mapjoin.smalltable.filesize=25000000;  -- 25 MB threshold
-- If the smaller table < 25 MB → Hive automatically uses map join
-- ⚠️ Increase this if you have slightly larger but still "small" dimension tables

-- FORCE map join for a specific query with hint:
SELECT /*+ MAPJOIN(airports) */
    b.booking_id,
    a.airport_name
FROM bookings b
JOIN airports a ON b.departure_airport = a.code;
-- airports table is small (100 airports) → broadcast to all mappers → NO shuffle!

-- BUCKET MAP JOIN (both tables bucketed by join key):
SET hive.optimize.bucketmapjoin=true;
-- Even faster: only load matching bucket from small table into mapper

-- PARTITION PRUNING: Hive SKIPS partitions not matching WHERE clause
-- ⚠️ ONLY WORKS if you filter on the PARTITION COLUMN!

-- ✓ GOOD — partition pruning works (only scans booking_date=2024-01-15):
SELECT COUNT(*) FROM bookings WHERE booking_date = '2024-01-15';

-- ✗ BAD — NO partition pruning (scans ALL partitions!):
SELECT COUNT(*) FROM bookings WHERE YEAR(booking_date) = 2024;
-- Hive applies function YEAR() to each row → can't prune without scanning all!
-- FIX: WHERE booking_date BETWEEN '2024-01-01' AND '2024-12-31'

-- ✗ ANOTHER TRAP — dynamic partition filter from subquery:
-- This may or may not prune depending on Hive version and optimizer
SELECT * FROM bookings b
JOIN date_dim d ON b.booking_date = d.date
WHERE d.year = 2024;
-- May not prune! Use: WHERE b.booking_date >= '2024-01-01'

-- Data skew in GROUP BY (one key dominates):
-- Example: 50% of bookings are from 'US' → one reducer handles all US data!

SET hive.groupby.skewindata=true;
-- What Hive does: 2-phase aggregation:
-- Phase 1: randomly distribute keys → partial aggregation in parallel
-- Phase 2: final aggregation by real key
-- Result: no single reducer bottleneck for skewed keys

-- Also for skewed JOIN:
SET hive.optimize.skewjoin=true;
SET hive.skewjoin.key=100000;  -- Keys with > 100k rows are treated as "skew keys"

-- ORDER BY vs SORT BY vs DISTRIBUTE BY vs CLUSTER BY:

ORDER BY column     -- Global sort (all data goes to ONE reducer) → SLOW for large data
SORT BY column      -- Local sort (each reducer sorts its own output) → FAST but not globally sorted
DISTRIBUTE BY col   -- Control which rows go to which reducer (like partitioner)
CLUSTER BY col      -- = DISTRIBUTE BY col SORT BY col (same column for both)

-- ⚡ INTERVIEW TRAP: "What's the difference between ORDER BY and SORT BY?"
-- ORDER BY: guaranteed global order, uses 1 reducer (slow for billions of rows!)
-- SORT BY: each reducer output is sorted, but globally NOT sorted (multiple reducers)
-- Use SORT BY when you just need local sorted chunks (e.g., before writing partitioned output)
-- Use ORDER BY only for final result sets that need to be globally ordered

-- WHAT HAPPENED:
-- Someone ran: hdfs dfs -rm -r /user/hive/warehouse/bookings/booking_date=2024-01-15/
-- Now this query returns 0 rows:
SELECT COUNT(*) FROM bookings WHERE booking_date = '2024-01-15';
-- Even though Metastore thinks the partition exists!

-- DIAGNOSIS:
-- 1. Check if partition exists in Metastore:
SHOW PARTITIONS bookings;
-- Shows: booking_date=2024-01-15 (it's there in Metastore!)
-- 2. Check if files exist in HDFS:
-- hdfs dfs -ls /user/hive/warehouse/bookings/booking_date=2024-01-15/
-- Output: No such file or directory ← CONFIRMED — data deleted!

-- FIX OPTIONS:
-- Option 1: Restore data from backup and MSCK REPAIR TABLE bookings
-- Option 2: If no backup — DROP the stale partition from Metastore:
ALTER TABLE bookings DROP PARTITION (booking_date='2024-01-15');
-- Now Hive is consistent again (it won't show non-existent partition)
-- Then: reload data into that partition from source

-- PREVENTION:
-- 1. NEVER run hdfs dfs -rm directly on Hive-managed table locations
-- 2. Use HDFS Trash (hdfs dfs -rm moves to .Trash, not permanent delete)
-- 3. Enable HDFS Snapshots for the warehouse directory

-- WHAT HAPPENED: Sqoop/Spark wrote new partition directory to HDFS
-- but forgot to register it in Hive Metastore

-- DIAGNOSIS:
SELECT * FROM bookings WHERE booking_date = '2024-03-01';
-- Returns 0 rows... but file exists in HDFS!

-- FIX — Option 1: MSCK REPAIR (for EXTERNAL tables, scans all partitions):
MSCK REPAIR TABLE bookings;

-- FIX — Option 2: ADD PARTITION (fast, for specific known partition):
ALTER TABLE bookings ADD IF NOT EXISTS PARTITION (booking_date='2024-03-01')
LOCATION '/user/hive/warehouse/bookings/booking_date=2024-03-01/';

-- ⚠️ MSCK REPAIR on a table with 10,000 partitions → scans 10,000 HDFS paths!
-- Use ALTER TABLE ADD PARTITION when you know exactly which partition to add

-- STEP 1: Enable EXPLAIN to see query plan:
EXPLAIN SELECT country, COUNT(*) FROM bookings GROUP BY country;
-- Look for: Map Join vs Shuffle Join, partition pruning status, # of Map/Reduce stages

-- STEP 2: Check query stats after run:
SET hive.stats.autogather=true;
-- After INSERT: Hive auto-collects stats (rows, bytes)

-- STEP 3: Check if vectorization is working:
EXPLAIN VECTORIZATION SELECT country, COUNT(*) FROM bookings_orc GROUP BY country;
-- Should show "Vectorized execution: true" for ORC tables

-- STEP 4: Check for data skew:
-- If one reducer takes much longer than others → skew problem
SET hive.groupby.skewindata=true;

-- STEP 5: Confirm Tez engine:
SET hive.execution.engine;
-- Should show: tez (if mr → change to tez immediately!)

# BASIC IMPORT: Oracle table → HDFS
sqoop import \
  --connect jdbc:oracle:thin:@//db-host:1521/AMADEUS \
  --username etl_user \
  --password 'secret' \
  --table FLIGHT_INVENTORY \
  --target-dir /data/raw/flight_inventory/ \
  --num-mappers 4             # 4 parallel import tasks

# IMPORT WITH HIVE INTEGRATION: directly to Hive table
sqoop import \
  --connect jdbc:mysql://db-host/bookings \
  --username user --password pass \
  --table bookings \
  --hive-import \             # create Hive table automatically
  --hive-table booking_db.bookings_raw \
  --create-hive-table \
  --num-mappers 8

# INCREMENTAL IMPORT (most important for production!):
# Only import NEW rows since last run (much faster than full import daily)

# Mode 1: append — for insert-only tables (new rows have higher IDs)
sqoop import \
  --connect jdbc:oracle:thin:@//db-host:1521/AMADEUS \
  --table BOOKINGS \
  --incremental append \
  --check-column booking_id \     # column to check for new rows
  --last-value 1000000 \          # only import booking_id > 1000000
  --target-dir /data/raw/bookings_incremental/

# Mode 2: lastmodified — for tables with an update timestamp
sqoop import \
  --connect jdbc:oracle:thin:@//db-host:1521/AMADEUS \
  --table BOOKINGS \
  --incremental lastmodified \
  --check-column updated_at \     # timestamp column
  --last-value "2024-01-15 00:00:00" \
  --target-dir /data/raw/bookings_incremental/

# --split-by: control how data is split across mappers (affects parallelism)
sqoop import \
  --table BOOKINGS \
  --split-by booking_date \       # each mapper handles a date range
  --num-mappers 8
# ⚠️ Default: split by primary key (must be numeric!)
# ⚠️ If no primary key and no --split-by → must use --num-mappers 1 (slow!)

<!-- workflow.xml — simple Hive workflow -->
<workflow-app name="daily-booking-report" xmlns="uri:oozie:workflow:0.5">

    <start to="hive-action"/>

    <!-- Action 1: Run Hive query -->
    <action name="hive-action">
        <hive xmlns="uri:oozie:hive-action:0.5">
            <job-tracker>${jobTracker}</job-tracker>
            <name-node>${nameNode}</name-node>
            <script>/user/oozie/scripts/aggregate_bookings.hql</script>
            <param>booking_date=${booking_date}</param>
        </hive>
        <ok to="sqoop-export"/>
        <error to="fail"/>
    </action>

    <!-- Action 2: Export result to Oracle -->
    <action name="sqoop-export">
        <sqoop xmlns="uri:oozie:sqoop-action:0.4">
            <command>export --connect jdbc:oracle:... --table REPORT_TABLE ...</command>
        </sqoop>
        <ok to="end"/>
        <error to="fail"/>
    </action>

    <kill name="fail">
        <message>Workflow failed: ${wf:errorMessage(wf:lastErrorNode())}</message>
    </kill>

    <end name="end"/>
</workflow-app>