eiri commented on a change in pull request #496: Couchdb 3287 pluggable storage engines URL: https://github.com/apache/couchdb/pull/496#discussion_r118570812
########## File path: src/couch/src/couch_bt_engine.erl ########## @@ -0,0 +1,956 @@ +% Licensed under the Apache License, Version 2.0 (the "License"); you may not +% use this file except in compliance with the License. You may obtain a copy of +% the License at +% +% http://www.apache.org/licenses/LICENSE-2.0 +% +% Unless required by applicable law or agreed to in writing, software +% distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +% WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +% License for the specific language governing permissions and limitations under +% the License. + +-module(couch_bt_engine). +-behavior(couch_db_engine). + +-export([ + exists/1, + + delete/3, + delete_compaction_files/3, + + init/2, + terminate/2, + handle_call/2, + handle_info/2, + + incref/1, + decref/1, + monitored_by/1, + + get_compacted_seq/1, + get_del_doc_count/1, + get_disk_version/1, + get_doc_count/1, + get_epochs/1, + get_last_purged/1, + get_purge_seq/1, + get_revs_limit/1, + get_security/1, + get_size_info/1, + get_update_seq/1, + get_uuid/1, + + set_revs_limit/2, + set_security/2, + + open_docs/2, + open_local_docs/2, + read_doc_body/2, + + serialize_doc/2, + write_doc_body/2, + write_doc_infos/4, + + commit_data/1, + + open_write_stream/2, + open_read_stream/2, + is_active_stream/2, + + fold_docs/4, + fold_local_docs/4, + fold_changes/5, + count_changes_since/2, + + start_compaction/4, + finish_compaction/4 +]). + + +-export([ + init_state/4 +]). + + +-export([ + id_tree_split/1, + id_tree_join/2, + id_tree_reduce/2, + + seq_tree_split/1, + seq_tree_join/2, + seq_tree_reduce/2, + + local_tree_split/1, + local_tree_join/2 +]). + + +% Used by the compactor +-export([ + set_update_seq/2, + copy_security/2 +]). + + +-include_lib("couch/include/couch_db.hrl"). +-include("couch_bt_engine.hrl"). + + +exists(FilePath) -> + case filelib:is_file(FilePath) of + true -> + true; + false -> + filelib:is_file(FilePath ++ ".compact") + end. + + +delete(RootDir, FilePath, Async) -> + %% Delete any leftover compaction files. If we don't do this a + %% subsequent request for this DB will try to open them to use + %% as a recovery. + delete_compaction_files(RootDir, FilePath, [{context, delete}]), + + % Delete the actual database file + couch_file:delete(RootDir, FilePath, Async). + + +delete_compaction_files(RootDir, FilePath, DelOpts) -> + lists:foreach(fun(Ext) -> + couch_file:delete(RootDir, FilePath ++ Ext, DelOpts) + end, [".compact", ".compact.data", ".compact.meta"]). + + +init(FilePath, Options) -> + {ok, Fd} = open_db_file(FilePath, Options), + Header = case lists:member(create, Options) of + true -> + delete_compaction_files(FilePath), + Header0 = couch_bt_engine_header:new(), + ok = couch_file:write_header(Fd, Header0), + Header0; + false -> + case couch_file:read_header(Fd) of + {ok, Header0} -> + Header0; + no_valid_header -> + delete_compaction_files(FilePath), + Header0 = couch_bt_engine_header:new(), + ok = couch_file:write_header(Fd, Header0), + Header0 + end + end, + {ok, init_state(FilePath, Fd, Header, Options)}. + + +terminate(_Reason, St) -> + % If the reason we died is because our fd disappeared + % then we don't need to try closing it again. + Ref = St#st.fd_monitor, + if Ref == closed -> ok; true -> + ok = couch_file:close(St#st.fd), + receive + {'DOWN', Ref, _, _, _} -> + ok + after 500 -> + ok + end + end, + couch_util:shutdown_sync(St#st.fd), + ok. + + +handle_call(Msg, St) -> + {stop, {invalid_call, Msg}, {invalid_call, Msg}, St}. + + +handle_info({'DOWN', Ref, _, _, _}, #st{fd_monitor=Ref} = St) -> + {stop, normal, St#st{fd=undefined, fd_monitor=closed}}. + + +incref(St) -> + {ok, St#st{fd_monitor = erlang:monitor(process, St#st.fd)}}. + + +decref(St) -> + true = erlang:demonitor(St#st.fd_monitor, [flush]), + ok. + + +monitored_by(St) -> + case erlang:process_info(St#st.fd, monitored_by) of + {monitored_by, Pids} -> + Pids; + _ -> + [] + end. + + +get_compacted_seq(#st{header = Header}) -> + couch_bt_engine_header:get(Header, compacted_seq). + + +get_del_doc_count(#st{} = St) -> + {ok, Reds} = couch_btree:full_reduce(St#st.id_tree), + element(2, Reds). + + +get_disk_version(#st{header = Header}) -> + couch_bt_engine_header:get(Header, disk_version). + + +get_doc_count(#st{} = St) -> + {ok, Reds} = couch_btree:full_reduce(St#st.id_tree), + element(1, Reds). + + +get_epochs(#st{header = Header}) -> + couch_bt_engine_header:get(Header, epochs). + + +get_last_purged(#st{header = Header} = St) -> + case couch_bt_engine_header:get(Header, purged_docs) of + nil -> + []; + Pointer -> + {ok, PurgeInfo} = couch_file:pread_term(St#st.fd, Pointer), + PurgeInfo + end. + + +get_purge_seq(#st{header = Header}) -> + couch_bt_engine_header:get(Header, purge_seq). + + +get_revs_limit(#st{header = Header}) -> + couch_bt_engine_header:get(Header, revs_limit). + + +get_size_info(#st{} = St) -> + {ok, FileSize} = couch_file:bytes(St#st.fd), + {ok, DbReduction} = couch_btree:full_reduce(St#st.id_tree), + SizeInfo0 = element(3, DbReduction), + SizeInfo = case SizeInfo0 of + SI when is_record(SI, size_info) -> + SI; + {AS, ES} -> + #size_info{active=AS, external=ES}; + AS -> + #size_info{active=AS} + end, + ActiveSize = active_size(St, SizeInfo), + ExternalSize = SizeInfo#size_info.external, + [ + {active, ActiveSize}, + {external, ExternalSize}, + {file, FileSize} + ]. + + +get_security(#st{header = Header} = St) -> + case couch_bt_engine_header:get(Header, security_ptr) of + undefined -> + []; + Pointer -> + {ok, SecProps} = couch_file:pread_term(St#st.fd, Pointer), + SecProps + end. + + +get_update_seq(#st{header = Header}) -> + couch_bt_engine_header:get(Header, update_seq). + + +get_uuid(#st{header = Header}) -> + couch_bt_engine_header:get(Header, uuid). + + +set_revs_limit(#st{header = Header} = St, RevsLimit) -> + NewSt = St#st{ + header = couch_bt_engine_header:set(Header, [ + {revs_limit, RevsLimit} + ]), + needs_commit = true + }, + {ok, increment_update_seq(NewSt)}. + + +set_security(#st{header = Header} = St, NewSecurity) -> + Options = [{compression, St#st.compression}], + {ok, Ptr, _} = couch_file:append_term(St#st.fd, NewSecurity, Options), + NewSt = St#st{ + header = couch_bt_engine_header:set(Header, [ + {security_ptr, Ptr} + ]), + needs_commit = true + }, + {ok, increment_update_seq(NewSt)}. + + +open_docs(#st{} = St, DocIds) -> + Results = couch_btree:lookup(St#st.id_tree, DocIds), + lists:map(fun + ({ok, FDI}) -> FDI; + (not_found) -> not_found + end, Results). + + +open_local_docs(#st{} = St, DocIds) -> + Results = couch_btree:lookup(St#st.local_tree, DocIds), + lists:map(fun + ({ok, #doc{} = Doc}) -> Doc; Review comment: @davisp yes, I understand that, what I meant: we are guarding on `Doc` to be `#doc` record here, but we are not doing the same above and just passing `FDI` as it is. It is not a big of a deal, just seems inconsistent and implies that there are more into it. ---------------------------------------------------------------- This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
