Repository: ClickHouse/clickhouse-connect (language: python)

Issue:
Timezone-aware `datetime` query parameters are internally cast to a wrong timezone
### Describe the bug

Timezone-aware `datetime` query parameters are internally cast to a wrong timezone.

### Steps to reproduce
1. Use a database whose timezone is EST (UTC-05:00):
```sql
SELECT timezone();
-- EST
```
2. Make a `Client` with default parameters:
```python
import clickhouse_connect

client = clickhouse_connect.get_client(
    username='...',
    password='...',
    host='...',
    port=443,
    database='...',
)
```
3. Make a query with a timezone-aware datetime parameter in UTC+3:
```python
import datetime
tzinfo = datetime.timezone(offset=datetime.timedelta(hours=3)) # UTC+3
dt = datetime.datetime(2024, 7, 23, 12, 0, 0, tzinfo=tzinfo)
result = client.query('select toString(%s)', parameters=(dt,))
print(result.result_rows[0][0])
# 2024-07-23 09:00:00
```

It should have converted UTC+3 to UTC-5 and get `2024-07-23 04:00:00`.

### Where it happens

As we can see in the debugger, the function `clickhouse_connect.driver.query.py:format_query_value` is casting the parameter to UTC:
```python
    if isinstance(value, datetime):
        if value.tzinfo is not None or server_tz != pytz.UTC:
            value = value.astimezone(server_tz) # <-- here
        return f"'{value.strftime('%Y-%m-%d %H:%M:%S')}'"
```

The `format_query_value()` was called from `finalize_query()`:

```python
def finalize_query(query: str, parameters: Optional[Union[Sequence, Dict[str, Any]]],
                   server_tz: Optional[tzinfo] = None) -> str:
    while query.endswith(';'):
        query = query[:-1]
    if not parameters:
        return query
    if hasattr(parameters, 'items'):
        return query % {k: format_query_value(v, server_tz) for k, v in parameters.items()}
    return query % tuple(format_query_value(v) for v in parameters) # <-- here
```

For some reason, `server_tz` wasn't passed to `format_query_value()` and `pytz.UTC` default value was used.

### Configuration

#### Environment
* clickhouse-connect version: 0.7.16
* Python version: 3.9.6
* Operating system: macOS Ventura 13.5.2

#### ClickHouse server
* ClickHouse Server version: 23.8.8.21.altinitystable
Streaming with query_row_block_stream crashes after few reads
<!-- delete unnecessary items -->
### Describe the bug

The streaming read is crashing after some time if there any processing in between reads.

```
Traceback (most recent call last):
  File "/usr/lib/python3.11/http/client.py", line 573, in _get_chunk_left
    chunk_left = self._read_next_chunk_size()
                 ^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/lib/python3.11/http/client.py", line 540, in _read_next_chunk_size
    return int(line, 16)
           ^^^^^^^^^^^^^
ValueError: invalid literal for int() with base 16: b'\xe9\xe9\x80\xe8\xf8\x8c\x05I"\xe7\x1bh\x9e\x0c\xa6\x06\xe3\x92\xa3?\x7f6:e\xec\x9d\xa8\x01d\xd3\x118{\xe1\xe9\xe60ta\x11T\xeb\xd0\x1d\x91\xed\xbf\x8aT5\xc8\x00\xceN\x1bA{\xe4\xf9\xeb\xd0\xc5\xba\xb
During handling of the above exception, another exception occurred:
Traceback (most recent call last):
  File "/usr/lib/python3.11/http/client.py", line 590, in _read_chunked
    chunk_left = self._get_chunk_left()
                 ^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/lib/python3.11/http/client.py", line 575, in _get_chunk_left
    raise IncompleteRead(b'')
http.client.IncompleteRead: IncompleteRead(0 bytes read)
The above exception was the direct cause of the following exception:
Traceback (most recent call last):
  File "/progai/.venv/lib/python3.11/site-packages/urllib3/response.py", line 444, in _error_catcher
    yield
  File "/progai/.venv/lib/python3.11/site-packages/urllib3/response.py", line 567, in read
    data = self._fp_read(amt) if not fp_closed else b""
           ^^^^^^^^^^^^^^^^^^
  File "/progai/.venv/lib/python3.11/site-packages/urllib3/response.py", line 533, in _fp_read
    return self._fp.read(amt) if amt is not None else self._fp.read()
           ^^^^^^^^^^^^^^^^^^
  File "/usr/lib/python3.11/http/client.py", line 467, in read
    return self._read_chunked(amt)
           ^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/lib/python3.11/http/client.py", line 605, in _read_chunked
    raise IncompleteRead(b''.join(value)) from exc
http.client.IncompleteRead: IncompleteRead(134 bytes read)
During handling of the above exception, another exception occurred:
Traceback (most recent call last):
  File "/progai/pipeline/airflow/dags/shared/female_names_processing.py", line 76, in execute
    for i, block in enumerate(stream):
  File "/progai/.venv/lib/python3.11/site-packages/clickhouse_connect/driver/common.py", line 201, in __next__
    return next(self.gen)
           ^^^^^^^^^^^^^^
  File "/progai/.venv/lib/python3.11/site-packages/clickhouse_connect/driver/query.py", line 296, in _row_block_stream
    for block in self._column_block_stream():
  File "/progai/.venv/lib/python3.11/site-packages/clickhouse_connect/driver/transform.py", line 75, in gen
    next_block = get_block()
                 ^^^^^^^^^^^
  File "/progai/.venv/lib/python3.11/site-packages/clickhouse_connect/driver/transform.py", line 50, in get_block
    column = col_type.read_column(source, num_rows, context)
             ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/progai/.venv/lib/python3.11/site-packages/clickhouse_connect/datatypes/base.py", line 143, in read_column
    return self.read_column_data(source, num_rows, ctx)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/progai/.venv/lib/python3.11/site-packages/clickhouse_connect/datatypes/base.py", line 158, in read_column_data
    column = self._read_column_binary(source, num_rows, ctx)
             ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/progai/.venv/lib/python3.11/site-packages/clickhouse_connect/datatypes/string.py", line 34, in _read_column_binary
    return source.read_str_col(num_rows, self._active_encoding(ctx))
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "clickhouse_connect/driverc/buffer.pyx", line 248, in clickhouse_connect.driverc.buffer.ResponseBuffer.read_str_col
  File "clickhouse_connect/driverc/buffer.pyx", line 134, in clickhouse_connect.driverc.buffer.ResponseBuffer._read_str_col
  File "clickhouse_connect/driverc/buffer.pyx", line 74, in clickhouse_connect.driverc.buffer.ResponseBuffer.read_bytes_c
  File "/progai/.venv/lib/python3.11/site-packages/clickhouse_connect/driver/httputil.py", line 200, in decompress
    chunk = response.read(chunk_size, decode_content=False)
            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/progai/.venv/lib/python3.11/site-packages/urllib3/response.py", line 566, in read
    with self._error_catcher():
  File "/usr/lib/python3.11/contextlib.py", line 158, in __exit__
    self.gen.throw(typ, value, traceback)
  File "/progai/.venv/lib/python3.11/site-packages/urllib3/response.py", line 461, in _error_catcher
    raise ProtocolError("Connection broken: %r" % e, e)
urllib3.exceptions.ProtocolError: ('Connection broken: IncompleteRead(134 bytes read)', IncompleteRead(134 bytes read))
```

### Steps to reproduce
1. start stream read
2. consume iterator with some work/sleep
3. soon it will crash with the error above

### Expected behaviour
I expect it to read the whole dataset. If I disable processing, the dataset is read fine (40 mln records).
This leads me to think that it is not related to actual data and response but something inside the implementation.

### Code example
```python
import clickhouse_connect
read_client = clickhouse_connect.get_client(
            host=host, username=login, password=password, port=port,
            connect_timeout=60 * 30, send_receive_timeout=60 * 30, client_name="airflow-read",
            settings={
                "session_timeout": 60 * 20,
            }
        )
qry = 'SELECT id, name FROM dev.tbl WHERE NOT empty(name)'
with read_client.query_row_block_stream(qry) as stream:
    for block in stream:
        rows = list(process(block, dataset))
```

### clickhouse-connect and/or ClickHouse server logs

### Configuration
#### Environment
* clickhouse-connect version: 0.7.19
* Python version: 3.11.10
* Operating system: Ubuntu 20.4


#### ClickHouse server
* ClickHouse Server version: 24.7.2 revision 54468
* ClickHouse Server non-default settings, if any:
* `CREATE TABLE` statements for tables involved:
* Sample data for these tables, use [clickhouse-obfuscator](https://github.com/ClickHouse/ClickHouse/blob/master/programs/obfuscator/Obfuscator.cpp#L42-L80) if necessary




Discussion / hints:
Thanks for the report!  `server_tz` has actually been broken for a while and no one seemed to notice.  This should be fixed in 0.7.17
@genzgd Could you please also pass the `server_tz` to `format_query_value()` in the case when `parameters` are not a dictionary:
```python
def finalize_query(query: str, parameters: Optional[Union[Sequence, Dict[str, Any]]],
                   server_tz: Optional[tzinfo] = None) -> str:
    while query.endswith(';'):
        query = query[:-1]
    if not parameters:
        return query
    if hasattr(parameters, 'items'):
        return query % {k: format_query_value(v, server_tz) for k, v in parameters.items()}
    return query % tuple(format_query_value(v) for v in parameters) # <<<---- EDIT THIS LINE
```


Fix patch:
diff --git a/.docker/clickhouse/single_node/config.xml b/.docker/clickhouse/single_node/config.xml
index 8b20b06b..c0f76f36 100644
--- a/.docker/clickhouse/single_node/config.xml
+++ b/.docker/clickhouse/single_node/config.xml
@@ -37,24 +37,5 @@
         <table>session_log</table>
     </session_log>
 
-    <http_options_response>
-        <header>
-            <name>Access-Control-Allow-Origin</name>
-            <value>*</value>
-        </header>
-        <header>
-            <name>Access-Control-Allow-Headers</name>
-            <value>accept, origin, x-requested-with, content-type, authorization</value>
-        </header>
-        <header>
-            <name>Access-Control-Allow-Methods</name>
-            <value>POST, GET, OPTIONS</value>
-        </header>
-        <header>
-            <name>Access-Control-Max-Age</name>
-            <value>86400</value>
-        </header>
-    </http_options_response>
-
     <custom_settings_prefixes>SQL_</custom_settings_prefixes>
 </clickhouse>
diff --git a/.docker/clickhouse/single_node_tls/Dockerfile b/.docker/clickhouse/single_node_tls/Dockerfile
index b027edb3..0fd1a6f5 100644
--- a/.docker/clickhouse/single_node_tls/Dockerfile
+++ b/.docker/clickhouse/single_node_tls/Dockerfile
@@ -1,4 +1,4 @@
-FROM clickhouse/clickhouse-server:24.3-alpine
+FROM clickhouse/clickhouse-server:24.8-alpine
 COPY .docker/clickhouse/single_node_tls/certificates /etc/clickhouse-server/certs
 RUN chown clickhouse:clickhouse -R /etc/clickhouse-server/certs \
     && chmod 600 /etc/clickhouse-server/certs/* \
diff --git a/.docker/clickhouse/single_node_tls/config.xml b/.docker/clickhouse/single_node_tls/config.xml
index 1bc4e92e..d25ffbaa 100644
--- a/.docker/clickhouse/single_node_tls/config.xml
+++ b/.docker/clickhouse/single_node_tls/config.xml
@@ -43,4 +43,6 @@
         <partition_by>toYYYYMM(event_date)</partition_by>
         <flush_interval_milliseconds>1000</flush_interval_milliseconds>
     </query_log>
+
+    <custom_settings_prefixes>SQL_</custom_settings_prefixes>
 </clickhouse>
diff --git a/.github/workflows/clickhouse_ci.yml b/.github/workflows/clickhouse_ci.yml
index e51b56aa..4b2cc45f 100644
--- a/.github/workflows/clickhouse_ci.yml
+++ b/.github/workflows/clickhouse_ci.yml
@@ -28,7 +28,7 @@ jobs:
       - name: "Add distribution info"  #  This lets SQLAlchemy find entry points
         run: python setup.py develop
 
-      - name: run ClickHouse Cloud SMT tests
+      - name: run ClickHouse Cloud tests
         env:
           CLICKHOUSE_CONNECT_TEST_PORT: 8443
           CLICKHOUSE_CONNECT_TEST_CLOUD: 'True'
@@ -42,18 +42,3 @@ jobs:
         run: pytest tests/integration_tests
       - name: remove latest container
         run: docker compose down -v
-
-      - name: run ClickHouse Cloud tests
-        env:
-          CLICKHOUSE_CONNECT_TEST_PORT: 8443
-          CLICKHOUSE_CONNECT_TEST_INSERT_QUORUM: 3
-          CLICKHOUSE_CONNECT_TEST_HOST: ${{ secrets.INTEGRATIONS_TEAM_TESTS_CLOUD_HOST }}
-          CLICKHOUSE_CONNECT_TEST_PASSWORD: ${{ secrets.INTEGRATIONS_TEAM_TESTS_CLOUD_PASSWORD }}
-        run: pytest tests/integration_tests
-
-      - name: Run ClickHouse Container (HEAD)
-        run: CLICKHOUSE_VERSION=head docker compose up -d clickhouse
-      - name: Run HEAD tests
-        run: pytest tests/integration_tests
-      - name: remove head container
-        run: docker compose down -v
diff --git a/.github/workflows/on_push.yml b/.github/workflows/on_push.yml
index 9c97fe16..49484dae 100644
--- a/.github/workflows/on_push.yml
+++ b/.github/workflows/on_push.yml
@@ -58,17 +58,15 @@ jobs:
     strategy:
       matrix:
         python-version:
-          - '3.8'
           - '3.9'
           - '3.10'
           - '3.11'
           - '3.12'
         clickhouse-version:
-          - '23.8'
-          - '23.12'
-          - '24.1'
-          - '24.2'
           - '24.3'
+          - '24.6'
+          - '24.7'
+          - '24.8'
           - latest
 
     name: Local Tests Py=${{ matrix.python-version }} CH=${{ matrix.clickhouse-version }}
@@ -99,14 +97,11 @@ jobs:
           sudo echo "127.0.0.1 server1.clickhouse.test" | sudo tee -a /etc/hosts
       - name: Run tests
         env:
+          CLICKHOUSE_CONNECT_TEST_TLS: 1
           CLICKHOUSE_CONNECT_TEST_DOCKER: 'False'
           CLICKHOUSE_CONNECT_TEST_FUZZ: 50
           SQLALCHEMY_SILENCE_UBER_WARNING: 1
         run: pytest tests
-      - name: Run TLS tests
-        env:
-          CLICKHOUSE_CONNECT_TEST_TLS: 1
-        run: pytest tests/tls
 
   check-secret:
     runs-on: ubuntu-latest
diff --git a/CHANGELOG.md b/CHANGELOG.md
index ba43c121..7c015be8 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -3,10 +3,75 @@
 ### WARNING -- Impending Breaking Change - Server Settings in DSN
 When creating a DBAPI Connection method using the Connection constructor or a SQLAlchemy DSN, the library currently
 converts any unrecognized keyword argument/query parameter to a ClickHouse server setting. Starting in the next minor
-release (0.8.0), unrecognized arguments/keywords for these methods of creating a DBAPI connection will raise an exception
+release (0.9.0), unrecognized arguments/keywords for these methods of creating a DBAPI connection will raise an exception
 instead of being passed as ClickHouse server settings. This is in conjunction with some refactoring in Client construction.
 The supported method of passing ClickHouse server settings is to prefix such arguments/query parameters with`ch_`.  
 
+## 0.8.0, 2024-09-26
+### Experimental Feature - "New" JSON/Dynamic/Variant DataTypes
+#### Usage Notes
+- JSON data can be inserted as either a Python dictionary or a JSON string containing a JSON object `{}`.  Other
+forms of JSON data are not supported
+- Valid formats for the JSON type are 'native', which returns a Python dictionary, or 'string', which returns a JSON string
+- Any value can be inserted into a Variant column, and ClickHouse will try to correctly determine the correct Variant
+Type for the value, based on its String representation.
+- More complete documentation for the new types will be provided in the future.
+
+#### Known limitations:
+- Each of these types must be enabled in the ClickHouse settings before using.  The "new" JSON type is available started
+with the 24.8 release
+- Returned JSON objects will only return the `max_dynamic_paths` number of elements (which defaults to 1024).  This
+will be fixed in a future release.
+- Inserts into `Dynamic` columns will always be the String representation of the Python value.  This will be fixed
+in a future release.
+- The implementation for the new types has not been optimized in C code, so performance may be somewhat slower than for
+simpler, established data types.
+
+This is the first time that a new `clickhouse_connect` features has been labeled "experimental", but these new
+datatypes are complex and still experimental in ClickHouse server.  Current test coverage for these types is also
+quite limited.  Please don't hesitate to report issues with the new types.
+
+### Bug Fixes
+- When operating ClickHouse Server in `strict` TLS mode, HTTPS connections [require](https://github.com/ClickHouse/poco/blob/master/NetSSL_OpenSSL/include/Poco/Net/Context.h#L84-L89) a client certificate even if that
+certificate is not used for authentication.  A new client parameter `tls_mode='strict'` can be used in this situation where
+username/password authentication is being used with client certificates.  Other valid values for the new `tls_mode` setting
+are `'proxy'` when TLS termination occurs at a proxy, and `'mutual'` to specify mutual TLS authentication is used by
+the ClickHouse server.  If `tls_mode` is not set, and a client certificate and key are provided, `mutual` is assumed.
+- The server timezone was not being used for parameter binding if parameters were sent as a list instead of a dictionary.
+This should fully fix the reopened https://github.com/ClickHouse/clickhouse-connect/issues/377.
+- String port numbers (such as from environmental variables) are now correctly interpreted to determine the correct interface/protocol.
+Fixes https://github.com/ClickHouse/clickhouse-connect/issues/395
+- Insert commands with a `SELECT FROM ... LIMIT 0` will no longer raise an exception.  Closes https://github.com/ClickHouse/clickhouse-connect/issues/389.
+
+### Improvements
+- Some low level errors for problems with Native format inserts and queries now include the relevant column name in the
+error message.  Thanks to [Angus Holder](https://github.com/angusholder) for the PR!
+- There is a new intermediate buffer for HTTP streaming/chunked queries.  The buffer will store raw data from the HTTP request
+until it is actually requested in a stream.  This allows some lag between reading the data from ClickHouse and processing
+the same data.  Previously, if processing the data stream fell 30 seconds behind the ClickHouse HTTP writes to the stream,
+the ClickHouse server would close the connection, aborting the query and stream processing.  This will now be mitigated by
+storing the data stream in the new intermediate buffer.  By default, this buffer is set to 10 megabytes, but for slow
+processing of large queries where memory is not an issue, the buffer size can be increasing using the new `common` setting
+`http_buffer_size`.  This is a fix in some cases of https://github.com/ClickHouse/clickhouse-connect/issues/399, but note that
+slow processing of large queries will still cause connection and processing failures if the data cannot be buffered.
+- It is now possible to correctly bind `DateTime64` type parameters when calling Client `query` methods through one of two approaches:
+  - Wrap the Python `datetime.datetime` value in the new DT64Param class, e.g.
+  ```python
+    query = 'SELECT {p1:DateTime64(3)}'  # Server side binding with dictionary
+    parameters={'p1': DT64Param(dt_value)}
+  
+    query = 'SELECT %s as string, toDateTime64(%s,6) as dateTime' # Client side binding with list 
+    parameters=['a string', DT64Param(datetime.now())]
+  ```
+  - If using a dictionary of parameter values, append the string `_64` to the parameter name
+  ```python
+    query = 'SELECT {p1:DateTime64(3)}, {a1:Array(DateTime(3))}'  # Server side binding with dictionary
+  
+    parameters={'p1_64': dt_value, 'a1_64': [dt_value1, dt_value2]}
+  ```
+  This closes https://github.com/ClickHouse/clickhouse-connect/issues/396, see also the similar issue https://github.com/ClickHouse/clickhouse-connect/issues/212
+
+
 ## 0.7.19, 2024-08-23
 ### Bug Fix
 - Insertion of large strings was triggering an exception. This has been fixed.
diff --git a/clickhouse_connect/__version__.py b/clickhouse_connect/__version__.py
index d9d76edb..c5e4522a 100644
--- a/clickhouse_connect/__version__.py
+++ b/clickhouse_connect/__version__.py
@@ -1,1 +1,1 @@
-version = '0.7.19'
+version = '0.8.0'
diff --git a/clickhouse_connect/cc_sqlalchemy/datatypes/base.py b/clickhouse_connect/cc_sqlalchemy/datatypes/base.py
index 0c0d25e6..ce73da14 100644
--- a/clickhouse_connect/cc_sqlalchemy/datatypes/base.py
+++ b/clickhouse_connect/cc_sqlalchemy/datatypes/base.py
@@ -5,7 +5,7 @@
 
 from clickhouse_connect.datatypes.base import ClickHouseType, TypeDef, EMPTY_TYPE_DEF
 from clickhouse_connect.datatypes.registry import parse_name, type_map
-from clickhouse_connect.driver.query import str_query_value
+from clickhouse_connect.driver.binding import str_query_value
 
 logger = logging.getLogger(__name__)
 
diff --git a/clickhouse_connect/cc_sqlalchemy/ddl/custom.py b/clickhouse_connect/cc_sqlalchemy/ddl/custom.py
index b7eee4ad..9f7c1e64 100644
--- a/clickhouse_connect/cc_sqlalchemy/ddl/custom.py
+++ b/clickhouse_connect/cc_sqlalchemy/ddl/custom.py
@@ -1,7 +1,7 @@
 from sqlalchemy.sql.ddl import DDL
 from sqlalchemy.exc import ArgumentError
 
-from clickhouse_connect.driver.query import quote_identifier
+from clickhouse_connect.driver.binding import quote_identifier
 
 
 #  pylint: disable=too-many-ancestors,abstract-method
@@ -31,7 +31,7 @@ def __init__(self, name: str, engine: str = None, zoo_path: str = None, shard_na
         super().__init__(stmt)
 
 
-#  pylint: disable=too-many-ancestors,abstract-method
+# pylint: disable=too-many-ancestors,abstract-method
 class DropDatabase(DDL):
     """
     Alternative DDL statement for built in SqlAlchemy DropSchema DDL class
diff --git a/clickhouse_connect/cc_sqlalchemy/dialect.py b/clickhouse_connect/cc_sqlalchemy/dialect.py
index 6b04c7e2..c9bb24c9 100644
--- a/clickhouse_connect/cc_sqlalchemy/dialect.py
+++ b/clickhouse_connect/cc_sqlalchemy/dialect.py
@@ -8,7 +8,7 @@
 from clickhouse_connect.cc_sqlalchemy.sql.ddlcompiler import ChDDLCompiler
 from clickhouse_connect.cc_sqlalchemy import ischema_names, dialect_name
 from clickhouse_connect.cc_sqlalchemy.sql.preparer import ChIdentifierPreparer
-from clickhouse_connect.driver.query import quote_identifier, format_str
+from clickhouse_connect.driver.binding import quote_identifier, format_str
 
 
 # pylint: disable=too-many-public-methods,no-self-use,unused-argument
diff --git a/clickhouse_connect/cc_sqlalchemy/sql/__init__.py b/clickhouse_connect/cc_sqlalchemy/sql/__init__.py
index 68becd54..00b9bc8c 100644
--- a/clickhouse_connect/cc_sqlalchemy/sql/__init__.py
+++ b/clickhouse_connect/cc_sqlalchemy/sql/__init__.py
@@ -2,7 +2,7 @@
 
 from sqlalchemy import Table
 
-from clickhouse_connect.driver.query import quote_identifier
+from clickhouse_connect.driver.binding import quote_identifier
 
 
 def full_table(table_name: str, schema: Optional[str] = None) -> str:
diff --git a/clickhouse_connect/cc_sqlalchemy/sql/ddlcompiler.py b/clickhouse_connect/cc_sqlalchemy/sql/ddlcompiler.py
index 5a972547..8a2180c4 100644
--- a/clickhouse_connect/cc_sqlalchemy/sql/ddlcompiler.py
+++ b/clickhouse_connect/cc_sqlalchemy/sql/ddlcompiler.py
@@ -2,7 +2,7 @@
 from sqlalchemy.sql.compiler import DDLCompiler
 
 from clickhouse_connect.cc_sqlalchemy.sql import  format_table
-from clickhouse_connect.driver.query import quote_identifier
+from clickhouse_connect.driver.binding import quote_identifier
 
 
 class ChDDLCompiler(DDLCompiler):
diff --git a/clickhouse_connect/cc_sqlalchemy/sql/preparer.py b/clickhouse_connect/cc_sqlalchemy/sql/preparer.py
index a31b3e7a..f53a2bde 100644
--- a/clickhouse_connect/cc_sqlalchemy/sql/preparer.py
+++ b/clickhouse_connect/cc_sqlalchemy/sql/preparer.py
@@ -1,6 +1,6 @@
 from sqlalchemy.sql.compiler import IdentifierPreparer
 
-from clickhouse_connect.driver.query import quote_identifier
+from clickhouse_connect.driver.binding import quote_identifier
 
 
 class ChIdentifierPreparer(IdentifierPreparer):
diff --git a/clickhouse_connect/common.py b/clickhouse_connect/common.py
index 1fda1ecf..ab960e76 100644
--- a/clickhouse_connect/common.py
+++ b/clickhouse_connect/common.py
@@ -81,3 +81,6 @@ def _init_common(name: str, options: Sequence[Any], default: Any):
 _init_common('use_protocol_version', (True, False), True)
 
 _init_common('max_error_size', (), 1024)
+
+# HTTP raw data buffer for streaming queries.  This should not be reduced below 64KB to ensure compatibility with LZ4 compression
+_init_common('http_buffer_size', (), 10 * 1024 * 1024)
diff --git a/clickhouse_connect/datatypes/__init__.py b/clickhouse_connect/datatypes/__init__.py
index aa9b8c2f..e4229f66 100644
--- a/clickhouse_connect/datatypes/__init__.py
+++ b/clickhouse_connect/datatypes/__init__.py
@@ -4,4 +4,6 @@
 import clickhouse_connect.datatypes.special
 import clickhouse_connect.datatypes.string
 import clickhouse_connect.datatypes.temporal
+import clickhouse_connect.datatypes.dynamic
 import clickhouse_connect.datatypes.registry
+import clickhouse_connect.datatypes.postinit
diff --git a/clickhouse_connect/datatypes/base.py b/clickhouse_connect/datatypes/base.py
index d7a3f963..ab43b4e1 100644
--- a/clickhouse_connect/datatypes/base.py
+++ b/clickhouse_connect/datatypes/base.py
@@ -3,7 +3,7 @@
 
 from abc import ABC
 from math import log
-from typing import NamedTuple, Dict, Type, Any, Sequence, MutableSequence, Optional, Union, Collection
+from typing import NamedTuple, Dict, Type, Any, Sequence, MutableSequence, Union, Collection
 
 from clickhouse_connect.driver.common import array_type, int_size, write_array, write_uint64, low_card_version
 from clickhouse_connect.driver.context import BaseQueryContext
@@ -94,6 +94,10 @@ def name(self):
             name = f'{wrapper}({name})'
         return name
 
+    @property
+    def insert_name(self):
+        return self.name
+
     def data_size(self, sample: Sequence) -> int:
         if self.low_card:
             values = set(sample)
@@ -104,10 +108,13 @@ def data_size(self, sample: Sequence) -> int:
             d_size += 1
         return d_size
 
-    def _data_size(self, _sample: Collection) -> int:
+    def _data_size(self, sample: Collection) -> int:
         if self.byte_size:
             return self.byte_size
-        return 0
+        total = 0
+        for x in sample:
+            total += len(str(x))
+        return total / len(sample) + 1
 
     def write_column_prefix(self, dest: bytearray):
         """
@@ -119,7 +126,7 @@ def write_column_prefix(self, dest: bytearray):
         if self.low_card:
             write_uint64(low_card_version, dest)
 
-    def read_column_prefix(self, source: ByteSource):
+    def read_column_prefix(self, source: ByteSource, _ctx: QueryContext):
         """
         Read the low cardinality version.  Like the write method, this has to happen immediately for container classes
         :param source: The native protocol binary read buffer
@@ -139,7 +146,7 @@ def read_column(self, source: ByteSource, num_rows: int, ctx: QueryContext) -> S
         :param ctx: QueryContext for query specific settings
         :return: The decoded column data as a sequence and the updated location pointer
         """
-        self.read_column_prefix(source)
+        self.read_column_prefix(source, ctx)
         return self.read_column_data(source, num_rows, ctx)
 
     def read_column_data(self, source: ByteSource, num_rows: int, ctx: QueryContext) -> Sequence:
@@ -274,18 +281,11 @@ def _write_column_low_card(self, column: Sequence, dest: bytearray, ctx: InsertC
         write_uint64(len(index), dest)
         self._write_column_binary(index, dest, ctx)
         write_uint64(len(keys), dest)
-        write_array(array_type(1 << ix_type, False), keys, dest, ctx)
+        write_array(array_type(1 << ix_type, False), keys, dest, ctx.column_name)
 
     def _active_null(self, _ctx: QueryContext) -> Any:
         return None
 
-    def _first_value(self, column: Sequence) -> Optional[Any]:
-        if self.nullable:
-            return next((x for x in column if x is not None), None)
-        if len(column):
-            return column[0]
-        return None
-
 
 EMPTY_TYPE_DEF = TypeDef()
 NULLABLE_TYPE_DEF = TypeDef(wrappers=('Nullable',))
@@ -338,7 +338,7 @@ def _finalize_column(self, column: Sequence, ctx: QueryContext) -> Sequence:
     def _write_column_binary(self, column: Union[Sequence, MutableSequence], dest: bytearray, ctx: InsertContext):
         if len(column) and self.nullable:
             column = [0 if x is None else x for x in column]
-        write_array(self._array_type, column, dest, ctx)
+        write_array(self._array_type, column, dest, ctx.column_name)
 
     def _active_null(self, ctx: QueryContext):
         if ctx.as_pandas and ctx.use_extended_dtypes:
diff --git a/clickhouse_connect/datatypes/container.py b/clickhouse_connect/datatypes/container.py
index 445b2414..b8244eef 100644
--- a/clickhouse_connect/datatypes/container.py
+++ b/clickhouse_connect/datatypes/container.py
@@ -3,11 +3,12 @@
 from typing import Sequence, Collection
 
 from clickhouse_connect.driver.insert import InsertContext
-from clickhouse_connect.driver.query import QueryContext, quote_identifier
+from clickhouse_connect.driver.query import QueryContext
+from clickhouse_connect.driver.binding import quote_identifier
 from clickhouse_connect.driver.types import ByteSource
 from clickhouse_connect.json_impl import any_to_json
 from clickhouse_connect.datatypes.base import ClickHouseType, TypeDef
-from clickhouse_connect.driver.common import must_swap
+from clickhouse_connect.driver.common import must_swap, first_value
 from clickhouse_connect.datatypes.registry import get_from_name
 
 logger = logging.getLogger(__name__)
@@ -22,8 +23,8 @@ def __init__(self, type_def: TypeDef):
         self.element_type = get_from_name(type_def.values[0])
         self._name_suffix = f'({self.element_type.name})'
 
-    def read_column_prefix(self, source: ByteSource):
-        return self.element_type.read_column_prefix(source)
+    def read_column_prefix(self, source: ByteSource, ctx:QueryContext):
+        return self.element_type.read_column_prefix(source, ctx)
 
     def _data_size(self, sample: Sequence) -> int:
         if len(sample) == 0:
@@ -102,7 +103,7 @@ def _data_size(self, sample: Collection) -> int:
         if len(sample) == 0:
             return 0
         elem_size = 0
-        is_dict = self.element_names and isinstance(self._first_value(list(sample)), dict)
+        is_dict = self.element_names and isinstance(first_value(list(sample), self.nullable), dict)
         for ix, e_type in enumerate(self.element_types):
             if e_type.byte_size > 0:
                 elem_size += e_type.byte_size
@@ -112,9 +113,9 @@ def _data_size(self, sample: Collection) -> int:
                 elem_size += e_type.data_size([x[ix] for x in sample])
         return elem_size
 
-    def read_column_prefix(self, source: ByteSource):
+    def read_column_prefix(self, source: ByteSource, ctx: QueryContext):
         for e_type in self.element_types:
-            e_type.read_column_prefix(source)
+            e_type.read_column_prefix(source, ctx)
 
     def read_column_data(self, source: ByteSource, num_rows: int, ctx: QueryContext):
         columns = []
@@ -138,7 +139,7 @@ def write_column_prefix(self, dest: bytearray):
             e_type.write_column_prefix(dest)
 
     def write_column_data(self, column: Sequence, dest: bytearray, ctx: InsertContext):
-        if self.element_names and isinstance(self._first_value(column), dict):
+        if self.element_names and isinstance(first_value(column, self.nullable), dict):
             columns = self.convert_dict_insert(column)
         else:
             columns = list(zip(*column))
@@ -180,9 +181,9 @@ def _data_size(self, sample: Collection) -> int:
             total += self.value_type.data_size(x.values())
         return total // len(sample)
 
-    def read_column_prefix(self, source: ByteSource):
-        self.key_type.read_column_prefix(source)
-        self.value_type.read_column_prefix(source)
+    def read_column_prefix(self, source: ByteSource, ctx: QueryContext):
+        self.key_type.read_column_prefix(source, ctx)
+        self.value_type.read_column_prefix(source, ctx)
 
     # pylint: disable=too-many-locals
     def read_column_data(self, source: ByteSource, num_rows: int, ctx: QueryContext):
@@ -237,8 +238,8 @@ def _data_size(self, sample: Collection) -> int:
         array_sample = [[tuple(sub_row[key] for key in keys) for sub_row in row] for row in sample]
         return self.tuple_array.data_size(array_sample)
 
-    def read_column_prefix(self, source: ByteSource):
-        self.tuple_array.read_column_prefix(source)
+    def read_column_prefix(self, source: ByteSource, ctx:QueryContext):
+        self.tuple_array.read_column_prefix(source, ctx)
 
     def read_column_data(self, source: ByteSource, num_rows: int, ctx: QueryContext):
         keys = self.element_names
@@ -252,64 +253,3 @@ def write_column_data(self, column: Sequence, dest: bytearray, ctx: InsertContex
         keys = self.element_names
         data = [[tuple(sub_row[key] for key in keys) for sub_row in row] for row in column]
         self.tuple_array.write_column_data(data, dest, ctx)
-
-
-class JSON(ClickHouseType):
-    python_type = dict
-    # Native is a Python type (primitive, dict, array), string is an actual JSON string
-    valid_formats = 'string', 'native'
-
-    def write_column_prefix(self, dest: bytearray):
-        dest.append(0x01)
-
-    def _data_size(self, sample: Collection) -> int:
-        if len(sample) == 0:
-            return 0
-        total = 0
-        for x in sample:
-            if isinstance(x, str):
-                total += len(x)
-            elif x:
-                total += len(any_to_json(x))
-        return total // len(sample) + 1
-
-    # pylint: disable=duplicate-code
-    def write_column_data(self, column: Sequence, dest: bytearray, ctx: InsertContext):
-        app = dest.append
-        first = self._first_value(column)
-        if isinstance(first, str) or self.write_format(ctx) == 'string':
-            for x in column:
-                v = x.encode()
-                sz = len(v)
-                while True:
-                    b = sz & 0x7f
-                    sz >>= 7
-                    if sz == 0:
-                        app(b)
-                        break
-                    app(0x80 | b)
-                dest += v
-        else:
-            to_json = any_to_json
-            for x in column:
-                v = to_json(x)
-                sz = len(v)
-                while True:
-                    b = sz & 0x7f
-                    sz >>= 7
-                    if sz == 0:
-                        app(b)
-                        break
-                    app(0x80 | b)
-                dest += v
-
-
-class Object(JSON):
-    python_type = dict
-
-    def __init__(self, type_def):
-        data_type = type_def.values[0].lower().replace(' ', '')
-        if data_type not in ("'json'", "nullable('json')"):
-            raise NotImplementedError('Only json or Nullable(json) Object type is currently supported')
-        super().__init__(type_def)
-        self._name_suffix = type_def.arg_str
diff --git a/clickhouse_connect/datatypes/dynamic.py b/clickhouse_connect/datatypes/dynamic.py
new file mode 100644
index 00000000..e32145f1
--- /dev/null
+++ b/clickhouse_connect/datatypes/dynamic.py
@@ -0,0 +1,257 @@
+from typing import List, Sequence, Collection
+
+from clickhouse_connect.datatypes.base import ClickHouseType, TypeDef
+from clickhouse_connect.datatypes.registry import get_from_name
+from clickhouse_connect.driver.common import unescape_identifier, first_value
+from clickhouse_connect.driver.ctypes import data_conv
+from clickhouse_connect.driver.errors import handle_error
+from clickhouse_connect.driver.exceptions import DataError
+from clickhouse_connect.driver.insert import InsertContext
+from clickhouse_connect.driver.query import QueryContext
+from clickhouse_connect.driver.types import ByteSource
+from clickhouse_connect.json_impl import any_to_json
+
+SHARED_DATA_TYPE: ClickHouseType
+STRING_DATA_TYPE: ClickHouseType
+
+class Variant(ClickHouseType):
+    _slots = 'element_types'
+    python_type = object
+
+    def __init__(self, type_def: TypeDef):
+        super().__init__(type_def)
+        self.element_types:List[ClickHouseType] = [get_from_name(name) for name in type_def.values]
+        self._name_suffix = f"({', '.join(ch_type.name for ch_type in self.element_types)})"
+
+    @property
+    def insert_name(self):
+        return 'String'
+
+    def read_column_prefix(self, source: ByteSource, ctx:QueryContext):
+        if source.read_uint64() != 0:
+            raise DataError(f'Unexpected discriminator format in Variant column {ctx.column_name}')
+
+    def read_column_data(self, source: ByteSource, num_rows: int, ctx: QueryContext) -> Sequence:
+        return read_variant_column(self.element_types, source, num_rows, ctx)
+
+    def write_column_data(self, column: Sequence, dest: bytearray, ctx: InsertContext):
+        write_str_values(self, column, dest, ctx)
+
+
+def read_variant_column(variant_types: List[ClickHouseType], source: ByteSource, num_rows:int, ctx: QueryContext) -> Sequence:
+    v_count = len(variant_types)
+    discriminators = source.read_array('B', num_rows)
+    # We have to count up how many of each discriminator there are in the block to read the sub columns correctly
+    disc_rows = [0] * v_count
+    for disc in discriminators:
+        if disc != 255:
+            disc_rows[disc] += 1
+    sub_columns: List[Sequence] = [[]] * v_count
+    # Read all the sub-columns
+    for ix in range(v_count):
+        if disc_rows[ix] > 0:
+            sub_columns[ix] = variant_types[ix].read_column_data(source, disc_rows[ix], ctx)
+    # Now we have to walk through each of the discriminators again to assign the correct value from
+    # the sub-column to the final result column
+    sub_indexes = [0] * v_count
+    col = []
+    app_col = col.append
+    for disc in discriminators:
+        if disc == 255:
+            app_col(None)
+        else:
+            app_col(sub_columns[disc][sub_indexes[disc]])
+            sub_indexes[disc] += 1
+    return col
+
+
+class Dynamic(ClickHouseType):
+    python_type = object
+
+    @property
+    def insert_name(self):
+        return 'String'
+
+    def __init__(self, type_def:TypeDef):
+        super().__init__(type_def)
+        if type_def.keys and type_def.keys[0] == 'max_types':
+            self._name_suffix = f'(max_types={type_def.values[0]})'
+
+    def read_column(self, source: ByteSource, num_rows:int, ctx:QueryContext):
+        variant_types = read_dynamic_prefix(source)
+        return read_variant_column(variant_types, source, num_rows, ctx)
+
+    def write_column_data(self, column: Sequence, dest: bytearray, ctx: InsertContext):
+        write_str_values(self, column, dest, ctx)
+
+
+def read_dynamic_prefix(source: ByteSource) -> List[ClickHouseType]:
+    if source.read_uint64() != 1:  # dynamic structure serialization version, currently only 1 is recognized
+        raise DataError('Unrecognized dynamic structure version')
+    source.read_leb128()  # max dynamic types, we ignore this value
+    num_variants = source.read_leb128()
+    variant_types = [get_from_name(source.read_leb128_str()) for _ in range(num_variants)]
+    variant_types.append(STRING_DATA_TYPE)
+    if source.read_uint64() != 0: # discriminator format, currently only 0 is recognized
+        raise DataError('Unexpected discriminator format in Variant column prefix')
+    return variant_types
+
+
+def json_sample_size(_, sample: Collection) -> int:
+    if len(sample) == 0:
+        return 0
+    total = 0
+    for x in sample:
+        if isinstance(x, str):
+            total += len(x)
+        elif x:
+            total += len(any_to_json(x))
+    return total // len(sample) + 1
+
+
+def write_json(ch_type:ClickHouseType, column: Sequence, dest: bytearray, ctx: InsertContext):
+    first = first_value(column, ch_type.nullable)
+    write_col = column
+    encoding = ctx.encoding or ch_type.encoding
+    if not isinstance(first, str) and ch_type.write_format(ctx) != 'string':
+        to_json = any_to_json
+        write_col = [to_json(v) for v in column]
+        encoding = None
+    handle_error(data_conv.write_str_col(write_col, ch_type.nullable, encoding, dest), ctx)
+
+
+def write_str_values(ch_type:ClickHouseType, column: Sequence, dest: bytearray, ctx: InsertContext):
+    encoding = ctx.encoding or ch_type.encoding
+    col = [''] * len(column)
+    for ix, v in enumerate(column):
+        if v is None:
+            col[ix] = 'NULL'
+        else:
+            col[ix] = str(v)
+    handle_error(data_conv.write_str_col(col, False, encoding, dest), ctx)
+
+
+class JSON(ClickHouseType):
+    _slots = 'typed_paths', 'typed_types'
+    python_type = dict
+    valid_formats = 'string', 'native'
+    _data_size = json_sample_size
+    write_column_data = write_json
+    shared_data_type: ClickHouseType
+    max_dynamic_paths = 0
+    max_dynamic_types = 0
+    typed_paths = []
+    typed_types = []
+    skips = []
+
+    def __init__(self, type_def:TypeDef):
+        super().__init__(type_def)
+        typed_paths = []
+        typed_types = []
+        skips = []
+        parts = []
+        for key, value in zip(type_def.keys, type_def.values):
+            if key == 'max_dynamic_paths':
+                try:
+                    self.max_dynamic_paths = int(value)
+                    parts.append(f'{key} = {value}')
+                    continue
+                except ValueError:
+                    pass
+            if key == 'max_dynamic_types':
+                try:
+                    self.max_dynamic_types = int(value)
+                    parts.append(f'{key} = {value}')
+                    continue
+                except ValueError:
+                    pass
+            if key == 'SKIP':
+                if value.startswith('REGEXP'):
+                    value = 'REGEXP ' + value[6:]
+                else:
+                    if not value.startswith("`"):
+                        value = f'`{value}`'
+                skips.append(value)
+            else:
+                key = unescape_identifier(key)
+                typed_paths.append(key)
+                typed_types.append(get_from_name(value))
+                key = f'`{key}`'
+            parts.append(f'{key} {value}')
+        if typed_paths:
+            self.typed_paths = typed_paths
+            self.typed_types = typed_types
+        if skips:
+            self.skips = skips
+        if parts:
+            self._name_suffix = f'({", ".join(parts)})'
+
+    @property
+    def insert_name(self):
+        return 'String'
+
+    # pylint: disable=too-many-locals
+    def read_column(self, source: ByteSource, num_rows: int, ctx: QueryContext):
+        if source.read_uint64() != 0: # object serialization version, currently only 0 is recognized
+            raise DataError(f'unrecognized object serialization version, column `{ctx.column_name}`')
+        source.read_leb128() # the max number of dynamic paths.  Used to preallocate storage in ClickHouse; we ignore it
+        dynamic_path_cnt = source.read_leb128()
+        dynamic_paths = [source.read_leb128_str() for _ in range(dynamic_path_cnt)]
+        for typed in self.typed_types:
+            typed.read_column_prefix(source, ctx)
+        dynamic_variants = [read_dynamic_prefix(source) for _ in range(dynamic_path_cnt)]
+        # C++ prefix read ends here
+
+        typed_columns = [ch_type.read_column_data(source, num_rows, ctx) for ch_type in self.typed_types]
+        dynamic_columns = [read_variant_column(dynamic_variants[ix], source, num_rows, ctx) for ix in range(dynamic_path_cnt)]
+        SHARED_DATA_TYPE.read_column_data(source, num_rows, ctx)
+        col = []
+        for row_num in range(num_rows):
+            top = {}
+            for ix, field in enumerate(self.typed_paths):
+                value = typed_columns[ix][row_num]
+                item = top
+                chain = field.split('.')
+                for key in chain[:-1]:
+                    child = item.get(key)
+                    if child is None:
+                        child = {}
+                        item[key] = child
+                    item = child
+                item[chain[-1]] = value
+            for ix, field in enumerate(dynamic_paths):
+                value = dynamic_columns[ix][row_num]
+                if value is None:
+                    continue
+                item = top
+                chain = field.split('.')
+                for key in chain[:-1]:
+                    child = item.get(key)
+                    if child is None:
+                        child = {}
+                        item[key] = child
+                    item = child
+                item[chain[-1]] = value
+            col.append(top)
+        if self.read_format(ctx) == 'string':
+            return [any_to_json(v) for v in col]
+        return col
+
+
+# Note that this type is deprecated and should not be used, it included for temporary backward compatibility only
+class Object(ClickHouseType):
+    python_type = dict
+    # Native is a Python type (primitive, dict, array), string is an actual JSON string
+    valid_formats = 'string', 'native'
+    _data_size = json_sample_size
+    write_column_data = write_json
+
+    def __init__(self, type_def):
+        data_type = type_def.values[0].lower().replace(' ', '')
+        if data_type not in ("'json'", "nullable('json')"):
+            raise NotImplementedError('Only json or Nullable(json) Object type is currently supported')
+        super().__init__(type_def)
+        self._name_suffix = type_def.arg_str
+
+    def write_column_prefix(self, dest: bytearray):
+        dest.append(0x01)
diff --git a/clickhouse_connect/datatypes/network.py b/clickhouse_connect/datatypes/network.py
index 3f2951f9..f4468636 100644
--- a/clickhouse_connect/datatypes/network.py
+++ b/clickhouse_connect/datatypes/network.py
@@ -3,7 +3,7 @@
 from typing import Union, MutableSequence, Sequence
 
 from clickhouse_connect.datatypes.base import ClickHouseType
-from clickhouse_connect.driver.common import write_array, int_size
+from clickhouse_connect.driver.common import write_array, int_size, first_value
 from clickhouse_connect.driver.insert import InsertContext
 from clickhouse_connect.driver.query import QueryContext
 from clickhouse_connect.driver.types import ByteSource
@@ -29,7 +29,7 @@ def _read_column_binary(self, source: ByteSource, num_rows: int, ctx: QueryConte
         return data_conv.read_ipv4_col(source, num_rows)
 
     def _write_column_binary(self, column: Union[Sequence, MutableSequence], dest: bytearray, ctx: InsertContext):
-        first = self._first_value(column)
+        first = first_value(column, self.nullable)
         if isinstance(first, str):
             fixed = 24, 16, 8, 0
             # pylint: disable=consider-using-generator
@@ -39,7 +39,7 @@ def _write_column_binary(self, column: Union[Sequence, MutableSequence], dest: b
                 column = [x._ip if x else 0 for x in column]
             else:
                 column = [x._ip for x in column]
-        write_array(self._array_type, column, dest, ctx)
+        write_array(self._array_type, column, dest, ctx.column_name)
 
     def _active_null(self, ctx: QueryContext):
         fmt = self.read_format(ctx)
@@ -103,7 +103,7 @@ def _read_binary_str(source: ByteSource, num_rows: int):
 
     def _write_column_binary(self, column: Union[Sequence, MutableSequence], dest: bytearray, ctx: InsertContext):
         v = V6_NULL
-        first = self._first_value(column)
+        first = first_value(column, self.nullable)
         v4mask = IPV4_V6_MASK
         af6 = socket.AF_INET6
         tov6 = socket.inet_pton
diff --git a/clickhouse_connect/datatypes/numeric.py b/clickhouse_connect/datatypes/numeric.py
index f943b2b0..0ff62b1d 100644
--- a/clickhouse_connect/datatypes/numeric.py
+++ b/clickhouse_connect/datatypes/numeric.py
@@ -4,7 +4,7 @@
 from math import nan
 
 from clickhouse_connect.datatypes.base import TypeDef, ArrayType, ClickHouseType
-from clickhouse_connect.driver.common import array_type, write_array, decimal_size, decimal_prec
+from clickhouse_connect.driver.common import array_type, write_array, decimal_size, decimal_prec, first_value
 from clickhouse_connect.driver.ctypes import numpy_conv, data_conv
 from clickhouse_connect.driver.insert import InsertContext
 from clickhouse_connect.driver.options import pd, np
@@ -98,7 +98,7 @@ def _read_column_binary(self, source: ByteSource, num_rows: int, ctx: QueryConte
     def _write_column_binary(self, column: Union[Sequence, MutableSequence], dest: bytearray, ctx: InsertContext):
         if len(column) == 0:
             return
-        first = self._first_value(column)
+        first = first_value(column, self.nullable)
         sz = self.byte_size
         signed = self._signed
         empty = bytes(b'\x00' * sz)
@@ -190,7 +190,7 @@ def _finalize_column(self, column: Sequence, ctx: QueryContext) -> Sequence:
         return column
 
     def _write_column_binary(self, column, dest, ctx):
-        write_array('B', [1 if x else 0 for x in column], dest, ctx)
+        write_array('B', [1 if x else 0 for x in column], dest, ctx.column_name)
 
 
 class Boolean(Bool):
@@ -218,15 +218,15 @@ def _read_column_binary(self, source: ByteSource, num_rows: int, ctx: QueryConte
         lookup = self._int_map.get
         return [lookup(x, None) for x in column]
 
-    def _write_column_binary(self, column: Union[Sequence, MutableSequence], dest: bytearray, ctx):
-        first = self._first_value(column)
+    def _write_column_binary(self, column: Union[Sequence, MutableSequence], dest: bytearray, ctx:InsertContext):
+        first = first_value(column, self.nullable)
         if first is None or not isinstance(first, str):
             if self.nullable:
                 column = [0 if not x else x for x in column]
-            write_array(self._array_type, column, dest, ctx)
+            write_array(self._array_type, column, dest, ctx.column_name)
         else:
             lookup = self._name_map.get
-            write_array(self._array_type, [lookup(x, 0) for x in column], dest, ctx)
+            write_array(self._array_type, [lookup(x, 0) for x in column], dest, ctx.column_name)
 
 
 class Enum8(Enum):
@@ -285,15 +285,15 @@ def _read_column_binary(self, source: ByteSource, num_rows: int, _ctx: QueryCont
                 app(dec(f'-{digits[:-scale]}.{digits[-scale:]}'))
         return new_col
 
-    def _write_column_binary(self, column: Union[Sequence, MutableSequence], dest: bytearray, _ctx):
-        with decimal.localcontext() as ctx:
-            ctx.prec = self.prec
+    def _write_column_binary(self, column: Union[Sequence, MutableSequence], dest: bytearray, ctx:InsertContext):
+        with decimal.localcontext() as dec_ctx:
+            dec_ctx.prec = self.prec
             dec = decimal.Decimal
             mult = self._mult
             if self.nullable:
-                write_array(self._array_type, [int(dec(str(x)) * mult) if x else 0 for x in column], dest, ctx)
+                write_array(self._array_type, [int(dec(str(x)) * mult) if x else 0 for x in column], dest, ctx.column_name)
             else:
-                write_array(self._array_type, [int(dec(str(x)) * mult) for x in column], dest, ctx)
+                write_array(self._array_type, [int(dec(str(x)) * mult) for x in column], dest, ctx.column_name)
 
     def _active_null(self, ctx: QueryContext):
         if ctx.use_none:
diff --git a/clickhouse_connect/datatypes/postinit.py b/clickhouse_connect/datatypes/postinit.py
new file mode 100644
index 00000000..3536c651
--- /dev/null
+++ b/clickhouse_connect/datatypes/postinit.py
@@ -0,0 +1,4 @@
+from clickhouse_connect.datatypes import registry, dynamic
+
+dynamic.SHARED_DATA_TYPE = registry.get_from_name('Array(String, String)')
+dynamic.STRING_DATA_TYPE = registry.get_from_name('String')
diff --git a/clickhouse_connect/datatypes/registry.py b/clickhouse_connect/datatypes/registry.py
index 52d10367..6fb20d8a 100644
--- a/clickhouse_connect/datatypes/registry.py
+++ b/clickhouse_connect/datatypes/registry.py
@@ -35,6 +35,12 @@ def parse_name(name: str) -> Tuple[str, str, TypeDef]:
     elif base.startswith('Tuple'):
         keys, values = parse_columns(base[5:])
         base = 'Tuple'
+    elif base.startswith('Variant'):
+        keys, values = parse_columns(base[7:])
+        base = 'Variant'
+    elif base.startswith('JSON') and len(base) > 4 and base[4] == '(':
+        keys, values = parse_columns(base[4:])
+        base = 'JSON'
     elif base == 'Point':
         values = ('Float64', 'Float64')
     else:
diff --git a/clickhouse_connect/datatypes/special.py b/clickhouse_connect/datatypes/special.py
index 29d28e33..9d296d00 100644
--- a/clickhouse_connect/datatypes/special.py
+++ b/clickhouse_connect/datatypes/special.py
@@ -3,6 +3,7 @@
 
 from clickhouse_connect.datatypes.base import TypeDef, ClickHouseType, ArrayType, UnsupportedType
 from clickhouse_connect.datatypes.registry import get_from_name
+from clickhouse_connect.driver.common import first_value
 from clickhouse_connect.driver.ctypes import data_conv
 from clickhouse_connect.driver.insert import InsertContext
 from clickhouse_connect.driver.query import QueryContext
@@ -37,7 +38,7 @@ def _read_binary_str(source: ByteSource, num_rows: int):
 
     # pylint: disable=too-many-branches
     def _write_column_binary(self, column: Union[Sequence, MutableSequence], dest: bytearray, ctx: InsertContext):
-        first = self._first_value(column)
+        first = first_value(column, self.nullable)
         empty = empty_uuid_b
         if isinstance(first, str) or self.write_format(ctx) == 'string':
             for v in column:
@@ -92,8 +93,8 @@ def __init__(self, type_def: TypeDef):
     def _data_size(self, sample: Sequence) -> int:
         return self.element_type.data_size(sample)
 
-    def read_column_prefix(self, source: ByteSource):
-        return self.element_type.read_column_prefix(source)
+    def read_column_prefix(self, source: ByteSource, ctx: QueryContext):
+        return self.element_type.read_column_prefix(source, ctx)
 
     def write_column_prefix(self, dest: bytearray):
         self.element_type.write_column_prefix(dest)
diff --git a/clickhouse_connect/datatypes/string.py b/clickhouse_connect/datatypes/string.py
index 897b5ec6..4b788651 100644
--- a/clickhouse_connect/datatypes/string.py
+++ b/clickhouse_connect/datatypes/string.py
@@ -1,5 +1,6 @@
 from typing import Sequence, MutableSequence, Union, Collection
 
+from clickhouse_connect.driver.common import first_value
 from clickhouse_connect.driver.ctypes import data_conv
 
 from clickhouse_connect.datatypes.base import ClickHouseType, TypeDef
@@ -44,9 +45,9 @@ def _finalize_column(self, column: Sequence, ctx: QueryContext) -> Sequence:
 
     def _write_column_binary(self, column: Union[Sequence, MutableSequence], dest: bytearray, ctx: InsertContext):
         encoding = None
-        if not isinstance(self._first_value(column), bytes):
+        if not isinstance(first_value(column, self.nullable), bytes):
             encoding = ctx.encoding or self.encoding
-        handle_error(data_conv.write_str_col(column, self.nullable, encoding, dest))
+        handle_error(data_conv.write_str_col(column, self.nullable, encoding, dest), ctx)
 
     def _active_null(self, ctx):
         if ctx.use_none:
@@ -91,7 +92,7 @@ def _write_column_binary(self, column: Union[Sequence, MutableSequence], dest: b
         empty = bytes((0,) * sz)
         str_enc = str.encode
         enc = ctx.encoding or self.encoding
-        first = self._first_value(column)
+        first = first_value(column, self.nullable)
         if isinstance(first, str) or self.write_format(ctx) == 'string':
             if self.nullable:
                 for x in column:
@@ -103,7 +104,7 @@ def _write_column_binary(self, column: Union[Sequence, MutableSequence], dest: b
                         except UnicodeEncodeError:
                             b = empty
                         if len(b) > sz:
-                            raise ctx.make_data_error(f'UTF-8 encoded FixedString value {b.hex(" ")} exceeds column size {sz}')
+                            raise ctx.data_error(f'UTF-8 encoded FixedString value {b.hex(" ")} exceeds column size {sz}')
                         ext(b)
                         ext(empty[:sz - len(b)])
             else:
@@ -113,7 +114,7 @@ def _write_column_binary(self, column: Union[Sequence, MutableSequence], dest: b
                     except UnicodeEncodeError:
                         b = empty
                     if len(b) > sz:
-                        raise ctx.make_data_error(f'UTF-8 encoded FixedString value {b.hex(" ")} exceeds column size {sz}')
+                        raise ctx.data_error(f'UTF-8 encoded FixedString value {b.hex(" ")} exceeds column size {sz}')
                     ext(b)
                     ext(empty[:sz - len(b)])
         elif self.nullable:
@@ -121,11 +122,11 @@ def _write_column_binary(self, column: Union[Sequence, MutableSequence], dest: b
                 if not b:
                     ext(empty)
                 elif len(b) != sz:
-                    raise ctx.make_data_error(f'Fixed String binary value {b.hex(" ")} does not match column size {sz}')
+                    raise ctx.data_error(f'Fixed String binary value {b.hex(" ")} does not match column size {sz}')
                 else:
                     ext(b)
         else:
             for b in column:
                 if len(b) != sz:
-                    raise ctx.make_data_error(f'Fixed String binary value {b.hex(" ")} does not match column size {sz}')
+                    raise ctx.data_error(f'Fixed String binary value {b.hex(" ")} does not match column size {sz}')
                 ext(b)
diff --git a/clickhouse_connect/datatypes/temporal.py b/clickhouse_connect/datatypes/temporal.py
index 76ef04e5..6359d5ba 100644
--- a/clickhouse_connect/datatypes/temporal.py
+++ b/clickhouse_connect/datatypes/temporal.py
@@ -4,7 +4,7 @@
 from typing import Union, Sequence, MutableSequence
 
 from clickhouse_connect.datatypes.base import TypeDef, ClickHouseType
-from clickhouse_connect.driver.common import write_array, np_date_types, int_size
+from clickhouse_connect.driver.common import write_array, np_date_types, int_size, first_value
 from clickhouse_connect.driver.exceptions import ProgrammingError
 from clickhouse_connect.driver.ctypes import data_conv, numpy_conv
 from clickhouse_connect.driver.insert import InsertContext
@@ -32,7 +32,7 @@ def _read_column_binary(self, source: ByteSource, num_rows: int, ctx: QueryConte
         return data_conv.read_date_col(source, num_rows)
 
     def _write_column_binary(self, column: Union[Sequence, MutableSequence], dest: bytearray, ctx: InsertContext):
-        first = self._first_value(column)
+        first = first_value(column, self.nullable)
         if isinstance(first, int) or self.write_format(ctx) == 'int':
             if self.nullable:
                 column = [x if x else 0 for x in column]
@@ -45,7 +45,7 @@ def _write_column_binary(self, column: Union[Sequence, MutableSequence], dest: b
                 column = [0 if x is None else (x - esd).days for x in column]
             else:
                 column = [(x - esd).days for x in column]
-        write_array(self._array_type, column, dest, ctx)
+        write_array(self._array_type, column, dest, ctx.column_name)
 
     def _active_null(self, ctx: QueryContext):
         fmt = self.read_format(ctx)
@@ -127,7 +127,7 @@ def _read_column_binary(self, source: ByteSource, num_rows: int, ctx: QueryConte
         return data_conv.read_datetime_col(source, num_rows, active_tz)
 
     def _write_column_binary(self, column: Union[Sequence, MutableSequence], dest: bytearray, ctx: InsertContext):
-        first = self._first_value(column)
+        first = first_value(column, self.nullable)
         if isinstance(first, int) or self.write_format(ctx) == 'int':
             if self.nullable:
                 column = [x if x else 0 for x in column]
@@ -136,7 +136,7 @@ def _write_column_binary(self, column: Union[Sequence, MutableSequence], dest: b
                 column = [int(x.timestamp()) if x else 0 for x in column]
             else:
                 column = [int(x.timestamp()) for x in column]
-        write_array(self._array_type, column, dest, ctx)
+        write_array(self._array_type, column, dest, ctx.column_name)
 
 
 class DateTime64(DateTimeBase):
@@ -202,7 +202,7 @@ def _read_binary_naive(self, column: Sequence):
         return new_col
 
     def _write_column_binary(self, column: Union[Sequence, MutableSequence], dest: bytearray, ctx: InsertContext):
-        first = self._first_value(column)
+        first = first_value(column, self.nullable)
         if isinstance(first, int) or self.write_format(ctx) == 'int':
             if self.nullable:
                 column = [x if x else 0 for x in column]
@@ -213,4 +213,4 @@ def _write_column_binary(self, column: Union[Sequence, MutableSequence], dest: b
                           for x in column]
             else:
                 column = [((int(x.timestamp()) * 1000000 + x.microsecond) * prec) // 1000000 for x in column]
-        write_array('q', column, dest, ctx)
+        write_array('q', column, dest, ctx.column_name)
diff --git a/clickhouse_connect/driver/__init__.py b/clickhouse_connect/driver/__init__.py
index 27e16733..fe3e4b3f 100644
--- a/clickhouse_connect/driver/__init__.py
+++ b/clickhouse_connect/driver/__init__.py
@@ -84,7 +84,7 @@ def create_client(*,
         database = database or parsed.path
         for k, v in parse_qs(parsed.query).items():
             kwargs[k] = v[0]
-    use_tls = str(secure).lower() == 'true' or interface == 'https' or (not interface and port in (443, 8443))
+    use_tls = str(secure).lower() == 'true' or interface == 'https' or (not interface and str(port) in ('443', '8443'))
     if not host:
         host = 'localhost'
     if not interface:
diff --git a/clickhouse_connect/driver/asyncclient.py b/clickhouse_connect/driver/asyncclient.py
index e2dc5b01..bf56c978 100644
--- a/clickhouse_connect/driver/asyncclient.py
+++ b/clickhouse_connect/driver/asyncclient.py
@@ -13,7 +13,7 @@
 from clickhouse_connect.driver.insert import InsertContext
 
 
-# pylint: disable=too-many-public-methods, too-many-instance-attributes, too-many-arguments, too-many-locals
+# pylint: disable=too-many-public-methods,too-many-instance-attributes,too-many-arguments,too-many-positional-arguments,too-many-locals
 class AsyncClient:
     """
     AsyncClient is a wrapper around the ClickHouse Client object that allows for async calls to the ClickHouse server.
diff --git a/clickhouse_connect/driver/binding.py b/clickhouse_connect/driver/binding.py
new file mode 100644
index 00000000..24522b08
--- /dev/null
+++ b/clickhouse_connect/driver/binding.py
@@ -0,0 +1,199 @@
+import ipaddress
+import re
+import uuid
+from datetime import tzinfo, datetime, date
+from enum import Enum
+from typing import Optional, Union, Sequence, Dict, Any, Tuple
+
+import pytz
+
+from clickhouse_connect import common
+from clickhouse_connect.driver.common import dict_copy
+from clickhouse_connect.json_impl import any_to_json
+
+BS = '\\'
+must_escape = (BS, '\'', '`', '\t', '\n')
+external_bind_re = re.compile(r'{.+:.+}')
+
+
+class DT64Param:
+    def __init__(self, value: datetime):
+        self.value = value
+
+    def format(self, tz: tzinfo, top_level:bool) -> str:
+        value = self.value
+        if tz:
+            value = value.astimezone(tz)
+        s = value.strftime('%Y-%m-%d %H:%M:%S.%f')
+        if top_level:
+            return s
+        return f"'{s}'"
+
+
+def quote_identifier(identifier: str):
+    first_char = identifier[0]
+    if first_char in ('`', '"') and identifier[-1] == first_char:
+        # Identifier is already quoted, assume that it's valid
+        return identifier
+    return f'`{escape_str(identifier)}`'
+
+
+def finalize_query(query: str, parameters: Optional[Union[Sequence, Dict[str, Any]]],
+                   server_tz: Optional[tzinfo] = None) -> str:
+    while query.endswith(';'):
+        query = query[:-1]
+    if not parameters:
+        return query
+    if hasattr(parameters, 'items'):
+        return query % {k: format_query_value(v, server_tz) for k, v in parameters.items()}
+    return query % tuple(format_query_value(v, server_tz) for v in parameters)
+
+
+# pylint: disable=too-many-locals,too-many-branches
+def bind_query(query: str, parameters: Optional[Union[Sequence, Dict[str, Any]]],
+               server_tz: Optional[tzinfo] = None) -> Tuple[str, Dict[str, str]]:
+    while query.endswith(';'):
+        query = query[:-1]
+    if not parameters:
+        return query, {}
+
+    binary_binds = None
+
+    if isinstance(parameters, dict):
+        params_copy = dict_copy(parameters)
+        binary_binds = {k: v for k, v in params_copy.items() if k.startswith('$') and k.endswith('$') and len(k) > 1}
+        for key in binary_binds.keys():
+            del params_copy[key]
+
+        final_params = {}
+        for k, v in params_copy.items():
+            if k.endswith('_64'):
+                if isinstance(v, datetime):
+                    k = k[:-3]
+                    v = DT64Param(v)
+                elif isinstance(v, list) and len(v) > 0 and isinstance(v[0], datetime):
+                    k = k[:-3]
+                    v = [DT64Param(x) for x in v]
+            final_params[k] = v
+        if external_bind_re.search(query) is None:
+            query, bound_params = finalize_query(query, final_params, server_tz), {}
+        else:
+            bound_params = {f'param_{k}': format_bind_value(v, server_tz) for k, v in final_params.items()}
+    else:
+        query, bound_params = finalize_query(query, parameters, server_tz), {}
+    if binary_binds:
+        binary_query = query.encode()
+        binary_indexes = {}
+        for k, v in binary_binds.items():
+            key = k.encode()
+            item_index = 0
+            while True:
+                item_index = binary_query.find(key, item_index)
+                if item_index == -1:
+                    break
+                binary_indexes[item_index + len(key)] = key, v
+                item_index += len(key)
+        query = b''
+        start = 0
+        for loc in sorted(binary_indexes.keys()):
+            key, value = binary_indexes[loc]
+            query += binary_query[start:loc] + value + key
+            start = loc
+        query += binary_query[start:]
+    return query, bound_params
+
+
+def format_str(value: str):
+    return f"'{escape_str(value)}'"
+
+
+def escape_str(value: str):
+    return ''.join(f'{BS}{c}' if c in must_escape else c for c in value)
+
+
+# pylint: disable=too-many-return-statements
+def format_query_value(value: Any, server_tz: tzinfo = pytz.UTC):
+    """
+    Format Python values in a ClickHouse query
+    :param value: Python object
+    :param server_tz: Server timezone for adjusting datetime values
+    :return: Literal string for python value
+    """
+    if value is None:
+        return 'NULL'
+    if isinstance(value, str):
+        return format_str(value)
+    if isinstance(value, DT64Param):
+        return value.format(server_tz, False)
+    if isinstance(value, datetime):
+        if value.tzinfo is not None or server_tz != pytz.UTC:
+            value = value.astimezone(server_tz)
+        return f"'{value.strftime('%Y-%m-%d %H:%M:%S')}'"
+    if isinstance(value, date):
+        return f"'{value.isoformat()}'"
+    if isinstance(value, list):
+        return f"[{', '.join(str_query_value(x, server_tz) for x in value)}]"
+    if isinstance(value, tuple):
+        return f"({', '.join(str_query_value(x, server_tz) for x in value)})"
+    if isinstance(value, dict):
+        if common.get_setting('dict_parameter_format') == 'json':
+            return format_str(any_to_json(value).decode())
+        pairs = [str_query_value(k, server_tz) + ':' + str_query_value(v, server_tz)
+                 for k, v in value.items()]
+        return f"{{{', '.join(pairs)}}}"
+    if isinstance(value, Enum):
+        return format_query_value(value.value, server_tz)
+    if isinstance(value, (uuid.UUID, ipaddress.IPv4Address, ipaddress.IPv6Address)):
+        return f"'{value}'"
+    return value
+
+
+def str_query_value(value: Any, server_tz: tzinfo = pytz.UTC):
+    return str(format_query_value(value, server_tz))
+
+
+# pylint: disable=too-many-branches
+def format_bind_value(value: Any, server_tz: tzinfo = pytz.UTC, top_level: bool = True):
+    """
+    Format Python values in a ClickHouse query
+    :param value: Python object
+    :param server_tz: Server timezone for adjusting datetime values
+    :param top_level: Flag for top level for nested structures
+    :return: Literal string for python value
+    """
+
+    def recurse(x):
+        return format_bind_value(x, server_tz, False)
+
+    if value is None:
+        return '\\N'
+    if isinstance(value, str):
+        if top_level:
+            # At the top levels, strings must not be surrounded by quotes
+            return escape_str(value)
+        return format_str(value)
+    if isinstance(value, DT64Param):
+        return value.format(server_tz, top_level)
+    if isinstance(value, datetime):
+        value = value.astimezone(server_tz)
+        val = value.strftime('%Y-%m-%d %H:%M:%S')
+        if top_level:
+            return val
+        return f"'{val}'"
+    if isinstance(value, date):
+        if top_level:
+            return value.isoformat()
+        return f"'{value.isoformat()}'"
+    if isinstance(value, list):
+        return f"[{', '.join(recurse(x) for x in value)}]"
+    if isinstance(value, tuple):
+        return f"({', '.join(recurse(x) for x in value)})"
+    if isinstance(value, dict):
+        if common.get_setting('dict_parameter_format') == 'json':
+            return any_to_json(value).decode()
+        pairs = [recurse(k) + ':' + recurse(v)
+                 for k, v in value.items()]
+        return f"{{{', '.join(pairs)}}}"
+    if isinstance(value, Enum):
+        return recurse(value.value)
+    return str(value)
diff --git a/clickhouse_connect/driver/client.py b/clickhouse_connect/driver/client.py
index ba3873cb..fe11c278 100644
--- a/clickhouse_connect/driver/client.py
+++ b/clickhouse_connect/driver/client.py
@@ -20,15 +20,15 @@
 from clickhouse_connect.driver.insert import InsertContext
 from clickhouse_connect.driver.summary import QuerySummary
 from clickhouse_connect.driver.models import ColumnDef, SettingDef, SettingStatus
-from clickhouse_connect.driver.query import QueryResult, to_arrow, to_arrow_batches, QueryContext, arrow_buffer, \
-    quote_identifier
+from clickhouse_connect.driver.query import QueryResult, to_arrow, to_arrow_batches, QueryContext, arrow_buffer
+from clickhouse_connect.driver.binding import quote_identifier
 
 io.DEFAULT_BUFFER_SIZE = 1024 * 256
 logger = logging.getLogger(__name__)
 arrow_str_setting = 'output_format_arrow_string_as_string'
 
 
-# pylint: disable=too-many-public-methods, too-many-instance-attributes
+# pylint: disable=too-many-public-methods,too-many-arguments,too-many-positional-arguments,too-many-instance-attributes
 class Client(ABC):
     """
     Base ClickHouse Connect client
@@ -93,6 +93,8 @@ def __init__(self,
             })
             if test_data[8:16] == b'\x01\x01\x05check':
                 self.protocol_version = PROTOCOL_VERSION_WITH_LOW_CARD
+        if self._setting_status('date_time_input_format').is_writable:
+            self.set_client_setting('date_time_input_format', 'best_effort')
         self.uri = uri
 
     def _validate_settings(self, settings: Optional[Dict[str, Any]]) -> Dict[str, str]:
@@ -171,7 +173,7 @@ def get_client_setting(self, key) -> Optional[str]:
         :return: The string value of the setting, if it exists, or None
         """
 
-    # pylint: disable=too-many-arguments,unused-argument,too-many-locals
+    # pylint: disable=unused-argument,too-many-locals
     def query(self,
               query: Optional[str] = None,
               parameters: Optional[Union[Sequence, Dict[str, Any]]] = None,
@@ -303,7 +305,7 @@ def raw_stream(self, query: str,
        :return: io.IOBase stream/iterator for the result
        """
 
-    # pylint: disable=duplicate-code,too-many-arguments,unused-argument
+    # pylint: disable=duplicate-code,unused-argument
     def query_np(self,
                  query: Optional[str] = None,
                  parameters: Optional[Union[Sequence, Dict[str, Any]]] = None,
@@ -341,7 +343,7 @@ def query_np_stream(self,
         """
         return self._context_query(locals(), use_numpy=True, streaming=True).np_stream
 
-    # pylint: disable=duplicate-code,too-many-arguments,unused-argument
+    # pylint: disable=duplicate-code,unused-argument
     def query_df(self,
                  query: Optional[str] = None,
                  parameters: Optional[Union[Sequence, Dict[str, Any]]] = None,
@@ -364,7 +366,7 @@ def query_df(self,
         """
         return self._context_query(locals(), use_numpy=True, as_pandas=True).df_result
 
-    # pylint: disable=duplicate-code,too-many-arguments,unused-argument
+    # pylint: disable=duplicate-code,unused-argument
     def query_df_stream(self,
                         query: Optional[str] = None,
                         parameters: Optional[Union[Sequence, Dict[str, Any]]] = None,
@@ -571,7 +573,6 @@ def ping(self) -> bool:
         :return: ClickHouse server is up and reachable
         """
 
-    # pylint: disable=too-many-arguments
     def insert(self,
                table: Optional[str] = None,
                data: Sequence[Sequence[Any]] = None,
@@ -777,11 +778,18 @@ def raw_insert(self, table: str,
         :param fmt: Valid clickhouse format
         """
 
+    @abstractmethod
     def close(self):
         """
         Subclass implementation to close the connection to the server/deallocate the client
         """
 
+    @abstractmethod
+    def close_connections(self):
+        """
+        Subclass implementation to disconnect all "re-used" client connections
+        """
+
     def _context_query(self, lcls: dict, **overrides):
         kwargs = lcls.copy()
         kwargs.pop('self')
diff --git a/clickhouse_connect/driver/common.py b/clickhouse_connect/driver/common.py
index 4e6149ab..404a09e8 100644
--- a/clickhouse_connect/driver/common.py
+++ b/clickhouse_connect/driver/common.py
@@ -2,13 +2,11 @@
 import struct
 import sys
 
-from typing import Sequence, MutableSequence, Dict, Optional, Union, Generator, TYPE_CHECKING
+from typing import Sequence, MutableSequence, Dict, Optional, Union, Generator
 
-from clickhouse_connect.driver.exceptions import ProgrammingError, StreamClosedError
+from clickhouse_connect.driver.exceptions import ProgrammingError, StreamClosedError, DataError
 from clickhouse_connect.driver.types import Closable
 
-if TYPE_CHECKING:
-    from clickhouse_connect.driver.insert import InsertContext
 
 # pylint: disable=invalid-name
 must_swap = sys.byteorder == 'big'
@@ -41,13 +39,13 @@ def array_type(size: int, signed: bool):
     return code if signed else code.upper()
 
 
-def write_array(code: str, column: Sequence, dest: MutableSequence, ctx: 'InsertContext'):
+def write_array(code: str, column: Sequence, dest: MutableSequence, col_name: Optional[str]=None):
     """
     Write a column of native Python data matching the array.array code
     :param code: Python array.array code matching the column data type
     :param column: Column of native Python values
     :param dest: Destination byte buffer
-    :param ctx: The InsertContext
+    :param col_name: Optional column name for error tracking
     """
     if len(column) and not isinstance(column[0], (int, float)):
         if code in ('f', 'F', 'd', 'D'):
@@ -58,7 +56,10 @@ def write_array(code: str, column: Sequence, dest: MutableSequence, ctx: 'Insert
         buff = struct.Struct(f'<{len(column)}{code}')
         dest += buff.pack(*column)
     except (TypeError, OverflowError, struct.error) as ex:
-        raise ctx.make_data_error('Unable to create Python array.  This is usually caused by trying to insert None ' +
+        col_msg = ''
+        if col_name:
+            col_msg = f' for source column `{col_name}`'
+        raise DataError(f'Unable to create Python array{col_msg}.  This is usually caused by trying to insert None ' +
                                   'values into a ClickHouse column that is not Nullable') from ex
 
 
@@ -138,6 +139,14 @@ def coerce_bool(val: Optional[Union[str, bool]]):
     return val is True or (isinstance(val, str) and val.lower() in ('true', '1', 'y', 'yes'))
 
 
+def first_value(column: Sequence, nullable:bool = True):
+    if nullable:
+        return next((x for x in column if x is not None), None)
+    if len(column):
+        return column[0]
+    return None
+
+
 class SliceView(Sequence):
     """
     Provides a view into a sequence rather than copying.  Borrows liberally from
diff --git a/clickhouse_connect/driver/context.py b/clickhouse_connect/driver/context.py
index 4f4835c8..00125a0b 100644
--- a/clickhouse_connect/driver/context.py
+++ b/clickhouse_connect/driver/context.py
@@ -36,15 +36,15 @@ def __init__(self,
                                                        for type_name, fmt in fmt.items()}
         self.query_formats = query_formats or {}
         self.column_formats = column_formats or {}
+        self.column_name = None
         self.encoding = encoding
         self.use_numpy = use_numpy
         self.use_extended_dtypes = use_extended_dtypes
         self._active_col_fmt = None
         self._active_col_type_fmts = _empty_map
-        self._column_name = None
 
     def start_column(self, name: str):
-        self._column_name = name
+        self.column_name = name
         self._active_col_fmt = self.col_simple_formats.get(name)
         self._active_col_type_fmts = self.col_type_formats.get(name, _empty_map)
 
diff --git a/clickhouse_connect/driver/ctypes.py b/clickhouse_connect/driver/ctypes.py
index e7bb607e..39e5954c 100644
--- a/clickhouse_connect/driver/ctypes.py
+++ b/clickhouse_connect/driver/ctypes.py
@@ -46,4 +46,4 @@ def connect_numpy():
              str(ex))
 
 
-connect_c_modules()
+# connect_c_modules()
diff --git a/clickhouse_connect/driver/errors.py b/clickhouse_connect/driver/errors.py
index 0d37890b..e5b3b646 100644
--- a/clickhouse_connect/driver/errors.py
+++ b/clickhouse_connect/driver/errors.py
@@ -1,3 +1,4 @@
+from clickhouse_connect.driver.context import BaseQueryContext
 from clickhouse_connect.driver.exceptions import DataError
 
 
@@ -8,6 +9,9 @@
 error_messages = {NONE_IN_NULLABLE_COLUMN: 'Invalid None value in non-Nullable column'}
 
 
-def handle_error(error_num: int):
+def handle_error(error_num: int, ctx: BaseQueryContext):
     if error_num > 0:
-        raise DataError(error_messages[error_num])
+        msg = error_messages[error_num]
+        if ctx.column_name:
+            msg = f'{msg}, column name: `{ctx.column_name}`'
+        raise DataError(msg)
diff --git a/clickhouse_connect/driver/httpclient.py b/clickhouse_connect/driver/httpclient.py
index bf3c0d86..001c578b 100644
--- a/clickhouse_connect/driver/httpclient.py
+++ b/clickhouse_connect/driver/httpclient.py
@@ -24,7 +24,8 @@
 from clickhouse_connect.driver.httputil import ResponseSource, get_pool_manager, get_response_data, \
     default_pool_manager, get_proxy_manager, all_managers, check_env_proxy, check_conn_expiration
 from clickhouse_connect.driver.insert import InsertContext
-from clickhouse_connect.driver.query import QueryResult, QueryContext, quote_identifier, bind_query
+from clickhouse_connect.driver.query import QueryResult, QueryContext
+from clickhouse_connect.driver.binding import quote_identifier, bind_query
 from clickhouse_connect.driver.summary import QuerySummary
 from clickhouse_connect.driver.transform import NativeTransform
 
@@ -44,7 +45,7 @@ class HttpClient(Client):
                                    'enable_http_compression'}
     _owns_pool_manager = False
 
-    # pylint: disable=too-many-arguments,too-many-locals,too-many-branches,too-many-statements,unused-argument
+    # pylint: disable=too-many-positional-arguments,too-many-arguments,too-many-locals,too-many-branches,too-many-statements,unused-argument
     def __init__(self,
                  interface: str,
                  host: str,
@@ -70,7 +71,8 @@ def __init__(self,
                  server_host_name: Optional[str] = None,
                  apply_server_timezone: Optional[Union[str, bool]] = None,
                  show_clickhouse_errors: Optional[bool] = None,
-                 autogenerate_session_id: Optional[bool] = None):
+                 autogenerate_session_id: Optional[bool] = None,
+                 tls_mode: Optional[str] = None):
         """
         Create an HTTP ClickHouse Connect client
         See clickhouse_connect.get_client for parameters
@@ -80,20 +82,20 @@ def __init__(self,
         ch_settings = dict_copy(settings, self.params)
         self.http = pool_mgr
         if interface == 'https':
+            if isinstance(verify, str) and verify.lower() == 'proxy':
+                verify = True
+                tls_mode = tls_mode or 'proxy'
             if not https_proxy:
                 https_proxy = check_env_proxy('https', host, port)
-            if https_proxy and isinstance(verify, str) and verify.lower() == 'proxy':
-                verify = 'proxy'
-            else:
-                verify = coerce_bool(verify)
-            if client_cert and verify != 'proxy':
+            verify = coerce_bool(verify)
+            if client_cert and (tls_mode is None or tls_mode == 'mutual'):
                 if not username:
                     raise ProgrammingError('username parameter is required for Mutual TLS authentication')
                 self.headers['X-ClickHouse-User'] = username
                 self.headers['X-ClickHouse-SSL-Certificate-Auth'] = 'on'
             # pylint: disable=too-many-boolean-expressions
             if not self.http and (server_host_name or ca_cert or client_cert or not verify or https_proxy):
-                options = {'verify': verify is not False}
+                options = {'verify': verify}
                 dict_add(options, 'ca_cert', ca_cert)
                 dict_add(options, 'client_cert', client_cert)
                 dict_add(options, 'client_cert_key', client_cert_key)
@@ -111,7 +113,7 @@ def __init__(self,
             else:
                 self.http = default_pool_manager()
 
-        if (not client_cert or verify == 'proxy') and username:
+        if (not client_cert or tls_mode in ('strict', 'proxy')) and username:
             self.headers['Authorization'] = 'Basic ' + b64encode(f'{username}:{password}'.encode()).decode()
         self.headers['User-Agent'] = common.build_client_name(client_name)
         self._read_format = self._write_format = 'Native'
@@ -157,7 +159,7 @@ def __init__(self,
                          server_host_name=server_host_name,
                          apply_server_timezone=apply_server_timezone,
                          show_clickhouse_errors=show_clickhouse_errors)
-        self.params = self._validate_settings(ch_settings)
+        self.params = dict_copy(self.params, self._validate_settings(ch_settings))
         comp_setting = self._setting_status('enable_http_compression')
         self._send_comp_setting = not comp_setting.is_set and comp_setting.is_writable
         if comp_setting.is_set or comp_setting.is_writable:
@@ -195,7 +197,7 @@ def _query_with_context(self, context: QueryContext) -> QueryResult:
             context.block_info = True
         params.update(context.bind_params)
         params.update(self._validate_settings(context.settings))
-        if columns_only_re.search(context.uncommented_query):
+        if not context.is_insert and columns_only_re.search(context.uncommented_query):
             response = self._raw_request(f'{context.final_query}\n FORMAT JSON',
                                          params, headers, retries=self.query_retries)
             json_result = json.loads(response.data)
@@ -513,6 +515,9 @@ def ping(self):
             logger.debug('ping failed', exc_info=True)
             return False
 
+    def close_connections(self):
+        self.http.clear()
+
     def close(self):
         if self._owns_pool_manager:
             self.http.clear()
diff --git a/clickhouse_connect/driver/httputil.py b/clickhouse_connect/driver/httputil.py
index 9a2b835c..7dd73114 100644
--- a/clickhouse_connect/driver/httputil.py
+++ b/clickhouse_connect/driver/httputil.py
@@ -6,7 +6,8 @@
 import sys
 import socket
 import time
-from typing import Dict, Any, Optional
+from collections import deque
+from typing import Dict, Any, Optional, Tuple, Callable
 
 import certifi
 import lz4.frame
@@ -192,34 +193,56 @@ class ResponseSource:
     def __init__(self, response: HTTPResponse, chunk_size: int = 1024 * 1024):
         self.response = response
         compression = response.headers.get('content-encoding')
+        decompress:Optional[Callable] = None
         if compression == 'zstd':
             zstd_decom = zstandard.ZstdDecompressor().decompressobj()
 
-            def decompress():
-                while True:
-                    chunk = response.read(chunk_size, decode_content=False)
-                    if not chunk:
-                        break
-                    yield zstd_decom.decompress(chunk)
+            def zstd_decompress(c: deque) -> Tuple[bytes, int]:
+                chunk = c.popleft()
+                return zstd_decom.decompress(chunk), len(chunk)
 
-            self.gen = decompress()
+            decompress = zstd_decompress
         elif compression == 'lz4':
             lz4_decom = lz4.frame.LZ4FrameDecompressor()
 
-            def decompress():
+            def lz_decompress(c: deque) -> Tuple[Optional[bytes], int]:
+                read_amt = 0
                 while lz4_decom.needs_input:
-                    data = self.response.read(chunk_size, decode_content=False)
+                    data = c.popleft()
+                    read_amt += len(data)
                     if lz4_decom.unused_data:
                         data = lz4_decom.unused_data + data
-                    if not data:
-                        return
-                    chunk = lz4_decom.decompress(data)
-                    if chunk:
-                        yield chunk
-
-            self.gen = decompress()
-        else:
-            self.gen = response.stream(amt=chunk_size, decode_content=True)
+                    return lz4_decom.decompress(data), read_amt
+                return None, 0
+
+            decompress = lz_decompress
+
+        buffer_size = common.get_setting('http_buffer_size')
+
+        def buffered():
+            chunks = deque()
+            done = False
+            current_size = 0
+            read_gen = response.read_chunked(chunk_size, decompress is None)
+            while True:
+                while not done and current_size < buffer_size:
+                    chunk = next(read_gen, None)
+                    if not chunk:
+                        done = True
+                        break
+                    chunks.append(chunk)
+                    current_size += len(chunk)
+                if len(chunks) == 0:
+                    return
+                if decompress:
+                    chunk, used = decompress(chunks)
+                    current_size -= used
+                else:
+                    chunk = chunks.popleft()
+                    current_size -= len(chunk)
+                yield chunk
+
+        self.gen = buffered()
 
     def close(self):
         self.response.drain_conn()
diff --git a/clickhouse_connect/driver/insert.py b/clickhouse_connect/driver/insert.py
index 701e0d88..8ca1ef9f 100644
--- a/clickhouse_connect/driver/insert.py
+++ b/clickhouse_connect/driver/insert.py
@@ -2,7 +2,7 @@
 from math import log
 from typing import Iterable, Sequence, Optional, Any, Dict, NamedTuple, Generator, Union, TYPE_CHECKING
 
-from clickhouse_connect.driver.query import quote_identifier
+from clickhouse_connect.driver.binding import quote_identifier
 
 from clickhouse_connect.driver.ctypes import data_conv
 from clickhouse_connect.driver.context import BaseQueryContext
@@ -199,5 +199,5 @@ def _convert_numpy(self, np_array):
         self.column_oriented = True
         return data
 
-    def make_data_error(self, error_message: str) -> DataError:
-        return DataError(f"Failed to write column '{self._column_name}': {error_message}")
+    def data_error(self, error_message: str) -> DataError:
+        return DataError(f"Failed to write column '{self.column_name}': {error_message}")
diff --git a/clickhouse_connect/driver/parser.py b/clickhouse_connect/driver/parser.py
index acdf9510..02bdc03c 100644
--- a/clickhouse_connect/driver/parser.py
+++ b/clickhouse_connect/driver/parser.py
@@ -142,7 +142,7 @@ def parse_columns(expr: str):
                 pos += 1
         else:
             if level == 0:
-                if char == ' ':
+                if char in (' ', '='):
                     if label and not named:
                         names.append(unescape_identifier(label))
                         label = ''
diff --git a/clickhouse_connect/driver/query.py b/clickhouse_connect/driver/query.py
index c11e128e..54edbeff 100644
--- a/clickhouse_connect/driver/query.py
+++ b/clickhouse_connect/driver/query.py
@@ -1,22 +1,18 @@
-import ipaddress
 import logging
 import re
-import uuid
 import pytz
 
-from enum import Enum
 from io import IOBase
 from typing import Any, Tuple, Dict, Sequence, Optional, Union, Generator
-from datetime import date, datetime, tzinfo
+from datetime import tzinfo
 
 from pytz.exceptions import UnknownTimeZoneError
 
-from clickhouse_connect import common
 from clickhouse_connect.driver import tzutil
+from clickhouse_connect.driver.binding import bind_query
 from clickhouse_connect.driver.common import dict_copy, empty_gen, StreamContext
 from clickhouse_connect.driver.external import ExternalData
 from clickhouse_connect.driver.types import Matrix, Closable
-from clickhouse_connect.json_impl import any_to_json
 from clickhouse_connect.driver.exceptions import StreamClosedError, ProgrammingError
 from clickhouse_connect.driver.options import check_arrow, pd_extended_dtypes
 from clickhouse_connect.driver.context import BaseQueryContext
@@ -29,7 +25,6 @@
 select_re = re.compile(r'(^|\s)SELECT\s', re.IGNORECASE)
 insert_re = re.compile(r'(^|\s)INSERT\s*INTO', re.IGNORECASE)
 command_re = re.compile(r'(^\s*)(' + commands + r')\s', re.IGNORECASE)
-external_bind_re = re.compile(r'{.+:.+}')
 
 
 # pylint: disable=too-many-instance-attributes
@@ -38,7 +33,7 @@ class QueryContext(BaseQueryContext):
     Argument/parameter object for queries.  This context is used to set thread/query specific formats
     """
 
-    # pylint: disable=duplicate-code,too-many-arguments,too-many-locals
+    # pylint: disable=duplicate-code,too-many-arguments,too-many-positional-arguments,too-many-locals
     def __init__(self,
                  query: Union[str, bytes] = '',
                  parameters: Optional[Dict[str, Any]] = None,
@@ -176,6 +171,7 @@ def active_tz(self, datatype_tz: Optional[tzinfo]):
             return None
         return active_tz
 
+    # pylint disable=too-many-positional-arguments
     def updated_copy(self,
                      query: Optional[Union[str, bytes]] = None,
                      parameters: Optional[Dict[str, Any]] = None,
@@ -343,159 +339,6 @@ def close(self):
             self._block_gen = None
 
 
-BS = '\\'
-must_escape = (BS, '\'', '`', '\t', '\n')
-
-
-def quote_identifier(identifier: str):
-    first_char = identifier[0]
-    if first_char in ('`', '"') and identifier[-1] == first_char:
-        # Identifier is already quoted, assume that it's valid
-        return identifier
-    return f'`{escape_str(identifier)}`'
-
-
-def finalize_query(query: str, parameters: Optional[Union[Sequence, Dict[str, Any]]],
-                   server_tz: Optional[tzinfo] = None) -> str:
-    while query.endswith(';'):
-        query = query[:-1]
-    if not parameters:
-        return query
-    if hasattr(parameters, 'items'):
-        return query % {k: format_query_value(v, server_tz) for k, v in parameters.items()}
-    return query % tuple(format_query_value(v) for v in parameters)
-
-
-def bind_query(query: str, parameters: Optional[Union[Sequence, Dict[str, Any]]],
-               server_tz: Optional[tzinfo] = None) -> Tuple[str, Dict[str, str]]:
-    while query.endswith(';'):
-        query = query[:-1]
-    if not parameters:
-        return query, {}
-
-    binary_binds = None
-    if isinstance(parameters, dict):
-        binary_binds = {k: v for k, v in parameters.items() if k.startswith('$') and k.endswith('$') and len(k) > 1}
-        for key in binary_binds.keys():
-            del parameters[key]
-    if external_bind_re.search(query) is None:
-        query, bound_params = finalize_query(query, parameters, server_tz), {}
-    else:
-        bound_params = {f'param_{k}': format_bind_value(v, server_tz) for k, v in parameters.items()}
-    if binary_binds:
-        binary_query = query.encode()
-        binary_indexes = {}
-        for k, v in binary_binds.items():
-            key = k.encode()
-            item_index = 0
-            while True:
-                item_index = binary_query.find(key, item_index)
-                if item_index == -1:
-                    break
-                binary_indexes[item_index + len(key)] = key, v
-                item_index += len(key)
-        query = b''
-        start = 0
-        for loc in sorted(binary_indexes.keys()):
-            key, value = binary_indexes[loc]
-            query += binary_query[start:loc] + value + key
-            start = loc
-        query += binary_query[start:]
-    return query, bound_params
-
-
-def format_str(value: str):
-    return f"'{escape_str(value)}'"
-
-
-def escape_str(value: str):
-    return ''.join(f'{BS}{c}' if c in must_escape else c for c in value)
-
-
-# pylint: disable=too-many-return-statements
-def format_query_value(value: Any, server_tz: tzinfo = pytz.UTC):
-    """
-    Format Python values in a ClickHouse query
-    :param value: Python object
-    :param server_tz: Server timezone for adjusting datetime values
-    :return: Literal string for python value
-    """
-    if value is None:
-        return 'NULL'
-    if isinstance(value, str):
-        return format_str(value)
-    if isinstance(value, datetime):
-        if value.tzinfo is not None or server_tz != pytz.UTC:
-            value = value.astimezone(server_tz)
-        return f"'{value.strftime('%Y-%m-%d %H:%M:%S')}'"
-    if isinstance(value, date):
-        return f"'{value.isoformat()}'"
-    if isinstance(value, list):
-        return f"[{', '.join(str_query_value(x, server_tz) for x in value)}]"
-    if isinstance(value, tuple):
-        return f"({', '.join(str_query_value(x, server_tz) for x in value)})"
-    if isinstance(value, dict):
-        if common.get_setting('dict_parameter_format') == 'json':
-            return format_str(any_to_json(value).decode())
-        pairs = [str_query_value(k, server_tz) + ':' + str_query_value(v, server_tz)
-                 for k, v in value.items()]
-        return f"{{{', '.join(pairs)}}}"
-    if isinstance(value, Enum):
-        return format_query_value(value.value, server_tz)
-    if isinstance(value, (uuid.UUID, ipaddress.IPv4Address, ipaddress.IPv6Address)):
-        return f"'{value}'"
-    return value
-
-
-def str_query_value(value: Any, server_tz: tzinfo = pytz.UTC):
-    return str(format_query_value(value, server_tz))
-
-
-# pylint: disable=too-many-branches
-def format_bind_value(value: Any, server_tz: tzinfo = pytz.UTC, top_level: bool = True):
-    """
-    Format Python values in a ClickHouse query
-    :param value: Python object
-    :param server_tz: Server timezone for adjusting datetime values
-    :param top_level: Flag for top level for nested structures
-    :return: Literal string for python value
-    """
-
-    def recurse(x):
-        return format_bind_value(x, server_tz, False)
-
-    if value is None:
-        return '\\N'
-    if isinstance(value, str):
-        if top_level:
-            # At the top levels, strings must not be surrounded by quotes
-            return escape_str(value)
-        return format_str(value)
-    if isinstance(value, datetime):
-        value = value.astimezone(server_tz)
-        val = value.strftime('%Y-%m-%d %H:%M:%S')
-        if top_level:
-            return val
-        return f"'{val}'"
-    if isinstance(value, date):
-        if top_level:
-            return value.isoformat()
-        return f"'{value.isoformat()}'"
-    if isinstance(value, list):
-        return f"[{', '.join(recurse(x) for x in value)}]"
-    if isinstance(value, tuple):
-        return f"({', '.join(recurse(x) for x in value)})"
-    if isinstance(value, dict):
-        if common.get_setting('dict_parameter_format') == 'json':
-            return any_to_json(value).decode()
-        pairs = [recurse(k) + ':' + recurse(v)
-                 for k, v in value.items()]
-        return f"{{{', '.join(pairs)}}}"
-    if isinstance(value, Enum):
-        return recurse(value.value)
-    return str(value)
-
-
 comment_re = re.compile(r"(\".*?\"|\'.*?\')|(/\*.*?\*/|(--\s)[^\n]*$)", re.MULTILINE | re.DOTALL)
 
 
diff --git a/clickhouse_connect/driver/tools.py b/clickhouse_connect/driver/tools.py
index 54b4b459..42480858 100644
--- a/clickhouse_connect/driver/tools.py
+++ b/clickhouse_connect/driver/tools.py
@@ -2,7 +2,7 @@
 
 from clickhouse_connect.driver import Client
 from clickhouse_connect.driver.summary import QuerySummary
-from clickhouse_connect.driver.query import quote_identifier
+from clickhouse_connect.driver.binding import quote_identifier
 
 
 def insert_file(client: Client,
diff --git a/clickhouse_connect/driver/transform.py b/clickhouse_connect/driver/transform.py
index 03206d0c..1d20d9c6 100644
--- a/clickhouse_connect/driver/transform.py
+++ b/clickhouse_connect/driver/transform.py
@@ -96,7 +96,7 @@ def chunk_gen():
                     col_enc = col_name.encode()
                     write_leb128(len(col_enc), output)
                     output += col_enc
-                    col_enc = col_type.name.encode()
+                    col_enc = col_type.insert_name.encode()
                     write_leb128(len(col_enc), output)
                     output += col_enc
                     context.start_column(col_name)
diff --git a/docker-compose.yml b/docker-compose.yml
index f9717c78..f8e678de 100644
--- a/docker-compose.yml
+++ b/docker-compose.yml
@@ -1,7 +1,6 @@
-version: '3.8'
 services:
   clickhouse:
-    image: 'clickhouse/clickhouse-server:${CLICKHOUSE_VERSION-24.3-alpine}'
+    image: 'clickhouse/clickhouse-server:${CLICKHOUSE_VERSION-24.8-alpine}'
     container_name: 'clickhouse-connect-clickhouse-server'
     ports:
       - '8123:8123'
diff --git a/examples/params_example.py b/examples/params_example.py
index 748faf02..88fcceb9 100644
--- a/examples/params_example.py
+++ b/examples/params_example.py
@@ -2,7 +2,7 @@
 
 from datetime import datetime, timedelta
 
-from clickhouse_connect.driver.query import finalize_query
+from clickhouse_connect.driver.binding import finalize_query
 
 select_template = """
   SELECT
diff --git a/pylintrc b/pylintrc
index 47b404ae..7cddf107 100644
--- a/pylintrc
+++ b/pylintrc
@@ -255,7 +255,10 @@ exclude-too-few-public-methods=
 ignored-parents=Visitable
 
 # Maximum number of arguments for function / method.
-max-args=8
+max-args=12
+
+# Maximum positional arguments for function / method
+max-positional-arguments=12
 
 # Maximum number of attributes for a class (see R0902).
 max-attributes=7
diff --git a/pyproject.toml b/pyproject.toml
index 99710365..d1af9faf 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -7,3 +7,4 @@ build-backend = "setuptools.build_meta"
 log_cli = true
 log_cli_level = "INFO"
 env_files = ["test.env"]
+asyncio_default_fixture_loop_scope = "session"


Test patch:
diff --git a/clickhouse_connect/tools/testing.py b/clickhouse_connect/tools/testing.py
index 7084c71a..56ced725 100644
--- a/clickhouse_connect/tools/testing.py
+++ b/clickhouse_connect/tools/testing.py
@@ -1,7 +1,7 @@
 from typing import Sequence, Optional, Union, Dict, Any
 
 from clickhouse_connect.driver import Client
-from clickhouse_connect.driver.query import quote_identifier, str_query_value
+from clickhouse_connect.driver.binding import quote_identifier, str_query_value
 
 
 class TableContext:
diff --git a/tests/integration_tests/conftest.py b/tests/integration_tests/conftest.py
index 0122feef..4fb428e1 100644
--- a/tests/integration_tests/conftest.py
+++ b/tests/integration_tests/conftest.py
@@ -3,17 +3,16 @@
 import random
 import time
 from subprocess import Popen, PIPE
-from typing import Iterator, NamedTuple, Sequence, Optional
+from typing import Iterator, NamedTuple, Sequence, Optional, Callable
 
 from pytest import fixture
 
-from clickhouse_connect import create_client
 from clickhouse_connect import common
 from clickhouse_connect.driver.common import coerce_bool
 from clickhouse_connect.driver.exceptions import OperationalError
 from clickhouse_connect.tools.testing import TableContext
 from clickhouse_connect.driver.httpclient import HttpClient
-from clickhouse_connect.driver import AsyncClient, Client
+from clickhouse_connect.driver import AsyncClient, Client, create_client
 from tests.helpers import PROJECT_ROOT_DIR
 
 
@@ -42,11 +41,12 @@ def test_config_fixture() -> Iterator[TestConfig]:
     docker = host == 'localhost' and coerce_bool(os.environ.get('CLICKHOUSE_CONNECT_TEST_DOCKER', 'False'))
     port = int(os.environ.get('CLICKHOUSE_CONNECT_TEST_PORT', '0'))
     if not port:
-        port = 10723 if docker else 8123
-    cloud = coerce_bool(os.environ.get('CLICKHOUSE_CONNECT_TEST_CLOUD', 'True'))
+        port = 8123
+    cloud = coerce_bool(os.environ.get('CLICKHOUSE_CONNECT_TEST_CLOUD', 'False'))
     username = os.environ.get('CLICKHOUSE_CONNECT_TEST_USER', 'default')
     password = os.environ.get('CLICKHOUSE_CONNECT_TEST_PASSWORD', '')
-    test_database = f'ch_connect__{random.randint(100000, 999999)}__{int(time.time() * 1000)}'
+    test_database = os.environ.get('CLICKHOUSE_CONNECT_TEST_DATABASE',
+                                   f'ch_connect__{random.randint(100000, 999999)}__{int(time.time() * 1000)}')
     compress = os.environ.get('CLICKHOUSE_CONNECT_TEST_COMPRESS', 'True')
     insert_quorum = int(os.environ.get('CLICKHOUSE_CONNECT_TEST_INSERT_QUORUM', '0'))
     proxy_address = os.environ.get('CLICKHOUSE_CONNECT_TEST_PROXY_ADDR', '')
@@ -59,6 +59,33 @@ def test_db_fixture(test_config: TestConfig) -> Iterator[str]:
     yield test_config.test_database or 'default'
 
 
+@fixture(scope='session', name='test_create_client')
+def test_create_client_fixture(test_config: TestConfig) -> Callable:
+    def f(**kwargs):
+        client = create_client(host=test_config.host,
+                               port=test_config.port,
+                               user=test_config.username,
+                               password=test_config.password,
+                               compress=test_config.compress,
+                               settings={'allow_suspicious_low_cardinality_types': 1},
+                               client_name='int_tests/test',
+                               **kwargs)
+        if client.min_version('22.8'):
+            client.set_client_setting('database_replicated_enforce_synchronous_settings', 1)
+        if client.min_version('24.8') and not test_config.cloud:
+            client.set_client_setting('allow_experimental_json_type', 1)
+            client.set_client_setting('allow_experimental_dynamic_type', 1)
+            client.set_client_setting('allow_experimental_variant_type', 1)
+        if test_config.insert_quorum:
+            client.set_client_setting('insert_quorum', test_config.insert_quorum)
+        elif test_config.cloud:
+            client.set_client_setting('select_sequential_consistency', 1)
+        client.database = test_config.test_database
+        return client
+
+    return f
+
+
 @fixture(scope='session', name='test_table_engine')
 def test_table_engine_fixture() -> Iterator[str]:
     yield 'MergeTree'
@@ -66,7 +93,7 @@ def test_table_engine_fixture() -> Iterator[str]:
 
 # pylint: disable=too-many-branches
 @fixture(scope='session', autouse=True, name='test_client')
-def test_client_fixture(test_config: TestConfig, test_db: str) -> Iterator[Client]:
+def test_client_fixture(test_config: TestConfig, test_create_client: Callable) -> Iterator[Client]:
     compose_file = f'{PROJECT_ROOT_DIR}/docker-compose.yml'
     if test_config.docker:
         run_cmd(['docker', 'compose', '-f', compose_file, 'down', '-v'])
@@ -74,45 +101,27 @@ def test_client_fixture(test_config: TestConfig, test_db: str) -> Iterator[Clien
         pull_result = run_cmd(['docker', 'compose', '-f', compose_file, 'pull'])
         if pull_result[0]:
             raise TestException(f'Failed to pull latest docker image(s): {pull_result[2]}')
-        up_result = run_cmd(['docker', 'compose', '-f', compose_file, 'up', '-d', 'clickhouse'])
+        up_result = run_cmd(['docker', 'compose', '-f', compose_file, 'up', '-d'])
         if up_result[0]:
             raise TestException(f'Failed to start docker: {up_result[2]}')
         time.sleep(5)
     tries = 0
+    if test_config.docker:
+        HttpClient.params = {'SQL_test_setting': 'value'}
+        HttpClient.valid_transport_settings.add('SQL_test')
     while True:
         tries += 1
         try:
-            if test_config.docker:
-                HttpClient.params = {'SQL_test_setting': 'value'}
-                HttpClient.valid_transport_settings.add('SQL_test')
-            client = create_client(
-                host=test_config.host,
-                port=test_config.port,
-                username=test_config.username,
-                password=test_config.password,
-                query_limit=0,
-                compress=test_config.compress,
-                client_name='int_tests/test',
-                settings={'allow_suspicious_low_cardinality_types': True,
-                          'insert_deduplicate': False,
-                          'async_insert': 0}
-            )
+            client = test_create_client()
             break
         except OperationalError as ex:
             if tries > 10:
                 raise TestException('Failed to connect to ClickHouse server after 30 seconds') from ex
             time.sleep(3)
-    if client.min_version('22.8'):
-        client.set_client_setting('database_replicated_enforce_synchronous_settings', 1)
-    if test_config.insert_quorum:
-        client.set_client_setting('insert_quorum', test_config.insert_quorum)
-    elif test_config.cloud:
-        client.set_client_setting('select_sequential_consistency', 1)
-    client.command(f'CREATE DATABASE IF NOT EXISTS {test_db}', use_database=False)
-    client.database = test_db
+    client.command(f'CREATE DATABASE IF NOT EXISTS {test_config.test_database}', use_database=False)
     yield client
 
-    client.command(f'DROP database IF EXISTS {test_db}', use_database=False)
+    # client.command(f'DROP database IF EXISTS {test_db}', use_database=False)
     if test_config.docker:
         down_result = run_cmd(['docker', 'compose', '-f', compose_file, 'down', '-v'])
         if down_result[0]:
diff --git a/tests/integration_tests/test_arrow.py b/tests/integration_tests/test_arrow.py
index d6d80e37..c7f80ff5 100644
--- a/tests/integration_tests/test_arrow.py
+++ b/tests/integration_tests/test_arrow.py
@@ -77,12 +77,10 @@ def test_arrow_map(test_client: Client, table_context: Callable):
                                           'update_time DateTime DEFAULT now()']):
         data = [[date(2023, 10, 15), 'C1', {'k': 2.5, 'd': 0, 'j': 0}],
                 [date(2023, 10, 16), 'C2', {'k': 3.5, 'd': 0, 'j': -.372}]]
-        insert_result = test_client.insert('test_arrow_map', data, column_names=('trade_date', 'code', 'kdj'))
-        assert 2 == insert_result.written_rows
+        test_client.insert('test_arrow_map', data, column_names=('trade_date', 'code', 'kdj'),
+                           settings={'insert_deduplication_token': '10381'})
         arrow_table = test_client.query_arrow('SELECT * FROM test_arrow_map ORDER BY trade_date',
                                               use_strings=True)
-        print(arrow_table)
         assert isinstance(arrow_table.schema, arrow.Schema)
-        insert_result = test_client.insert_arrow('test_arrow_map', arrow_table)
+        test_client.insert_arrow('test_arrow_map', arrow_table, settings={'insert_deduplication_token': '10382'})
         assert 4 == test_client.command('SELECT count() FROM test_arrow_map')
-        assert 2 == insert_result.written_rows
diff --git a/tests/integration_tests/test_client.py b/tests/integration_tests/test_client.py
index aba0b2bc..b9417e6b 100644
--- a/tests/integration_tests/test_client.py
+++ b/tests/integration_tests/test_client.py
@@ -103,6 +103,9 @@ def test_get_columns_only(test_client: Client):
     assert result.column_names == ('name', 'database')
     assert len(result.result_set) == 0
 
+    test_client.query('CREATE TABLE IF NOT EXISTS test_zero_insert (v Int8) ENGINE MergeTree() ORDER BY tuple()')
+    test_client.query('INSERT INTO test_zero_insert SELECT 1 LIMIT 0')
+
 
 def test_no_limit(test_client: Client):
     old_limit = test_client.query_limit
diff --git a/tests/integration_tests/test_dynamic.py b/tests/integration_tests/test_dynamic.py
new file mode 100644
index 00000000..bad75b12
--- /dev/null
+++ b/tests/integration_tests/test_dynamic.py
@@ -0,0 +1,102 @@
+import datetime
+from ipaddress import IPv4Address
+from typing import Callable
+from uuid import UUID
+
+import pytest
+
+from clickhouse_connect.datatypes.format import set_write_format
+from clickhouse_connect.driver import Client
+
+
+def test_variant(test_client: Client, table_context: Callable):
+    if not test_client.get_client_setting('allow_experimental_variant_type'):
+        pytest.skip(f'New Variant type not available in this version: {test_client.server_version}')
+    with table_context('basic_variants', [
+        'key Int32',
+        'v1 Variant(UInt64, String, Array(UInt64), UUID)',
+        'v2 Variant(IPv4, Decimal(10, 2))']):
+        data = [[1, 58322, None],
+                [2, 'a string', 55.2],
+                [3, 'bef56f14-0870-4f82-a35e-9a47eff45a5b', 777.25],
+                [4, [120, 250], '243.12.55.44']
+                ]
+        test_client.insert('basic_variants', data)
+        result = test_client.query('SELECT * FROM basic_variants ORDER BY key').result_set
+        assert result[2][1] == UUID('bef56f14-0870-4f82-a35e-9a47eff45a5b')
+        assert result[2][2] == 777.25
+        assert result[3][1] == [120, 250]
+        assert result[3][2] == IPv4Address('243.12.55.44')
+
+
+def test_dynamic(test_client: Client, table_context: Callable):
+    if not test_client.get_client_setting('allow_experimental_dynamic_type'):
+        pytest.skip(f'New Dynamic type not available in this version: {test_client.server_version}')
+    with table_context('basic_dynamic', [
+        'key UInt64',
+        'v1 Dynamic',
+        'v2 Dynamic']):
+        data = [[1, 58322, 15.5],
+                [3, 'bef56f14-0870-4f82-a35e-9a47eff45a5b', 777.25],
+                [2, 'a string', 55.2],
+                [4, [120, 250], 577.22]
+                ]
+        test_client.insert('basic_dynamic', data)
+        result = test_client.query('SELECT * FROM basic_dynamic ORDER BY key').result_set
+        assert result[2][1] == 'bef56f14-0870-4f82-a35e-9a47eff45a5b'
+        assert result[3][1] == '[120, 250]'
+        assert result[2][2] == '777.25'
+
+
+def test_basic_json(test_client: Client, table_context: Callable):
+    if not test_client.get_client_setting('allow_experimental_json_type'):
+        pytest.skip(f'New JSON type not available in this version: {test_client.server_version}')
+    with table_context('new_json_basic', [
+        'key Int32',
+        'value JSON',
+        "null_value JSON"
+    ]):
+        jv3 = {'key3': 752, 'value.2': 'v2_rules', 'blank': None}
+        jv1 = {'key1': 337, 'value.2': 'vvvv', 'HKD@spéçiäl': 'Special K', 'blank': 'not_really_blank'}
+        njv2 = {'nk1': -302, 'nk2': {'sub1': 372, 'sub2': 'a string'}}
+        njv3 = {'nk1': 5832.44, 'nk2': {'sub1': 47788382, 'sub2': 'sub2val', 'sub3': 'sub3str', 'space key': 'spacey'}}
+        test_client.insert('new_json_basic', [
+            [5, jv1, None],
+            [20, None, njv2],
+            [25, jv3, njv3]])
+
+        result = test_client.query('SELECT * FROM new_json_basic ORDER BY key').result_set
+        json1 = result[0][1]
+        assert json1['HKD@spéçiäl'] == 'Special K'
+        assert 'key3' not in json1
+        json2 = result[1][2]
+        assert json2['nk1'] == -302.0
+        assert json2['nk2']['sub2'] == 'a string'
+        assert json2['nk2'].get('sub3') is None
+        json3 = result[2][1]
+        assert json3['value']['2'] == 'v2_rules'
+        assert 'blank' not in json3
+        assert 'key1' not in json3
+        assert json3['key3'] == 752
+        null_json3 = result[2][2]
+        assert null_json3['nk2']['space key'] == 'spacey'
+
+        set_write_format('JSON', 'string')
+        test_client.insert('new_json_basic', [[999, '{"key4": 283, "value.2": "str_value"}', '{"nk1":53}']])
+        result = test_client.query('SELECT value.key4, null_value.nk1 FROM new_json_basic ORDER BY key').result_set
+        assert result[3][0] == 283
+        assert result[3][1] == 53
+
+
+def test_typed_json(test_client: Client, table_context: Callable):
+    if not test_client.get_client_setting('allow_experimental_json_type'):
+        pytest.skip(f'New JSON type not available in this version: {test_client.server_version}')
+    with table_context('new_json_typed', [
+        'key Int32',
+        'value JSON(max_dynamic_paths=150, `a.b` DateTime64(3), SKIP a.c)'
+    ]):
+        v1 = '{"a":{"b":"2020-10-15T10:15:44.877", "c":"skip_me"}}'
+        test_client.insert('new_json_typed', [[1, v1]])
+        result = test_client.query('SELECT * FROM new_json_typed ORDER BY key')
+        json1 = result.result_set[0][1]
+        assert json1['a']['b'] == datetime.datetime(2020, 10, 15, 10, 15, 44, 877000)
diff --git a/tests/integration_tests/test_external_data.py b/tests/integration_tests/test_external_data.py
index 167620b4..4ebfb688 100644
--- a/tests/integration_tests/test_external_data.py
+++ b/tests/integration_tests/test_external_data.py
@@ -55,6 +55,8 @@ def test_external_parquet(test_config: TestConfig, test_client: Client):
     if test_config.cloud:
         pytest.skip('External data join not working in SMT, skipping')
     movies_file = f'{Path(__file__).parent}/movies.parquet'
+    test_client.command('DROP TABLE IF EXISTS movies')
+    test_client.command('DROP TABLE IF EXISTS num')
     test_client.command("""
 CREATE TABLE IF NOT EXISTS num (number UInt64, t String)
 ENGINE = MergeTree
diff --git a/tests/integration_tests/test_inserts.py b/tests/integration_tests/test_inserts.py
index 99d8898b..2be13b16 100644
--- a/tests/integration_tests/test_inserts.py
+++ b/tests/integration_tests/test_inserts.py
@@ -12,8 +12,9 @@ def test_insert(test_client: Client, test_table_engine: str):
         test_client.command('DROP TABLE IF EXISTS test_system_insert SYNC')
     test_client.command(f'CREATE TABLE test_system_insert AS system.tables Engine {test_table_engine} ORDER BY name')
     tables_result = test_client.query('SELECT * from system.tables')
-    insert_result = test_client.insert(table='test_system_insert', column_names='*', data=tables_result.result_set)
-    assert int(tables_result.summary['read_rows']) == insert_result.written_rows
+    test_client.insert(table='test_system_insert', column_names='*', data=tables_result.result_set)
+    copy_result = test_client.command('SELECT count() from test_system_insert')
+    assert tables_result.row_count == copy_result
     test_client.command('DROP TABLE IF EXISTS test_system_insert')
 
 
diff --git a/tests/integration_tests/test_multithreading.py b/tests/integration_tests/test_multithreading.py
index 8f628f03..1b01eaf4 100644
--- a/tests/integration_tests/test_multithreading.py
+++ b/tests/integration_tests/test_multithreading.py
@@ -1,10 +1,15 @@
 import threading
 
+import pytest
+
 from clickhouse_connect.driver import Client
 from clickhouse_connect.driver.exceptions import ProgrammingError
+from tests.integration_tests.conftest import TestConfig
 
 
-def test_threading_error(test_client: Client):
+def test_threading_error(test_config: TestConfig, test_client: Client):
+    if test_config.cloud:
+        pytest.skip('Skipping threading test in ClickHouse Cloud')
     thrown = None
 
     class QueryThread (threading.Thread):
diff --git a/tests/integration_tests/test_native.py b/tests/integration_tests/test_native.py
index 6fd96076..c65a20a4 100644
--- a/tests/integration_tests/test_native.py
+++ b/tests/integration_tests/test_native.py
@@ -58,10 +58,10 @@ def test_nulls(test_client: Client, table_context: Callable):
         assert result[3] == (4, 'nonnull2', None)
 
 
-def test_json(test_client: Client, table_context: Callable):
-    if not coerce_bool(os.environ.get('CLICKHOUSE_CONNECT_TEST_JSON_TYPE')):
+def test_old_json(test_client: Client, table_context: Callable):
+    if not coerce_bool(os.environ.get('CLICKHOUSE_CONNECT_TEST_OLD_JSON_TYPE')):
         pytest.skip('Deprecated JSON type not tested')
-    with table_context('native_json_test', [
+    with table_context('old_json_test', [
         'key Int32',
         'value JSON',
         'e2 Int32',
@@ -71,12 +71,12 @@ def test_json(test_client: Client, table_context: Callable):
         jv3 = {'key3': 752, 'value.2': 'v2_rules', 'blank': None}
         njv2 = {'nk1': -302, 'nk2': {'sub1': 372, 'sub2': 'a string'}}
         njv3 = {'nk1': 5832.44, 'nk2': {'sub1': 47788382, 'sub2':'sub2val', 'sub3': 'sub3str', 'space key': 'spacey'}}
-        test_client.insert('native_json_test', [
+        test_client.insert('old_json_test', [
             [5, jv1, -44, None],
             [20, None, 5200, njv2],
             [25, jv3, 7302, njv3]])
 
-        result = test_client.query('SELECT * FROM native_json_test ORDER BY key')
+        result = test_client.query('SELECT * FROM old_json_test ORDER BY key')
         json1 = result.result_set[0][1]
         assert json1['HKD@spéçiäl'] == 'Special K'
         assert json1['key3'] == 0
@@ -167,14 +167,12 @@ def test_tuple_inserts(test_client: Client, table_context: Callable):
     with table_context('insert_tuple_test', ['key Int32', 'named Tuple(fl Float64, `ns space` Nullable(String))',
                                              'unnamed Tuple(Float64, Nullable(String))']):
         data = [[1, (3.55, 'str1'), (555, None)], [2, (-43.2, None), (0, 'str2')]]
-        result = test_client.insert('insert_tuple_test', data)
-        assert 2 == result.written_rows
+        test_client.insert('insert_tuple_test', data, settings={'insert_deduplication_token': 5772})
 
         data = [[1, {'fl': 3.55, 'ns space': 'str1'}, (555, None)], [2, {'fl': -43.2}, (0, 'str2')]]
-        result = test_client.insert('insert_tuple_test', data)
-        assert 2 == result.written_rows
-
+        test_client.insert('insert_tuple_test', data, settings={'insert_deduplication_token': 5773})
         query_result = test_client.query('SELECT * FROM insert_tuple_test ORDER BY key').result_rows
+        assert len(query_result) == 4
         assert query_result[0] == query_result[1]
         assert query_result[2] == query_result[3]
 
@@ -182,10 +180,10 @@ def test_tuple_inserts(test_client: Client, table_context: Callable):
 def test_point_inserts(test_client: Client, table_context: Callable):
     with table_context('insert_point_test', ['key Int32', 'point Point']):
         data = [[1, (3.55, 3.55)], [2, (4.55, 4.55)]]
-        result = test_client.insert('insert_point_test', data)
-        assert 2 == result.written_rows
+        test_client.insert('insert_point_test', data)
 
         query_result = test_client.query('SELECT * FROM insert_point_test ORDER BY key').result_rows
+        assert len(query_result) == 2
         assert query_result[0] == (1, (3.55, 3.55))
         assert query_result[1] == (2, (4.55, 4.55))
 
diff --git a/tests/integration_tests/test_pandas.py b/tests/integration_tests/test_pandas.py
index fba17594..6447cc26 100644
--- a/tests/integration_tests/test_pandas.py
+++ b/tests/integration_tests/test_pandas.py
@@ -10,6 +10,7 @@
 from clickhouse_connect.driver.exceptions import DataError
 from clickhouse_connect.driver.options import np, pd
 from tests.helpers import random_query
+from tests.integration_tests.conftest import TestConfig
 from tests.integration_tests.datasets import null_ds, null_ds_columns, null_ds_types
 
 pytestmark = pytest.mark.skipif(pd is None, reason='Pandas package not installed')
@@ -258,7 +259,7 @@ def test_pandas_row_df(test_client: Client, table_context:Callable):
         assert str(result_df.dtypes[2]) == 'string'
         assert result_df.iloc[0]['key'] == 2
         assert result_df.iloc[0]['dt'] == pd.Timestamp(2023, 10, 15, 14, 50, 2, 4038)
-        assert result_df.iloc[0]['fs'] == 'bit\0\0'
+        assert result_df.iloc[0]['fs'] == 'bit'
         assert len(result_df) == 1
         assert source_df.equals(df)
 
@@ -276,7 +277,9 @@ def test_pandas_null_strings(test_client: Client, table_context:Callable):
             test_client.insert_df('test_pandas_null_strings', df)
 
 
-def test_pandas_small_blocks(test_client: Client):
+def test_pandas_small_blocks(test_config: TestConfig, test_client: Client):
+    if test_config.cloud:
+        pytest.skip('Skipping performance test in ClickHouse Cloud')
     res = test_client.query_df('SELECT number, randomString(512) FROM numbers(1000000)',
                                settings={'max_block_size': 250})
     assert len(res) == 1000000
diff --git a/tests/integration_tests/test_params.py b/tests/integration_tests/test_params.py
index 4d0ca65d..98c4fabe 100644
--- a/tests/integration_tests/test_params.py
+++ b/tests/integration_tests/test_params.py
@@ -2,6 +2,7 @@
 from typing import Callable
 
 from clickhouse_connect.driver import Client
+from clickhouse_connect.driver.binding import DT64Param
 
 
 def test_params(test_client: Client, table_context: Callable):
@@ -61,3 +62,27 @@ def test_params(test_client: Client, table_context: Callable):
     result = test_client.query(
         'SELECT count() FROM system.tables WHERE total_rows > %(p_0)d and total_rows < %(p_1)f', parameters=num_params)
     assert result.first_row[0] > 0
+
+
+def test_datetime_64_params(test_client: Client):
+    dt_values = [datetime(2023, 6, 1, 7, 40, 2, 250306), datetime(2023, 8, 17, 20, 0, 10, 777722)]
+    dt_params = {f'd{ix}': DT64Param(v) for ix, v in enumerate(dt_values)}
+    result = test_client.query('SELECT {d0:DateTime64(3)}, {d1:Datetime64(9)}', parameters=dt_params).first_row
+    assert result[0] == dt_values[0].replace(microsecond=250000)
+    assert result[1] == dt_values[1]
+
+    result = test_client.query('SELECT {a1:Array(DateTime64(6))}', parameters={'a1': [dt_params['d0'], dt_params['d1']]}).first_row
+    assert result[0] == dt_values
+
+    dt_params = {f'd{ix}_64': v for ix, v in enumerate(dt_values)}
+    result = test_client.query('SELECT {d0:DateTime64(3)}, {d1:Datetime64(9)}', parameters=dt_params).first_row
+    assert result[0] == dt_values[0].replace(microsecond=250000)
+    assert result[1] == dt_values[1]
+
+    result = test_client.query('SELECT {a1:Array(DateTime64(6))}',
+                               parameters={'a1_64': dt_values}).first_row
+    assert result[0] == dt_values
+
+    dt_params = [DT64Param(v) for v in dt_values]
+    result = test_client.query("SELECT %s as string, toDateTime64(%s,6) as dateTime", parameters = dt_params).first_row
+    assert result == ('2023-06-01 07:40:02.250306', dt_values[1])
diff --git a/tests/integration_tests/test_raw_insert.py b/tests/integration_tests/test_raw_insert.py
index 5e729a0e..518088ac 100644
--- a/tests/integration_tests/test_raw_insert.py
+++ b/tests/integration_tests/test_raw_insert.py
@@ -24,11 +24,10 @@ def test_raw_insert_compression(test_client: Client, table_context: Callable):
     with open(data_file, mode='rb') as movies_file:
         data = movies_file.read()
     with table_context('test_gzip_movies', ['movie String', 'year UInt16', 'rating Decimal32(3)']):
-        insert_result = test_client.raw_insert('test_gzip_movies', None, data, fmt='CSV', compression='gzip',
+        test_client.raw_insert('test_gzip_movies', None, data, fmt='CSV', compression='gzip',
                                                settings={'input_format_allow_errors_ratio': .2,
                                                          'input_format_allow_errors_num': 5}
                                                )
-        assert 248 == insert_result.written_rows
         res = test_client.query(
             'SELECT count() as count, sum(rating) as rating, max(year) as year FROM test_gzip_movies').first_item
         assert res['count'] == 248
diff --git a/tests/integration_tests/test_session_id.py b/tests/integration_tests/test_session_id.py
new file mode 100644
index 00000000..84c683f9
--- /dev/null
+++ b/tests/integration_tests/test_session_id.py
@@ -0,0 +1,78 @@
+import uuid
+
+from typing import Callable
+
+import pytest
+
+from clickhouse_connect.driver import create_async_client
+from tests.integration_tests.conftest import TestConfig
+
+SESSION_KEY = 'session_id'
+
+
+def test_client_default_session_id(test_create_client: Callable):
+    # by default, the sync client will autogenerate the session id
+    client = test_create_client()
+    session_id = client.get_client_setting(SESSION_KEY)
+    try:
+        uuid.UUID(session_id)
+    except ValueError:
+        pytest.fail(f"Invalid session_id: {session_id}")
+    client.close()
+
+
+def test_client_autogenerate_session_id(test_create_client: Callable):
+    client = test_create_client()
+    session_id = client.get_client_setting(SESSION_KEY)
+    try:
+        uuid.UUID(session_id)
+    except ValueError:
+        pytest.fail(f"Invalid session_id: {session_id}")
+
+
+def test_client_custom_session_id(test_create_client: Callable):
+    session_id = 'custom_session_id'
+    client = test_create_client(session_id=session_id)
+    assert client.get_client_setting(SESSION_KEY) == session_id
+    client.close()
+
+
+@pytest.mark.asyncio
+async def test_async_client_default_session_id(test_config: TestConfig):
+    # by default, the async client will NOT autogenerate the session id
+    async_client = await create_async_client(database=test_config.test_database,
+                                             host=test_config.host,
+                                             port=test_config.port,
+                                             user=test_config.username,
+                                             password=test_config.password)
+    assert async_client.get_client_setting(SESSION_KEY) is None
+    async_client.close()
+
+
+@pytest.mark.asyncio
+async def test_async_client_autogenerate_session_id(test_config: TestConfig):
+    async_client = await create_async_client(database=test_config.test_database,
+                                             host=test_config.host,
+                                             port=test_config.port,
+                                             user=test_config.username,
+                                             password=test_config.password,
+                                             autogenerate_session_id=True)
+    session_id = async_client.get_client_setting(SESSION_KEY)
+    try:
+        uuid.UUID(session_id)
+    except ValueError:
+        pytest.fail(f"Invalid session_id: {session_id}")
+    async_client.close()
+
+
+@pytest.mark.asyncio
+async def test_async_client_custom_session_id(test_config: TestConfig):
+    session_id = 'custom_session_id'
+    async_client = await create_async_client(database=test_config.test_database,
+                                             host=test_config.host,
+                                             port=test_config.port,
+                                             user=test_config.username,
+                                             password=test_config.password,
+                                             session_id=session_id)
+    assert async_client.get_client_setting(SESSION_KEY) == session_id
+    async_client.close()
diff --git a/tests/tls/test_tls.py b/tests/integration_tests/test_tls.py
similarity index 84%
rename from tests/tls/test_tls.py
rename to tests/integration_tests/test_tls.py
index 3786e20b..378c064a 100644
--- a/tests/tls/test_tls.py
+++ b/tests/integration_tests/test_tls.py
@@ -18,24 +18,24 @@ def test_basic_tls():
         pytest.skip('TLS tests not enabled')
     client = get_client(interface='https', host=host, port=10843, verify=False)
     assert client.command("SELECT 'insecure'") == 'insecure'
-    client.http.clear()
+    client.close_connections()
 
     client = get_client(interface='https', host=host, port=10843, ca_cert=f'{cert_dir}ca.crt')
     assert client.command("SELECT 'verify_server'") == 'verify_server'
-    client.http.clear()
+    client.close_connections()
 
     try:
         get_client(interface='https', host='localhost', port=10843, ca_cert=f'{cert_dir}ca.crt')
         pytest.fail('Expected TLS exception with a different hostname')
     except OperationalError as ex:
-        assert isinstance(ex.__cause__.reason, SSLError)
-    client.http.clear()
+        assert isinstance(ex.__cause__.reason, SSLError) # pylint: disable=no-member
+    client.close_connections()
 
     try:
-        get_client(interface='https', host=host, port=10843)
+        get_client(interface='https', host='localhost', port=10843)
         pytest.fail('Expected TLS exception with a self-signed cert')
     except OperationalError as ex:
-        assert isinstance(ex.__cause__.reason, SSLError)
+        assert isinstance(ex.__cause__.reason, SSLError) # pylint: disable=no-member
 
 
 def test_mutual_tls():
diff --git a/tests/integration_tests/test_tools.py b/tests/integration_tests/test_tools.py
index 854b231a..6ed546af 100644
--- a/tests/integration_tests/test_tools.py
+++ b/tests/integration_tests/test_tools.py
@@ -9,10 +9,9 @@
 def test_csv_upload(test_client: Client, table_context: Callable):
     data_file = f'{Path(__file__).parent}/movies.csv.gz'
     with table_context('test_csv_upload', ['movie String', 'year UInt16', 'rating Decimal32(3)']):
-        insert_result = insert_file(test_client, 'test_csv_upload', data_file,
+        insert_file(test_client, 'test_csv_upload', data_file,
                                     settings={'input_format_allow_errors_ratio': .2,
                                               'input_format_allow_errors_num': 5})
-        assert 248 == insert_result.written_rows
         res = test_client.query(
             'SELECT count() as count, sum(rating) as rating, max(year) as year FROM test_csv_upload').first_item
         assert res['count'] == 248
@@ -23,9 +22,8 @@ def test_parquet_upload(test_config: TestConfig, test_client: Client, table_cont
     data_file = f'{Path(__file__).parent}/movies.parquet'
     full_table = f'{test_config.test_database}.test_parquet_upload'
     with table_context(full_table, ['movie String', 'year UInt16', 'rating Float64']):
-        insert_result = insert_file(test_client, full_table, data_file, 'Parquet',
+        insert_file(test_client, full_table, data_file, 'Parquet',
                                     settings={'output_format_parquet_string_as_string': 1})
-        assert 250 == insert_result.written_rows
         res = test_client.query(
             f'SELECT count() as count, sum(rating) as rating, max(year) as year FROM {full_table}').first_item
         assert res['count'] == 250
diff --git a/tests/test_requirements.txt b/tests/test_requirements.txt
index 1b87e750..b0e06c6e 100644
--- a/tests/test_requirements.txt
+++ b/tests/test_requirements.txt
@@ -3,7 +3,7 @@ urllib3>=1.26
 setuptools
 certifi
 sqlalchemy>1.3.21,<2.0
-cython==3.0.10
+cython==3.0.11
 pyarrow
 pytest
 pytest-asyncio
diff --git a/tests/unit_tests/test_driver/test_params.py b/tests/unit_tests/test_driver/test_params.py
index f10b8d1f..2e29ef60 100644
--- a/tests/unit_tests/test_driver/test_params.py
+++ b/tests/unit_tests/test_driver/test_params.py
@@ -2,7 +2,7 @@
 
 import pytest
 
-from clickhouse_connect.driver.query import finalize_query, format_bind_value
+from clickhouse_connect.driver.binding import finalize_query, format_bind_value
 
 
 def test_finalize():
diff --git a/tests/unit_tests/test_driver/test_parser.py b/tests/unit_tests/test_driver/test_parser.py
index 672f27de..24202b1d 100644
--- a/tests/unit_tests/test_driver/test_parser.py
+++ b/tests/unit_tests/test_driver/test_parser.py
@@ -22,6 +22,26 @@ def test_map_type():
     assert ch_type.name == 'Map(String, Decimal(5, 5))'
 
 
+def test_variant_type():
+    ch_type = get_from_name('Variant(UInt64, String, Array(UInt64))')
+    assert ch_type.name == 'Variant(UInt64, String, Array(UInt64))'
+
+
+def test_json_type():
+    names = ['JSON',
+             'JSON(max_dynamic_paths=100, a.b UInt32, SKIP `a.e`)',
+             "JSON(max_dynamic_types = 55, SKIP REGEXP 'a[efg]')",
+             'JSON(max_dynamic_types = 33, `a.b` UInt64, b.c String)']
+    parsed = ['JSON',
+               'JSON(max_dynamic_paths = 100, `a.b` UInt32, SKIP `a.e`)',
+               "JSON(max_dynamic_types = 55, SKIP REGEXP 'a[efg]')",
+               'JSON(max_dynamic_types = 33, `a.b` UInt64, `b.c` String)'
+               ]
+    for name, x in zip(names, parsed):
+        ch_type = get_from_name(name)
+        assert x == ch_type.name
+
+
 def test_remove_comments():
     sql = """SELECT -- 6dcd92a04feb50f14bbcf07c661680ba
 * FROM benchmark_results /*With an inline comment */ WHERE result = 'True'
diff --git a/tests/unit_tests/test_driver/test_session_id.py b/tests/unit_tests/test_driver/test_session_id.py
deleted file mode 100644
index 2a632e64..00000000
--- a/tests/unit_tests/test_driver/test_session_id.py
+++ /dev/null
@@ -1,62 +0,0 @@
-import uuid
-
-import pytest
-
-from clickhouse_connect.driver import create_async_client, create_client
-
-SESSION_KEY = 'session_id'
-
-
-def test_client_default_session_id():
-    # by default, the sync client will autogenerate the session id
-    client = create_client()
-    session_id = client.get_client_setting(SESSION_KEY)
-    try:
-        uuid.UUID(session_id)
-    except ValueError:
-        pytest.fail(f"Invalid session_id: {session_id}")
-    client.close()
-
-
-def test_client_autogenerate_session_id():
-    client = create_client(autogenerate_session_id=True)
-    session_id = client.get_client_setting(SESSION_KEY)
-    try:
-        uuid.UUID(session_id)
-    except ValueError:
-        pytest.fail(f"Invalid session_id: {session_id}")
-    client.close()
-
-
-def test_client_custom_session_id():
-    session_id = 'custom_session_id'
-    client = create_client(session_id=session_id)
-    assert client.get_client_setting(SESSION_KEY) == session_id
-    client.close()
-
-
-@pytest.mark.asyncio
-async def test_async_client_default_session_id():
-    # by default, the async client will NOT autogenerate the session id
-    async_client = await create_async_client()
-    assert async_client.get_client_setting(SESSION_KEY) is None
-    async_client.close()
-
-
-@pytest.mark.asyncio
-async def test_async_client_autogenerate_session_id():
-    async_client = await create_async_client(autogenerate_session_id=True)
-    session_id = async_client.get_client_setting(SESSION_KEY)
-    try:
-        uuid.UUID(session_id)
-    except ValueError:
-        pytest.fail(f"Invalid session_id: {session_id}")
-    async_client.close()
-
-
-@pytest.mark.asyncio
-async def test_async_client_custom_session_id():
-    session_id = 'custom_session_id'
-    async_client = await create_async_client(session_id=session_id)
-    assert async_client.get_client_setting(SESSION_KEY) == session_id
-    async_client.close()