Introduction

The Challenge

• Multiple tenants with parent-child relationships
• Complex cross-tenant associations and polymorphic relationships
• Several huge tables (10+ million rows each)
• Critical data that must be preserved during tenant extraction
• Extraction and import operations that were becoming prohibitively slow

System Architecture Overview

┌─────────────────┐     ┌─────────────────┐     ┌─────────────────┐
│  CLI Interface  │────▶│  Extract Runner │────▶│     Exporter    │
└─────────────────┘     └─────────────────┘     └─────────────────┘
                                                        │
                                                        ▼
┌─────────────────┐     ┌─────────────────┐     ┌─────────────────┐
│  SQL Generator  │◀────│  Database Util  │◀────│  Query Config   │
└─────────────────┘     └─────────────────┘     └─────────────────┘

• CLI Interface: Provides command-line options and argument parsing
• Extract Runner: Orchestrates the overall extraction process
• Exporter: Handles the core data export logic
• Database Util: Provides database connectivity and utility functions
• Query Config: Stores SQL queries in YAML for maintainability
• SQL Generator: Creates SQL scripts for extraction and import

My Approach: Smart Data Filtering for Tenant Extraction

1. Audit Logs Table

# In queries.yml - Using CTEs to filter logs by entity type
activity_logs: |
  WITH type_a_logs AS (
    SELECT logs.* FROM activity_logs logs 
    WHERE logs.entity_type = 'TypeA' 
    AND logs.entity_id IN (
      SELECT id FROM entity_a WHERE tenant_id IN (
        SELECT id FROM tenants WHERE parent_id IN (SELECT id FROM extraction_tenant_ids)
      )
    )
    ORDER BY logs.id DESC LIMIT 100000
  ),
  type_b_logs AS (
    SELECT logs.* FROM activity_logs logs 
    WHERE logs.entity_type = 'TypeB' 
    AND logs.entity_id IN (
      SELECT id FROM entity_b WHERE tenant_id IN (
        SELECT id FROM tenants WHERE parent_id IN (SELECT id FROM extraction_tenant_ids)
      )
    )
    ORDER BY logs.id DESC LIMIT 100000
  ),
  type_c_logs AS (
    SELECT logs.* FROM activity_logs logs 
    WHERE logs.entity_type = 'TypeC' 
    AND logs.entity_id IN (
      SELECT id FROM entity_c WHERE owner_id IN (
        SELECT owner_id FROM tenants WHERE id IN (SELECT id FROM extraction_tenant_ids)
      )
    )
    ORDER BY logs.id DESC LIMIT 100000
  )
  SELECT * FROM type_a_logs
  UNION ALL
  SELECT * FROM type_b_logs
  UNION ALL
  SELECT * FROM type_c_logs

• Creates separate CTEs for each entity type I care about (TypeA, TypeB, TypeC)
• For each entity type, find only entities related to the tenant being extracted
• Limits to the 100,000 most recent logs per entity type
• Combine the results with UNION ALL

2. Historical Transaction Table

# In queries.yml - Filtering to only recent transactions
transaction_history: "SELECT * FROM transaction_history WHERE tenant_id IN (SELECT id FROM extraction_tenant_ids) AND created_at > (NOW() - INTERVAL '1 month')"

3. Worker Task Records

# In queries.yml - Limiting to the most recent task records
worker_tasks: "SELECT * FROM worker_tasks WHERE tenant_id IN (SELECT id FROM extraction_tenant_ids) ORDER BY created_at DESC LIMIT 10000"

Technical Implementation Details

Table Selection Logic

# First, get tables with direct tenant relationships
tenant_tables = db.tables_with_column("tenant_id")

# Next, get tables with secondary relationships
secondary_tables = db.tables_with_column("secondary_id")

# Process standard tables with simple filtering
tenant_tables.each do |table_name|
  next if table_name == "tenants" # Already processed
  process_table(f, table_name)
end

# Process special case tables with custom queries
SPECIAL_TABLES.each do |table_name|
  process_table(f, table_name)
end

Batched Processing for Large Tables

def export_large_table(table_name, query)
  batch_size = 10_000
  offset = 0
  total_rows = 0
  
  loop do
    batch_query = "#{query} LIMIT #{batch_size} OFFSET #{offset}"
    rows = db.connection.exec(batch_query)
    break if rows.count == 0
    
    # Process this batch
    export_rows_to_csv(table_name, rows)
    
    total_rows += rows.count
    offset += batch_size
    puts "  Exported #{total_rows} rows from #{table_name}"
  end
end

Parallelization Strategy for Performance

Parallel Export Process

def process_tables_in_parallel(tables)
  puts "Processing #{tables.size} tables in parallel with #{@parallel_workers} workers..."
  
  Parallel.each(tables, in_processes: @parallel_workers) do |table_name|
    begin
      # Create a separate exporter instance for this table
      table_exporter = TableExporter.new(@db, @config, table_name, @output_dir, @progress_file, @queries)
      table_exporter.process
    rescue => e
      puts "Error processing table #{table_name}: #{e.message}"
      # Log the error but continue with other tables
      next
    end
  end
end

1. Dynamic Worker Count: The number of parallel workers is configurable based on the available CPU cores and memory.
2. Independent Table Exporters: Each table gets its own exporter instance to avoid sharing state.
3. Isolation of Failures: Errors in one table's export won't affect other tables.
4. Progress Tracking: Each exporter updates a shared progress file to enable resuming interrupted operations.

Parallel Import Strategy

def import_data_in_parallel
  # First, import tables that other tables depend on
  import_priority_tables

  # Get remaining tables to import
  remaining_tables = @tables - @processed_tables
  
  # Group tables for parallel import
  table_groups = create_table_groups(remaining_tables)
  
  table_groups.each_with_index do |group, index|
    puts "Importing table group #{index + 1}/#{table_groups.size} (#{group.size} tables)..."
    
    Parallel.each(group, in_processes: @parallel_workers) do |table|
      import_table(table)
    end
  end
end

def create_table_groups(tables)
  # Use foreign key relationships to group tables that can be safely imported in parallel
  groups = []
  dependency_graph = build_dependency_graph(tables)
  
  until dependency_graph.empty?
    # Find tables with no dependencies
    independent_tables = dependency_graph.keys.select { |t| dependency_graph[t].empty? }
    groups << independent_tables
    
    # Remove these tables from the graph
    independent_tables.each { |t| dependency_graph.delete(t) }
    
    # Remove these tables from dependencies
    dependency_graph.each { |_, deps| deps.subtract(independent_tables) }
  end
  
  groups
end

1. Dependency-Aware Parallelization: Tables are grouped based on their foreign key relationships to avoid constraint violations.
2. Priority Tables First: Core tables that other tables depend on are imported first, sequentially.
3. Group-Based Processing: Tables are imported in waves, each containing tables that can be safely imported in parallel.
4. Transaction Isolation: Each table import runs in its own transaction to prevent deadlocks.

Improving Error Resilience for Tenant Imports

# Breaking sequence updates into batches with error handling
def update_sequences
  puts "Updating sequences..."
  batch_size = 50

  sequences.each_slice(batch_size) do |batch|
    # Create a transaction for each batch
    db.connection.exec("BEGIN;")
    
    batch.each do |sequence|
      begin
        # Update each sequence to match the imported data
        sql = "SELECT setval('#{sequence[:name]}', COALESCE((SELECT MAX(#{sequence[:column]}) FROM #{sequence[:table]}), 1), true)"
        db.connection.exec(sql)
        puts "  Updated sequence #{sequence[:name]}"
      rescue => e
        # Log error but continue with other sequences
        puts "  Error updating sequence #{sequence[:name]}: #{e.message}"
      end
    end
    
    # Commit the batch
    db.connection.exec("COMMIT;")
  end
  
  puts "Sequences updated."
end

• Processes sequences in batches to avoid overwhelming the database
• It uses individual error handling so a single failure doesn't stop the entire process
• Wraps each batch in a transaction for better performance

def run_post_import_finalization
  begin
    # Re-enable constraints in a separate transaction
    db.connection.exec("SET session_replication_role = 'origin';")
    puts "Re-enabled database constraints"
    
    # Update sequences to match imported data
    update_sequences
    
    # Refresh materialized views if they exist
    if table_exists?("materialized_views_meta")
      puts "Refreshing materialized views..."
      db.connection.exec("SELECT refresh_all_materialized_views()")
    end
    
    puts "Post-import finalization completed successfully"
    true
  rescue => e
    puts "Error during post-import finalization: #{e.message}"
    puts "Attempting to continue despite error..."
    # Still try to enable constraints even if other steps fail
    begin
      db.connection.exec("SET session_replication_role = 'origin';")
    rescue => inner_e
      puts "Fatal error: Failed to re-enable constraints: #{inner_e.message}"
    end
    false
  end
end

Architecture Lessons

1. Be selective about data: Not all historical data is necessary for a functioning tenant environment.
2. Use database features effectively: CTEs, window functions, and other PostgreSQL features can help you filter data efficiently.
3. Prioritize error resilience: Assume failures will happen and design your system to continue despite them.
4. Batch operations for large datasets: Breaking work into smaller transactions reduces memory usage and improves reliability.
5. Configuration-driven design: Storing queries in a YAML file rather than hardcoding them made the system more maintainable and adaptable.
6. Progress tracking and resumability: Storing progress information allowed the system to resume interrupted extractions, a crucial feature for large tenants.

Conclusion