We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
✨ enable SQLAlchemy logging
### Thanks for stopping by to let us know something could be better!

**Is your feature request related to a problem? Please describe.** _A clear and concise description of what the problem is. Ex. I'm always frustrated when [...]_

**Describe the solution you'd like** _A clear and concise description of what you want to happen._

**Describe alternatives you've considered** _A clear and concise description of any alternative solutions or features you've considered._

**Additional context** _Add any other context or screenshots about the feature request here._


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `opteryx/connectors/sql_connector.py`
Content:
```
1 # Licensed under the Apache License, Version 2.0 (the "License");
2 # you may not use this file except in compliance with the License.
3 # You may obtain a copy of the License at
4 #
5 #     http://www.apache.org/licenses/LICENSE-2.0
6 #
7 # Unless required by applicable law or agreed to in writing, software
8 # distributed under the License is distributed on an "AS IS" BASIS,
9 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
10 # See the License for the specific language governing permissions and
11 # limitations under the License.
12 
13 """
14 The SQL Connector downloads data from remote servers and converts them
15 to pyarrow tables so they can be processed as per any other data source.
16 """
17 import time
18 from decimal import Decimal
19 from typing import Any
20 from typing import Dict
21 from typing import Generator
22 from typing import Tuple
23 
24 import pyarrow
25 from orso import DataFrame
26 from orso.schema import ConstantColumn
27 from orso.schema import FlatColumn
28 from orso.schema import RelationSchema
29 from orso.tools import random_string
30 from orso.types import PYTHON_TO_ORSO_MAP
31 
32 from opteryx.connectors.base.base_connector import DEFAULT_MORSEL_SIZE
33 from opteryx.connectors.base.base_connector import INITIAL_CHUNK_SIZE
34 from opteryx.connectors.base.base_connector import MIN_CHUNK_SIZE
35 from opteryx.connectors.base.base_connector import BaseConnector
36 from opteryx.connectors.capabilities import PredicatePushable
37 from opteryx.exceptions import MissingDependencyError
38 from opteryx.exceptions import UnmetRequirementError
39 from opteryx.managers.expression import Node
40 from opteryx.managers.expression import NodeType
41 from opteryx.third_party.query_builder import Query
42 
43 
44 def _handle_operand(operand: Node, parameters: dict) -> Tuple[Any, dict]:
45     if operand.node_type == NodeType.IDENTIFIER:
46         return operand.source_column, parameters
47 
48     literal = operand.value
49     if hasattr(literal, "item"):
50         literal = literal.item()
51 
52     name = random_string(8)
53     parameters[name] = literal
54     return f":{name}", parameters
55 
56 
57 class SqlConnector(BaseConnector, PredicatePushable):
58     __mode__ = "Sql"
59 
60     PUSHABLE_OPS: Dict[str, bool] = {
61         "Eq": True,
62         "NotEq": True,
63         "Gt": True,
64         "GtEq": True,
65         "Lt": True,
66         "LtEq": True,
67         "Like": True,
68         "NotLike": True,
69     }
70 
71     OPS_XLAT: Dict[str, str] = {
72         "Eq": "=",
73         "NotEq": "!=",
74         "Gt": ">",
75         "GtEq": ">=",
76         "Lt": "<",
77         "LtEq": "<=",
78         "Like": "LIKE",
79         "NotLike": "NOT LIKE",
80         "IsTrue": "IS TRUE",
81         "IsNotTrue": "IS NOT TRUE",
82         "IsFalse": "IS FALSE",
83         "IsNotFalse": "IS NOT FALSE",
84         "IsNull": "IS NULL",
85         "IsNotNull": "IS NOT NULL",
86     }
87 
88     def __init__(self, *args, connection: str = None, engine=None, **kwargs):
89         BaseConnector.__init__(self, **kwargs)
90         PredicatePushable.__init__(self, **kwargs)
91 
92         try:
93             from sqlalchemy import MetaData
94             from sqlalchemy import create_engine
95             from sqlalchemy.pool import NullPool
96         except ImportError as err:  # pragma: nocover
97             raise MissingDependencyError(err.name) from err
98 
99         if engine is None and connection is None:  # pragma: no cover
100             raise UnmetRequirementError(
101                 "SQL Connections require either a SQL Alchemy connection string in the 'connection' parameter, or a SQL Alchemy Engine in the 'engine' parameter."
102             )
103 
104         # create the SqlAlchemy engine
105         if engine is None:
106             self._engine = create_engine(connection, poolclass=NullPool)
107         else:
108             self._engine = engine
109 
110         self.schema = None  # type: ignore
111         self.metadata = MetaData()
112 
113     def can_push(self, operator: Node, types: set = None) -> bool:
114         if super().can_push(operator, types):
115             return True
116         return operator.condition.node_type == NodeType.UNARY_OPERATOR
117 
118     def read_dataset(  # type:ignore
119         self,
120         *,
121         columns: list = None,
122         predicates: list = None,
123         chunk_size: int = INITIAL_CHUNK_SIZE,  # type:ignore
124     ) -> Generator[pyarrow.Table, None, None]:  # type:ignore
125         from sqlalchemy.sql import text
126 
127         self.chunk_size = chunk_size
128         result_schema = self.schema
129 
130         query_builder = Query().FROM(self.dataset)
131 
132         # Update the SQL and the target morsel schema if we've pushed a projection
133         if columns:
134             column_names = [col.name for col in columns]
135             query_builder.add("SELECT", *column_names)
136             result_schema.columns = [  # type:ignore
137                 col for col in self.schema.columns if col.name in column_names  # type:ignore
138             ]
139         elif self.schema.columns:  # type:ignore
140             query_builder.add("SELECT", "*")
141         else:
142             query_builder.add("SELECT", "1")
143             self.schema.columns.append(ConstantColumn(name="1", value=1))  # type:ignore
144 
145         # Update SQL if we've pushed predicates
146         parameters: dict = {}
147         for predicate in predicates:
148             if predicate.node_type == NodeType.UNARY_OPERATOR:
149                 operand = predicate.centre.current_name
150                 operator = self.OPS_XLAT[predicate.value]
151 
152                 query_builder.WHERE(f"{operand} {operator}")
153             else:
154                 left_operand = predicate.left
155                 right_operand = predicate.right
156                 operator = self.OPS_XLAT[predicate.value]
157 
158                 left_value, parameters = _handle_operand(left_operand, parameters)
159                 right_value, parameters = _handle_operand(right_operand, parameters)
160 
161                 query_builder.WHERE(f"{left_value} {operator} {right_value}")
162 
163         at_least_once = False
164 
165         convert_time = 0.0
166 
167         with self._engine.connect() as conn:
168             # DEBUG: log ("READ DATASET\n", str(query_builder))
169             # DEBUG: log ("PARAMETERS\n", parameters)
170             # Execution Options allows us to handle datasets larger than memory
171             result = conn.execution_options(stream_results=True, max_row_buffer=10000).execute(
172                 text(str(query_builder)), parameters=parameters
173             )
174 
175             while True:
176                 batch_rows = result.fetchmany(self.chunk_size)
177                 if not batch_rows:
178                     break
179 
180                 # convert the SqlAlchemy Results to Arrow using Orso
181                 b = time.monotonic_ns()
182                 morsel = DataFrame(schema=result_schema, rows=batch_rows).arrow()
183                 convert_time += time.monotonic_ns() - b
184                 yield morsel
185                 at_least_once = True
186 
187                 # Dynamically adjust chunk size based on the data size, we start by downloading
188                 # 500 records to get an idea of the row size, assuming these 500 are
189                 # representative, we work out how many rows fit into 16Mb (check setting).
190                 # Don't keep recalculating, this is not a cheap operation and it's predicting
191                 # the future so isn't going to ever be 100% correct
192                 if self.chunk_size == INITIAL_CHUNK_SIZE and morsel.nbytes > 0:
193                     self.chunk_size = int(len(morsel) // (morsel.nbytes / DEFAULT_MORSEL_SIZE)) + 1
194                     self.chunk_size = (self.chunk_size // MIN_CHUNK_SIZE) * MIN_CHUNK_SIZE
195                     self.chunk_size = max(self.chunk_size, MIN_CHUNK_SIZE)
196                     # DEBUG: log (f"CHANGING CHUNK SIZE TO {self.chunk_size} was {INITIAL_CHUNK_SIZE}.")
197 
198         if not at_least_once:
199             yield DataFrame(schema=result_schema).arrow()
200 
201         # DEBUG: log (f"time spent converting: {convert_time/1e9}s")
202 
203     def get_dataset_schema(self) -> RelationSchema:
204         from sqlalchemy import Table
205 
206         if self.schema:
207             return self.schema
208 
209         # Try to read the schema from the metastore
210         self.schema = self.read_schema_from_metastore()
211         if self.schema:
212             return self.schema
213 
214         # get the schema from the dataset
215         # DEBUG: log ("GET SQL SCHEMA:", self.dataset)
216         try:
217             table = Table(self.dataset, self.metadata, autoload_with=self._engine)
218 
219             self.schema = RelationSchema(
220                 name=table.name,
221                 columns=[
222                     FlatColumn(
223                         name=column.name,
224                         type=PYTHON_TO_ORSO_MAP[column.type.python_type],
225                         precision=(
226                             None
227                             if column.type.python_type != Decimal
228                             else column.type.precision  # type:ignore
229                         ),
230                         scale=(
231                             None
232                             if column.type.python_type != Decimal
233                             else column.type.scale  # type:ignore
234                         ),
235                         nullable=column.nullable,
236                     )
237                     for column in table.columns
238                 ],
239             )
240         except Exception as err:
241             # Fall back to getting the schema from the first row, this is the column names, and where
242             # possible, column types.
243             # DEBUG: log (f"APPROXIMATING SCHEMA OF {self.dataset} BECAUSE OF {type(err).__name__}({err})")
244             from sqlalchemy.sql import text
245 
246             with self._engine.connect() as conn:
247                 query = Query().SELECT("*").FROM(self.dataset).LIMIT("1")
248                 # DEBUG: log ("READ ROW\n", str(query))
249                 row = conn.execute(text(str(query))).fetchone()._asdict()
250                 # DEBUG: log ("ROW:", row)
251                 self.schema = RelationSchema(
252                     name=self.dataset,
253                     columns=[
254                         FlatColumn(
255                             name=column,
256                             type=0 if value is None else PYTHON_TO_ORSO_MAP[type(value)],
257                         )
258                         for column, value in row.items()
259                     ],
260                 )
261 
262         return self.schema
263
```

Path: `opteryx/__version__.py`
Content:
```
1 __build__ = 340
2 
3 # Licensed under the Apache License, Version 2.0 (the "License");
4 # you may not use this file except in compliance with the License.
5 # You may obtain a copy of the License at
6 #
7 #     http://www.apache.org/licenses/LICENSE-2.0
8 #
9 # Unless required by applicable law or agreed to in writing, software
10 # distributed under the License is distributed on an "AS IS" BASIS,
11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 # See the License for the specific language governing permissions and
13 # limitations under the License.
14 
15 """
16 Store the version here so:
17 1) we don't load dependencies by storing it in __init__.py
18 2) we can import it in setup.py for the same reason
19 """
20 from enum import Enum  # isort: skip
21 
22 
23 class VersionStatus(Enum):
24     ALPHA = "alpha"
25     BETA = "beta"
26     RELEASE = "release"
27 
28 
29 _major = 0
30 _minor = 13
31 _revision = 4
32 _status = VersionStatus.ALPHA
33 
34 __author__ = "@joocer"
35 __version__ = f"{_major}.{_minor}.{_revision}" + (
36     f"-{_status.value}.{__build__}" if _status != VersionStatus.RELEASE else ""
37 )
38
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/opteryx/__version__.py b/opteryx/__version__.py
--- a/opteryx/__version__.py
+++ b/opteryx/__version__.py
@@ -1,4 +1,4 @@
-__build__ = 340
+__build__ = 341
 
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
diff --git a/opteryx/connectors/sql_connector.py b/opteryx/connectors/sql_connector.py
--- a/opteryx/connectors/sql_connector.py
+++ b/opteryx/connectors/sql_connector.py
@@ -14,6 +14,7 @@
 The SQL Connector downloads data from remote servers and converts them
 to pyarrow tables so they can be processed as per any other data source.
 """
+import os
 import time
 from decimal import Decimal
 from typing import Any
@@ -40,6 +41,8 @@
 from opteryx.managers.expression import NodeType
 from opteryx.third_party.query_builder import Query
 
+DEBUG_ENABLED = os.environ.get("OPTERYX_DEBUG") is not None
+
 
 def _handle_operand(operand: Node, parameters: dict) -> Tuple[Any, dict]:
     if operand.node_type == NodeType.IDENTIFIER:
@@ -103,7 +106,7 @@
 
         # create the SqlAlchemy engine
         if engine is None:
-            self._engine = create_engine(connection, poolclass=NullPool)
+            self._engine = create_engine(connection, poolclass=NullPool, echo=DEBUG_ENABLED)
         else:
             self._engine = engine
 
@@ -168,7 +171,7 @@
             # DEBUG: log ("READ DATASET\n", str(query_builder))
             # DEBUG: log ("PARAMETERS\n", parameters)
             # Execution Options allows us to handle datasets larger than memory
-            result = conn.execution_options(stream_results=True, max_row_buffer=10000).execute(
+            result = conn.execution_options(stream_results=True, max_row_buffer=25000).execute(
                 text(str(query_builder)), parameters=parameters
             )

{"golden_diff": "diff --git a/opteryx/__version__.py b/opteryx/__version__.py\n--- a/opteryx/__version__.py\n+++ b/opteryx/__version__.py\n@@ -1,4 +1,4 @@\n-__build__ = 340\n+__build__ = 341\n \n # Licensed under the Apache License, Version 2.0 (the \"License\");\n # you may not use this file except in compliance with the License.\ndiff --git a/opteryx/connectors/sql_connector.py b/opteryx/connectors/sql_connector.py\n--- a/opteryx/connectors/sql_connector.py\n+++ b/opteryx/connectors/sql_connector.py\n@@ -14,6 +14,7 @@\n The SQL Connector downloads data from remote servers and converts them\n to pyarrow tables so they can be processed as per any other data source.\n \"\"\"\n+import os\n import time\n from decimal import Decimal\n from typing import Any\n@@ -40,6 +41,8 @@\n from opteryx.managers.expression import NodeType\n from opteryx.third_party.query_builder import Query\n \n+DEBUG_ENABLED = os.environ.get(\"OPTERYX_DEBUG\") is not None\n+\n \n def _handle_operand(operand: Node, parameters: dict) -> Tuple[Any, dict]:\n     if operand.node_type == NodeType.IDENTIFIER:\n@@ -103,7 +106,7 @@\n \n         # create the SqlAlchemy engine\n         if engine is None:\n-            self._engine = create_engine(connection, poolclass=NullPool)\n+            self._engine = create_engine(connection, poolclass=NullPool, echo=DEBUG_ENABLED)\n         else:\n             self._engine = engine\n \n@@ -168,7 +171,7 @@\n             # DEBUG: log (\"READ DATASET\\n\", str(query_builder))\n             # DEBUG: log (\"PARAMETERS\\n\", parameters)\n             # Execution Options allows us to handle datasets larger than memory\n-            result = conn.execution_options(stream_results=True, max_row_buffer=10000).execute(\n+            result = conn.execution_options(stream_results=True, max_row_buffer=25000).execute(\n                 text(str(query_builder)), parameters=parameters\n             )\n", "issue": "\u2728 enable SQLAlchemy logging\n### Thanks for stopping by to let us know something could be better!\r\n\r\n**Is your feature request related to a problem? Please describe.** _A clear and concise description of what the problem is. Ex. I'm always frustrated when [...]_\r\n\r\n**Describe the solution you'd like** _A clear and concise description of what you want to happen._\r\n\r\n**Describe alternatives you've considered** _A clear and concise description of any alternative solutions or features you've considered._\r\n\r\n**Additional context** _Add any other context or screenshots about the feature request here._\r\n\n", "before_files": [{"content": "# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#     http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\n\"\"\"\nThe SQL Connector downloads data from remote servers and converts them\nto pyarrow tables so they can be processed as per any other data source.\n\"\"\"\nimport time\nfrom decimal import Decimal\nfrom typing import Any\nfrom typing import Dict\nfrom typing import Generator\nfrom typing import Tuple\n\nimport pyarrow\nfrom orso import DataFrame\nfrom orso.schema import ConstantColumn\nfrom orso.schema import FlatColumn\nfrom orso.schema import RelationSchema\nfrom orso.tools import random_string\nfrom orso.types import PYTHON_TO_ORSO_MAP\n\nfrom opteryx.connectors.base.base_connector import DEFAULT_MORSEL_SIZE\nfrom opteryx.connectors.base.base_connector import INITIAL_CHUNK_SIZE\nfrom opteryx.connectors.base.base_connector import MIN_CHUNK_SIZE\nfrom opteryx.connectors.base.base_connector import BaseConnector\nfrom opteryx.connectors.capabilities import PredicatePushable\nfrom opteryx.exceptions import MissingDependencyError\nfrom opteryx.exceptions import UnmetRequirementError\nfrom opteryx.managers.expression import Node\nfrom opteryx.managers.expression import NodeType\nfrom opteryx.third_party.query_builder import Query\n\n\ndef _handle_operand(operand: Node, parameters: dict) -> Tuple[Any, dict]:\n    if operand.node_type == NodeType.IDENTIFIER:\n        return operand.source_column, parameters\n\n    literal = operand.value\n    if hasattr(literal, \"item\"):\n        literal = literal.item()\n\n    name = random_string(8)\n    parameters[name] = literal\n    return f\":{name}\", parameters\n\n\nclass SqlConnector(BaseConnector, PredicatePushable):\n    __mode__ = \"Sql\"\n\n    PUSHABLE_OPS: Dict[str, bool] = {\n        \"Eq\": True,\n        \"NotEq\": True,\n        \"Gt\": True,\n        \"GtEq\": True,\n        \"Lt\": True,\n        \"LtEq\": True,\n        \"Like\": True,\n        \"NotLike\": True,\n    }\n\n    OPS_XLAT: Dict[str, str] = {\n        \"Eq\": \"=\",\n        \"NotEq\": \"!=\",\n        \"Gt\": \">\",\n        \"GtEq\": \">=\",\n        \"Lt\": \"<\",\n        \"LtEq\": \"<=\",\n        \"Like\": \"LIKE\",\n        \"NotLike\": \"NOT LIKE\",\n        \"IsTrue\": \"IS TRUE\",\n        \"IsNotTrue\": \"IS NOT TRUE\",\n        \"IsFalse\": \"IS FALSE\",\n        \"IsNotFalse\": \"IS NOT FALSE\",\n        \"IsNull\": \"IS NULL\",\n        \"IsNotNull\": \"IS NOT NULL\",\n    }\n\n    def __init__(self, *args, connection: str = None, engine=None, **kwargs):\n        BaseConnector.__init__(self, **kwargs)\n        PredicatePushable.__init__(self, **kwargs)\n\n        try:\n            from sqlalchemy import MetaData\n            from sqlalchemy import create_engine\n            from sqlalchemy.pool import NullPool\n        except ImportError as err:  # pragma: nocover\n            raise MissingDependencyError(err.name) from err\n\n        if engine is None and connection is None:  # pragma: no cover\n            raise UnmetRequirementError(\n                \"SQL Connections require either a SQL Alchemy connection string in the 'connection' parameter, or a SQL Alchemy Engine in the 'engine' parameter.\"\n            )\n\n        # create the SqlAlchemy engine\n        if engine is None:\n            self._engine = create_engine(connection, poolclass=NullPool)\n        else:\n            self._engine = engine\n\n        self.schema = None  # type: ignore\n        self.metadata = MetaData()\n\n    def can_push(self, operator: Node, types: set = None) -> bool:\n        if super().can_push(operator, types):\n            return True\n        return operator.condition.node_type == NodeType.UNARY_OPERATOR\n\n    def read_dataset(  # type:ignore\n        self,\n        *,\n        columns: list = None,\n        predicates: list = None,\n        chunk_size: int = INITIAL_CHUNK_SIZE,  # type:ignore\n    ) -> Generator[pyarrow.Table, None, None]:  # type:ignore\n        from sqlalchemy.sql import text\n\n        self.chunk_size = chunk_size\n        result_schema = self.schema\n\n        query_builder = Query().FROM(self.dataset)\n\n        # Update the SQL and the target morsel schema if we've pushed a projection\n        if columns:\n            column_names = [col.name for col in columns]\n            query_builder.add(\"SELECT\", *column_names)\n            result_schema.columns = [  # type:ignore\n                col for col in self.schema.columns if col.name in column_names  # type:ignore\n            ]\n        elif self.schema.columns:  # type:ignore\n            query_builder.add(\"SELECT\", \"*\")\n        else:\n            query_builder.add(\"SELECT\", \"1\")\n            self.schema.columns.append(ConstantColumn(name=\"1\", value=1))  # type:ignore\n\n        # Update SQL if we've pushed predicates\n        parameters: dict = {}\n        for predicate in predicates:\n            if predicate.node_type == NodeType.UNARY_OPERATOR:\n                operand = predicate.centre.current_name\n                operator = self.OPS_XLAT[predicate.value]\n\n                query_builder.WHERE(f\"{operand} {operator}\")\n            else:\n                left_operand = predicate.left\n                right_operand = predicate.right\n                operator = self.OPS_XLAT[predicate.value]\n\n                left_value, parameters = _handle_operand(left_operand, parameters)\n                right_value, parameters = _handle_operand(right_operand, parameters)\n\n                query_builder.WHERE(f\"{left_value} {operator} {right_value}\")\n\n        at_least_once = False\n\n        convert_time = 0.0\n\n        with self._engine.connect() as conn:\n            # DEBUG: log (\"READ DATASET\\n\", str(query_builder))\n            # DEBUG: log (\"PARAMETERS\\n\", parameters)\n            # Execution Options allows us to handle datasets larger than memory\n            result = conn.execution_options(stream_results=True, max_row_buffer=10000).execute(\n                text(str(query_builder)), parameters=parameters\n            )\n\n            while True:\n                batch_rows = result.fetchmany(self.chunk_size)\n                if not batch_rows:\n                    break\n\n                # convert the SqlAlchemy Results to Arrow using Orso\n                b = time.monotonic_ns()\n                morsel = DataFrame(schema=result_schema, rows=batch_rows).arrow()\n                convert_time += time.monotonic_ns() - b\n                yield morsel\n                at_least_once = True\n\n                # Dynamically adjust chunk size based on the data size, we start by downloading\n                # 500 records to get an idea of the row size, assuming these 500 are\n                # representative, we work out how many rows fit into 16Mb (check setting).\n                # Don't keep recalculating, this is not a cheap operation and it's predicting\n                # the future so isn't going to ever be 100% correct\n                if self.chunk_size == INITIAL_CHUNK_SIZE and morsel.nbytes > 0:\n                    self.chunk_size = int(len(morsel) // (morsel.nbytes / DEFAULT_MORSEL_SIZE)) + 1\n                    self.chunk_size = (self.chunk_size // MIN_CHUNK_SIZE) * MIN_CHUNK_SIZE\n                    self.chunk_size = max(self.chunk_size, MIN_CHUNK_SIZE)\n                    # DEBUG: log (f\"CHANGING CHUNK SIZE TO {self.chunk_size} was {INITIAL_CHUNK_SIZE}.\")\n\n        if not at_least_once:\n            yield DataFrame(schema=result_schema).arrow()\n\n        # DEBUG: log (f\"time spent converting: {convert_time/1e9}s\")\n\n    def get_dataset_schema(self) -> RelationSchema:\n        from sqlalchemy import Table\n\n        if self.schema:\n            return self.schema\n\n        # Try to read the schema from the metastore\n        self.schema = self.read_schema_from_metastore()\n        if self.schema:\n            return self.schema\n\n        # get the schema from the dataset\n        # DEBUG: log (\"GET SQL SCHEMA:\", self.dataset)\n        try:\n            table = Table(self.dataset, self.metadata, autoload_with=self._engine)\n\n            self.schema = RelationSchema(\n                name=table.name,\n                columns=[\n                    FlatColumn(\n                        name=column.name,\n                        type=PYTHON_TO_ORSO_MAP[column.type.python_type],\n                        precision=(\n                            None\n                            if column.type.python_type != Decimal\n                            else column.type.precision  # type:ignore\n                        ),\n                        scale=(\n                            None\n                            if column.type.python_type != Decimal\n                            else column.type.scale  # type:ignore\n                        ),\n                        nullable=column.nullable,\n                    )\n                    for column in table.columns\n                ],\n            )\n        except Exception as err:\n            # Fall back to getting the schema from the first row, this is the column names, and where\n            # possible, column types.\n            # DEBUG: log (f\"APPROXIMATING SCHEMA OF {self.dataset} BECAUSE OF {type(err).__name__}({err})\")\n            from sqlalchemy.sql import text\n\n            with self._engine.connect() as conn:\n                query = Query().SELECT(\"*\").FROM(self.dataset).LIMIT(\"1\")\n                # DEBUG: log (\"READ ROW\\n\", str(query))\n                row = conn.execute(text(str(query))).fetchone()._asdict()\n                # DEBUG: log (\"ROW:\", row)\n                self.schema = RelationSchema(\n                    name=self.dataset,\n                    columns=[\n                        FlatColumn(\n                            name=column,\n                            type=0 if value is None else PYTHON_TO_ORSO_MAP[type(value)],\n                        )\n                        for column, value in row.items()\n                    ],\n                )\n\n        return self.schema\n", "path": "opteryx/connectors/sql_connector.py"}, {"content": "__build__ = 340\n\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#     http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\n\"\"\"\nStore the version here so:\n1) we don't load dependencies by storing it in __init__.py\n2) we can import it in setup.py for the same reason\n\"\"\"\nfrom enum import Enum  # isort: skip\n\n\nclass VersionStatus(Enum):\n    ALPHA = \"alpha\"\n    BETA = \"beta\"\n    RELEASE = \"release\"\n\n\n_major = 0\n_minor = 13\n_revision = 4\n_status = VersionStatus.ALPHA\n\n__author__ = \"@joocer\"\n__version__ = f\"{_major}.{_minor}.{_revision}\" + (\n    f\"-{_status.value}.{__build__}\" if _status != VersionStatus.RELEASE else \"\"\n)\n", "path": "opteryx/__version__.py"}], "after_files": [{"content": "# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#     http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\n\"\"\"\nThe SQL Connector downloads data from remote servers and converts them\nto pyarrow tables so they can be processed as per any other data source.\n\"\"\"\nimport os\nimport time\nfrom decimal import Decimal\nfrom typing import Any\nfrom typing import Dict\nfrom typing import Generator\nfrom typing import Tuple\n\nimport pyarrow\nfrom orso import DataFrame\nfrom orso.schema import ConstantColumn\nfrom orso.schema import FlatColumn\nfrom orso.schema import RelationSchema\nfrom orso.tools import random_string\nfrom orso.types import PYTHON_TO_ORSO_MAP\n\nfrom opteryx.connectors.base.base_connector import DEFAULT_MORSEL_SIZE\nfrom opteryx.connectors.base.base_connector import INITIAL_CHUNK_SIZE\nfrom opteryx.connectors.base.base_connector import MIN_CHUNK_SIZE\nfrom opteryx.connectors.base.base_connector import BaseConnector\nfrom opteryx.connectors.capabilities import PredicatePushable\nfrom opteryx.exceptions import MissingDependencyError\nfrom opteryx.exceptions import UnmetRequirementError\nfrom opteryx.managers.expression import Node\nfrom opteryx.managers.expression import NodeType\nfrom opteryx.third_party.query_builder import Query\n\nDEBUG_ENABLED = os.environ.get(\"OPTERYX_DEBUG\") is not None\n\n\ndef _handle_operand(operand: Node, parameters: dict) -> Tuple[Any, dict]:\n    if operand.node_type == NodeType.IDENTIFIER:\n        return operand.source_column, parameters\n\n    literal = operand.value\n    if hasattr(literal, \"item\"):\n        literal = literal.item()\n\n    name = random_string(8)\n    parameters[name] = literal\n    return f\":{name}\", parameters\n\n\nclass SqlConnector(BaseConnector, PredicatePushable):\n    __mode__ = \"Sql\"\n\n    PUSHABLE_OPS: Dict[str, bool] = {\n        \"Eq\": True,\n        \"NotEq\": True,\n        \"Gt\": True,\n        \"GtEq\": True,\n        \"Lt\": True,\n        \"LtEq\": True,\n        \"Like\": True,\n        \"NotLike\": True,\n    }\n\n    OPS_XLAT: Dict[str, str] = {\n        \"Eq\": \"=\",\n        \"NotEq\": \"!=\",\n        \"Gt\": \">\",\n        \"GtEq\": \">=\",\n        \"Lt\": \"<\",\n        \"LtEq\": \"<=\",\n        \"Like\": \"LIKE\",\n        \"NotLike\": \"NOT LIKE\",\n        \"IsTrue\": \"IS TRUE\",\n        \"IsNotTrue\": \"IS NOT TRUE\",\n        \"IsFalse\": \"IS FALSE\",\n        \"IsNotFalse\": \"IS NOT FALSE\",\n        \"IsNull\": \"IS NULL\",\n        \"IsNotNull\": \"IS NOT NULL\",\n    }\n\n    def __init__(self, *args, connection: str = None, engine=None, **kwargs):\n        BaseConnector.__init__(self, **kwargs)\n        PredicatePushable.__init__(self, **kwargs)\n\n        try:\n            from sqlalchemy import MetaData\n            from sqlalchemy import create_engine\n            from sqlalchemy.pool import NullPool\n        except ImportError as err:  # pragma: nocover\n            raise MissingDependencyError(err.name) from err\n\n        if engine is None and connection is None:  # pragma: no cover\n            raise UnmetRequirementError(\n                \"SQL Connections require either a SQL Alchemy connection string in the 'connection' parameter, or a SQL Alchemy Engine in the 'engine' parameter.\"\n            )\n\n        # create the SqlAlchemy engine\n        if engine is None:\n            self._engine = create_engine(connection, poolclass=NullPool, echo=DEBUG_ENABLED)\n        else:\n            self._engine = engine\n\n        self.schema = None  # type: ignore\n        self.metadata = MetaData()\n\n    def can_push(self, operator: Node, types: set = None) -> bool:\n        if super().can_push(operator, types):\n            return True\n        return operator.condition.node_type == NodeType.UNARY_OPERATOR\n\n    def read_dataset(  # type:ignore\n        self,\n        *,\n        columns: list = None,\n        predicates: list = None,\n        chunk_size: int = INITIAL_CHUNK_SIZE,  # type:ignore\n    ) -> Generator[pyarrow.Table, None, None]:  # type:ignore\n        from sqlalchemy.sql import text\n\n        self.chunk_size = chunk_size\n        result_schema = self.schema\n\n        query_builder = Query().FROM(self.dataset)\n\n        # Update the SQL and the target morsel schema if we've pushed a projection\n        if columns:\n            column_names = [col.name for col in columns]\n            query_builder.add(\"SELECT\", *column_names)\n            result_schema.columns = [  # type:ignore\n                col for col in self.schema.columns if col.name in column_names  # type:ignore\n            ]\n        elif self.schema.columns:  # type:ignore\n            query_builder.add(\"SELECT\", \"*\")\n        else:\n            query_builder.add(\"SELECT\", \"1\")\n            self.schema.columns.append(ConstantColumn(name=\"1\", value=1))  # type:ignore\n\n        # Update SQL if we've pushed predicates\n        parameters: dict = {}\n        for predicate in predicates:\n            if predicate.node_type == NodeType.UNARY_OPERATOR:\n                operand = predicate.centre.current_name\n                operator = self.OPS_XLAT[predicate.value]\n\n                query_builder.WHERE(f\"{operand} {operator}\")\n            else:\n                left_operand = predicate.left\n                right_operand = predicate.right\n                operator = self.OPS_XLAT[predicate.value]\n\n                left_value, parameters = _handle_operand(left_operand, parameters)\n                right_value, parameters = _handle_operand(right_operand, parameters)\n\n                query_builder.WHERE(f\"{left_value} {operator} {right_value}\")\n\n        at_least_once = False\n\n        convert_time = 0.0\n\n        with self._engine.connect() as conn:\n            # DEBUG: log (\"READ DATASET\\n\", str(query_builder))\n            # DEBUG: log (\"PARAMETERS\\n\", parameters)\n            # Execution Options allows us to handle datasets larger than memory\n            result = conn.execution_options(stream_results=True, max_row_buffer=25000).execute(\n                text(str(query_builder)), parameters=parameters\n            )\n\n            while True:\n                batch_rows = result.fetchmany(self.chunk_size)\n                if not batch_rows:\n                    break\n\n                # convert the SqlAlchemy Results to Arrow using Orso\n                b = time.monotonic_ns()\n                morsel = DataFrame(schema=result_schema, rows=batch_rows).arrow()\n                convert_time += time.monotonic_ns() - b\n                yield morsel\n                at_least_once = True\n\n                # Dynamically adjust chunk size based on the data size, we start by downloading\n                # 500 records to get an idea of the row size, assuming these 500 are\n                # representative, we work out how many rows fit into 16Mb (check setting).\n                # Don't keep recalculating, this is not a cheap operation and it's predicting\n                # the future so isn't going to ever be 100% correct\n                if self.chunk_size == INITIAL_CHUNK_SIZE and morsel.nbytes > 0:\n                    self.chunk_size = int(len(morsel) // (morsel.nbytes / DEFAULT_MORSEL_SIZE)) + 1\n                    self.chunk_size = (self.chunk_size // MIN_CHUNK_SIZE) * MIN_CHUNK_SIZE\n                    self.chunk_size = max(self.chunk_size, MIN_CHUNK_SIZE)\n                    # DEBUG: log (f\"CHANGING CHUNK SIZE TO {self.chunk_size} was {INITIAL_CHUNK_SIZE}.\")\n\n        if not at_least_once:\n            yield DataFrame(schema=result_schema).arrow()\n\n        # DEBUG: log (f\"time spent converting: {convert_time/1e9}s\")\n\n    def get_dataset_schema(self) -> RelationSchema:\n        from sqlalchemy import Table\n\n        if self.schema:\n            return self.schema\n\n        # Try to read the schema from the metastore\n        self.schema = self.read_schema_from_metastore()\n        if self.schema:\n            return self.schema\n\n        # get the schema from the dataset\n        # DEBUG: log (\"GET SQL SCHEMA:\", self.dataset)\n        try:\n            table = Table(self.dataset, self.metadata, autoload_with=self._engine)\n\n            self.schema = RelationSchema(\n                name=table.name,\n                columns=[\n                    FlatColumn(\n                        name=column.name,\n                        type=PYTHON_TO_ORSO_MAP[column.type.python_type],\n                        precision=(\n                            None\n                            if column.type.python_type != Decimal\n                            else column.type.precision  # type:ignore\n                        ),\n                        scale=(\n                            None\n                            if column.type.python_type != Decimal\n                            else column.type.scale  # type:ignore\n                        ),\n                        nullable=column.nullable,\n                    )\n                    for column in table.columns\n                ],\n            )\n        except Exception as err:\n            # Fall back to getting the schema from the first row, this is the column names, and where\n            # possible, column types.\n            # DEBUG: log (f\"APPROXIMATING SCHEMA OF {self.dataset} BECAUSE OF {type(err).__name__}({err})\")\n            from sqlalchemy.sql import text\n\n            with self._engine.connect() as conn:\n                query = Query().SELECT(\"*\").FROM(self.dataset).LIMIT(\"1\")\n                # DEBUG: log (\"READ ROW\\n\", str(query))\n                row = conn.execute(text(str(query))).fetchone()._asdict()\n                # DEBUG: log (\"ROW:\", row)\n                self.schema = RelationSchema(\n                    name=self.dataset,\n                    columns=[\n                        FlatColumn(\n                            name=column,\n                            type=0 if value is None else PYTHON_TO_ORSO_MAP[type(value)],\n                        )\n                        for column, value in row.items()\n                    ],\n                )\n\n        return self.schema\n", "path": "opteryx/connectors/sql_connector.py"}, {"content": "__build__ = 341\n\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#     http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\n\"\"\"\nStore the version here so:\n1) we don't load dependencies by storing it in __init__.py\n2) we can import it in setup.py for the same reason\n\"\"\"\nfrom enum import Enum  # isort: skip\n\n\nclass VersionStatus(Enum):\n    ALPHA = \"alpha\"\n    BETA = \"beta\"\n    RELEASE = \"release\"\n\n\n_major = 0\n_minor = 13\n_revision = 4\n_status = VersionStatus.ALPHA\n\n__author__ = \"@joocer\"\n__version__ = f\"{_major}.{_minor}.{_revision}\" + (\n    f\"-{_status.value}.{__build__}\" if _status != VersionStatus.RELEASE else \"\"\n)\n", "path": "opteryx/__version__.py"}]}