Data Fragmentation in Distributed Databases

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
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94

-- Query Localization Examples
 
-- ================================================================
-- HORIZONTAL FRAGMENT LOCALIZATION
-- ================================================================
 
-- Original global query
-- SELECT customer_name, total_amount
-- FROM Orders
-- WHERE region = 'DE' AND order_date >= '2024-01-01';
 
-- After localization (only EU fragment accessed):
SELECT customer_name, total_amount
FROM Orders_EU
WHERE region = 'DE' AND order_date >= '2024-01-01';
-- Orders_NA and Orders_APAC eliminated
 
 
-- ================================================================
-- VERTICAL FRAGMENT LOCALIZATION
-- ================================================================
 
-- Given vertical fragments:
-- Customers_Contact: tid, customer_id, name, email, phone
-- Customers_Demographics: tid, age, gender, income_bracket
-- Customers_Preferences: tid, newsletter, language, timezone
 
-- Original query only needs contact info:
-- SELECT name, email FROM Customers WHERE customer_id = 12345;
 
-- After localization (only Contact fragment accessed):
SELECT name, email
FROM Customers_Contact
WHERE customer_id = 12345;
-- No join needed! Demographics and Preferences fragments not accessed
 
 
-- Query needing multiple vertical fragments:
-- SELECT name, age, newsletter FROM Customers WHERE customer_id = 12345;
 
-- After localization (two fragments, minimal attributes):
SELECT c.name, d.age, p.newsletter
FROM Customers_Contact c
JOIN Customers_Demographics d ON c.tid = d.tid
JOIN Customers_Preferences p ON c.tid = p.tid
WHERE c.customer_id = 12345;
-- Only necessary fragments joined
 
 
-- ================================================================
-- HYBRID FRAGMENT LOCALIZATION
-- ================================================================
 
-- Given HV fragmentation:
-- Orders_NA_Billing: region IN (NA), billing attributes
-- Orders_NA_Shipping: region IN (NA), shipping attributes
-- Orders_EU_Billing: region IN (EU), billing attributes
-- Orders_EU_Shipping: region IN (EU), shipping attributes
 
-- Query: Get billing info for US orders
-- SELECT order_id, amount, tax FROM Orders WHERE region = 'US';
 
-- After localization:
-- 1. Eliminate EU fragments (region mismatch)
-- 2. Eliminate Shipping fragments (attributes not needed)
 
SELECT order_id, amount, tax
FROM Orders_NA_Billing
WHERE region = 'US';
-- Only one fragment accessed out of four!
 
 
-- ================================================================
-- LOCALIZATION OPTIMIZATION VIEW
-- ================================================================
 
-- Create a view that the optimizer can localize
CREATE VIEW global_orders AS
SELECT * FROM Orders_NA
UNION ALL
SELECT * FROM Orders_EU
UNION ALL
SELECT * FROM Orders_APAC;
 
-- PostgreSQL optimizer will perform constraint exclusion
-- when querying this view with partition-compatible predicates
EXPLAIN (COSTS OFF)
SELECT * FROM global_orders WHERE region = 'JP';
 
-- OUTPUT (ideal):
-- Append
--   ->  Seq Scan on orders_apac
--         Filter: (region = 'JP')
-- (NA and EU partitions excluded)

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
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84

-- Semi-Join Reduction Strategy
 
-- Scenario: Reconstruct customer data from vertical fragments
-- Customers_Contact (Site A): 10M rows, contains email for filtering
-- Customers_Financial (Site B): 10M rows, contains salary data
 
-- Query: Get names and salaries for customers with @gmail.com email
-- SELECT c.name, f.salary
-- FROM Customers_Contact c
-- JOIN Customers_Financial f ON c.tid = f.tid
-- WHERE c.email LIKE '%@gmail.com';
 
 
-- ================================================================
-- NAIVE APPROACH (what NOT to do)
-- ================================================================
 
-- Step 1: Ship all of Customers_Financial (10M rows × 200B = 2GB) to Site A
-- Step 2: Join locally
-- Step 3: Filter by email
-- Result: Most transferred data is discarded!
 
 
-- ================================================================
-- SEMI-JOIN APPROACH (optimized)
-- ================================================================
 
-- Step 1: At Site A, find qualifying TIDs (site-local query)
CREATE TEMP TABLE qualifying_tids AS
SELECT tid
FROM Customers_Contact
WHERE email LIKE '%@gmail.com';
-- Result: ~100K TIDs (1% selectivity, 8 bytes each = 800KB)
 
 
-- Step 2: Send TID set to Site B (800KB transfer)
-- (In practice, this might be a federated query or explicit data transfer)
 
 
-- Step 3: At Site B, filter Customers_Financial using received TIDs
CREATE TEMP TABLE filtered_financial AS
SELECT f.tid, f.salary
FROM Customers_Financial f
WHERE f.tid IN (SELECT tid FROM qualifying_tids);
-- Result: ~100K rows (1% of table, 100K × 200B = 20MB)
 
 
-- Step 4: Send filtered result back to Site A (20MB transfer)
 
 
-- Step 5: At Site A, complete the join locally
SELECT c.name, f.salary
FROM Customers_Contact c
JOIN filtered_financial f ON c.tid = f.tid
WHERE c.email LIKE '%@gmail.com';
 
 
-- Total transfer: 800KB + 20MB ≈ 21MB
-- vs. Naive: 2GB
-- Savings: ~99%!
 
 
-- ================================================================
-- BLOOM FILTER OPTIMIZATION
-- ================================================================
 
-- For very large TID sets, sending the actual set is expensive
-- Use a Bloom filter: probabilistic set membership with false positives
 
-- Build Bloom filter for qualifying TIDs (fixed size, e.g., 1MB)
-- Send Bloom filter instead of TID list
 
-- At Site B, filter using Bloom filter
-- Some false positives will be included, but join will filter them out
 
-- PostgreSQL extension for Bloom filters
CREATE EXTENSION bloom;
 
-- Create bloom index on TID
CREATE INDEX idx_financial_tid_bloom ON Customers_Financial 
    USING bloom (tid) WITH (length=1024);
 
-- With proper integration, the optimizer can use Bloom filters
-- for semi-join reduction in distributed queries

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
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145

-- Materialized View Reconstruction Strategies
 
-- ================================================================
-- FULL RECONSTRUCTION MATERIALIZATION
-- ================================================================
 
-- Vertical fragments:
-- Customers_Contact: tid, customer_id, name, email
-- Customers_Demographics: tid, age, income_bracket
-- Customers_Activity: tid, last_login, purchase_count
 
-- Materialized view for complete customer record
CREATE MATERIALIZED VIEW customers_full AS
SELECT 
    c.tid,
    c.customer_id,
    c.name,
    c.email,
    d.age,
    d.income_bracket,
    a.last_login,
    a.purchase_count
FROM Customers_Contact c
JOIN Customers_Demographics d ON c.tid = d.tid
JOIN Customers_Activity a ON c.tid = a.tid
WITH DATA;
 
-- Create indexes for common query patterns
CREATE INDEX idx_customers_full_email ON customers_full(email);
CREATE INDEX idx_customers_full_income ON customers_full(income_bracket);
CREATE INDEX idx_customers_full_active ON customers_full(last_login) 
    WHERE last_login > CURRENT_DATE - INTERVAL '30 days';
 
-- Refresh strategy: full refresh nightly
REFRESH MATERIALIZED VIEW customers_full;
 
 
-- ================================================================
-- PARTIAL RECONSTRUCTION FOR HOT QUERIES
-- ================================================================
 
-- Horizontal fragments by region (NA, EU, APAC)
-- Common report needs only NA data with specific attributes
 
CREATE MATERIALIZED VIEW customers_na_report AS
SELECT 
    customer_id,
    name,
    email,
    state,
    revenue_ytd,
    last_order_date
FROM Customers_NA
WHERE status = 'active'
WITH DATA;
 
-- Refresh more frequently for operational data
-- Using CONCURRENTLY for zero-downtime refresh
CREATE UNIQUE INDEX idx_customers_na_report_pk ON customers_na_report(customer_id);
REFRESH MATERIALIZED VIEW CONCURRENTLY customers_na_report;
 
 
-- ================================================================
-- INCREMENTAL MATERIALIZATION
-- ================================================================
 
-- For real-time or near-real-time requirements
-- Use triggers to maintain materialized view
 
-- Create the materialized table manually
CREATE TABLE customers_realtime (
    tid BIGINT PRIMARY KEY,
    customer_id BIGINT UNIQUE,
    name VARCHAR(100),
    email VARCHAR(255),
    age INTEGER,
    last_login TIMESTAMP,
    updated_at TIMESTAMP DEFAULT NOW()
);
 
-- Trigger function to propagate changes
CREATE OR REPLACE FUNCTION sync_customers_realtime()
RETURNS TRIGGER AS $$
BEGIN
    IF TG_TABLE_NAME = 'customers_contact' THEN
        INSERT INTO customers_realtime (tid, customer_id, name, email)
        VALUES (NEW.tid, NEW.customer_id, NEW.name, NEW.email)
        ON CONFLICT (tid) DO UPDATE SET
            customer_id = EXCLUDED.customer_id,
            name = EXCLUDED.name,
            email = EXCLUDED.email,
            updated_at = NOW();
    ELSIF TG_TABLE_NAME = 'customers_demographics' THEN
        UPDATE customers_realtime 
        SET age = NEW.age, updated_at = NOW()
        WHERE tid = NEW.tid;
    ELSIF TG_TABLE_NAME = 'customers_activity' THEN
        UPDATE customers_realtime 
        SET last_login = NEW.last_login, updated_at = NOW()
        WHERE tid = NEW.tid;
    END IF;
    RETURN NEW;
END;
$$ LANGUAGE plpgsql;
 
-- Attach triggers to vertical fragments
CREATE TRIGGER tr_sync_contact
AFTER INSERT OR UPDATE ON customers_contact
FOR EACH ROW EXECUTE FUNCTION sync_customers_realtime();
 
CREATE TRIGGER tr_sync_demographics
AFTER INSERT OR UPDATE ON customers_demographics
FOR EACH ROW EXECUTE FUNCTION sync_customers_realtime();
 
CREATE TRIGGER tr_sync_activity
AFTER INSERT OR UPDATE ON customers_activity
FOR EACH ROW EXECUTE FUNCTION sync_customers_realtime();
 
 
-- ================================================================
-- AGGREGATED RECONSTRUCTION
-- ================================================================
 
-- Sometimes full reconstruction isn't needed—only aggregates
 
CREATE MATERIALIZED VIEW regional_sales_summary AS
SELECT 
    region,
    DATE_TRUNC('month', order_date) AS month,
    COUNT(*) AS order_count,
    SUM(amount) AS total_revenue,
    AVG(amount) AS avg_order_value
FROM (
    SELECT * FROM Orders_NA
    UNION ALL
    SELECT * FROM Orders_EU
    UNION ALL
    SELECT * FROM Orders_APAC
) all_orders
GROUP BY region, DATE_TRUNC('month', order_date)
WITH DATA;
 
-- Much smaller than full reconstruction!
-- Refresh daily for reporting freshness
REFRESH MATERIALIZED VIEW regional_sales_summary;

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
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122

-- Consistent Reconstruction Strategies
 
-- ================================================================
-- SNAPSHOT ISOLATION
-- ================================================================
 
-- Use a single transaction with REPEATABLE READ for consistent reconstruction
BEGIN TRANSACTION ISOLATION LEVEL REPEATABLE READ;
 
-- All these queries see the same snapshot
SELECT c.name, c.email 
FROM Customers_Contact c 
WHERE c.tid IN (SELECT tid FROM active_customers);
 
SELECT d.age, d.income_bracket 
FROM Customers_Demographics d 
WHERE d.tid IN (SELECT tid FROM active_customers);
 
SELECT a.last_login, a.purchase_count 
FROM Customers_Activity a 
WHERE a.tid IN (SELECT tid FROM active_customers);
 
COMMIT;
-- All three queries see database as of transaction start
 
 
-- ================================================================
-- VERSIONED RECONSTRUCTION
-- ================================================================
 
-- Add version/timestamp metadata to fragments
ALTER TABLE Customers_Contact ADD COLUMN version_ts TIMESTAMP DEFAULT NOW();
ALTER TABLE Customers_Demographics ADD COLUMN version_ts TIMESTAMP DEFAULT NOW();
ALTER TABLE Customers_Activity ADD COLUMN version_ts TIMESTAMP DEFAULT NOW();
 
-- Reconstruct as of a specific point in time
-- (Requires temporal tables or versioning extension)
 
-- PostgreSQL temporal tables approach
CREATE EXTENSION temporal_tables;
 
-- Create history tables for versioning
CREATE TABLE Customers_Contact_History (LIKE Customers_Contact);
ALTER TABLE Customers_Contact_History ADD COLUMN valid_from TIMESTAMP;
ALTER TABLE Customers_Contact_History ADD COLUMN valid_until TIMESTAMP;
 
-- Query as of specific timestamp
CREATE OR REPLACE FUNCTION reconstruct_customer_at(
    p_tid BIGINT,
    p_as_of TIMESTAMP
) RETURNS TABLE (
    name VARCHAR,
    email VARCHAR,
    age INTEGER,
    last_login TIMESTAMP
) AS $$
BEGIN
    RETURN QUERY
    SELECT 
        c.name, c.email, d.age, a.last_login
    FROM 
        (SELECT * FROM Customers_Contact_History 
         WHERE tid = p_tid 
           AND valid_from <= p_as_of 
           AND (valid_until IS NULL OR valid_until > p_as_of)) c
    JOIN
        (SELECT * FROM Customers_Demographics_History 
         WHERE tid = p_tid 
           AND valid_from <= p_as_of 
           AND (valid_until IS NULL OR valid_until > p_as_of)) d
        ON c.tid = d.tid
    JOIN
        (SELECT * FROM Customers_Activity_History 
         WHERE tid = p_tid 
           AND valid_from <= p_as_of 
           AND (valid_until IS NULL OR valid_until > p_as_of)) a
        ON c.tid = a.tid;
END;
$$ LANGUAGE plpgsql;
 
 
-- ================================================================
-- LAG DETECTION
-- ================================================================
 
-- Monitor reconstruction consistency by tracking fragment timestamps
 
CREATE TABLE fragment_metadata (
    fragment_name VARCHAR PRIMARY KEY,
    last_write_ts TIMESTAMP,
    replica_lag_ms INTEGER
);
 
-- Update metadata on each write (via triggers or application)
 
-- Before reconstruction, check for unacceptable lag
CREATE OR REPLACE FUNCTION check_reconstruction_consistency(
    max_lag_ms INTEGER DEFAULT 1000
) RETURNS BOOLEAN AS $$
DECLARE
    max_observed_lag INTEGER;
BEGIN
    SELECT MAX(replica_lag_ms) INTO max_observed_lag
    FROM fragment_metadata
    WHERE fragment_name IN ('Customers_Contact', 'Customers_Demographics', 'Customers_Activity');
    
    IF max_observed_lag > max_lag_ms THEN
        RAISE WARNING 'Reconstruction may be inconsistent. Max lag: % ms', max_observed_lag;
        RETURN FALSE;
    END IF;
    
    RETURN TRUE;
END;
$$ LANGUAGE plpgsql;
 
-- Usage before reconstruction queries
DO $$
BEGIN
    IF NOT check_reconstruction_consistency(500) THEN
        RAISE EXCEPTION 'Fragments too inconsistent for reconstruction';
    END IF;
END $$;

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
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142

import asyncio
import asyncpg
from typing import List, AsyncIterator
from dataclasses import dataclass
 
@dataclass
class FragmentConfig:
    name: str
    connection_string: str
    predicate: str
 
class DistributedReconstructor:
    """Parallel and pipelined reconstruction from distributed fragments."""
    
    def __init__(self, fragments: List[FragmentConfig]):
        self.fragments = fragments
        self.pools = {}
    
    async def initialize(self):
        """Create connection pools for each fragment site."""
        for frag in self.fragments:
            self.pools[frag.name] = await asyncpg.create_pool(
                frag.connection_string,
                min_size=2,
                max_size=10
            )
    
    async def close(self):
        """Close all connection pools."""
        for pool in self.pools.values():
            await pool.close()
    
    async def reconstruct_horizontal(
        self,
        query_predicates: dict,
        attributes: List[str]
    ) -> AsyncIterator[dict]:
        """
        Reconstruct from horizontal fragments with parallel access.
        Yields results as they arrive (pipelined).
        """
        # Build query for each fragment
        async def query_fragment(frag: FragmentConfig):
            # Check if fragment can be eliminated
            if not self._predicates_compatible(query_predicates, frag.predicate):
                return  # Fragment eliminated
            
            pool = self.pools[frag.name]
            async with pool.acquire() as conn:
                # Build and execute query
                sql = self._build_query(frag.name, attributes, query_predicates)
                async for record in conn.cursor(sql):
                    yield record
        
        # Execute all fragment queries in parallel
        tasks = [query_fragment(f) for f in self.fragments]
        
        # Merge results as they complete (pipelined)
        for task in asyncio.as_completed([
            asyncio.create_task(self._collect(t)) for t in tasks
        ]):
            records = await task
            for record in records:
                yield record
    
    async def reconstruct_vertical(
        self,
        tid_filter: str,
        fragment_attributes: dict  # {fragment_name: [attributes]}
    ) -> AsyncIterator[dict]:
        """
        Reconstruct from vertical fragments using parallel access and merge.
        """
        async def query_vertical_fragment(frag_name: str, attrs: List[str]):
            pool = self.pools[frag_name]
            async with pool.acquire() as conn:
                sql = f"""
                    SELECT tid, {', '.join(attrs)}
                    FROM {frag_name}
                    WHERE {tid_filter}
                """
                return await conn.fetch(sql)
        
        # Query all vertical fragments in parallel
        tasks = {
            frag: asyncio.create_task(
                query_vertical_fragment(frag, attrs)
            )
            for frag, attrs in fragment_attributes.items()
        }
        
        # Wait for all results
        results = {}
        for frag, task in tasks.items():
            results[frag] = {r['tid']: dict(r) for r in await task}
        
        # Merge on TID
        primary_frag = list(fragment_attributes.keys())[0]
        for tid, primary_record in results[primary_frag].items():
            merged = primary_record.copy()
            for frag in list(fragment_attributes.keys())[1:]:
                if tid in results[frag]:
                    merged.update(results[frag][tid])
            yield merged
    
    def _predicates_compatible(self, query_pred: dict, frag_pred: str) -> bool:
        """Check if query predicates are compatible with fragment definition."""
        # Simplified - real implementation needs SQL predicate analysis
        return True  # Placeholder
    
    def _build_query(self, table: str, attrs: List[str], preds: dict) -> str:
        """Build SQL query for fragment."""
        where_clause = ' AND '.join(f"{k} = ${i + 1}" for i, k in enumerate(preds))
        return f"SELECT {', '.join(attrs)} FROM {table} WHERE {where_clause}"
    
    async def _collect(self, async_gen) -> List:
        """Collect async generator results into list."""
        return [item async for item in async_gen]
 
 
# Usage example
async def main():
    fragments = [
        FragmentConfig("orders_na", "postgresql://db1/orders", "region IN ('US', 'CA')"),
        FragmentConfig("orders_eu", "postgresql://db2/orders", "region IN ('UK', 'DE')"),
        FragmentConfig("orders_apac", "postgresql://db3/orders", "region IN ('JP', 'AU')"),
    ]
    
    reconstructor = DistributedReconstructor(fragments)
    await reconstructor.initialize()
    
    try:
        # Parallel reconstruction with predicates
        async for order in reconstructor.reconstruct_horizontal(
            {"region": "US", "status": "pending"},
            ["order_id", "customer_id", "amount"]
        ):
            print(f"Order: {order}")
    finally:
        await reconstructor.close()
 
# asyncio.run(main())