377 lines
18 KiB
Python
377 lines
18 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""Test indexes"""
|
|
|
|
import pytest
|
|
|
|
from pyrseas.testutils import DatabaseToMapTestCase
|
|
from pyrseas.testutils import InputMapToSqlTestCase, fix_indent
|
|
|
|
CREATE_TABLE_STMT = "CREATE TABLE t1 (c1 integer, c2 text)"
|
|
CREATE_STMT = "CREATE INDEX t1_idx ON t1 (c1)"
|
|
COMMENT_STMT = "COMMENT ON INDEX sd.t1_idx IS 'Test index t1_idx'"
|
|
|
|
|
|
class IndexToMapTestCase(DatabaseToMapTestCase):
|
|
"""Test mapping of created indexes"""
|
|
|
|
def test_index_1(self):
|
|
"Map a single-column index"
|
|
dbmap = self.to_map([CREATE_TABLE_STMT, CREATE_STMT])
|
|
expmap = {'columns': [{'c1': {'type': 'integer'}},
|
|
{'c2': {'type': 'text'}}],
|
|
'indexes': {'t1_idx': {'keys': ['c1']}}}
|
|
assert dbmap['schema sd']['table t1'] == expmap
|
|
|
|
def test_index_2(self):
|
|
"Map a two-column index"
|
|
stmts = ["CREATE TABLE t1 (c1 INTEGER, c2 CHAR(5), c3 TEXT)",
|
|
"CREATE UNIQUE INDEX t1_idx ON t1 (c1, c2)"]
|
|
dbmap = self.to_map(stmts)
|
|
expmap = {'columns': [{'c1': {'type': 'integer'}},
|
|
{'c2': {'type': 'character(5)'}},
|
|
{'c3': {'type': 'text'}}],
|
|
'indexes': {'t1_idx': {'keys': ['c1', 'c2'],
|
|
'unique': True}}}
|
|
assert dbmap['schema sd']['table t1'] == expmap
|
|
|
|
def test_index_3(self):
|
|
"Map a table with a unique index and a non-unique GIN index"
|
|
stmts = ["CREATE TABLE t1 (c1 INTEGER, c2 CHAR(5), c3 tsvector)",
|
|
"CREATE UNIQUE INDEX t1_idx_1 ON t1 (c1, c2)",
|
|
"CREATE INDEX t1_idx_2 ON t1 USING gin (c3)"]
|
|
dbmap = self.to_map(stmts)
|
|
expmap = {'columns': [{'c1': {'type': 'integer'}},
|
|
{'c2': {'type': 'character(5)'}},
|
|
{'c3': {'type': 'tsvector'}}],
|
|
'indexes': {'t1_idx_1': {'keys': ['c1', 'c2'],
|
|
'unique': True},
|
|
't1_idx_2': {'keys': ['c3'],
|
|
'access_method': 'gin'}}}
|
|
assert dbmap['schema sd']['table t1'] == expmap
|
|
|
|
def test_map_index_partial(self):
|
|
"Map a table with a partial index"
|
|
dbmap = self.to_map([CREATE_TABLE_STMT,
|
|
"CREATE INDEX t1_idx ON t1 (c2) WHERE c1 > 42"])
|
|
expmap = {'columns': [{'c1': {'type': 'integer'}},
|
|
{'c2': {'type': 'text'}}],
|
|
'indexes': {'t1_idx': {'keys': ['c2'],
|
|
'predicate': '(c1 > 42)'}}}
|
|
assert dbmap['schema sd']['table t1'] == expmap
|
|
|
|
def test_index_function_simple(self):
|
|
"Map an index using a function -- issue #98"
|
|
stmts = [CREATE_TABLE_STMT, "CREATE INDEX t1_idx ON t1 ((lower(c2)))"]
|
|
dbmap = self.to_map(stmts)
|
|
expmap = {'columns': [{'c1': {'type': 'integer'}},
|
|
{'c2': {'type': 'text'}}],
|
|
'indexes': {'t1_idx': {'keys': [
|
|
{'lower(c2)': {'type': 'expression'}}]}}}
|
|
assert dbmap['schema sd']['table t1'] == expmap
|
|
|
|
def test_index_function_complex(self):
|
|
"Map indexes using nested functions and complex arguments"
|
|
stmts = ["CREATE TABLE t1 (c1 integer, c2 text, c3 date)",
|
|
"CREATE INDEX t1_idx1 ON t1 (substring(c2 from position("
|
|
"'_begin' in c2)), substring(c2 from position('_end' in "
|
|
"c2)))",
|
|
"CREATE INDEX t1_idx2 ON t1 (extract(month from c3), "
|
|
"extract(day from c3))"]
|
|
dbmap = self.to_map(stmts)
|
|
expr_type = {'type': 'expression'}
|
|
key1 = [{"SUBSTRING(c2 FROM POSITION(('_begin'::text) IN (c2)))":
|
|
expr_type},
|
|
{"SUBSTRING(c2 FROM POSITION(('_end'::text) IN (c2)))":
|
|
expr_type}]
|
|
key2 = [{"EXTRACT(month FROM c3)": expr_type},
|
|
{"EXTRACT(day FROM c3)": expr_type}]
|
|
if self.db.version < 150000:
|
|
key1 = [{'"substring"(c2, "position"(c2, \'_begin\''
|
|
'::text))': expr_type},
|
|
{'"substring"(c2, "position"(c2, \'_end\''
|
|
'::text))': expr_type}]
|
|
key2 = [{"date_part('month'::text, c3)": {
|
|
'type': 'expression'}},
|
|
{"date_part('day'::text, c3)": {
|
|
'type': 'expression'}}]
|
|
expmap = {'columns': [{'c1': {'type': 'integer'}},
|
|
{'c2': {'type': 'text'}},
|
|
{'c3': {'type': 'date'}}],
|
|
'indexes': {'t1_idx1': {'keys': key1},
|
|
't1_idx2': {'keys': key2}}}
|
|
assert dbmap['schema sd']['table t1'] == expmap
|
|
|
|
def test_index_col_opts(self):
|
|
"Map an index with various column options"
|
|
stmts = ["CREATE TABLE t1 (c1 cidr, c2 text)",
|
|
"CREATE INDEX t1_idx ON t1 (c1 cidr_ops NULLS FIRST, "
|
|
"c2 DESC)"]
|
|
dbmap = self.to_map(stmts)
|
|
expmap = {'columns': [{'c1': {'type': 'cidr'}},
|
|
{'c2': {'type': 'text'}}],
|
|
'indexes': {'t1_idx': {'keys': [
|
|
{'c1': {'opclass': 'cidr_ops', 'nulls': 'first'}},
|
|
{'c2': {'order': 'desc'}}]}}}
|
|
assert dbmap['schema sd']['table t1'] == expmap
|
|
|
|
def test_index_mixed(self):
|
|
"Map indexes using functions, a regular column and expressions"
|
|
stmts = ["CREATE TABLE t1 (c1 integer, c2 text, c3 text)",
|
|
"CREATE INDEX t1_idx ON t1 (btrim(c3, 'x') NULLS FIRST, c1, "
|
|
"lower(c2) DESC)",
|
|
"CREATE INDEX t1_idx2 ON t1 ((c2 || ', ' || c3), "
|
|
"(c3 || ' ' || c2))"]
|
|
dbmap = self.to_map(stmts)
|
|
expmap = {'columns': [{'c1': {'type': 'integer'}},
|
|
{'c2': {'type': 'text'}},
|
|
{'c3': {'type': 'text'}}],
|
|
'indexes': {'t1_idx': {
|
|
'keys': [{"btrim(c3, 'x'::text)": {
|
|
'type': 'expression', 'nulls': 'first'}},
|
|
'c1', {'lower(c2)': {
|
|
'type': 'expression', 'order': 'desc'}}]},
|
|
't1_idx2': {
|
|
'keys': [{"(((c2 || ', '::text) || c3))": {
|
|
'type': 'expression'}},
|
|
{"(((c3 || ' '::text) || c2))": {
|
|
'type': 'expression'}}]}}}
|
|
assert dbmap['schema sd']['table t1'] == expmap
|
|
|
|
def test_map_index_cluster(self):
|
|
"Map a table with an index and cluster on it"
|
|
dbmap = self.to_map([CREATE_TABLE_STMT, CREATE_STMT,
|
|
"CLUSTER t1 USING t1_idx"])
|
|
expmap = {'columns': [{'c1': {'type': 'integer'}},
|
|
{'c2': {'type': 'text'}}],
|
|
'indexes': {'t1_idx': {'keys': ['c1'], 'cluster': True}}}
|
|
assert dbmap['schema sd']['table t1'] == expmap
|
|
|
|
def test_map_index_comment(self):
|
|
"Map an index comment"
|
|
dbmap = self.to_map([CREATE_TABLE_STMT, CREATE_STMT, COMMENT_STMT])
|
|
assert dbmap['schema sd']['table t1']['indexes']['t1_idx'][
|
|
'description'] == 'Test index t1_idx'
|
|
|
|
def test_map_multicol_index_with_exprs(self):
|
|
"Map a multicol index with expressions"
|
|
dbmap = self.to_map(["CREATE TABLE holiday (id serial PRIMARY KEY,"
|
|
"date date NOT NULL, recurring boolean NOT NULL)",
|
|
"CREATE UNIQUE INDEX unique_date ON holiday (("
|
|
"CASE WHEN recurring THEN (0)::double precision "
|
|
"ELSE date_part('year'::text, date) END), "
|
|
"date_part('month'::text, date), "
|
|
"date_part('day'::text, date))"])
|
|
fmt = "(\nCASE\n %s\n %s\nEND)"
|
|
assert dbmap['schema sd']['table holiday']['indexes'][
|
|
'unique_date'] == {
|
|
'keys': [
|
|
{fmt % ("WHEN recurring THEN (0)::double precision",
|
|
"ELSE date_part('year'::text, date)"):
|
|
{'type': 'expression'}},
|
|
{"date_part('month'::text, date)": {'type': 'expression'}},
|
|
{"date_part('day'::text, date)": {'type': 'expression'}}],
|
|
'unique': True}
|
|
|
|
|
|
class IndexToSqlTestCase(InputMapToSqlTestCase):
|
|
"""Test SQL generation from input indexes"""
|
|
|
|
def test_create_table_with_index(self):
|
|
"Create new table with a single column index"
|
|
inmap = self.std_map()
|
|
inmap['schema sd'].update({'table t1': {
|
|
'columns': [{'c1': {'type': 'integer'}},
|
|
{'c2': {'type': 'text'}}],
|
|
'indexes': {'t1_idx': {'keys': ['c1']}}}})
|
|
sql = self.to_sql(inmap)
|
|
assert fix_indent(sql[0]) == "CREATE TABLE sd.t1 (c1 integer, c2 text)"
|
|
assert sql[1] == "CREATE INDEX t1_idx ON sd.t1 (c1)"
|
|
|
|
def test_add_index(self):
|
|
"Add a two-column unique index to an existing table"
|
|
stmts = ["CREATE TABLE t1 (c1 INTEGER NOT NULL, "
|
|
"c2 INTEGER NOT NULL, c3 TEXT)"]
|
|
inmap = self.std_map()
|
|
inmap['schema sd'].update({'table t1': {
|
|
'columns': [{'c1': {'type': 'integer', 'not_null': True}},
|
|
{'c2': {'type': 'integer', 'not_null': True}},
|
|
{'c3': {'type': 'text'}}],
|
|
'indexes': {'t1_idx': {'keys': ['c2', 'c1'], 'unique': True}}}})
|
|
sql = self.to_sql(inmap, stmts)
|
|
assert sql == ["CREATE UNIQUE INDEX t1_idx ON sd.t1 (c2, c1)"]
|
|
|
|
def test_add_index_schema(self):
|
|
"Add an index to an existing table in a non-default schema"
|
|
stmts = ["CREATE SCHEMA s1",
|
|
"CREATE TABLE s1.t1 (c1 INTEGER NOT NULL, "
|
|
"c2 INTEGER NOT NULL, c3 TEXT)"]
|
|
inmap = self.std_map()
|
|
inmap.update({'schema s1': {'table t1': {
|
|
'columns': [{'c1': {'type': 'integer', 'not_null': True}},
|
|
{'c2': {'type': 'integer', 'not_null': True}},
|
|
{'c3': {'type': 'text'}}],
|
|
'indexes': {'t1_idx': {'keys': ['c2', 'c1'], 'unique': True}}}}})
|
|
sql = self.to_sql(inmap, stmts)
|
|
assert sql == ["CREATE UNIQUE INDEX t1_idx ON s1.t1 (c2, c1)"]
|
|
|
|
def test_add_index_back_compat(self):
|
|
"Add a index to an existing table accepting back-compatible spec"
|
|
stmts = ["CREATE TABLE t1 (c1 INTEGER NOT NULL, "
|
|
"c2 INTEGER NOT NULL, c3 TEXT)"]
|
|
inmap = self.std_map()
|
|
inmap['schema sd'].update({'table t1': {
|
|
'columns': [{'c1': {'type': 'integer', 'not_null': True}},
|
|
{'c2': {'type': 'integer', 'not_null': True}},
|
|
{'c3': {'type': 'text'}}],
|
|
'indexes': {'t1_idx': {'columns': ['c2', 'c1'], 'unique': True,
|
|
'access_method': 'btree'}}}})
|
|
sql = self.to_sql(inmap, stmts)
|
|
assert sql == ["CREATE UNIQUE INDEX t1_idx ON sd.t1 (c2, c1)"]
|
|
|
|
def test_bad_index(self):
|
|
"Fail on creating an index without columns or expression"
|
|
inmap = self.std_map()
|
|
inmap['schema sd'].update({'table t1': {
|
|
'columns': [{'c1': {'type': 'integer'}}, {'c2': {'type': 'text'}}],
|
|
'indexes': {'t1_idx': {'access_method': 'btree'}}}})
|
|
with pytest.raises(KeyError):
|
|
self.to_sql(inmap)
|
|
|
|
def test_create_partial(self):
|
|
"Create a partial index"
|
|
inmap = self.std_map()
|
|
inmap['schema sd'].update({'table t1': {
|
|
'columns': [{'c1': {'type': 'integer'}},
|
|
{'c2': {'type': 'text'}}],
|
|
'indexes': {'t1_idx': {'keys': ['c2'],
|
|
'predicate': '(c1 > 42)'}}}})
|
|
sql = self.to_sql(inmap, [CREATE_TABLE_STMT])
|
|
assert fix_indent(sql[0]) == \
|
|
"CREATE INDEX t1_idx ON sd.t1 (c2) WHERE (c1 > 42)"
|
|
|
|
def test_create_index_function(self):
|
|
"Create an index which uses a function"
|
|
inmap = self.std_map()
|
|
inmap['schema sd'].update({'table t1': {
|
|
'columns': [{'c1': {'type': 'integer'}}, {'c2': {'type': 'text'}}],
|
|
'indexes': {'t1_idx': {'keys': [{
|
|
'lower(c2)': {'type': 'expression'}}]}}}})
|
|
sql = self.to_sql(inmap)
|
|
assert sql[1] == "CREATE INDEX t1_idx ON sd.t1 (lower(c2))"
|
|
|
|
def test_create_index_col_opts(self):
|
|
"Create table and an index with column options"
|
|
inmap = self.std_map()
|
|
inmap['schema sd'].update({'table t1': {
|
|
'columns': [{'c1': {'type': 'cidr'}}, {'c2': {'type': 'text'}}],
|
|
'indexes': {'t1_idx': {'keys': [{'c1': {
|
|
'opclass': 'cidr_ops', 'nulls': 'first'}}, 'c2']}}}})
|
|
sql = self.to_sql(inmap)
|
|
assert sql[1] == "CREATE INDEX t1_idx ON sd.t1 " \
|
|
"(c1 cidr_ops NULLS FIRST, c2)"
|
|
|
|
def test_create_index_mixed(self):
|
|
"Create indexes using functions, a regular column and expressions"
|
|
stmts = ["CREATE TABLE t1 (c1 integer, c2 text, c3 text)"]
|
|
inmap = self.std_map()
|
|
inmap['schema sd'].update({'table t1': {
|
|
'columns': [{'c1': {'type': 'integer'}}, {'c2': {'type': 'text'}},
|
|
{'c3': {'type': 'text'}}],
|
|
'indexes': {'t1_idx': {'keys': [
|
|
{"btrim(c3, 'x'::text)": {'type': 'expression',
|
|
'nulls': 'first'}}, 'c1',
|
|
{'lower(c2)': {'type': 'expression', 'order': 'desc'}}]},
|
|
't1_idx2': {'keys': [
|
|
{"(((c2 || ', '::text) || c3))": {'type': 'expression'}},
|
|
{"(((c3 || ' '::text) || c2))": {
|
|
'type': 'expression'}}]}}}})
|
|
sql = sorted(self.to_sql(inmap, stmts))
|
|
assert sql[0] == "CREATE INDEX t1_idx ON sd.t1 (" \
|
|
"btrim(c3, 'x'::text) NULLS FIRST, c1, lower(c2) DESC)"
|
|
assert sql[1] == "CREATE INDEX t1_idx2 ON sd.t1 (" \
|
|
"(((c2 || ', '::text) || c3)), (((c3 || ' '::text) || c2)))"
|
|
|
|
def test_create_table_with_index_clustered(self):
|
|
"Create new table clustered on a single column index"
|
|
inmap = self.std_map()
|
|
inmap['schema sd'].update({'table t1': {
|
|
'columns': [{'c1': {'type': 'integer'}},
|
|
{'c2': {'type': 'text'}}],
|
|
'indexes': {'t1_idx': {'keys': ['c1'], 'cluster': True}}}})
|
|
sql = self.to_sql(inmap)
|
|
assert sql[2] == "CLUSTER sd.t1 USING t1_idx"
|
|
|
|
def test_cluster_table_with_index(self):
|
|
"Change a table with an index to cluster on it"
|
|
stmts = [CREATE_TABLE_STMT, CREATE_STMT]
|
|
inmap = self.std_map()
|
|
inmap['schema sd'].update({'table t1': {
|
|
'columns': [{'c1': {'type': 'integer'}}, {'c2': {'type': 'text'}}],
|
|
'indexes': {'t1_idx': {'keys': ['c1'], 'cluster': True}}}})
|
|
sql = self.to_sql(inmap, stmts)
|
|
assert sql[0] == "CLUSTER sd.t1 USING t1_idx"
|
|
|
|
def test_uncluster_table_with_index(self):
|
|
"Change a table clustered on an index to remove cluster"
|
|
stmts = [CREATE_TABLE_STMT, CREATE_STMT, "CLUSTER t1 USING t1_idx"]
|
|
inmap = self.std_map()
|
|
inmap['schema sd'].update({'table t1': {
|
|
'columns': [{'c1': {'type': 'integer'}}, {'c2': {'type': 'text'}}],
|
|
'indexes': {'t1_idx': {'keys': ['c1']}}}})
|
|
sql = self.to_sql(inmap, stmts)
|
|
assert fix_indent(sql[0]) == "ALTER TABLE sd.t1 SET WITHOUT CLUSTER"
|
|
|
|
def test_comment_on_index(self):
|
|
"Create a comment for an existing index"
|
|
stmts = [CREATE_TABLE_STMT, CREATE_STMT]
|
|
inmap = self.std_map()
|
|
inmap['schema sd'].update({'table t1': {
|
|
'columns': [{'c1': {'type': 'integer'}}, {'c2': {'type': 'text'}}],
|
|
'indexes': {'t1_idx': {'keys': ['c1'],
|
|
'description': 'Test index t1_idx'}}}})
|
|
sql = self.to_sql(inmap, stmts)
|
|
assert sql == [COMMENT_STMT]
|
|
|
|
def test_drop_index_comment(self):
|
|
"Drop the comment on an existing index"
|
|
stmts = [CREATE_TABLE_STMT, CREATE_STMT, COMMENT_STMT]
|
|
inmap = self.std_map()
|
|
inmap['schema sd'].update({'table t1': {
|
|
'columns': [{'c1': {'type': 'integer'}}, {'c2': {'type': 'text'}}],
|
|
'indexes': {'t1_idx': {'keys': ['c1']}}}})
|
|
sql = self.to_sql(inmap, stmts)
|
|
assert sql == ["COMMENT ON INDEX sd.t1_idx IS NULL"]
|
|
|
|
def test_change_index_comment(self):
|
|
"Change existing comment on an index"
|
|
stmts = [CREATE_TABLE_STMT, CREATE_STMT, COMMENT_STMT]
|
|
inmap = self.std_map()
|
|
inmap['schema sd'].update({'table t1': {
|
|
'columns': [{'c1': {'type': 'integer'}}, {'c2': {'type': 'text'}}],
|
|
'indexes': {'t1_idx': {'keys': ['c1'],
|
|
'description': 'Changed index t1_idx'}}}})
|
|
sql = self.to_sql(inmap, stmts)
|
|
assert sql == ["COMMENT ON INDEX sd.t1_idx IS 'Changed index t1_idx'"]
|
|
|
|
def test_change_index_keys(self):
|
|
"Change keys of an existing index"
|
|
stmts = [CREATE_TABLE_STMT, CREATE_STMT]
|
|
inmap = self.std_map()
|
|
inmap['schema sd'].update({'table t1': {
|
|
'columns': [{'c1': {'type': 'integer'}}, {'c2': {'type': 'text'}}],
|
|
'indexes': {'t1_idx': {'keys': ['btrim(c1)']}}}})
|
|
sql = self.to_sql(inmap, stmts)
|
|
assert sql == ["DROP INDEX sd.t1_idx",
|
|
"CREATE INDEX t1_idx ON sd.t1 (btrim(c1))"]
|
|
|
|
def test_change_order_index_keys(self):
|
|
"Change keys order of an existing index"
|
|
stmts = [CREATE_TABLE_STMT, "CREATE INDEX t1_idx ON t1 (c1, c2)"]
|
|
inmap = self.std_map()
|
|
inmap['schema sd'].update({'table t1': {
|
|
'columns': [{'c1': {'type': 'integer'}}, {'c2': {'type': 'text'}}],
|
|
'indexes': {'t1_idx': {'keys': ['c2', 'c1']}}}})
|
|
sql = self.to_sql(inmap, stmts)
|
|
assert sql == ["DROP INDEX sd.t1_idx",
|
|
"CREATE INDEX t1_idx ON sd.t1 (c2, c1)"]
|