%% Copyright (c) 2024-2026 Benoit Chesneau %% %% This file is provided to you 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(compaction_filter). -include_lib("eunit/include/eunit.hrl"). %% Test declarative rules: key prefix filter filter_key_prefix_test() -> DbPath = "compaction_filter_prefix.test", rocksdb_test_util:rm_rf(DbPath), {ok, Db} = rocksdb:open(DbPath, [ {create_if_missing, true}, %% Configure for easier compaction triggering {write_buffer_size, 64 * 1024}, % Small write buffer {level0_file_num_compaction_trigger, 1}, {compaction_filter, #{ rules => [{key_prefix, <<"tmp_">>}] }} ]), %% Write more data to trigger flush and compaction lists:foreach(fun(N) -> Key = iolist_to_binary(["tmp_key", integer_to_list(N)]), Value = iolist_to_binary(["value", integer_to_list(N), binary:copy(<<"x">>, 1000)]), ok = rocksdb:put(Db, Key, Value, []) end, lists:seq(1, 100)), lists:foreach(fun(N) -> Key = iolist_to_binary(["keep_key", integer_to_list(N)]), Value = iolist_to_binary(["value", integer_to_list(N), binary:copy(<<"y">>, 1000)]), ok = rocksdb:put(Db, Key, Value, []) end, lists:seq(1, 100)), %% Flush and compact with force to ensure filter runs on all data ok = rocksdb:flush(Db, []), timer:sleep(100), ok = rocksdb:compact_range(Db, <<"keep_key">>, <<"tmp_key~">>, [{bottommost_level_compaction, force}]), %% Check results - tmp_ keys should be deleted TmpResult = rocksdb:get(Db, <<"tmp_key50">>, []), KeepResult = rocksdb:get(Db, <<"keep_key50">>, []), %% Log results for debugging io:format("tmp_key50 result: ~p~n", [TmpResult]), io:format("keep_key50 result: ~p~n", [KeepResult]), %% Verify keep keys are still there {ok, _} = KeepResult, %% tmp_ keys should be deleted after forced compaction not_found = TmpResult, ok = rocksdb:close(Db), ok = destroy_and_rm(DbPath). %% Test declarative rules: key suffix filter filter_key_suffix_test() -> DbPath = "compaction_filter_suffix.test", rocksdb_test_util:rm_rf(DbPath), {ok, Db} = rocksdb:open(DbPath, [ {create_if_missing, true}, {write_buffer_size, 64 * 1024}, {level0_file_num_compaction_trigger, 1}, {compaction_filter, #{ rules => [{key_suffix, <<"_expired">>}] }} ]), %% Write enough data to trigger compaction lists:foreach(fun(N) -> Key = iolist_to_binary(["key", integer_to_list(N), "_expired"]), Value = iolist_to_binary(["value", integer_to_list(N), binary:copy(<<"x">>, 1000)]), ok = rocksdb:put(Db, Key, Value, []) end, lists:seq(1, 100)), lists:foreach(fun(N) -> Key = iolist_to_binary(["key", integer_to_list(N), "_active"]), Value = iolist_to_binary(["value", integer_to_list(N), binary:copy(<<"y">>, 1000)]), ok = rocksdb:put(Db, Key, Value, []) end, lists:seq(1, 100)), %% Flush and compact with force to ensure filter runs on all data ok = rocksdb:flush(Db, []), timer:sleep(100), ok = rocksdb:compact_range(Db, <<"key">>, <<"key~">>, [{bottommost_level_compaction, force}]), %% Verify active keys are kept {ok, _} = rocksdb:get(Db, <<"key50_active">>, []), %% Expired keys should be deleted after forced compaction not_found = rocksdb:get(Db, <<"key50_expired">>, []), ok = rocksdb:close(Db), ok = destroy_and_rm(DbPath). %% Test declarative rules: value empty filter %% Tests that keys with empty values are deleted during compaction. filter_value_empty_test() -> DbPath = "compaction_filter_empty.test", rocksdb_test_util:rm_rf(DbPath), {ok, Db} = rocksdb:open(DbPath, [ {create_if_missing, true}, {write_buffer_size, 64 * 1024}, {level0_file_num_compaction_trigger, 1}, {compaction_filter, #{ rules => [{value_empty}] }} ]), %% Write empty values with small padding in key to ensure SST creation (with sync) lists:foreach(fun(N) -> Key = iolist_to_binary(["empty_key", integer_to_list(N), binary:copy(<<"p">>, 100)]), ok = rocksdb:put(Db, Key, <<>>, [{sync, true}]) end, lists:seq(1, 100)), %% Write non-empty values (with sync) lists:foreach(fun(N) -> Key = iolist_to_binary(["nonempty_key", integer_to_list(N)]), Value = iolist_to_binary(["value", integer_to_list(N), binary:copy(<<"x">>, 1000)]), ok = rocksdb:put(Db, Key, Value, [{sync, true}]) end, lists:seq(1, 100)), %% Flush and wait a moment ok = rocksdb:flush(Db, []), timer:sleep(100), %% Compact specific key ranges (not the whole db) ok = rocksdb:compact_range(Db, <<"empty_key">>, <<"empty_key~">>, [{bottommost_level_compaction, force}]), ok = rocksdb:compact_range(Db, <<"nonempty_key">>, <<"nonempty_key~">>, [{bottommost_level_compaction, force}]), %% ASSERT: Verify non-empty values are kept {ok, _} = rocksdb:get(Db, <<"nonempty_key50">>, []), {ok, _} = rocksdb:get(Db, <<"nonempty_key1">>, []), %% ASSERT: Empty value keys should be deleted EmptyKey1 = iolist_to_binary(["empty_key1", binary:copy(<<"p">>, 100)]), EmptyKey50 = iolist_to_binary(["empty_key50", binary:copy(<<"p">>, 100)]), not_found = rocksdb:get(Db, EmptyKey1, []), not_found = rocksdb:get(Db, EmptyKey50, []), ok = rocksdb:close(Db), ok = destroy_and_rm(DbPath). %% Test declarative rules: multiple rules %% Tests that multiple rules can be combined. Each rule should match %% different patterns (prefix, suffix, empty value). filter_multiple_rules_test() -> DbPath = "compaction_filter_multi.test", rocksdb_test_util:rm_rf(DbPath), {ok, Db} = rocksdb:open(DbPath, [ {create_if_missing, true}, {write_buffer_size, 64 * 1024}, {level0_file_num_compaction_trigger, 1}, {compaction_filter, #{ rules => [ {key_prefix, <<"tmp_">>}, {key_suffix, <<"_old">>} ] }} ]), %% Write data that matches prefix rule lists:foreach(fun(N) -> Key = iolist_to_binary(["tmp_key", integer_to_list(N)]), Value = iolist_to_binary(["value", integer_to_list(N), binary:copy(<<"x">>, 1000)]), ok = rocksdb:put(Db, Key, Value, []) end, lists:seq(1, 100)), %% Write data that matches suffix rule lists:foreach(fun(N) -> Key = iolist_to_binary(["key", integer_to_list(N), "_old"]), Value = iolist_to_binary(["value", integer_to_list(N), binary:copy(<<"y">>, 1000)]), ok = rocksdb:put(Db, Key, Value, []) end, lists:seq(1, 100)), %% Write data that should be kept lists:foreach(fun(N) -> Key = iolist_to_binary(["keep_key", integer_to_list(N)]), Value = iolist_to_binary(["value", integer_to_list(N), binary:copy(<<"z">>, 1000)]), ok = rocksdb:put(Db, Key, Value, []) end, lists:seq(1, 100)), %% Flush and compact with force to ensure filter runs on all data ok = rocksdb:flush(Db, []), timer:sleep(100), ok = rocksdb:compact_range(Db, <<"keep">>, <<"tmp~">>, [{bottommost_level_compaction, force}]), %% Verify kept keys are still there {ok, _} = rocksdb:get(Db, <<"keep_key50">>, []), %% Matching keys should be deleted after forced compaction not_found = rocksdb:get(Db, <<"tmp_key50">>, []), not_found = rocksdb:get(Db, <<"key50_old">>, []), ok = rocksdb:close(Db), ok = destroy_and_rm(DbPath). %% Test TTL from key - expired keys %% The TTL is extracted from the first 8 bytes of the key (big-endian timestamp) filter_ttl_from_key_test() -> DbPath = "compaction_filter_ttl.test", rocksdb_test_util:rm_rf(DbPath), {ok, Db} = rocksdb:open(DbPath, [ {create_if_missing, true}, {write_buffer_size, 64 * 1024}, {level0_file_num_compaction_trigger, 1}, {compaction_filter, #{ rules => [{ttl_from_key, 0, 8, 1}] % First 8 bytes = timestamp, 1 second TTL }} ]), %% Create keys with expired timestamps (10 seconds ago to ensure expiry) ExpiredTs = erlang:system_time(second) - 10, lists:foreach(fun(N) -> Key = <>, Value = iolist_to_binary(["value", integer_to_list(N), binary:copy(<<"x">>, 1000)]), ok = rocksdb:put(Db, Key, Value, []) end, lists:seq(1, 100)), %% Create keys with valid timestamps (1 hour in future) ValidTs = erlang:system_time(second) + 3600, lists:foreach(fun(N) -> Key = <>, Value = iolist_to_binary(["value", integer_to_list(N), binary:copy(<<"y">>, 1000)]), ok = rocksdb:put(Db, Key, Value, []) end, lists:seq(1, 100)), %% Flush and force compaction ok = rocksdb:flush(Db, []), timer:sleep(100), ok = rocksdb:compact_range(Db, <>, <>, [{bottommost_level_compaction, force}]), %% ASSERT: Valid keys should remain ValidKey1 = <>, ValidKey50 = <>, {ok, _} = rocksdb:get(Db, ValidKey1, []), {ok, _} = rocksdb:get(Db, ValidKey50, []), %% ASSERT: Expired keys should be deleted ExpiredKey1 = <>, ExpiredKey50 = <>, not_found = rocksdb:get(Db, ExpiredKey1, []), not_found = rocksdb:get(Db, ExpiredKey50, []), ok = rocksdb:close(Db), ok = destroy_and_rm(DbPath). %% Test Erlang callback mode - basic handler %% This test verifies that the Erlang handler actually filters keys during compaction. filter_erlang_handler_test() -> DbPath = "compaction_filter_handler.test", rocksdb_test_util:rm_rf(DbPath), %% Start handler that removes keys starting with "delete_" Self = self(), Handler = spawn_link(fun() -> filter_handler_loop(Self) end), {ok, Db} = rocksdb:open(DbPath, [ {create_if_missing, true}, {write_buffer_size, 64 * 1024}, {level0_file_num_compaction_trigger, 1}, {compaction_filter, #{ handler => Handler, batch_size => 10, timeout => 5000 % 5 second timeout }} ]), %% Write data to delete lists:foreach(fun(N) -> Key = iolist_to_binary(["delete_key", integer_to_list(N)]), Value = iolist_to_binary(["value", integer_to_list(N), binary:copy(<<"x">>, 1000)]), ok = rocksdb:put(Db, Key, Value, []) end, lists:seq(1, 50)), %% Write data to keep lists:foreach(fun(N) -> Key = iolist_to_binary(["keep_key", integer_to_list(N)]), Value = iolist_to_binary(["value", integer_to_list(N), binary:copy(<<"y">>, 1000)]), ok = rocksdb:put(Db, Key, Value, []) end, lists:seq(1, 50)), %% Flush and force compaction ok = rocksdb:flush(Db, []), timer:sleep(100), ok = rocksdb:compact_range(Db, <<"delete">>, <<"keep~">>, [{bottommost_level_compaction, force}]), %% Wait for handler to process and collect total processed count TotalProcessed = collect_handler_processed(0, 2000), io:format("Handler processed ~p total keys~n", [TotalProcessed]), %% Verify handler was invoked - should have processed at least some keys ?assert(TotalProcessed > 0), %% ASSERT: delete_ keys should be removed not_found = rocksdb:get(Db, <<"delete_key1">>, []), not_found = rocksdb:get(Db, <<"delete_key25">>, []), not_found = rocksdb:get(Db, <<"delete_key50">>, []), %% ASSERT: keep_ keys should still exist {ok, _} = rocksdb:get(Db, <<"keep_key1">>, []), {ok, _} = rocksdb:get(Db, <<"keep_key25">>, []), {ok, _} = rocksdb:get(Db, <<"keep_key50">>, []), %% Clean up Handler ! stop, ok = rocksdb:close(Db), ok = destroy_and_rm(DbPath). %% Helper to collect all handler_processed messages collect_handler_processed(Total, Timeout) -> receive {handler_processed, Count} -> collect_handler_processed(Total + Count, Timeout) after Timeout -> Total end. filter_handler_loop(Parent) -> receive {compaction_filter, BatchRef, Keys} -> Decisions = lists:map(fun({_Level, Key, _Value}) -> case Key of <<"delete_", _/binary>> -> remove; _ -> keep end end, Keys), rocksdb:compaction_filter_reply(BatchRef, Decisions), Parent ! {handler_processed, length(Keys)}, filter_handler_loop(Parent); stop -> ok after 60000 -> Parent ! {handler_processed, 0} end. %% Test timeout handling - handler that doesn't respond should not crash filter_handler_timeout_test() -> DbPath = "compaction_filter_timeout.test", rocksdb_test_util:rm_rf(DbPath), %% Handler that never responds SlowHandler = spawn(fun() -> receive _ -> timer:sleep(infinity) end end), {ok, Db} = rocksdb:open(DbPath, [ {create_if_missing, true}, {write_buffer_size, 64 * 1024}, {level0_file_num_compaction_trigger, 1}, {compaction_filter, #{ handler => SlowHandler, timeout => 100 % 100ms timeout }} ]), %% Write enough data to trigger compaction lists:foreach(fun(N) -> Key = iolist_to_binary(["key", integer_to_list(N)]), Value = iolist_to_binary(["value", integer_to_list(N), binary:copy(<<"x">>, 1000)]), ok = rocksdb:put(Db, Key, Value, []) end, lists:seq(1, 100)), %% Flush and force compaction - should NOT hang or crash ok = rocksdb:flush(Db, []), ok = rocksdb:compact_range(Db, undefined, undefined, [{bottommost_level_compaction, force}]), %% Keys should still exist (timeout = keep) {ok, _} = rocksdb:get(Db, <<"key50">>, []), exit(SlowHandler, kill), ok = rocksdb:close(Db), ok = destroy_and_rm(DbPath). %% Test dead handler doesn't crash filter_handler_dead_test() -> DbPath = "compaction_filter_dead.test", rocksdb_test_util:rm_rf(DbPath), %% Handler that dies immediately Handler = spawn(fun() -> ok end), timer:sleep(50), % Ensure it's dead {ok, Db} = rocksdb:open(DbPath, [ {create_if_missing, true}, {write_buffer_size, 64 * 1024}, {level0_file_num_compaction_trigger, 1}, {compaction_filter, #{handler => Handler}} ]), %% Write enough data to trigger compaction lists:foreach(fun(N) -> Key = iolist_to_binary(["key", integer_to_list(N)]), Value = iolist_to_binary(["value", integer_to_list(N), binary:copy(<<"x">>, 1000)]), ok = rocksdb:put(Db, Key, Value, []) end, lists:seq(1, 100)), %% Flush and force compaction - should NOT crash ok = rocksdb:flush(Db, []), ok = rocksdb:compact_range(Db, undefined, undefined, [{bottommost_level_compaction, force}]), %% Keys preserved (dead handler = keep) {ok, _} = rocksdb:get(Db, <<"key50">>, []), ok = rocksdb:close(Db), ok = destroy_and_rm(DbPath). %% Test handler change_value decision - handler can modify values during compaction filter_handler_change_value_test() -> DbPath = "compaction_filter_change_value.test", rocksdb_test_util:rm_rf(DbPath), Self = self(), Handler = spawn_link(fun() -> change_value_handler_loop(Self) end), {ok, Db} = rocksdb:open(DbPath, [ {create_if_missing, true}, {write_buffer_size, 64 * 1024}, {level0_file_num_compaction_trigger, 1}, {compaction_filter, #{ handler => Handler, batch_size => 10, timeout => 5000 }} ]), %% Write data to transform lists:foreach(fun(N) -> Key = iolist_to_binary(["transform_key", integer_to_list(N)]), Value = iolist_to_binary(["original_value", integer_to_list(N), binary:copy(<<"x">>, 1000)]), ok = rocksdb:put(Db, Key, Value, []) end, lists:seq(1, 50)), %% Write data to keep unchanged lists:foreach(fun(N) -> Key = iolist_to_binary(["normal_key", integer_to_list(N)]), Value = iolist_to_binary(["normal_value", integer_to_list(N), binary:copy(<<"y">>, 1000)]), ok = rocksdb:put(Db, Key, Value, []) end, lists:seq(1, 50)), %% Flush and force compaction ok = rocksdb:flush(Db, []), timer:sleep(100), ok = rocksdb:compact_range(Db, <<"normal">>, <<"transform~">>, [{bottommost_level_compaction, force}]), %% Wait for handler to process TotalProcessed = collect_handler_processed(0, 2000), io:format("Change value handler processed ~p total keys~n", [TotalProcessed]), ?assert(TotalProcessed > 0), %% ASSERT: transform_ keys should have modified values {ok, Value1} = rocksdb:get(Db, <<"transform_key1">>, []), ?assertEqual(<<"MODIFIED">>, Value1), {ok, Value25} = rocksdb:get(Db, <<"transform_key25">>, []), ?assertEqual(<<"MODIFIED">>, Value25), %% ASSERT: normal_ keys should have original values (unchanged) {ok, NormalValue} = rocksdb:get(Db, <<"normal_key1">>, []), ?assertMatch(<<"normal_value1", _/binary>>, NormalValue), Handler ! stop, ok = rocksdb:close(Db), ok = destroy_and_rm(DbPath). change_value_handler_loop(Parent) -> receive {compaction_filter, BatchRef, Keys} -> Decisions = lists:map(fun({_Level, Key, _Value}) -> case Key of <<"transform_", _/binary>> -> {change_value, <<"MODIFIED">>}; _ -> keep end end, Keys), rocksdb:compaction_filter_reply(BatchRef, Decisions), Parent ! {handler_processed, length(Keys)}, change_value_handler_loop(Parent); stop -> ok after 60000 -> ok end. %% Test that handler is actually invoked with all keys filter_handler_invocation_test() -> DbPath = "compaction_filter_invocation.test", rocksdb_test_util:rm_rf(DbPath), Self = self(), %% Use ETS to track received keys Tab = ets:new(received_keys, [set, public]), Handler = spawn_link(fun() -> tracking_handler_loop(Self, Tab) end), {ok, Db} = rocksdb:open(DbPath, [ {create_if_missing, true}, {write_buffer_size, 64 * 1024}, {level0_file_num_compaction_trigger, 1}, {compaction_filter, #{ handler => Handler, batch_size => 20, timeout => 5000 }} ]), %% Write known keys - need enough data to trigger actual compaction ExpectedKeys = [iolist_to_binary(["key", integer_to_list(N)]) || N <- lists:seq(1, 100)], lists:foreach(fun(Key) -> Value = iolist_to_binary([Key, binary:copy(<<"x">>, 1000)]), ok = rocksdb:put(Db, Key, Value, []) end, ExpectedKeys), %% Flush and force compaction ok = rocksdb:flush(Db, []), ok = rocksdb:compact_range(Db, undefined, undefined, [{bottommost_level_compaction, force}]), %% Wait for handler TotalProcessed = collect_handler_processed(0, 2000), io:format("Tracking handler processed ~p keys~n", [TotalProcessed]), ?assert(TotalProcessed > 0), %% Verify all keys were seen by handler ReceivedKeys = [K || {K} <- ets:tab2list(Tab)], io:format("Received keys count: ~p~n", [length(ReceivedKeys)]), %% All expected keys should have been received lists:foreach(fun(Key) -> ?assert(lists:member(Key, ReceivedKeys)) end, ExpectedKeys), Handler ! stop, ets:delete(Tab), ok = rocksdb:close(Db), ok = destroy_and_rm(DbPath). tracking_handler_loop(Parent, Tab) -> receive {compaction_filter, BatchRef, Keys} -> %% Track all received keys lists:foreach(fun({_Level, Key, _Value}) -> ets:insert(Tab, {Key}) end, Keys), %% Keep all keys Decisions = [keep || _ <- Keys], rocksdb:compaction_filter_reply(BatchRef, Decisions), Parent ! {handler_processed, length(Keys)}, tracking_handler_loop(Parent, Tab); stop -> ok after 60000 -> ok end. %% Test key_contains rule filter_key_contains_test() -> DbPath = "compaction_filter_contains.test", rocksdb_test_util:rm_rf(DbPath), {ok, Db} = rocksdb:open(DbPath, [ {create_if_missing, true}, {write_buffer_size, 64 * 1024}, {level0_file_num_compaction_trigger, 1}, {compaction_filter, #{ rules => [{key_contains, <<"_session_">>}] }} ]), %% Write keys containing _session_ pattern lists:foreach(fun(N) -> Key = iolist_to_binary(["user_session_", integer_to_list(N), "_data"]), Value = iolist_to_binary(["session_data", integer_to_list(N), binary:copy(<<"x">>, 1000)]), ok = rocksdb:put(Db, Key, Value, []) end, lists:seq(1, 100)), %% Write keys NOT containing the pattern lists:foreach(fun(N) -> Key = iolist_to_binary(["user_data_", integer_to_list(N)]), Value = iolist_to_binary(["user_data", integer_to_list(N), binary:copy(<<"y">>, 1000)]), ok = rocksdb:put(Db, Key, Value, []) end, lists:seq(1, 100)), %% Flush and force compaction ok = rocksdb:flush(Db, []), timer:sleep(100), ok = rocksdb:compact_range(Db, <<"user">>, <<"user~">>, [{bottommost_level_compaction, force}]), %% ASSERT: keys containing _session_ should be deleted not_found = rocksdb:get(Db, <<"user_session_50_data">>, []), not_found = rocksdb:get(Db, <<"user_session_1_data">>, []), %% ASSERT: keys NOT containing pattern should remain {ok, _} = rocksdb:get(Db, <<"user_data_50">>, []), {ok, _} = rocksdb:get(Db, <<"user_data_1">>, []), ok = rocksdb:close(Db), ok = destroy_and_rm(DbPath). %% Test value_prefix rule filter_value_prefix_test() -> DbPath = "compaction_filter_value_prefix.test", rocksdb_test_util:rm_rf(DbPath), {ok, Db} = rocksdb:open(DbPath, [ {create_if_missing, true}, {write_buffer_size, 64 * 1024}, {level0_file_num_compaction_trigger, 1}, {compaction_filter, #{ rules => [{value_prefix, <<"DELETED:">>}] }} ]), %% Write keys with values starting with DELETED: lists:foreach(fun(N) -> Key = iolist_to_binary(["marked_key", integer_to_list(N)]), Value = iolist_to_binary(["DELETED:", integer_to_list(N), binary:copy(<<"x">>, 1000)]), ok = rocksdb:put(Db, Key, Value, []) end, lists:seq(1, 100)), %% Write keys with normal values lists:foreach(fun(N) -> Key = iolist_to_binary(["normal_key", integer_to_list(N)]), Value = iolist_to_binary(["ACTIVE:", integer_to_list(N), binary:copy(<<"y">>, 1000)]), ok = rocksdb:put(Db, Key, Value, []) end, lists:seq(1, 100)), %% Flush and force compaction ok = rocksdb:flush(Db, []), timer:sleep(100), ok = rocksdb:compact_range(Db, <<"marked">>, <<"normal~">>, [{bottommost_level_compaction, force}]), %% ASSERT: keys with DELETED: value prefix should be removed not_found = rocksdb:get(Db, <<"marked_key50">>, []), not_found = rocksdb:get(Db, <<"marked_key1">>, []), %% ASSERT: keys with normal values should remain {ok, _} = rocksdb:get(Db, <<"normal_key50">>, []), {ok, _} = rocksdb:get(Db, <<"normal_key1">>, []), ok = rocksdb:close(Db), ok = destroy_and_rm(DbPath). %% Test always_delete rule filter_always_delete_test() -> DbPath = "compaction_filter_always_delete.test", rocksdb_test_util:rm_rf(DbPath), {ok, Db} = rocksdb:open(DbPath, [ {create_if_missing, true}, {write_buffer_size, 64 * 1024}, {level0_file_num_compaction_trigger, 1}, {compaction_filter, #{ rules => [{always_delete}] }} ]), %% Write various keys with sync to ensure durability lists:foreach(fun(N) -> Key = iolist_to_binary(["key", integer_to_list(N)]), Value = iolist_to_binary(["value", integer_to_list(N), binary:copy(<<"x">>, 1000)]), ok = rocksdb:put(Db, Key, Value, [{sync, true}]) end, lists:seq(1, 100)), %% Flush and wait a moment ok = rocksdb:flush(Db, []), timer:sleep(100), %% Compact specific key range (not the whole db) ok = rocksdb:compact_range(Db, <<"key">>, <<"key~">>, [{bottommost_level_compaction, force}]), %% ASSERT: ALL keys should be deleted not_found = rocksdb:get(Db, <<"key1">>, []), not_found = rocksdb:get(Db, <<"key50">>, []), not_found = rocksdb:get(Db, <<"key100">>, []), ok = rocksdb:close(Db), ok = destroy_and_rm(DbPath). %% Test handler with invalid decisions - should default to keep filter_handler_invalid_decision_test() -> DbPath = "compaction_filter_invalid.test", rocksdb_test_util:rm_rf(DbPath), Self = self(), Handler = spawn_link(fun() -> invalid_decision_handler_loop(Self) end), {ok, Db} = rocksdb:open(DbPath, [ {create_if_missing, true}, {write_buffer_size, 64 * 1024}, {level0_file_num_compaction_trigger, 1}, {compaction_filter, #{ handler => Handler, batch_size => 10, timeout => 5000 }} ]), %% Write data lists:foreach(fun(N) -> Key = iolist_to_binary(["key", integer_to_list(N)]), Value = iolist_to_binary(["value", integer_to_list(N), binary:copy(<<"x">>, 1000)]), ok = rocksdb:put(Db, Key, Value, []) end, lists:seq(1, 50)), %% Flush and force compaction ok = rocksdb:flush(Db, []), ok = rocksdb:compact_range(Db, undefined, undefined, [{bottommost_level_compaction, force}]), %% Wait for handler TotalProcessed = collect_handler_processed(0, 2000), io:format("Invalid decision handler processed ~p keys~n", [TotalProcessed]), %% ASSERT: All keys should be kept (invalid decisions default to keep) {ok, _} = rocksdb:get(Db, <<"key1">>, []), {ok, _} = rocksdb:get(Db, <<"key25">>, []), {ok, _} = rocksdb:get(Db, <<"key50">>, []), Handler ! stop, ok = rocksdb:close(Db), ok = destroy_and_rm(DbPath). invalid_decision_handler_loop(Parent) -> receive {compaction_filter, BatchRef, Keys} -> %% Return invalid decisions - should default to keep Decisions = [invalid_decision || _ <- Keys], rocksdb:compaction_filter_reply(BatchRef, Decisions), Parent ! {handler_processed, length(Keys)}, invalid_decision_handler_loop(Parent); stop -> ok after 60000 -> ok end. %% Test compaction filter with column families using rule-based filters filter_column_family_test() -> DbPath = "compaction_filter_cf.test", rocksdb_test_util:rm_rf(DbPath), %% First create the DB with column families {ok, Db0} = rocksdb:open(DbPath, [{create_if_missing, true}]), {ok, _CF1} = rocksdb:create_column_family(Db0, "cf1", []), {ok, _CF2} = rocksdb:create_column_family(Db0, "cf2", []), ok = rocksdb:close(Db0), %% Reopen with CF-specific filters - use distinct option lists CF1Opts = [ {write_buffer_size, 64 * 1024}, {compaction_filter, #{rules => [{key_prefix, <<"tmp_">>}]}} ], CF2Opts = [ {write_buffer_size, 64 * 1024}, {compaction_filter, #{rules => [{key_prefix, <<"old_">>}]}} ], {ok, Db, [_DefaultCF, CF1, CF2]} = rocksdb:open(DbPath, [ {create_if_missing, true} ], [ {"default", []}, {"cf1", CF1Opts}, {"cf2", CF2Opts} ]), %% Write to CF1 (rule-based filter for tmp_ prefix) lists:foreach(fun(N) -> TmpKey = iolist_to_binary([<<"tmp_key">>, integer_to_binary(N)]), KeepKey = iolist_to_binary([<<"keep_key">>, integer_to_binary(N)]), Value = iolist_to_binary([<<"value">>, integer_to_binary(N), binary:copy(<<"x">>, 1000)]), ok = rocksdb:put(Db, CF1, TmpKey, Value, []), ok = rocksdb:put(Db, CF1, KeepKey, Value, []) end, lists:seq(1, 100)), %% Write to CF2 (rule-based filter for old_ prefix) lists:foreach(fun(N) -> OldKey = iolist_to_binary([<<"old_key">>, integer_to_binary(N)]), NewKey = iolist_to_binary([<<"new_key">>, integer_to_binary(N)]), Value = iolist_to_binary([<<"value">>, integer_to_binary(N), binary:copy(<<"y">>, 1000)]), ok = rocksdb:put(Db, CF2, OldKey, Value, []), ok = rocksdb:put(Db, CF2, NewKey, Value, []) end, lists:seq(1, 100)), %% Flush and compact both CFs with specific bounds ok = rocksdb:flush(Db, CF1, []), ok = rocksdb:flush(Db, CF2, []), %% Compact CF1: keys range from keep_key1 to tmp_key99 (alphabetically) ok = rocksdb:compact_range(Db, CF1, <<"keep_key1">>, <<"tmp_key99">>, [{bottommost_level_compaction, force}]), %% Compact CF2: keys range from new_key1 to old_key99 (alphabetically) ok = rocksdb:compact_range(Db, CF2, <<"new_key1">>, <<"old_key99">>, [{bottommost_level_compaction, force}]), %% ASSERT CF1: tmp_ keys deleted, keep_ keys remain not_found = rocksdb:get(Db, CF1, <<"tmp_key15">>, []), {ok, _} = rocksdb:get(Db, CF1, <<"keep_key15">>, []), %% ASSERT CF2: old_ keys deleted, new_ keys remain not_found = rocksdb:get(Db, CF2, <<"old_key15">>, []), {ok, _} = rocksdb:get(Db, CF2, <<"new_key15">>, []), ok = rocksdb:close(Db), ok = destroy_and_rm(DbPath). %% Helper function destroy_and_rm(DbPath) -> rocksdb:destroy(DbPath, []), rocksdb_test_util:rm_rf(DbPath).