This is an automated email from the ASF dual-hosted git repository. nickva pushed a commit to branch fix-compression-settings in repository https://gitbox.apache.org/repos/asf/couchdb.git
commit 238e633a9260bcb6cc9fcb12c9f82f1221d9e520 Author: Nick Vatamaniuc <[email protected]> AuthorDate: Sun Jun 7 00:46:02 2026 -0400 Always apply user configured compression Previously, even when user picked a non-default compression algorithm in some instances, like the purge trees and emsort, snappy would still be used such. Make sure to always respect user's choice, and if they picked something non-default always use that. [1] There is a good reason to avoid snappy as doesn't properly yield when encoding or decoding. In a low concurrency environment it might not matter, but as the concurrency ramps up it could lead to scheduler collapse (it could manifest as high p90+ latencies while cpus usage remains low). A similar effect is illustrated in scheduler responsiveness test for jiffy vs other C json libraries in https://github.com/nickva/bench#scheduler-responsiveness-test --- src/couch/src/couch_bt_engine.erl | 12 ++++++++---- src/couch/src/couch_bt_engine_compactor.erl | 18 +++++++++++------- src/couch/src/couch_emsort.erl | 21 ++++++++++++++++----- .../test/eunit/couch_bt_engine_compactor_tests.erl | 10 +++++++--- 4 files changed, 42 insertions(+), 19 deletions(-) diff --git a/src/couch/src/couch_bt_engine.erl b/src/couch/src/couch_bt_engine.erl index 072a75c74..4bdef732d 100644 --- a/src/couch/src/couch_bt_engine.erl +++ b/src/couch/src/couch_bt_engine.erl @@ -844,7 +844,7 @@ init_state(FilePath, Fd, Header0, Options) -> Header1 = couch_bt_engine_header:upgrade(Header0), Header2 = set_default_security_object(Fd, Header1, Compression, Options), - Header = upgrade_purge_info(Fd, Header2), + Header = upgrade_purge_info(Fd, Header2, Compression), IdTreeState = couch_bt_engine_header:id_tree_state(Header), {ok, IdTree} = couch_btree:open(IdTreeState, Fd, [ @@ -877,6 +877,7 @@ init_state(FilePath, Fd, Header0, Options) -> {split, fun ?MODULE:purge_tree_split/1}, {join, fun ?MODULE:purge_tree_join/2}, {reduce, fun ?MODULE:purge_tree_reduce/2}, + {compression, Compression}, {cache_depth, btree_cache_depth()} ]), @@ -885,6 +886,7 @@ init_state(FilePath, Fd, Header0, Options) -> {split, fun ?MODULE:purge_seq_tree_split/1}, {join, fun ?MODULE:purge_seq_tree_join/2}, {reduce, fun ?MODULE:purge_tree_reduce/2}, + {compression, Compression}, {cache_depth, btree_cache_depth()} ]), @@ -943,7 +945,7 @@ set_default_security_object(Fd, Header, Compression, Options) -> % This function is here, and not in couch_bt_engine_header % because it requires modifying file contents -upgrade_purge_info(Fd, Header) -> +upgrade_purge_info(Fd, Header, Compression) -> case couch_bt_engine_header:get(Header, purge_tree_state) of nil -> Header; @@ -979,7 +981,8 @@ upgrade_purge_info(Fd, Header) -> {ok, PurgeTree} = couch_btree:open(nil, Fd, [ {split, fun ?MODULE:purge_tree_split/1}, {join, fun ?MODULE:purge_tree_join/2}, - {reduce, fun ?MODULE:purge_tree_reduce/2} + {reduce, fun ?MODULE:purge_tree_reduce/2}, + {compression, Compression} ]), {ok, PurgeTree2} = couch_btree:add(PurgeTree, Infos), PurgeTreeSt = couch_btree:get_state(PurgeTree2), @@ -987,7 +990,8 @@ upgrade_purge_info(Fd, Header) -> {ok, PurgeSeqTree} = couch_btree:open(nil, Fd, [ {split, fun ?MODULE:purge_seq_tree_split/1}, {join, fun ?MODULE:purge_seq_tree_join/2}, - {reduce, fun ?MODULE:purge_tree_reduce/2} + {reduce, fun ?MODULE:purge_tree_reduce/2}, + {compression, Compression} ]), {ok, PurgeSeqTree2} = couch_btree:add(PurgeSeqTree, Infos), PurgeSeqTreeSt = couch_btree:get_state(PurgeSeqTree2), diff --git a/src/couch/src/couch_bt_engine_compactor.erl b/src/couch/src/couch_bt_engine_compactor.erl index 12c1ea3ac..ee349337d 100644 --- a/src/couch/src/couch_bt_engine_compactor.erl +++ b/src/couch/src/couch_bt_engine_compactor.erl @@ -509,7 +509,9 @@ copy_docs(St, #st{} = NewSt, MixedInfos, Retry) -> ), EMSortFd = couch_emsort:get_fd(NewSt#st.id_tree), - {ok, LocSizes} = couch_file:append_terms(EMSortFd, NewInfos), + {ok, LocSizes} = couch_file:append_terms( + EMSortFd, NewInfos, [{compression, NewSt#st.compression}] + ), EMSortEntries = lists:zipwith( fun(FDI, {Loc, _}) -> #full_doc_info{ @@ -603,7 +605,8 @@ copy_meta_data(#comp_st{new_st = St} = CompSt) -> {ok, IdTree0} = couch_btree:open(DstState, Fd, [ {split, fun couch_bt_engine:id_tree_split/1}, {join, fun couch_bt_engine:id_tree_join/2}, - {reduce, fun couch_bt_engine:id_tree_reduce/2} + {reduce, fun couch_bt_engine:id_tree_reduce/2}, + {compression, St#st.compression} ]), {ok, Iter} = couch_emsort:iter(Src), Acc0 = #merge_st{ @@ -689,20 +692,21 @@ commit_compaction_data(#st{header = OldHeader} = St0, Fd) -> bind_emsort(St2, MetaFd, MetaState). bind_emsort(St, Fd, nil) -> - {ok, Ems} = couch_emsort:open(Fd), - St#st{id_tree = Ems}; + bind_emsort(St, Fd, []); bind_emsort(St, Fd, {BB, _} = Root) when is_list(BB) -> % Upgrade clause when we find old compaction files bind_emsort(St, Fd, [{root, Root}]); -bind_emsort(St, Fd, State) -> - {ok, Ems} = couch_emsort:open(Fd, State), +bind_emsort(St, Fd, State) when is_list(State) -> + Options = [{compression, St#st.compression} | State], + {ok, Ems} = couch_emsort:open(Fd, Options), St#st{id_tree = Ems}. bind_id_tree(St, Fd, State) -> {ok, IdBtree} = couch_btree:open(State, Fd, [ {split, fun couch_bt_engine:id_tree_split/1}, {join, fun couch_bt_engine:id_tree_join/2}, - {reduce, fun couch_bt_engine:id_tree_reduce/2} + {reduce, fun couch_bt_engine:id_tree_reduce/2}, + {compression, St#st.compression} ]), St#st{id_tree = IdBtree}. diff --git a/src/couch/src/couch_emsort.erl b/src/couch/src/couch_emsort.erl index 4250b45e9..6a7856839 100644 --- a/src/couch/src/couch_emsort.erl +++ b/src/couch/src/couch_emsort.erl @@ -133,13 +133,16 @@ -export([add/2, merge/2, iter/1, next/1]). -export([num_kvs/1, num_merges/1]). +-include_lib("couch/include/couch_db.hrl"). + -record(ems, { fd, root, bb_chunk = 10, chain_chunk = 100, num_kvs = 0, - num_bb = 0 + num_bb = 0, + compression = ?DEFAULT_COMPRESSION }). -define(REPORT_INTERVAL, 1000). @@ -161,7 +164,9 @@ set_options(Ems, [{back_bone_chunk, Count} | Rest]) when is_integer(Count) -> set_options(Ems, [{num_kvs, NumKVs} | Rest]) when is_integer(NumKVs) -> set_options(Ems#ems{num_kvs = NumKVs}, Rest); set_options(Ems, [{num_bb, NumBB} | Rest]) when is_integer(NumBB) -> - set_options(Ems#ems{num_bb = NumBB}, Rest). + set_options(Ems#ems{num_bb = NumBB}, Rest); +set_options(Ems, [{compression, Comp} | Rest]) -> + set_options(Ems#ems{compression = Comp}, Rest). get_fd(#ems{fd = Fd}) -> Fd. @@ -231,7 +236,9 @@ write_kvs(Ems, KVs) -> {[], nil}, lists:sort(KVs) ), - {ok, Final, _} = couch_file:append_term(Ems#ems.fd, {LastKVs, LastPos}), + {ok, Final, _} = couch_file:append_term( + Ems#ems.fd, {LastKVs, LastPos}, [{compression, Ems#ems.compression}] + ), Final. decimate(#ems{root = {_BB, nil}} = Ems, _Reporter) -> @@ -274,7 +281,9 @@ merge_chains(Ems, Choose, BB, Reporter) -> merge_chains(Ems, Choose, Chains, {[], nil}, Reporter, 0). merge_chains(Ems, _Choose, [], ChainAcc, _Reporter, _Count) -> - {ok, CPos, _} = couch_file:append_term(Ems#ems.fd, ChainAcc), + {ok, CPos, _} = couch_file:append_term( + Ems#ems.fd, ChainAcc, [{compression, Ems#ems.compression}] + ), CPos; merge_chains(#ems{chain_chunk = CC} = Ems, Choose, Chains, Acc, Reporter, Count0) -> {KV, RestChains} = choose_kv(Choose, Ems, Chains), @@ -327,7 +336,9 @@ ins_big_chain(Rest, Chain, Acc) -> lists:reverse(Acc, [Chain | Rest]). append_item(Ems, {List, Prev}, Pos, Size) when length(List) >= Size -> - {ok, PrevList, _} = couch_file:append_term(Ems#ems.fd, {List, Prev}), + {ok, PrevList, _} = couch_file:append_term( + Ems#ems.fd, {List, Prev}, [{compression, Ems#ems.compression}] + ), {[Pos], PrevList}; append_item(_Ems, {List, Prev}, Pos, _Size) -> {[Pos | List], Prev}. diff --git a/src/couch/test/eunit/couch_bt_engine_compactor_tests.erl b/src/couch/test/eunit/couch_bt_engine_compactor_tests.erl index a7fec17db..63df30fb0 100644 --- a/src/couch/test/eunit/couch_bt_engine_compactor_tests.erl +++ b/src/couch/test/eunit/couch_bt_engine_compactor_tests.erl @@ -70,7 +70,7 @@ is_compacting_works(DbName) -> {_Pid, Ref} = spawn_monitor(fun() -> compact_db(DbName) end), receive {in_emsort_bind, From} -> - % When emsort:open(Fd) is called the files should + % When emsort:open(Fd, Options) is called the files should % have been created already ?assert(couch_db:is_compacting(DbName)), From ! {please_continue, self()} @@ -145,11 +145,15 @@ wait_db_compact_done(DbName, N) -> end. wait_in_emsort_bind(WaitingPid) when is_pid(WaitingPid) -> - meck:expect(couch_emsort, open, fun(Fd) -> + meck:expect(couch_emsort, open, fun(Fd, Options) -> + % Intercept first call to bind only the rest should pass through + meck:expect(couch_emsort, open, fun(F, O) -> + meck:passthrough([F, O]) + end), WaitingPid ! {in_emsort_bind, self()}, receive {please_continue, WaitingPid} -> ok end, - meck:passthrough([Fd]) + meck:passthrough([Fd, Options]) end).
