All committers have signed the CLA.

Hey @yokofly, thanks for the thorough review — you caught three real issues and I've addressed all of them in commit a812443.On the dead settings: you're right, the static constexpr values were making the knobs completely non-functional. I've replaced them with live reads from storage_stream.getContext()->getSettingsRef(), so dynamic_commit_row_threshold and dynamic_commit_byte_threshold can now actually be tuned at runtime via SET or server config as intended.On the worker retry loop: the original infinite while (!isStopped()) was the real liveness problem — under TOO_MANY_PARTS pressure it would hold a pool thread indefinitely. I've capped it at MAX_COMMIT_RETRIES = 10 (about 20 seconds worst case). After the cap, we log the give-up and always call progressSequences() so the producer side is never permanently blocked. The data can be recovered on next restart. On the mixed non-JSON/JSON batch issue: you were correct that mergeBlocks() ran before the hasDynamicSubcolumns() check, which silently let a non-JSON block accumulate into a JSON-thresholded batch. I've added an explicit boundary check before the merge — if the in-flight block and incoming record differ in dynamic-subcolumn presence, we flush the current block and start fresh. JSON and non-JSON now stay strictly separated.

CLA has been signed. Let me know if anything else needs addressing.

CLA has been signed. Let me know if anything else needs addressing.

I also noticed the license/CLA check was still pending @SBALAVIGNESH123

You're absolutely right — good catch. Calling progressSequences() unconditionally after retry exhaustion would let
commitSNLocal() persist a sequence number for data that was never written to disk.Fixed in 600fdcb: I now track success with a bool committed flag and only call progressSequences() on the success path. On the failure path, we log the dropped SN range and release the pool thread without advancing sequence state — NativeLog will replay the missing range on restart.
Regarding the CLA — I signed it earlier through the CLA Assistant link. Let me know if it's still showing as pending and I have re-triggered it.

CLA assistant checks commit identities in the PR, not just the PR opener. If the commit email is missing or not linked to the contributor’s GitHub account, the CLA check can trigger but fail to recognize the signer, leaving the status pending.

I have read the CLA Document and I hereby sign the CLA

Thanks for the fix. license/cla is pending because commit author emails are inconsistent in this PR.

Current mismatch:

• 2c284cf uses [email protected]
• later commits use [email protected]

Please do one of these, then push again:

1. Add/sign CLA for [email protected] (GitHub account email), or
2. Rewrite that commit author to your signed email and git push --force-with-lease.

After pushing, please re-check: #1124 (comment)

I have read the CLA Document and I hereby sign the CLA

Thanks for catching this — it's exactly the kind of subtle liveness issue that's hard to spot without knowing the full state machine. Fixed in d9a4d0c: the failure path now calls failSequencesWithLockHeld() which handles two cases since async workers can complete in any order — if the failed entry is at the front it pops it directly (without touching last_sn) and calls the new shared drainWithLockHeld() loop to resume; if it arrives out of order it goes into a new failed_sns set, mirroring how local_committed_sns already handles out-of-order successes. The drainWithLockHeld() loop processes both sets — advancing last_sn for committed entries and silently skipping failed ones — so the SN state machine stays unblocked regardless of completion order, and last_sn always excludes failed ranges so NativeLog replays them on restart. Really appreciate the depth of review on this one.

Thanks for iterating on this. I think my earlier review pulled the change too far into async failure-handling. After stepping back to the original #1113 issue and re-reading the earlier discussion (#1113 (comment)), I think the simpler direction is the right one here.

The core issue is the dynamic/JSON path creating a tiny-part storm. I'd recommend narrowing this PR back to that:

1. keep batching for dynamic/JSON blocks
2. for the dynamic/JSON path only, commit synchronously instead of through the async commit pool
3. keep SN advancement strictly tied to successful write completion
4. drop the async failure-unblocking state machine from this PR

That still addresses the original incident, while keeping the recovery semantics much simpler and safer to maintain.

Good direction — agreed. Simplified in e0270dc: dynamic/JSON blocks now commit synchronously on the polling thread. Since the caller already batches to the configured row/byte threshold (8192 rows or 16MB), each synchronous write is a reasonably-sized part rather than a 1-row flush. SN advancement is strictly tied to successful write completion. On exception, it propagates to the NativeLog consumer which re-reads from the last committed SN on restart. The async failure state machine is dropped entirely from this PR.

Thanks, this is much closer to the shape I had in mind. The dynamic/JSON path is now batched and synchronous, which makes the SN/recovery behavior much simpler than the earlier async failure-unblocking approach.

I think there is just one remaining piece to tighten.

In the new synchronous dynamic/JSON path, the comment says a write failure will propagate to the NativeLog consumer and recovery will resume from the last committed SN. But StreamCallbackData::doCommit() still catches stream_shard_store->commit(...) and only logs the exception, so that failure is currently swallowed.

To make the dynamic/JSON path robust, I think it would help to do two small things together:

1. for the synchronous dynamic/JSON path only, add seq_pair to outstanding_sns only after the write succeeds, right before progressSequences()
2. when that synchronous dynamic/JSON write fails, explicitly stop/fail the shard instead of only logging, so restart/recovery semantics take over cleanly from the last committed SN

That keeps the dynamic/JSON path simple: no front gap in outstanding_sns, no partial SN progress, and no ambiguity about what should happen after a failed write.

For coverage, I think it would also be good to add a few user-visible checks under tests/queries_ported/0_stateless, for example:

1. dynamic/JSON batching threshold behavior
2. mixed JSON/non-JSON boundary behavior
3. historical reads via table(stream) after those inserts

The restart/failure path probably still needs a more targeted integration/restart-style test, but the stateless historical-query coverage would already make this PR much easier to validate.

Fixed in d27f92c — you're right that the recovery path was broken on both ends. In StreamShardStore::doCommit, the sync JSON path now pushes to outstanding_sns only after onFinish() returns successfully, so a failed write leaves no dangling entry in the SN deque and the exception propagates upward without any partial state. In StreamCallbackData::doCommit, the catch block now logs at FATAL and re-throws instead of silently swallowing — the exception reaches the NativeLog consumer which stops the shard and restarts cleanly from the last committed SN, which is exactly the recovery behavior the comment promised. Also added two stateless tests in tests/queries_ported/0_stateless —
99175_dynamic_json_batching.sql covers JSON batch inserts and historical reads via table(stream), and 99176_dynamic_json_nonjson_boundary.sql verifies that JSON and non-JSON streams stay correctly separated and are both readable after inserts.

The main fix looks right to me now. I’m only working through a few last-mile cleanup/scope details so the final diff stays narrow and easier to maintain. The core dynamic/JSON batching + synchronous commit direction looks good.

No further action needed on your side. I’m moving the final version to #1127 because of the external-contributor workflow here. Once CI is green there, I’m comfortable approving it.

(I’m not sure whether my first comment was posted successfully, so I’m reposting it here.)

The main fix looks right to me now. I’m only working through a few last-mile cleanup/scope details so the final diff stays narrow and easier to maintain. The core dynamic/JSON batching + synchronous commit direction looks good.

No further action needed on your side. I’m moving the final version to #1127 because of the external-contributor workflow here. Once CI is green there, I’m comfortable approving it.

@yuzifeng1984, please review, check pr desc first #1124 (comment)

I do not fully understand the original storage saving design and thus need some more time to think about this.

I went through the changes which looks doing below separate things:

1. Merge small json's
2. Refactor asynchronous commit scheduling.
3. Enhancement exception and error handling.

Initial thought: these issues may be triggered by the same use case; while putting them in multiple enhancements will be better to understand and reasoning.

Thanks @yuzifeng1984 for the thorough review — great questions. Let me address everything together:

On std::exchange vs direct move: std::exchange(batch, {}) guarantees batch is left in a known empty state, not just a moved-from state. In the main polling loop, the code already does this manually (auto current_batch = std::move(batch); batch = SchemaRecordPtrs{}; batch.reserve(batch_size);). The std::exchange in the SCOPE_EXIT final-batch path is the same idea, just more concise since no reserve() is needed there.

On the sleep/retry blocking the commit pool: Agreed — the sleep_for(2000ms) in the async retry loop is a pre-existing issue that predates this PR. This PR intentionally leaves the non-JSON async retry path unchanged to keep the diff narrow and focused on the dynamic/JSON tiny-part storm fix. Fixing the async retry with proper re-submission + backpressure would be a good follow-up, but it's a broader change to the commit pool scheduling model.

On when the exception stops the background thread: The synchronous dynamic/JSON commit path calls stream_shard_store->commit(...) directly (not via the async pool). If that write fails — for example, disk full, filesystem error, or a TOO_MANY_PARTS exception — it propagates through StreamCallbackData::doCommit(), which logs and re-throws. The exception then reaches backgroundPollNativeLog() where the stream_commit.commit() call is wrapped in a try/catch that catches the failure, logs "Stopping consume loop after commit failure", and returns — stopping the shard's consume loop. On restart, NativeLog replays from the last committed SN, so no data is lost.

On the impact of merging all JSON blocks without thresholds: Without thresholds, JSON blocks just accumulate indefinitely in the commit() loop until the batch ends, creating one massive part. This has two problems: (1) memory pressure — the entire catch-up backlog stays in memory before any write, which could itself cause OOM; and (2) commit latency — one massive write blocks the polling thread for a long time, stalling SN progress. The thresholds (8192 rows / 16MB, read live from storage_stream.getContext()->getSettingsRef()) keep each part bounded — large enough to avoid the tiny-part storm, small enough to bound memory and latency.

On the redundant comment (nit): Fair point — will remove.

On splitting into separate enhancements: That's a fair observation. The changes do touch three areas, but they're tightly coupled around one failure mode:

Batching JSON blocks — the core fix. Without it, every JSON record creates a 1-row part, causing the part storm.
Synchronous commit for the JSON path — required because of the batching. The original async path pushes to outstanding_sns before the write, then calls progressSequences() on completion. For synchronous JSON writes, we push to outstanding_sns after success so there's never a gap-entry from a failed write. This is simpler and safer than adapting the async state machine for batched JSON.
Re-throwing on commit failure — required to make the sync path safe. Previously StreamCallbackData::doCommit() caught and swallowed the exception. Without re-throwing, a failed synchronous write would be silently lost and the consume loop would continue with inconsistent SN state.
They could theoretically be split, but (2) and (3) don't make sense without (1), and (1) without (2) would require the complex async failure-handling that @yokofly asked to remove. Happy to hear your thoughts if you see a cleaner split.

I think sync commit for the dynamic/JSON path should stay in this PR, because it simplifies the fix rather than expanding scope: SN advancement stays tied to successful writes, and we avoid reintroducing more complex async failure-handling.

That also seems consistent with the earlier note in #1113 (comment) that sync commit would be a minor enhancement.

@SBALAVIGNESH123 Thanks for the clarification.

My thought to split is that squashing JSON tiny block is clear enhancement; while introducing sync write, the flow is quite complex and need more review. Background task works differently by whether block has json type, async block write may encourter the same exceptions you mentioned or others. It will retry but not quit. I am thinking if stop background polling and rely on restart is a good solution.

Good point — you're right, batch goes out of scope right after this, so std::exchange is unnecessary here. A plain std::move(batch) is sufficient. Thanks for catching that.

Thanks for the thoughtful feedback. @yokofly covered the main reasoning above — the sync approach was specifically chosen to avoid the async failure-handling complexity from the earlier iterations of this PR.

On stop + restart: agreed it's a heavy response — but for synchronous writes, the block is gone after a failed write, so there's no clean in-place retry without re-reading from NativeLog. Happy to iterate on this if you and @yokofly agree on a different direction.

Hi @SBALAVIGNESH123 Let us discuss about this. Will reply to you later.

Hi @SBALAVIGNESH123 After discussion, we think:

1. In this PR, let's focus on tiny-part problem by merging the JSON records if applicable.
2. We will do some performance / stress tests once the PR is merged and see how the JSON cases are handled.
3. We plan to do an enhancement to support synchronous submit in general. Possibly have a flag to switch beween sync and async mode. We are doing more evaluation on this currently. So let's keep the commit logic unchanged for now.

If you have any ideas, please share them.

Thanks @yuzifeng1984 — understood. I'll narrow this PR to just the JSON batching logic:

Keep the dynamic_commit_row_threshold / dynamic_commit_byte_threshold batching
Keep the JSON/non-JSON boundary flush
Revert the synchronous commit path back to async
Revert the re-throw in StreamCallbackData::doCommit()
Will push the updated version shortly.

On the sync/async enhancement idea: one approach could be a per-stream setting like commit_mode = 'async' | 'sync' (defaulting to async for backward compatibility). For streams with hasDynamicSubcolumns(), the engine could default to
sync automatically since the part-count sensitivity makes synchronous writes safer. This would let operators override per-stream without changing the global behavior. Happy to help with that in a follow-up PR if useful.

Updated in 448029c. Narrowed to JSON batching only:

Kept: dynamic_commit_row_threshold / dynamic_commit_byte_threshold batching, JSON/non-JSON boundary flush, settings wiring, stateless tests
Reverted: synchronous commit path, retry cap, exponential backoff, StreamCallbackData re-throw — commit logic is unchanged from develop

The behavior is the same — both paths push to outstanding_sns under the lock and call progressSequencesWithLockHeld
for empty blocks. The restructuring was a leftover from the sync commit iteration. Happy to revert to the original single-lock-scope structure if you'd prefer the diff to stay minimal.

@yokofly The PR LGTM. Please help to progress.

Thanks @yokofly and @yuzifeng1984 for the review and the merge! Tracking the OOM back to the per-record commit in the dynamic subcolumns path was a great debugging exercise — I really enjoyed digging into it. Between this fix and the NATS JetStream connector (#1111), I've spent quite a bit of time exploring Proton internals — the storage commit pipeline, NativeLog integration, external stream framework, and the settings system.
I'm also an active competitive programmer (recently placed 2nd out of 83 in a TopCoder challenge), so I'm comfortable working on performance-sensitive C++ systems.If there are any opportunities to contribute more formally — contract work, part-time, internships, or other roles — I'd definitely be interested. I'm flexible with timings and availability, and happy to adapt to what works best for the team. And if it helps to assess fit first — feel free to assign me a task or issue with a deadline; I'm happy to do it.Of course, I completely understand if the timing isn't right — I'm happy to keep picking up issues from the backlog regardless.

Thanks again for being welcoming to contributors — it really makes a difference.

Thanks @yokofly and @yuzifeng1984 for the review and the merge! Tracking the OOM back to the per-record commit in the dynamic subcolumns path was a great debugging exercise — I really enjoyed digging into it. Between this fix and the NATS JetStream connector (#1111), I've spent quite a bit of time exploring Proton internals — the storage commit pipeline, NativeLog integration, external stream framework, and the settings system. I'm also an active competitive programmer (recently placed 2nd out of 83 in a TopCoder challenge), so I'm comfortable working on performance-sensitive C++ systems.If there are any opportunities to contribute more formally — contract work, part-time, internships, or other roles — I'd definitely be interested. I'm flexible with timings and availability, and happy to adapt to what works best for the team. And if it helps to assess fit first — feel free to assign me a task or issue with a deadline; I'm happy to do it.Of course, I completely understand if the timing isn't right — I'm happy to keep picking up issues from the backlog regardless.

Thanks again for being welcoming to contributors — it really makes a difference.

Thanks for the contributions and for your interest. @SBALAVIGNESH123

At the moment, we do not have budget or headcount for contract, part-time, internship, or other formal roles. We still appreciate your contributions, and you’re welcome to keep contributing through the normal OSS workflow.

Separately, if you’re looking for paid open-source opportunities more broadly, you may also want to keep an eye on programs like Google Summer of Code or the Linux Foundation mentorship programs for projects that participate.

Thanks for the honest response — I completely understand the constraints. I'll keep contributing through the OSS workflow and appreciate the team being so responsive and welcoming