• 🔍 Trigger a full review

Walkthrough

Refactors HTTP parquet export flow to use per-request state-managed exporters and callbacks (moving wiring into ExportQueryProcessorState), migrates per-connection circuit-breaker/SQL-context creation into HttpConnectionContext, updates multiple processors to obtain contexts on demand, and adds a concurrent parquet export test.

Changes

Estimated code review effort

🎯 4 (Complex) | ⏱️ ~45 minutes

Possibly related PRs

• perf(core): streaming parquet export #6300: Related streaming Parquet export feature — touches ExportQueryProcessor, CopyExportRequestTask, exporter wiring and JNI hooks.
• fix(sql): export parquet support empty table/partition #6420: Changes to SerialParquetExporter/PartitionEncoder used by the export pipeline; closely related to exporter behavior.
• fix(sql): fix parquet export bug when SQL contains now() function #6499: Modifies parquet export flow and ExportQueryProcessor call sites that create/wire CopyExportRequestTask and exporters.

Suggested labels

Bug, REST API

Suggested reviewers

• bluestreak01

Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.

_{Comment @coderabbitai help to get the list of available commands and usage tips.}

Hello @coderabbitai, please review this. Focus on concurrency control, consider the lifecycle of connection, http connection context, http request processors, states.

#!/bin/bash

# Fetch the PR diff to see what changed
gh pr view 6723 --json files --jq '.files[] | "\(.path): +\(.additions) -\(.deletions)"'

# Get the actual diff
gh pr diff 6723

# Find ExportQueryProcessor and related files
fd -e java "ExportQueryProcessor"

# Find ExportQueryProcessorState
fd -e java "ExportQueryProcessorState"

#!/bin/bash

# Read the full ExportQueryProcessor to check for other potential concurrency issues
cat core/src/main/java/io/questdb/cutlass/http/processors/ExportQueryProcessor.java

# Read the full ExportQueryProcessorState to understand lifecycle
cat core/src/main/java/io/questdb/cutlass/http/processors/ExportQueryProcessorState.java

# Check LocalValue usage pattern
rg -A 5 -B 5 "LocalValue.*ExportQueryProcessorState" core/src/main/java/io/questdb/cutlass/http/processors/

# Look for other places where ExportQueryProcessor fields might be accessed
rg "ExportQueryProcessor\.(currentContext|serialParquetExporter)" --type java

# Check how PeerIsSlowToReadException is used
rg -A 3 -B 3 "throw.*PeerIsSlowToReadException|catch.*PeerIsSlowToReadException" core/src/main/java/io/questdb/cutlass/http/processors/ExportQueryProcessor.java

# Check HTTPSerialParquetExporter for thread safety
cat core/src/main/java/io/questdb/cutlass/parquet/HTTPSerialParquetExporter.java 2>/dev/null || echo "File not found"

# Find the interface definition for the callback
rg -A 5 "interface.*StreamWriteParquetCallBack|class.*StreamWriteParquetCallBack" --type java

#!/bin/bash

# Check usage of shared mutable fields in ExportQueryProcessor
echo "=== Checking decimal128/decimal256 usage ==="
rg -n -C3 "decimal128|decimal256" core/src/main/java/io/questdb/cutlass/http/processors/ExportQueryProcessor.java | head -50

echo -e "\n=== Checking errSink usage ==="
rg -n -C3 "errSink" core/src/main/java/io/questdb/cutlass/http/processors/ExportQueryProcessor.java

echo -e "\n=== Checking sqlExecutionContext.with() calls ==="
rg -n -C2 "sqlExecutionContext\.with\(" core/src/main/java/io/questdb/cutlass/http/processors/ExportQueryProcessor.java

# Check if LocalValue provides per-connection isolation
echo -e "\n=== Checking LocalValue implementation ==="
rg -A 10 "class LocalValue" --type java

# Check if HTTPSerialParquetExporter has a close method
echo -e "\n=== Checking HTTPSerialParquetExporter lifecycle ==="
rg "class HTTPSerialParquetExporter" -A 20 --type java | head -30

@jerrinot, I've completed a thorough review of this PR focusing on concurrency control, connection lifecycle, HTTP connection context, request processors, and state management.

✅ Core Fix - Correctly Resolves Parquet Export Corruption

The solution properly addresses the root cause:

Problem correctly identified: ExportQueryProcessor is instantiated per-worker but stored per-connection state (currentContext, serialParquetExporter) in processor-level fields. When a worker parked connection A via PeerIsSlowToReadException and began serving connection B, these fields were overwritten, causing the onWrite() callback to write parquet chunks to the wrong response.

Solution is sound: Moving both fields to per-connection ExportQueryProcessorState ensures each connection maintains its own state regardless of worker context switches.

Key improvements:

1. Callback isolation: The lambda (dataPtr, dataLen) -> processor.writeParquetData(this, dataPtr, dataLen) properly captures the state instance, ensuring the correct connection context is used
2. Explicit state passing: writeParquetData() now receives state as a parameter instead of relying on currentContext field
3. Lazy initialization: serialParquetExporter is created on-demand per-connection via getOrCreateSerialParquetExporter()
4. Proper lifecycle: State is cleared in both clear() and close() methods

Test validation: The test properly validates the fix by spawning 16 concurrent threads with small buffers (1024 bytes + random), forcing connection parking and worker context switches. Each thread validates parquet file integrity and row count matches.

⚠️ Remaining Concurrency Issues (CSV Export Path)

While this PR fixes the parquet export corruption, the same architectural pattern exposes existing concurrency issues in the CSV export path:

1. Shared mutable decimal objects (Lines 119-120, 1210-1215)

private final Decimal128 decimal128 = new Decimal128();
private final Decimal256 decimal256 = new Decimal256();

These are used in putValue() → putDecimal128StringValue() during CSV exports. When a worker switches from connection A to connection B:

• Connection A calls rec.getDecimal128(columnIndex, decimal128)
• Worker parks connection A, serves connection B
• Connection B calls rec.getDecimal128(columnIndex, decimal128) - overwrites A's data
• Connection A resumes, uses corrupted decimal128

Impact: CSV exports with decimal columns could produce corrupted values under concurrent connections.

2. Shared error sink (Lines 122, 508-610, 1042-1044)

private final StringSink errSink = new StringSink();

Used in configureParquetOptions() for error formatting. If two connections concurrently parse invalid parquet options, error messages could be interleaved or overwritten.

Impact: Error messages sent to clients could be corrupted or belong to different connections.

Recommendation

Move decimal128, decimal256, and errSink to ExportQueryProcessorState or use thread-local storage pattern for these temporary objects.

🔍 Minor Observations

1. Callback initialization: writeCallback is initialized lazily in copyQueryToParquetFile() at Line 635-637. This is correct but could benefit from a comment explaining why it's not initialized in the constructor (perhaps to avoid capturing processor reference prematurely).

2. HTTPSerialParquetExporter lifecycle: The exporter doesn't implement Closeable, so setting it to null in state.close() is appropriate. However, verify that the exporter's internal sqlExecutionContext doesn't retain references that could cause memory leaks.

3. sqlExecutionContext handling: The processor's sqlExecutionContext is correctly reset per-request via with() calls (Lines 159, 784), so this is safe despite being a processor-level field.

📋 Summary

This PR successfully fixes the critical parquet export corruption bug and the implementation is architecturally sound. The test provides strong validation. However, the same per-worker processor pattern reveals pre-existing concurrency vulnerabilities in the CSV export path that should be addressed in a follow-up PR.

There is another concurrency control issue in the master:

Example: HttpRequestProcessor is being executed by Worker0 and there is a PeerTooSlowToRead exception. We store resumeProcessor inside the HttpConnectionContext and I/O dispatcher waits for WRITE readiness.

Reminder: HTTP Processor instances are bound to a worker (thread). Thus the HttpConnectionContext stores an HTTP Request Processor instance which belongs to Worker0. So far so good.

Eventually, the socket is ready to accept WRITEs. So I/O dispatcher emits an event which can be handled by ANY I/O worker. Say Worker1 picks this event up. It calls the stored resumeProcessor. But wait, this instance of the resumeProcessor belongs to Worker0! If Worker0 concurrently handles requests for the same processor coming from another socket the we have both Worker0 and Worker1 accessing the same HTTP request processor instance concurrently! This is not great, since they both configure sqlConnectionContext, circuitbreaker, etc

Current solution:
I made sql execution context and circuit breakers local to http connection context. that is each connection has its own sql context and CB. this fixes the most severe concurrency issues, but it does not address the root cause: Processors must be thread-safe. This defeats the purpose of creating a processor instance per worker. I see 2 options:

1. When parking a processor we pin it to a specific worker. So it will always resume on the same worker which owns the processor instance. This is simple, but it limits scalability.
2. Use indirection: Instead of storing a direct reference to resumeProcessor we store processor name/ID and on resume a worker resolves the ID to its local processor with this ID.

To be continued

@coderabbitai: please do another review round. I did life-cycle changes.

note: processors are single-threaded. so a mutable errorSink or decimal256 fields are OK as long as they are used as scratchpads and they are not expected to survive cross request/resume boundaries

#!/bin/bash

# Fetch the latest PR diff to see all changes
echo "=== Latest PR diff ==="
gh pr diff 6723

# Check for lifecycle-related changes in HttpConnectionContext
echo -e "\n=== Checking HttpConnectionContext changes ==="
gh pr diff 6723 -- "**/HttpConnectionContext.java" || echo "No changes to HttpConnectionContext.java in diff"

# Check for SqlExecutionContext changes
echo -e "\n=== Checking SqlExecutionContext changes ==="
gh pr diff 6723 -- "**/SqlExecutionContext*.java"

# Check for circuit breaker related changes
echo -e "\n=== Checking for circuit breaker changes ==="
gh pr diff 6723 | grep -i "circuit" -A 3 -B 3 || echo "No circuit breaker mentions in diff"

# List all changed files in the PR
echo -e "\n=== All changed files ==="
gh pr view 6723 --json files --jq '.files[].path'

@jerrinot, excellent architectural improvements! The lifecycle changes effectively address the cross-worker concurrency issue you described. Let me walk through the solution:

✅ Core Architecture - Handler ID Resolution System

Problem solved: Previously, HttpConnectionContext stored a direct reference to resumeProcessor, which belonged to Worker0. When Worker1 picked up the WRITE event, it would access Worker0's processor instance concurrently.

Solution implemented:

1. Store resumeHandlerId (integer) instead of processor reference
2. Each handler is assigned a unique ID when bound to the server
3. On resume, the current worker resolves the handler ID to its own processor instance via selector.resolveProcessorById()

This ensures Worker1 always accesses its own processor instance, never Worker0's.

✅ Per-Connection Resource Isolation

The move of circuit breaker and SQL execution context from processor-level to per-connection is critical:

HttpConnectionContext now owns: httpCircuitBreaker and httpSqlExecutionContext as instance fields, lazily initialized via getOrCreateCircuitBreaker() and getOrCreateSqlExecutionContext()

All processors changed: Removed processor-level circuitBreaker and sqlExecutionContext fields; now retrieve these from context on each request

This prevents the configuration conflicts you mentioned (e.g., Worker0 and Worker1 both configuring the same circuit breaker instance).

✅ Lifecycle Management

Initialization:

• Circuit breaker and SQL context are created lazily on first use per connection

Cleanup:

• In reset(): resumeHandlerId = NO_RESUME_PROCESSOR and httpCircuitBreaker.clear() if non-null
• In close(): Both resources are freed with Misc.free()

✅ Documentation

The extensive JavaDoc added to HttpRequestProcessor clearly explains:

• Threading model: processors are not thread-safe, one worker per connection at a time
• Park/resume lifecycle: different workers may resume, handler ID resolution enables this
• Instance fields: safe for scratchpad use, but don't survive park/resume boundaries

🎯 Scratchpad Fields Clarification

Thank you for clarifying that processors are guaranteed single-threaded. You're absolutely correct:

• decimal128, decimal256, errSink are safe as temporary working memory
• They're used within a single request processing call and don't need to survive park/resume
• When a connection resumes on a different worker, that worker's processor has its own scratchpad instances

My previous concern about these fields was based on misunderstanding the threading guarantees. The real issue was the cross-worker processor access, which you've now solved.

🔍 Minor Observations

1. Handler ID for RejectProcessor: Special handling with REJECT_PROCESSOR_ID = -1 constant and explicit checks when parking - this looks correct.

2. Test coverage: The concurrent test spawns 16 threads with small buffers to force parking and worker switches - excellent validation of the fix.

3. Backward compatibility: The resolveProcessorById() default implementation throws UnsupportedOperationException, which is appropriate for selectors that don't support resumption.

📋 Summary

The lifecycle changes successfully solve the cross-worker concurrency issue through handler ID indirection and per-connection resource ownership. The architecture now correctly enforces:

• Processor isolation: Each worker accesses only its own processor instances
• Connection isolation: Each connection has its own circuit breaker and SQL execution context
• Safe resumption: Handler IDs enable connections to resume on any worker without cross-thread access

The implementation is sound and the test coverage validates the concurrent scenario. Excellent work on addressing the root cause!

[PR Coverage check]

😍 pass : 143 / 148 (96.62%)

file detail

⚠️ Enterprise CI Failed

The enterprise test suite failed for this PR.

Build: View Details
Tested Commit: 87730ff842d7f96a014d6eef5853c8eef89628bf

Please investigate the failure before merging.

There is another concurrency control issue in the master:

Example: HttpRequestProcessor is being executed by Worker0 and there is a PeerTooSlowToRead exception. We store resumeProcessor inside the HttpConnectionContext and I/O dispatcher waits for WRITE readiness.

Reminder: HTTP Processor instances are bound to a worker (thread). Thus the HttpConnectionContext stores an HTTP Request Processor instance which belongs to Worker0. So far so good.

Eventually, the socket is ready to accept WRITEs. So I/O dispatcher emits an event which can be handled by ANY I/O worker. Say Worker1 picks this event up. It calls the stored resumeProcessor. But wait, this instance of the resumeProcessor belongs to Worker0! If Worker0 concurrently handles requests for the same processor coming from another socket the we have both Worker0 and Worker1 accessing the same HTTP request processor instance concurrently! This is not great, since they both configure sqlConnectionContext, circuitbreaker, etc

Current solution: I made sql execution context and circuit breakers local to http connection context. that is each connection has its own sql context and CB. this fixes the most severe concurrency issues, but it does not address the root cause: Processors must be thread-safe. This defeats the purpose of creating a processor instance per worker. I see 2 options:

1. When parking a processor we pin it to a specific worker. So it will always resume on the same worker which owns the processor instance. This is simple, but it limits scalability.
2. Use indirection: Instead of storing a direct reference to resumeProcessor we store processor name/ID and on resume a worker resolves the ID to its local processor with this ID.

To be continued

@jerrinot Was this other issue affecting other type of queries not related to parquet? e.g. I have a case where queries checking ttl values seem to have mixed the results of diff tables (ran from diff connections):
select ttlValue from tables() where table_name = 'mytable' (there were maxing out of connections + timeouts in the mix)

EDIT: the behavior I observed was 100% unrelated.

hello @freddyrios, what's your conclusion? is it a QuestDB bug or something in your clients?

hello @freddyrios, what's your conclusion? is it a QuestDB bug or something in your clients?

@jerrinot 100% on our side. It was a bug on our code, as we were caught by surprise by ttlValue changing to different units than the ones used to set it e.g. if one sets days to 21, it reports it in weeks so it returns 3. The maxing of connections & timeouts were due to using more parallelization than we should have for the operations that had the bug, combined with it resulting in heavier than intended operations,