import pytest import redis import math import statistics import time from collections import defaultdict from utils import set_hertz from test_helper_classes import _insert_data from test_ts_range import build_expected_aligned_data from includes import * @skip(asan=True) def test_mrange_with_expire_cmd(env): set_hertz(env) with env.getClusterConnectionIfNeeded() as r, env.getConnection(1) as r1: assert r.execute_command("TS.ADD", "X" ,"*" ,"1" ,"LABELS", "type", "DELAYED") assert r.execute_command("TS.ADD", "Y" ,"*" ,"1" ,"LABELS", "type", "DELAYED") assert r.execute_command("TS.ADD", "Z" ,"*" ,"1" ,"LABELS", "type", "DELAYED") current_ts = time.time() assert r.execute_command("EXPIRE","X", 5) assert r.execute_command("EXPIRE","Y", 6) assert r.execute_command("EXPIRE","Z", 7) while time.time() < current_ts + 10: reply = r1.execute_command('TS.mrange', '-', '+', 'FILTER', 'type=DELAYED') assert(len(reply)>=0 and len(reply)<=3) assert r.execute_command("PING") def testWithMultiExec(env): if env.shardsCount < 2: env.skip() if not env.is_cluster(): env.skip() with env.getConnection() as r: r.execute_command('multi', ) r.execute_command('TS.mrange', '-', '+', 'FILTER', 'name=bob') if is_rlec(): with pytest.raises(redis.ResponseError): r.execute_command('exec') else: res = r.execute_command('exec') assert type(res[0]) is redis.ResponseError @skip(asan=True) def test_mrange_expire_issue549(env): env.skipOnDebugger() set_hertz(env) with env.getClusterConnectionIfNeeded() as r: assert r.execute_command('ts.add', 'k1', 1, 10, 'LABELS', 'l', '1') == 1 assert r.execute_command('ts.add', 'k2', 2, 20, 'LABELS', 'l', '1') == 2 assert r.execute_command('expire', 'k1', '1') == 1 for i in range(0, 5000): assert env.getConnection().execute_command('ts.mrange - + aggregation avg 10 withlabels filter l=1') is not None def test_range_by_labels(): start_ts = 1511885909 samples_count = 50 for mode in ["UNCOMPRESSED", "COMPRESSED"]: env = Env() with env.getClusterConnectionIfNeeded() as r, env.getConnection(1) as r1: assert r.execute_command('TS.CREATE', 'tester1', mode, 'LABELS', 'name', 'bob', 'class', 'middle', 'generation', 'x') assert r.execute_command('TS.CREATE', 'tester2', mode, 'LABELS', 'name', 'rudy', 'class', 'junior', 'generation', 'x') assert r.execute_command('TS.CREATE', 'tester3', mode, 'LABELS', 'name', 'fabi', 'class', 'top', 'generation', 'x') _insert_data(r, 'tester1', start_ts, samples_count, 5) _insert_data(r, 'tester2', start_ts, samples_count, 15) _insert_data(r, 'tester3', start_ts, samples_count, 25) expected_result = [[start_ts + i, str(5).encode('ascii')] for i in range(samples_count)] actual_result = r1.execute_command('TS.mrange', start_ts, start_ts + samples_count, 'FILTER', 'name=bob') assert [[b'tester1', [], expected_result]] == actual_result expected_result.reverse() actual_result = r1.execute_command('TS.mrevrange', start_ts, start_ts + samples_count, 'FILTER', 'name=bob') assert [[b'tester1', [], expected_result]] == actual_result def build_expected(val, time_bucket): return [[int(i - i % time_bucket), str(val).encode('ascii')] for i in range(start_ts, start_ts + samples_count + 1, time_bucket)] actual_result = r1.execute_command('TS.mrange', start_ts, start_ts + samples_count, 'AGGREGATION', 'LAST', 5, 'FILTER', 'generation=x') expected_result = [[b'tester1', [], build_expected(5, 5)], [b'tester2', [], build_expected(15, 5)], [b'tester3', [], build_expected(25, 5)], ] env.assertEqual(sorted(expected_result), sorted(actual_result)) assert expected_result[1:] == sorted(r1.execute_command('TS.mrange', start_ts, start_ts + samples_count, 'AGGREGATION', 'LAST', 5, 'FILTER', 'generation=x', 'class!=middle'), key=lambda x:x[0]) actual_result = r1.execute_command('TS.mrange', start_ts, start_ts + samples_count, 'COUNT', 3, 'AGGREGATION', 'LAST', 5, 'FILTER', 'generation=x') assert expected_result[0][2][:3] == sorted(actual_result, key=lambda x:x[0])[0][2] actual_result = r1.execute_command('TS.mrange', start_ts, start_ts + samples_count, 'AGGREGATION', 'COUNT', 5, 'FILTER', 'generation=x') assert [[1511885905, b'1']] == actual_result[0][2][:1] assert expected_result[0][2][1:9] == actual_result[0][2][1:9] actual_result = r1.execute_command('TS.mrange', start_ts, start_ts + samples_count, 'AGGREGATION', 'COUNT', 3, 'COUNT', 4, 'FILTER', 'generation=x') assert 4 == len(actual_result[0][2]) # just checking that agg count before count works actual_result = r1.execute_command('TS.mrange', start_ts, start_ts + samples_count, 'COUNT', 4, 'AGGREGATION', 'COUNT', 3, 'FILTER', 'generation=x') assert 4 == len(actual_result[0][2]) # just checking that agg count after count works actual_result = r1.execute_command('TS.mrange', start_ts, start_ts + samples_count, 'AGGREGATION', 'COUNT', 3, 'FILTER', 'generation=x') assert 18 == len(actual_result[0][2]) # just checking that agg count before count works with pytest.raises(redis.ResponseError) as excinfo: assert r1.execute_command('TS.mrange', start_ts, start_ts + samples_count, 'AGGREGATION', 'invalid', 3, 'FILTER', 'generation=x') with pytest.raises(redis.ResponseError) as excinfo: assert r1.execute_command('TS.mrange', start_ts, start_ts + samples_count, 'AGGREGATION', 'AVG', 'string', 'FILTER', 'generation=x') with pytest.raises(redis.ResponseError) as excinfo: assert r1.execute_command('TS.mrange', start_ts, start_ts + samples_count, 'COUNT', 'string', 'FILTER', 'generation=x') with pytest.raises(redis.ResponseError) as excinfo: assert r1.execute_command('TS.mrange', '-', '+' ,'FILTER') # missing args with pytest.raises(redis.ResponseError) as excinfo: assert r1.execute_command('TS.mrange', '-', '+', 'RETLIF') # no filter word with pytest.raises(redis.ResponseError) as excinfo: assert r1.execute_command('TS.mrange', 'string', start_ts + samples_count, 'FILTER', 'generation=x') with pytest.raises(redis.ResponseError) as excinfo: assert r1.execute_command('TS.mrange', start_ts, 'string', 'FILTER', 'generation=x') with pytest.raises(redis.ResponseError) as excinfo: assert r1.execute_command('TS.mrange', start_ts, start_ts + samples_count, 'FILTER', 'generation+x') with pytest.raises(redis.ResponseError) as excinfo: assert r1.execute_command('TS.mrange', start_ts, start_ts + samples_count, 'FILTER', 'generation!=x') # issue 414 with pytest.raises(redis.ResponseError) as excinfo: assert r1.execute_command('TS.mrange', start_ts, start_ts + samples_count, 'FILTER', 'name=(bob,rudy,)') with pytest.raises(redis.ResponseError) as excinfo: assert r1.execute_command('TS.mrange', start_ts, start_ts + samples_count, 'FILTER', 'name=(bob,,rudy)') # test SELECTED_LABELS with pytest.raises(redis.ResponseError) as excinfo: assert r1.execute_command('TS.mrange', start_ts, start_ts + samples_count, 'SELECTED_LABELS', 'filter', 'k!=5') with pytest.raises(redis.ResponseError) as excinfo: assert r1.execute_command('TS.mrange', start_ts, start_ts + samples_count, 'SELECTED_LABELS', 'filter', 'k!=5') with pytest.raises(redis.ResponseError) as excinfo: assert r1.execute_command('TS.mrange', start_ts, start_ts + samples_count, 'SELECTED_LABELS', 'WITHLABELS', 'filter', 'k!=5') with pytest.raises(redis.ResponseError) as excinfo: assert r1.execute_command('TS.mrange', start_ts, start_ts + samples_count, 'WITHLABELS', 'SELECTED_LABELS', 'filter', 'k!=5') env.flush() def test_mrange_filterby(env): start_ts = 1511885909 samples_count = 50 with env.getClusterConnectionIfNeeded() as r, env.getConnection(1) as r1: assert r.execute_command('TS.CREATE', 'tester1', 'LABELS', 'name', 'bob', 'class', 'middle', 'generation', 'x') assert r.execute_command('TS.CREATE', 'tester2', 'LABELS', 'name', 'rudy', 'class', 'junior', 'generation', 'x') assert r.execute_command('TS.CREATE', 'tester3', 'LABELS', 'name', 'fabi', 'class', 'top', 'generation', 'x') _insert_data(r, 'tester1', start_ts, samples_count, 5) _insert_data(r, 'tester2', start_ts, samples_count, 15) _insert_data(r, 'tester3', start_ts, samples_count, 25) with pytest.raises(redis.ResponseError) as excinfo: assert r1.execute_command('TS.mrange', start_ts, start_ts + samples_count, 'FILTER_BY_VALUE', "a", 1 ,'FILTER', 'name=bob') with pytest.raises(redis.ResponseError) as excinfo: assert r1.execute_command('TS.mrange', start_ts, start_ts + samples_count, 'FILTER_BY_VALUE', "a", "a" ,'FILTER', 'name=bob') with pytest.raises(redis.ResponseError) as excinfo: assert r1.execute_command('TS.mrange', start_ts, start_ts + samples_count, 'FILTER_BY_VALUE', 1, "a" ,'FILTER', 'name=bob') expected_result = [[b'tester1', [], []], [b'tester2', [], [[start_ts + i, str(15).encode('ascii')] for i in range(samples_count)]], [b'tester3', [], []], ] actual_result = r1.execute_command('TS.mrange', start_ts, start_ts + samples_count, 'FILTER_BY_VALUE', 10, 20,'FILTER', 'generation=x') env.assertEqual(sorted(actual_result), sorted(expected_result)) expected_result = [[b'tester1', [], []], [b'tester2', [], [[start_ts + i, str(15).encode('ascii')] for i in range(9, 12)]], [b'tester3', [], []], ] actual_result = r1.execute_command('TS.mrange', start_ts, start_ts + samples_count, 'FILTER_BY_TS', start_ts+9, start_ts+10, start_ts+11, 'FILTER_BY_VALUE', 10, 20,'FILTER', 'generation=x') env.assertEqual(sorted(actual_result), sorted(expected_result)) actual_result = r1.execute_command('TS.mrange', start_ts + 1000000, start_ts + 1000000 + samples_count, 'FILTER_BY_TS', start_ts + 1000000, 'FILTER', 'generation=x') assert sorted(actual_result) == sorted([[b'tester1', [], []], [b'tester2', [], []], [b'tester3', [], []]]) assert r.execute_command('TS.CREATE', 'tester4', 'LABELS', 'name', 'fabi', 'class', 'top', 'generation', 'z') r.execute_command('ts.add', 'tester4', 1, 1) r.execute_command('ts.add', 'tester4', 8, 8) actual_result = r1.execute_command('TS.mrange', 4, 6, 'FILTER_BY_TS', 4, 'FILTER', 'generation=z') assert actual_result == [[b'tester4', [], []]] def test_mrange_withlabels(env): start_ts = 1511885909 samples_count = 50 with env.getClusterConnectionIfNeeded() as r, env.getConnection(1) as r1: assert r.execute_command('TS.CREATE', 'tester1', 'LABELS', 'name', 'bob', 'class', 'middle', 'generation', 'x') assert r.execute_command('TS.CREATE', 'tester2', 'LABELS', 'name', 'rudy', 'class', 'junior', 'generation', 'x') assert r.execute_command('TS.CREATE', 'tester3', 'LABELS', 'name', 'fabi', 'class', 'top', 'generation', 'x') _insert_data(r, 'tester1', start_ts, samples_count, 5) _insert_data(r, 'tester2', start_ts, samples_count, 15) _insert_data(r, 'tester3', start_ts, samples_count, 25) expected_result = [[start_ts + i, str(5).encode('ascii')] for i in range(samples_count)] actual_result = r1.execute_command('TS.mrange', start_ts, start_ts + samples_count, 'WITHLABELS', 'FILTER', 'name=bob') assert [[b'tester1', [[b'name', b'bob'], [b'class', b'middle'], [b'generation', b'x']], expected_result]] == actual_result actual_result = r1.execute_command('TS.mrange', start_ts, start_ts + samples_count, 'SELECTED_LABELS', 'name', 'generation', 'FILTER', 'name=bob') assert [[b'tester1', [[b'name', b'bob'], [b'generation', b'x']], expected_result]] == actual_result actual_result = r1.execute_command('TS.mrange', start_ts + 1, start_ts + samples_count, 'WITHLABELS', 'AGGREGATION', 'COUNT', 1, 'FILTER', 'generation=x') # assert the labels length is 3 (name,class,generation) for each of the returned time-series try: assert len(actual_result[0][1]) == 3 and len(actual_result[1][1]) == 3 and len(actual_result[2][1]) == 3 except Exception as ex: print(str(actual_result)) res = r.execute_command('TS.INFO', 'tester1') print(str(res)) res = r.execute_command('TS.INFO', 'tester2') print(str(res)) res = r.execute_command('TS.INFO', 'tester3') print(str(res)) raise ex assert len(actual_result[0][1]) == 3 assert len(actual_result[1][1]) == 3 assert len(actual_result[2][1]) == 3 def test_multilabel_filter(env): with env.getClusterConnectionIfNeeded() as r, env.getConnection(1) as r1: assert r.execute_command('TS.CREATE', 'tester1', 'LABELS', 'name', 'bob', 'class', 'middle', 'generation', 'x', 'special', 'yes') assert r.execute_command('TS.CREATE', 'tester2', 'LABELS', 'name', 'rudy', 'class', 'junior', 'generation', 'x') assert r.execute_command('TS.CREATE', 'tester3', 'LABELS', 'name', 'fabi', 'class', 'top', 'generation', 'x') assert r.execute_command('TS.ADD', 'tester1', 0, 1) == 0 assert r.execute_command('TS.ADD', 'tester2', 0, 2) == 0 assert r.execute_command('TS.ADD', 'tester3', 0, 3) == 0 actual_result = r1.execute_command('TS.mrange', '-', '+', 'WITHLABELS', 'FILTER', 'name=(bob,rudy)') assert set(item[0] for item in actual_result) == set([b'tester1', b'tester2']) actual_result = r1.execute_command('TS.mrange', 0, '+', 'WITHLABELS', 'FILTER', 'name=(bob,rudy)', 'class!=(middle,top)') assert actual_result[0][0] == b'tester2' actual_result = r1.execute_command('TS.mget', 'WITHLABELS', 'FILTER', 'name=(bob,rudy)') assert set(item[0] for item in actual_result) == set([b'tester1', b'tester2']) actual_result = r1.execute_command('TS.mget', 'WITHLABELS', 'FILTER', 'name=(bob,rudy)', 'class!=(middle,top)') assert actual_result[0][0] == b'tester2' actual_result = r1.execute_command('TS.mget', 'WITHLABELS', 'FILTER', 'name=(bob,rudy)', 'class!=(middle,top)', 'generation=y') assert actual_result == [] actual_result = r1.execute_command('TS.mget', 'WITHLABELS', 'FILTER', 'name=(bob,rudy)', 'class!=(middle,top)', 'generation!=x') assert actual_result == [] actual_result = r1.execute_command('TS.mget', 'WITHLABELS', 'FILTER', 'name=(bob,rudy)', 'class!=(middle,top)', 'generation=') assert actual_result == [] actual_result = r1.execute_command('TS.mget', 'WITHLABELS', 'FILTER', 'name=(bob)', 'class=(middle,top,junior)', 'generation!=y') assert actual_result[0][0] == b'tester1' actual_result = r1.execute_command('TS.mget', 'WITHLABELS', 'FILTER', 'name=(bob)', 'class=(middle,top,junior)', 'generation!=') assert actual_result[0][0] == b'tester1' actual_result = r1.execute_command('TS.mget', 'WITHLABELS', 'FILTER', 'class=(top,junior,middle)', 'generation!=', 'special!=') assert actual_result[0][0] == b'tester1' def test_large_key_value_pairs(env): with env.getClusterConnectionIfNeeded() as r, env.getConnection(1) as r1: number_series = 100 for i in range(0,number_series): assert r.execute_command('TS.CREATE', f"ts-{i}", 'LABELS', 'baseAsset', '17049', 'counterAsset', '840', 'source', '1000', 'dataType', 'PRICE_TICK') kv_label1 = 'baseAsset=(13830,10249,16019,10135,17049,10777,10138,11036,11292,15778,11043,10025,11436,12207,13359,10807,12216,11833,10170,10811,12864,12738,10053,11334,12487,12619,12364,13266,11219,15827,12374,11223,10071,12249,11097,14430,13282,16226,13667,11365,12261,12646,12650,12397,12785,13941,10231,16254,12159,15103)' kv_label2 = 'counterAsset=(840)' kv_label3 = 'source=(1000)' kv_label4 = 'dataType=(PRICE_TICK)' kv_labels = [kv_label1, kv_label2, kv_label3, kv_label4] for kv_label in kv_labels: res = r1.execute_command('TS.MRANGE', '-', '+', 'FILTER', kv_label1) assert len(res) == number_series def ensure_replies_series_match(env,series_array_1, series_array_2): for ts in series_array_1: ts_name = ts[0] ts_labels =ts[1] ts_values =ts[2] for comparison_ts in series_array_2: comparison_ts_name = comparison_ts[0] comparison_ts_labels =comparison_ts[1] comparison_ts_values =comparison_ts[2] if ts_name == comparison_ts_name: env.assertEqual(ts_labels,comparison_ts_labels) env.assertEqual(ts_values,comparison_ts_values) def test_non_local_data(env): with env.getClusterConnectionIfNeeded() as r: r.execute_command('TS.ADD', '{host1}_metric_1', 1 ,100, 'LABELS', 'metric', 'cpu') r.execute_command('TS.ADD', '{host1}_metric_2', 2 ,40, 'LABELS', 'metric', 'cpu') r.execute_command('TS.ADD', '{host1}_metric_1', 2, 95) r.execute_command('TS.ADD', '{host1}_metric_1', 10, 99) previous_results = [] # ensure that initiating the query on different shards always replies with the same series for shard in range(0, env.shardsCount): shard_conn = env.getConnection(shard) actual_result = shard_conn.execute_command('TS.MRANGE - + FILTER metric=cpu') env.assertEqual(len(actual_result),2) for previous_result in previous_results: ensure_replies_series_match(env,previous_result,actual_result) previous_results.append(actual_result) def test_non_local_filtered_data(): env = Env() with env.getClusterConnectionIfNeeded() as r: r.execute_command('TS.ADD', '{host1}_metric_1', 1 ,100, 'LABELS', 'metric', 'cpu') r.execute_command('TS.ADD', '{host1}_metric_2', 2 ,40, 'LABELS', 'metric', 'cpu') r.execute_command('TS.ADD', '{host1}_metric_1', 2, 95) r.execute_command('TS.ADD', '{host1}_metric_1', 10, 99) previous_results = [] # ensure that initiating the query on different shards always replies with the same series for shard in range(0, env.shardsCount): shard_conn = env.getConnection(shard) # send undordered timestamps to test for sorting actual_result = shard_conn.execute_command('TS.MRANGE - + FILTER_BY_TS 11 5 25 55 101 18 9 1900 2 FILTER metric=cpu') env.assertEqual(len(actual_result),2) # ensure reply is properly filtered by TS for serie in actual_result: serie_ts = serie[2] # ensure only timestamp 2 is present on reply env.assertEqual(len(serie_ts),1) env.assertEqual(serie_ts[0][0],2) for previous_result in previous_results: ensure_replies_series_match(env,previous_result,actual_result) previous_results.append(actual_result) def test_non_local_filtered_labels(env): with env.getClusterConnectionIfNeeded() as r: r.execute_command('TS.ADD', '{host1}_metric_1', 1 ,100, 'LABELS', 'metric', 'cpu', '') r.execute_command('TS.ADD', '{host1}_metric_2', 2 ,40, 'LABELS', 'metric', 'cpu') r.execute_command('TS.ADD', '{host1}_metric_1', 2, 95) r.execute_command('TS.ADD', '{host1}_metric_1', 10, 99) previous_results = [] # ensure that initiating the query on different shards always replies with the same series for shard in range(0, env.shardsCount): shard_conn = env.getConnection(shard) actual_result = shard_conn.execute_command('TS.MRANGE - + FILTER_BY_TS 2 SELECTED_LABELS metric FILTER metric=cpu') env.assertEqual(len(actual_result),2) for previous_result in previous_results: ensure_replies_series_match(env,previous_result,actual_result) previous_results.append(actual_result) def test_mrange_align(): start_ts = 1511885909 samples_count = 50 env = Env(decodeResponses=True) with env.getClusterConnectionIfNeeded() as r, env.getConnection(1) as r1: assert r.execute_command('TS.CREATE', 'tester1', 'LABELS', 'name', 'bob', 'class', 'middle', 'generation', 'x') assert r.execute_command('TS.CREATE', 'tester2', 'LABELS', 'name', 'rudy', 'class', 'junior', 'generation', 'x') assert r.execute_command('TS.CREATE', 'tester3', 'LABELS', 'name', 'fabi', 'class', 'top', 'generation', 'x') _insert_data(r, 'tester1', start_ts, samples_count, 5) _insert_data(r, 'tester2', start_ts, samples_count, 15) _insert_data(r, 'tester3', start_ts, samples_count, 25) end_ts = start_ts + samples_count agg_bucket_size = 15 expected_start_result = [ ['tester1', [], build_expected_aligned_data(start_ts, start_ts + samples_count, agg_bucket_size, start_ts)], ['tester2', [], build_expected_aligned_data(start_ts, start_ts + samples_count, agg_bucket_size, start_ts)], ['tester3', [], build_expected_aligned_data(start_ts, start_ts + samples_count, agg_bucket_size, start_ts)], ] expected_end_result = [ ['tester1', [], build_expected_aligned_data(start_ts, start_ts + samples_count, agg_bucket_size, end_ts)], ['tester2', [], build_expected_aligned_data(start_ts, start_ts + samples_count, agg_bucket_size, end_ts)], ['tester3', [], build_expected_aligned_data(start_ts, start_ts + samples_count, agg_bucket_size, end_ts)], ] assert expected_start_result == decode_if_needed(sorted(r1.execute_command('TS.mrange', start_ts, start_ts + samples_count, 'ALIGN', '-', 'AGGREGATION', 'COUNT', agg_bucket_size, 'FILTER', 'generation=x'))) assert expected_end_result == decode_if_needed(sorted(r1.execute_command('TS.mrange', start_ts, start_ts + samples_count, 'ALIGN', '+', 'AGGREGATION', 'COUNT', agg_bucket_size, 'FILTER', 'generation=x'))) def groupby(data): result = defaultdict(lambda: 0) for key, labels, samples in data: for sample in samples: result[sample[0]] = max(result[sample[0]], int(sample[1])) return [[s[0], str(s[1])] for s in result.items()] expected_groupby_start_result = [['generation=x', [], groupby(expected_start_result)]] expected_groupby_end_result = [['generation=x', [], groupby(expected_end_result)]] assert expected_groupby_start_result == decode_if_needed(r1.execute_command('TS.mrange', start_ts, start_ts + samples_count, 'ALIGN', '-', 'AGGREGATION', 'COUNT', agg_bucket_size, 'FILTER', 'generation=x', 'GROUPBY', 'generation', 'REDUCE', 'max')) assert expected_groupby_end_result == decode_if_needed(r1.execute_command('TS.mrange', start_ts, start_ts + samples_count, 'ALIGN', '+', 'AGGREGATION', 'COUNT', agg_bucket_size, 'FILTER', 'generation=x', 'GROUPBY', 'generation', 'REDUCE', 'max')) def test_mrange_partial_range(): start_ts = 0 samples_count = 50 env = Env(decodeResponses=True) with env.getClusterConnectionIfNeeded() as r, env.getConnection(1) as r1: assert r.execute_command('TS.CREATE', 'tester1{1}', 'LABELS', 'name', 'bob') assert r.execute_command('TS.CREATE', 'tester2{3}', 'LABELS', 'name', 'fabi') _insert_data(r, 'tester1{1}', start_ts, samples_count, 5) _insert_data(r, 'tester2{3}', start_ts, samples_count, 15) exp = [['tester1{1}', [], [[0, '2'], [2, '2'], [4, '2'], [6, '2'], [8, '2'], [10, '1']]]] res = decode_if_needed(sorted(r1.execute_command('TS.mrange', start_ts, start_ts + 10, 'ALIGN', '-', 'AGGREGATION', 'COUNT', 2, 'FILTER', 'name=bob'))) exp = [['tester2{3}', [], [[0, '2'], [2, '2'], [4, '2'], [6, '2'], [8, '2'], [10, '1']]]] res = decode_if_needed(sorted(r1.execute_command('TS.mrange', start_ts, start_ts + 10, 'ALIGN', '-', 'AGGREGATION', 'COUNT', 2, 'FILTER', 'name=fabi'))) assert res == exp def test_latest_flag_mrange(): env = Env(decodeResponses=True) key1 = 't1{1}' key2 = 't2{1}' key3 = 't3{1}' key4 = 't4{1}' with env.getClusterConnectionIfNeeded() as r: assert r.execute_command('TS.CREATE', key1) assert r.execute_command('TS.CREATE', key2, 'LABELS', 'is_compaction', 'true') assert r.execute_command('TS.CREATE', key3) assert r.execute_command('TS.CREATE', key4, 'LABELS', 'is_compaction', 'true') assert r.execute_command('TS.CREATERULE', key1, key2, 'AGGREGATION', 'SUM', 10) assert r.execute_command('TS.CREATERULE', key3, key4, 'AGGREGATION', 'SUM', 10) assert r.execute_command('TS.add', key1, 1, 1) assert r.execute_command('TS.add', key1, 2, 3) assert r.execute_command('TS.add', key1, 11, 7) assert r.execute_command('TS.add', key1, 13, 1) res = r.execute_command('TS.range', key1, 0, 20) assert r.execute_command('TS.add', key3, 1, 1) assert r.execute_command('TS.add', key3, 2, 3) assert r.execute_command('TS.add', key3, 11, 7) assert r.execute_command('TS.add', key3, 13, 1) res = env.getConnection(1).execute_command('TS.mrange', 0, 10, 'FILTER', 'is_compaction=true') assert res == [['t2{1}', [], [[0, '4']]], ['t4{1}', [], [[0, '4']]]] or res == [[b't2{1}', [], [[0, b'4']]], [b't4{1}', [], [[0, b'4']]]] res = env.getConnection(1).execute_command('TS.mrange', 0, 10, 'LATEST', 'FILTER', 'is_compaction=true') assert res == [['t2{1}', [], [[0, '4'], [10, '8']]], ['t4{1}', [], [[0, '4'], [10, '8']]]] or res == [[b't2{1}', [], [[0, b'4'], [10, b'8']]], [b't4{1}', [], [[0, b'4'], [10, b'8']]]] res = env.getConnection(1).execute_command('TS.mrange', 0, 10, 'FILTER', 'is_compaction=true', 'GROUPBY', 'is_compaction', 'REDUCE', 'sum') assert res == [['is_compaction=true', [], [[0, '8']]]] or res == [[b'is_compaction=true', [], [[0, b'8']]]] res = env.getConnection(1).execute_command('TS.mrange', 0, 10, 'LATEST', 'FILTER', 'is_compaction=true', 'GROUPBY', 'is_compaction', 'REDUCE', 'sum') assert res == [['is_compaction=true', [], [[0, '8'], [10, '16']]]] or res == [[b'is_compaction=true', [], [[0, b'8'], [10, b'16']]]] # make sure LATEST haven't changed anything in the keys res = r.execute_command('TS.range', key2, 0, 10) assert res == [[0, '4']] or res == [[0, b'4']] res = r.execute_command('TS.range', key1, 0, 20) assert res == [[1, '1'], [2, '3'], [11, '7'], [13, '1']] or res == [[1, b'1'], [2, b'3'], [11, b'7'], [13, b'1']] def test_latest_flag_mrevrange(): env = Env(decodeResponses=True) key1 = 't1{1}' key2 = 't2{1}' key3 = 't3{1}' key4 = 't4{1}' with env.getClusterConnectionIfNeeded() as r: assert r.execute_command('TS.CREATE', key1) assert r.execute_command('TS.CREATE', key2, 'LABELS', 'is_compaction', 'true') assert r.execute_command('TS.CREATE', key3) assert r.execute_command('TS.CREATE', key4, 'LABELS', 'is_compaction', 'true') assert r.execute_command('TS.CREATERULE', key1, key2, 'AGGREGATION', 'SUM', 10) assert r.execute_command('TS.CREATERULE', key3, key4, 'AGGREGATION', 'SUM', 10) assert r.execute_command('TS.add', key1, 1, 1) assert r.execute_command('TS.add', key1, 2, 3) assert r.execute_command('TS.add', key1, 11, 7) assert r.execute_command('TS.add', key1, 13, 1) res = r.execute_command('TS.range', key1, 0, 20) assert r.execute_command('TS.add', key3, 1, 1) assert r.execute_command('TS.add', key3, 2, 3) assert r.execute_command('TS.add', key3, 11, 7) assert r.execute_command('TS.add', key3, 13, 1) res = env.getConnection(1).execute_command('TS.mrevrange', 0, 10, 'FILTER', 'is_compaction=true') assert res == [['t2{1}', [], [[0, '4']]], ['t4{1}', [], [[0, '4']]]] or res == [[b't2{1}', [], [[0, b'4']]], [b't4{1}', [], [[0, b'4']]]] res = env.getConnection(1).execute_command('TS.mrevrange', 0, 10, 'LATEST', 'FILTER', 'is_compaction=true') assert res == [['t2{1}', [], [[10, '8'], [0, '4']]], ['t4{1}', [], [[10, '8'], [0, '4']]]] or res == [[b't2{1}', [], [[10, b'8'], [0, b'4']]], [b't4{1}', [], [[10, b'8'], [0, b'4']]]] res = env.getConnection(1).execute_command('TS.mrevrange', 0, 10, 'FILTER', 'is_compaction=true', 'GROUPBY', 'is_compaction', 'REDUCE', 'sum') assert res == [['is_compaction=true', [], [[0, '8']]]] or res == [[b'is_compaction=true', [], [[0, b'8']]]] res = env.getConnection(1).execute_command('TS.mrevrange', 0, 10, 'LATEST', 'FILTER', 'is_compaction=true', 'GROUPBY', 'is_compaction', 'REDUCE', 'sum') assert res == [['is_compaction=true', [], [[10, '16'], [0, '8']]]] or res == [[b'is_compaction=true', [], [[10, b'16'], [0, b'8']]]] # make sure LATEST haven't changed anything in the keys res = r.execute_command('TS.range', key2, 0, 10) assert res == [[0, '4']] or res == [[0, b'4']] res = r.execute_command('TS.range', key1, 0, 20) assert res == [[1, '1'], [2, '3'], [11, '7'], [13, '1']] or res == [[1, b'1'], [2, b'3'], [11, b'7'], [13, b'1']] def test_mrange_nan_handling(): """Test MRANGE with NaN values across multiple series""" import math env = Env(decodeResponses=True) with env.getClusterConnectionIfNeeded() as r, env.getConnection(1) as r1: for encoding in ['compressed', 'uncompressed']: env.flush() # Create two series with same labels r.execute_command('TS.CREATE', 'ts1{a}', 'ENCODING', encoding, 'LABELS', 'sensor', 'temp') r.execute_command('TS.CREATE', 'ts2{a}', 'ENCODING', encoding, 'LABELS', 'sensor', 'temp') # ts1: bucket 0-99 mixed, bucket 100-199 NaN-only, bucket 200-299 valid for ts, val in [(10, 10), (20, 'nan'), (30, 20), (40, 'nan'), (50, 30)]: r.execute_command('TS.ADD', 'ts1{a}', ts, val) for ts in [110, 120, 130]: r.execute_command('TS.ADD', 'ts1{a}', ts, 'nan') for ts, val in [(210, 40), (220, 50)]: r.execute_command('TS.ADD', 'ts1{a}', ts, val) # ts2: bucket 0-99 valid, bucket 100-199 mixed, bucket 200-299 NaN-only for ts, val in [(10, 100), (50, 200)]: r.execute_command('TS.ADD', 'ts2{a}', ts, val) for ts, val in [(110, 'nan'), (150, 300), (180, 'nan')]: r.execute_command('TS.ADD', 'ts2{a}', ts, val) for ts in [210, 250]: r.execute_command('TS.ADD', 'ts2{a}', ts, 'nan') # Test 1: sum ignores NaN, count counts only valid samples res = r1.execute_command('TS.MRANGE', 0, 299, 'AGGREGATION', 'sum', 100, 'FILTER', 'sensor=temp') for series in res: if series[0] == 'ts1{a}': assert series[2] == [[0, '60'], [200, '90']] # bucket 100 skipped (NaN-only) elif series[0] == 'ts2{a}': assert series[2] == [[0, '300'], [100, '300']] # bucket 200 skipped (NaN-only) # Test 2: with EMPTY flag, NaN-only buckets appear res = r1.execute_command('TS.MRANGE', 0, 299, 'AGGREGATION', 'sum', 100, 'EMPTY', 'FILTER', 'sensor=temp') for series in res: if series[0] == 'ts1{a}': assert len(series[2]) == 3 assert series[2][1][0] == 100 and float(series[2][1][1]) == 0 # empty sum = 0 elif series[0] == 'ts2{a}': assert len(series[2]) == 3 assert series[2][2][0] == 200 and float(series[2][2][1]) == 0 # Test 3: avg returns NaN for NaN-only buckets with EMPTY res = r1.execute_command('TS.MRANGE', 0, 299, 'AGGREGATION', 'avg', 100, 'EMPTY', 'FILTER', 'sensor=temp') for series in res: if series[0] == 'ts1{a}': assert math.isnan(float(series[2][1][1])) # bucket 100 is NaN elif series[0] == 'ts2{a}': assert math.isnan(float(series[2][2][1])) # bucket 200 is NaN # Test 4: GROUPBY with REDUCE - NaN values should be skipped in reduction res = r1.execute_command('TS.MRANGE', 0, 299, 'AGGREGATION', 'sum', 100, 'FILTER', 'sensor=temp', 'GROUPBY', 'sensor', 'REDUCE', 'sum') assert len(res) == 1 assert res[0][0] == 'sensor=temp' # bucket 0: 60+300=360, bucket 100: 300 (ts1 has no valid), bucket 200: 90 (ts2 has no valid) sums = {int(s[0]): float(s[1]) for s in res[0][2]} assert sums[0] == 360 assert sums.get(100) == 300 assert sums.get(200) == 90 # Test 5: count aggregation res = r1.execute_command('TS.MRANGE', 0, 299, 'AGGREGATION', 'count', 100, 'FILTER', 'sensor=temp') for series in res: if series[0] == 'ts1{a}': assert series[2] == [[0, '3'], [200, '2']] # 3 valid in bucket 0, 2 in bucket 200 elif series[0] == 'ts2{a}': assert series[2] == [[0, '2'], [100, '1']] # 2 valid in bucket 0, 1 in bucket 100 # Test 6: min/max with mixed buckets res = r1.execute_command('TS.MRANGE', 0, 99, 'AGGREGATION', 'min', 100, 'FILTER', 'sensor=temp') for series in res: if series[0] == 'ts1{a}': assert float(series[2][0][1]) == 10 # min of valid values elif series[0] == 'ts2{a}': assert float(series[2][0][1]) == 100 res = r1.execute_command('TS.MRANGE', 0, 99, 'AGGREGATION', 'max', 100, 'FILTER', 'sensor=temp') for series in res: if series[0] == 'ts1{a}': assert float(series[2][0][1]) == 30 elif series[0] == 'ts2{a}': assert float(series[2][0][1]) == 200 # ── Shard pre-aggregation (Level 1) ────────────────────────────────────────── # These tests verify that MRANGE/MREVRANGE with AGGREGATION (no GROUPBY) # returns correct results when aggregation is performed on each shard before # the coordinator merges the per-series results. def _setup_shard_agg_data(r): """ Two series, each with 6 samples spread across three 10ms buckets: s1: t=0→10, t=5→20, t=10→30, t=15→40, t=20→50, t=25→60 s2: t=0→100, t=5→200, t=10→300, t=15→400, t=20→500, t=25→600 bucket 0–9: s1=[10,20] s2=[100,200] bucket 10–19: s1=[30,40] s2=[300,400] bucket 20–29: s1=[50,60] s2=[500,600] """ r.execute_command('TS.CREATE', 's1', 'LABELS', 'grp', 'A') r.execute_command('TS.CREATE', 's2', 'LABELS', 'grp', 'A') for ts, v in [(0, 10), (5, 20), (10, 30), (15, 40), (20, 50), (25, 60)]: r.execute_command('TS.ADD', 's1', ts, v) for ts, v in [(0, 100), (5, 200), (10, 300), (15, 400), (20, 500), (25, 600)]: r.execute_command('TS.ADD', 's2', ts, v) def _series_values(result, name): for series in result: if series[0] == name.encode() or series[0] == name: return [(int(ts), float(val)) for ts, val in series[2]] raise AssertionError(f"{name!r} not found in result") def test_shard_agg_avg(): env = Env() with env.getClusterConnectionIfNeeded() as r, env.getConnection(1) as r1: _setup_shard_agg_data(r) res = r1.execute_command('TS.MRANGE', 0, 29, 'AGGREGATION', 'avg', 10, 'FILTER', 'grp=A') s1 = _series_values(res, 's1') s2 = _series_values(res, 's2') assert s1 == [(0, 15.0), (10, 35.0), (20, 55.0)] assert s2 == [(0, 150.0), (10, 350.0), (20, 550.0)] def test_shard_agg_sum(): env = Env() with env.getClusterConnectionIfNeeded() as r, env.getConnection(1) as r1: _setup_shard_agg_data(r) res = r1.execute_command('TS.MRANGE', 0, 29, 'AGGREGATION', 'sum', 10, 'FILTER', 'grp=A') s1 = _series_values(res, 's1') s2 = _series_values(res, 's2') assert s1 == [(0, 30.0), (10, 70.0), (20, 110.0)] assert s2 == [(0, 300.0), (10, 700.0), (20, 1100.0)] def test_shard_agg_min(): env = Env() with env.getClusterConnectionIfNeeded() as r, env.getConnection(1) as r1: _setup_shard_agg_data(r) res = r1.execute_command('TS.MRANGE', 0, 29, 'AGGREGATION', 'min', 10, 'FILTER', 'grp=A') s1 = _series_values(res, 's1') s2 = _series_values(res, 's2') assert s1 == [(0, 10.0), (10, 30.0), (20, 50.0)] assert s2 == [(0, 100.0), (10, 300.0), (20, 500.0)] def test_shard_agg_max(): env = Env() with env.getClusterConnectionIfNeeded() as r, env.getConnection(1) as r1: _setup_shard_agg_data(r) res = r1.execute_command('TS.MRANGE', 0, 29, 'AGGREGATION', 'max', 10, 'FILTER', 'grp=A') s1 = _series_values(res, 's1') s2 = _series_values(res, 's2') assert s1 == [(0, 20.0), (10, 40.0), (20, 60.0)] assert s2 == [(0, 200.0), (10, 400.0), (20, 600.0)] def test_shard_agg_count(): env = Env() with env.getClusterConnectionIfNeeded() as r, env.getConnection(1) as r1: _setup_shard_agg_data(r) res = r1.execute_command('TS.MRANGE', 0, 29, 'AGGREGATION', 'count', 10, 'FILTER', 'grp=A') s1 = _series_values(res, 's1') s2 = _series_values(res, 's2') assert s1 == [(0, 2.0), (10, 2.0), (20, 2.0)] assert s2 == [(0, 2.0), (10, 2.0), (20, 2.0)] def test_shard_agg_first(): env = Env() with env.getClusterConnectionIfNeeded() as r, env.getConnection(1) as r1: _setup_shard_agg_data(r) res = r1.execute_command('TS.MRANGE', 0, 29, 'AGGREGATION', 'first', 10, 'FILTER', 'grp=A') s1 = _series_values(res, 's1') s2 = _series_values(res, 's2') assert s1 == [(0, 10.0), (10, 30.0), (20, 50.0)] assert s2 == [(0, 100.0), (10, 300.0), (20, 500.0)] def test_shard_agg_last(): env = Env() with env.getClusterConnectionIfNeeded() as r, env.getConnection(1) as r1: _setup_shard_agg_data(r) res = r1.execute_command('TS.MRANGE', 0, 29, 'AGGREGATION', 'last', 10, 'FILTER', 'grp=A') s1 = _series_values(res, 's1') s2 = _series_values(res, 's2') assert s1 == [(0, 20.0), (10, 40.0), (20, 60.0)] assert s2 == [(0, 200.0), (10, 400.0), (20, 600.0)] def test_shard_agg_range(): env = Env() with env.getClusterConnectionIfNeeded() as r, env.getConnection(1) as r1: _setup_shard_agg_data(r) res = r1.execute_command('TS.MRANGE', 0, 29, 'AGGREGATION', 'range', 10, 'FILTER', 'grp=A') s1 = _series_values(res, 's1') s2 = _series_values(res, 's2') # range = max - min per bucket: s1=[20-10,40-30,60-50]=[10,10,10] assert s1 == [(0, 10.0), (10, 10.0), (20, 10.0)] assert s2 == [(0, 100.0), (10, 100.0), (20, 100.0)] def test_shard_agg_std_p(): env = Env() with env.getClusterConnectionIfNeeded() as r, env.getConnection(1) as r1: _setup_shard_agg_data(r) res = r1.execute_command('TS.MRANGE', 0, 29, 'AGGREGATION', 'std.p', 10, 'FILTER', 'grp=A') s1 = _series_values(res, 's1') s2 = _series_values(res, 's2') for bucket_vals, bucket_result in [([10, 20], s1[0]), ([30, 40], s1[1]), ([50, 60], s1[2])]: assert math.isclose(bucket_result[1], statistics.pstdev(bucket_vals), rel_tol=1e-9) for bucket_vals, bucket_result in [([100, 200], s2[0]), ([300, 400], s2[1]), ([500, 600], s2[2])]: assert math.isclose(bucket_result[1], statistics.pstdev(bucket_vals), rel_tol=1e-9) def test_shard_agg_std_s(): env = Env() with env.getClusterConnectionIfNeeded() as r, env.getConnection(1) as r1: _setup_shard_agg_data(r) res = r1.execute_command('TS.MRANGE', 0, 29, 'AGGREGATION', 'std.s', 10, 'FILTER', 'grp=A') s1 = _series_values(res, 's1') s2 = _series_values(res, 's2') for bucket_vals, bucket_result in [([10, 20], s1[0]), ([30, 40], s1[1]), ([50, 60], s1[2])]: assert math.isclose(bucket_result[1], statistics.stdev(bucket_vals), rel_tol=1e-9) for bucket_vals, bucket_result in [([100, 200], s2[0]), ([300, 400], s2[1]), ([500, 600], s2[2])]: assert math.isclose(bucket_result[1], statistics.stdev(bucket_vals), rel_tol=1e-9) def test_shard_agg_var_p(): env = Env() with env.getClusterConnectionIfNeeded() as r, env.getConnection(1) as r1: _setup_shard_agg_data(r) res = r1.execute_command('TS.MRANGE', 0, 29, 'AGGREGATION', 'var.p', 10, 'FILTER', 'grp=A') s1 = _series_values(res, 's1') s2 = _series_values(res, 's2') for bucket_vals, bucket_result in [([10, 20], s1[0]), ([30, 40], s1[1]), ([50, 60], s1[2])]: assert math.isclose(bucket_result[1], statistics.pvariance(bucket_vals), rel_tol=1e-9) def test_shard_agg_var_s(): env = Env() with env.getClusterConnectionIfNeeded() as r, env.getConnection(1) as r1: _setup_shard_agg_data(r) res = r1.execute_command('TS.MRANGE', 0, 29, 'AGGREGATION', 'var.s', 10, 'FILTER', 'grp=A') s1 = _series_values(res, 's1') s2 = _series_values(res, 's2') for bucket_vals, bucket_result in [([10, 20], s1[0]), ([30, 40], s1[1]), ([50, 60], s1[2])]: assert math.isclose(bucket_result[1], statistics.variance(bucket_vals), rel_tol=1e-9) def test_shard_agg_mrevrange(): """MREVRANGE with aggregation should return buckets in descending order.""" env = Env() with env.getClusterConnectionIfNeeded() as r, env.getConnection(1) as r1: _setup_shard_agg_data(r) res = r1.execute_command('TS.MREVRANGE', 0, 29, 'AGGREGATION', 'avg', 10, 'FILTER', 'grp=A') s1 = _series_values(res, 's1') assert s1 == [(20, 55.0), (10, 35.0), (0, 15.0)] def test_shard_agg_count_limit(): """COUNT cap applied after shard aggregation.""" env = Env() with env.getClusterConnectionIfNeeded() as r, env.getConnection(1) as r1: _setup_shard_agg_data(r) res = r1.execute_command('TS.MRANGE', 0, 29, 'AGGREGATION', 'sum', 10, 'COUNT', 2, 'FILTER', 'grp=A') s1 = _series_values(res, 's1') assert len(s1) == 2 assert s1[0] == (0, 30.0) assert s1[1] == (10, 70.0) def test_shard_agg_time_range(): """Partial time range only returns buckets that overlap the range.""" env = Env() with env.getClusterConnectionIfNeeded() as r, env.getConnection(1) as r1: _setup_shard_agg_data(r) # Only request bucket 10–19 res = r1.execute_command('TS.MRANGE', 10, 19, 'AGGREGATION', 'avg', 10, 'FILTER', 'grp=A') s1 = _series_values(res, 's1') s2 = _series_values(res, 's2') assert s1 == [(10, 35.0)] assert s2 == [(10, 350.0)] def test_shard_agg_single_sample_per_bucket(): """When each bucket has exactly one sample, all agg types return that sample.""" env = Env() with env.getClusterConnectionIfNeeded() as r, env.getConnection(1) as r1: r.execute_command('TS.CREATE', 'solo', 'LABELS', 'type', 'single') for ts, v in [(0, 42), (10, 99), (20, 7)]: r.execute_command('TS.ADD', 'solo', ts, v) for agg in ['avg', 'sum', 'min', 'max', 'first', 'last']: res = r1.execute_command('TS.MRANGE', 0, 29, 'AGGREGATION', agg, 10, 'FILTER', 'type=single') vals = _series_values(res, 'solo') assert vals == [(0, 42.0), (10, 99.0), (20, 7.0)], f"failed for agg={agg}" def test_shard_agg_multiple_series_independent(): """Each series is aggregated independently — no cross-series interference.""" env = Env() with env.getClusterConnectionIfNeeded() as r, env.getConnection(1) as r1: # s_high has large values, s_low has small values — min of s_high should never equal s_low r.execute_command('TS.CREATE', 's_high', 'LABELS', 'band', 'test') r.execute_command('TS.CREATE', 's_low', 'LABELS', 'band', 'test') for ts in [0, 5]: r.execute_command('TS.ADD', 's_high', ts, 1000 + ts) r.execute_command('TS.ADD', 's_low', ts, ts + 1) res = r1.execute_command('TS.MRANGE', 0, 9, 'AGGREGATION', 'min', 10, 'FILTER', 'band=test') high = _series_values(res, 's_high') low = _series_values(res, 's_low') assert high == [(0, 1000.0)] assert low == [(0, 1.0)] # ── GROUPBY + shard pre-aggregation (Option A) ─────────────────────────────── # Option A: avg-of-per-series-avgs (not a true weighted average). # To distinguish Option A from Option B we use unequal sample counts per series # within the same bucket: # s1: [10] → avg=10, count=1 # s2: [100, 200] → avg=150, count=2 # # Option A: avg(10, 150) = 80 # Option B: (10+100+200)/3 ≈ 103.33 # # The correct result for our implementation is 80. def _setup_groupby_unequal_data(r): """s1 and s2 share label grp=X, bucket 0–9, unequal sample counts.""" r.execute_command('TS.CREATE', 'g1', 'LABELS', 'grp', 'X') r.execute_command('TS.CREATE', 'g2', 'LABELS', 'grp', 'X') r.execute_command('TS.ADD', 'g1', 0, 10) r.execute_command('TS.ADD', 'g2', 0, 100) r.execute_command('TS.ADD', 'g2', 5, 200) def test_groupby_agg_option_a_avg(): """GROUPBY REDUCE avg with unequal bucket sizes confirms Option A (avg-of-avgs).""" env = Env() with env.getClusterConnectionIfNeeded() as r, env.getConnection(1) as r1: _setup_groupby_unequal_data(r) res = r1.execute_command( 'TS.MRANGE', 0, 9, 'AGGREGATION', 'avg', 10, 'WITHLABELS', 'FILTER', 'grp=X', 'GROUPBY', 'grp', 'REDUCE', 'avg') assert len(res) == 1 vals = [(int(ts), float(v)) for ts, v in res[0][2]] # Option A: avg(10, 150) = 80 (NOT 103.33 which would be Option B) assert vals == [(0, 80.0)] def test_groupby_agg_sum(): """GROUPBY REDUCE sum: sum of per-series sums per bucket.""" env = Env() with env.getClusterConnectionIfNeeded() as r, env.getConnection(1) as r1: _setup_groupby_unequal_data(r) res = r1.execute_command( 'TS.MRANGE', 0, 9, 'AGGREGATION', 'sum', 10, 'WITHLABELS', 'FILTER', 'grp=X', 'GROUPBY', 'grp', 'REDUCE', 'sum') vals = [(int(ts), float(v)) for ts, v in res[0][2]] # g1 sum=10, g2 sum=300 → REDUCE sum = 310 assert vals == [(0, 310.0)] def test_groupby_agg_max(): """GROUPBY REDUCE max: max of per-series maxes per bucket.""" env = Env() with env.getClusterConnectionIfNeeded() as r, env.getConnection(1) as r1: _setup_groupby_unequal_data(r) res = r1.execute_command( 'TS.MRANGE', 0, 9, 'AGGREGATION', 'max', 10, 'WITHLABELS', 'FILTER', 'grp=X', 'GROUPBY', 'grp', 'REDUCE', 'max') vals = [(int(ts), float(v)) for ts, v in res[0][2]] # g1 max=10, g2 max=200 → REDUCE max = 200 assert vals == [(0, 200.0)] def test_groupby_agg_min(): """GROUPBY REDUCE min: min of per-series mins per bucket.""" env = Env() with env.getClusterConnectionIfNeeded() as r, env.getConnection(1) as r1: _setup_groupby_unequal_data(r) res = r1.execute_command( 'TS.MRANGE', 0, 9, 'AGGREGATION', 'min', 10, 'WITHLABELS', 'FILTER', 'grp=X', 'GROUPBY', 'grp', 'REDUCE', 'min') vals = [(int(ts), float(v)) for ts, v in res[0][2]] # g1 min=10, g2 min=100 → REDUCE min = 10 assert vals == [(0, 10.0)] def test_groupby_agg_multiple_buckets(): """GROUPBY REDUCE avg across multiple buckets to verify per-bucket correctness.""" env = Env() with env.getClusterConnectionIfNeeded() as r, env.getConnection(1) as r1: r.execute_command('TS.CREATE', 'mb1', 'LABELS', 'grp', 'Y') r.execute_command('TS.CREATE', 'mb2', 'LABELS', 'grp', 'Y') # bucket 0: mb1=[10,20] avg=15, mb2=[30,40] avg=35 # bucket 10: mb1=[50] avg=50, mb2=[60,70] avg=65 for ts, v in [(0, 10), (5, 20), (10, 50)]: r.execute_command('TS.ADD', 'mb1', ts, v) for ts, v in [(0, 30), (5, 40), (10, 60), (15, 70)]: r.execute_command('TS.ADD', 'mb2', ts, v) res = r1.execute_command( 'TS.MRANGE', 0, 19, 'AGGREGATION', 'avg', 10, 'WITHLABELS', 'FILTER', 'grp=Y', 'GROUPBY', 'grp', 'REDUCE', 'avg') vals = {int(ts): float(v) for ts, v in res[0][2]} assert vals[0] == 25.0 # avg(15, 35) = 25 assert vals[10] == 57.5 # avg(50, 65) = 57.5 # ── Multi-aggregator (no GROUPBY) ──────────────────────────────────────────── # AGGREGATION avg,sum 10 → each sample is [ts, avg_val, sum_val] def _series_multi_values(result, name): """Return list of (ts, [v1, v2, ...]) tuples from a multi-agg result.""" for series in result: if series[0] == name.encode() or series[0] == name: return [(int(sample[0]), [float(v) for v in sample[1:]]) for sample in series[2]] raise AssertionError(f"{name!r} not found in result") def test_multi_agg_avg_sum(): """AGGREGATION avg,sum 10 returns [ts, avg, sum] per sample.""" env = Env() with env.getClusterConnectionIfNeeded() as r, env.getConnection(1) as r1: _setup_shard_agg_data(r) res = r1.execute_command('TS.MRANGE', 0, 29, 'AGGREGATION', 'avg,sum', 10, 'FILTER', 'grp=A') s1 = _series_multi_values(res, 's1') # bucket 0: avg=15, sum=30; bucket 10: avg=35, sum=70; bucket 20: avg=55, sum=110 assert s1 == [(0, [15.0, 30.0]), (10, [35.0, 70.0]), (20, [55.0, 110.0])] def test_multi_agg_min_max_count(): """AGGREGATION min,max,count 10 returns [ts, min, max, count] per sample.""" env = Env() with env.getClusterConnectionIfNeeded() as r, env.getConnection(1) as r1: _setup_shard_agg_data(r) res = r1.execute_command('TS.MRANGE', 0, 9, 'AGGREGATION', 'min,max,count', 10, 'FILTER', 'grp=A') s1 = _series_multi_values(res, 's1') # bucket 0: min=10, max=20, count=2 assert s1 == [(0, [10.0, 20.0, 2.0])] def test_multi_agg_mrevrange(): """Multi-agg with MREVRANGE returns buckets in descending order.""" env = Env() with env.getClusterConnectionIfNeeded() as r, env.getConnection(1) as r1: _setup_shard_agg_data(r) res = r1.execute_command('TS.MREVRANGE', 0, 29, 'AGGREGATION', 'avg,sum', 10, 'FILTER', 'grp=A') s1 = _series_multi_values(res, 's1') assert s1 == [(20, [55.0, 110.0]), (10, [35.0, 70.0]), (0, [15.0, 30.0])] def test_multi_agg_groupby_blocked(): """Multiple aggregators with GROUPBY should return an error.""" env = Env() with env.getClusterConnectionIfNeeded() as r, env.getConnection(1) as r1: _setup_shard_agg_data(r) with pytest.raises(redis.ResponseError): r1.execute_command( 'TS.MRANGE', 0, 29, 'AGGREGATION', 'avg,sum', 10, 'FILTER', 'grp=A', 'GROUPBY', 'grp', 'REDUCE', 'avg') # ── EMPTY gap-filling edge cases via MRANGE ────────────────────────────────── # Setup: two samples at t=10→100 and t=20→110. # # Three edge cases for EMPTY gap-filling: # "between" — query range entirely between the two samples (t=11..16) # "before_first"— range starts before first sample (t=8..12) # "after_last" — range starts between samples, ends past last (t=18..22) # # Each test creates the key-under-test plus two filler series that share the # same label, so the MRANGE fan-out touches multiple series. # Assertion: MRANGE result for the key-under-test == TS.RANGE result. _EMPTY_FILL_LABEL = 'scenario=empty_fill' _EMPTY_FILL_AGGS = ['last', 'avg', 'first', 'sum', 'min', 'max', 'count'] def _empty_fill_create(r, key): r.execute_command('TS.CREATE', key, 'LABELS', 'scenario', 'empty_fill') r.execute_command('TS.ADD', key, 10, 100) r.execute_command('TS.ADD', key, 20, 110) def _empty_fill_filler(r, key): """Series with same label; data outside all three case ranges.""" r.execute_command('TS.CREATE', key, 'LABELS', 'scenario', 'empty_fill') r.execute_command('TS.ADD', key, 50, 42) r.execute_command('TS.ADD', key, 60, 43) def _mrange_key_samples(mrange_result, key): for series in mrange_result: if series[0] in (key, key.encode()): return series[2] raise AssertionError(f'{key!r} not found in MRANGE result') def test_mrange_empty_fill_range_between_samples(): """MRANGE with EMPTY: query range lies entirely between two samples (t=11..16).""" env = Env() with env.getClusterConnectionIfNeeded() as r, env.getConnection(1) as r1: _empty_fill_create(r, 'ef_between') _empty_fill_filler(r, 'ef_between_f1') _empty_fill_filler(r, 'ef_between_f2') for agg in _EMPTY_FILL_AGGS: range_res = r.execute_command('TS.RANGE', 'ef_between', 11, 16, 'AGGREGATION', agg, 1, 'EMPTY') mrange_res = r1.execute_command('TS.MRANGE', 11, 16, 'AGGREGATION', agg, 1, 'EMPTY', 'FILTER', _EMPTY_FILL_LABEL) assert _mrange_key_samples(mrange_res, 'ef_between') == range_res, \ f'between-samples mismatch for agg={agg}' def test_mrange_empty_fill_range_starts_before_first_sample(): """MRANGE with EMPTY: query starts before the first sample (t=8..12).""" env = Env() with env.getClusterConnectionIfNeeded() as r, env.getConnection(1) as r1: _empty_fill_create(r, 'ef_before') _empty_fill_filler(r, 'ef_before_f1') _empty_fill_filler(r, 'ef_before_f2') for agg in _EMPTY_FILL_AGGS: range_res = r.execute_command('TS.RANGE', 'ef_before', 8, 12, 'AGGREGATION', agg, 1, 'EMPTY') mrange_res = r1.execute_command('TS.MRANGE', 8, 12, 'AGGREGATION', agg, 1, 'EMPTY', 'FILTER', _EMPTY_FILL_LABEL) assert _mrange_key_samples(mrange_res, 'ef_before') == range_res, \ f'before-first-sample mismatch for agg={agg}' def test_mrange_empty_fill_range_ends_after_last_sample(): """MRANGE with EMPTY: query starts between samples and ends past the last one (t=18..22).""" env = Env() with env.getClusterConnectionIfNeeded() as r, env.getConnection(1) as r1: _empty_fill_create(r, 'ef_after') _empty_fill_filler(r, 'ef_after_f1') _empty_fill_filler(r, 'ef_after_f2') for agg in _EMPTY_FILL_AGGS: range_res = r.execute_command('TS.RANGE', 'ef_after', 18, 22, 'AGGREGATION', agg, 1, 'EMPTY') mrange_res = r1.execute_command('TS.MRANGE', 18, 22, 'AGGREGATION', agg, 1, 'EMPTY', 'FILTER', _EMPTY_FILL_LABEL) assert _mrange_key_samples(mrange_res, 'ef_after') == range_res, \ f'after-last-sample mismatch for agg={agg}' # ───────────────────────────────────────────────────────────────────────────── # EXCLUDEEMPTY tests # ───────────────────────────────────────────────────────────────────────────── def _excl_keys(res): """Return sorted list of key names from RESP2 or RESP3 MRANGE result.""" if isinstance(res, dict): return sorted(k.decode() if isinstance(k, bytes) else k for k in res.keys()) return sorted(s[0].decode() if isinstance(s[0], bytes) else s[0] for s in res) def _excl_samples(res, key): """Return sample list for key from RESP2 or RESP3 MRANGE result.""" key_b = key.encode() if isinstance(key, str) else key if isinstance(res, dict): entry = res.get(key_b) or res.get(key) return entry[-1] # last element is always samples for s in res: if s[0] == key_b or s[0] == key: return s[2] raise AssertionError(repr(key) + ' not found in result') def test_excludeempty_basic(env): """Series with no samples in range are excluded; series with data are returned.""" with env.getClusterConnectionIfNeeded() as r, env.getConnection(1) as r1: r.execute_command('TS.CREATE', 'xeb1', 'LABELS', 'g', 'xeb') r.execute_command('TS.CREATE', 'xeb2', 'LABELS', 'g', 'xeb') r.execute_command('TS.CREATE', 'xeb3', 'LABELS', 'g', 'xeb') r.execute_command('TS.ADD', 'xeb1', 10, 1.0) r.execute_command('TS.ADD', 'xeb1', 20, 2.0) r.execute_command('TS.ADD', 'xeb2', 1000, 99.0) # outside range r.execute_command('TS.ADD', 'xeb3', 15, 3.0) res = r1.execute_command('TS.MRANGE', 1, 100, 'EXCLUDEEMPTY', 'FILTER', 'g=xeb') assert _excl_keys(res) == ['xeb1', 'xeb3'] res_all = r1.execute_command('TS.MRANGE', 1, 100, 'FILTER', 'g=xeb') assert len(res_all) == 3 def test_excludeempty_all_empty(env): """When every series is empty in range the reply is an empty array.""" with env.getClusterConnectionIfNeeded() as r, env.getConnection(1) as r1: r.execute_command('TS.CREATE', 'xee1', 'LABELS', 'g', 'xee') r.execute_command('TS.CREATE', 'xee2', 'LABELS', 'g', 'xee') r.execute_command('TS.ADD', 'xee1', 9999, 1.0) r.execute_command('TS.ADD', 'xee2', 9999, 2.0) res = r1.execute_command('TS.MRANGE', 1, 100, 'EXCLUDEEMPTY', 'FILTER', 'g=xee') assert res == [] or res == {} def test_excludeempty_all_have_data(env): """When every series has data the result matches a plain MRANGE.""" with env.getClusterConnectionIfNeeded() as r, env.getConnection(1) as r1: r.execute_command('TS.CREATE', 'xea1', 'LABELS', 'g', 'xea') r.execute_command('TS.CREATE', 'xea2', 'LABELS', 'g', 'xea') r.execute_command('TS.ADD', 'xea1', 10, 1.0) r.execute_command('TS.ADD', 'xea2', 20, 2.0) res_excl = r1.execute_command('TS.MRANGE', 1, 100, 'EXCLUDEEMPTY', 'FILTER', 'g=xea') res_plain = r1.execute_command('TS.MRANGE', 1, 100, 'FILTER', 'g=xea') assert _excl_keys(res_excl) == _excl_keys(res_plain) def test_excludeempty_single_series_empty(env): """Only series: empty → reply is [].""" with env.getClusterConnectionIfNeeded() as r, env.getConnection(1) as r1: r.execute_command('TS.CREATE', 'xes1', 'LABELS', 'g', 'xes1') r.execute_command('TS.ADD', 'xes1', 500, 7.0) res = r1.execute_command('TS.MRANGE', 1, 10, 'EXCLUDEEMPTY', 'FILTER', 'g=xes1') assert res == [] or res == {} def test_excludeempty_single_series_nonempty(env): """Only series: has data → returned normally.""" with env.getClusterConnectionIfNeeded() as r, env.getConnection(1) as r1: r.execute_command('TS.CREATE', 'xes2', 'LABELS', 'g', 'xes2') r.execute_command('TS.ADD', 'xes2', 5, 7.0) res = r1.execute_command('TS.MRANGE', 1, 10, 'EXCLUDEEMPTY', 'FILTER', 'g=xes2') assert _excl_keys(res) == ['xes2'] def test_excludeempty_withlabels(env): """EXCLUDEEMPTY + WITHLABELS: non-empty series include their labels.""" with env.getClusterConnectionIfNeeded() as r, env.getConnection(1) as r1: r.execute_command('TS.CREATE', 'xewl1', 'LABELS', 'g', 'xewl', 'name', 'alpha') r.execute_command('TS.CREATE', 'xewl2', 'LABELS', 'g', 'xewl', 'name', 'beta') r.execute_command('TS.ADD', 'xewl1', 5, 1.0) r.execute_command('TS.ADD', 'xewl2', 5000, 2.0) # outside range res = r1.execute_command('TS.MRANGE', 1, 100, 'WITHLABELS', 'EXCLUDEEMPTY', 'FILTER', 'g=xewl') assert _excl_keys(res) == ['xewl1'] labels = dict(next(s[1] for s in res if s[0] == b'xewl1')) assert labels[b'name'] == b'alpha' def test_excludeempty_with_count(env): """EXCLUDEEMPTY + COUNT: count applies to samples within the kept series.""" with env.getClusterConnectionIfNeeded() as r, env.getConnection(1) as r1: r.execute_command('TS.CREATE', 'xec1', 'LABELS', 'g', 'xec') r.execute_command('TS.CREATE', 'xec2', 'LABELS', 'g', 'xec') for i in range(1, 11): r.execute_command('TS.ADD', 'xec1', i, float(i)) r.execute_command('TS.ADD', 'xec2', 9999, 1.0) res = r1.execute_command('TS.MRANGE', 1, 100, 'COUNT', 3, 'EXCLUDEEMPTY', 'FILTER', 'g=xec') assert _excl_keys(res) == ['xec1'] assert len(_excl_samples(res, 'xec1')) == 3 def test_excludeempty_with_aggregation(env): """EXCLUDEEMPTY + AGGREGATION: series with no raw samples in range are excluded.""" with env.getClusterConnectionIfNeeded() as r, env.getConnection(1) as r1: r.execute_command('TS.CREATE', 'xeag1', 'LABELS', 'g', 'xeag') r.execute_command('TS.CREATE', 'xeag2', 'LABELS', 'g', 'xeag') for i in range(0, 100, 10): r.execute_command('TS.ADD', 'xeag1', i, float(i)) r.execute_command('TS.ADD', 'xeag2', 9999, 1.0) res = r1.execute_command( 'TS.MRANGE', 0, 99, 'AGGREGATION', 'avg', 50, 'EXCLUDEEMPTY', 'FILTER', 'g=xeag') assert _excl_keys(res) == ['xeag1'] def test_excludeempty_with_filter_by_value(env): """EXCLUDEEMPTY + FILTER_BY_VALUE: series whose values all fall outside the filter are excluded.""" with env.getClusterConnectionIfNeeded() as r, env.getConnection(1) as r1: r.execute_command('TS.CREATE', 'xefv1', 'LABELS', 'g', 'xefv') r.execute_command('TS.CREATE', 'xefv2', 'LABELS', 'g', 'xefv') r.execute_command('TS.ADD', 'xefv1', 10, 5.0) r.execute_command('TS.ADD', 'xefv1', 20, 15.0) r.execute_command('TS.ADD', 'xefv2', 10, 100.0) r.execute_command('TS.ADD', 'xefv2', 20, 200.0) res = r1.execute_command( 'TS.MRANGE', 1, 100, 'FILTER_BY_VALUE', 0, 20, 'EXCLUDEEMPTY', 'FILTER', 'g=xefv') assert _excl_keys(res) == ['xefv1'] def test_excludeempty_with_filter_by_ts(env): """EXCLUDEEMPTY + FILTER_BY_TS: series with no matching timestamps are excluded.""" with env.getClusterConnectionIfNeeded() as r, env.getConnection(1) as r1: r.execute_command('TS.CREATE', 'xeft1', 'LABELS', 'g', 'xeft') r.execute_command('TS.CREATE', 'xeft2', 'LABELS', 'g', 'xeft') r.execute_command('TS.ADD', 'xeft1', 10, 1.0) r.execute_command('TS.ADD', 'xeft1', 20, 2.0) r.execute_command('TS.ADD', 'xeft2', 30, 3.0) # not in FILTER_BY_TS list res = r1.execute_command( 'TS.MRANGE', 1, 100, 'FILTER_BY_TS', 10, 20, 'EXCLUDEEMPTY', 'FILTER', 'g=xeft') assert _excl_keys(res) == ['xeft1'] def test_excludeempty_mrevrange(env): """EXCLUDEEMPTY works with TS.MREVRANGE and returns samples in reverse order.""" with env.getClusterConnectionIfNeeded() as r, env.getConnection(1) as r1: r.execute_command('TS.CREATE', 'xer1', 'LABELS', 'g', 'xer') r.execute_command('TS.CREATE', 'xer2', 'LABELS', 'g', 'xer') r.execute_command('TS.ADD', 'xer1', 10, 1.0) r.execute_command('TS.ADD', 'xer1', 20, 2.0) r.execute_command('TS.ADD', 'xer1', 30, 3.0) r.execute_command('TS.ADD', 'xer2', 9999, 1.0) res = r1.execute_command('TS.MREVRANGE', 1, 100, 'EXCLUDEEMPTY', 'FILTER', 'g=xer') assert _excl_keys(res) == ['xer1'] timestamps = [s[0] for s in _excl_samples(res, 'xer1')] assert timestamps == sorted(timestamps, reverse=True) def test_excludeempty_boundary_timestamps(env): """Series with a sample exactly at the range boundary is included.""" with env.getClusterConnectionIfNeeded() as r, env.getConnection(1) as r1: r.execute_command('TS.CREATE', 'xebt1', 'LABELS', 'g', 'xebt') r.execute_command('TS.CREATE', 'xebt2', 'LABELS', 'g', 'xebt') r.execute_command('TS.ADD', 'xebt1', 1, 1.0) # exactly at start r.execute_command('TS.ADD', 'xebt2', 100, 2.0) # exactly at end res = r1.execute_command('TS.MRANGE', 1, 100, 'EXCLUDEEMPTY', 'FILTER', 'g=xebt') assert _excl_keys(res) == ['xebt1', 'xebt2'] res_out = r1.execute_command('TS.MRANGE', 2, 99, 'EXCLUDEEMPTY', 'FILTER', 'g=xebt') assert res_out == [] or res_out == {} def test_excludeempty_error_with_groupby(env): """EXCLUDEEMPTY combined with GROUPBY must return an error.""" with env.getClusterConnectionIfNeeded() as r, env.getConnection(1) as r1: r.execute_command('TS.CREATE', 'xegb1', 'LABELS', 'g', 'xegb') with pytest.raises(redis.ResponseError) as exc: r1.execute_command( 'TS.MRANGE', 1, 100, 'EXCLUDEEMPTY', 'FILTER', 'g=xegb', 'GROUPBY', 'g', 'REDUCE', 'sum') assert 'EXCLUDEEMPTY' in str(exc.value) def test_excludeempty_mixed_large(env): """10 series: alternating empty/non-empty — exactly the non-empty ones come back.""" with env.getClusterConnectionIfNeeded() as r, env.getConnection(1) as r1: expected = [] for i in range(10): key = f'xeml{i}' r.execute_command('TS.CREATE', key, 'LABELS', 'g', 'xeml') if i % 2 == 0: r.execute_command('TS.ADD', key, 50, float(i)) expected.append(key) else: r.execute_command('TS.ADD', key, 9999, float(i)) res = r1.execute_command('TS.MRANGE', 1, 100, 'EXCLUDEEMPTY', 'FILTER', 'g=xeml') assert _excl_keys(res) == sorted(expected) def _run_excludeempty_agg_type(env, agg_type): label_val = f'xeat_{agg_type}' with env.getClusterConnectionIfNeeded() as r, env.getConnection(1) as r1: r.execute_command('TS.CREATE', f'xeat1_{agg_type}', 'LABELS', 'g', label_val) r.execute_command('TS.CREATE', f'xeat2_{agg_type}', 'LABELS', 'g', label_val) r.execute_command('TS.ADD', f'xeat1_{agg_type}', 10, 3.0) r.execute_command('TS.ADD', f'xeat1_{agg_type}', 20, 5.0) r.execute_command('TS.ADD', f'xeat2_{agg_type}', 9999, 1.0) res = r1.execute_command( 'TS.MRANGE', 1, 100, 'AGGREGATION', agg_type, 100, 'EXCLUDEEMPTY', 'FILTER', f'g={label_val}') assert _excl_keys(res) == [f'xeat1_{agg_type}'], f'agg={agg_type}' assert len(_excl_samples(res, f'xeat1_{agg_type}')) == 1, f'agg={agg_type}' def test_excludeempty_agg_type_avg(env): _run_excludeempty_agg_type(env, 'avg') def test_excludeempty_agg_type_sum(env): _run_excludeempty_agg_type(env, 'sum') def test_excludeempty_agg_type_min(env): _run_excludeempty_agg_type(env, 'min') def test_excludeempty_agg_type_max(env): _run_excludeempty_agg_type(env, 'max') def test_excludeempty_agg_type_count(env): _run_excludeempty_agg_type(env, 'count') def test_excludeempty_agg_type_first(env): _run_excludeempty_agg_type(env, 'first') def test_excludeempty_agg_type_last(env): _run_excludeempty_agg_type(env, 'last') def test_excludeempty_agg_empty_flag_both_sides(env): """AGGREGATION EMPTY + data on BOTH sides of range: NaN buckets are produced → reply is non-empty → series is KEPT by EXCLUDEEMPTY (reply-level semantics).""" with env.getClusterConnectionIfNeeded() as r, env.getConnection(1) as r1: r.execute_command('TS.CREATE', 'xebs', 'LABELS', 'g', 'xebs') r.execute_command('TS.ADD', 'xebs', 10, 1.0) # before range r.execute_command('TS.ADD', 'xebs', 9999, 2.0) # after range # No raw samples in [1000, 5000], but EMPTY fills NaN buckets because # there's data on both sides → reply is non-empty → series must be kept. res = r1.execute_command( 'TS.MRANGE', 1000, 5000, 'AGGREGATION', 'avg', 1000, 'EMPTY', 'EXCLUDEEMPTY', 'FILTER', 'g=xebs') assert _excl_keys(res) == ['xebs'] samples = _excl_samples(res, 'xebs') assert len(samples) > 0 assert all(math.isnan(float(v)) for _, v in samples) def test_excludeempty_agg_empty_flag_one_side(env): """AGGREGATION EMPTY + data on ONE side only: edge buckets dropped → no NaN buckets in range → series is EXCLUDED by EXCLUDEEMPTY.""" with env.getClusterConnectionIfNeeded() as r, env.getConnection(1) as r1: r.execute_command('TS.CREATE', 'xeos', 'LABELS', 'g', 'xeos') r.execute_command('TS.ADD', 'xeos', 9999, 2.0) # after range only res = r1.execute_command( 'TS.MRANGE', 1000, 5000, 'AGGREGATION', 'avg', 1000, 'EMPTY', 'EXCLUDEEMPTY', 'FILTER', 'g=xeos') assert res == [] or res == {} def test_excludeempty_resp3(env): """EXCLUDEEMPTY returns correct keys under RESP3 (map reply).""" from utils import is_resp3_possible if not is_resp3_possible(env): env.skip() env3 = Env(protocol=3) with env3.getClusterConnectionIfNeeded() as r, env3.getConnection(1) as r1: r.execute_command('TS.CREATE', 'xr3a', 'LABELS', 'g', 'xr3') r.execute_command('TS.CREATE', 'xr3b', 'LABELS', 'g', 'xr3') r.execute_command('TS.ADD', 'xr3a', 10, 1.0) r.execute_command('TS.ADD', 'xr3b', 9999, 2.0) # outside range res = r1.execute_command('TS.MRANGE', 1, 100, 'EXCLUDEEMPTY', 'FILTER', 'g=xr3') assert isinstance(res, dict) assert _excl_keys(res) == ['xr3a']