Skip to content
Closed
Show file tree
Hide file tree
Changes from 9 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 3 additions & 2 deletions integration/compatibility/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -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},
]


Expand Down
Binary file modified integration/compatibility/aggregate-answers.pickle.gz
Binary file not shown.
Binary file added integration/compatibility/alias-answers.pickle.gz
Binary file not shown.
87 changes: 80 additions & 7 deletions integration/compatibility/data_sets.py
Original file line number Diff line number Diff line change
Expand Up @@ -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 = {
Expand Down Expand Up @@ -211,6 +212,8 @@ def unbytes(b):
return b.decode("utf-8")
else:
return b


class ClientSystem:
def __init__(self, address):
self.address = address
Expand Down Expand Up @@ -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)
Expand All @@ -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
Expand All @@ -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."""
Expand Down
1 change: 1 addition & 0 deletions integration/compatibility/generate.py
Original file line number Diff line number Diff line change
Expand Up @@ -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 = ["+", "-", "*", "/", "^"]
Expand Down
81 changes: 81 additions & 0 deletions integration/compatibility/generate_alias.py
Original file line number Diff line number Diff line change
@@ -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"])


Binary file modified integration/compatibility/text-search-answers.pickle.gz
Binary file not shown.
107 changes: 101 additions & 6 deletions integration/compatibility_test.py
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down Expand Up @@ -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'))
Expand Down Expand Up @@ -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"],
Expand All @@ -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(
Expand Down
Loading
Loading