polygon/example.py at main · polygon-sql/polygon · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
import logging

from polygon.environment import Environment
from polygon.logger import logger
from polygon.testers.mysql_tester import DB_CONFIG, MySQLTester


def main():
    logger.setLevel(logging.INFO)

    # specify the schema and constraints
    schema = [
        {
            "TableName": "Customers",
            "PKeys": [
                {
                    "Name": "customer_id",
                    "Type": "int"
                }
            ],
            "FKeys": [],
            "Others": [
                {
                    "Name": "customer_name",
                    "Type": "varchar"
                },
                {
                    "Name": "email",
                    "Type": "varchar"
                }
            ]
        },
        {
            "TableName": "Contacts",
            "PKeys": [
                {
                    "Name": "user_id",
                    "Type": "int"
                },
                {
                    "Name": "contact_email",
                    "Type": "varchar"
                }
            ],
            "FKeys": [],
            "Others": [
                {
                    "Name": "contact_name",
                    "Type": "varchar"
                }
            ]
        },
        {
            "TableName": "Invoices",
            "PKeys": [
                {
                    "Name": "invoice_id",
                    "Type": "int"
                }
            ],
            "FKeys": [
                {
                    "FName": "user_id",
                    "PName": "customer_id",
                    "PTable": "0"
                }
            ],
            "Others": [
                {
                    "Name": "price",
                    "Type": "int"
                }
            ]
        }
    ]
    constraints = [{'distinct': ['Customers.customer_id']},
                   {'distinct': ['Contacts.user_id', 'Contacts.contact_email']}, {'distinct': ['Invoices.invoice_id']},
                   {'gt': ['Invoices.price', 0]}, {'eq': ['Contacts.user_id', 'Customers.customer_id']},
                   {'eq': ['Invoices.user_id', 'Customers.customer_id']},
                   {'join': ['Customers.customer_name', 'Contacts.contact_name']},
                   {'join': ['Customers.email', 'Contacts.contact_email']}, {
                       'if': [{'col1a': 'Customers.customer_name'}, {'comp1': 'equal'},
                              {'col1b': 'Contacts.contact_name'}, {'col2a': 'Customers.email'}, {'comp2': 'equal'},
                              {'col2b': 'Contacts.contact_email'}]}]

    # create an environment
    env = Environment(schema, constraints, bound=2, time_budget=60)

    # specify the queries
    queries = [
        """
        SELECT INVOICE_ID, T.CUSTOMER_NAME, PRICE, CONTACTS_CNT, TRUSTED_CONTACTS_CNT
        FROM INVOICES I LEFT JOIN (
          SELECT A.CUSTOMER_ID, CUSTOMER_NAME, COUNT(CONTACT_NAME) AS CONTACTS_CNT,
                 SUM(CASE WHEN CONTACT_EMAIL IN (SELECT EMAIL FROM CUSTOMERS) THEN 1 ELSE 0 END) AS TRUSTED_CONTACTS_CNT
          FROM CUSTOMERS A LEFT JOIN CONTACTS B
          ON A.CUSTOMER_ID = B.USER_ID
          GROUP BY A.CUSTOMER_ID, CUSTOMER_NAME
          ) T
          ON I.USER_ID = T.CUSTOMER_ID
          ORDER BY 1
        """,
        """
        SELECT INVOICE_ID, C.CUSTOMER_NAME, PRICE, COUNT(T.CONTACT_NAME) CONTACTS_CNT, COUNT(C1.EMAIL) TRUSTED_CONTACTS_CNT
        FROM INVOICES I LEFT JOIN CUSTOMERS C
        ON I.USER_ID = C.CUSTOMER_ID
        LEFT JOIN CONTACTS T ON C.CUSTOMER_ID = T.USER_ID
        LEFT JOIN CUSTOMERS C1 ON T.CONTACT_EMAIL = C1.EMAIL
        GROUP BY 1,2,3
        ORDER BY INVOICE_ID
        """
    ]

    # run
    eq, cex, checking_time, total_time, ret = env.check(*queries, use_precise_encoding=False)
    print(f'Running time: {total_time}')
    print(f'Result: ', end='')

    if eq is None:
        print('ERR')
    else:
        if not eq:
            print('NEQ')
            logger.info(f'Counterexample: {cex}')
            with MySQLTester(DB_CONFIG, schema) as tester:
                tester.create_all_databases([cex], constraints)
                rejected = tester.test_pair(*queries)
                print('Refuted:', rejected)
        else:
            print('EQ')


if __name__ == '__main__':
    main()