import apexbase as ab
from apexbase import ApexClient
import tempfile
import pytest
class TestColumnEncoding:
def test_create_index_on_string_column(self):
with tempfile.TemporaryDirectory() as tmpdir:
client = ab.ApexClient(dirpath=tmpdir)
client.create_table('test')
data = [{'col': f'value_{i}'} for i in range(100)]
client.store(data)
client.execute('CREATE INDEX idx_col ON test (col)')
print('Index created successfully')
result = client.execute("SELECT * FROM test WHERE col = 'value_50'")
assert len(result) == 1
assert result[0]['col'] == 'value_50'
client.close()
def test_create_index_on_dict_encoded_column(self):
with tempfile.TemporaryDirectory() as tmpdir:
client = ab.ApexClient(dirpath=tmpdir)
client.create_table('test')
data = [{'category': f'cat_{i % 5}'} for i in range(1000)]
client.store(data)
client.execute('CREATE INDEX idx_category ON test (category)')
print('Index on dict-encoded column created successfully')
result = client.execute("SELECT * FROM test WHERE category = 'cat_2'")
assert len(result) == 200
client.close()
def test_unicode_string_column(self):
with tempfile.TemporaryDirectory() as tmpdir:
client = ab.ApexClient(dirpath=tmpdir)
client.create_table('test')
data = [
{'text': 'ä½ å¥½ä¸–ç•Œ'},
{'text': 'Hello World'},
{'text': '日本語テスト'},
{'text': '🔒 Emoji Test'},
]
client.store(data)
client.execute('CREATE INDEX idx_text ON test (text)')
print('Index on Unicode column created successfully')
result = client.execute("SELECT * FROM test WHERE text = 'ä½ å¥½ä¸–ç•Œ'")
assert len(result) == 1
client.close()
def test_special_characters_column(self):
with tempfile.TemporaryDirectory() as tmpdir:
client = ab.ApexClient(dirpath=tmpdir)
client.create_table('test')
data = [
{'text': 'line1\nline2'},
{'text': 'tab\there'},
{'text': 'quote"here'},
{'text': 'backslash\\here'},
]
client.store(data)
client.execute('CREATE INDEX idx_text ON test (text)')
print('Index on special chars column created successfully')
client.close()
def test_large_string_column(self):
with tempfile.TemporaryDirectory() as tmpdir:
client = ab.ApexClient(dirpath=tmpdir)
client.create_table('test')
large_text = 'x' * 10000
data = [
{'text': large_text},
{'text': 'short'},
{'text': 'medium' * 100},
]
client.store(data)
client.execute('CREATE INDEX idx_text ON test (text)')
print('Index on large string column created successfully')
client.close()
def test_multiple_string_columns(self):
with tempfile.TemporaryDirectory() as tmpdir:
client = ab.ApexClient(dirpath=tmpdir)
client.create_table('test')
data = []
for i in range(500):
data.append({
'low_card': f'type_{i % 3}', 'med_card': f'group_{i % 50}', 'high_card': f'unique_{i}', })
client.store(data)
client.execute('CREATE INDEX idx_low ON test (low_card)')
client.execute('CREATE INDEX idx_med ON test (med_card)')
client.execute('CREATE INDEX idx_high ON test (high_card)')
print('Indexes on multiple string columns created successfully')
result1 = client.execute("SELECT * FROM test WHERE low_card = 'type_1'")
assert len(result1) > 0
result2 = client.execute("SELECT * FROM test WHERE med_card = 'group_25'")
assert len(result2) > 0
client.close()
def test_mixed_type_columns_with_string(self):
with tempfile.TemporaryDirectory() as tmpdir:
client = ab.ApexClient(dirpath=tmpdir)
client.create_table('test')
data = []
for i in range(200):
data.append({
'id': i,
'name': f'name_{i}',
'value': float(i) * 1.5,
'active': i % 2 == 0,
})
client.store(data)
client.execute('CREATE INDEX idx_name ON test (name)')
print('Index on mixed-type table string column created successfully')
result = client.execute("SELECT * FROM test WHERE name = 'name_100'")
assert len(result) == 1
assert result[0]['id'] == 100
client.close()
if __name__ == '__main__':
pytest.main([__file__, '-v'])