Processing Pipeline API

Namespace: fq::processing

---

ProcessingPipelineInterface

Abstract interface for processing pipeline, hiding concrete implementation (factory pattern).

Factory Creation

auto pipeline = fq::processing::createProcessingPipeline();

Interface Methods

class ProcessingPipelineInterface {
public:
    virtual void setInputPath(const std::string& path) = 0;
    virtual void setOutputPath(const std::string& path) = 0;
    virtual void setProcessingConfig(const ProcessingConfig& config) = 0;
    virtual void addReadPredicate(std::unique_ptr<ReadPredicateInterface> predicate) = 0;
    virtual void addReadMutator(std::unique_ptr<ReadMutatorInterface> mutator) = 0;
    virtual auto run() -> ProcessingStatistics = 0;
};

---

ProcessingConfig

Pipeline configuration parameters.

Parameter	Type	Description
batchSize	size_t	Number of reads per batch
threadCount	size_t	Number of parallel threads
executionBackend	ExecutionBackend	Currently supports OneTbb
memoryResourcePolicy	MemoryResourcePolicy	Currently supports ObjectPool
allocationTelemetryEnabled	bool	Enables memory telemetry
readChunkBytes	size_t	Read chunk size
zlibBufferBytes	size_t	zlib buffer
writerBufferBytes	size_t	Writer buffer
batchCapacityBytes	size_t	Batch memory limit
memoryLimitBytes	size_t	Total memory limit
maxInFlightBatches	size_t	Max concurrent batches

---

ProcessingStatistics

Processing result statistics.

Field	Type	Description
totalReads	uint64_t	Total input reads
passedReads	uint64_t	Reads passed filter
filteredReads	uint64_t	Reads filtered out
modifiedReads	uint64_t	Reads trimmed or otherwise modified
errorReads	uint64_t	Error reads
inputBytes	uint64_t	Input bytes
outputBytes	uint64_t	Output bytes
elapsedMs	uint64_t	Total time (milliseconds)
processingTimeMs	double	Processing time in milliseconds (floating-point)
throughputMbps	double	Throughput (MB/s)
allocationTelemetryEnabled	bool	Whether memory telemetry was enabled
memoryResourcePolicy	MemoryResourcePolicy	Resolved memory policy
resolvedMaxInFlightBatches	size_t	Resolved in-flight batch limit

auto getPassRate() const -> double;
auto getFilterRate() const -> double;
auto toString() const -> std::string;

---

Parallel Processing Mechanism

When threadCount > 1, uses tbb::parallel_pipeline:

Source (serial_in_order) → Processing (parallel) → Sink (serial_in_order)
    Read FastqBatch            Parallel filter/trim         Ordered write results

Built-in backpressure mechanism, controls memory usage via maxInFlightBatches.

---

ReadPredicateInterface — Filter Predicate

class ReadPredicateInterface {
public:
    virtual auto evaluate(const fq::io::FastqRecord& read) const -> bool = 0;
};

Built-in Implementations

Class	Description
MinQualityPredicate	Minimum average quality filter
MinLengthPredicate	Minimum read length filter
MaxLengthPredicate	Maximum read length filter
MaxNRatioPredicate	Maximum N base ratio filter

pipeline->addReadPredicate(
    std::make_unique<MinQualityPredicate>(20.0, 33));
pipeline->addReadPredicate(
    std::make_unique<MinLengthPredicate>(50));

---

ReadMutatorInterface — Read Mutator

class ReadMutatorInterface {
public:
    virtual void process(fq::io::FastqRecord& read) = 0;
};

Built-in Implementations

Class	Description
QualityTrimmer	Quality trim (Both/FivePrime/ThreePrime modes)
LengthTrimmer	Length trim (FixedLength/MaxLength/FromStart/FromEnd)
AdapterTrimmer	Adapter trim
PolyTailTrimmer	polyG / bounded polyX tail trimming

pipeline->addReadMutator(
    std::make_unique<QualityTrimmer>(
        20.0, 50,
        QualityTrimmer::TrimMode::Both, 33));

---

Complete Example

#include <fqtools/fq.h>

auto pipeline = fq::processing::createProcessingPipeline();
pipeline->setInputPath("input.fastq");
pipeline->setOutputPath("output.fastq");

fq::processing::ProcessingConfig config;
config.batchSize = 10000;
config.threadCount = 4;
config.executionBackend = fq::processing::ExecutionBackend::OneTbb;
pipeline->setProcessingConfig(config);

pipeline->addReadPredicate(
    std::make_unique<fq::processing::MinQualityPredicate>(20.0, 33));
pipeline->addReadPredicate(
    std::make_unique<fq::processing::MinLengthPredicate>(50));
pipeline->addReadMutator(
    std::make_unique<fq::processing::QualityTrimmer>(
        20.0, 50, fq::processing::QualityTrimmer::TrimMode::Both, 33));
pipeline->addReadMutator(
    std::make_unique<fq::processing::AdapterTrimmer>(
        std::vector<std::string>{"AGATCGGAAGAGC"}, 6, 1));

auto stats = pipeline->run();
std::cout << stats.toString() << "\n";