Feature/alias as index attribute

integration/compatibility/__init__.py

@@ -9,8 +9,9 @@
 # ANSWER_FILE_NAME) and add an entry here. regenerate.sh and
 # compatibility_test.py both read from this list.
 GENERATORS = [
-    {"generator": "generate.py",      "answers": "aggregate-answers.pickle.gz",   "cluster": True},
-    {"generator": "generate_text.py", "answers": "text-search-answers.pickle.gz", "cluster": False},
+    {"generator": "generate.py",       "answers": "aggregate-answers.pickle.gz",   "cluster": True},
+    {"generator": "generate_alias.py", "answers": "alias-answers.pickle.gz",       "cluster": True},
+    {"generator": "generate_text.py",  "answers": "text-search-answers.pickle.gz", "cluster": False},
 ]
 
 

integration/compatibility/data_sets.py

@@ -10,6 +10,7 @@
 
 SETS_KEY = lambda key_type: f"{key_type} sets"
 CREATES_KEY = lambda key_type: f"{key_type} creates"
+SETUP_KEY = lambda key_type: f"{key_type} setup"
 
 # Text data configuration
 TEXT_SCHEMA = {
@@ -211,6 +212,8 @@ def unbytes(b):
         return b.decode("utf-8")
     else:
         return b
+
+
 class ClientSystem:
     def __init__(self, address):
         self.address = address
@@ -621,18 +624,63 @@ def generate_doc(doc_id):
     return data
 
 ### Helper Functions ###
+def compute_alias_data():
+    """Return alias compatibility dataset in the standard compute_data_sets() shape.
+    Supports both hash and json key types.
+    """
+    data = {"alias": {}}
+    for key_type in ["hash", "json"]:
+        if key_type == "hash":
+            data["alias"][CREATES_KEY(key_type)] = [
+                "FT.CREATE hash_idx1 ON HASH PREFIX 1 adoc: SCHEMA price NUMERIC category TAG",
+                "FT.CREATE hash_idx2 ON HASH PREFIX 1 empty: SCHEMA price NUMERIC category TAG",
+            ]
+            data["alias"][SETS_KEY(key_type)] = [
+                (f"adoc:{i}", {"price": str(i * 10),
+                               "category": "electronics" if i % 2 == 0 else "books"})
+                for i in range(5)
+            ]
+            data["alias"][SETUP_KEY(key_type)] = [
+                ["FT.ALIASUPDATE", "alias_search", "hash_idx1"],
+                ["FT.ALIASUPDATE", "alias_agg",    "hash_idx1"],
+            ]
+        else:
+            data["alias"][CREATES_KEY(key_type)] = [
+                "FT.CREATE json_idx1 ON JSON PREFIX 1 jdoc: SCHEMA $.price AS price NUMERIC $.category AS category TAG",
+                "FT.CREATE json_idx2 ON JSON PREFIX 1 jempty: SCHEMA $.price AS price NUMERIC $.category AS category TAG",
+            ]
+            data["alias"][SETS_KEY(key_type)] = [
+                (f"jdoc:{i}", {"price": i * 10,
+                               "category": "electronics" if i % 2 == 0 else "books"})
+                for i in range(5)
+            ]
+            data["alias"][SETUP_KEY(key_type)] = [
+                ["FT.ALIASUPDATE", "alias_search", "json_idx1"],
+                ["FT.ALIASUPDATE", "alias_agg",    "json_idx1"],
+            ]
+    return data
+
+
 def load_data(client, data_set, key_type, data_source=None, schema_type="default"):
     # Auto-detect data source based on data_set name
     if data_source is None:
-        data_source = "text" if data_set in TEXT_DATASETS else "vector"
+        if data_set == "alias":
+            data_source = "alias"
+        elif data_set in TEXT_DATASETS:
+            data_source = "text"
+        else:
+            data_source = "vector"
 
     match data_source:
+        case "alias":
+            data = compute_alias_data()
         case "vector":
             data = compute_data_sets()
         case "text":
             data = compute_text_data_sets(data_set, schema_type=schema_type)
         case _:
             raise ValueError(f"Unknown data source: {data_source}")
+
     load_list = data[data_set][SETS_KEY(key_type)]
     for create_index_cmd in data[data_set][CREATES_KEY(key_type)]:
         client.execute_command(create_index_cmd)
@@ -648,7 +696,23 @@ def load_data(client, data_set, key_type, data_source=None, schema_type="default
                 pipe.execute_command(*["JSON.SET", cmd[0], "$", json.dumps(cmd[1])])
         pipe.execute()
 
-    # client.wait_for_indexing_done(f"{key_type}_idx1")
+    # Run any post-load setup commands
+    for setup_cmd in data[data_set].get(SETUP_KEY(key_type), []):
+        client.execute_command(*setup_cmd)
+    # Verify that each alias expected to be live after setup actually resolves.
+    if data[data_set].get(SETUP_KEY(key_type)):
+        setup_cmds = data[data_set][SETUP_KEY(key_type)]
+        live_aliases: set[str] = set()
+        for cmd in setup_cmds:
+            verb = cmd[0].upper()
+            alias = cmd[1]
+            if verb in ("FT.ALIASADD", "FT.ALIASUPDATE"):
+                live_aliases.add(alias)
+            elif verb == "FT.ALIASDEL":
+                live_aliases.discard(alias)
+        for alias in live_aliases:
+            client.execute_command("FT.INFO", alias)
+
     print(f"setup_data completed {data_set} {key_type}")
 
     # Print loaded data for debugging
@@ -662,22 +726,31 @@ def load_data(client, data_set, key_type, data_source=None, schema_type="default
             print(f"{s}:{load_list[s][0]}:  ", k)
     return len(load_list)
 
-def load_data_cluster(cluster_client, test_case, data_set, key_type):
-    data = compute_data_sets()
+def load_data_cluster(cluster_client, test_case, data_set_name, key_type):
+    if data_set_name == "alias":
+        data = compute_alias_data()
+    elif data_set_name in TEXT_DATASETS:
+        data = compute_text_data_sets(data_set_name)
+    else:
+        data = compute_data_sets()
 
     primary0 = test_case.new_client_for_primary(0)
-    for create_cmd in data[data_set][CREATES_KEY(key_type)]:
+    for create_cmd in data[data_set_name][CREATES_KEY(key_type)]:
         primary0.execute_command(create_cmd)
 
-    for key, fields in data[data_set][SETS_KEY(key_type)]:
+    for key, fields in data[data_set_name][SETS_KEY(key_type)]:
         if key_type == "hash":
             cluster_client.hset(key, mapping=fields)
         else:
             cluster_client.execute_command(
                 "JSON.SET", key, "$", json.dumps(fields)
             )
 
-    print(f"cluster load completed {data_set} {key_type}")
+    # Run any post-load setup commands (e.g. alias creation) via primary 0
+    for setup_cmd in data[data_set_name].get(SETUP_KEY(key_type), []):
+        primary0.execute_command(*setup_cmd)
+
+    print(f"cluster load completed {data_set_name} {key_type}")
 
 def extract_vocab_from_text_data(dataset_name, key_type):
     """Extract unique words from TEXT fields in a text data set."""

integration/compatibility/generate.py

@@ -307,6 +307,7 @@ def test_aggregate_numeric_dyadic_operators_sortable_numbers(self, key_type, dia
                 f"ft.aggregate {key_type}_idx1  * load 3 @__key @n1 @n2 apply @n1{op}@n2 as nn"
             )
 
+    @pytest.mark.skip(reason="Needs research")
     def test_aggregate_numeric_triadic_operators(self, key_type, dialect):
         self.setup_data("hard numbers", key_type)
         dyadic = ["+", "-", "*", "/", "^"]

integration/compatibility/generate_alias.py

@@ -0,0 +1,81 @@
+import pytest
+from .data_sets import *
+from .generate import BaseCompatibilityTest
+
+'''
+Capture alias compatibility answers from Redisearch
+'''
+
+
+@pytest.mark.parametrize("key_type", ["json", "hash"])
+class TestAliasCompatibility(BaseCompatibilityTest):
+    # Alias management & error cases
+
+    ANSWER_FILE_NAME = "alias-answers.pickle.gz"
+
+    @pytest.fixture(autouse=True)
+    def _setup_alias_data(self, key_type):
+        """Load alias dataset for the current key_type (runs after setup_method)."""
+        load_data(self.client, "alias", key_type)
+        self.data_set_name = "alias"
+        self.key_type = key_type
+
+    def test_aliasadd_search(self, key_type):
+        """FT.SEARCH via alias returns same results as via index name."""
+        self.execute_command(["FT.INFO", "alias_search"])
+        self.execute_command(["FT.SEARCH", "alias_search", "@price:[0 +inf]"])
+
+    def test_aliasadd_aggregate(self, key_type):
+        """FT.AGGREGATE via alias returns same results as via index name."""
+        self.execute_command(["FT.INFO", "alias_agg"])
+        self.execute_command(["FT.AGGREGATE", "alias_agg", "@category:{electronics}",
+            "LOAD", "1", "@category",
+            "GROUPBY", "1", "@category",
+            "REDUCE", "COUNT", "0", "AS", "count",
+        ])
+
+    def test_aliasadd_collides_with_existing_index(self, key_type):
+        """FT.ALIASADD where alias name matches a different existing index."""
+        if key_type == "hash":
+            self.client.execute_command(
+                "FT.CREATE", "second_idx", "ON", "HASH", "PREFIX", "1", "bdoc:",
+                "SCHEMA", "val", "NUMERIC")
+        else:
+            self.client.execute_command(
+                "FT.CREATE", "second_idx", "ON", "JSON", "PREFIX", "1", "bdoc:",
+                "SCHEMA", "$.val", "AS", "val", "NUMERIC")
+        self.execute_command(["FT.ALIASADD", "second_idx", f"{key_type}_idx1"])
+
+    def test_aliasupdate_collides_with_existing_index(self, key_type):
+        """FT.ALIASUPDATE where alias name matches a different existing index."""
+        if key_type == "hash":
+            self.client.execute_command(
+                "FT.CREATE", "second_idx", "ON", "HASH", "PREFIX", "1", "bdoc:",
+                "SCHEMA", "val", "NUMERIC")
+        else:
+            self.client.execute_command(
+                "FT.CREATE", "second_idx", "ON", "JSON", "PREFIX", "1", "bdoc:",
+                "SCHEMA", "$.val", "AS", "val", "NUMERIC")
+        self.execute_command(["FT.ALIASUPDATE", "second_idx", f"{key_type}_idx1"])
+
+    def test_aliasadd_duplicate(self, key_type):
+        """FT.ALIASADD with an already-existing alias returns an error."""
+        self.execute_command(["FT.ALIASADD", "alias_search", f"{key_type}_idx1"])
+
+    def test_aliasadd_nonexistent_index(self, key_type):
+        """FT.ALIASADD for a non-existent index returns an error."""
+        self.execute_command(["FT.ALIASADD", "new_alias", "no_such_index"])
+
+    def test_aliasadd_alias_to_alias(self, key_type):
+        """FT.ALIASADD pointing to an existing alias returns an error."""
+        self.execute_command(["FT.ALIASADD", "chain_alias", "alias_search"])
+
+    def test_aliasdel_nonexistent(self, key_type):
+        """FT.ALIASDEL on a non-existent alias returns an error."""
+        self.execute_command(["FT.ALIASDEL", "no_such_alias"])
+
+    def test_aliasupdate_nonexistent_index(self, key_type):
+        """FT.ALIASUPDATE for a non-existent index returns an error."""
+        self.execute_command(["FT.ALIASUPDATE", "new_alias", "no_such_index"])
+
+

integration/compatibility_test.py

@@ -315,6 +315,32 @@ def compare_results(expected, results):
         print(TEST_MARKER)
         return False
 
+    # Alias mutation commands return b'OK', not a result list.
+    if _is_simple_compare_cmd(cmd):
+        # FT.INFO contains dynamic values (memory, timing), just verify both
+        # succeed with the same index name, or both return the same error.
+        if cmd[0].upper() == "FT.INFO":
+            exp = expected["result"]
+            act = results["result"]
+            # Both errors -> compare error text
+            if isinstance(exp, Exception) and isinstance(act, Exception):
+                return str(exp) == str(act)
+            # One error, one success -> mismatch
+            if isinstance(exp, Exception) or isinstance(act, Exception):
+                print(f"FT.INFO mismatch: expected={exp!r} got={act!r}")
+                return False
+            # Both lists - compare index name (first element)
+            if isinstance(exp, list) and isinstance(act, list) and len(exp) > 0 and len(act) > 0:
+                if exp[0] != act[0]:
+                    print(f"FT.INFO index name mismatch: expected={exp[0]!r} got={act[0]!r}")
+                    return False
+                return True
+            return exp == act
+        match = expected["result"] == results["result"]
+        if not match:
+            print(f"Simple result mismatch: expected={expected['result']!r} got={results['result']!r}")
+        return match
+
     # Output raw results
     # print("Raw expected result:", expected["result"])
     rl = unpack_result(cmd, expected["key_type"], expected["result"], sortkeys)
@@ -385,9 +411,34 @@ def mark_as_failed(testname):
     wrong_answers += 1
     assert not StopOnFailure, "Test failed, stopping execution"
 
+_SIMPLE_COMPARE_CMDS = {"FT.ALIASADD", "FT.ALIASDEL", "FT.ALIASUPDATE", "FT.INFO"}
+
+def _is_simple_compare_cmd(cmd):
+    """Return True if cmd should use direct equality comparison (not search/aggregate unpacking)."""
+    return cmd and cmd[0].upper() in _SIMPLE_COMPARE_CMDS
+
+_ALIAS_MUTATION_VERBS = {"FT.ALIASADD", "FT.ALIASDEL", "FT.ALIASUPDATE"}
+
+def _is_alias_mutation_cmd(cmd):
+    """Return True if cmd is an alias mutation command (needs cluster routing)."""
+    return cmd and cmd[0].upper() in _ALIAS_MUTATION_VERBS
+
+def _first_index_name(data_set_name, key_type):
+    """Return the first index name created by a dataset's CREATES_KEY commands."""
+    if data_set_name == "alias":
+        data = compute_alias_data()
+    elif data_set_name in TEXT_DATASETS:
+        data = compute_text_data_sets(data_set_name)
+    else:
+        data = compute_data_sets()
+    create_cmd = data[data_set_name][CREATES_KEY(key_type)][0]
+    # FT.CREATE <index_name> ... — index name is the second token.
+    return create_cmd.split()[1]
+
 def do_answer(client, expected, data_set):
     global correct_answers, failed_tests, passed_tests
-    if (expected['data_set_name'], expected['key_type'], expected.get('schema_type')) != data_set:
+    reload = expected.get('reload', False)
+    if reload or (expected['data_set_name'], expected['key_type'], expected.get('schema_type')) != data_set:
         print("Loading data set:", expected['data_set_name'], "key type:", expected['key_type'])
         client.execute_command("FLUSHALL SYNC")
         load_data(client, expected['data_set_name'], expected['key_type'], schema_type=expected.get('schema_type', 'default'))
@@ -433,20 +484,27 @@ def do_answer(client, expected, data_set):
     return data_set
 
 def drop_index_cluster(test_case, key_type):
-    index_name = "json_idx1" if key_type == "json" else "hash_idx1"
     primary0 = test_case.new_client_for_primary(0)
     try:
-        primary0.execute_command("FT.DROPINDEX", index_name)
-        print(f"Dropped index {index_name}")
+        indexes = primary0.execute_command("FT._LIST")
+        for index_name in indexes:
+            if isinstance(index_name, bytes):
+                index_name = index_name.decode()
+            try:
+                primary0.execute_command("FT.DROPINDEX", index_name)
+                print(f"Dropped index {index_name}")
+            except valkey.ResponseError:
+                pass
     except valkey.ResponseError:
-        pass  # index may not exist yet
+        pass
 
 def do_answer_cluster(cluster_client, expected, data_set, test_case):
     global correct_answers, failed_tests, passed_tests
 
     next_data_set = (expected["data_set_name"], expected["key_type"])
+    reload = expected.get("reload", False)
 
-    if data_set != next_data_set:
+    if reload or data_set != next_data_set:
         print(
             "Loading CLUSTER data set:",
             expected["data_set_name"],
@@ -467,6 +525,43 @@ def do_answer_cluster(cluster_client, expected, data_set, test_case):
 
         data_set = next_data_set
 
+    # Replay alias commands via primary 0 (can't be slot-routed by cluster client).
+    if _is_alias_mutation_cmd(expected["cmd"]):
+        result = {}
+        result["cmd"] = expected["cmd"]
+        try:
+            # Route to primary 0 — cluster client can't slot FT.ALIAS* commands.
+            primary0 = test_case.new_client_for_primary(0)
+            result["result"] = primary0.execute_command(*expected["cmd"])
+            result["exception"] = False
+            print(f"Alias cmd (cluster): {expected['cmd']} -> {result['result']!r}")
+        except valkey.ResponseError as e:
+            result["result"] = {}
+            result["exception"] = True
+            print(f"Alias cmd (cluster) error: {expected['cmd']} -> {e}")
+
+        if compare_results(expected, result):
+            mark_as_passed(expected.get('testname', f"alias_cmd_{expected['cmd'][0]}"))
+        else:
+            mark_as_failed(expected.get('testname', f"alias_cmd_{expected['cmd'][0]}"))
+
+        # Wait for alias to be visible on all primaries after successful add/update.
+        cmd = expected["cmd"]
+        if not result["exception"] and len(cmd) >= 2 and cmd[0].upper() in ("FT.ALIASADD", "FT.ALIASUPDATE"):
+            alias_name = cmd[1]
+            def _alias_visible_on_all():
+                for i in range(test_case.CLUSTER_SIZE):
+                    try:
+                        test_case.new_client_for_primary(i).execute_command(
+                            "FT.INFO", alias_name
+                        )
+                    except valkey.ResponseError:
+                        return False
+                return True
+            waiters.wait_for_true(_alias_visible_on_all, timeout=15)
+
+        return data_set
+
     result = {}
     try:
         print(