from common import *

DEFAULT_LIMIT = 10

def _setup_index_and_data(env, docs):
    env.expect('FT.CREATE', 'idx', 'ON', 'HASH',
                        'SCHEMA', 'title', 'TEXT', 'SORTABLE',
                        'brand', 'TEXT', 'NOSTEM', 'SORTABLE',
                        'description', 'TEXT', 'price', 'NUMERIC',
                        'categories', 'TAG').ok()
    conn = env.getClusterConnectionIfNeeded()

    for i in range(docs):
        title = f'Game {i}'
        brand = f'Brand {i % 25}'
        description = f'Description for game {i}'
        price = i
        category = f'Category {i % 5}'
        conn.execute_command(
            'HSET', f'doc_{i}', 'title', title, 'brand', brand,
            'description', description, 'price', price, 'categories', category)


def _get_total_results(res) -> int:
    # Extract the total_results from the query response
    if isinstance(res, dict):
        return res['total_results']
    else:
        return res[0]


def _get_results(res):
    # Extract the results from the query response
    if isinstance(res, dict):
        return res['results']
    else:
        return res[1:]


def _get_cluster_RP_profile(env, res) -> list:
    # Extract the RP types from the profile response
    shard_RP_and_count = []
    if isinstance(res, dict):
        for i in range(len(res['Profile']['Shards'])):
            shard = res['Profile']['Shards'][i]['Result processors profile']
            shard_RP_and_count.append([(item['Type'], item['Results processed']) for item in shard])

        # sort shard by the number of results processed by the first RP
        shard_RP_and_count.sort(key=lambda x: x[0][1])
        # Extract the RP types from the coordinator
        coord = res['Profile']['Coordinator']['Result processors profile']
        coord_RP_and_count = [(item['Type'], item['Results processed']) for item in coord]
        return [shard_RP_and_count, coord_RP_and_count]

    else:
        for i in range(len(res[1][1])):
            shard = res[1][1][i][19]
            shard_RP_and_count.append([(item[1], item[5]) for item in shard])

        # sort shard by the number of results processed by the first RP
        shard_RP_and_count.sort(key=lambda x: x[0][1])
        # Extract the RP types from the coordinator
        coord = res[1][3][13]
        coord_RP_and_count = [(item[1], item[5]) for item in coord]
        return [shard_RP_and_count, coord_RP_and_count]


def _get_standalone_RP_profile(env, res) -> list:
    if isinstance(res, dict):
        profile = res['Profile']['Shards'][0]['Result processors profile']
        RP_and_count = [(item['Type'], item['Results processed']) for item in profile]
        return RP_and_count
    else:
        profile = res[1][1][0][13]
        RP_and_count = [(item[1], item[5]) for item in profile]
        return RP_and_count


def _translate_query_to_profile_query(query) -> list:
    profile = ['FT.PROFILE']
    profile.append(query[1])        # index name
    profile.append(query[0][3:])    # command
    profile.append('QUERY')
    profile.extend(query[2:])       # query
    return profile


def _test_limit00(protocol):
    env = Env(protocol=protocol)
    docs = 2265
    _setup_index_and_data(env, docs)

    for on_timeout_policy in ['return', 'fail']:
        config_cmd = ['CONFIG', 'SET', 'search-on-timeout', on_timeout_policy]
        verify_command_OK_on_all_shards(env, *config_cmd)
        queries_and_results = [
            (['FT.AGGREGATE', 'idx', '*', 'WITHCOUNT', 'LIMIT', 0, 0], docs),
            (['FT.AGGREGATE', 'idx', '*', 'WITHOUTCOUNT', 'LIMIT', 0, 0], ANY),
            # WITHOUTCOUNT is implied by default
            (['FT.AGGREGATE', 'idx', '*', 'LIMIT', 0, 0], ANY),
        ]

        for query, expected_results in queries_and_results:
            cmd=' '.join(str(x) for x in query)
            for dialect in [1, 2, 3, 4]:
                config_cmd = ['CONFIG', 'SET', 'search-default-dialect', dialect]
                verify_command_OK_on_all_shards(env, *config_cmd)
                res = env.cmd(*query)
                total_results = _get_total_results(res)
                results = _get_results(res)

                # Verify results
                env.assertEqual(
                    total_results, expected_results,
                    message=f'{cmd}: total_results != expected. Dialect: {dialect}')
                env.assertEqual(
                    len(results), 0,
                    message=f'{cmd}: len(results) != 0. Dialect: {dialect}')


def test_limit00_resp3():
    _test_limit00(3)


def test_limit00_resp2():
    _test_limit00(2)


def _test_withcount(protocol):
    env = Env(protocol=protocol)
    docs = 2265
    _setup_index_and_data(env, docs)

    queries_and_results = [
        # query, total_results, length of results

        # WITHCOUNT
        # No sorter, no limit, returns all results
        (['FT.AGGREGATE', 'idx', '*', 'WITHCOUNT'], docs, docs),

        # WITHCOUNT + LIMIT
        # No sorter, limit results
        # total_results = number of documents matching the query up to the LIMIT
        # length of results = min(total_results, LIMIT)
        (['FT.AGGREGATE', 'idx', '*', 'WITHCOUNT', 'LIMIT', 0, 50], docs, 50),
        (['FT.AGGREGATE', 'idx', '*', 'WITHCOUNT', 'LIMIT', 0, int(docs/2)], docs, int(docs/2)),
        (['FT.AGGREGATE', 'idx', '*', 'WITHCOUNT', 'LIMIT', 0, docs*4], docs, docs),
        (['FT.AGGREGATE', 'idx', '*', 'WITHCOUNT', 'LIMIT', 10, 50], docs, 50),
        (['FT.AGGREGATE', 'idx', '*', 'WITHCOUNT', 'LIMIT', 100, docs], docs, docs - 100),
        (['FT.AGGREGATE', 'idx', '@price:[1, 100]', 'WITHCOUNT', 'LIMIT', 0, docs], 100, 100),
        (['FT.AGGREGATE', 'idx', '*', 'WITHCOUNT', 'LIMIT', docs, docs*2], docs, 0),

        # WITHCOUNT + SORTBY 0
        # Sorter without keys, no sorter, no limiter
        (['FT.AGGREGATE', 'idx', '*', 'WITHCOUNT', 'SORTBY', '0'], docs, docs),

        # WITHCOUNT + SORTBY 0 + MAX
        (['FT.AGGREGATE', 'idx', '*', 'WITHCOUNT', 'SORTBY', '0', 'MAX', 3], docs, 3),
        (['FT.AGGREGATE', 'idx', '*', 'WITHCOUNT', 'SORTBY', '0', 'MAX', 30], docs, 30),

        # WITHCOUNT + SORTBY
        # Sorter, limit results to DEFAULT_LIMIT
        # total_results = docs, length of results = DEFAULT_LIMIT
        (['FT.AGGREGATE', 'idx', '*', 'WITHCOUNT', 'SORTBY', '1', '@title'], docs, DEFAULT_LIMIT),
        (['FT.AGGREGATE', 'idx', '*', 'WITHCOUNT', 'SORTBY', '1', '@price'], docs, DEFAULT_LIMIT),
        (['FT.AGGREGATE', 'idx', '*', 'WITHCOUNT', 'SORTBY', '2', '@title', 'ASC'], docs, DEFAULT_LIMIT),
        (['FT.AGGREGATE', 'idx', '*', 'WITHCOUNT', 'SORTBY', '2', '@price', 'ASC'], docs, DEFAULT_LIMIT),

        # WITHCOUNT + SORTBY + MAX
        # total_results = docs, length of results = MAX
        (['FT.AGGREGATE', 'idx', '*', 'WITHCOUNT', 'SORTBY', '1', '@title', 'MAX', 3], docs, 3),
        (['FT.AGGREGATE', 'idx', '*', 'WITHCOUNT', 'SORTBY', '1', '@price', 'MAX', 4], docs, 4),
        (['FT.AGGREGATE', 'idx', '*', 'WITHCOUNT', 'SORTBY', '1', '@title', 'MAX', 30], docs, 30),
        (['FT.AGGREGATE', 'idx', '*', 'WITHCOUNT', 'SORTBY', '1', '@price', 'MAX', 40], docs, 40),

        # WITHCOUNT + SORTBY + LIMIT
        # total_results = number of documents matching the query up to the LIMIT
        # length of results = min(total_results, LIMIT)
        (['FT.AGGREGATE', 'idx', '*', 'WITHCOUNT', 'SORTBY', 1, '@title', 'LIMIT', 0, 50], docs, 50),
        (['FT.AGGREGATE', 'idx', '*', 'WITHCOUNT', 'SORTBY', 1, '@price', 'LIMIT', 0, 50], docs, 50),
        (['FT.AGGREGATE', 'idx', '*', 'WITHCOUNT', 'SORTBY', 1, '@title', 'LIMIT', 0, int(docs/2)], docs, int(docs/2)),
        (['FT.AGGREGATE', 'idx', '*', 'WITHCOUNT', 'SORTBY', 1, '@title', 'LIMIT', 0, docs*4], docs, docs),
        (['FT.AGGREGATE', 'idx', '*', 'WITHCOUNT', 'SORTBY', 1, '@title', 'LIMIT', 10, 50], docs, 50),
        (['FT.AGGREGATE', 'idx', '*', 'WITHCOUNT', 'SORTBY', 1, '@title', 'LIMIT', 100, docs], docs, docs - 100),
        (['FT.AGGREGATE', 'idx', '@price:[1, 100]', 'WITHCOUNT', 'SORTBY', 1, '@title', 'LIMIT', 0, docs], 100, 100),

        # WITHCOUNT + SORTBY + MAX + LIMIT
        # total_results = docs, length of results = LIMIT
        (['FT.AGGREGATE', 'idx', '*', 'WITHCOUNT', 'SORTBY', '1', '@title', 'MAX', 3, 'LIMIT', 0, 50], docs, 50),
        (['FT.AGGREGATE', 'idx', '*', 'WITHCOUNT', 'SORTBY', '1', '@title', 'MAX', docs*2, 'LIMIT', 0, 50], docs, 50),

        # WITHCOUNT + LOAD
        (['FT.AGGREGATE', 'idx', '*', 'WITHCOUNT', 'LOAD', 1, '@title'], docs, docs),
        (['FT.AGGREGATE', 'idx', '*', 'WITHCOUNT', 'LOAD', 1, '@price'], docs, docs),

        # WITHCOUNT + LOAD + LIMIT
        (['FT.AGGREGATE', 'idx', '*', 'WITHCOUNT', 'LOAD', 1, '@title', 'LIMIT', 0, 50], docs, 50),
        (['FT.AGGREGATE', 'idx', '*', 'WITHCOUNT', 'LOAD', 1, '@title', 'LIMIT', 100, docs], docs, docs - 100),

        # WITHCOUNT + GROUPBY
        (['FT.AGGREGATE', 'idx', '*', 'WITHCOUNT', 'GROUPBY', 1, '@brand'], 25, 25),

        # WITHCOUNT + GROUPBY + LIMIT
        (['FT.AGGREGATE', 'idx', '*', 'WITHCOUNT', 'GROUPBY', 1, '@brand', 'LIMIT', 0, 12], 25, 12),

        # WITHCOUNT + GROUPBY + SORTBY + LIMIT
        (['FT.AGGREGATE', 'idx', '*', 'WITHCOUNT', 'GROUPBY', 1, '@brand', 'SORTBY', 1, '@brand', 'LIMIT', 0, 11], 25, 11),
        (['FT.AGGREGATE', 'idx', '*', 'WITHCOUNT', 'GROUPBY', 1, '@brand', 'SORTBY', 1, '@brand', 'LIMIT', 0, 50], 25, 25),

        # WITHCOUNT + SORTBY + MAX -> GROUPBY (high-cardinality — fan-in reduction)
        (['FT.AGGREGATE', 'idx', '*', 'WITHCOUNT', 'SORTBY', 1, '@title', 'MAX', 50,
          'GROUPBY', 1, '@price', 'REDUCE', 'COUNT', 0, 'AS', 'cnt'], 50, 50),

        # WITHCOUNT + ADDSCORES
        (['FT.AGGREGATE', 'idx', '*', 'WITHCOUNT', 'ADDSCORES'], docs, docs),

        # WITHCOUNT + ADDSCORES + SORTBY
        (['FT.AGGREGATE', 'idx', '*', 'WITHCOUNT', 'ADDSCORES', 'SORTBY', 1, '@title'], docs, DEFAULT_LIMIT),

        # WITHCOUNT + ADDSCORES + LIMIT
        (['FT.AGGREGATE', 'idx', '*', 'WITHCOUNT', 'ADDSCORES', 'LIMIT', 0, 50], docs, 50),
        (['FT.AGGREGATE', 'idx', '*', 'WITHCOUNT', 'ADDSCORES', 'LIMIT', 10, 50], docs, 50),

        # WITHCOUNT + FILTER
        (['FT.AGGREGATE', 'idx', '*', 'WITHCOUNT', 'LOAD', 1, '@price', 'FILTER', '@price < 200'], 200, 200),
        (['FT.AGGREGATE', 'idx', '*', 'WITHCOUNT', 'LOAD', 1, '@price', 'FILTER', '@price >= 0'], docs, docs),

        # WITHCOUNT + FILTER + LIMIT
        (['FT.AGGREGATE', 'idx', '*', 'WITHCOUNT', 'LOAD', 1, '@price', 'FILTER', '@price < 200', 'LIMIT', 0, 50], 200, 50),
        (['FT.AGGREGATE', 'idx', '*', 'WITHCOUNT', 'LOAD', 1, '@price', 'FILTER', '@price < 200', 'LIMIT', 20, 50], 200, 50),
    ]

    for query, expected_total_results, expected_results in queries_and_results:
        cmd=' '.join(str(x) for x in query)
        for dialect in [1, 2, 3, 4]:
            config_cmd = ['CONFIG', 'SET', 'search-default-dialect', dialect]
            verify_command_OK_on_all_shards(env, *config_cmd)
            res = env.cmd(*query)
            total_results = _get_total_results(res)
            results = _get_results(res)

            # Verify results
            env.assertEqual(
                total_results, expected_total_results,
                message=f'{cmd}: total_results != expected. Dialect: {dialect}')
            env.assertEqual(
                len(results), expected_results,
                message=f'{cmd}: len(results) != expected. Dialect: {dialect}')

def test_withcount_resp3():
    _test_withcount(3)


def test_withcount_resp2():
    _test_withcount(2)


def _test_withoutcount(protocol):
    env = Env(protocol=protocol)
    docs = 2265
    _setup_index_and_data(env, docs)

    queries_and_results = [
        # WITHOUTCOUNT
        (['FT.AGGREGATE', 'idx', '*', 'WITHOUTCOUNT'], docs),

        # WITHOUTCOUNT + LIMIT
        (['FT.AGGREGATE', 'idx', '*', 'WITHOUTCOUNT', 'LIMIT', 0, 50], 50),
        (['FT.AGGREGATE', 'idx', '*', 'WITHOUTCOUNT', 'LIMIT', 0, int(docs/2)], int(docs/2)),
        (['FT.AGGREGATE', 'idx', '*', 'WITHOUTCOUNT', 'LIMIT', 0, docs*4], docs),
        (['FT.AGGREGATE', 'idx', '*', 'WITHOUTCOUNT', 'LIMIT', 10, 50], 50),
        (['FT.AGGREGATE', 'idx', '*', 'WITHOUTCOUNT', 'LIMIT', 100, docs], docs - 100),
        (['FT.AGGREGATE', 'idx', '*', 'WITHOUTCOUNT', 'LIMIT', docs, docs*2], 0),

        # WITHOUTCOUNT + SORTBY 0
        (['FT.AGGREGATE', 'idx', '*', 'WITHOUTCOUNT', 'SORTBY', '0'], docs),

        # WITHOUTCOUNT + SORTBY 0 + MAX
        (['FT.AGGREGATE', 'idx', '*', 'WITHOUTCOUNT', 'SORTBY', '0', 'MAX', 3], 3),
        (['FT.AGGREGATE', 'idx', '*', 'WITHOUTCOUNT', 'SORTBY', '0', 'MAX', 30], 30),

        # WITHOUTCOUNT + SORTBY - backwards compatible, returns only 10 results
        (['FT.AGGREGATE', 'idx', '*', 'WITHOUTCOUNT', 'SORTBY', '1', '@title'], DEFAULT_LIMIT),
        # (['FT.AGGREGATE', 'idx', '*', 'WITHOUTCOUNT', 'SORTBY', '1', '@price'], DEFAULT_LIMIT), # crash
        (['FT.AGGREGATE', 'idx', '*', 'WITHOUTCOUNT', 'SORTBY', '2', '@title', 'ASC'], DEFAULT_LIMIT),
        # (['FT.AGGREGATE', 'idx', '*', 'WITHOUTCOUNT', 'SORTBY', '2', '@price', 'ASC'], DEFAULT_LIMIT), # crash

        # WITHOUTCOUNT + SORTBY + MAX
        # total_results = docs, length of results = MAX
        (['FT.AGGREGATE', 'idx', '*', 'WITHOUTCOUNT', 'SORTBY', '1', '@title', 'MAX', 3], 3),
        (['FT.AGGREGATE', 'idx', '*', 'WITHOUTCOUNT', 'SORTBY', '1', '@price', 'MAX', 4], 4),
        (['FT.AGGREGATE', 'idx', '*', 'WITHOUTCOUNT', 'SORTBY', '1', '@title', 'MAX', 30], 30),
        (['FT.AGGREGATE', 'idx', '*', 'WITHOUTCOUNT', 'SORTBY', '1', '@price', 'MAX', 40], 40),

        # WITHOUTCOUNT + SORTBY + LIMIT
        (['FT.AGGREGATE', 'idx', '*', 'WITHOUTCOUNT', 'SORTBY', 1, '@title', 'LIMIT', 0, 50], 50),
        (['FT.AGGREGATE', 'idx', '*', 'WITHOUTCOUNT', 'SORTBY', 1, '@price', 'LIMIT', 0, 50], 50),
        (['FT.AGGREGATE', 'idx', '*', 'WITHOUTCOUNT', 'SORTBY', 1, '@title', 'LIMIT', 0, int(docs/2)], int(docs/2)),
        (['FT.AGGREGATE', 'idx', '*', 'WITHOUTCOUNT', 'SORTBY', 1, '@title', 'LIMIT', 0, docs*4], docs),
        (['FT.AGGREGATE', 'idx', '*', 'WITHOUTCOUNT', 'SORTBY', 1, '@title', 'LIMIT', 10, 50], 50),
        (['FT.AGGREGATE', 'idx', '*', 'WITHOUTCOUNT', 'SORTBY', 1, '@title', 'LIMIT', 100, docs], docs - 100),
        (['FT.AGGREGATE', 'idx', '@price:[1, 100]', 'WITHOUTCOUNT', 'SORTBY', 1, '@title', 'LIMIT', 0, docs], 100),

        # WITHOUTCOUNT + SORTBY + MAX + LIMIT
        # total_results = docs, length of results = LIMIT
        (['FT.AGGREGATE', 'idx', '*', 'WITHOUTCOUNT', 'SORTBY', '1', '@title', 'MAX', 3, 'LIMIT', 0, 50], 50),
        (['FT.AGGREGATE', 'idx', '*', 'WITHOUTCOUNT', 'SORTBY', '1', '@title', 'MAX', docs*2, 'LIMIT', 0, 50], 50),

        # WITHOUTCOUNT + LOAD
        (['FT.AGGREGATE', 'idx', '*', 'WITHOUTCOUNT', 'LOAD', 1, '@title'], docs),
        (['FT.AGGREGATE', 'idx', '*', 'WITHOUTCOUNT', 'LOAD', 1, '@price'], docs),

        # WITHOUTCOUNT + LOAD + LIMIT
        (['FT.AGGREGATE', 'idx', '*', 'WITHOUTCOUNT', 'LOAD', 1, '@title', 'LIMIT', 0, 50], 50),
        (['FT.AGGREGATE', 'idx', '*', 'WITHOUTCOUNT', 'LOAD', 1, '@title', 'LIMIT', 100, docs], docs - 100),

        # WITHOUTCOUNT + GROUPBY
        (['FT.AGGREGATE', 'idx', '*', 'WITHOUTCOUNT', 'GROUPBY', 1, '@brand'], 25),

        # WITHOUTCOUNT + GROUPBY + LIMIT
        (['FT.AGGREGATE', 'idx', '*', 'WITHOUTCOUNT', 'GROUPBY', 1, '@brand', 'LIMIT', 0, 12], 12),

        # WITHOUTCOUNT + GROUPBY + SORTBY + LIMIT
        (['FT.AGGREGATE', 'idx', '*', 'WITHOUTCOUNT', 'GROUPBY', 1, '@brand', 'SORTBY', 1, '@brand', 'LIMIT', 0, 11], 11),
        (['FT.AGGREGATE', 'idx', '*', 'WITHOUTCOUNT', 'GROUPBY', 1, '@brand', 'SORTBY', 1, '@brand', 'LIMIT', 0, 50], 25),

        # WITHOUTCOUNT + ADDSCORES
        (['FT.AGGREGATE', 'idx', '*', 'WITHOUTCOUNT', 'ADDSCORES'], docs),

        # WITHOUTCOUNT + ADDSCORES + SORTBY
        (['FT.AGGREGATE', 'idx', '*', 'WITHOUTCOUNT', 'ADDSCORES', 'SORTBY', 1, '@title'], 10),

        # WITHOUTCOUNT + ADDSCORES + LIMIT
        (['FT.AGGREGATE', 'idx', '*', 'WITHOUTCOUNT', 'ADDSCORES', 'LIMIT', 0, 50], 50),
        (['FT.AGGREGATE', 'idx', '*', 'WITHOUTCOUNT', 'ADDSCORES', 'LIMIT', 10, 50], 50),
    ]

    for query, expected_results in queries_and_results:
        cmd=' '.join(str(x) for x in query)

        # create a query without WITHOUTCOUNT to test the default behavior
        query_default = query.copy()
        query_default.remove('WITHOUTCOUNT')
        cmd_default=' '.join(str(x) for x in query_default)
        res_default = env.cmd(*query_default)

        for dialect in [1, 2, 3, 4]:
            config_cmd = ['CONFIG', 'SET', 'search-default-dialect', dialect]
            verify_command_OK_on_all_shards(env, *config_cmd)
            res = env.cmd(*query)

            results = _get_results(res)
            results_default = _get_results(res_default)

            # Verify only the length of results, don't verify total_results
            env.assertEqual(
                len(results), expected_results,
                message=f'{cmd}: len(results) != expected. Dialect: {dialect}')

            # Compare with the query without WITHOUTCOUNT
            env.assertEqual(
                len(results_default), expected_results,
                message=f'{cmd_default}: len(results) != results_default. Dialect: {dialect}')

def test_withoutcount_resp3():
    _test_withoutcount(3)

def test_withoutcount_resp2():
    _test_withoutcount(2)


def _test_profile(protocol):
    env = Env(protocol=protocol)
    docs = 3100
    _setup_index_and_data(env, docs)

    queries_and_profiles = [
        # query,
        # RESP2/RESP3 Standalone,
        # RESP2/RESP3 [[shard[0], shard[1], shard[2]], coordinator]

        # WITHCOUNT
        # No sorter, no limit, returns all results
        (['FT.AGGREGATE', 'idx', '*', 'WITHCOUNT'],
         [('Index', 3100), ('Depleter', 3100)],
         [[[('Index', 1027), ('Depleter', 1027)],
           [('Index', 1032), ('Depleter', 1032)],
           [('Index', 1041), ('Depleter', 1041)]],
           [('Network', 3100), ('Depleter', 3100)]]),

        # WITHCOUNT + LIMIT 0 0
        (['FT.AGGREGATE', 'idx', '*', 'WITHCOUNT', 'LIMIT', 0, 0],
         [('Index', 3100), ('Counter', 1)],
         [[[('Index', 1027), ('Depleter', 1027)],
           [('Index', 1032), ('Depleter', 1032)],
           [('Index', 1041), ('Depleter', 1041)]],
           [('Network', 3100), ('Counter', 1)]]),

        # WITHCOUNT + LIMIT
        # No sorter, limit results
        (['FT.AGGREGATE', 'idx', '*', 'WITHCOUNT', 'LIMIT', 0, 50],
         [('Index', 3100), ('Depleter', 49), ('Pager/Limiter', 50)],
         [[[('Index', 1027), ('Depleter', 49), ('Pager/Limiter', 50)],
           [('Index', 1032), ('Depleter', 49), ('Pager/Limiter', 50)],
           [('Index', 1041), ('Depleter', 49), ('Pager/Limiter', 50)]],
           [('Network', 150), ('Depleter', 49), ('Pager/Limiter', 50)]]),

        # WITHCOUNT + SORTBY 0
        # Sorter without keys, default limit
        (['FT.AGGREGATE', 'idx', '*', 'WITHCOUNT', 'SORTBY', '0'],
         [('Index', 3100), ('Depleter', 3100)],
         [[[('Index', 1027), ('Depleter', 1027)],
           [('Index', 1032), ('Depleter', 1032)],
           [('Index', 1041), ('Depleter', 1041)]],
           [('Network', 3100), ('Depleter', 3100)]]),

        # WITHCOUNT + SORTBY 0 + MAX
        (['FT.AGGREGATE', 'idx', '*', 'WITHCOUNT', 'SORTBY', '0', 'MAX', 3],
         [('Index', 3100), ('Depleter', 2), ('Pager/Limiter', 3)],
         [[[('Index', 1027), ('Depleter', 2), ('Pager/Limiter', 3)],
           [('Index', 1032), ('Depleter', 2), ('Pager/Limiter', 3)],
           [('Index', 1041), ('Depleter', 2), ('Pager/Limiter', 3)]],
           [('Network', 9), ('Depleter', 2), ('Pager/Limiter', 3)]]),

        # WITHCOUNT + SORTBY 0 + LIMIT
        (['FT.AGGREGATE', 'idx', '*', 'WITHCOUNT', 'SORTBY', '0', 'LIMIT', 0, 50],
         [('Index', 3100), ('Depleter', 49), ('Pager/Limiter', 50)],
         [[[('Index', 1027), ('Depleter', 49), ('Pager/Limiter', 50)],
           [('Index', 1032), ('Depleter', 49), ('Pager/Limiter', 50)],
           [('Index', 1041), ('Depleter', 49), ('Pager/Limiter', 50)]],
           [('Network', 150), ('Depleter', 49), ('Pager/Limiter', 50)]]),

        # WITHCOUNT + SORTBY
        # Sorter, limit results to DEFAULT_LIMIT
        (['FT.AGGREGATE', 'idx', '*', 'WITHCOUNT', 'SORTBY', '1', '@title'],
         [('Index', 3100), ('Sorter', 10)],
         [[[('Index', 1027), ('Sorter', 10), ('Loader', 10)],
           [('Index', 1032), ('Sorter', 10), ('Loader', 10)],
           [('Index', 1041), ('Sorter', 10), ('Loader', 10)]],
           [('Network', 30), ('Sorter', 10)]]),

        # WITHCOUNT + SORTBY + LIMIT
        (['FT.AGGREGATE', 'idx', '*', 'WITHCOUNT', 'SORTBY', 1, '@title', 'LIMIT', 0, 50],
         [('Index', 3100), ('Sorter', 50)],
         [[[('Index', 1027), ('Sorter', 50), ('Loader', 50)],
         [('Index', 1032), ('Sorter', 50), ('Loader', 50)],
         [('Index', 1041), ('Sorter', 50), ('Loader', 50)]],
         [('Network', 150), ('Sorter', 50)]]),

        # WITHCOUNT + SORTBY + MAX
        (['FT.AGGREGATE', 'idx', '*', 'WITHCOUNT', 'SORTBY', 1, '@title', 'MAX', 50],
         [('Index', 3100), ('Sorter', 50)],
         [[[('Index', 1027), ('Sorter', 50), ('Loader', 50)],
           [('Index', 1032), ('Sorter', 50), ('Loader', 50)],
           [('Index', 1041), ('Sorter', 50), ('Loader', 50)]],
           [('Network', 150), ('Sorter', 50)]]),

        # WITHCOUNT + LOAD
        (['FT.AGGREGATE', 'idx', '*', 'WITHCOUNT', 'LOAD', 1, '@title'],
         [('Index', 3100), ('Loader', 3100), ('Depleter', 3100)],
         [[[('Index', 1027), ('Loader', 1027), ('Depleter', 1027)],
           [('Index', 1032), ('Loader', 1032), ('Depleter', 1032)],
           [('Index', 1041), ('Loader', 1041), ('Depleter', 1041)]],
           [('Network', 3100), ('Depleter', 3100)]]),

        # WITHCOUNT + LOAD + LIMIT
        (['FT.AGGREGATE', 'idx', '*', 'WITHCOUNT', 'LOAD', 1, '@title', 'LIMIT', 0, 50],
         [('Index', 3100), ('Loader', 3100), ('Depleter', 49), ('Pager/Limiter', 50)],
         [[[('Index', 1027), ('Loader', 1027), ('Depleter', 49), ('Pager/Limiter', 50)],
           [('Index', 1032), ('Loader', 1032), ('Depleter', 49), ('Pager/Limiter', 50)],
           [('Index', 1041), ('Loader', 1041), ('Depleter', 49), ('Pager/Limiter', 50)]],
           [('Network', 150), ('Depleter', 49), ('Pager/Limiter', 50)]]),

        # WITHCOUNT + GROUPBY
        (['FT.AGGREGATE', 'idx', '*', 'WITHCOUNT', 'GROUPBY', 1, '@brand'],
         [('Index', 3100), ('Grouper', 25)],
         [[[('Index', 1027), ('Grouper', 25)],
           [('Index', 1032), ('Grouper', 25)],
           [('Index', 1041), ('Grouper', 25)]],
           [('Network', 75), ('Grouper', 25)]]),

        # WITHCOUNT + GROUPBY + SORTBY
        (['FT.AGGREGATE', 'idx', '*', 'WITHCOUNT', 'GROUPBY', 1, '@brand', 'SORTBY', 1, '@brand'],
         [('Index', 3100), ('Grouper', 25), ('Sorter', 10)],
         [[[('Index', 1027), ('Grouper', 25)],
           [('Index', 1032), ('Grouper', 25)],
           [('Index', 1041), ('Grouper', 25)]],
           [('Network', 75), ('Grouper', 25), ('Sorter', 10)]]),

        # WITHCOUNT + GROUPBY + LIMIT
        (['FT.AGGREGATE', 'idx', '*', 'WITHCOUNT', 'GROUPBY', 1, '@brand', 'LIMIT', 0, 50],
         [('Index', 3100), ('Grouper', 25), ('Pager/Limiter', 25)],
         [[[('Index', 1027), ('Grouper', 25)],
           [('Index', 1032), ('Grouper', 25)],
           [('Index', 1041), ('Grouper', 25)]],
           [('Network', 75), ('Grouper', 25), ('Pager/Limiter', 25)]]),

        # WITHCOUNT + GROUPBY + LIMIT (stop calling before EOF)
        (['FT.AGGREGATE', 'idx', '*', 'WITHCOUNT', 'GROUPBY', 1, '@brand', 'LIMIT', 0, 25],
         [('Index', 3100), ('Grouper', 24), ('Pager/Limiter', 25)],
         [[[('Index', 1027), ('Grouper', 25)],
           [('Index', 1032), ('Grouper', 25)],
           [('Index', 1041), ('Grouper', 25)]],
           [('Network', 75), ('Grouper', 24), ('Pager/Limiter', 25)]]),

        # WITHCOUNT + GROUPBY + SORTBY + LIMIT
        (['FT.AGGREGATE', 'idx', '*', 'WITHCOUNT', 'GROUPBY', 1, '@brand', 'SORTBY', 1, '@brand', 'LIMIT', 0, 50],
         [('Index', 3100), ('Grouper', 25), ('Sorter', 25)],
         [[[('Index', 1027), ('Grouper', 25)],
           [('Index', 1032), ('Grouper', 25)],
           [('Index', 1041), ('Grouper', 25)]],
           [('Network', 75), ('Grouper', 25), ('Sorter', 25)]]),

        # WITHCOUNT + ADDSCORES
        (['FT.AGGREGATE', 'idx', '*', 'WITHCOUNT', 'ADDSCORES'],
         [('Index', 3100), ('Scorer', 3100), ('Depleter', 3100)],
         [[[('Index', 1027), ('Scorer', 1027), ('Depleter', 1027)],
           [('Index', 1032), ('Scorer', 1032), ('Depleter', 1032)],
           [('Index', 1041), ('Scorer', 1041), ('Depleter', 1041)]],
           [('Network', 3100), ('Depleter', 3100)]]),

        # WITHCOUNT + ADDSCORES + SORTBY
        (['FT.AGGREGATE', 'idx', '*', 'WITHCOUNT', 'ADDSCORES', 'SORTBY', 1, '@title'],
         [('Index', 3100), ('Scorer', 3100), ('Sorter', 10)],
         [[[('Index', 1027), ('Scorer', 1027), ('Sorter', 10), ('Loader', 10)],
           [('Index', 1032), ('Scorer', 1032), ('Sorter', 10), ('Loader', 10)],
           [('Index', 1041), ('Scorer', 1041), ('Sorter', 10), ('Loader', 10)]],
           [('Network', 30), ('Sorter', 10)]]),

        # WITHCOUNT + ADDSCORES + LIMIT
        (['FT.AGGREGATE', 'idx', '*', 'WITHCOUNT', 'ADDSCORES', 'LIMIT', 0, 50],
         [('Index', 3100), ('Scorer', 3100), ('Depleter', 49), ('Pager/Limiter', 50)],
         [[[('Index', 1027), ('Scorer', 1027), ('Depleter', 49), ('Pager/Limiter', 50)],
           [('Index', 1032), ('Scorer', 1032), ('Depleter', 49), ('Pager/Limiter', 50)],
           [('Index', 1041), ('Scorer', 1041), ('Depleter', 49), ('Pager/Limiter', 50)]],
           [('Network', 150), ('Depleter', 49), ('Pager/Limiter', 50)]]),

        # WITHCOUNT + FILTER
        (['FT.AGGREGATE', 'idx', '*', 'WITHCOUNT', 'LOAD', 1, '@price', 'FILTER', '@price < 200'],
         [('Index', 3100), ('Loader', 3100), ('Filter - Predicate <', 200), ('Depleter', 200)],
         [[[('Index', 1027), ('Loader', 1027), ('Filter - Predicate <', 64), ('Depleter', 64)],
           [('Index', 1032), ('Loader', 1032), ('Filter - Predicate <', 68), ('Depleter', 68)],
           [('Index', 1041), ('Loader', 1041), ('Filter - Predicate <', 68), ('Depleter', 68)]],
           [('Network', 200), ('Depleter', 200)]]),

        # WITHCOUNT + FILTER + LIMIT
        (['FT.AGGREGATE', 'idx', '*', 'WITHCOUNT', 'LOAD', 1, '@price', 'FILTER', '@price < 200', 'LIMIT', 0, 50],
         [('Index', 3100), ('Loader', 3100), ('Filter - Predicate <', 200), ('Depleter', 49), ('Pager/Limiter', 50)],
         [[[('Index', 1027), ('Loader', 1027), ('Filter - Predicate <', 64), ('Depleter', 49), ('Pager/Limiter', 50)],
           [('Index', 1032), ('Loader', 1032), ('Filter - Predicate <', 68), ('Depleter', 49), ('Pager/Limiter', 50)],
           [('Index', 1041), ('Loader', 1041), ('Filter - Predicate <', 68), ('Depleter', 49), ('Pager/Limiter', 50)]],
           [('Network', 150), ('Depleter', 49), ('Pager/Limiter', 50)]]),

        # WITHCOUNT + SORTBY -> GROUPBY
        (['FT.AGGREGATE', 'idx', '*', 'WITHCOUNT', 'SORTBY', 1, '@title',
          'GROUPBY', 1, '@brand', 'REDUCE', 'COUNT', 0, 'AS', 'cnt'],
         [('Index', 3100), ('Sorter', 10), ('Grouper', 7)],
         [[[('Index', 1027), ('Sorter', 10), ('Loader', 10)],
           [('Index', 1032), ('Sorter', 10), ('Loader', 10)],
           [('Index', 1041), ('Sorter', 10), ('Loader', 10)]],
           [('Network', 30), ('Sorter', 10), ('Grouper', 7)]]),

        # WITHCOUNT + SORTBY + MAX -> GROUPBY
        (['FT.AGGREGATE', 'idx', '*', 'WITHCOUNT', 'SORTBY', 1, '@title', 'MAX', 50,
          'GROUPBY', 1, '@brand', 'REDUCE', 'COUNT', 0, 'AS', 'cnt'],
         [('Index', 3100), ('Sorter', 50), ('Grouper', 25)],
         [[[('Index', 1027), ('Sorter', 50), ('Loader', 50)],
           [('Index', 1032), ('Sorter', 50), ('Loader', 50)],
           [('Index', 1041), ('Sorter', 50), ('Loader', 50)]],
           [('Network', 150), ('Sorter', 50), ('Grouper', 25)]]),

        # SORTBY+MAX before GROUPBY on high-cardinality field — demonstrates fan-in reduction.
        # Without SORTBY+MAX, GROUPBY @price sends ~1000 groups per shard (Network ~3100).
        # With SORTBY MAX 50, each shard sends only 50 sorted docs (Network 150) — a ~20x reduction.
        (['FT.AGGREGATE', 'idx', '*', 'WITHCOUNT', 'SORTBY', 1, '@title', 'MAX', 50,
          'GROUPBY', 1, '@price', 'REDUCE', 'COUNT', 0, 'AS', 'cnt'],
         [('Index', 3100), ('Sorter', 50), ('Loader', 50), ('Grouper', 50)],
         [[[('Index', 1027), ('Sorter', 50), ('Loader', 50)],
           [('Index', 1032), ('Sorter', 50), ('Loader', 50)],
           [('Index', 1041), ('Sorter', 50), ('Loader', 50)]],
           [('Network', 150), ('Sorter', 50), ('Grouper', 50)]]),

        # WITHCOUNT + SORTBY + MAX -> GROUPBY + REDUCE -> SORTBY -> LIMIT
        (['FT.AGGREGATE', 'idx', '*', 'WITHCOUNT', 'SORTBY', 2, '@price', 'DESC', 'MAX', 100,
          'GROUPBY', 1, '@brand', 'REDUCE', 'COUNT', 0, 'AS', 'cnt',
          'SORTBY', 2, '@cnt', 'DESC', 'LIMIT', 0, 10],
         [('Index', 3100), ('Loader', 3100), ('Sorter', 100), ('Grouper', 25), ('Sorter', 10)],
         [[[('Index', 1027), ('Loader', 1027), ('Sorter', 100), ('Loader', 100)],
           [('Index', 1032), ('Loader', 1032), ('Sorter', 100), ('Loader', 100)],
           [('Index', 1041), ('Loader', 1041), ('Sorter', 100), ('Loader', 100)]],
           [('Network', 300), ('Sorter', 100), ('Grouper', 25), ('Sorter', 10)]]),

        # WITHCOUNT + LOAD -> SORTBY + MAX -> GROUPBY + REDUCE -> FILTER
        (['FT.AGGREGATE', 'idx', '*', 'WITHCOUNT', 'LOAD', 1, '@price',
          'SORTBY', 2, '@price', 'DESC', 'MAX', 200,
          'GROUPBY', 1, '@brand', 'REDUCE', 'COUNT', 0, 'AS', 'cnt',
          'FILTER', '@cnt > 5'],
         [('Index', 3100), ('Loader', 3100), ('Sorter', 200), ('Grouper', 25), ('Filter - Predicate >', 25)],
         [[[('Index', 1027), ('Loader', 1027), ('Sorter', 200), ('Loader', 200)],
           [('Index', 1032), ('Loader', 1032), ('Sorter', 200), ('Loader', 200)],
           [('Index', 1041), ('Loader', 1041), ('Sorter', 200), ('Loader', 200)]],
           [('Network', 600), ('Sorter', 200), ('Grouper', 25), ('Filter - Predicate >', 25)]]),

        # WITHCOUNT + GROUPBY -> GROUPBY (re-grouping)
        (['FT.AGGREGATE', 'idx', '*', 'WITHCOUNT',
          'GROUPBY', 1, '@brand', 'REDUCE', 'COUNT', 0, 'AS', 'cnt',
          'GROUPBY', 1, '@cnt', 'REDUCE', 'COUNT', 0, 'AS', 'num_brands'],
         [('Index', 3100), ('Grouper', 25), ('Grouper', 1)],
         [[[('Index', 1027), ('Grouper', 25)],
           [('Index', 1032), ('Grouper', 25)],
           [('Index', 1041), ('Grouper', 25)]],
           [('Network', 75), ('Grouper', 25), ('Grouper', 1)]]),

        # WITHCOUNT + GROUPBY -> SORTBY -> GROUPBY (mixed pipeline)
        (['FT.AGGREGATE', 'idx', '*', 'WITHCOUNT',
          'GROUPBY', 1, '@brand', 'REDUCE', 'COUNT', 0, 'AS', 'cnt',
          'SORTBY', 2, '@cnt', 'DESC',
          'GROUPBY', 1, '@cnt', 'REDUCE', 'COUNT', 0, 'AS', 'num_brands'],
         [('Index', 3100), ('Grouper', 25), ('Sorter', 10), ('Grouper', 1)],
         [[[('Index', 1027), ('Grouper', 25)],
           [('Index', 1032), ('Grouper', 25)],
           [('Index', 1041), ('Grouper', 25)]],
           [('Network', 75), ('Grouper', 25), ('Sorter', 10), ('Grouper', 1)]]),

        # ----------------------------------------------------------------------
        # WITHOUTCOUNT
        (['FT.AGGREGATE', 'idx', '*', 'WITHOUTCOUNT'],
         [('Index', 3100)],
         [[[('Index', 1027)],
           [('Index', 1032)],
           [('Index', 1041)]],
           [('Network', 3100)]]),

        # WITHOUTCOUNT implicit (by default)
        (['FT.AGGREGATE', 'idx', '*'],
         [('Index', 3100)],
         [[[('Index', 1027)],
           [('Index', 1032)],
           [('Index', 1041)]],
           [('Network', 3100)]]),

        # WITHOUTCOUNT + LIMIT
        (['FT.AGGREGATE', 'idx', '*', 'WITHOUTCOUNT', 'LIMIT', 0, 50],
         [('Index', 49), ('Pager/Limiter', 50)],
         [[[('Index', 49), ('Pager/Limiter', 50)],
           [('Index', 49), ('Pager/Limiter', 50)],
           [('Index', 49), ('Pager/Limiter', 50)]],
           [('Network', 49), ('Pager/Limiter', 50)]]),

         # WITHOUTCOUNT (implicit) + LIMIT
        (['FT.AGGREGATE', 'idx', '*', 'LIMIT', 0, 50],
         [('Index', 49), ('Pager/Limiter', 50)],
         [[[('Index', 49), ('Pager/Limiter', 50)],
           [('Index', 49), ('Pager/Limiter', 50)],
           [('Index', 49), ('Pager/Limiter', 50)]],
           [('Network', 49), ('Pager/Limiter', 50)]]),

        # WITHOUTCOUNT + SORTBY 0
        (['FT.AGGREGATE', 'idx', '*', 'WITHOUTCOUNT', 'SORTBY', '0'],
         [('Index', 3100)],
         [[[('Index', 1027)],
           [('Index', 1032)],
           [('Index', 1041)]],
           [('Network', 3100)]]),

        # WITHOUTCOUNT + SORTBY 0 + MAX
        (['FT.AGGREGATE', 'idx', '*', 'WITHOUTCOUNT', 'SORTBY', '0', 'MAX', 3],
         [('Index', 2), ('Pager/Limiter', 3)],
         [[[('Index', 2), ('Pager/Limiter', 3)],
           [('Index', 2), ('Pager/Limiter', 3)],
           [('Index', 2), ('Pager/Limiter', 3)]],
           [('Network', 2), ('Pager/Limiter', 3)]]),

        # WITHOUTCOUNT + SORTBY 0 + LIMIT
        (['FT.AGGREGATE', 'idx', '*', 'WITHOUTCOUNT', 'SORTBY', '0', 'LIMIT', 0, 50],
         [('Index', 49), ('Pager/Limiter', 50)],
         [[[('Index', 49), ('Pager/Limiter', 50)], [('Index', 49), ('Pager/Limiter', 50)], [('Index', 49), ('Pager/Limiter', 50)]], [('Network', 49), ('Pager/Limiter', 50)]]),

        # WITHOUTCOUNT + SORTBY
        (['FT.AGGREGATE', 'idx', '*', 'WITHOUTCOUNT', 'SORTBY', '1', '@title'],
         [('Index', 3100), ('Sorter', 10)],
         [[[('Index', 1027), ('Sorter', 10), ('Loader', 10)],
           [('Index', 1032), ('Sorter', 10), ('Loader', 10)],
           [('Index', 1041), ('Sorter', 10), ('Loader', 10)]],
           [('Network', 30), ('Sorter', 10)]]),

        # WITHOUTCOUNT + SORTBY + LIMIT
        (['FT.AGGREGATE', 'idx', '*', 'WITHOUTCOUNT', 'SORTBY', 1, '@title', 'LIMIT', 0, 50],
         [('Index', 3100), ('Sorter', 50)],
         [[[('Index', 1027), ('Sorter', 50), ('Loader', 50)],
           [('Index', 1032), ('Sorter', 50), ('Loader', 50)],
           [('Index', 1041), ('Sorter', 50), ('Loader', 50)]],
           [('Network', 150), ('Sorter', 50)]]),

        # WITHOUTCOUNT + SORTBY + MAX
        (['FT.AGGREGATE', 'idx', '*', 'WITHOUTCOUNT', 'SORTBY', 1, '@title', 'MAX', 50],
         [('Index', 3100), ('Sorter', 50)],
         [[[('Index', 1027), ('Sorter', 50), ('Loader', 50)],
           [('Index', 1032), ('Sorter', 50), ('Loader', 50)],
           [('Index', 1041), ('Sorter', 50), ('Loader', 50)]],
           [('Network', 150), ('Sorter', 50)]]),

        # WITHOUTCOUNT + GROUPBY + LIMIT (stop calling before EOF)
        (['FT.AGGREGATE', 'idx', '*', 'WITHOUTCOUNT', 'GROUPBY', 1, '@brand', 'LIMIT', 0, 25],
         [('Index', 3100), ('Grouper', 24), ('Pager/Limiter', 25)],
         [[[('Index', 1027), ('Grouper', 25)],
           [('Index', 1032), ('Grouper', 25)],
           [('Index', 1041), ('Grouper', 25)]],
           [('Network', 75), ('Grouper', 24), ('Pager/Limiter', 25)]]),

         # WITHOUTCOUNT + LOAD
        (['FT.AGGREGATE', 'idx', '*', 'WITHOUTCOUNT', 'LOAD', 1, '@title'],
         [('Index', 3100), ('Loader', 3100)],
         [[[('Index', 1027), ('Loader', 1027)],
           [('Index', 1032), ('Loader', 1032)],
           [('Index', 1041), ('Loader', 1041)]],
           [('Network', 3100)]]),

         # WITHOUTCOUNT + GROUPBY
        (['FT.AGGREGATE', 'idx', '*', 'WITHOUTCOUNT', 'GROUPBY', 1, '@brand'],
         [('Index', 3100), ('Grouper', 25)],
         [[[('Index', 1027), ('Grouper', 25)],
           [('Index', 1032), ('Grouper', 25)],
           [('Index', 1041), ('Grouper', 25)]],
           [('Network', 75), ('Grouper', 25)]]),

        # WITHOUTCOUNT + GROUPBY + SORTBY
        (['FT.AGGREGATE', 'idx', '*', 'WITHOUTCOUNT', 'GROUPBY', 1, '@brand', 'SORTBY', 1, '@brand'],
         [('Index', 3100), ('Grouper', 25), ('Sorter', 10)],
         [[[('Index', 1027), ('Grouper', 25)],
           [('Index', 1032), ('Grouper', 25)],
           [('Index', 1041), ('Grouper', 25)]],
           [('Network', 75), ('Grouper', 25), ('Sorter', 10)]]),

        # WITHOUTCOUNT + GROUPBY + LIMIT
        (['FT.AGGREGATE', 'idx', '*', 'WITHOUTCOUNT', 'GROUPBY', 1, '@brand', 'LIMIT', 0, 50],
         [('Index', 3100), ('Grouper', 25), ('Pager/Limiter', 25)],
         [[[('Index', 1027), ('Grouper', 25)],
           [('Index', 1032), ('Grouper', 25)],
           [('Index', 1041), ('Grouper', 25)]],
           [('Network', 75), ('Grouper', 25), ('Pager/Limiter', 25)]]),

        # WITHOUTCOUNT + GROUPBY + SORTBY + LIMIT
        (['FT.AGGREGATE', 'idx', '*', 'WITHOUTCOUNT', 'GROUPBY', 1, '@brand', 'SORTBY', 1, '@brand', 'LIMIT', 0, 50],
         [('Index', 3100), ('Grouper', 25), ('Sorter', 25)],
         [[[('Index', 1027), ('Grouper', 25)],
           [('Index', 1032), ('Grouper', 25)],
           [('Index', 1041), ('Grouper', 25)]],
           [('Network', 75), ('Grouper', 25), ('Sorter', 25)]]),

        # WITHOUTCOUNT + ADDSCORES
        (['FT.AGGREGATE', 'idx', '*', 'WITHOUTCOUNT', 'ADDSCORES'],
         [('Index', 3100), ('Scorer', 3100)],
         [[[('Index', 1027), ('Scorer', 1027)],
           [('Index', 1032), ('Scorer', 1032)],
           [('Index', 1041), ('Scorer', 1041)]],
           [('Network', 3100)]]),

        # WITHOUTCOUNT + ADDSCORES + SORTBY
        (['FT.AGGREGATE', 'idx', '*', 'WITHOUTCOUNT', 'ADDSCORES', 'SORTBY', 1, '@title'],
         [('Index', 3100), ('Scorer', 3100), ('Sorter', 10)],
         [[[('Index', 1027), ('Scorer', 1027), ('Sorter', 10), ('Loader', 10)],
           [('Index', 1032), ('Scorer', 1032), ('Sorter', 10), ('Loader', 10)],
           [('Index', 1041), ('Scorer', 1041), ('Sorter', 10), ('Loader', 10)]],
           [('Network', 30), ('Sorter', 10)]]),

        # WITHOUTCOUNT + ADDSCORES + LIMIT
        (['FT.AGGREGATE', 'idx', '*', 'WITHOUTCOUNT', 'ADDSCORES', 'LIMIT', 0, 50],
         [('Index', 49), ('Scorer', 49), ('Pager/Limiter', 50)],
         [[[('Index', 49), ('Scorer', 49), ('Pager/Limiter', 50)],
           [('Index', 49), ('Scorer', 49), ('Pager/Limiter', 50)],
           [('Index', 49), ('Scorer', 49), ('Pager/Limiter', 50)]],
           [('Network', 49), ('Pager/Limiter', 50)]]),

        # WITHOUTCOUNT + FILTER
        (['FT.AGGREGATE', 'idx', '*', 'WITHOUTCOUNT', 'LOAD', 1, '@price', 'FILTER', '@price < 200'],
         [('Index', 3100), ('Loader', 3100), ('Filter - Predicate <', 200)],
         [[[('Index', 1027), ('Loader', 1027), ('Filter - Predicate <', 64)],
           [('Index', 1032), ('Loader', 1032), ('Filter - Predicate <', 68)],
           [('Index', 1041), ('Loader', 1041), ('Filter - Predicate <', 68)]],
           [('Network', 200)]]),

        # WITHOUTCOUNT + FILTER + LIMIT
        (['FT.AGGREGATE', 'idx', '*', 'WITHOUTCOUNT', 'LOAD', 1, '@price', 'FILTER', '@price < 200', 'LIMIT', 0, 50],
         [('Index', 49), ('Loader', 49), ('Filter - Predicate <', 49), ('Pager/Limiter', 50)],
         [[[('Index', 49), ('Loader', 49), ('Filter - Predicate <', 49), ('Pager/Limiter', 50)],
           [('Index', 49), ('Loader', 49), ('Filter - Predicate <', 49), ('Pager/Limiter', 50)],
           [('Index', 49), ('Loader', 49), ('Filter - Predicate <', 49), ('Pager/Limiter', 50)]],
           [('Network', 49), ('Pager/Limiter', 50)]]),

        # WITHOUTCOUNT + SORTBY + MAX -> GROUPBY
        (['FT.AGGREGATE', 'idx', '*', 'WITHOUTCOUNT', 'SORTBY', 1, '@title', 'MAX', 50,
          'GROUPBY', 1, '@brand', 'REDUCE', 'COUNT', 0, 'AS', 'cnt'],
         [('Index', 49), ('Pager/Limiter', 50), ('Grouper', 25)],
         [[[('Index', 1027), ('Sorter', 50), ('Loader', 50)],
           [('Index', 1032), ('Sorter', 50), ('Loader', 50)],
           [('Index', 1041), ('Sorter', 50), ('Loader', 50)]],
           [('Network', 150), ('Sorter', 50), ('Grouper', 25)]]),

        # WITHOUTCOUNT + SORTBY + MAX -> GROUPBY + REDUCE -> SORTBY -> LIMIT
        (['FT.AGGREGATE', 'idx', '*', 'WITHOUTCOUNT', 'SORTBY', 2, '@price', 'DESC', 'MAX', 100,
          'GROUPBY', 1, '@brand', 'REDUCE', 'COUNT', 0, 'AS', 'cnt',
          'SORTBY', 2, '@cnt', 'DESC', 'LIMIT', 0, 10],
         [('Index', 3100), ('Loader', 3100), ('Sorter', 100), ('Grouper', 25), ('Sorter', 10)],
         [[[('Index', 1027), ('Loader', 1027), ('Sorter', 100), ('Loader', 100)],
           [('Index', 1032), ('Loader', 1032), ('Sorter', 100), ('Loader', 100)],
           [('Index', 1041), ('Loader', 1041), ('Sorter', 100), ('Loader', 100)]],
           [('Network', 300), ('Sorter', 100), ('Grouper', 25), ('Sorter', 10)]]),

        # WITHOUTCOUNT + LOAD -> SORTBY + MAX -> GROUPBY + REDUCE -> FILTER
        (['FT.AGGREGATE', 'idx', '*', 'WITHOUTCOUNT', 'LOAD', 1, '@price',
          'SORTBY', 2, '@price', 'DESC', 'MAX', 200,
          'GROUPBY', 1, '@brand', 'REDUCE', 'COUNT', 0, 'AS', 'cnt',
          'FILTER', '@cnt > 5'],
         [('Index', 199), ('Loader', 199), ('Pager/Limiter', 200), ('Grouper', 25), ('Filter - Predicate >', 25)],
         [[[('Index', 1027), ('Loader', 1027), ('Sorter', 200), ('Loader', 200)],
           [('Index', 1032), ('Loader', 1032), ('Sorter', 200), ('Loader', 200)],
           [('Index', 1041), ('Loader', 1041), ('Sorter', 200), ('Loader', 200)]],
           [('Network', 600), ('Sorter', 200), ('Grouper', 25), ('Filter - Predicate >', 25)]]),

        # WITHOUTCOUNT + GROUPBY -> GROUPBY (re-grouping)
        (['FT.AGGREGATE', 'idx', '*', 'WITHOUTCOUNT',
          'GROUPBY', 1, '@brand', 'REDUCE', 'COUNT', 0, 'AS', 'cnt',
          'GROUPBY', 1, '@cnt', 'REDUCE', 'COUNT', 0, 'AS', 'num_brands'],
         [('Index', 3100), ('Grouper', 25), ('Grouper', 1)],
         [[[('Index', 1027), ('Grouper', 25)],
           [('Index', 1032), ('Grouper', 25)],
           [('Index', 1041), ('Grouper', 25)]],
           [('Network', 75), ('Grouper', 25), ('Grouper', 1)]]),

        # MOD-14849: WITHOUTCOUNT + SORTBY (no MAX) + GROUPBY returns
        # "Success (not an error)". Uncomment when MOD-14849 is fixed.
        #
        # # WITHOUTCOUNT + SORTBY -> GROUPBY
        # (['FT.AGGREGATE', 'idx', '*', 'WITHOUTCOUNT', 'SORTBY', 1, '@title',
        #   'GROUPBY', 1, '@brand', 'REDUCE', 'COUNT', 0, 'AS', 'cnt'],
        #  [<TBD standalone profile>],
        #  [<TBD cluster profile>]),
        #
        # # WITHOUTCOUNT + GROUPBY -> SORTBY -> GROUPBY (mixed pipeline)
        # (['FT.AGGREGATE', 'idx', '*', 'WITHOUTCOUNT',
        #   'GROUPBY', 1, '@category', 'REDUCE', 'COUNT', 0, 'AS', 'cnt',
        #   'SORTBY', 2, '@cnt', 'DESC',
        #   'GROUPBY', 1, '@cnt', 'REDUCE', 'COUNT', 0, 'AS', 'num_categories'],
        #  [<TBD standalone profile>],
        #  [<TBD cluster profile>]),

    ]

    for (query, standalone, cluster) in queries_and_profiles:
        cmd=' '.join(str(x) for x in query)
        ftprofile = _translate_query_to_profile_query(query)
        res = env.cmd(*ftprofile)

        if env.isCluster():
            message = f'{cmd}: RP_list != expected: RESP{env.protocol}, Cluster'
            cluster_RP_list = _get_cluster_RP_profile(env, res)
            env.assertEqual(cluster_RP_list, cluster,
                            message=message)
        else:
            message = f'{cmd}: RP_list != expected: RESP{env.protocol}, Standalone'
            standalone_RP_list = _get_standalone_RP_profile(env, res)
            env.assertEqual(standalone_RP_list, standalone,
                            message=message)


def test_profile_resp2():
    _test_profile(2)

def test_profile_resp3():
    _test_profile(3)

def _test_withcursor(protocol):
    env = Env(protocol=protocol)
    docs = 25
    _setup_index_and_data(env, docs)

    # WITHCOUNT + WITHCURSOR was previously rejected at parse time; make sure
    # the combination is now accepted and that every chunk reports the same
    # total_results (the full pipeline count, not the chunk size).
    queries = [
        ['FT.AGGREGATE', 'idx', '*', 'WITHCOUNT', 'WITHCURSOR', 'COUNT', 5],
        ['FT.AGGREGATE', 'idx', '*', 'WITHCURSOR', 'COUNT', 5, 'WITHCOUNT'],
        ['FT.AGGREGATE', 'idx', '*', 'WITHCURSOR', 'WITHCOUNT'],
    ]
    for query in queries:
        res, cursor = env.cmd(*query)
        first_total = _get_total_results(res)
        env.assertEqual(first_total, docs,
                        message=f'{query}: first chunk total_results')
        rows_seen = len(_get_results(res))
        while cursor != 0:
            res, cursor = env.cmd('FT.CURSOR', 'READ', 'idx', str(cursor))
            env.assertEqual(_get_total_results(res), first_total,
                            message=f'{query}: per-chunk total_results drift')
            rows_seen += len(_get_results(res))
        env.assertEqual(rows_seen, docs,
                        message=f'{query}: total rows across chunks')

def test_withcursor_resp2():
    _test_withcursor(2)

def test_withcursor_resp3():
    _test_withcursor(3)

def _test_pagers(protocol):
    env = Env(protocol=protocol)
    docs = 10
    _setup_index_and_data(env, docs)

    queries = [
        ['FT.AGGREGATE', 'idx', '*', 'WITHCOUNT', 'LOAD', 1, '@title'],
        ['FT.AGGREGATE', 'idx', '*', 'WITHCOUNT', 'LOAD', 1, '@title', 'SORTBY', 1, '@title'],
        ['FT.AGGREGATE', 'idx', '*', 'WITHCOUNT', 'LOAD', 1, '@title', 'GROUPBY', 1, '@brand'],
        ['FT.AGGREGATE', 'idx', '*', 'WITHCOUNT', 'ADDSCORES', 'LOAD', 1, '@title'],
        ['FT.AGGREGATE', 'idx', '*', 'WITHCOUNT', 'ADDSCORES', 'LOAD', 1, '@title', 'SORTBY', 2, '@price', 'DESC'],
    ]
    for query in queries:
        limit = 6
        offset = 2
        query1 = query + ['LIMIT', 0, limit]
        query2 = query + ['LIMIT', offset, limit]
        res1 = env.cmd(*query1)
        res2 = env.cmd(*query2)

        # Compare total_results
        total_results1 = _get_total_results(res1)
        total_results2 = _get_total_results(res2)
        env.assertEqual(total_results1, total_results2)

        # Compare length of results
        results1 = _get_results(res1)
        results2 = _get_results(res2)
        env.assertEqual(len(results1), len(results2))

        # Compare common part of the results (order-sensitive only for SORTBY;
        # GROUPBY without SORTBY has no deterministic iteration order in cluster mode)
        if 'SORTBY' in query:
            env.assertEqual(results1[offset:limit + offset + 1],
                            results2[0:limit - offset], message=query)

def test_pagers_resp2():
    _test_pagers(2)

def test_pagers_resp3():
    _test_pagers(3)