Distributed Query System with Join Enrichment and Enhanced Streaming

[olsonanl]

Summary

This PR introduces a comprehensive distributed query system along with join enrichment capabilities, improved streaming with backpressure, and new media format support. The changes enable:

• Distributed Query System: Parallel querying across Solr shards for improved performance on large result sets
• Join Enrichment: Post-query field enrichment from related collections (e.g., genome fields in genome_feature queries)
• Enhanced Streaming: Native Node.js streams with proper backpressure handling, replacing event-stream
• Genbank Serializer: New media format for genome export
• Enhanced FASTA Serializers: Streaming with metadata enrichment and configurable headers

Statistics

• 80 commits ahead of alpha
• 72 files changed: 18,160 insertions, 6,756 deletions
• New components: 15+ new files in lib/distributed/
• Test coverage: Comprehensive test suites in tests/test-distributed/ and tests/test-join/

---

Major Components

1. Distributed Query System (lib/distributed/)

Component	Purpose
DistributedQueryManager.js	High-level orchestrator for distributed queries
ParallelQueryCoordinator.js	Unordered parallel query execution across shards
MergeSortStream.js	K-way merge sort for globally sorted output
ShardCursorStream.js	Cursor-based pagination per shard
SolrClusterClient.js	Cluster state discovery and shard management
DirectSolrClient.js	Direct shard replica queries (bypasses load balancer)
DistributedQueryConfig.js	Runtime configuration management

Decision flow for distributed queries:

1. Collection must be in enabledCollections (or not in disabledCollections)
2. Request limit must exceed minLimitThreshold (default: 10,000)
3. Method must be query or stream
4. Can be overridden via X-Distributed-Query header or ?distributed= param

2. Join Enrichment System

Component	Purpose
middleware/JoinFieldInjector.js	Pre-query: injects join key fields into fl=
middleware/JoinEnrichment.js	Post-query: fetches and merges related data
lib/BatchJoiner.js	Efficient batch lookup infrastructure with caching

Supported joins:

genome_feature → genome: genome_name, taxon_id, genome_status, strain
pathway → genome: genome_name, taxon_id
subsystem → genome: genome_name, taxon_id
sp_gene → genome: genome_name, taxon_id
genome_amr → genome: genome_name, taxon_id

3. Enhanced Media Handlers

• Backpressure handling: All streaming handlers now use util/streamWithBackpressure.js
• FASTA improvements: Configurable headers via http_fasta_* parameters, genome metadata enrichment
• Genbank format: New media/genbank.js for genome export with multi-contig support

---

Middleware Chain Changes

// NEW middleware chain in routes/dataType.js
router.use([
  RQLQueryParser,
  DecorateQuery,
  Limiter,
  JoinFieldInjector,    // NEW: Injects join key fields
  DistributedQuery,     // NEW: Routes large queries to distributed system
  ShardsPreference,
  // ...
  APIMethodHandler,     // MODIFIED: Skip logic for distributed queries
  reqCounter,
  ExtractCustomFields,
  ContentRange,
  JoinEnrichment,       // NEW: Post-query field enrichment
  media
])

---

Risk Assessment

🔴 HIGH RISK: Distributed Query Bypass

Impact: Queries meeting threshold criteria bypass standard APIMethodHandler and execute through the new distributed system.

Trigger conditions (all must be true):

• distributedQuery.enabled: true (default)
• Query limit ≥ 10,000 rows (configurable)
• Collection not in disabledCollections
• Method is query or stream

Risks:

1. Untested code path for production workloads
2. Forces streaming mode for large query requests
3. Requires SSL/TLS configuration for direct shard connections
4. Different error handling and response characteristics

Mitigation:

{
  "distributedQuery": {
    "enabled": false
  }
}

Or per-request: X-Distributed-Query: false header

---

🟡 MEDIUM RISK: Media Handler Changes

Impact: All streaming responses use new backpressure-aware implementation.

Changes:

• event-stream.mapSync() → streamWithBackpressure()
• Promise-based error handling
• Different first-doc metadata handling

Affected formats: JSON, CSV, TSV, GFF, NDJSON, FASTA

Testing required:

• Small result sets (<100 docs)
• Large result sets (>10,000 docs)
• Error conditions (Solr timeout, network failure)
• Client disconnect handling

---

🟡 MEDIUM RISK: Join Enrichment

Impact: Queries requesting joinable fields trigger secondary Solr lookups.

Risks:

1. Additional latency from secondary queries
2. Requires DirectSolrClient infrastructure
3. Response contains fields not in source collection

Mitigation:

{
  "joinEnrichment": {
    "enabled": false
  }
}

---

🟢 LOW RISK: http-params Changes

Impact: New parameter namespaces http_fasta_* and http_genbank_* extracted before Solr query.

Risk: Minimal - new parameters only, no existing behavior change.

---

Configuration

Add to p3api.conf:

{
  "distributedQuery": {
    "enabled": false,
    "minLimitThreshold": 10000,
    "enabledCollections": [],
    "maxParallelism": 8,
    "cursorBatchSize": 2000,
    "rejectUnauthorized": false,
    "ca": "/path/to/ca-cert.pem"
  },
  "joinEnrichment": {
    "enabled": true,
    "cacheSize": 200
  }
}

---

Deployment Recommendations

Phase 1: Safe Merge

1. Merge with distributedQuery.enabled: false
2. Join enrichment and media handler improvements become active
3. Test FASTA, CSV, and JSON exports thoroughly

Phase 2: Gradual Distributed Query Enablement

1. Set enabledCollections: ["genome"] (single collection)
2. Monitor for errors and performance differences
3. Expand collection list as confidence builds

Phase 3: Full Enablement

1. Set enabled: true with appropriate threshold
2. Consider starting with threshold of 25,000+

---

Test Plan

• Run existing test suite: npm test
• Run distributed query tests: npx mocha tests/test-distributed/
• Run join enrichment tests: npx mocha tests/test-join/
• Manual testing of FASTA exports with genome metadata
• Manual testing of Genbank exports
• Load testing with large result sets
• Verify error handling on Solr failures

---

Documentation

New documentation added:

• Docs/DISTRIBUTED_QUERY_DOCS.md - System overview
• Docs/DISTRIBUTED_QUERY_QUICKSTART.md - Quick start guide
• Docs/DISTRIBUTED_QUERY_SPEC.md - Technical specification
• API_REFERENCE.md - API documentation

---

🤖 Generated with Claude Code

[olsonanl]

Configuration File Changes

This PR modifies the configuration structure in config.js and p3api.conf.sample. Here are the detailed changes:

---

Removed: collectionUniqueKeys

The hardcoded collectionUniqueKeys mapping has been removed from config.js. This included 35 collection-to-unique-key mappings:

// REMOVED from config.js
collectionUniqueKeys: {
  antibiotics: 'pubchem_cid',
  genome: 'genome_id',
  genome_feature: 'feature_id',
  taxonomy: 'taxon_id',
  // ... 31 more entries
}

Why removed: The distributed query system now dynamically discovers unique keys from Solr schema via SolrClusterClient.getUniqueKey(), making static configuration unnecessary.

---

Added: distributedQuery Section

New configuration block controlling the distributed query system:

distributedQuery: {
  // Integration settings
  enabled: true,                    // Master switch for distributed queries
  minLimitThreshold: 10000,         // Minimum rows to trigger distributed mode
  enabledCollections: ['genome_feature'],  // Whitelist (empty = all allowed)
  disabledCollections: [],          // Blacklist
  exposeMetadataHeaders: true,      // Add X-Distributed-Query headers to response

  // Performance tuning
  maxParallelism: 8,                // Max concurrent shard queries
  cursorBatchSize: 2000,            // Docs per cursor fetch

  // Retry configuration
  maxRetries: 3,                    // Retry attempts on shard failure
  initialRetryDelayMs: 100,         // Initial backoff delay

  // Cache TTLs
  schemaCacheTTLMinutes: 60,        // Schema cache lifetime
  clusterStatusCacheTTLSeconds: 60, // Cluster state cache lifetime

  // Memory limits
  maxMergeSortHeapDocs: 10000,      // Max docs in merge-sort heap
  maxMemoryMB: 32,                  // Memory limit for query operations

  // SSL/TLS for direct shard connections
  rejectUnauthorized: true,         // Set false for self-signed certs
  ca: null,                         // Path to CA cert file or PEM content

  // Administration
  excludeNodes: [],                 // Regex patterns for nodes to exclude
  adminUsers: []                    // Users allowed to modify runtime config
}

---

Added: joinEnrichment Section

New configuration block for post-query field enrichment:

joinEnrichment: {
  enabled: true,           // Master switch for join enrichment
  cacheSize: 200,          // LRU cache size per target collection
  collections: {
    // Define joinable fields per source collection
    genome_feature: {
      joinableFields: {
        genome_name: { from: 'genome', via: 'genome_id', field: 'genome_name' },
        taxon_id: { from: 'genome', via: 'genome_id', field: 'taxon_id' },
        genome_status: { from: 'genome', via: 'genome_id', field: 'genome_status' },
        strain: { from: 'genome', via: 'genome_id', field: 'strain' }
      }
    },
    pathway: {
      joinableFields: {
        genome_name: { from: 'genome', via: 'genome_id', field: 'genome_name' },
        taxon_id: { from: 'genome', via: 'genome_id', field: 'taxon_id' }
      }
    },
    subsystem: { /* similar */ },
    sp_gene: { /* similar */ },
    genome_amr: { /* similar */ }
  }
}

Join field structure:

• from: Source collection to fetch from (e.g., genome)
• via: Join key field present in both collections (e.g., genome_id)
• field: Field to fetch from source collection

---

Added: sequenceJoin Section (in sample config)

Configuration for FASTA sequence lookups:

sequenceJoin: {
  batchSize: 200,        // Docs per sequence lookup batch
  prefetchBatches: 2     // Number of batches to prefetch
}

---

Added: genomeMetadataCache Section (in sample config)

Configuration for genome metadata caching in FASTA/Genbank serializers:

genomeMetadataCache: {
  maxSize: 100,          // Max cached genome records
  batchSize: 50          // Genomes per lookup batch
}

---

Recommended Production Configuration

For a safe initial deployment, use these settings:

{
  "distributedQuery": {
    "enabled": false,
    "minLimitThreshold": 25000,
    "enabledCollections": [],
    "rejectUnauthorized": false,
    "ca": "/path/to/solr-ca.pem"
  },
  "joinEnrichment": {
    "enabled": true,
    "cacheSize": 500
  },
  "sequenceJoin": {
    "batchSize": 200,
    "prefetchBatches": 2
  }
}

This disables distributed queries initially while enabling the join enrichment feature which has lower risk.