# feat\(database\): reimplement parallelization in batching \(\#15\) · flambo/atlantis

[View on GitCafe](https://git.cafe/flambo/atlantis/commit/16ebb0b541b3c46476a1c326ab6a91680f9b00ef)

Repository: [flambo/atlantis](https://git.cafe/flambo/atlantis)

Visibility: public

Requested revision: 16ebb0b541b3c46476a1c326ab6a91680f9b00ef

Requested commit: 16ebb0b541b3c46476a1c326ab6a91680f9b00ef

Commit: 16ebb0b541b3c46476a1c326ab6a91680f9b00ef

Tree: deab3ea42d8013436f0a5127d53812386db6798f

Author: oliver

Committer: GitHub

## Message

````
feat(database): reimplement parallelization in batching (#15)

<!-- greptile_comment -->

<h3>Greptile Summary</h3>


This PR reimplements parallelization across all five database processing modules (`flow_db.py`, `ip_db.py`, `protocol_db.py`, `spectrum_db.py`, `structure_db.py`) by introducing a consistent worker-batch pattern that separates file processing from database operations.

**Key changes:**
- Refactored each module to use `multiprocessing.Pool` with configurable `MAX_WORKERS` and `BATCH_SIZE`
- Separated processing logic into `process_file_worker()` functions that run without DB connections
- Consolidated DB operations into `batch_insert_results()` functions that handle transaction commits per batch
- Added new `batch_mark_processed()` function in `discovery.py` for batch status updates
- Removed unused `aggregate_buckets()` functions from `ip_db.py` and `protocol_db.py`
- Changed `compute_spectrum()` and `compute_structure_function()` signatures to accept lists instead of sets

**Performance improvements:**
- Parallel processing of files within batches using process pools
- Batch database commits reduce transaction overhead
- Improved logging with batch progress indicators and table name prefixes

<h3>Confidence Score: 5/5</h3>


- This PR is safe to merge with minimal risk
- The refactoring follows a consistent pattern across all modules, properly separates concerns between processing and database operations, and maintains backward compatibility. The code uses validated parameters and proper error handling throughout.
- No files require special attention

<h3>Important Files Changed</h3>




| Filename | Overview |
|----------|----------|
| netflow-db/discovery.py | Added `batch_mark_processed` function for batch status updates; uses validated table names with f-string interpolation for column names |
| netflow-db/flow_db.py | Refactored to use multiprocessing with worker/batch pattern; processing logic separated from DB operations |
| netflow-db/ip_db.py | Refactored to parallel batch processing; removed unused `aggregate_buckets` function |
| netflow-db/protocol_db.py | Refactored to parallel batch processing; removed unused `aggregate_buckets` function |
| netflow-db/spectrum_db.py | Refactored to parallel batch processing with worker pattern; `compute_spectrum` now accepts list instead of set |
| netflow-db/structure_db.py | Refactored to parallel batch processing with worker pattern; `compute_structure_function` now accepts list instead of set |

</details>



<h3>Sequence Diagram</h3>

```mermaid
sequenceDiagram
    participant Main as process_pending_files()
    participant Pool as Multiprocessing Pool
    participant Worker as process_file_worker()
    participant Batch as batch_insert_results()
    participant Mark as batch_mark_processed()
    participant DB as SQLite Database
    
    Main->>DB: get_files_needing_processing()
    DB-->>Main: pending files list
    
    loop For each batch (BATCH_SIZE)
        Main->>Pool: Create Pool(MAX_WORKERS)
        Pool->>Worker: Map tasks to workers (parallel)
        
        par Parallel Processing
            Worker->>Worker: Extract file_path, router, timestamp
            Worker->>Worker: Check if file exists
            alt File exists
                Worker->>Worker: Run nfdump/subprocess
                Worker->>Worker: Parse output
                Worker-->>Pool: Return result (success, data)
            else File missing (gap)
                Worker->>Worker: Create zero/empty data
                Worker-->>Pool: Return result (success, placeholder)
            end
        end
        
        Pool-->>Main: Batch results
        Main->>Batch: batch_insert_results(conn, results)
        
        loop For each successful result
            Batch->>DB: INSERT OR REPLACE stats
        end
        Batch->>DB: commit()
        
        Main->>Mark: batch_mark_processed(conn, table_name, results)
        
        loop For each result
            Mark->>DB: UPDATE processed_files SET status
        end
        Mark->>DB: UPDATE processed_at (fully processed files)
        Mark->>DB: commit()
        
        Main->>Main: Update stats counters
    end
    
    Main-->>Main: Return stats
```

<!-- greptile_other_comments_section -->

<!-- /greptile_comment -->
````

## Parents

- [2170a101181c89da09f2c3a17edf528707c0ef8d](https://git.cafe/flambo/atlantis/commit/2170a101181c89da09f2c3a17edf528707c0ef8d?format=markdown)

- [d5246f289038af916ea3867458f325e40c33258f](https://git.cafe/flambo/atlantis/commit/d5246f289038af916ea3867458f325e40c33258f?format=markdown)

[Source at this commit](https://git.cafe/flambo/atlantis/tree/16ebb0b541b3c46476a1c326ab6a91680f9b00ef?format=markdown)