Index Updates and Delta Indexing

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

// 1. Full document update (replaces entire document)
PUT /products/_doc/12345
{
  "product_id": "12345",
  "name": "Wireless Headphones",
  "price": 79.99,
  "inventory": 150,
  "category": "electronics",
  "last_updated": "2024-01-15T10:30:00Z"
}
 
// 2. Partial update (preferred for single field changes)
// Still rewrites document internally, but API is simpler
POST /products/_update/12345
{
  "doc": {
    "price": 69.99,
    "last_updated": "2024-01-15T11:00:00Z"
  }
}
 
// 3. Scripted update (for computed changes)
POST /products/_update/12345
{
  "script": {
    "source": "ctx._source.inventory -= params.sold",
    "params": { "sold": 5 }
  }
}
 
// 4. Upsert (update or insert if not exists)
POST /products/_update/12345
{
  "doc": {
    "price": 69.99,
    "last_updated": "2024-01-15T11:00:00Z"
  },
  "upsert": {
    "product_id": "12345",
    "name": "New Product",
    "price": 69.99,
    "inventory": 100,
    "last_updated": "2024-01-15T11:00:00Z"
  }
}
 
// 5. Bulk update (efficient for multiple documents)
POST /_bulk
{"update": {"_index": "products", "_id": "12345"}}
{"doc": {"price": 69.99}}
{"update": {"_index": "products", "_id": "12346"}}
{"doc": {"price": 89.99}}
{"update": {"_index": "products", "_id": "12347"}}
{"doc": {"price": 49.99}}

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
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206

"""
Production Delta Indexing Pipeline
 
This implementation demonstrates patterns used at companies
processing millions of document updates per hour.
"""
 
from dataclasses import dataclass
from datetime import datetime, timedelta
from typing import Generator, Optional
import json
 
 
@dataclass
class ChangeEvent:
    """Represents a single change in the source system."""
    event_id: str          # Unique identifier for idempotency
    entity_type: str       # e.g., 'product', 'user', 'order'
    entity_id: str         # Primary key of changed entity
    operation: str         # 'INSERT', 'UPDATE', 'DELETE'
    timestamp: datetime    # When the change occurred
    payload: Optional[dict]  # Changed data (may be partial)
 
 
@dataclass
class IndexCheckpoint:
    """Tracks progress through the change stream."""
    last_event_id: str
    last_timestamp: datetime
    events_processed: int
    last_updated: datetime
 
 
class DeltaIndexingPipeline:
    """
    Coordinates delta indexing from change detection through
    bulk indexing with fault tolerance and checkpointing.
    """
    
    def __init__(
        self,
        change_source: 'ChangeSource',
        document_enricher: 'DocumentEnricher', 
        search_client: 'SearchClient',
        checkpoint_store: 'CheckpointStore',
        batch_size: int = 1000,
        max_batch_wait_seconds: int = 5
    ):
        self.change_source = change_source
        self.enricher = document_enricher
        self.search = search_client
        self.checkpoints = checkpoint_store
        self.batch_size = batch_size
        self.max_wait = max_batch_wait_seconds
        
    async def run(self) -> None:
        """
        Main loop: consume changes, enrich, transform, index.
        
        This loop is designed to run continuously in production,
        handling backpressure and failures gracefully.
        """
        while True:
            checkpoint = await self.checkpoints.load()
            
            try:
                async for batch in self._consume_batches(checkpoint):
                    # Separate deletes from upserts
                    deletes = [c for c in batch if c.operation == 'DELETE']
                    upserts = [c for c in batch if c.operation != 'DELETE']
                    
                    # Process upserts: enrich, transform, index
                    if upserts:
                        documents = await self._process_upserts(upserts)
                        await self._bulk_index(documents)
                    
                    # Process deletes: remove from index
                    if deletes:
                        await self._bulk_delete(
                            [c.entity_id for c in deletes]
                        )
                    
                    # Checkpoint after successful batch
                    new_checkpoint = IndexCheckpoint(
                        last_event_id=batch[-1].event_id,
                        last_timestamp=batch[-1].timestamp,
                        events_processed=(
                            checkpoint.events_processed + len(batch)
                        ),
                        last_updated=datetime.utcnow()
                    )
                    await self.checkpoints.save(new_checkpoint)
                    checkpoint = new_checkpoint
                    
            except Exception as e:
                # Log error, wait, retry from checkpoint
                await self._handle_failure(e, checkpoint)
                
    async def _consume_batches(
        self, 
        checkpoint: IndexCheckpoint
    ) -> Generator[list[ChangeEvent], None, None]:
        """
        Consume changes in batches, balancing latency and throughput.
        
        Uses time-based batching: emit batch when either size limit
        or time limit is reached, whichever comes first.
        """
        batch = []
        batch_start = datetime.utcnow()
        
        async for change in self.change_source.stream(checkpoint):
            batch.append(change)
            
            batch_age = (datetime.utcnow() - batch_start).seconds
            
            if len(batch) >= self.batch_size or batch_age >= self.max_wait:
                yield batch
                batch = []
                batch_start = datetime.utcnow()
                
        # Emit remaining items
        if batch:
            yield batch
            
    async def _process_upserts(
        self, 
        changes: list[ChangeEvent]
    ) -> list[dict]:
        """
        Enrich and transform changes into index-ready documents.
        
        Handles the common case where change events contain only IDs,
        requiring enrichment from the source database.
        """
        # Deduplicate: if same entity changed multiple times in batch,
        # only process the latest
        latest_changes = {}
        for change in changes:
            key = f"{change.entity_type}:{change.entity_id}"
            existing = latest_changes.get(key)
            if not existing or change.timestamp > existing.timestamp:
                latest_changes[key] = change
                
        # Enrich: fetch full documents from source
        enriched = await self.enricher.enrich_batch(
            list(latest_changes.values())
        )
        
        # Transform: convert to search schema
        documents = []
        for entity_id, data in enriched.items():
            doc = self._transform_to_search_schema(data)
            documents.append(doc)
            
        return documents
        
    async def _bulk_index(self, documents: list[dict]) -> None:
        """
        Write documents to search index using bulk API.
        
        Implements retry with exponential backoff for transient failures.
        """
        for attempt in range(3):
            try:
                result = await self.search.bulk_index(documents)
                
                if result.errors:
                    # Handle partial failures
                    failed_ids = [
                        item['_id'] 
                        for item in result.items 
                        if 'error' in item
                    ]
                    raise PartialIndexingError(failed_ids)
                    
                return
                
            except TransientError as e:
                wait = 2 ** attempt
                await asyncio.sleep(wait)
                
        raise IndexingError("Max retries exceeded")
        
    async def _bulk_delete(self, entity_ids: list[str]) -> None:
        """Remove deleted entities from search index."""
        await self.search.bulk_delete(entity_ids)
        
    def _transform_to_search_schema(self, data: dict) -> dict:
        """
        Map source data to search index schema.
        
        This is where you apply field mappings, compute derived fields,
        and normalize data for search.
        """
        return {
            '_id': data['id'],
            'name': data['name'],
            'description': data.get('description', ''),
            'price': float(data['price']),
            'category': data['category']['name'],
            'brand': data.get('brand', {}).get('name'),
            'in_stock': data.get('inventory', 0) > 0,
            'popularity_score': self._compute_popularity(data),
            'indexed_at': datetime.utcnow().isoformat()
        }

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
146
147
148
149
150
151

/**
 * Performance-optimized delta indexing with parallel processing.
 * 
 * Achieves 50,000+ documents/second on typical hardware.
 */
 
interface PerformanceConfig {
  // Parallelism settings
  enrichmentConcurrency: number;    // Parallel enrichment workers
  indexingConcurrency: number;      // Parallel bulk request streams
  
  // Batch settings
  enrichmentBatchSize: number;      // IDs per enrichment query
  indexingBatchSize: number;        // Documents per bulk request
  
  // Caching settings
  relationCacheTTLSeconds: number;  // TTL for category/brand cache
  relationCacheMaxSize: number;     // Max cached relations
}
 
const PRODUCTION_CONFIG: PerformanceConfig = {
  enrichmentConcurrency: 8,
  indexingConcurrency: 4,
  enrichmentBatchSize: 500,
  indexingBatchSize: 2000,
  relationCacheTTLSeconds: 300,
  relationCacheMaxSize: 100_000,
};
 
class OptimizedDeltaPipeline {
  private enrichmentSemaphore: Semaphore;
  private indexingSemaphore: Semaphore;
  private relationCache: LRUCache<string, Relation>;
  
  constructor(private config: PerformanceConfig) {
    this.enrichmentSemaphore = new Semaphore(config.enrichmentConcurrency);
    this.indexingSemaphore = new Semaphore(config.indexingConcurrency);
    this.relationCache = new LRUCache({
      maxSize: config.relationCacheMaxSize,
      ttl: config.relationCacheTTLSeconds * 1000,
    });
  }
  
  /**
   * Process changes with optimal parallelism.
   * 
   * Uses a pipeline pattern: enrichment and indexing
   * happen concurrently on different batches.
   */
  async processBatch(changes: ChangeEvent[]): Promise<void> {
    // Split into enrichment batches
    const enrichmentBatches = chunk(
      changes, 
      this.config.enrichmentBatchSize
    );
    
    // Process enrichment batches in parallel
    const enrichedDocs = await Promise.all(
      enrichmentBatches.map(batch => 
        this.enrichmentSemaphore.runExclusive(() =>
          this.enrichBatch(batch)
        )
      )
    );
    
    // Flatten and split into indexing batches
    const allDocs = enrichedDocs.flat();
    const indexingBatches = chunk(allDocs, this.config.indexingBatchSize);
    
    // Index batches in parallel
    await Promise.all(
      indexingBatches.map(batch =>
        this.indexingSemaphore.runExclusive(() =>
          this.indexBatch(batch)
        )
      )
    );
  }
  
  /**
   * Batch enrichment with caching for relations.
   */
  private async enrichBatch(changes: ChangeEvent[]): Promise<Document[]> {
    const entityIds = changes.map(c => c.entity_id);
    
    // Single query for all entities
    const entities = await this.db.query(`
      SELECT * FROM products 
      WHERE id = ANY($1::uuid[])
    `, [entityIds]);
    
    // Collect unique relation IDs
    const categoryIds = new Set(entities.map(e => e.category_id));
    const brandIds = new Set(entities.map(e => e.brand_id).filter(Boolean));
    
    // Fetch uncached relations
    const uncachedCategoryIds = [...categoryIds].filter(
      id => !this.relationCache.has(`category:${id}`)
    );
    const uncachedBrandIds = [...brandIds].filter(
      id => !this.relationCache.has(`brand:${id}`)
    );
    
    if (uncachedCategoryIds.length > 0) {
      const categories = await this.db.query(
        'SELECT * FROM categories WHERE id = ANY($1::uuid[])',
        [uncachedCategoryIds]
      );
      categories.forEach(c => 
        this.relationCache.set(`category:${c.id}`, c)
      );
    }
    
    if (uncachedBrandIds.length > 0) {
      const brands = await this.db.query(
        'SELECT * FROM brands WHERE id = ANY($1::uuid[])',
        [uncachedBrandIds]
      );
      brands.forEach(b => 
        this.relationCache.set(`brand:${b.id}`, b)
      );
    }
    
    // Build enriched documents
    return entities.map(entity => ({
      ...entity,
      category: this.relationCache.get(`category:${entity.category_id}`),
      brand: entity.brand_id 
        ? this.relationCache.get(`brand:${entity.brand_id}`) 
        : null,
    }));
  }
  
  /**
   * Bulk index with compression and connection reuse.
   */
  private async indexBatch(documents: Document[]): Promise<void> {
    const body = documents.flatMap(doc => [
      { index: { _index: 'products', _id: doc.id } },
      this.transformDocument(doc)
    ]);
    
    await this.esClient.bulk({
      body,
      refresh: false,  // Don't refresh after each bulk
      timeout: '30s',
      // Enable request compression
      headers: { 'Content-Encoding': 'gzip' }
    });
  }
}

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

// Debezium PostgreSQL connector configuration
// Captures changes from products, categories, and brands tables
 
{
  "name": "search-indexing-connector",
  "config": {
    // Connector class for PostgreSQL
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    
    // Database connection
    "database.hostname": "primary.db.example.com",
    "database.port": "5432",
    "database.user": "replication_user",
    "database.password": "${secrets.db_password}",
    "database.dbname": "ecommerce",
    
    // Logical replication slot name
    "slot.name": "search_indexing_slot",
    "plugin.name": "pgoutput",
    
    // What to capture
    "table.include.list": "public.products,public.categories,public.brands",
    
    // Kafka topic routing
    "topic.prefix": "cdc",
    // Produces topics: cdc.public.products, cdc.public.categories, etc.
    
    // Schema handling
    "key.converter": "org.apache.kafka.connect.json.JsonConverter",
    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
    "key.converter.schemas.enable": "false",
    "value.converter.schemas.enable": "true",
    
    // Snapshot mode: what to do on first start
    "snapshot.mode": "initial",  // Take snapshot then stream changes
    
    // Heartbeat for health monitoring
    "heartbeat.interval.ms": "10000",
    
    // Transforms for convenience
    "transforms": "unwrap",
    "transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
    "transforms.unwrap.add.fields": "op,ts_ms",
    "transforms.unwrap.delete.handling.mode": "rewrite"
  }
}
 
// Sample Debezium change event (after ExtractNewRecordState transform)
{
  "id": "12345",
  "name": "Wireless Headphones",
  "price": 79.99,
  "category_id": "cat-001",
  "brand_id": "brand-abc",
  "updated_at": "2024-01-15T10:30:00.000000Z",
  "__op": "u",       // 'c' = create, 'u' = update, 'd' = delete
  "__ts_ms": 1705315800000,
  "__deleted": "false"
}